Compatibility with "Try to send VDP_END with the last bytes"

Ref varnish-cache c2d45aefb563bb105f524b67b84d20649ecb85a2

src/foreign/from_cache_esi_deliver.c

@@ -176,13 +176,6 @@ vped_gzgz_init(struct vdp_ctx *vdc, void **priv, struct objcore *oc)
 	return (0);
 }
 
-/*
- * we removed a NULL VDP_FLUSH from _fini because in our code it goes to the
- * _to_parent vdp on the same request, rather than one request up as in the
- * original code.
- *
- * thus we need to flush at the end of each call
- */
 static int v_matchproto_(vdp_bytes_f)
 vped_gzgz_bytes(struct vdp_ctx *vdx, enum vdp_action act, void **priv,
     const void *ptr, ssize_t len)
@@ -192,6 +185,9 @@ vped_gzgz_bytes(struct vdp_ctx *vdx, enum vdp_action act, void **priv,
 	ssize_t dl;
 	ssize_t l;
 
+	// XXX would like to know which is the last
+	// segment sent to push FLUSH/END with it
+
 	CAST_OBJ_NOTNULL(foo, *priv, VPED_GZGZ_PRIV_MAGIC);
 	pp = ptr;
 	if (len > 0) {
@@ -212,7 +208,7 @@ vped_gzgz_bytes(struct vdp_ctx *vdx, enum vdp_action act, void **priv,
 		if (dl > 0) {
 			if (dl > len)
 				dl = len;
-			if (VDP_bytes(vdx, act, pp, dl))
+			if (VDP_bytes(vdx, VDP_NULL, pp, dl))
 				return(-1);
 			foo->ll += dl;
 			len -= dl;
@@ -235,7 +231,7 @@ vped_gzgz_bytes(struct vdp_ctx *vdx, enum vdp_action act, void **priv,
 		if (dl > 0) {
 			if (dl > len)
 				dl = len;
-			if (VDP_bytes(vdx, act, pp, dl))
+			if (VDP_bytes(vdx, VDP_NULL, pp, dl))
 				return (-1);
 			foo->ll += dl;
 			len -= dl;
@@ -318,7 +314,9 @@ vped_gzgz_bytes(struct vdp_ctx *vdx, enum vdp_action act, void **priv,
 		}
 	}
 	assert(len == 0);
-	return (VDP_bytes(vdx, VDP_FLUSH, NULL, 0));
+	if (act != VDP_END)
+		act = VDP_FLUSH;
+	return (VDP_bytes(vdx, act, NULL, 0));
 }
 
 static int v_matchproto_(vdp_fini_f)

src/node.c

@@ -668,6 +668,22 @@ fini_data(struct vdp_ctx *vdx, struct node *node)
  * (unlocked)
  */
 
+static int
+node_bytes(struct vdp_ctx *vdx, struct bytes_tree *tree,
+    const struct node *node, enum vdp_action act, const void *ptr, ssize_t sz)
+{
+	if (node->is_end) {
+		AZ(tree->end_sent);
+		assert(tree->end == node);
+		tree->end_sent = 1;
+		act = VDP_END;
+	} else if (act == VDP_END) {
+		act = VDP_FLUSH;
+	}
+
+	return (VDP_bytes(vdx, act, ptr, sz));
+}
+
 static int
 push_final(struct vdp_ctx *vdx, struct bytes_tree *tree,
     struct node *node, const struct node *next)
@@ -690,6 +706,7 @@ push_final(struct vdp_ctx *vdx, struct bytes_tree *tree,
 
 	AZ(pthread_mutex_unlock(&node->final.fi_mtx));
 
+	WRONG("ACT");
 	assert(node->final.fi_state == FI_DONE);
 	return (tree->retval);
 }
@@ -768,7 +785,6 @@ push_crc(struct vdp_ctx *vdx, struct bytes_tree *tree,
 	assert(node->type == T_CRC);
 
 	(void) vdx;
-	(void) tree;
 
 	nex = node->parent;
 	CHECK_OBJ_NOTNULL(nex, NODE_MAGIC);
@@ -797,8 +813,9 @@ push_crc(struct vdp_ctx *vdx, struct bytes_tree *tree,
 			parent_gzip->l_crc += nex->nexus.gzip.l_crc;
 		} else {
 			gzip_tailbuf(tailbuf, &nex->nexus.gzip);
-			return (VDP_bytes(nex->nexus.req->vdc,
-			    VDP_FLUSH, tailbuf, sizeof(tailbuf)));
+			return (node_bytes(nex->nexus.req->vdc,
+			    tree, node, VDP_FLUSH,
+			    tailbuf, sizeof(tailbuf)));
 		}
 		break;
 	default:
@@ -818,8 +835,6 @@ push_data(struct vdp_ctx *vdx, struct bytes_tree *tree,
 
 	assert(node->type == T_DATA);
 
-	(void) tree;
-
 	AN(node->parent);
 	assert(node->parent->type == T_NEXUS);
 
@@ -839,7 +854,7 @@ push_data(struct vdp_ctx *vdx, struct bytes_tree *tree,
 		return (0);
 
 	vdx = node->parent->nexus.req->vdc;
-	return (VDP_bytes(vdx, act, p, node->data.len));
+	return (node_bytes(vdx, tree, node, act, p, node->data.len));
 }
 
 /*
@@ -962,6 +977,8 @@ worklist_set_delivered(struct vdp_ctx *vdx, struct bytes_tree *tree,
 		(tree)->front_owner = wrk;				\
 	} while(0)
 
+#define set_end(tree, node) (tree)->end = node;
+
 static void
 worklist_unpend(const struct vdp_ctx *vdx, struct bytes_tree *tree,
     struct node_head *work)
@@ -1049,6 +1066,8 @@ worklist_unpend(const struct vdp_ctx *vdx, struct bytes_tree *tree,
 
 			VSLdbg(vdx, "bytes_unpend: reached root");
 			set_front(tree, tree->root, vdx->wrk);
+			if (tree->end != NULL)
+				tree->end->is_end = 1;
 			return;
 		}
 
@@ -1090,6 +1109,7 @@ worklist_unpend(const struct vdp_ctx *vdx, struct bytes_tree *tree,
 
 		VSTAILQ_INSERT_TAIL(work, node, unpend);
 		set_unpending(tree, node);
+		set_end(tree, node);
 
 		// for more iterations, also accept UNPEND
 		// for now, we use any

src/node.h

@@ -41,6 +41,9 @@ struct bytes_tree {
 	struct node		*front;
 	const struct worker	*front_owner;
 	const struct worker	*unpend_owner;
+	// candidate for last ever node
+	struct node		*end;
+	int			end_sent;
 	int			npending;
 	int			retval;
 };
@@ -184,6 +187,7 @@ struct node {
 	enum n_type		type;
 	enum n_state		state;
 	enum n_alloc		allocator;
+	unsigned		is_end:1;
 	VSTAILQ_ENTRY(node)	sibling;
 	VSTAILQ_ENTRY(node)	unpend;
 	struct node		*parent;

src/vdp_pesi.c

@@ -908,6 +908,9 @@ vdp_pesi_bytes(struct vdp_ctx *vdx, enum vdp_action act, void **priv,
 
 	CHECK_OBJ_NOTNULL(vdx, VDP_CTX_MAGIC);
 
+	if (act == VDP_END)
+		act = VDP_FLUSH;
+
 	AN(priv);
 	CAST_OBJ_NOTNULL(pesi, *priv, PESI_MAGIC);
 	CHECK_OBJ_NOTNULL(pesi->pecx, PECX_MAGIC);
@@ -1125,6 +1128,8 @@ vdp_pesi_bytes(struct vdp_ctx *vdx, enum vdp_action act, void **priv,
 						&tree->tree_lock, 0));
 				tree_deliver(vdx, tree);
 			}
+			if (! tree->end_sent)
+				(void) VDP_bytes(vdx, VDP_END, NULL, 0);
 			Lck_Unlock(&tree->tree_lock);
 
 			return (tree->retval);