rework tree traversal

src/vdp_pesi.c

@@ -1302,15 +1302,8 @@ ved_decode_len(struct req *req, const uint8_t **pp)
 	return (l);
 }
 
-/* VDP pesi_buf */
-
 /*
- * to minimize lock contention, we walk the tree looking for work
- *
- * because pointers in data nodes may depend on objcore refs, we return when we
- * are done with a subtree (iow, when we need to go to an undelivered node)
- *
- * so basically we work depth first, then right, but not up
+ * to minimize lock contention, we walk the tree and gather work on a list
  */
 
 static void
@@ -1337,8 +1330,6 @@ bytes_unpend_worklist(struct req *req, struct bytes_tree *tree,
 	CHECK_OBJ_NOTNULL(tree->front, NODE_MAGIC);
 	CHECK_OBJ_NOTNULL(tree->root, NODE_MAGIC);
 
-	assert(req == tree->root->nexus.req);
-
 	assert(req->esi_level == 0);
 
 	if (tree->root->state == ST_DELIVERED) {
@@ -1347,6 +1338,8 @@ bytes_unpend_worklist(struct req *req, struct bytes_tree *tree,
 		return;
 	}
 
+	/* XXX what if we are a different worker and within a tree ?*/
+
 	/* when starting, we do not want to see any ST_UNPENDING nodes */
 	check = CHK_DELI;
 
@@ -1366,26 +1359,28 @@ bytes_unpend_worklist(struct req *req, struct bytes_tree *tree,
 				continue;
 			}
 
-			if (node->parent && node->parent->state != ST_OPEN) {
-				if (VSTAILQ_EMPTY(work)) {
-					node = node->parent;
-					continue;
-				}
-				/* need to DELIVER children of this parent
-				 * first, so we can unref any storage
-				 * XXX this comment is probably obsolete -
-				 * rethink
-				 */
-				tree->front = node->parent;
+			if (node->parent && node->parent->state == ST_OPEN) {
+				tree->front = node;
 				VSLdbgv(req, "bytes_unpend: return at "
-					"node %p state %u", node, node->state);
+					"node %p state %u - await sibling",
+					node, node->state);
 				return;
 			}
 
-			tree->front = node;
-			VSLdbgv(req, "bytes_unpend: return at "
-				"node %p state %u - await sibling",
-				node, node->state);
+			/* up and right */
+			node = node->parent;
+			while (node != NULL) {
+				next = VSTAILQ_NEXT(node, sibling);
+				if (next != NULL) {
+					node = next;
+					continue;
+				}
+				node = node->parent;
+			}
+
+			AZ(node);
+			VSLdbgv(req, "bytes_unpend: reached root");
+			tree->front = tree->root;
 			return;
 		}
 
@@ -1399,6 +1394,7 @@ bytes_unpend_worklist(struct req *req, struct bytes_tree *tree,
 
 		assert_node(node, CHK_PEND);
 
+		/* XXX only enter ST_OPEN if we own it */
 		if (node->type == T_NEXUS) {
 			node = VSTAILQ_FIRST(&node->nexus.children);
 			continue;