Put V1P(ipe) processing firmly below VBE, and access it through VDI.

bin/varnishd/cache/cache.h

@@ -731,7 +731,7 @@ void V1D_Deliver(struct req *, struct busyobj *);
 
 /* cache_http1_pipe.c */
 void V1P_Init(void);
-void V1P_Process(struct req *req, struct busyobj *bo);
+void V1P_Process(struct req *req, struct busyobj *bo, int fd);
 
 /* cache_req_body.c */
 int VRB_Ignore(struct req *req);

bin/varnishd/cache/cache_backend.c

@@ -488,6 +488,19 @@ VRT_fini_vbe(VRT_CTX, struct director *d)
 	d->priv = NULL;
 }
 
+static void
+vbe_dir_http1pipe(const struct director *d, struct req *req, struct busyobj *bo)
+{
+	int i;
+
+	CHECK_OBJ_NOTNULL(req, REQ_MAGIC);
+	CHECK_OBJ_NOTNULL(bo, BUSYOBJ_MAGIC);
+
+	i = vbe_dir_getfd(d, bo);
+	V1P_Process(req, bo, i);
+	vbe_dir_finish(d, bo->wrk, bo);
+}
+
 void
 VRT_init_vbe(VRT_CTX, struct director **bp, int idx,
     const void *priv)
@@ -505,7 +518,7 @@ VRT_init_vbe(VRT_CTX, struct director **bp, int idx,
 	vs->dir.priv = vs;
 	vs->dir.name = "simple";
 	REPLACE(vs->dir.vcl_name, t->vcl_name);
-	vs->dir.gethttp1fd = vbe_dir_getfd;
+	vs->dir.http1pipe = vbe_dir_http1pipe;
 	vs->dir.healthy = vbe_dir_healthy;
 	vs->dir.gethdrs = vbe_dir_gethdrs;
 	vs->dir.getbody = vbe_dir_getbody;

bin/varnishd/cache/cache_director.c

@@ -130,17 +130,18 @@ VDI_Finish(struct worker *wrk, struct busyobj *bo)
 /* Get a connection --------------------------------------------------*/
 
 int
-VDI_GetHttp1Fd(struct worker *wrk, struct busyobj *bo)
+VDI_Http1Pipe(struct req *req, struct busyobj *bo)
 {
 	const struct director *d;
 
-	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
+	CHECK_OBJ_NOTNULL(req, REQ_MAGIC);
 	CHECK_OBJ_NOTNULL(bo, BUSYOBJ_MAGIC);
 
-	d = vdi_resolve(wrk, bo);
-	if (d == NULL || d->gethttp1fd == NULL)
+	d = vdi_resolve(req->wrk, bo);
+	if (d == NULL || d->http1pipe == NULL)
 		return (-1);
-	return (d->gethttp1fd(d, bo));
+	d->http1pipe(d, req, bo);
+	return (0);
 }
 
 /* Check health --------------------------------------------------------

bin/varnishd/cache/cache_director.h

@@ -42,11 +42,13 @@
  * backends to use.
  */
 
-typedef int vdi_gethttp1fd_f(const struct director *, struct busyobj *);
+
 typedef unsigned vdi_healthy_f(const struct director *, const struct busyobj *,
     double *changed);
+
 typedef const struct director *vdi_resolve_f(const struct director *,
     struct worker *, struct busyobj *);
+
 typedef int vdi_gethdrs_f(const struct director *, struct worker *,
     struct busyobj *);
 typedef int vdi_getbody_f(const struct director *, struct worker *,
@@ -54,12 +56,15 @@ typedef int vdi_getbody_f(const struct director *, struct worker *,
 typedef void vdi_finish_f(const struct director *, struct worker *,
     struct busyobj *);
 
+typedef void vdi_http1pipe_f(const struct director *, struct req *,
+    struct busyobj *);
+
 struct director {
 	unsigned		magic;
 #define DIRECTOR_MAGIC		0x3336351d
 	const char		*name;
 	char			*vcl_name;
-	vdi_gethttp1fd_f	*gethttp1fd;
+	vdi_http1pipe_f		*http1pipe;
 	vdi_healthy_f		*healthy;
 	vdi_resolve_f		*resolve;
 	vdi_gethdrs_f		*gethdrs;
@@ -69,10 +74,14 @@ struct director {
 };
 
 /* cache_director.c */
+
 int VDI_GetHdr(struct worker *wrk, struct busyobj *bo);
 int VDI_GetBody(struct worker *wrk, struct busyobj *bo);
 void VDI_Finish(struct worker *wrk, struct busyobj *bo);
-int VDI_GetHttp1Fd(struct worker *wrk, struct busyobj *);
+
+int VDI_Http1Pipe(struct req *, struct busyobj *);
+
 int VDI_Healthy(const struct director *, const struct busyobj *);
+
 void VBE_Init(void);
 

bin/varnishd/cache/cache_req_fsm.c

@@ -43,6 +43,7 @@
 #include <stdlib.h>
 
 #include "cache.h"
+#include "cache_director.h"
 
 #include "hash/hash_slinger.h"
 #include "vcl.h"
@@ -516,7 +517,8 @@ cnt_pipe(struct worker *wrk, struct req *req)
 		INCOMPL();
 	assert(wrk->handling == VCL_RET_PIPE);
 
-	V1P_Process(req, bo);
+	if (VDI_Http1Pipe(req, bo) < 0)
+		VSLb(bo->vsl, SLT_VCL_Error, "Backend does not support pipe");
 	http_Teardown(bo->bereq);
 	VBO_DerefBusyObj(wrk, &bo);
 	THR_SetBusyobj(NULL);

bin/varnishd/http1/cache_http1_pipe.c

@@ -93,11 +93,11 @@ pipecharge(struct req *req, const struct acct_pipe *a, struct VSC_C_vbe *b)
 }
 
 void
-V1P_Process(struct req *req, struct busyobj *bo)
+V1P_Process(struct req *req, struct busyobj *bo, int fd)
 {
 	struct worker *wrk;
 	struct pollfd fds[2];
-	int i, fd;
+	int i;
 	struct acct_pipe acct_pipe;
 
 	CHECK_OBJ_NOTNULL(req, REQ_MAGIC);
@@ -112,7 +112,6 @@ V1P_Process(struct req *req, struct busyobj *bo)
 	acct_pipe.req = req->acct.req_hdrbytes;
 	req->acct.req_hdrbytes = 0;
 
-	fd = VDI_GetHttp1Fd(wrk, bo);
 	if (fd < 0) {
 		pipecharge(req, &acct_pipe, NULL);
 		SES_Close(req->sp, SC_OVERLOAD);
@@ -172,7 +171,6 @@ V1P_Process(struct req *req, struct busyobj *bo)
 	pipecharge(req, &acct_pipe, bo->htc->vbc->backend->vsc);
 	SES_Close(req->sp, SC_TX_PIPE);
 	bo->doclose = SC_TX_PIPE;
-	VDI_Finish(bo->wrk, bo);
 }
 
 /*--------------------------------------------------------------------*/