Fix a race condition where we do not allow reuse of a stream-allowance

as soon as a RST_STREAM has been received.

Testcase by:	xcir

Fixes:	#2923

bin/varnishd/http2/cache_http2.h

@@ -148,6 +148,7 @@ struct h2_sess {
 
 	struct sess			*sess;
 	int				refcnt;
+	int				pending_kills;
 	uint32_t			highest_stream;
 	int				bogosity;
 	int				do_sweep;
@@ -228,8 +229,7 @@ void H2_Send(struct worker *, struct h2_req *,
 struct h2_req * h2_new_req(const struct worker *, struct h2_sess *,
     unsigned stream, struct req *);
 void h2_del_req(struct worker *, const struct h2_req *);
-void h2_kill_req(struct worker *, const struct h2_sess *,
-    struct h2_req *, h2_error);
+void h2_kill_req(struct worker *, struct h2_sess *, struct h2_req *, h2_error);
 int h2_rxframe(struct worker *, struct h2_sess *);
 h2_error h2_set_setting(struct h2_sess *, const uint8_t *);
 void h2_req_body(struct req*);

bin/varnishd/http2/cache_http2_proto.c

@@ -197,6 +197,8 @@ h2_del_req(struct worker *wrk, const struct h2_req *r2)
 	Lck_Lock(&sp->mtx);
 	assert(h2->refcnt > 0);
 	--h2->refcnt;
+	if (h2->pending_kills > 0)
+		h2->pending_kills--;
 	/* XXX: PRIORITY reshuffle */
 	VTAILQ_REMOVE(&h2->streams, r2, list);
 	Lck_Unlock(&sp->mtx);
@@ -206,14 +208,16 @@ h2_del_req(struct worker *wrk, const struct h2_req *r2)
 }
 
 void
-h2_kill_req(struct worker *wrk, const struct h2_sess *h2,
+h2_kill_req(struct worker *wrk, struct h2_sess *h2,
     struct h2_req *r2, h2_error h2e)
 {
 
 	ASSERT_RXTHR(h2);
 	AN(h2e);
 	Lck_Lock(&h2->sess->mtx);
-	VSLb(h2->vsl, SLT_Debug, "KILL st=%u state=%d", r2->stream, r2->state);
+	VSLb(h2->vsl, SLT_Debug, "KILL st=%u state=%d sched=%d",
+	    r2->stream, r2->state, r2->scheduled);
+	h2->pending_kills++;
 	if (r2->error == NULL)
 		r2->error = h2e;
 	if (r2->scheduled) {
@@ -634,7 +638,8 @@ h2_rx_headers(struct worker *wrk, struct h2_sess *h2, struct h2_req *r2)
 	if (r2 == NULL) {
 		if (h2->rxf_stream <= h2->highest_stream)
 			return (H2CE_PROTOCOL_ERROR);	// rfc7540,l,1153,1158
-		if (h2->refcnt > h2->local_settings.max_concurrent_streams) {
+		if (h2->refcnt - h2->pending_kills >
+		    h2->local_settings.max_concurrent_streams) {
 			VSLb(h2->vsl, SLT_Debug,
 			     "H2: stream %u: Hit maximum number of "
 			     "concurrent streams", h2->rxf_stream);

bin/varnishtest/tests/r02923.vtc

+varnishtest "race cond in max_concurrent_streams when request after receiving RST_STREAM"
+
+barrier b1 sock 6
+barrier b2 sock 6
+barrier b3 sock 6
+barrier b4 sock 6
+
+server s1 {
+    rxreq
+    txresp
+} -start
+
+varnish v1 -cliok "param.set feature +http2"
+varnish v1 -cliok "param.set debug +syncvsl"
+varnish v1 -cliok "param.set h2_max_concurrent_streams 3"
+
+varnish v1 -vcl+backend {
+    import vtc;
+
+    sub vcl_recv {
+    }
+
+    sub vcl_backend_fetch {
+	vtc.barrier_sync("${b1_sock}");
+	vtc.barrier_sync("${b2_sock}");
+	vtc.barrier_sync("${b3_sock}");
+	vtc.barrier_sync("${b4_sock}");
+    }
+} -start
+
+client c1 {
+    txpri
+    stream 0 rxsettings -run
+
+    stream 1 {
+	txreq
+	barrier b1 sync
+	barrier b2 sync
+	barrier b3 sync
+	barrier b4 sync
+	rxresp
+	expect resp.status == 200
+    } -start
+
+    stream 3 {
+	barrier b1 sync
+	txreq
+	txrst -err 0x8
+	barrier b2 sync
+	barrier b3 sync
+	barrier b4 sync
+    } -start
+
+    stream 5 {
+	barrier b1 sync
+	barrier b2 sync
+	txreq
+	txrst -err 0x8
+	barrier b3 sync
+	barrier b4 sync
+    } -start
+
+    stream 7 {
+	barrier b1 sync
+	barrier b2 sync
+	barrier b3 sync
+	txreq
+	barrier b4 sync
+	rxresp
+	expect resp.status == 200
+    } -start
+
+    stream 9 {
+	barrier b1 sync
+	barrier b2 sync
+	barrier b3 sync
+	barrier b4 sync
+	txreq
+	rxresp
+	expect resp.status == 200
+    } -start
+} -run