handle "final" (pass = OC_F_PRIVATE / _HFM / _HFP) requests differently

geoffs load tests have shown that my previous idea of trying to fool
varnish-cache code in temporarily changing the objcore flags was bad and
leads to all kinds of nasty races.

So so we add yet another method which not only keeps the subreq intact,
but also halts its thread until deliery is ready.

This requires another thread to exist for longer, hopyfully will get us
towards the goals described in d9c36c7e.

src/vdp_pesi.c

@@ -97,7 +97,8 @@ enum n_type {
 	T_INVALID = 0,
 	T_NEXUS,
 	T_DATA,
-	T_SUBREQ
+	T_SUBREQ,
+	T_FINAL	// non-ESI pass / hfm / hfp
 };
 
 /*
@@ -150,6 +151,10 @@ struct node_data {
 	enum vdp_action		act;
 };
 
+/*
+ * we transfer the sub-request, boc and oc to the topreq thread, delivery
+ * happens there
+ */
 struct node_subreq {
 	struct req		*req;
 	struct boc		*boc;
@@ -161,6 +166,22 @@ struct node_subreq {
 	pthread_cond_t		cond;
 };
 
+// XXX integrate with n_state?
+enum fi_state {
+	FI_READY	= 0,
+	FI_GO,			// topreq signalling req to deliver
+	FI_DONE,		// req signalling topreq it is done
+	FI_DESTROYED		// cond/mtx destroyed (fini_final())
+};
+
+/* we block the sub-thread when it's ready for delivery and continue when the
+ * topreqp tells it to */
+struct node_final {
+	enum fi_state		fi_state;
+	pthread_mutex_t	fi_mtx;
+	pthread_cond_t		fi_cond;
+};
+
 struct node {
 	unsigned		magic;
 #define NODE_MAGIC 0xe31edef3
@@ -177,6 +198,7 @@ struct node {
 		struct node_nexus	nexus;	// T_NEXUS
 		struct node_data	data;	// T_DATA
 		struct node_subreq	subreq; // T_SUBREQ
+		struct node_final	final;	// T_FINAL
 	};
 };
 
@@ -241,7 +263,6 @@ struct pesi {
 	unsigned		magic;
 #define PESI_MAGIC 0xa6ba54a0
 	unsigned		bypass:1;
-	unsigned		keep_req:1;
 	unsigned		has_task:1;
 
 	struct pesi_tree	*pesi_tree;
@@ -256,6 +277,8 @@ static void req_fini(struct req *, struct worker *);
 
 // node finalizers
 static void
+fini_final(struct req *req, struct node *node);
+static void
 fini_subreq(struct req *req, struct node *node);
 static void
 fini_data(struct req *req, struct node *node);
@@ -411,6 +434,9 @@ node_free(struct req *req, struct node *node)
 			node_free(req, child);
 
 	switch (node->type) {
+	case T_FINAL:
+		fini_final(req, node);
+		break;
 	case T_SUBREQ:
 		fini_subreq(req, node);
 		break;
@@ -447,10 +473,12 @@ node_insert(struct bytes_tree *tree, struct node *parent,
 		       node->state == ST_OPEN);
 		break;
 	case T_DATA:
-	case T_SUBREQ:
 		assert(node->state == ST_DATA);
 		break;
+	case T_SUBREQ:
+	case T_FINAL:
 	default:
+		/* cannot insert SUBREQ and FINAL yet */
 		INCOMPL();
 	}
 
@@ -583,7 +611,8 @@ set_unpending(struct bytes_tree *tree, struct node *node)
 
 	assert(node->state == ST_DATA);
 	assert(node->type == T_DATA ||
-	       node->type == T_SUBREQ);
+	       node->type == T_SUBREQ ||
+	       node->type == T_FINAL);
 
 	node->state = ST_UNPENDING;
 
@@ -796,14 +825,18 @@ ved_task(struct worker *wrk, void *priv)
 
 	VCL_Rel(&req->vcl);
 
-	if (pesi->keep_req == 0) {
-		VSLdbg(req, "Done PESI keep_req == 0");
-		req_fini(req, wrk);
-		pesi = NULL;
-		pesi_finish(pesi_tree, req->transport_priv);
-	}
-	else {
-		VSLdbg(req, "Done PESI keep_req == 1");
+	switch (node->type) {
+	case T_FINAL:
+		/* XXX WORK AROUND that we can't run VDP_close yet,
+		 * so the pesi_destroy from vdp_pesi_fini() has not run
+		 * otherwise this would be the same as the default case
+		 */
+		assert(pesi == req->transport_priv);
+		req->transport_priv = NULL;
+		pesi_destroy(&pesi);
+		pesi_finish(pesi_tree, pesi);
+		break;
+	case T_SUBREQ:
 		assert (node->type == T_SUBREQ);
 		pesi_finish(pesi_tree, req->transport_priv);
 		// XXX MAKE EFFICIENT
@@ -812,6 +845,11 @@ ved_task(struct worker *wrk, void *priv)
 		node->subreq.done = 1;
 		AZ(pthread_cond_signal(&node->subreq.cond));
 		Lck_Unlock(&pesi_tree->tree->tree_lock);
+		break;
+	default:
+		req_fini(req, wrk);
+		pesi = NULL;
+		pesi_finish(pesi_tree, req->transport_priv);
 	}
 
 	wrk->task.func = NULL;
@@ -1128,7 +1166,7 @@ bytes_unpend_worklist(struct req *req, struct bytes_tree *tree,
 
 		assert(node->state == ST_DATA);
 
-		if (node->type == T_SUBREQ) {
+		if (node->type == T_SUBREQ || node->type == T_FINAL) {
 			VSTAILQ_INSERT_TAIL(work, node, unpend);
 			set_unpending(tree, node);
 			check = CHK_ANY;
@@ -1232,6 +1270,36 @@ subreq_fixup(struct node *node, struct req *req)
  *
  * must be idempotent
  */
+static void
+fini_final(struct req *req, struct node *node)
+{
+	(void) req;
+
+	assert(node->type == T_FINAL);
+
+	if (node->final.fi_state == FI_DESTROYED)
+		return;
+
+	AZ(pthread_mutex_lock(&node->final.fi_mtx));
+
+	if (node->final.fi_state < FI_GO) {
+		node->final.fi_state = FI_GO;
+		AZ(pthread_cond_signal(&node->final.fi_cond));
+	}
+
+	if (node->final.fi_state < FI_DONE)
+		AZ(pthread_cond_wait(&node->final.fi_cond,
+				     &node->final.fi_mtx));
+
+	AZ(pthread_mutex_unlock(&node->final.fi_mtx));
+
+	assert(node->final.fi_state == FI_DONE);
+
+	AZ(pthread_cond_destroy(&node->final.fi_cond));
+	AZ(pthread_mutex_destroy(&node->final.fi_mtx));
+
+	node->final.fi_state = FI_DESTROYED;
+}
 
 static void
 fini_subreq(struct req *req, struct node *node)
@@ -1294,6 +1362,32 @@ fini_data(struct req *req, struct node *node)
  * node delivery (not holding tree lock)
  */
 
+static int
+push_final(struct req *req, struct bytes_tree *tree,
+    struct node *node, const struct node *next)
+{
+	(void) req;
+	(void) tree;
+	(void) next;
+
+	assert(node->type == T_FINAL);
+	assert(node->final.fi_state == FI_READY);
+
+	AZ(pthread_mutex_lock(&node->final.fi_mtx));
+
+	node->final.fi_state = FI_GO;
+	AZ(pthread_cond_signal(&node->final.fi_cond));
+
+	if (node->final.fi_state < FI_DONE)
+		AZ(pthread_cond_wait(&node->final.fi_cond,
+				     &node->final.fi_mtx));
+
+	AZ(pthread_mutex_unlock(&node->final.fi_mtx));
+
+	assert(node->final.fi_state == FI_DONE);
+	return (tree->retval);
+}
+
 static int
 push_subreq(struct req *req, struct bytes_tree *tree,
     struct node *node, const struct node *next)
@@ -1392,6 +1486,9 @@ bytes_push_worklist(struct req *req, struct bytes_tree *tree,
 		assert(node->state == ST_UNPENDING);
 
 		switch (node->type) {
+		case T_FINAL:
+			retval = push_final(req, tree, node, next);
+			break;
 		case T_SUBREQ:
 			retval = push_subreq(req, tree, node, next);
 			break;
@@ -1420,6 +1517,9 @@ bytes_push_worklist(struct req *req, struct bytes_tree *tree,
 
 	VSTAILQ_FOREACH(node, work, unpend) {
 		switch (node->type) {
+		case T_FINAL:
+			fini_final(req, node);
+			break;
 		case T_SUBREQ:
 			fini_subreq(req, node);
 			break;
@@ -2034,9 +2134,10 @@ vdp_pesi_bytes(struct req *req, enum vdp_action act, void **priv,
 	 *
 	 * TODO: take pesi out of the VDPs
 	 */
-	if (node->type == T_SUBREQ) {
-		VSLdbg(req, "ved_vdp: T_SUBREQ");
-		AN(node->subreq.done);
+	if (node->type == T_SUBREQ || node->type == T_FINAL) {
+		VSLdbg(req, "ved_vdp: T_SUBREQ / T_FINAL");
+		if (node->type == T_SUBREQ)
+			AN(node->subreq.done);
 		req->acct.resp_bodybytes += len;
 		return (VDP_bytes(req->topreq, act, ptr, len));
 	}
@@ -2322,8 +2423,57 @@ vped_deliver(struct req *req, struct boc *boc, int wantbody)
 	if (boc == NULL && ObjGetLen(req->wrk, req->objcore) == 0)
 		return;
 
-	/* XXX WIP / STILL HACKY */
-	if (!ObjHasAttr(req->wrk, req->objcore, OA_ESIDATA)) {
+	if ((req->objcore->flags & OC_F_FINAL) != 0 &&
+	    !ObjHasAttr(req->wrk, req->objcore, OA_ESIDATA)) {
+		VSLdbg(req, "vped_deliver: T_FINAL");
+
+		/* XXX WIP / STILL HACKY */
+		XXXAZ(push_vdps_NOesi(req));
+
+		node = pesi->pecx->node;
+
+		/* XXX TODO CHANGE NODE TYPES -- this is a NEXUS atm */
+		assert(node->type == T_NEXUS);
+		assert(node->state == ST_PRIVATE);
+		assert(node->nexus.npending_private == 0);
+
+		/* XXX can reduce locking */
+		Lck_Lock(&tree->tree_lock);
+
+		node->type = T_FINAL;
+		node->state = ST_DATA;
+
+		AZ(pthread_mutex_init(&node->final.fi_mtx, NULL));
+		AZ(pthread_cond_init(&node->final.fi_cond, NULL));
+
+		assert(node->final.fi_state == FI_READY);
+
+		if (node->parent == NULL ||
+		    node->parent->state != ST_PRIVATE)
+			AZ(pthread_cond_signal(&tree->cond));
+		Lck_Unlock(&tree->tree_lock);
+
+		AZ(pthread_mutex_lock(&node->final.fi_mtx));
+		if (node->final.fi_state < FI_GO)
+			AZ(pthread_cond_wait(&node->final.fi_cond,
+					     &node->final.fi_mtx));
+
+		assert(node->final.fi_state == FI_GO);
+
+		(void)VDP_DeliverObj(req);
+
+		// XXX closes topreq VDPs, TODO
+		// VDP_close(req);
+
+		node->final.fi_state = FI_DONE;
+		AZ(pthread_cond_signal(&node->final.fi_cond));
+		AZ(pthread_mutex_unlock(&node->final.fi_mtx));
+
+		return;
+	}
+	else if (!ObjHasAttr(req->wrk, req->objcore, OA_ESIDATA)) {
+		VSLdbg(req, "vped_deliver: T_SUBREQ");
+		/* XXX WIP / STILL HACKY */
 		XXXAZ(push_vdps_NOesi(req));
 
 		node = pesi->pecx->node;
@@ -2361,9 +2511,6 @@ vped_deliver(struct req *req, struct boc *boc, int wantbody)
 
 		AZ(pthread_cond_init(&node->subreq.cond, NULL));
 
-		// XXX rename
-		pesi->keep_req = 1;
-
 		if (node->parent == NULL ||
 		    node->parent->state != ST_PRIVATE)
 			AZ(pthread_cond_signal(&tree->cond));
@@ -2379,6 +2526,8 @@ vped_deliver(struct req *req, struct boc *boc, int wantbody)
 		return;
 	}
 
+	VSLdbg(req, "vped_deliver: ESI");
+
 	/* XXX needed ??? */
 	i = ObjCheckFlag(req->wrk, req->objcore, OF_GZIPED);
 	VSLdbgv(req, "vped_deliver: OF_GZIPED=%d tree->isgzip=%d RES_ESI=%d",