HTTP/2 write buffering

This commit adds the ability to consolidate HTTP/2 write requests if
there is already one in flight. If it is the case, the code will
consolidate multiple subsequent write request into a larger one
allowing to utilise the network in a more efficient way by creating
larger TCP packets as well as by reducing TLS records overhead (by
creating large TLS records instead of multiple small ones).

This optimisation is especially efficient for clients, creating many
concurrent HTTP/2 streams over a transport connection at once.  This
way, the code might create a small amount of multi-kilobyte requests
instead of many 50-120 byte ones.

In fact, it turned out to work so well that I had to add a work-around
to the code to ensure compatibility with the flamethrower, which, at
the time of writing, does not support TLS records larger than two
kilobytes. Now the code tries to flush the write buffer after 1.5
kilobyte, which is still pretty adequate for our use case.

Essentially, this commit implements a recommendation given by nghttp2
library:

https://nghttp2.org/documentation/nghttp2_session_mem_send.html

lib/isc/netmgr/http.c

@@ -58,6 +58,19 @@
 #define MIN_SUCCESSFUL_HTTP_STATUS (200)
 #define MAX_SUCCESSFUL_HTTP_STATUS (299)
 
+/* This definition sets the upper limit of pending write buffer to an
+ * adequate enough value. That is done mostly to fight a limitation
+ * for a max TLS record size in flamethrower (2K).  In a perfect world
+ * this constant should not be required, if we ever move closer to
+ * that state, the constant, and corresponding code, should be
+ * removed. For now the limit seems adequate enough to fight
+ * "tinygrams" problem. */
+#define FLUSH_HTTP_WRITE_BUFFER_AFTER (1536)
+
+/* This switch is here mostly to test the code interoperability with
+ * buggy implementations */
+#define ENABLE_HTTP_WRITE_BUFFERING 1
+
 #define SUCCESSFUL_HTTP_STATUS(code)             \
 	((code) >= MIN_SUCCESSFUL_HTTP_STATUS && \
 	 (code) <= MAX_SUCCESSFUL_HTTP_STATUS)
@@ -130,6 +143,9 @@ struct isc_nm_http_session {
 	size_t bufsize;
 
 	isc_tlsctx_t *tlsctx;
+
+	ISC_LIST(isc__nm_uvreq_t) pending_write_callbacks;
+	isc_buffer_t *pending_write_data;
 };
 
 typedef enum isc_http_error_responses {
@@ -151,6 +167,7 @@ typedef struct isc_http_send_req {
 	isc_region_t data;
 	isc_nm_cb_t cb;
 	void *cbarg;
+	ISC_LIST(isc__nm_uvreq_t) pending_write_callbacks;
 } isc_http_send_req_t;
 
 static bool
@@ -251,6 +268,7 @@ new_session(isc_mem_t *mctx, isc_tlsctx_t *tctx,
 	isc_mem_attach(mctx, &session->mctx);
 	ISC_LIST_INIT(session->cstreams);
 	ISC_LIST_INIT(session->sstreams);
+	ISC_LIST_INIT(session->pending_write_callbacks);
 
 	*sessionp = session;
 }
@@ -421,6 +439,8 @@ finish_http_session(isc_nm_http_session_t *session) {
 	}
 
 	if (session->handle != NULL) {
+		isc__nm_uvreq_t *cbreq;
+
 		if (!session->closed) {
 			session->closed = true;
 			isc_nm_cancelread(session->handle);
@@ -431,6 +451,22 @@ finish_http_session(isc_nm_http_session_t *session) {
 		} else {
 			server_call_failed_read_cb(ISC_R_UNEXPECTED, session);
 		}
+
+		cbreq = ISC_LIST_HEAD(session->pending_write_callbacks);
+		while (cbreq != NULL) {
+			isc__nm_uvreq_t *next = ISC_LIST_NEXT(cbreq, link);
+			ISC_LIST_UNLINK(session->pending_write_callbacks, cbreq,
+					link);
+			isc__nm_sendcb(cbreq->handle->sock, cbreq,
+				       ISC_R_UNEXPECTED, false);
+			cbreq = next;
+		}
+		ISC_LIST_INIT(session->pending_write_callbacks);
+
+		if (session->pending_write_data != NULL) {
+			isc_buffer_free(&session->pending_write_data);
+		}
+
 		isc_nmhandle_detach(&session->handle);
 	}
 
@@ -895,6 +931,7 @@ http_writecb(isc_nmhandle_t *handle, isc_result_t result, void *arg) {
 	isc_http_send_req_t *req = (isc_http_send_req_t *)arg;
 	isc_nm_http_session_t *session = req->session;
 	isc_nmhandle_t *transphandle = req->transphandle;
+	isc__nm_uvreq_t *cbreq;
 
 	REQUIRE(VALID_HTTP2_SESSION(session));
 	REQUIRE(VALID_NMHANDLE(handle));
@@ -903,6 +940,14 @@ http_writecb(isc_nmhandle_t *handle, isc_result_t result, void *arg) {
 		INSIST(session->handle == handle);
 	}
 
+	cbreq = ISC_LIST_HEAD(req->pending_write_callbacks);
+	while (cbreq != NULL) {
+		isc__nm_uvreq_t *next = ISC_LIST_NEXT(cbreq, link);
+		ISC_LIST_UNLINK(req->pending_write_callbacks, cbreq, link);
+		isc__nm_sendcb(cbreq->handle->sock, cbreq, result, false);
+		cbreq = next;
+	}
+
 	if (req->cb != NULL) {
 		req->cb(req->httphandle, result, req->cbarg);
 		isc_nmhandle_detach(&req->httphandle);
@@ -911,8 +956,8 @@ http_writecb(isc_nmhandle_t *handle, isc_result_t result, void *arg) {
 	isc_mem_put(session->mctx, req->data.base, req->data.length);
 	isc_mem_put(session->mctx, req, sizeof(*req));
 
-	http_do_bio(session, NULL, NULL, NULL);
 	session->sending--;
+	http_do_bio(session, NULL, NULL, NULL);
 	isc_nmhandle_detach(&transphandle);
 	if (result != ISC_R_SUCCESS && session->sending == 0) {
 		finish_http_session(session);
@@ -920,31 +965,147 @@ http_writecb(isc_nmhandle_t *handle, isc_result_t result, void *arg) {
 	isc__nm_httpsession_detach(&session);
 }
 
+static void
+move_pending_send_callbacks(isc_nm_http_session_t *session,
+			    isc_http_send_req_t *send) {
+	REQUIRE(sizeof(session->pending_write_callbacks) ==
+		sizeof(send->pending_write_callbacks));
+	memmove(&send->pending_write_callbacks,
+		&session->pending_write_callbacks,
+		sizeof(session->pending_write_callbacks));
+	ISC_LIST_INIT(session->pending_write_callbacks);
+}
+
 static bool
 http_send_outgoing(isc_nm_http_session_t *session, isc_nmhandle_t *httphandle,
 		   isc_nm_cb_t cb, void *cbarg) {
 	isc_http_send_req_t *send = NULL;
-	const uint8_t *data = NULL;
-	size_t pending;
+	size_t total = 0;
+	uint8_t tmp_data[UINT16_MAX] = { 0 };
+	uint8_t *prepared_data = &tmp_data[0];
+#ifdef ENABLE_HTTP_WRITE_BUFFERING
+	size_t max_total_write_size = 0;
+#endif /* ENABLE_HTTP_WRITE_BUFFERING */
 
 	if (!http_session_active(session) ||
-	    !nghttp2_session_want_write(session->ngsession))
+	    (!nghttp2_session_want_write(session->ngsession) &&
+	     session->pending_write_data == NULL))
 	{
+		INSIST(session->pending_write_data == NULL);
 		return (false);
 	}
 
-	pending = nghttp2_session_mem_send(session->ngsession, &data);
-	if (pending == 0) {
+	while (nghttp2_session_want_write(session->ngsession)) {
+		const uint8_t *data = NULL;
+		const size_t pending =
+			nghttp2_session_mem_send(session->ngsession, &data);
+		const size_t new_total = total + pending;
+
+		/* reallocate buffer if required */
+		if (new_total > sizeof(tmp_data)) {
+			uint8_t *old_prepared_data = prepared_data;
+			const bool allocated = prepared_data != tmp_data;
+
+			prepared_data = isc_mem_get(session->mctx, new_total);
+			memmove(prepared_data, old_prepared_data, total);
+			if (allocated) {
+				isc_mem_put(session->mctx, old_prepared_data,
+					    total);
+			}
+		}
+		memmove(&prepared_data[total], data, pending);
+		total = new_total;
+	}
+
+#ifdef ENABLE_HTTP_WRITE_BUFFERING
+	if (session->pending_write_data != NULL) {
+		max_total_write_size =
+			isc_buffer_usedlength(session->pending_write_data) +
+			total;
+	} else {
+		max_total_write_size = total;
+	}
+
+	/* Here we are trying to flush the pending writes buffer earlier
+	 * to avoid hitting stupid limitations on record size within some
+	 * tools (e.g. flamethrower). */
+	if (max_total_write_size >= FLUSH_HTTP_WRITE_BUFFER_AFTER) {
+		if (session->pending_write_data == NULL) {
+			isc_buffer_allocate(session->mctx,
+					    &session->pending_write_data,
+					    max_total_write_size);
+		}
+
+		isc_buffer_putmem(session->pending_write_data, prepared_data,
+				  total);
+
+		total = max_total_write_size;
+		prepared_data = isc_buffer_base(session->pending_write_data);
+	} else if (session->sending > 0 && total > 0) {
+		if (cb != NULL) {
+			isc__nm_uvreq_t *newcb = isc__nm_uvreq_get(
+				httphandle->sock->mgr, httphandle->sock);
+
+			INSIST(total != 0);
+			INSIST(VALID_NMHANDLE(httphandle));
+			newcb->cb.send = cb;
+			newcb->cbarg = cbarg;
+			isc_nmhandle_attach(httphandle, &newcb->handle);
+			ISC_LIST_APPEND(session->pending_write_callbacks, newcb,
+					link);
+		}
+
+		if (session->pending_write_data == NULL) {
+			isc_buffer_allocate(session->mctx,
+					    &session->pending_write_data,
+					    total);
+			isc_buffer_setautorealloc(session->pending_write_data,
+						  true);
+		}
+
+		isc_buffer_putmem(session->pending_write_data, prepared_data,
+				  total);
+		return (false);
+	} else if (session->sending == 0 && total == 0 &&
+		   session->pending_write_data &&
+		   (total = isc_buffer_usedlength(
+			    session->pending_write_data)) > 0)
+	{
+		isc_region_t region = { 0 };
+		INSIST(prepared_data == &tmp_data[0]);
+		isc_buffer_usedregion(session->pending_write_data, &region);
+		INSIST(total == region.length);
+		prepared_data = region.base;
+	}
+#endif /* ENABLE_HTTP_WRITE_BUFFERING */
+
+	if (total == 0) {
+		INSIST(prepared_data == &tmp_data[0]);
 		/* No data returned */
 		return (false);
 	}
 
 	send = isc_mem_get(session->mctx, sizeof(*send));
-	*send = (isc_http_send_req_t){
-		.data.base = isc_mem_get(session->mctx, pending),
-		.data.length = pending,
-	};
-	memmove(send->data.base, data, pending);
+	if (prepared_data == &tmp_data[0]) {
+		*send = (isc_http_send_req_t){
+			.data.base = isc_mem_get(session->mctx, total),
+			.data.length = total,
+		};
+		memmove(send->data.base, tmp_data, total);
+	} else if (session->pending_write_data != NULL) {
+		*send = (isc_http_send_req_t){
+			.data.base = isc_mem_get(session->mctx, total),
+			.data.length = total,
+		};
+		memmove(send->data.base,
+			isc_buffer_base(session->pending_write_data), total);
+		isc_buffer_free(&session->pending_write_data);
+	} else {
+		*send = (isc_http_send_req_t){
+			.data.base = prepared_data,
+			.data.length = total,
+		};
+	}
 	isc_nmhandle_attach(session->handle, &send->transphandle);
 	isc__nm_httpsession_attach(session, &send->session);
 
@@ -955,6 +1116,8 @@ http_send_outgoing(isc_nm_http_session_t *session, isc_nmhandle_t *httphandle,
 		isc_nmhandle_attach(httphandle, &send->httphandle);
 	}
 
+	move_pending_send_callbacks(session, send);
+
 	session->sending++;
 	isc_nm_send(session->handle, &send->data, http_writecb, send);
 	return (true);
@@ -975,8 +1138,9 @@ http_do_bio(isc_nm_http_session_t *session, isc_nmhandle_t *send_httphandle,
 			finish_http_session(session);
 		}
 		return;
-	} else if ((nghttp2_session_want_read(session->ngsession) == 0 &&
-		    nghttp2_session_want_write(session->ngsession) == 0))
+	} else if (nghttp2_session_want_read(session->ngsession) == 0 &&
+		   nghttp2_session_want_write(session->ngsession) == 0 &&
+		   session->pending_write_data == NULL)
 	{
 		session->closing = true;
 		return;