h2: Keep track of streams entering waiting lists

But operate on a different objhead reference guarded by the h2_sess lock
instead. This will allow coordination with the h2 session thread.

Author: dridi

bin/varnishd/cache/cache.h

@@ -466,11 +466,13 @@ struct req {
 	void			*transport_priv;
 
 	VTAILQ_ENTRY(req)	w_list;
+	VTAILQ_ENTRY(req)	t_list;
 
 	struct objcore		*body_oc;
 
-	/* The busy objhead we sleep on */
+	/* The busy objhead we sleep on, referenced up to twice */
 	struct objhead		*hash_objhead;
+	struct objhead		*transport_objhead;
 
 	/* Built Vary string */
 	uint8_t			*vary_b;

bin/varnishd/http2/cache_http2.h

@@ -134,6 +134,7 @@ struct h2_req {
 	int				counted;
 	struct h2_sess			*h2sess;
 	struct req			*req;
+	VTAILQ_HEAD(, req)		waitinglist;
 	double				t_send;
 	double				t_winupd;
 	pthread_cond_t			*cond;

bin/varnishd/http2/cache_http2_proto.c

@@ -159,6 +159,7 @@ h2_new_req(struct h2_sess *h2, unsigned stream, struct req *req)
 	r2->r_window = h2->local_settings.initial_window_size;
 	r2->t_window = h2->remote_settings.initial_window_size;
 	req->transport_priv = r2;
+	VTAILQ_INIT(&r2->waitinglist);
 	Lck_Lock(&h2->sess->mtx);
 	if (stream)
 		h2->open_streams++;
@@ -182,6 +183,7 @@ h2_del_req(struct worker *wrk, struct h2_req *r2)
 	ASSERT_RXTHR(h2);
 	sp = h2->sess;
 	Lck_Lock(&sp->mtx);
+	assert(VTAILQ_EMPTY(&r2->waitinglist));
 	assert(h2->refcnt > 0);
 	--h2->refcnt;
 	/* XXX: PRIORITY reshuffle */

bin/varnishd/http2/cache_http2_session.c

@@ -36,6 +36,7 @@
 #include <stdio.h>
 
 #include "cache/cache_transport.h"
+#include "cache/cache_objhead.h"
 #include "http2/cache_http2.h"
 
 #include "vend.h"
@@ -444,17 +445,59 @@ h2_new_session(struct worker *wrk, void *arg)
 static void
 h2_waitlist(struct worker *wrk, struct req *req)
 {
+	struct objhead *oh;
+	struct h2_req *r2;
 
 	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
 	CHECK_OBJ_NOTNULL(req, REQ_MAGIC);
+	oh = req->hash_objhead;
+	CHECK_OBJ_NOTNULL(oh, OBJHEAD_MAGIC);
+	CAST_OBJ_NOTNULL(r2, req->top->topreq->transport_priv, H2_REQ_MAGIC);
+
+	if (DO_DEBUG(DBG_WAITINGLIST))
+		VSLb(req->vsl, SLT_Debug, "on h2 waiting list <%p>", r2);
+
+	Lck_AssertHeld(&oh->mtx);
+	AN(req->waitinglist);
+	AZ(req->wrk);
+
+	assert(oh->refcnt > 0);
+	oh->refcnt++;
+
+	Lck_Lock(&r2->h2sess->sess->mtx);
+	AZ(req->transport_objhead);
+	req->transport_objhead = oh;
+	VTAILQ_INSERT_TAIL(&r2->waitinglist, req, t_list);
+	Lck_Unlock(&r2->h2sess->sess->mtx);
 }
 
 static void
 h2_reembark(struct worker *wrk, struct req *req)
 {
+	struct objhead *oh;
+	struct h2_req *r2;
 
 	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
 	CHECK_OBJ_NOTNULL(req, REQ_MAGIC);
+	CAST_OBJ_NOTNULL(r2, req->top->topreq->transport_priv, H2_REQ_MAGIC);
+
+	if (DO_DEBUG(DBG_WAITINGLIST))
+		VSLb(req->vsl, SLT_Debug, "off h2 waiting list <%p>", r2);
+
+	Lck_Lock(&r2->h2sess->sess->mtx);
+	VTAILQ_REMOVE(&r2->waitinglist, req, t_list);
+	oh = req->transport_objhead;
+	req->transport_objhead = NULL;
+	Lck_Unlock(&r2->h2sess->sess->mtx);
+
+	CHECK_OBJ_NOTNULL(oh, OBJHEAD_MAGIC);
+	Lck_Lock(&oh->mtx);
+	AN(req->hash_objhead);
+	AZ(req->waitinglist);
+	AZ(req->wrk);
+	assert(oh->refcnt > 1);
+	oh->refcnt--;
+	Lck_Unlock(&oh->mtx);
 
 	if (IS_TOPREQ(req))
 		AZ(Pool_Task(req->sp->pool, req->task, TASK_QUEUE_RUSH));