add (sub)tree pruning

once a subtree is completely delivered, we do still need the T_NEXUS at
the head of the (sub)tree for integrity (linkage to its siblings), but
anything below can be completely fini'd and freed.

src/node.c

@@ -102,6 +102,7 @@ node_fini(struct req *req, struct node *node)
 		/* root req owned by varnish-cache */
 		if (node->parent == NULL)
 			break;
+		assert(VSTAILQ_EMPTY(&node->nexus.children));
 		req_fini(&node->nexus.req, req->wrk);
 		break;
 	default:
@@ -109,31 +110,30 @@ node_fini(struct req *req, struct node *node)
 	}
 }
 
-/* recursively fini all nodes */
 void
-tree_fini(struct req *req, struct node *node)
+tree_prune(struct req *req, struct node *node)
 {
 	struct node *child;
 
-	if (node->type == T_NEXUS)
-		VSTAILQ_FOREACH(child, &node->nexus.children, sibling)
-			tree_fini(req, child);
+	assert(node->type == T_NEXUS);
 
-	node_fini(req, node);
+	while((child = VSTAILQ_FIRST(&node->nexus.children)) != NULL) {
+		VSTAILQ_REMOVE_HEAD(&node->nexus.children, sibling);
+		tree_free(req, child);
+	}
+
+	node->state = ST_PRUNED;
 }
 
 /* recursively free all nodes == free the tree */
 void
 tree_free(struct req *req, struct node *node)
 {
-	struct node *child, *tmp;
-
 	AN(mempool);
 	MPL_AssertSane(node);
 
 	if (node->type == T_NEXUS)
-		VSTAILQ_FOREACH_SAFE(child, &node->nexus.children, sibling, tmp)
-			tree_free(req, child);
+		tree_prune(req, node);
 
 	node_fini(req, node);
 	MPL_Free(mempool, node);
@@ -729,44 +729,42 @@ worklist_push(struct req *req, struct bytes_tree *tree,
 	 */
 
 	VSTAILQ_FOREACH(node, work, unpend) {
+		assert(node->type != T_NEXUS);
 		node_fini(req, node);
-
-		if (VSTAILQ_NEXT(node, sibling) != NULL)
-			continue;
-
-		parent = node->parent;
-		/* don't touch the root */
-		if (parent->parent == NULL)
-			continue;
-
-		if (parent->state < ST_CLOSED)
-			continue;
-
-		/*
-		 * fini the now unneeded parent ASAP
-		 *
-		 * this is separate from set_delived() because
-		 * - set_delivered() happens under the lock
-		 * - set_delivered() can also be called from
-		 *   within the parent's FSM
-		 */
-		node_fini(req, parent);
 	}
 
 	return (retval);
 }
 
 static void
-worklist_set_delivered(struct bytes_tree *tree,
+worklist_set_delivered(struct req *req, struct bytes_tree *tree,
     struct node_head *work)
 {
-	struct node *node;
+	struct node *node, *parent;
 
 	CHECK_OBJ_NOTNULL(tree, BYTES_TREE_MAGIC);
 	Lck_AssertHeld(&tree->tree_lock);
 
-	VSTAILQ_FOREACH(node, work, unpend)
+	VSTAILQ_FOREACH(node, work, unpend) {
 		set_delivered(tree, node);
+
+		if (VSTAILQ_NEXT(node, sibling) != NULL)
+			continue;
+
+		parent = node->parent;
+
+		if (parent->state < ST_CLOSED)
+			continue;
+
+		assert(parent->state == ST_DELIVERED);
+		parent->state = ST_PRUNED;
+		Lck_Unlock(&tree->tree_lock);
+
+		tree_prune(req, parent);
+		node_fini(req, parent);
+
+		Lck_Lock(&tree->tree_lock);
+	}
 }
 
 /*
@@ -802,7 +800,8 @@ worklist_unpend(struct req *req, struct bytes_tree *tree,
 	CHECK_OBJ_NOTNULL(tree->front, NODE_MAGIC);
 	CHECK_OBJ_NOTNULL(tree->root, NODE_MAGIC);
 
-	if (tree->root->state == ST_DELIVERED) {
+	if (tree->root->state == ST_DELIVERED ||
+	    tree->root->state == ST_PRUNED) {
 		VSLdbg(req, "bytes_unpend: whole tree is delivered");
 		assert_node(tree->root, CHK_DELI);
 		return;
@@ -830,7 +829,8 @@ worklist_unpend(struct req *req, struct bytes_tree *tree,
 		assert_node(node, CHK_ANY);
 
 		while (node->state == ST_DELIVERED ||
-		       node->state == ST_UNPENDING) {
+		       node->state == ST_UNPENDING ||
+		       node->state == ST_PRUNED) {
 			assert_node(node, check);
 			VSLdbgv(req, "bytes_unpend: delivered/unpending "
 				"node %p state %u", node, node->state);
@@ -852,6 +852,7 @@ worklist_unpend(struct req *req, struct bytes_tree *tree,
 			/* up and right */
 			node = node->parent;
 			while (node != NULL) {
+				assert(node->state != ST_PRUNED);
 				next = VSTAILQ_NEXT(node, sibling);
 				if (next != NULL) {
 					node = next;
@@ -966,7 +967,7 @@ tree_deliver(struct req *req, struct bytes_tree *tree)
 			tree->retval = retval;
 			break;
 		}
-		worklist_set_delivered(tree, &work);
+		worklist_set_delivered(req, tree, &work);
 	} while (1);
 
 	tree->unpend_owner = NULL;

src/node.h

@@ -89,6 +89,12 @@ enum n_type {
  *               any type
  *
  *		 for T_NEXUS, means "anything below is delivered"
+ *
+ * ST_PRUNED: nodes below have been freed, no attempt must
+ *            be made to access any children of this node
+ *            (similar to ST_PRIVATE)
+ *
+ *            T_NEXUS only
  */
 enum n_state {
 	ST_INVALID = 0,
@@ -97,7 +103,8 @@ enum n_state {
 	ST_OPEN,
 	ST_CLOSED,
 	ST_UNPENDING,
-	ST_DELIVERED
+	ST_DELIVERED,
+	ST_PRUNED
 } __attribute__ ((__packed__));
 
 VSTAILQ_HEAD(node_head, node);
@@ -252,7 +259,7 @@ void set_closed(struct bytes_tree *, struct node *, const struct worker *);
 
 //--------------
 
-void tree_fini(struct req *, struct node *);
+void tree_prune(struct req *, struct node *);
 void tree_free(struct req *, struct node *);
 
 //--------------

src/node_assert.h

@@ -61,8 +61,7 @@ assert_nexus(struct node *node, enum check_state check)
 	VSTAILQ_FOREACH(child, &node->nexus.children, sibling) {
 		assert_node(child, check);
 		if (check == CHK_ORDER &&
-		    child->state != ST_DELIVERED &&
-		    child->state != ST_UNPENDING)
+		    child->state <= ST_CLOSED)
 			check = CHK_PEND;
 	}
 }
@@ -77,11 +76,14 @@ assert_node(struct node *node, enum check_state check)
 	case CHK_ORDER:
 		break;
 	case CHK_PEND:
-		assert(node->state != ST_DELIVERED &&
-		       node->state != ST_UNPENDING);
+		// not ST_UNPENDING ST_DELIVERED ST_PRUNED
+		assert(node->state <= ST_CLOSED);
 		break;
 	case CHK_DELI:
-		assert(node->state == ST_DELIVERED);
+		if (node->state == ST_PRUNED)
+			assert(node->type == T_NEXUS);
+		else
+			assert(node->state == ST_DELIVERED);
 		break;
 	default:
 		INCOMPL();
@@ -89,6 +91,7 @@ assert_node(struct node *node, enum check_state check)
 
 	switch (node->state) {
 	case ST_PRIVATE:
+	case ST_PRUNED:
 		assert(node->type == T_NEXUS);
 		/* hands off all nexus fields */
 		break;

src/state.dot

@@ -6,8 +6,6 @@ digraph bytes_node_state {
 	node_insert -> ST_PRIVATE
 	vdp_pesi_init -> ST_PRIVATE
 
-	tree_free [label="tree_free()"]
-
 	subgraph cluster_T_NEXUS {
 		label="T_NEXUS"
 
@@ -15,6 +13,7 @@ digraph bytes_node_state {
 		ST_PRIVATE -> ST_CLOSED	[label="set_closed()"]
 
 		ST_OPEN -> ST_CLOSED		[label="set_closed()"]
+
 	}
 
 	subgraph cluster_leaf {
@@ -28,7 +27,10 @@ digraph bytes_node_state {
 
 	ST_DATA -> ST_DELIVERED	[label="set_delivered()\nif NULL data"]
 
-	ST_DELIVERED -> tree_free
+	ST_DELIVERED -> ST_PRUNED	[label="  tree_free()"]
 
+	# mutate
 	ST_PRIVATE -> ST_DATA		[label="node_mutate_*()"]
+
+
 }
\ No newline at end of file

src/vdp_pesi.c

@@ -774,7 +774,7 @@ vdp_pesi_fini(struct req *req, void **priv)
 	 *
 	 * we likely need more error handling
 	 */
-	tree_fini(req, bytes_tree->root);
+	tree_prune(req, bytes_tree->root);
 
 	AZ(pesi_tree->task_running);
 	assert(VTAILQ_EMPTY(&pesi_tree->task_head));
@@ -855,7 +855,7 @@ vdp_pesi_bytes(struct req *req, enum vdp_action act, void **priv,
 	 * see e00031.vtc and e00033.vtc
 	 */
 	if (req->esi_level == 0 &&
-	    node->state == ST_DELIVERED &&
+	    node->state >= ST_DELIVERED &&
 	    pecx->state == 99) {
 		return (VDP_bytes(req, act, ptr, len));
 	}
@@ -1036,7 +1036,7 @@ vdp_pesi_bytes(struct req *req, enum vdp_action act, void **priv,
 
 			tree_deliver(req, tree);
 			while (!tree->retval
-			       && (tree->root->state != ST_DELIVERED ||
+			       && (tree->root->state < ST_DELIVERED ||
 				   tree->npending > 0)) {
 				VSLdbgv(req, "vdp_pesi_bytes: waiting - "
 					"state=%d npending=%d",