Further flesh out the WRW chunked support

bin/varnishd/cache.h

@@ -790,6 +790,7 @@ void WRK_SumStat(struct worker *w);
 
 #define WRW_IsReleased(w)	((w)->wrw.wfd == NULL)
 void WRW_Chunked(struct worker *w);
+void WRW_EndChunk(struct worker *w);
 void WRW_Reserve(struct worker *w, int *fd);
 unsigned WRW_Flush(struct worker *w);
 unsigned WRW_FlushRelease(struct worker *w);

bin/varnishd/cache_center.c

@@ -241,6 +241,7 @@ cnt_deliver(struct sess *sp)
 	RES_WriteObj(sp);
 
 	assert(WRW_IsReleased(sp->wrk));
+	assert(sp->wrk->wrw.ciov == sp->wrk->wrw.siov);
 	(void)HSH_Deref(sp->wrk, NULL, &sp->obj);
 	http_Setup(sp->wrk->resp, NULL);
 	sp->step = STP_DONE;
@@ -1153,6 +1154,7 @@ cnt_recv(struct sess *sp)
 	CHECK_OBJ_NOTNULL(sp, SESS_MAGIC);
 	CHECK_OBJ_NOTNULL(sp->vcl, VCL_CONF_MAGIC);
 	AZ(sp->obj);
+	assert(sp->wrk->wrw.ciov == sp->wrk->wrw.siov);
 
 	/* By default we use the first backend */
 	AZ(sp->director);

bin/varnishd/cache_wrw.c

@@ -115,9 +115,9 @@ WRW_Flush(struct worker *w)
 	if (wrw->ciov < wrw->siov)
 		assert(wrw->niov < wrw->siov);
 
-	if (*wrw->wfd >= 0 && wrw->niov > 0 && wrw->werr == 0) {
-		if (wrw->ciov < wrw->siov && wrw->liov > 0) {
-			bprintf(cbuf, "%jx\r\n", (intmax_t)wrw->cliov);
+	if (*wrw->wfd >= 0 && wrw->liov > 0 && wrw->werr == 0) {
+		if (wrw->ciov < wrw->siov && wrw->cliov > 0) {
+			bprintf(cbuf, "00%jx\r\n", (intmax_t)wrw->cliov);
 			i = strlen(cbuf);
 			wrw->iov[wrw->ciov].iov_base = cbuf;
 			wrw->iov[wrw->ciov].iov_len = i;
@@ -126,6 +126,9 @@ WRW_Flush(struct worker *w)
 			wrw->iov[wrw->niov].iov_base = cbuf + i - 2;
 			wrw->iov[wrw->niov++].iov_len = 2;
 			wrw->liov += 2;
+		} else if (wrw->ciov < wrw->siov) {
+			wrw->iov[wrw->ciov].iov_base = cbuf;
+			wrw->iov[wrw->ciov].iov_len = 0;
 		}
 		i = writev(*wrw->wfd, wrw->iov, wrw->niov);
 		if (i != wrw->liov) {
@@ -211,8 +214,24 @@ WRW_Chunked(struct worker *w)
 	if (wrw->niov + 3 >= wrw->siov)
 		(void)WRW_Flush(w);
 	wrw->ciov = wrw->niov++;
-	wrw->cliov = wrw->liov;
+	wrw->cliov = 0;
+	assert(wrw->ciov < wrw->siov);
+}
+
+void
+WRW_EndChunk(struct worker *w)
+{
+	struct wrw *wrw;
+
+	CHECK_OBJ_NOTNULL(w, WORKER_MAGIC);
+	wrw = &w->wrw;
+
 	assert(wrw->ciov < wrw->siov);
+	(void)WRW_Flush(w);
+	wrw->ciov = wrw->siov;
+	(void)WRW_Flush(w);
+	wrw->cliov = 0;
+	WRW_Write(w, "0\r\n\r\n", -1);
 }