for tree traversal, register the thread which got to the front

src/vdp_pesi.c

@@ -246,6 +246,7 @@ struct bytes_tree {
 	pthread_cond_t		cond;
 	struct node		*root;
 	struct node		*front;
+	struct worker		*front_owner;
 	struct worker		*unpend_owner;
 	int			npending;
 	int			retval;
@@ -1306,6 +1307,11 @@ ved_decode_len(struct req *req, const uint8_t **pp)
  * to minimize lock contention, we walk the tree and gather work on a list
  */
 
+#define set_front(tree, node, wrk) do {					\
+		(tree)->front = node;					\
+		(tree)->front_owner = wrk;				\
+	} while(0)
+
 static void
 bytes_unpend_worklist(struct req *req, struct bytes_tree *tree,
     struct node_head *work)
@@ -1363,7 +1369,7 @@ bytes_unpend_worklist(struct req *req, struct bytes_tree *tree,
 				VSLdbgv(req, "bytes_unpend: return at "
 					"node %p state %u - await sibling",
 					node, node->state);
-				tree->front = node;
+				set_front(tree, node, req->wrk);
 				return;
 			}
 
@@ -1380,7 +1386,7 @@ bytes_unpend_worklist(struct req *req, struct bytes_tree *tree,
 
 			AZ(node);
 			VSLdbgv(req, "bytes_unpend: reached root");
-			tree->front = tree->root;
+			set_front(tree, tree->root, req->wrk);
 			return;
 		}
 
@@ -1388,7 +1394,7 @@ bytes_unpend_worklist(struct req *req, struct bytes_tree *tree,
 			VSLdbgv(req, "bytes_unpend: return at "
 				"node %p state %u - private",
 				node, node->state);
-			tree->front = node;
+			set_front(tree, node, req->wrk);
 			return;
 		}
 
@@ -1402,7 +1408,7 @@ bytes_unpend_worklist(struct req *req, struct bytes_tree *tree,
 
 		if (node->state == ST_OPEN) {
 			VSLdbg(req, "bytes_unpend: init node no children");
-			tree->front = node;
+			set_front(tree, node, req->wrk);
 			return;
 		}
 
@@ -1438,6 +1444,7 @@ bytes_unpend_worklist(struct req *req, struct bytes_tree *tree,
 		check = CHK_ANY;
 	}
 }
+#undef set_front
 
 /*
  * restore the original subreq state and take care that we do not mess up flags,