Add req.filters and bereq.filters

bin/varnishd/cache/cache.h

@@ -394,8 +394,11 @@ struct busyobj {
 	struct sess		*sp;
 	struct worker		*wrk;
 
+	/* beresp.body */
 	struct vfp_ctx		*vfc;
 	const char		*vfp_filter_list;
+	/* bereq.body */
+	const char		*vdp_filter_list;
 
 	struct ws		ws[1];
 	uintptr_t		ws_bo;
@@ -537,9 +540,11 @@ struct req {
 	struct objcore		*objcore;
 	struct objcore		*stale_oc;
 
-	/* Deliver pipeline */
+	/* resp.body */
 	struct vdp_ctx		*vdc;
 	const char		*vdp_filter_list;
+	/* req.body */
+	const char		*vfp_filter_list;
 
 	/* Transaction VSL buffer */
 	struct vsl_log		vsl[1];

bin/varnishd/cache/cache_deliver_proc.c

@@ -249,8 +249,8 @@ VDP_Close(struct vdp_ctx *vdc, struct objcore *oc, struct boc *boc)
 
 /*--------------------------------------------------------------------*/
 
-static int v_matchproto_(objiterate_f)
-vdp_objiterate(void *priv, unsigned flush, const void *ptr, ssize_t len)
+int v_matchproto_(objiterate_f)
+VDP_ObjIterate(void *priv, unsigned flush, const void *ptr, ssize_t len)
 {
 	enum vdp_action act;
 
@@ -278,7 +278,7 @@ VDP_DeliverObj(struct vdp_ctx *vdc, struct objcore *oc)
 	vdc->hp = NULL;
 	vdc->clen = NULL;
 	final = oc->flags & OC_F_TRANSIENT ? 1 : 0;
-	r = ObjIterate(vdc->wrk, oc, vdc, vdp_objiterate, final);
+	r = ObjIterate(vdc->wrk, oc, vdc, VDP_ObjIterate, final);
 	if (r < 0)
 		return (r);
 	return (0);

bin/varnishd/cache/cache_panic.c

@@ -377,6 +377,10 @@ pan_busyobj(struct vsb *vsb, const struct busyobj *bo)
 		VSB_printf(vsb, "vfp_filter_list = \"%s\",\n",
 		    bo->vfp_filter_list);
 	}
+	if (bo->vdp_filter_list != NULL) {
+		VSB_printf(vsb, "vdp_filter_list = \"%s\",\n",
+		    bo->vdp_filter_list);
+	}
 
 	WS_Panic(vsb, bo->ws);
 	VSB_printf(vsb, "ws_bo = %p,\n", (void *)bo->ws_bo);
@@ -464,6 +468,15 @@ pan_req(struct vsb *vsb, const struct req *req)
 	else
 		VSB_printf(vsb, "step = %s\n", req->req_step->name);
 
+	if (req->vfp_filter_list != NULL) {
+		VSB_printf(vsb, "vfp_filter_list = \"%s\",\n",
+		    req->vfp_filter_list);
+	}
+	if (req->vdp_filter_list != NULL) {
+		VSB_printf(vsb, "vdp_filter_list = \"%s\",\n",
+		    req->vdp_filter_list);
+	}
+
 	VSB_printf(vsb, "req_body = %s,\n",
 	    req->req_body_status ? req->req_body_status->name : "NULL");
 

bin/varnishd/cache/cache_req.c

@@ -252,6 +252,7 @@ Req_Rollback(VRT_CTX)
 		VCL_TaskEnter(req->top->privs);
 	HTTP_Clone(req->http, req->http0);
 	req->vdp_filter_list = NULL;
+	req->vfp_filter_list = NULL;
 	req->vcf = NULL;
 	if (WS_Overflowed(req->ws))
 		req->wrk->stats->ws_client_overflow++;

bin/varnishd/cache/cache_req_body.c

@@ -91,6 +91,16 @@ vrb_pull(struct req *req, ssize_t maxsize, objiterate_f *func, void *priv)
 	INIT_OBJ(ctx, VRT_CTX_MAGIC);
 	VCL_Req2Ctx(ctx, req);
 
+	if (req->vfp_filter_list != NULL &&
+	    VCL_StackVFP(vfc, req->vcl, req->vfp_filter_list)) {
+		(void)VFP_Error(vfc, "req.body filters failed");
+		req->req_body_status = BS_ERROR;
+		HSH_DerefBoc(req->wrk, req->body_oc);
+		AZ(HSH_DerefObjCore(req->wrk, &req->body_oc, 0));
+		return (-1);
+	}
+
+
 	if (VFP_Open(ctx, vfc) < 0) {
 		req->req_body_status = BS_ERROR;
 		HSH_DerefBoc(req->wrk, req->body_oc);

bin/varnishd/cache/cache_req_fsm.c

@@ -909,6 +909,8 @@ cnt_recv_prep(struct req *req, const char *ci)
 	req->is_hitpass = 0;
 	req->err_code = 0;
 	req->err_reason = NULL;
+
+	req->vfp_filter_list = NULL;
 }
 
 /*--------------------------------------------------------------------

bin/varnishd/cache/cache_varnishd.h

@@ -194,6 +194,7 @@ uint64_t VDP_Close(struct vdp_ctx *, struct objcore *, struct boc *);
 void VDP_Panic(struct vsb *vsb, const struct vdp_ctx *vdc);
 int VDP_Push(VRT_CTX, struct vdp_ctx *, struct ws *, const struct vdp *,
     void *priv);
+int VDP_ObjIterate(void *priv, unsigned flush, const void *ptr, ssize_t len);
 int VDP_DeliverObj(struct vdp_ctx *vdc, struct objcore *oc);
 extern const struct vdp VDP_gunzip;
 extern const struct vdp VDP_esi;

bin/varnishd/cache/cache_vrt_filter.c

@@ -400,6 +400,14 @@ VBF_Get_Filter_List(struct busyobj *bo)
 	return (filter_on_ws(bo->ws, vbf_default_filter_list, bo));
 }
 
+static const char *
+bereq_Empty_Filter(struct busyobj *bo)
+{
+
+	(void)bo;
+	return ("");
+}
+
 /*--------------------------------------------------------------------
  */
 
@@ -433,9 +441,25 @@ resp_Get_Filter_List(struct req *req)
 	return (filter_on_ws(req->ws, resp_default_filter_list, req));
 }
 
+static const char *
+req_Empty_Filter(struct req *req)
+{
+
+	(void)req;
+	return ("");
+}
+
+/*--------------------------------------------------------------------*/
+static int
+req_filter_can(struct req *req) {
+	CHECK_OBJ_NOTNULL(req, REQ_MAGIC);
+
+	return (req->req_body_status->avail == 1);
+}
+
 /*--------------------------------------------------------------------*/
 
-#define FILTER_VAR(vcl, in, func, fld)					\
+#define FILTER_VAR(vcl, in, func, cond, fld)				\
 	VCL_STRING							\
 	VRT_r_##vcl##_filters(VRT_CTX)					\
 	{								\
@@ -453,12 +477,18 @@ resp_Get_Filter_List(struct req *req)
 									\
 		(void)str;						\
 		CHECK_OBJ_NOTNULL(ctx, VRT_CTX_MAGIC);			\
+		if (! (cond)) {						\
+			VRT_fail(ctx, #vcl ".filters not settable");	\
+			return;						\
+		}							\
 		b = VRT_StrandsWS(ctx->in->ws, str, s);			\
 		if (b == NULL)						\
 			WS_MarkOverflow(ctx->in->ws);			\
 		else							\
 			ctx->in->fld = b;				\
 	}
 
-FILTER_VAR(beresp, bo, VBF_Get_Filter_List, vfp_filter_list)
-FILTER_VAR(resp, req, resp_Get_Filter_List, vdp_filter_list)
+FILTER_VAR(bereq, bo, bereq_Empty_Filter, 1, vdp_filter_list)
+FILTER_VAR(beresp, bo, VBF_Get_Filter_List, 1, vfp_filter_list)
+FILTER_VAR(req, req, req_Empty_Filter, req_filter_can(ctx->req), vfp_filter_list)
+FILTER_VAR(resp, req, resp_Get_Filter_List, 1, vdp_filter_list)

bin/varnishd/http1/cache_http1_fetch.c

@@ -42,25 +42,14 @@
 
 #include "cache_http1.h"
 
-/*--------------------------------------------------------------------
- * Pass the request body to the backend
- */
-
-static int v_matchproto_(objiterate_f)
-vbf_iter_req_body(void *priv, unsigned flush, const void *ptr, ssize_t l)
+static int
+v1f_stackv1l(struct vdp_ctx *vdc, struct busyobj *bo)
 {
-	struct busyobj *bo;
+	struct vrt_ctx ctx[1];
 
-	CAST_OBJ_NOTNULL(bo, priv, BUSYOBJ_MAGIC);
-
-	if (l > 0) {
-		if (DO_DEBUG(DBG_SLOW_BEREQ))
-			VTIM_sleep(1.0);
-		(void)V1L_Write(bo->wrk, ptr, l);
-		if (flush && V1L_Flush(bo->wrk) != SC_NULL)
-			return (-1);
-	}
-	return (0);
+	INIT_OBJ(ctx, VRT_CTX_MAGIC);
+	VCL_Bo2Ctx(ctx, bo);
+	return (VDP_Push(ctx, vdc, ctx->ws, VDP_v1l, NULL));
 }
 
 /*--------------------------------------------------------------------
@@ -80,7 +69,8 @@ V1F_SendReq(struct worker *wrk, struct busyobj *bo, uint64_t *ctr_hdrbytes,
 	ssize_t i;
 	uint64_t bytes, hdrbytes;
 	struct http_conn *htc;
-	int do_chunked = 0;
+	struct vdp_ctx vdc[1];
+	intmax_t cl;
 
 	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
 	CHECK_OBJ_NOTNULL(bo, BUSYOBJ_MAGIC);
@@ -93,13 +83,41 @@ V1F_SendReq(struct worker *wrk, struct busyobj *bo, uint64_t *ctr_hdrbytes,
 	assert(*htc->rfd > 0);
 	hp = bo->bereq;
 
-	if (bo->req != NULL && !bo->req->req_body_status->length_known) {
-		http_PrintfHeader(hp, "Transfer-Encoding: chunked");
-		do_chunked = 1;
+	if (bo->bereq_body != NULL)
+		cl = ObjGetLen(wrk, bo->bereq_body);
+	else if (bo->req != NULL && !bo->req->req_body_status->length_known)
+		cl = -1;
+	else if (bo->req != NULL) {
+		cl = http_GetContentLength(bo->req->http);
+		if (cl < 0)
+			cl = 0;
+	}
+	else
+		cl = 0;
+
+	VDP_Init(vdc, wrk, bo->vsl, NULL, bo, &cl);
+	if (bo->vdp_filter_list != NULL &&
+	    VCL_StackVDP(vdc, bo->vcl, bo->vdp_filter_list, NULL, bo)) {
+		VSLb(bo->vsl, SLT_FetchError, "Failure to push processors");
+		VSLb_ts_busyobj(bo, "Bereq", W_TIM_real(wrk));
+		htc->doclose = SC_OVERLOAD;
+		return (-1);
 	}
+	if (v1f_stackv1l(vdc, bo)) {
+		VSLb(bo->vsl, SLT_FetchError, "Failure to push V1L");
+		VSLb_ts_busyobj(bo, "Bereq", W_TIM_real(wrk));
+		(void) VDP_Close(vdc, NULL, NULL);
+		htc->doclose = SC_OVERLOAD;
+		return (-1);
+	}
+
+	assert(cl >= -1);
+	if (cl < 0)
+		http_PrintfHeader(hp, "Transfer-Encoding: chunked");
 
 	VTCP_blocking(*htc->rfd);	/* XXX: we should timeout instead */
 	/* XXX: need a send_timeout for the backend side */
+	// XXX cache_param->http1_iovs ?
 	V1L_Open(wrk, wrk->aws, htc->rfd, bo->vsl, nan(""), 0);
 	hdrbytes = HTTP1_Write(wrk, hp, HTTP1_Req);
 
@@ -108,16 +126,14 @@ V1F_SendReq(struct worker *wrk, struct busyobj *bo, uint64_t *ctr_hdrbytes,
 
 	if (bo->bereq_body != NULL) {
 		AZ(bo->req);
-		AZ(do_chunked);
+		assert(cl >= 0);
 		(void)ObjIterate(bo->wrk, bo->bereq_body,
-		    bo, vbf_iter_req_body, 0);
+		    vdc, VDP_ObjIterate, 0);
 	} else if (bo->req != NULL &&
 	    bo->req->req_body_status != BS_NONE) {
-		if (DO_DEBUG(DBG_FLUSH_HEAD))
-			(void)V1L_Flush(wrk);
-		if (do_chunked)
+		if (cl < 0)
 			V1L_Chunked(wrk);
-		i = VRB_Iterate(wrk, bo->vsl, bo->req, vbf_iter_req_body, bo);
+		i = VRB_Iterate(wrk, bo->vsl, bo->req, VDP_ObjIterate, vdc);
 
 		if (bo->req->req_body_status != BS_CACHED)
 			bo->no_retry = "req.body not cached";
@@ -138,20 +154,16 @@ V1F_SendReq(struct worker *wrk, struct busyobj *bo, uint64_t *ctr_hdrbytes,
 			    errno, VAS_errtxt(errno));
 			bo->req->doclose = SC_RX_BODY;
 		}
-		if (do_chunked)
+		if (cl < 0)
 			V1L_EndChunk(wrk);
 	}
 
 	sc = V1L_Close(wrk, &bytes);
 	CHECK_OBJ_NOTNULL(sc, STREAM_CLOSE_MAGIC);
 
 	/* Bytes accounting */
-	if (bytes < hdrbytes)
-		*ctr_hdrbytes += bytes;
-	else {
-		*ctr_hdrbytes += hdrbytes;
-		*ctr_bodybytes += bytes - hdrbytes;
-	}
+	*ctr_hdrbytes += hdrbytes;
+	*ctr_bodybytes += VDP_Close(vdc, NULL, NULL);
 
 	if (sc == SC_NULL && i < 0)
 		sc = SC_TX_ERROR;

bin/varnishtest/tests/m00048.vtc

@@ -66,7 +66,7 @@ varnish v1 -vcl+backend {
 
 	sub vcl_deliver {
 		if (req.http.Rot13) {
-			set resp.filters = "rot13 debug.pedantic";
+			set resp.filters = "debug.slow rot13 debug.pedantic";
 		}
 	}
 }
@@ -90,7 +90,7 @@ varnish v1 -vcl {
 	sub vcl_synth {
 		set resp.body = "Ponto Facto, Caesar Transit!";
 		if (req.http.Rot13) {
-			set resp.filters += "rot13 debug.chunked debug.pedantic";
+			set resp.filters += "debug.slow rot13 debug.chunked debug.pedantic";
 		}
 		return (deliver);
 	}