http2: Move all socket operations to the session thread

bin/varnishd/Makefile.am

@@ -81,6 +81,7 @@ varnishd_SOURCES = \
 	http2/cache_http2_hpack.c \
 	http2/cache_http2_panic.c \
 	http2/cache_http2_proto.c \
+	http2/cache_http2_reqbody.c \
 	http2/cache_http2_send.c \
 	http2/cache_http2_session.c \
 	mgt/mgt_child.c \

bin/varnishd/cache/cache_session.c

@@ -258,18 +258,13 @@ HTC_Status(enum htc_status_e e, const char **name, const char **desc)
 void
 HTC_RxInit(struct http_conn *htc, struct ws *ws)
 {
-	unsigned rollback;
 	int l;
 
 	CHECK_OBJ_NOTNULL(htc, HTTP_CONN_MAGIC);
 	htc->ws = ws;
 
-	/* NB: HTTP/1 keep-alive triggers a rollback, so does the first
-	 * request of a session or an h2 request where the rollback is a
-	 * no-op in terms of workspace usage.
-	 */
-	rollback = !strcasecmp(ws->id, "req") && htc->body_status == NULL;
-	l = WS_Pipeline(htc->ws, htc->pipeline_b, htc->pipeline_e, rollback);
+	l = WS_Pipeline(htc->ws, htc->pipeline_b, htc->pipeline_e,
+	    htc->pipeline_snap);
 	xxxassert(l >= 0);
 
 	htc->rxbuf_b = WS_Reservation(ws);
@@ -410,6 +405,38 @@ HTC_RxStuff(struct http_conn *htc, htc_complete_f *func,
 	}
 }
 
+/*--------------------------------------------------------------------
+ * Prune a vector of struct iovec
+ */
+
+void
+VIOV_prune(struct iovec *iov, unsigned *n, size_t l)
+{
+	unsigned u;
+
+	if (l == 0)
+		return;
+
+	AN(iov);
+	AN(n);
+
+	u = 0;
+	while (l > 0) {
+		assert(u < *n);
+		if (iov[u].iov_len <= l) {
+			l -= iov[u].iov_len;
+			u++;
+		} else {
+			iov[u].iov_base = (char *)iov[u].iov_base + l;
+			iov[u].iov_len -= l;
+			break;
+		}
+	}
+
+	memmove(iov, &iov[u], (*n - u) * sizeof *iov);
+	*n -= u;
+}
+
 /*--------------------------------------------------------------------
  * Get a new session, preferably by recycling an already ready one
  *

bin/varnishd/cache/cache_varnishd.h

@@ -104,6 +104,7 @@ struct http_conn {
 	char			*rxbuf_e;
 	char			*pipeline_b;
 	char			*pipeline_e;
+	uintptr_t		pipeline_snap;
 	ssize_t			content_length;
 	void			*priv;
 
@@ -462,6 +463,8 @@ enum htc_status_e HTC_RxStuff(struct http_conn *, htc_complete_f *,
     vtim_real *t1, vtim_real *t2, vtim_real ti, vtim_real tn, vtim_dur td,
     int maxbytes);
 
+void VIOV_prune(struct iovec *iov, unsigned *n, size_t l);
+
 #define SESS_ATTR(UP, low, typ, len)					\
 	int SES_Set_##low(const struct sess *sp, const typ *src);	\
 	int SES_Reserve_##low(struct sess *sp, typ **dst, ssize_t *sz);
@@ -567,7 +570,9 @@ WS_IsReserved(const struct ws *ws)
 
 void *WS_AtOffset(const struct ws *ws, unsigned off, unsigned len);
 unsigned WS_ReservationOffset(const struct ws *ws);
-int WS_Pipeline(struct ws *, const void *b, const void *e, unsigned rollback);
+
+extern uintptr_t const ws_pipeline_rollback;
+int WS_Pipeline(struct ws *, const void *b, const void *e, uintptr_t snap);
 
 /* cache_ws_common.c */
 void WS_Id(const struct ws *ws, char *id);

bin/varnishd/cache/cache_ws.c

@@ -136,14 +136,16 @@ WS_Reset(struct ws *ws, uintptr_t pp)
  */
 
 int
-WS_Pipeline(struct ws *ws, const void *b, const void *e, unsigned rollback)
+WS_Pipeline(struct ws *ws, const void *b, const void *e, uintptr_t snap)
 {
 	unsigned r, l;
 
 	WS_Assert(ws);
 
-	if (rollback)
+	if (snap == ws_pipeline_rollback)
 		WS_Rollback(ws, 0);
+	else if (snap != 0)
+		WS_Rollback(ws, snap);
 
 	r = WS_ReserveAll(ws);
 

bin/varnishd/cache/cache_ws_common.c

@@ -37,6 +37,8 @@
 
 #include "cache_varnishd.h"
 
+uintptr_t const ws_pipeline_rollback = (uintptr_t)&ws_pipeline_rollback;
+
 void
 WS_Id(const struct ws *ws, char *id)
 {

bin/varnishd/cache/cache_ws_emu.c

@@ -222,7 +222,7 @@ WS_Reset(struct ws *ws, uintptr_t pp)
 }
 
 int
-WS_Pipeline(struct ws *ws, const void *b, const void *e, unsigned rollback)
+WS_Pipeline(struct ws *ws, const void *b, const void *e, uintptr_t snap)
 {
 	void *tmp;
 	unsigned r, l;
@@ -248,8 +248,10 @@ WS_Pipeline(struct ws *ws, const void *b, const void *e, unsigned rollback)
 		tmp = NULL;
 	}
 
-	if (rollback)
+	if (snap == ws_pipeline_rollback)
 		WS_Rollback(ws, 0);
+	else if (snap != 0)
+		WS_Rollback(ws, snap);
 
 	r = WS_ReserveAll(ws);
 

bin/varnishd/http1/cache_http1_fsm.c

@@ -111,6 +111,7 @@ http1_new_session(struct worker *wrk, void *arg)
 	sp = req->sp;
 	CHECK_OBJ_NOTNULL(sp, SESS_MAGIC);
 
+	req->htc->pipeline_snap = ws_pipeline_rollback;
 	HTC_RxInit(req->htc, req->ws);
 
 	sz = sizeof u;

bin/varnishd/http2/cache_http2.h

@@ -35,6 +35,9 @@ struct h2h_decode;
 struct h2_frame_s;
 
 #include "hpack/vhp.h"
+#include "vefd.h"
+
+#define H2_TX_BUFSIZE                  1024
 
 /**********************************************************************/
 
@@ -140,42 +143,44 @@ struct h2_req {
 	int				counted;
 	struct h2_sess			*h2sess;
 	struct req			*req;
-	double				t_send;
-	double				t_winupd;
-	pthread_cond_t			*cond;
+	vtim_real			t_send;
+	vtim_real			t_win_low;
 	VTAILQ_ENTRY(h2_req)		list;
-	int64_t				t_window;
-	int64_t				r_window;
 
-	/* Where to wake this stream up */
-	struct worker			*wrk;
+	int64_t				tx_window;
+	int64_t				rx_window;
 
 	struct h2_rxbuf			*rxbuf;
+	struct h2_reqbody_waiter        *reqbody_waiter;
+	h2_error                        async_error;
 
-	VTAILQ_ENTRY(h2_req)		tx_list;
 	h2_error			error;
 };
 
 VTAILQ_HEAD(h2_req_s, h2_req);
 
+struct h2_send_large;
+VTAILQ_HEAD(h2_send_large_s, h2_send_large);
+
 struct h2_sess {
 	unsigned			magic;
 #define H2_SESS_MAGIC			0xa16f7e4b
 
+	unsigned			expect_settings_next;
+
 	pthread_t			rxthr;
-	pthread_cond_t			*cond;
-	pthread_cond_t			winupd_cond[1];
 
 	struct sess			*sess;
 	int				refcnt;
 	int				open_streams;
-	int				winup_streams;
+	int				win_low_streams;
 	uint32_t			highest_stream;
-	int				goaway;
 	int				bogosity;
-	int				do_sweep;
 
-	struct h2_req			*req0;
+	struct vefd                     efd[1];
+
+	int64_t				tx_window;
+	int64_t				rx_window;
 
 	struct h2_req_s			streams;
 
@@ -186,6 +191,23 @@ struct h2_sess {
 	struct h2h_decode		*decode;
 	struct vht_table		dectbl[1];
 
+	vtim_real			deadline;
+
+	struct iovec			tx_vec[2]; /* Must be 2 wide */
+	unsigned			tx_nvec;
+
+	unsigned			tx_stopped;
+
+	uint8_t				*tx_s_start;
+	uint8_t				*tx_s_end;
+	uint8_t				*tx_s_head;
+	uint8_t				*tx_s_mark;
+
+	struct h2_send_large_s		tx_l_queue;
+	struct h2_send_large		*tx_l_current;
+	uint8_t				tx_l_hdrbuf[9];
+	char				tx_l_stuck;
+
 	unsigned			rxf_len;
 	unsigned			rxf_type;
 	unsigned			rxf_flags;
@@ -195,11 +217,8 @@ struct h2_sess {
 	struct h2_settings		remote_settings;
 	struct h2_settings		local_settings;
 
-	struct req			*new_req;
+	struct h2_req			*hpack_lock;
 	vtim_real			t1;	// t_first for new_req
-	uint32_t			goaway_last_stream;
-
-	VTAILQ_HEAD(,h2_req)		txqueue;
 
 	h2_error			error;
 
@@ -213,7 +232,17 @@ struct h2_sess {
 	vtim_real			last_rst;
 };
 
-#define ASSERT_RXTHR(h2) do {assert(h2->rxthr == pthread_self());} while(0)
+#define ASSERT_H2_SESS(h2)						\
+	do {								\
+		CHECK_OBJ_NOTNULL(h2, H2_SESS_MAGIC);			\
+		assert(pthread_equal(h2->rxthr, pthread_self()));	\
+	} while (0)
+
+#define ASSERT_H2_REQ(h2) \
+	do {								\
+		CHECK_OBJ_NOTNULL(h2, H2_SESS_MAGIC);			\
+		assert(!pthread_equal(h2->rxthr, pthread_self()));	\
+	} while (0)
 
 /* http2/cache_http2_panic.c */
 #ifdef TRANSPORT_MAGIC
@@ -235,7 +264,6 @@ struct h2h_decode {
 	unsigned			has_scheme:1;
 	h2_error			error;
 	enum vhd_ret_e			vhd_ret;
-	struct ws			*ws;
 	char				*out;
 	int64_t				limit;
 	size_t				out_l;
@@ -244,38 +272,42 @@ struct h2h_decode {
 	struct vhd_decode		vhd[1];
 };
 
-void h2h_decode_hdr_init(const struct h2_sess *h2);
-h2_error h2h_decode_hdr_fini(const struct h2_sess *h2);
+void h2h_decode_hdr_init(struct h2_sess *h2, struct h2_req *);
+h2_error h2h_decode_hdr_fini(struct h2_sess *h2);
 h2_error h2h_decode_bytes(struct h2_sess *h2, const uint8_t *ptr,
     size_t len);
 
 /* cache_http2_send.c */
-void H2_Send_Get(struct worker *, struct h2_sess *, struct h2_req *);
-void H2_Send_Rel(struct h2_sess *, const struct h2_req *);
-
-void H2_Send_Frame(struct worker *, struct h2_sess *,
-    h2_frame type, uint8_t flags, uint32_t len, uint32_t stream,
-    const void *);
-
-void H2_Send_RST(struct worker *wrk, struct h2_sess *h2,
-    const struct h2_req *r2, uint32_t stream, h2_error h2e);
-
-void H2_Send(struct worker *, struct h2_req *, h2_frame type, uint8_t flags,
-    uint32_t len, const void *, uint64_t *acct);
+int H2_Send_RST(struct h2_sess *h2, uint32_t stream, h2_error h2e);
+int H2_Send_SETTINGS(struct h2_sess *h2, uint8_t flags, ssize_t len,
+    const uint8_t *buf);
+int H2_Send_PING(struct h2_sess *h2, uint8_t flags, uint64_t data);
+int H2_Send_GOAWAY(struct h2_sess *h2, uint32_t last_stream_id, h2_error h2e);
+int H2_Send_WINDOW_UPDATE(struct h2_sess *h2, uint32_t stream, uint32_t incr);
+int H2_Send(struct vsl_log *vsl, struct h2_req *r2, h2_frame ftyp,
+    uint8_t flags, uint32_t len, const void *ptr);
+ssize_t H2_Send_TxStuff(struct h2_sess *h2);
+int H2_Send_Something(struct h2_sess *h2);
+int H2_Send_Pending(struct h2_sess *h2);
+void H2_Send_Shutdown(struct h2_sess *h2);
+void H2_Send_Stop(struct h2_sess *h2);
 
 /* cache_http2_proto.c */
-struct h2_req * h2_new_req(struct h2_sess *, unsigned stream, struct req *);
-h2_error h2_stream_tmo(struct h2_sess *, const struct h2_req *, vtim_real);
-void h2_del_req(struct worker *, struct h2_req *);
-void h2_kill_req(struct worker *, struct h2_sess *, struct h2_req *, h2_error);
-int h2_rxframe(struct worker *, struct h2_sess *);
+const char *h2_framename(int frame);
+h2_error h2_errcheck(const struct h2_req *r2);
+void h2_async_error(struct h2_req *r2, h2_error h2e);
+void h2_attention(struct h2_sess *h2);
+void h2_stream_setstate(struct h2_req *r2, enum h2_stream_e state);
+void h2_run(struct worker *wrk, struct h2_sess *h2);
+struct h2_req * h2_new_req(struct h2_sess *, unsigned stream, struct req **);
+void h2_kill_req(struct worker *, struct h2_sess *, struct h2_req **, h2_error);
 h2_error h2_set_setting(struct h2_sess *, const uint8_t *);
-void h2_req_body(struct req*);
 task_func_t h2_do_req;
 #ifdef TRANSPORT_MAGIC
 vtr_req_fail_f h2_req_fail;
 #endif
 
-/* cache_http2_session.c */
-void
-H2S_Lock_VSLb(const struct h2_sess *, enum VSL_tag_e, const char *, ...);
+/* cache_http2_reqbody.c */
+h2_error h2_reqbody_data(struct worker *, struct h2_sess *, struct h2_req *);
+void h2_reqbody(struct req *);
+void h2_reqbody_kick(struct h2_req *r2);