Set parent in node_new to insert T_NEXUS nodes only when they are read

In order to properly handle vped_include() failing, we need a way to
insert a node only when it's ready and free it when not.

Thus, we link to the parent in node_new(), but do not insert the
child to the parent's list. As before, this happens in node_insert.

On the way, we also add node_free() to free a node without finalizing
it.

Partly fixes !11

src/node.c

@@ -198,6 +198,25 @@ vmod_workspace_prealloc(VRT_CTX, VCL_BYTES min_free, VCL_INT max_nodes)
 	ws_max_nodes = max_nodes;
 }
 
+void
+node_free(struct node **nodep)
+{
+	struct node *node;
+
+	TAKE_OBJ_NOTNULL(node, nodep, NODE_MAGIC);
+	switch (node->allocator) {
+	case NA_WS:
+		return;
+	case NA_MPL:
+		AN(mempool);
+		MPL_AssertSane(node);
+		MPL_Free(mempool, node);
+		return;
+	default:
+		INCOMPL();
+	}
+}
+
 static struct node *
 node_alloc(struct pesi *pesi)
 {
@@ -226,12 +245,16 @@ node_alloc(struct pesi *pesi)
 }
 
 struct node *
-node_new(struct pesi *pesi, enum n_type type, enum n_state state)
+node_new(struct pesi *pesi, struct node *parent,
+    enum n_type type, enum n_state state)
 {
 	struct node *node;
 
+	CHECK_OBJ_ORNULL(parent, NODE_MAGIC);
+
 	node = node_alloc(pesi);
 	AN(node);
+	node->parent = parent;
 	node->type = type;
 	node->state = state;
 	switch (type) {
@@ -299,17 +322,7 @@ tree_free(struct vdp_ctx *vdx, struct node *node)
 
 	node_fini(vdx, node);
 
-	switch (node->allocator) {
-	case NA_WS:
-		return;
-	case NA_MPL:
-		AN(mempool);
-		MPL_AssertSane(node);
-		MPL_Free(mempool, node);
-		return;
-	default:
-		INCOMPL();
-	}
+	node_free(&node);
 }
 
 /* ============================================================
@@ -317,42 +330,22 @@ tree_free(struct vdp_ctx *vdx, struct node *node)
  */
 
 void
-node_insert(const struct bytes_tree *tree, struct node *parent,
-    struct node *node)
+node_insert(const struct bytes_tree *tree, struct node *node)
 {
+	struct node *parent;
+
 	CHECK_OBJ_NOTNULL(tree, BYTES_TREE_MAGIC);
-	CHECK_OBJ_NOTNULL(parent, NODE_MAGIC);
 	CHECK_OBJ_NOTNULL(node, NODE_MAGIC);
 
-	assert(parent->type == T_NEXUS);
+	parent = node->parent;
+	CHECK_OBJ_NOTNULL(parent, NODE_MAGIC);
 
+	assert(parent->type == T_NEXUS);
 	assert(parent->state == ST_PRIVATE);
-
-	switch (node->type) {
-	case T_NEXUS:
-		assert(node->state == ST_PRIVATE);
-		assert(node->nexus.gzip.magic == NEXUS_GZIP_MAGIC);
-		break;
-	case T_CRC:
-	case T_DATA:
-		assert(node->state == ST_DATA);
-		break;
-	case T_SUBREQ:
-	default:
-		/* cannot insert T_SUBREQ yet */
-		INCOMPL();
-	}
-
-	AZ(node->parent);
-
-	node->parent = parent;
-
-	VSTAILQ_INSERT_TAIL(&parent->nexus.children, node, sibling);
-
+	parent->nexus.npending_private++;
 	assert(tree->npending >= 0);
 
-	assert(parent->state == ST_PRIVATE);
-	parent->nexus.npending_private++;
+	VSTAILQ_INSERT_TAIL(&parent->nexus.children, node, sibling);
 
 	VSLdbgv(tree->root->nexus.req, "node_insert: node=%p parent=%p "
 	    "parent->state=%d parent->npending_private=%d "

src/node.h

@@ -241,8 +241,10 @@ void node_fill_nodestock(struct ws *, struct node_head *);
 //--------------
 
 struct pesi;
-struct node *node_new(struct pesi *pesi, enum n_type type, enum n_state state);
-void node_insert(const struct bytes_tree *, struct node *, struct node *);
+struct node *node_new(struct pesi *pesi, struct node *parent,
+    enum n_type type, enum n_state state);
+void node_free(struct node **nodep);
+void node_insert(const struct bytes_tree *, struct node *);
 void set_closed(struct bytes_tree *, struct node *);
 
 //--------------

src/tests/r03865.vtc

@@ -12,7 +12,8 @@ varnish v1 -cliok "param.set feature +esi_disable_xml_check"
 varnish v1 -cliok "param.set feature +esi_include_onerror"
 varnish v1 -vcl+backend {
 	import ${vmod_pesi};
-
+	import ${vmod_pesi_debug};
+	include "debug.inc.vcl";
 	sub vcl_backend_fetch {
 		if (bereq.url == "/fail") {
 			return (error(604));
@@ -43,6 +44,20 @@ client c1 {
 	expect resp.body == "before "
 } -run
 
+varnish v1 -cliok "param.set max_esi_depth 0"
+
+client c1 {
+	txreq -url "/abort"
+	non_fatal
+	rxresphdrs
+	expect resp.status == 200
+	rxchunk
+	expect_close
+	expect resp.body ~ "^(before )?$"
+} -run
+
+varnish v1 -cliok "param.set max_esi_depth 1"
+
 varnish v1 -vsl_catchup
 
 server s1 -wait
@@ -60,3 +75,12 @@ client c1 {
 	rxresp
 	expect resp.body == "before FOOBAR after"
 } -run
+
+varnish v1 -cliok "param.set max_esi_depth 0"
+
+client c1 {
+	fatal
+	txreq -url "/continue"
+	rxresp
+	expect resp.body == "before  after"
+} -run

src/vdp_pesi.c

@@ -283,7 +283,7 @@ vped_task(struct worker *wrk, void *priv)
 
 static int
 vped_include(struct req *preq, const char *src, const char *host,
-	     struct pesi *pesi, struct node *node)
+	     struct pesi *pesi, struct node *node, int gzip)
 {
 	struct worker *wrk;
 	struct sess *sp;
@@ -355,11 +355,8 @@ vped_include(struct req *preq, const char *src, const char *host,
 	/* Don't allow Range */
 	http_Unset(req->http, H_Range);
 
-	CHECK_OBJ_NOTNULL(node->parent, NODE_MAGIC);
-	assert(node->parent->type == T_NEXUS);
-
 	/* Set Accept-Encoding according to what we want */
-	if (node->parent->nexus.gzip.is)
+	if (gzip)
 		http_ForceHeader(req->http, H_Accept_Encoding, "gzip");
 	else
 		http_Unset(req->http, H_Accept_Encoding);
@@ -500,7 +497,7 @@ pesi_buf_bytes(struct vdp_ctx *vdx, enum vdp_action act, void **priv,
 	CHECK_OBJ_NOTNULL(req, REQ_MAGIC);
 
 	VSLdbg(vdx, "bytes_add: adding data to node");
-	node = node_new(pesi, T_DATA, ST_DATA);
+	node = node_new(pesi, parent, T_DATA, ST_DATA);
 	CHECK_OBJ_NOTNULL(node, NODE_MAGIC);
 	node->data.len = len;
 	node->data.act = act;
@@ -556,7 +553,7 @@ pesi_buf_bytes(struct vdp_ctx *vdx, enum vdp_action act, void **priv,
 		memcpy(p, ptr, len);
 	}
 
-	node_insert(tree, parent, node);
+	node_insert(tree, node);
 
 	VSLdbgv(vdx, "bytes_add to %s parent: exit",
 	    parent->state == ST_PRIVATE ? "private" : "open");
@@ -636,7 +633,7 @@ root_node_new(struct pesi *pesi, struct req *req)
 {
 	struct node *root_node;
 
-	root_node = node_new(pesi, T_NEXUS, ST_PRIVATE);
+	root_node = node_new(pesi, NULL, T_NEXUS, ST_PRIVATE);
 	CHECK_OBJ_NOTNULL(root_node, NODE_MAGIC);
 	root_node->nexus.req = req;
 	root_node->nexus.npending_private = 1;
@@ -917,10 +914,11 @@ vdp_pesi_bytes(struct vdp_ctx *vdx, enum vdp_action act, void **priv,
 			if (*pecx->p == VEC_GZ) {
 				if (parent_gzip == NULL) {
 					AZ(child);
-					child = node_new(pesi, T_CRC, ST_DATA);
+					child = node_new(pesi, node,
+					    T_CRC, ST_DATA);
 					CHECK_OBJ_NOTNULL(child, NODE_MAGIC);
 					child->crc.ctype = GZIP_HDR;
-					node_insert(tree, node, child);
+					node_insert(tree, child);
 					child = NULL;
 				}
 				node->nexus.gzip.is = 1;
@@ -954,12 +952,13 @@ vdp_pesi_bytes(struct vdp_ctx *vdx, enum vdp_action act, void **priv,
 						return (-1);
 
 					AZ(child);
-					child = node_new(pesi, T_CRC, ST_DATA);
+					child = node_new(pesi, node,
+					    T_CRC, ST_DATA);
 					CHECK_OBJ_NOTNULL(child, NODE_MAGIC);
 					child->crc.ctype = UPDATE;
 					child->crc.icrc = vbe32dec(pecx->p);
 					child->crc.l_icrc = l;
-					node_insert(tree, node, child);
+					node_insert(tree, child);
 					child = NULL;
 
 					pecx->p += 4;
@@ -989,22 +988,35 @@ vdp_pesi_bytes(struct vdp_ctx *vdx, enum vdp_action act, void **priv,
 				Debug("INCL [%s][%s] BEGIN\n", q, pecx->p);
 
 				AZ(child);
-				child = node_new(pesi, T_NEXUS, ST_PRIVATE);
+				child = node_new(pesi, node,
+				    T_NEXUS, ST_PRIVATE);
 				CHECK_OBJ_NOTNULL(child, NODE_MAGIC);
 
-				node_insert(tree, node, child);
-
 				pesi->pecx->incl_cont = incl_cont;
 				VSLdbgv(vdx, "vped_vdp: call vped_include "
 				       "incl_cont=%d", incl_cont);
 				parallel =
 					vped_include(req, (const char*)q,
 						     (const char*)pecx->p,
-						     pesi, child);
-				child = NULL;
+						     pesi, child,
+						     node->nexus.gzip.is
+						);
 				VSLdbgv(vdx, "vped_vdp: vped_include()=%d",
 					parallel);
-				assert(parallel >= 0);
+				if (parallel >= 0) {
+					node_insert(tree, child);
+					child = NULL;
+				}
+				else if (! incl_cont && tree->retval == 0) {
+					Lck_Lock(&tree->tree_lock);
+					if (tree->retval == 0)
+						tree->retval = -1;
+					Lck_Unlock(&tree->tree_lock);
+					node_free(&child);
+				} else
+					node_free(&child);
+
+				AZ(child);
 				Debug("INCL [%s][%s] END\n", q, pecx->p);
 				pecx->p = r + 1;
 				break;
@@ -1018,10 +1030,10 @@ vdp_pesi_bytes(struct vdp_ctx *vdx, enum vdp_action act, void **priv,
 		case 2:
 			if (node->nexus.gzip.is) {
 				AZ(child);
-				child = node_new(pesi, T_CRC, ST_DATA);
+				child = node_new(pesi, node, T_CRC, ST_DATA);
 				CHECK_OBJ_NOTNULL(child, NODE_MAGIC);
 				child->crc.ctype = FINAL;
-				node_insert(tree, node, child);
+				node_insert(tree, child);
 				child = NULL;
 			}
 			assert(pecx->p >= pecx->e);