Tighten up the Session pulldown sequence and let GOWAY do it.

bin/varnishd/http2/cache_http2.h

@@ -220,5 +220,3 @@ void h2_del_req(struct worker *, struct h2_req *);
 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*);
-h2_error H2_StreamError(uint32_t);
-h2_error H2_ConnectionError(uint32_t);

bin/varnishd/http2/cache_http2_deliver.c

@@ -81,15 +81,17 @@ h2_bytes(struct req *req, enum vdp_action act, void **priv,
 	CHECK_OBJ_NOTNULL(req, REQ_MAGIC);
 	CAST_OBJ_NOTNULL(r2, req->transport_priv, H2_REQ_MAGIC);
 	(void)priv;
+
 	if (act == VDP_INIT)
 		return (0);
+	if (r2->error)
+		return (-1);
 	H2_Send_Get(req->wrk, r2->h2sess, r2);
 	H2_Send(req->wrk, r2,
 	    H2_F_DATA,
 	    act == VDP_FINI ? H2FF_DATA_END_STREAM : H2FF_NONE,
 	    len, ptr);
 	H2_Send_Rel(r2->h2sess, r2);
-
 	return (0);
 }
 

bin/varnishd/http2/cache_http2_proto.c

@@ -95,8 +95,8 @@ static const h2_error stream_errors[] = {
 
 #define NSTREAMERRORS (sizeof(stream_errors)/sizeof(stream_errors[0]))
 
-h2_error
-H2_StreamError(uint32_t u)
+static h2_error
+h2_streamerror(uint32_t u)
 {
 	if (u < NSTREAMERRORS && stream_errors[u] != NULL)
 		return (stream_errors[u]);
@@ -120,8 +120,8 @@ static const h2_error conn_errors[] = {
 
 #define NCONNERRORS (sizeof(conn_errors)/sizeof(conn_errors[0]))
 
-h2_error
-H2_ConnectionError(uint32_t u)
+static h2_error
+h2_connectionerror(uint32_t u)
 {
 	if (u < NCONNERRORS && conn_errors[u] != NULL)
 		return (conn_errors[u]);
@@ -150,7 +150,6 @@ h2_new_req(const struct worker *wrk, struct h2_sess *h2,
 	r2->stream = stream;
 	r2->req = req;
 	req->transport_priv = r2;
-	// XXX: ordering ?
 	Lck_Lock(&h2->sess->mtx);
 	VTAILQ_INSERT_TAIL(&h2->streams, r2, list);
 	Lck_Unlock(&h2->sess->mtx);
@@ -276,9 +275,10 @@ h2_rx_rst_stream(struct worker *wrk, struct h2_sess *h2, struct h2_req *r2)
 	if (r2 == NULL)
 		return (0);
 	Lck_Lock(&h2->sess->mtx);
-	r2->error = H2_StreamError(vbe32dec(h2->rxf_data));
+	r2->error = h2_streamerror(vbe32dec(h2->rxf_data));
 	if (r2->wrk != NULL)
 		AZ(pthread_cond_signal(&r2->wrk->cond));
+	Lck_Unlock(&h2->sess->mtx);
 	return (0);
 }
 
@@ -292,8 +292,9 @@ h2_rx_goaway(struct worker *wrk, struct h2_sess *h2, struct h2_req *r2)
 	(void)wrk;
 	(void)r2;
 	h2->goaway_last_stream = vbe32dec(h2->rxf_data);
-	h2->error = H2_ConnectionError(vbe32dec(h2->rxf_data + 4));
-	return (0);
+	h2->error = h2_connectionerror(vbe32dec(h2->rxf_data + 4));
+	VSLb(h2->vsl, SLT_Debug, "GOAWAY %s", h2->error->name);
+	return (h2->error);
 }
 
 /**********************************************************************
@@ -562,7 +563,7 @@ h2_rx_data(struct worker *wrk, struct h2_sess *h2, struct h2_req *r2)
 	h2->mailcall = r2;
 	Lck_Lock(&h2->sess->mtx);
 	AZ(pthread_cond_broadcast(h2->cond));
-	while (h2->mailcall != NULL)
+	while (h2->mailcall != NULL && h2->error == 0 && r2->error == 0)
 		AZ(Lck_CondWait(h2->cond, &h2->sess->mtx, 0));
 	Lck_Unlock(&h2->sess->mtx);
 	return (0);
@@ -587,21 +588,26 @@ h2_vfp_body(struct vfp_ctx *vc, struct vfp_entry *vfe, void *ptr, ssize_t *lp)
 	*lp = 0;
 
 	Lck_Lock(&h2->sess->mtx);
-	while (h2->mailcall != r2)
+	while (h2->mailcall != r2 && h2->error == 0 && r2->error == 0)
 		AZ(Lck_CondWait(h2->cond, &h2->sess->mtx, 0));
-	if (l > h2->rxf_len)
-		l = h2->rxf_len;
-	if (l > 0) {
-		memcpy(ptr, h2->rxf_data, l);
-		h2->rxf_data += l;
-		h2->rxf_len -= l;
-	}
-	*lp = l;
-	if (h2->rxf_len == 0) {
-		if (h2->rxf_flags & H2FF_DATA_END_STREAM)
-			retval = VFP_END;
+	if (h2->mailcall == r2) {
+		assert(h2->mailcall == r2);
+		if (l > h2->rxf_len)
+			l = h2->rxf_len;
+		if (l > 0) {
+			memcpy(ptr, h2->rxf_data, l);
+			h2->rxf_data += l;
+			h2->rxf_len -= l;
+		}
+		*lp = l;
+		if (h2->rxf_len == 0) {
+			if (h2->rxf_flags & H2FF_DATA_END_STREAM)
+				retval = VFP_END;
+		}
 		h2->mailcall = NULL;
 		AZ(pthread_cond_broadcast(h2->cond));
+	} else {
+		retval = VFP_ERROR;
 	}
 	Lck_Unlock(&h2->sess->mtx);
 	return (retval);
@@ -734,6 +740,7 @@ h2_rxframe(struct worker *wrk, struct h2_sess *h2)
 	if (hs != HTC_S_COMPLETE) {
 		Lck_Lock(&h2->sess->mtx);
 		VSLb(h2->vsl, SLT_Debug, "H2: No frame (hs=%d)", hs);
+		h2->error = H2CE_NO_ERROR;
 		Lck_Unlock(&h2->sess->mtx);
 		return (0);
 	}
@@ -786,7 +793,8 @@ h2_rxframe(struct worker *wrk, struct h2_sess *h2)
 	}
 
 	h2e = h2_procframe(wrk, h2, h2f);
-	if (h2e) {
+	if (h2->error == 0 && h2e) {
+		h2->error = h2e;
 		vbe32enc(b, h2->highest_stream);
 		vbe32enc(b + 4, h2e->val);
 		H2_Send_Get(wrk, h2, h2->req0);

bin/varnishd/http2/cache_http2_session.c

@@ -289,7 +289,7 @@ h2_new_session(struct worker *wrk, void *arg)
 	struct req *req;
 	struct sess *sp;
 	struct h2_sess *h2;
-	struct h2_req *r2, *r22;
+	struct h2_req *r2;
 	uintptr_t wsp;
 
 	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
@@ -332,12 +332,30 @@ h2_new_session(struct worker *wrk, void *arg)
 		HTC_RxInit(h2->htc, h2->ws);
 	}
 
+	AN(h2->error);
 	/* Delete all idle streams */
-	VTAILQ_FOREACH_SAFE(r2, &h2->streams, list, r22) {
-		if (r2->state == H2_S_IDLE)
-			h2_del_req(wrk, r2);
+	Lck_Lock(&sp->mtx);
+	VSLb(h2->vsl, SLT_Debug, "H2 CLEANUP %p", h2->error);
+	VTAILQ_FOREACH(r2, &h2->streams, list) {
+		if (r2->error == 0)
+			r2->error = h2->error;
+		if (r2->wrk != NULL)
+			AZ(pthread_cond_signal(&r2->wrk->cond));
+	}
+	while (1) {
+		VTAILQ_FOREACH(r2, &h2->streams, list)
+			if (r2->state == H2_S_IDLE && r2 != h2->req0)
+				break;
+		if (r2 == NULL)
+			break;
+		Lck_Unlock(&sp->mtx);
+		h2_del_req(wrk, r2);
+		Lck_Lock(&sp->mtx);
 	}
+	VSLb(h2->vsl, SLT_Debug, "H2CLEAN done");
+	Lck_Unlock(&sp->mtx);
 	h2->cond = NULL;
+	h2_del_req(wrk, h2->req0);
 }
 
 struct transport H2_transport = {

bin/varnishtest/tests/t02008.vtc

+varnishtest "Test GOAWAY/session cleanup"
+
+server s1 {
+	rxreq
+	txresp -hdr "Content-Type: text/plain" -body response
+} -start
+
+varnish v1 -vcl+backend {} -start
+varnish v1 -cliok {param.set feature +http2}
+varnish v1 -cliok "param.set debug +syncvsl"
+
+client c1 {
+	stream 1 {
+		txprio
+	} -run
+	stream 3 {
+		txreq
+
+		# First, HTTP checks
+		rxresp
+		expect resp.http.content-Type == "text/plain"
+
+		# Then, payload checks
+		expect resp.body == response
+	} -run
+	stream 5 {
+		txprio
+	} -run
+	stream 0 {
+		txgoaway -err 7
+	} -run
+	expect_close
+} -run
+
+delay .1
+
+client c1 {
+	stream 1 {
+		txreq
+
+		# First, HTTP checks
+		rxresp
+		expect resp.http.content-Type == "text/plain"
+
+		# Then, payload checks
+		expect resp.body == response
+	} -run
+} -run
+