move do_stream from worker to busyobj

bin/varnishd/cache/cache.h

@@ -332,7 +332,6 @@ struct worker {
 	struct http		*beresp;
 	struct vbc		*vbc;
 
-	unsigned		do_stream;
 	unsigned		do_close;
 	char			*h_content_length;
 
@@ -508,6 +507,7 @@ struct busyobj {
 	unsigned		do_esi;
 	unsigned		do_gzip;
 	unsigned		do_gunzip;
+	unsigned		do_stream;
 };
 
 /* Object structure --------------------------------------------------*/

bin/varnishd/cache/cache_center.c

@@ -179,7 +179,7 @@ cnt_prepresp(struct sess *sp)
 	CHECK_OBJ_NOTNULL(wrk->obj, OBJECT_MAGIC);
 	CHECK_OBJ_NOTNULL(sp->vcl, VCL_CONF_MAGIC);
 
-	if (wrk->do_stream)
+	if (wrk->busyobj != NULL && wrk->busyobj->do_stream)
 		AssertObjCorePassOrBusy(wrk->obj->objcore);
 
 	wrk->res_mode = 0;
@@ -188,7 +188,7 @@ cnt_prepresp(struct sess *sp)
 		wrk->res_mode |= RES_LEN;
 
 	if (wrk->busyobj != NULL &&
-	    (wrk->h_content_length != NULL || !wrk->do_stream) &&
+	    (wrk->h_content_length != NULL || !wrk->busyobj->do_stream) &&
 	    !wrk->busyobj->do_gzip && !wrk->busyobj->do_gunzip)
 		wrk->res_mode |= RES_LEN;
 
@@ -214,7 +214,8 @@ cnt_prepresp(struct sess *sp)
 	}
 
 	if (!(wrk->res_mode & (RES_LEN|RES_CHUNKED|RES_EOF))) {
-		if (wrk->obj->len == 0 && !wrk->do_stream)
+		if (wrk->obj->len == 0 &&
+		    (wrk->busyobj == NULL || !wrk->busyobj->do_stream))
 			/*
 			 * If the object is empty, neither ESI nor GUNZIP
 			 * can make it any different size
@@ -246,7 +247,7 @@ cnt_prepresp(struct sess *sp)
 	case VCL_RET_RESTART:
 		if (sp->restarts >= cache_param->max_restarts)
 			break;
-		if (wrk->do_stream) {
+		if (wrk->busyobj->do_stream) {
 			VDI_CloseFd(wrk);
 			HSH_Drop(wrk);
 		} else {
@@ -264,7 +265,7 @@ cnt_prepresp(struct sess *sp)
 	default:
 		WRONG("Illegal action in vcl_deliver{}");
 	}
-	if (wrk->do_stream) {
+	if (wrk->busyobj != NULL && wrk->busyobj->do_stream) {
 		AssertObjCorePassOrBusy(wrk->obj->objcore);
 		sp->step = STP_STREAMBODY;
 	} else {
@@ -343,8 +344,6 @@ cnt_done(struct sess *sp)
 
 	wrk->busyobj = NULL;
 
-	wrk->do_stream = 0;
-
 	SES_Charge(sp);
 
 	/* If we did an ESI include, don't mess up our state */
@@ -458,8 +457,6 @@ cnt_error(struct sess *sp)
 	wrk = sp->wrk;
 	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
 
-	wrk->do_stream = 0;
-
 	if (wrk->obj == NULL) {
 		HSH_Prealloc(sp);
 		New_BusyObj(wrk);
@@ -780,9 +777,9 @@ cnt_fetchbody(struct sess *sp)
 		wrk->busyobj->vfp = &vfp_testgzip;
 
 	if (wrk->busyobj->do_esi || sp->esi_level > 0)
-		wrk->do_stream = 0;
+		wrk->busyobj->do_stream = 0;
 	if (!sp->wantbody)
-		wrk->do_stream = 0;
+		wrk->busyobj->do_stream = 0;
 
 	l = http_EstimateWS(wrk->beresp,
 	    pass ? HTTPH_R_PASS : HTTPH_A_INS, &nhttp);
@@ -871,11 +868,11 @@ cnt_fetchbody(struct sess *sp)
 	if (wrk->obj->response == 200 &&
 	    sp->http->conds &&
 	    RFC2616_Do_Cond(sp))
-		wrk->do_stream = 0;
+		wrk->busyobj->do_stream = 0;
 
 	AssertObjCorePassOrBusy(wrk->obj->objcore);
 
-	if (wrk->do_stream) {
+	if (wrk->busyobj->do_stream) {
 		sp->step = STP_PREPRESP;
 		return (0);
 	}
@@ -1062,8 +1059,6 @@ cnt_hit(struct sess *sp)
 
 	assert(!(wrk->obj->objcore->flags & OC_F_PASS));
 
-	AZ(wrk->do_stream);
-
 	VCL_hit_method(sp);
 
 	if (sp->handling == VCL_RET_DELIVER) {
@@ -1468,9 +1463,6 @@ cnt_recv(struct sess *sp)
 		return (0);
 	}
 
-	/* Zap these, in case we came here through restart */
-	wrk->do_stream = 0;
-
 	if (cache_param->http_gzip_support &&
 	     (recv_handling != VCL_RET_PIPE) &&
 	     (recv_handling != VCL_RET_PASS)) {
@@ -1651,7 +1643,6 @@ CNT_Session(struct sess *sp)
 	    sp->step == STP_LOOKUP ||
 	    sp->step == STP_RECV);
 
-	AZ(wrk->do_stream);
 	AZ(wrk->obj);
 	AZ(wrk->objcore);
 
@@ -1705,7 +1696,6 @@ CNT_Session(struct sess *sp)
 	WSL_Flush(wrk, 0);
 	AZ(wrk->obj);
 	AZ(wrk->objcore);
-	AZ(wrk->do_stream);
 #define ACCT(foo)	AZ(wrk->acct_tmp.foo);
 #include "tbl/acct_fields.h"
 #undef ACCT

bin/varnishd/cache/cache_esi_deliver.c

@@ -91,8 +91,6 @@ ved_include(struct sess *sp, const char *src, const char *host)
 	/* Client content already taken care of */
 	http_Unset(sp->http, H_Content_Length);
 
-	sp->wrk->do_stream = 0;
-
 	sxid = sp->xid;
 	while (1) {
 		sp->wrk = w;

bin/varnishd/cache/cache_fetch.c

@@ -126,7 +126,7 @@ vfp_nop_bytes(struct worker *w, struct http_conn *htc, ssize_t bytes)
 		st->len += wl;
 		w->busyobj->fetch_obj->len += wl;
 		bytes -= wl;
-		if (w->do_stream)
+		if (w->busyobj->do_stream)
 			RES_StreamPoll(w);
 	}
 	return (1);
@@ -601,7 +601,7 @@ FetchBody(struct worker *w, struct object *obj)
 		uu = 0;
 		VTAILQ_FOREACH(st, &obj->store, list)
 			uu += st->len;
-		if (w->do_stream)
+		if (w->busyobj->do_stream)
 			/* Streaming might have started freeing stuff */
 			assert (uu <= obj->len);
 

bin/varnishd/cache/cache_gzip.c

@@ -493,7 +493,7 @@ vfp_gunzip_bytes(struct worker *w, struct http_conn *htc, ssize_t bytes)
 		if (i != VGZ_OK && i != VGZ_END)
 			return(FetchError(w, "Gunzip data error"));
 		w->busyobj->fetch_obj->len += dl;
-		if (w->do_stream)
+		if (w->busyobj->do_stream)
 			RES_StreamPoll(w);
 	}
 	assert(i == Z_OK || i == Z_STREAM_END);
@@ -569,7 +569,7 @@ vfp_gzip_bytes(struct worker *w, struct http_conn *htc, ssize_t bytes)
 		i = VGZ_Gzip(vg, &dp, &dl, VGZ_NORMAL);
 		assert(i == Z_OK);
 		w->busyobj->fetch_obj->len += dl;
-		if (w->do_stream)
+		if (w->busyobj->do_stream)
 			RES_StreamPoll(w);
 	}
 	return (1);
@@ -597,7 +597,7 @@ vfp_gzip_end(struct worker *w)
 		i = VGZ_Gzip(vg, &dp, &dl, VGZ_FINISH);
 		w->busyobj->fetch_obj->len += dl;
 	} while (i != Z_STREAM_END);
-	if (w->do_stream)
+	if (w->busyobj->do_stream)
 		RES_StreamPoll(w);
 	VGZ_UpdateObj(vg, w->busyobj->fetch_obj);
 	if (VGZ_Destroy(&vg, -1) != VGZ_END)
@@ -655,7 +655,7 @@ vfp_testgzip_bytes(struct worker *w, struct http_conn *htc, ssize_t bytes)
 		VGZ_Ibuf(vg, st->ptr + st->len, wl);
 		st->len += wl;
 		w->busyobj->fetch_obj->len += wl;
-		if (w->do_stream)
+		if (w->busyobj->do_stream)
 			RES_StreamPoll(w);
 
 		while (!VGZ_IbufEmpty(vg)) {

bin/varnishd/cache/cache_panic.c

@@ -216,6 +216,7 @@ pan_busyobj(const struct busyobj *bo)
 	if (bo->do_gzip)	VSB_printf(pan_vsp, "    do_gzip\n");
 	if (bo->do_gunzip)	VSB_printf(pan_vsp, "    do_gunzip\n");
 	if (bo->do_esi)		VSB_printf(pan_vsp, "    do_esi\n");
+	if (bo->do_stream)	VSB_printf(pan_vsp, "    do_stream\n");
 	VSB_printf(pan_vsp, "    bodystatus = %d,\n", bo->body_status);
 	VSB_printf(pan_vsp, "    },\n");
 }
@@ -258,7 +259,6 @@ pan_sess(const struct sess *sp)
 	    sp->restarts, sp->esi_level);
 
 	VSB_printf(pan_vsp, "  flags = ");
-	if (sp->wrk->do_stream)	VSB_printf(pan_vsp, " do_stream");
 	if (sp->wrk->do_close)	VSB_printf(pan_vsp, " do_close");
 	VSB_printf(pan_vsp, "\n");
 	pan_busyobj(sp->wrk->busyobj);

bin/varnishd/cache/cache_vrt_var.c

@@ -197,7 +197,7 @@ VRT_r_##dir##_##onm(const struct sess *sp)				\
 VBERESP(beresp, unsigned, do_esi, busyobj->do_esi)
 VBERESP(beresp, unsigned, do_gzip, busyobj->do_gzip)
 VBERESP(beresp, unsigned, do_gunzip, busyobj->do_gunzip)
-VBERESP(beresp, unsigned, do_stream, do_stream)
+VBERESP(beresp, unsigned, do_stream, busyobj->do_stream)
 
 /*--------------------------------------------------------------------*/