[master] 57c814de0 ws: Generalize WS_Pipeline() beyond req keep-alive

[Dridi Boukelmoune]

commit 57c814de07e875af8d8ab7d55a7e873b6b3e66f4
Author: Dridi Boukelmoune <dridi.boukelmoune at gmail.com>
Date:   Mon Jun 17 10:50:50 2024 +0200

    ws: Generalize WS_Pipeline() beyond req keep-alive
    
    This puts the caller in charge of triggering the workspace rollback
    before moving pipelined bytes around. This will open the door to
    pipelining for req and beresp trailers.
    
    The workspace emulator version of WS_ReqPipeline() had a systematic
    rollback instead of a conditional one. The reason why this didn't pose
    a problem was that fetch tasks never need pipelining and always take
    the shortcut where the rollback was properly optional.
    
    This is a good occasion to simplify the emulated variant and document
    why it is so different from the original.
    
    Co-authored-by: Walid Boudebouda <walid.boudebouda at gmail.com>

diff --git a/bin/varnishd/cache/cache_session.c b/bin/varnishd/cache/cache_session.c
index 43d5ab60e..898242da3 100644
--- a/bin/varnishd/cache/cache_session.c
+++ b/bin/varnishd/cache/cache_session.c
@@ -258,12 +258,20 @@ HTC_Status(enum htc_status_e e, const char **name, const char **desc)
 void
 HTC_RxInit(struct http_conn *htc, struct ws *ws)
 {
-	unsigned l;
+	unsigned rollback;
+	int l;
 
 	CHECK_OBJ_NOTNULL(htc, HTTP_CONN_MAGIC);
 	htc->ws = ws;
 
-	l = WS_ReqPipeline(htc->ws, htc->pipeline_b, htc->pipeline_e);
+	/* 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);
+	xxxassert(l >= 0);
+
 	htc->rxbuf_b = WS_Reservation(ws);
 	htc->rxbuf_e = htc->rxbuf_b + l;
 	htc->pipeline_b = NULL;
diff --git a/bin/varnishd/cache/cache_varnishd.h b/bin/varnishd/cache/cache_varnishd.h
index 9bd5e2490..2469cdcec 100644
--- a/bin/varnishd/cache/cache_varnishd.h
+++ b/bin/varnishd/cache/cache_varnishd.h
@@ -555,7 +555,7 @@ WS_IsReserved(const struct ws *ws)
 
 void *WS_AtOffset(const struct ws *ws, unsigned off, unsigned len);
 unsigned WS_ReservationOffset(const struct ws *ws);
-unsigned WS_ReqPipeline(struct ws *, const void *b, const void *e);
+int WS_Pipeline(struct ws *, const void *b, const void *e, unsigned rollback);
 
 /* cache_ws_common.c */
 void WS_Id(const struct ws *ws, char *id);
diff --git a/bin/varnishd/cache/cache_ws.c b/bin/varnishd/cache/cache_ws.c
index d6ff67370..3f2cc5309 100644
--- a/bin/varnishd/cache/cache_ws.c
+++ b/bin/varnishd/cache/cache_ws.c
@@ -135,17 +135,15 @@ WS_Reset(struct ws *ws, uintptr_t pp)
  * may not originate from the same workspace.
  */
 
-unsigned
-WS_ReqPipeline(struct ws *ws, const void *b, const void *e)
+int
+WS_Pipeline(struct ws *ws, const void *b, const void *e, unsigned rollback)
 {
 	unsigned r, l;
 
 	WS_Assert(ws);
 
-	if (!strcasecmp(ws->id, "req"))
+	if (rollback)
 		WS_Rollback(ws, 0);
-	else
-		AZ(b);
 
 	r = WS_ReserveAll(ws);
 
@@ -156,7 +154,8 @@ WS_ReqPipeline(struct ws *ws, const void *b, const void *e)
 
 	AN(e);
 	l = pdiff(b, e);
-	assert(l <= r);
+	if (l > r)
+		return (-1);
 	memmove(ws->f, b, l);
 	return (l);
 }
diff --git a/bin/varnishd/cache/cache_ws_emu.c b/bin/varnishd/cache/cache_ws_emu.c
index 830f8c450..767839d1e 100644
--- a/bin/varnishd/cache/cache_ws_emu.c
+++ b/bin/varnishd/cache/cache_ws_emu.c
@@ -221,45 +221,46 @@ WS_Reset(struct ws *ws, uintptr_t pp)
 	WS_Assert(ws);
 }
 
-unsigned
-WS_ReqPipeline(struct ws *ws, const void *b, const void *e)
+int
+WS_Pipeline(struct ws *ws, const void *b, const void *e, unsigned rollback)
 {
-	struct ws_emu *we;
-	struct ws_alloc *wa;
-	unsigned l;
+	void *tmp;
+	unsigned r, l;
 
 	WS_Assert(ws);
 	AZ(ws->f);
 	AZ(ws->r);
 
-	if (strcasecmp(ws->id, "req"))
-		AZ(b);
-
-	if (b == NULL) {
+	/* NB: the pipeline cannot be moved if it comes from the same
+	 * workspace because a rollback would free the memory. This is
+	 * emulated with two copies instead.
+	 */
+
+	if (b != NULL) {
+		AN(e);
+		l = pdiff(b, e);
+		tmp = malloc(l);
+		AN(tmp);
+		memcpy(tmp, b, l);
+	} else {
 		AZ(e);
-		if (!strcasecmp(ws->id, "req"))
-			WS_Rollback(ws, 0);
-		(void)WS_ReserveAll(ws);
-		return (0);
+		l = 0;
+		tmp = NULL;
 	}
 
-	we = ws_emu(ws);
-	ALLOC_OBJ(wa, WS_ALLOC_MAGIC);
-	AN(wa);
-	wa->len = we->len;
-	wa->ptr = malloc(wa->len);
-	AN(wa->ptr);
+	if (rollback)
+		WS_Rollback(ws, 0);
 
-	AN(e);
-	l = pdiff(b, e);
-	assert(l <= wa->len);
-	memcpy(wa->ptr, b, l);
+	r = WS_ReserveAll(ws);
 
-	WS_Rollback(ws, 0);
-	ws->f = wa->ptr;
-	ws->r = ws->f + wa->len;
-	VTAILQ_INSERT_TAIL(&we->head, wa, list);
-	WS_Assert(ws);
+	if (l > r) {
+		free(tmp);
+		return (-1);
+	}
+
+	if (l > 0)
+		memcpy(ws->f, tmp, l);
+	free(tmp);
 	return (l);
 }