move vcl0 to req->top

Also fix some errors in vcl0 handling:

- Only the top request may release vcl0 because it owns it
- because we can re-embark for ESI, we can not assert that vcl0 is NULL
  in CNT_Embark()

passes tests/r02849.vtc again.

still fails r03003.vtc, which will get fixed in a follow up commit

Fixes #3019 with test case by @Dridi

bin/varnishd/cache/cache.h

@@ -445,6 +445,7 @@ struct reqtop {
 	unsigned		magic;
 #define REQTOP_MAGIC		0x57fbda52
 	struct req		*topreq;
+	struct vcl		*vcl0;
 };
 
 struct req {
@@ -457,7 +458,6 @@ struct req {
 	unsigned		restarts;
 	unsigned		esi_level;
 	struct reqtop		*top;	/* esi_level == 0 request */
-	struct vcl		*vcl0;
 
 #define REQ_FLAG(l, r, w, d) unsigned	l:1;
 #include "tbl/req_flags.h"

bin/varnishd/cache/cache_esi_deliver.c

@@ -172,8 +172,9 @@ ved_include(struct req *preq, const char *src, const char *host,
 	req->req_body_status = REQ_BODY_NONE;
 
 	AZ(req->vcl);
-	if (req->vcl0)
-		req->vcl = req->vcl0;
+	AN(req->top);
+	if (req->top->vcl0)
+		req->vcl = req->top->vcl0;
 	else
 		req->vcl = preq->vcl;
 	VCL_Ref(req->vcl);

bin/varnishd/cache/cache_panic.c

@@ -493,6 +493,7 @@ pan_top(struct vsb *vsb, const struct reqtop *top)
 		return;
 	VSB_indent(vsb, 2);
 	pan_req(vsb, top->topreq);
+	VCL_Panic(vsb, "vcl0", top->vcl0);
 	VSB_indent(vsb, -2);
 	VSB_cat(vsb, "},\n");
 }
@@ -559,7 +560,6 @@ pan_req(struct vsb *vsb, const struct req *req)
 		pan_vdp(vsb, req->vdc);
 
 	VCL_Panic(vsb, "vcl", req->vcl);
-	VCL_Panic(vsb, "vcl0", req->vcl0);
 
 	if (req->body_oc != NULL)
 		pan_objcore(vsb, "BODY", req->body_oc);

bin/varnishd/cache/cache_req.c

@@ -207,7 +207,8 @@ Req_Cleanup(struct sess *sp, struct worker *wrk, struct req *req)
 	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
 	CHECK_OBJ_NOTNULL(req, REQ_MAGIC);
 	assert(sp == req->sp);
-	AZ(req->vcl0);
+	if (IS_TOPREQ(req))
+		AZ(req->top->vcl0);
 
 	req->director_hint = NULL;
 	req->restarts = 0;

bin/varnishd/cache/cache_req_fsm.c

@@ -1052,7 +1052,6 @@ CNT_Embark(struct worker *wrk, struct req *req)
 		VSLb(req->vsl, SLT_VCL_use, "%s", VCL_Name(req->vcl));
 	}
 
-	AZ(req->vcl0);
 	AN(req->vcl);
 }
 
@@ -1063,7 +1062,6 @@ CNT_Request(struct req *req)
 	enum req_fsm_nxt nxt;
 
 	CHECK_OBJ_NOTNULL(req, REQ_MAGIC);
-	assert(IS_TOPREQ(req) || req->vcl0 == NULL);
 
 	wrk = req->wrk;
 	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
@@ -1106,8 +1104,8 @@ CNT_Request(struct req *req)
 	}
 	wrk->vsl = NULL;
 	if (nxt == REQ_FSM_DONE) {
-		if (IS_TOPREQ(req) && req->vcl0 != NULL)
-			VCL_Rel(&req->vcl0);
+		if (IS_TOPREQ(req) && req->top->vcl0 != NULL)
+			VCL_Rel(&req->top->vcl0);
 		VCL_TaskLeave(req->vcl, req->privs);
 		AN(req->vsl->wid);
 		VRB_Free(req);

bin/varnishd/cache/cache_vpi.c

@@ -91,12 +91,14 @@ VPI_vcl_select(VRT_CTX, VCL_VCL vcl)
 
 	CHECK_OBJ_NOTNULL(vcl, VCL_MAGIC);
 
-	if (IS_TOPREQ(req) && req->vcl0 != NULL)
+	if (IS_TOPREQ(req) && req->top->vcl0 != NULL)
 		return;		// Illegal, req-FSM will fail this later.
 
 	VCL_TaskLeave(req->vcl, req->privs);
 	if (IS_TOPREQ(req)) {
-		req->vcl0 = req->vcl;
+		AN(req->top);
+		AZ(req->top->vcl0);
+		req->top->vcl0 = req->vcl;
 		req->vcl = NULL;
 	} else {
 		VCL_Rel(&req->vcl);

bin/varnishd/http1/cache_http1_fsm.c

@@ -414,7 +414,7 @@ HTTP1_Session(struct worker *wrk, struct req *req)
 				VCL_TaskEnter(req->vcl, req->privs);
 			if (CNT_Request(req) == REQ_FSM_DISEMBARK)
 				return;
-			AZ(req->vcl0);
+			AZ(req->top->vcl0);
 			req->task.func = NULL;
 			req->task.priv = NULL;
 			AZ(req->ws->r);

bin/varnishd/http2/cache_http2_proto.c

@@ -529,7 +529,7 @@ h2_do_req(struct worker *wrk, void *priv)
 	wrk->stats->client_req++;
 	if (CNT_Request(req) != REQ_FSM_DISEMBARK) {
 		AZ(req->ws->r);
-		AZ(req->vcl0);
+		AZ(req->top->vcl0);
 		h2 = r2->h2sess;
 		CHECK_OBJ_NOTNULL(h2, H2_SESS_MAGIC);
 		Lck_Lock(&h2->sess->mtx);

bin/varnishtest/tests/r03019.vtc

+varnishtest "return(vcl) then reembark"
+
+barrier b1 cond 2
+
+server s1 {
+	rxreq
+	barrier b1 sync
+	txresp
+} -start
+
+varnish v1 -vcl+backend ""
+varnish v1 -cliok "vcl.label lbl vcl1"
+varnish v1 -vcl {
+	backend be { .host = "${bad_backend}"; }
+
+	sub vcl_recv {
+		return (vcl(lbl));
+	}
+} -start
+
+client c1 {
+	txreq
+	rxresp
+	expect resp.status == 200
+} -start
+
+client c2 {
+	barrier b1 sync
+	txreq
+	rxresp
+	expect resp.status == 200
+} -start
+
+client c1 -wait
+client c2 -wait