New pipe_sess_max runtime parameter

Optionally, it can limit the number of concurrent pipe transactions as
they each monopolize one worker thread.

bin/varnishd/cache/cache_req_fsm.c

@@ -696,22 +696,24 @@ cnt_pipe(struct worker *wrk, struct req *req)
 	VCL_pipe_method(req->vcl, wrk, req, bo, NULL);
 
 	switch (wrk->handling) {
-	case VCL_RET_FAIL:
-		req->req_step = R_STP_VCLFAIL;
-		nxt = REQ_FSM_MORE;
-		break;
 	case VCL_RET_SYNTH:
 		req->req_step = R_STP_SYNTH;
 		nxt = REQ_FSM_MORE;
 		break;
 	case VCL_RET_PIPE:
-		XXXAZ(V1P_Enter());
-		AZ(bo->req);
-		bo->req = req;
-		bo->wrk = wrk;
-		SES_Close(req->sp, VDI_Http1Pipe(req, bo));
-		nxt = REQ_FSM_DONE;
-		V1P_Leave();
+		if (V1P_Enter() == 0) {
+			AZ(bo->req);
+			bo->req = req;
+			bo->wrk = wrk;
+			SES_Close(req->sp, VDI_Http1Pipe(req, bo));
+			nxt = REQ_FSM_DONE;
+			V1P_Leave();
+			break;
+		}
+		/* fall through */
+	case VCL_RET_FAIL:
+		req->req_step = R_STP_VCLFAIL;
+		nxt = REQ_FSM_MORE;
 		break;
 	default:
 		WRONG("Illegal return from vcl_pipe{}");

bin/varnishd/http1/cache_http1_pipe.c

@@ -65,11 +65,16 @@ rdf(int fd0, int fd1, uint64_t *pcnt)
 int
 V1P_Enter(void)
 {
+	int retval = 0;
 
 	Lck_Lock(&pipestat_mtx);
-	VSC_C_main->n_pipe++;
+	if (cache_param->pipe_sess_max == 0 ||
+	    VSC_C_main->n_pipe < cache_param->pipe_sess_max)
+		VSC_C_main->n_pipe++;
+	else
+		retval = -1;
 	Lck_Unlock(&pipestat_mtx);
-	return (0);
+	return (retval);
 }
 
 void

bin/varnishtest/tests/v00059.vtc

-varnishtest "n_pipe gauge"
+varnishtest "concurrent pipe limit"
 
 barrier b1 cond 2
 barrier b2 cond 2
@@ -36,6 +36,7 @@ varnish v1 -expect MAIN.n_pipe == 0
 client c1 {
 	txreq -url "/c1"
 	rxresp
+	expect resp.status == 200
 } -start
 
 barrier b1 sync
@@ -45,10 +46,20 @@ varnish v1 -expect MAIN.n_pipe == 1
 client c2 {
 	txreq -url "/c2"
 	rxresp
+	expect resp.status == 200
 } -start
 
 barrier b3 sync
 varnish v1 -expect MAIN.n_pipe == 2
+
+varnish v1 -cliok "param.set pipe_sess_max 2"
+
+client c3 {
+	txreq
+	rxresp
+	expect resp.status == 503
+} -run
+
 barrier b2 sync
 varnish v1 -expect MAIN.n_pipe == 1
 barrier b4 sync

include/tbl/params.h

@@ -914,6 +914,20 @@ PARAM(
 	/* func */	NULL
 )
 
+PARAM(
+	/* name */	pipe_sess_max,
+	/* typ */	uint,
+	/* min */	"0",
+	/* max */	NULL,
+	/* default */	"0",
+	/* units */	"connections",
+	/* flags */	0,
+	/* s-text */
+	"Maximum number of sessions dedicated to pipe transactions.",
+	/* l-text */	"",
+	/* func */	NULL
+)
+
 PARAM(
 	/* name */	pipe_timeout,
 	/* typ */	timeout,