Remove ST_OPEN state

Now that we got rid of front unpending, we can also remove all
the complications due to the ST_OPEN state.

Begin with removing that state itself.

src/node.c

@@ -300,15 +300,11 @@ node_insert(struct bytes_tree *tree, struct node *parent,
 
 	assert(parent->type == T_NEXUS);
 
-	if (parent->state == ST_OPEN)
-		Lck_Lock(&tree->tree_lock);
-	else
-		assert(parent->state == ST_PRIVATE);
+	assert(parent->state == ST_PRIVATE);
 
 	switch (node->type) {
 	case T_NEXUS:
-		assert(node->state == ST_PRIVATE ||
-		       node->state == ST_OPEN);
+		assert(node->state == ST_PRIVATE);
 		assert(node->nexus.gzip.magic == NEXUS_GZIP_MAGIC);
 		break;
 	case T_CRC:
@@ -331,19 +327,8 @@ node_insert(struct bytes_tree *tree, struct node *parent,
 
 	assert(tree->npending >= 0);
 
-	if (parent->state == ST_PRIVATE) {
-		parent->nexus.npending_private++;
-	}
-	else {
-		assert(parent->state == ST_OPEN);
-		AZ(parent->nexus.npending_private);
-		tree->npending++;
-
-		if (node->state == ST_DATA)
-			AZ(pthread_cond_signal(&tree->cond));
-
-		Lck_Unlock(&tree->tree_lock);
-	}
+	assert(parent->state == ST_PRIVATE);
+	parent->nexus.npending_private++;
 
 	VSLdbgv(tree->root->nexus.req, "node_insert: node=%p parent=%p "
 	    "parent->state=%d parent->npending_private=%d "
@@ -377,7 +362,7 @@ set_unpending(const struct bytes_tree *tree, struct node *node)
 	if (parent == NULL)
 		return;
 
-	assert(parent->state == ST_OPEN || parent->state == ST_CLOSED);
+	assert(parent->state == ST_CLOSED);
 
 }
 
@@ -422,9 +407,9 @@ set_delivered(struct bytes_tree *tree, struct node *node)
 		return;
 	}
 
-	assert(parent->state == ST_OPEN || parent->state == ST_CLOSED);
+	assert(parent->state == ST_CLOSED);
 
-	if (parent->state == ST_OPEN || VSTAILQ_NEXT(node, sibling) != NULL) {
+	if (VSTAILQ_NEXT(node, sibling) != NULL) {
 		VSL0dbg("set_delivered: stop parent=%p state=%d "
 		    "sibling %p", parent, parent->state,
 		    VSTAILQ_NEXT(node, sibling));
@@ -434,32 +419,7 @@ set_delivered(struct bytes_tree *tree, struct node *node)
 	set_delivered(tree, parent);
 }
 
-/* ST_PRIVATE -> ST_OPEN transition
- *
- * child nodes may get unpended by owner
- * any access locked only
- */
-
-static void
-set_open(struct bytes_tree *tree, struct node *node, const struct worker *wrk)
-{
-	CHECK_OBJ_NOTNULL(tree, BYTES_TREE_MAGIC);
-	CHECK_OBJ_NOTNULL(node, NODE_MAGIC);
-
-	Lck_AssertHeld(&tree->tree_lock);
-
-	VSL0dbg("set_open node=%p wrk=%p", node, wrk);
-
-	assert(node->state == ST_PRIVATE);
-	AZ(node->nexus.owner);
-
-	node->state = ST_OPEN;
-	tree->npending += node->nexus.npending_private;
-	node->nexus.npending_private = 0;
-	node->nexus.owner = wrk;
-}
-
-/* (ST_PRIVATE | ST_OPEN) -> CLOSE transition
+/* ST_PRIVATE -> transition
  *
  * possible states and transitions
  *
@@ -468,7 +428,7 @@ set_open(struct bytes_tree *tree, struct node *node, const struct worker *wrk)
  */
 
 void
-set_closed(struct bytes_tree *tree, struct node *node, const struct worker *wrk)
+set_closed(struct bytes_tree *tree, struct node *node)
 {
 	struct node *child;
 
@@ -476,17 +436,14 @@ set_closed(struct bytes_tree *tree, struct node *node, const struct worker *wrk)
 	CHECK_OBJ_NOTNULL(node, NODE_MAGIC);
 	Lck_AssertHeld(&tree->tree_lock);
 
-	if (node->state == ST_PRIVATE)
-		set_open(tree, node, wrk);
+	VSL0dbg("set_closed node=%p", node);
 
-	VSL0dbg("set_closed node=%p wrk=%p", node, wrk);
-
-	assert(node->state == ST_OPEN);
-	assert(node->nexus.owner == wrk);
+	assert(node->state == ST_PRIVATE);
+	tree->npending += node->nexus.npending_private;
+	node->nexus.npending_private = 0;
 	assert_node(node, CHK_ORDER);
 
 	node->state = ST_CLOSED;
-	node->nexus.owner = NULL;
 
 	/* unpending may wait for this node to close */
 	AZ(pthread_cond_signal(&tree->cond));
@@ -940,14 +897,6 @@ worklist_unpend(const struct vdp_ctx *vdx, struct bytes_tree *tree,
 		return;
 	}
 
-	/*
-	 * if unpending changes between threads, we need to make sure that we
-	 * only descend into ST_OPEN T_NEXUS which we own, so re-start at the
-	 * root
-	 *
-	 * this might seem inefficient, but we skip over any delivered subtrees,
-	 * so we will not walk the whole tree each time
-	 */
 	AN(tree->front);
 
 	if (tree->front_owner == vdx->wrk)
@@ -972,14 +921,6 @@ worklist_unpend(const struct vdp_ctx *vdx, struct bytes_tree *tree,
 				continue;
 			}
 
-			if (node->parent && node->parent->state == ST_OPEN) {
-				VSLdbgv(vdx, "bytes_unpend: return at "
-					"node %p state %u - await sibling",
-					node, node->state);
-				set_front(tree, node, vdx->wrk);
-				return;
-			}
-
 			/* up and right */
 			node = node->parent;
 			while (node != NULL) {
@@ -1012,27 +953,13 @@ worklist_unpend(const struct vdp_ctx *vdx, struct bytes_tree *tree,
 
 		assert_node(node, CHK_PEND);
 
-		/* for ST_OPEN, only descend if we own it */
 		if (node->type == T_NEXUS) {
+			assert(node->state == ST_CLOSED);
 			next = VSTAILQ_FIRST(&node->nexus.children);
-
-			switch (node->state) {
-			case ST_CLOSED:
-				// closed node with no children must not exist
-				AN(next);
-				node = next;
-				continue;
-			case ST_OPEN:
-				set_front(tree, node, vdx->wrk);
-				AN(node->nexus.owner);
-				if (node->nexus.owner == vdx->wrk) {
-					node = next;
-					continue;
-				}
-				return;
-			default:
-				INCOMPL();
-			}
+			// closed node with no children must not exist
+			AN(next);
+			node = next;
+			continue;
 		}
 
 		assert(node->state == ST_DATA);

src/node.h

@@ -72,13 +72,6 @@ enum n_type {
  *
  *             T_NEXUS only
  *
- * ST_OPEN:    may receive pushes creating children
- *             owning thread may run front delivery
- *             any changes below locked only
- *
- *             T_NEXUS only
- *
- *
  * ST_CLOSED:  the request is done, no pushes can occur,
  *             unpending can proceed upwards
  *
@@ -104,7 +97,6 @@ enum n_state {
 	ST_INVALID = 0,
 	ST_DATA,
 	ST_PRIVATE,
-	ST_OPEN,
 	ST_CLOSED,
 	ST_UNPENDING,
 	ST_DELIVERED,
@@ -115,7 +107,6 @@ struct node_nexus {
 	struct node_head	children;
 	struct objcore		*oc;
 	struct req		*req;
-	const struct worker	*owner; // ST_OPEN only
 
 	/* number of nodes pending under this node while state == ST_PRIVATE */
 	int			npending_private;
@@ -252,7 +243,7 @@ void node_fill_nodestock(struct ws *, struct node_head *);
 struct pesi;
 struct node *node_alloc(struct pesi *);
 void node_insert(struct bytes_tree *, struct node *, struct node *);
-void set_closed(struct bytes_tree *, struct node *, const struct worker *);
+void set_closed(struct bytes_tree *, struct node *);
 
 //--------------
 

src/node_assert.h

@@ -110,13 +110,7 @@ assert_node(const struct node *node, enum check_state check)
 	case ST_UNPENDING:
 		assert(node->type != T_NEXUS);
 		break;
-	case ST_OPEN:
-		AN(node->nexus.owner);
-		AZ(node->nexus.npending_private);
-		assert_nexus(node, CHK_ORDER);
-		break;
 	case ST_CLOSED:
-		AZ(node->nexus.owner);
 		AZ(node->nexus.npending_private);
 		assert_nexus(node, CHK_ORDER);
 		break;

src/state.dot

@@ -9,11 +9,7 @@ digraph bytes_node_state {
 	subgraph cluster_T_NEXUS {
 		label="T_NEXUS"
 
-		ST_PRIVATE -> ST_OPEN		[label=" set_open()"]
 		ST_PRIVATE -> ST_CLOSED	[label="set_closed()"]
-
-		ST_OPEN -> ST_CLOSED		[label=" set_closed()"]
-
 	}
 
 	subgraph cluster_leaf {

src/tests/permutations.vtc

@@ -46,6 +46,8 @@ varnish v1 -arg "-a 127.0.0.1:0 -a self=${tmpdir}/v1.sock,mode=777" \
 	import ${vmod_pesi};
 	import ${vmod_pesi_debug};
 	include "debug.inc.vcl";
+	import vtc;
+	import std;
 
         backend self {
             .path = "${tmpdir}/v1.sock";

src/vdp_pesi.c

@@ -216,8 +216,7 @@ vped_task(struct worker *wrk, void *priv)
 		break;
 	case T_NEXUS:
 		assert(node->nexus.req == req);
-		assert(node->state == ST_OPEN ||
-		       node->state == ST_PRIVATE);
+		assert(node->state == ST_PRIVATE);
 
 		if (VSTAILQ_FIRST(&node->nexus.children) != NULL) {
 			/*
@@ -229,7 +228,7 @@ vped_task(struct worker *wrk, void *priv)
 			pesi = NULL;
 
 			Lck_Lock(&pesi_tasks->bytes_tree->tree_lock);
-			set_closed(pesi_tasks->bytes_tree, node, wrk);
+			set_closed(pesi_tasks->bytes_tree, node);
 			Lck_Unlock(&pesi_tasks->bytes_tree->tree_lock);
 			AZ(pesi);
 			task_fini(pesi_tasks, pesi);
@@ -900,7 +899,7 @@ vdp_pesi_bytes(struct vdp_ctx *vdx, enum vdp_action act, void **priv,
 		return (tree->retval);
 	}
 
-	assert(node->state == ST_OPEN || node->state == ST_PRIVATE);
+	assert(node->state == ST_PRIVATE);
 
 	assert(ObjHasAttr(req->wrk, req->objcore, OA_ESIDATA));
 
@@ -1076,7 +1075,7 @@ vdp_pesi_bytes(struct vdp_ctx *vdx, enum vdp_action act, void **priv,
 			 * fini time, the V1D is already closed
 			 */
 			Lck_Lock(&tree->tree_lock);
-			set_closed(tree, node, req->wrk);
+			set_closed(tree, node);
 
 			assert_vdp_next_not(req, &VDP_pesi);
 			vped_close_vdp(req, 1, &VDP_pesi_buf);