Polish up bo->state before we unleash streaming

bin/varnishd/cache/cache.h

@@ -490,10 +490,10 @@ oc_getlru(const struct objcore *oc)
  */
 
 enum busyobj_state_e {
-	BOS_INVALID = 0,
-	BOS_FETCHING,
-	BOS_FAILED,
-	BOS_FINISHED
+	BOS_INVALID = 0,	/* don't touch (yet) */
+	BOS_FETCHING,		/* beresp.* can be examined */
+	BOS_FINISHED,		/* object is complete */
+	BOS_FAILED,		/* something went wrong */
 };
 
 struct busyobj {
@@ -797,6 +797,9 @@ struct busyobj *VBO_GetBusyObj(struct worker *, struct req *);
 void VBO_DerefBusyObj(struct worker *wrk, struct busyobj **busyobj);
 void VBO_Free(struct busyobj **vbo);
 void VBO_extend(const struct busyobj *, ssize_t);
+void VBO_setstate(struct busyobj *bo, enum busyobj_state_e next);
+void VBO_waitstate(struct busyobj *bo, enum busyobj_state_e want);
+
 
 /* cache_http1_fetch.c [V1F] */
 int V1F_fetch_hdr(struct worker *wrk, struct busyobj *bo, struct req *req);

bin/varnishd/cache/cache_busyobj.c

@@ -222,3 +222,26 @@ VBO_extend(const struct busyobj *bo, ssize_t l)
 	assert(l > 0);
 	bo->fetch_obj->len += l;
 }
+
+void
+VBO_setstate(struct busyobj *bo, enum busyobj_state_e next)
+{
+	Lck_Lock(&bo->mtx);
+	VSLb(bo->vsl, SLT_Debug, "XXX: %d -> %d", bo->state, next);
+	assert(next > bo->state);
+	bo->state = next;
+	AZ(pthread_cond_signal(&bo->cond));
+	Lck_Unlock(&bo->mtx);
+}
+
+void
+VBO_waitstate(struct busyobj *bo, enum busyobj_state_e want)
+{
+	Lck_Lock(&bo->mtx);
+	while (1) {
+		if (bo->state >= want)
+			break;
+		(void)Lck_CondWait(&bo->cond, &bo->mtx, NULL);
+	}
+	Lck_Unlock(&bo->mtx);
+}

bin/varnishd/cache/cache_fetch.c

@@ -117,8 +117,10 @@ vbf_stp_startfetch(struct worker *wrk, struct busyobj *bo)
 
 	http_PrintfHeader(bo->bereq,
 	    "X-Varnish: %u", bo->vsl->wid & VSL_IDENTMASK);
-	if (wrk->handling == VCL_RET_ABANDON)
+	if (wrk->handling == VCL_RET_ABANDON) {
+		VBO_setstate(bo, BOS_FAILED);
 		return (F_STP_ABANDON);
+	}
 	assert (wrk->handling == VCL_RET_FETCH);
 	return (F_STP_FETCHHDR);
 }
@@ -141,6 +143,8 @@ vbf_stp_fetchhdr(struct worker *wrk, struct busyobj *bo)
 	if (!bo->do_pass)
 		vbf_release_req(bo); /* XXX: retry ?? */
 
+	assert(bo->state == BOS_INVALID);
+
 	i = V1F_fetch_hdr(wrk, bo, bo->req);
 	/*
 	 * If we recycle a backend connection, there is a finite chance
@@ -204,6 +208,7 @@ vbf_stp_fetchhdr(struct worker *wrk, struct busyobj *bo)
 	if (wrk->handling == VCL_RET_DELIVER)
 		return (F_STP_FETCH);
 	if (wrk->handling == VCL_RET_RETRY) {
+		assert(bo->state == BOS_INVALID);
 		bo->retries++;
 		if (bo->retries <= cache_param->max_retries) {
 			VDI_CloseFd(&bo->vbc);
@@ -352,6 +357,7 @@ vbf_stp_fetch(struct worker *wrk, struct busyobj *bo)
 		AZ(HSH_Deref(&wrk->stats, bo->fetch_objcore, NULL));
 		bo->fetch_objcore = NULL;
 		VDI_CloseFd(&bo->vbc);
+		VBO_setstate(bo, BOS_FAILED);
 		return (F_STP_ABANDON);
 	}
 	CHECK_OBJ_NOTNULL(obj, OBJECT_MAGIC);
@@ -407,6 +413,9 @@ vbf_stp_fetch(struct worker *wrk, struct busyobj *bo)
 	if (bo->vfp == NULL)
 		bo->vfp = &VFP_nop;
 
+	assert(bo->state == BOS_INVALID);
+	VBO_setstate(bo, BOS_FETCHING);
+
 	V1F_fetch_body(wrk, bo);
 
 	assert(bo->refcount >= 1);
@@ -419,6 +428,7 @@ vbf_stp_fetch(struct worker *wrk, struct busyobj *bo)
 		(void)HSH_Deref(&wrk->stats, NULL, &obj);
 		return (F_STP_ABANDON);
 	}
+	VBO_setstate(bo, BOS_FINISHED);
 
 	VBO_DerefBusyObj(wrk, &bo);	// XXX ?
 	return (F_STP_DONE);
@@ -433,7 +443,7 @@ vbf_stp_abandon(struct worker *wrk, struct busyobj *bo)
 	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
 	CHECK_OBJ_NOTNULL(bo, BUSYOBJ_MAGIC);
 
-	bo->state = BOS_FAILED;
+	assert(bo->state == BOS_FAILED);
 	vbf_release_req(bo);
 	VBO_DerefBusyObj(wrk, &bo);	// XXX ?
 	return (F_STP_DONE);

bin/varnishd/cache/cache_fetch_proc.c

@@ -57,7 +57,7 @@ VFP_Error2(struct busyobj *bo, const char *error, const char *more)
 {
 
 	CHECK_OBJ_NOTNULL(bo, BUSYOBJ_MAGIC);
-	if (bo->state == BOS_FETCHING) {
+	if (bo->state < BOS_FAILED) {
 		if (more == NULL)
 			VSLb(bo->vsl, SLT_FetchError, "%s", error);
 		else

bin/varnishd/cache/cache_http1_fetch.c

@@ -324,7 +324,7 @@ V1F_fetch_body(struct worker *wrk, struct busyobj *bo)
 	CHECK_OBJ_NOTNULL(obj, OBJECT_MAGIC);
 	CHECK_OBJ_NOTNULL(obj->http, HTTP_MAGIC);
 
-	assert(bo->state == BOS_INVALID);
+	assert(bo->state == BOS_FETCHING);
 
 	/*
 	 * XXX: The busyobj needs a dstat, but it is not obvious which one
@@ -337,8 +337,6 @@ V1F_fetch_body(struct worker *wrk, struct busyobj *bo)
 	AZ(bo->vgz_rx);
 	AZ(VTAILQ_FIRST(&obj->store));
 
-	bo->state = BOS_FETCHING;
-
 	/* XXX: pick up estimate from objdr ? */
 	cl = 0;
 	cls = bo->should_close;
@@ -441,9 +439,6 @@ V1F_fetch_body(struct worker *wrk, struct busyobj *bo)
 			http_PrintfHeader(obj->http,
 			    "Content-Length: %zd", obj->len);
 		}
-
-		/* XXX: Atomic assignment, needs volatile/membar ? */
-		bo->state = BOS_FINISHED;
 	}
 	if (obj->objcore->objhead != NULL)
 		HSH_Complete(obj->objcore);

bin/varnishd/cache/cache_req_fsm.c

@@ -117,9 +117,8 @@ cnt_stream(struct worker *wrk, struct req *req)
 	}
 	assert(wrk->handling == VCL_RET_DELIVER);
 
-	while (bo->state < BOS_FAILED)
-		(void)usleep(10000);
-	assert(bo->state >= BOS_FAILED);
+	VBO_waitstate(bo, BOS_FINISHED);
+	assert(bo->state >= BOS_FINISHED);
 
 	if (bo->state == BOS_FAILED) {
 		(void)HSH_Deref(&wrk->stats, NULL, &req->obj);
@@ -369,7 +368,7 @@ cnt_fetch(struct worker *wrk, struct req *req)
 	AN(req->busyobj);
 	assert(req->busyobj->refcount > 0);
 	(void)HTTP1_DiscardReqBody(req);
-	while (req->busyobj->state < BOS_FAILED) {
+	while (req->busyobj->state < BOS_FINISHED) {
 		printf("YYY\n");
 		(void)usleep(100000);
 	}