param: New deadline_pipe parameter

bin/varnishd/cache/cache.h

@@ -419,6 +419,7 @@ struct busyobj {
 	vtim_dur		connect_timeout;
 	vtim_dur		first_byte_timeout;
 	vtim_dur		between_bytes_timeout;
+	vtim_dur		task_deadline;
 
 	/* Timers */
 	vtim_real		t_first;	/* First timestamp logged */

bin/varnishd/cache/cache_backend.c

@@ -366,6 +366,7 @@ vbe_dir_http1pipe(VRT_CTX, VCL_BACKEND d)
 	struct backend *bp;
 	struct v1p_acct v1a;
 	struct pfd *pfd;
+	vtim_real deadline;
 
 	CHECK_OBJ_NOTNULL(ctx, VRT_CTX_MAGIC);
 	CHECK_OBJ_NOTNULL(d, DIRECTOR_MAGIC);
@@ -390,12 +391,18 @@ vbe_dir_http1pipe(VRT_CTX, VCL_BACKEND d)
 		i = V1F_SendReq(ctx->req->wrk, ctx->bo,
 		    &v1a.bereq, &v1a.out);
 		VSLb_ts_req(ctx->req, "Pipe", W_TIM_real(ctx->req->wrk));
-		if (i == 0)
-			V1P_Process(ctx->req, *PFD_Fd(pfd), &v1a);
+		if (i == 0) {
+			deadline = ctx->bo->task_deadline;
+			if (isnan(deadline))
+				deadline = cache_param->pipe_task_deadline;
+			if (deadline > 0.)
+				deadline += ctx->req->sp->t_idle;
+			retval = V1P_Process(ctx->req, *PFD_Fd(pfd), &v1a,
+			    deadline);
+		}
 		VSLb_ts_req(ctx->req, "PipeSess", W_TIM_real(ctx->req->wrk));
-		ctx->bo->htc->doclose = SC_TX_PIPE;
+		ctx->bo->htc->doclose = retval;
 		vbe_dir_finish(ctx, d);
-		retval = SC_TX_PIPE;
 	}
 	V1P_Charge(ctx->req, &v1a, bp->vsc);
 	CHECK_OBJ_NOTNULL(retval, STREAM_CLOSE_MAGIC);

bin/varnishd/cache/cache_req_fsm.c

@@ -795,6 +795,7 @@ cnt_pipe(struct worker *wrk, struct req *req)
 	}
 
 	bo->wrk = wrk;
+	bo->task_deadline = NAN; /* XXX: copy req->task_deadline */
 	if (WS_Overflowed(req->ws))
 		wrk->vpi->handling = VCL_RET_FAIL;
 	else

bin/varnishd/cache/cache_vrt_var.c

@@ -409,6 +409,8 @@ VRT_r_bereq_##which(VRT_CTX)					\
 BEREQ_TIMEOUT(connect_timeout)
 BEREQ_TIMEOUT(first_byte_timeout)
 BEREQ_TIMEOUT(between_bytes_timeout)
+BEREQ_TIMEOUT(task_deadline)
+
 
 /*--------------------------------------------------------------------*/
 

bin/varnishd/http1/cache_http1.h

@@ -54,7 +54,8 @@ struct v1p_acct {
 
 int V1P_Enter(void);
 void V1P_Leave(void);
-void V1P_Process(const struct req *, int fd, struct v1p_acct *);
+stream_close_t V1P_Process(const struct req *, int fd, struct v1p_acct *,
+    vtim_real deadline);
 void V1P_Charge(struct req *, const struct v1p_acct *, struct VSC_vbe *);
 
 /* cache_http1_line.c */

bin/varnishd/http1/cache_http1_pipe.c

@@ -40,6 +40,7 @@
 
 #include "cache_http1.h"
 #include "vtcp.h"
+#include "vtim.h"
 
 #include "VSC_vbe.h"
 
@@ -113,10 +114,13 @@ V1P_Charge(struct req *req, const struct v1p_acct *a, struct VSC_vbe *b)
 	Lck_Unlock(&pipestat_mtx);
 }
 
-void
-V1P_Process(const struct req *req, int fd, struct v1p_acct *v1a)
+stream_close_t
+V1P_Process(const struct req *req, int fd, struct v1p_acct *v1a,
+    vtim_real deadline)
 {
 	struct pollfd fds[2];
+	vtim_dur tmo, tmo_task;
+	stream_close_t sc;
 	int i, j;
 
 	CHECK_OBJ_NOTNULL(req, REQ_MAGIC);
@@ -128,7 +132,7 @@ V1P_Process(const struct req *req, int fd, struct v1p_acct *v1a)
 		    req->htc->pipeline_e - req->htc->pipeline_b);
 		VTCP_Assert(j);
 		if (j < 0)
-			return;
+			return (SC_OVERLOAD);
 		req->htc->pipeline_b = NULL;
 		req->htc->pipeline_e = NULL;
 		v1a->in += j;
@@ -139,11 +143,21 @@ V1P_Process(const struct req *req, int fd, struct v1p_acct *v1a)
 	fds[1].fd = req->sp->fd;
 	fds[1].events = POLLIN;
 
+	sc = SC_TX_PIPE;
 	while (fds[0].fd > -1 || fds[1].fd > -1) {
 		fds[0].revents = 0;
 		fds[1].revents = 0;
-		i = poll(fds, 2,
-		    (int)(cache_param->pipe_timeout * 1e3));
+		tmo = cache_param->pipe_timeout;
+		if (tmo == 0.)
+			tmo = -1.;
+		if (deadline > 0.) {
+			tmo_task = deadline - VTIM_real();
+			tmo = (tmo > 0.) ? vmin(tmo, tmo_task) : tmo_task;
+			tmo = vmax(tmo, 0.);
+		}
+		i = poll(fds, 2, (int)(tmo * 1e3));
+		if (i == 0)
+			sc = SC_RX_TIMEOUT;
 		if (i < 1)
 			break;
 		if (fds[0].revents &&
@@ -165,6 +179,8 @@ V1P_Process(const struct req *req, int fd, struct v1p_acct *v1a)
 			fds[1].fd = -1;
 		}
 	}
+
+	return (sc);
 }
 
 /*--------------------------------------------------------------------*/

bin/varnishtest/tests/s00013.vtc

@@ -0,0 +1,75 @@
+varnishtest "pipe timeouts"
+
+server s1 {
+	rxreq
+	txresp -hdr "transfer-encoding: chunked"
+	delay 1.1
+	close
+
+	loop 3 {
+		accept
+		rxreq
+		txresp -hdr "transfer-encoding: chunked"
+		expect_close
+	}
+
+	accept
+	non_fatal
+	rxreq
+	txresp -hdr "transfer-encoding: chunked"
+	loop 20 {
+		chunkedlen 1
+		delay 0.1
+	}
+} -start
+
+varnish v1 -cliok "param.set pipe_timeout 0s"
+varnish v1 -cliok "param.set pipe_task_deadline 0s"
+varnish v1 -vcl+backend {
+	sub vcl_pipe {
+		if (req.method == "TMO") {
+			set bereq.task_deadline = 1.1s;
+		}
+	}
+} -start
+
+logexpect l1 -v v1 -g raw -q SessClose {
+	expect 1000 * SessClose {^TX_PIPE 1\.}
+	expect 1003 * SessClose {^RX_TIMEOUT 0\.}
+	expect 1006 * SessClose {^RX_TIMEOUT 1\.}
+	expect 1009 * SessClose {^RX_TIMEOUT 1\.}
+	expect 1012 * SessClose {^RX_TIMEOUT 1\.}
+} -start
+
+client c1 {
+	non_fatal
+	txreq -method PIPE
+	rxresp
+} -run
+
+varnish v1 -cliok "param.set pipe_timeout 500ms"
+varnish v1 -cliok "param.set pipe_task_deadline 0s"
+client c1 -run
+
+varnish v1 -cliok "param.set pipe_timeout 0s"
+varnish v1 -cliok "param.set pipe_task_deadline 1.1s"
+client c1 -run
+
+varnish v1 -cliok "param.set pipe_timeout 0s"
+varnish v1 -cliok "param.set pipe_task_deadline 0s"
+
+client c2 {
+	non_fatal
+	txreq -method TMO
+	rxresp
+} -run
+
+varnish v1 -cliok "param.set pipe_timeout 500ms"
+varnish v1 -cliok "param.set pipe_task_deadline 1.1s"
+client c1 -run
+
+logexpect l1 -wait
+
+varnish v1 -expect MAIN.s_pipe == 5
+varnish v1 -expect MAIN.sc_tx_pipe == 1
+varnish v1 -expect MAIN.sc_rx_timeout == 4

doc/sphinx/reference/varnishd.rst

@@ -540,6 +540,59 @@ to load the configurations at startup.
 RUN TIME PARAMETERS
 ===================
 
+Runtime parameters can either be set during startup with the ``-p`` command
+line option for ``varnishd(1)`` or through the CLI using the ``param.set`` or
+``param.reset`` commands. They can be locked during startup with the ``-r``
+command line option.
+
+Run Time Parameter Units
+------------------------
+
+There are different types of parameters that may accept a list of specific
+values, or optionally take a unit suffix.
+
+bool
+~~~~
+
+A boolean parameter accepts the values ``on`` and ``off``.
+
+It will also recognize the following values:
+
+- ``yes`` and ``no``
+- ``true`` and ``false``
+- ``enable`` and ``disable``
+
+bytes
+~~~~~
+
+A bytes parameter requires one of the following units suffixes:
+
+- ``b`` (bytes)
+- ``k`` (kibibytes, 1024 bytes)
+- ``m`` (mebibytes, 1024 kibibytes)
+- ``g`` (gibibytes, 1024 mebibytes)
+- ``t`` (tebibytes, 1024 gibibytes)
+- ``p`` (pebibytes, 1024 tebibytes)
+
+Multiplicator units may be appended with an extra ``b``. For example ``32k``
+is equivalent to ``32kb``. Bytes units are case-insensitive.
+
+seconds
+~~~~~~~
+
+A duration parameter may accept the following units suffixes:
+
+- ``ms`` (milliseconds)
+- ``s`` (seconds)
+- ``m`` (minutes)
+- ``h`` (hours)
+- ``d`` (days)
+- ``w`` (weeks)
+- ``y`` (years)
+
+If the parameter is a timeout or a deadline, a value of zero (when allowed)
+disables the effect of the parameter.
+
 Run Time Parameter Flags
 ------------------------
 

doc/sphinx/reference/vcl_var.rst

@@ -840,6 +840,20 @@ bereq.retries
 	A count of how many times this request has been retried.
 
 
+.. _bereq.task_deadline:
+
+bereq.task_deadline
+
+	Type: DURATION
+
+	Readable from: vcl_pipe
+
+	Writable from: vcl_pipe
+
+	Deadline for pipe sessions, defaults ``0s``, which falls back to the
+        ``pipe_task_deadline`` parameter, see :ref:`varnishd(1)`
+
+
 .. _bereq.time:
 
 bereq.time

include/tbl/params.h

@@ -766,6 +766,18 @@ PARAM_SIMPLE(
 	"Maximum number of sessions dedicated to pipe transactions."
 )
 
+PARAM_SIMPLE(
+	/* name */	pipe_task_deadline,
+	/* type */	timeout,
+	/* min */	"0.000",
+	/* max */	NULL,
+	/* def */	"0s",
+	/* units */	"seconds",
+	/* descr */
+	"Deadline for PIPE sessions. Regardless of activity in either "
+	"direction after this many seconds, the session is closed."
+)
+
 PARAM_SIMPLE(
 	/* name */	pipe_timeout,
 	/* type */	timeout,