Unbusy the objectcore when we start fetching the body.

Don't deref the objcore until the busyobj dies.

Apply two workarounds for now:  Don't return non-busy object with
a busyobj in HSH_Lookup(), and trade the busyobj ref for an objcore ref
in cnt_fetchbody()

Getting closer...

bin/varnishd/cache/cache_busyobj.c

@@ -38,6 +38,8 @@
 
 #include "cache.h"
 
+#include "hash/hash_slinger.h"
+
 static struct mempool		*vbopool;
 
 /*--------------------------------------------------------------------
@@ -157,6 +159,7 @@ VBO_DerefBusyObj(struct worker *wrk, struct busyobj **pbo)
 	bo = *pbo;
 	*pbo = NULL;
 	CHECK_OBJ_NOTNULL(bo, BUSYOBJ_MAGIC);
+	CHECK_OBJ_ORNULL(bo->fetch_obj, OBJECT_MAGIC);
 	Lck_Lock(&bo->mtx);
 	assert(bo->refcount > 0);
 	r = --bo->refcount;
@@ -167,6 +170,11 @@ VBO_DerefBusyObj(struct worker *wrk, struct busyobj **pbo)
 
 	VSL_Flush(bo->vsl, 0);
 
+	if (bo->fetch_obj != NULL && bo->fetch_obj->objcore != NULL) {
+		AN(wrk);
+		(void)HSH_Deref(&wrk->stats, NULL, &bo->fetch_obj);
+	}
+
 	memset(&bo->refcount, 0,
 	    sizeof *bo - offsetof(struct busyobj, refcount));
 

bin/varnishd/cache/cache_center.c

@@ -906,6 +906,13 @@ cnt_fetchbody(struct sess *sp, struct worker *wrk, struct req *req)
 
 	assert(bo->refcount == 2);	/* one for each thread */
 
+	if (req->obj->objcore != NULL) {
+		EXP_Insert(req->obj);
+		AN(req->obj->objcore->ban);
+		AZ(req->obj->ws_o->overflow);
+		HSH_Unbusy(&wrk->stats, req->obj->objcore);
+	}
+
 	if (Pool_Task(wrk->pool, &bo->fetch_task, POOL_NO_QUEUE))
 		FetchBody(wrk, bo);
 
@@ -923,6 +930,9 @@ cnt_fetchbody(struct sess *sp, struct worker *wrk, struct req *req)
 		return (0);
 	}
 
+	if (req->obj->objcore != NULL)
+		HSH_Ref(req->obj->objcore);
+
 	VBO_DerefBusyObj(wrk, &req->busyobj);
 	wrk->acct_tmp.fetch++;
 	sp->step = STP_PREPRESP;

bin/varnishd/cache/cache_fetch.c

@@ -35,8 +35,6 @@
 
 #include "cache.h"
 
-#include "hash/hash_slinger.h"
-
 #include "cache_backend.h"
 #include "vcli_priv.h"
 #include "vct.h"
@@ -557,7 +555,6 @@ FetchBody(struct worker *wrk, void *priv)
 	CHECK_OBJ_NOTNULL(obj->http, HTTP_MAGIC);
 
 	assert(bo->state == BOS_INVALID);
-	bo->state = BOS_FINISHED;
 
 	/*
 	 * XXX: The busyobj needs a dstat, but it is not obvious which one
@@ -638,8 +635,8 @@ FetchBody(struct worker *wrk, void *priv)
 	if (bo->state == BOS_FAILED) {
 		wrk->stats.fetch_failed++;
 		VDI_CloseFd(&bo->vbc);
+		obj->exp.ttl = -1.;
 		obj->len = 0;
-		HSH_Drop(wrk, &obj);
 	} else {
 		assert(bo->state == BOS_FETCHING);
 
@@ -669,24 +666,19 @@ FetchBody(struct worker *wrk, void *priv)
 			    "Content-Length: %zd", obj->len);
 		}
 
-
 		if (cls)
 			VDI_CloseFd(&bo->vbc);
 		else
 			VDI_RecycleFd(&bo->vbc);
 
-		if (obj->objcore != NULL) {
-			EXP_Insert(obj);
-			AN(obj->objcore->ban);
-			AZ(obj->ws_o->overflow);
-			HSH_Unbusy(&wrk->stats, obj->objcore);
-			obj->objcore->busyobj = NULL;
-			VMB();
-		}
 
 		/* XXX: Atomic assignment, needs volatile/membar ? */
 		bo->state = BOS_FINISHED;
-
+	}
+	if (obj->objcore != NULL) {
+		VMB();
+		obj->objcore->busyobj = NULL;
+		VMB();
 	}
 	bo->stats = NULL;
 	VBO_DerefBusyObj(wrk, &bo);

bin/varnishd/cache/cache_hash.c

@@ -328,7 +328,7 @@ HSH_Lookup(struct sess *sp)
 		CHECK_OBJ_NOTNULL(oc, OBJCORE_MAGIC);
 		assert(oc->objhead == oh);
 
-		if (oc->flags & OC_F_BUSY) {
+		if (oc->flags & OC_F_BUSY || oc->busyobj != NULL) {
 			CHECK_OBJ_ORNULL(oc->busyobj, BUSYOBJ_MAGIC);
 			if (req->hash_ignore_busy)
 				continue;
@@ -379,7 +379,7 @@ HSH_Lookup(struct sess *sp)
 	AZ(req->objcore);
 	if (oc == NULL			/* We found no live object */
 	    && grace_oc != NULL		/* There is a grace candidate */
-	    && (busy_found 		/* Somebody else is already busy */
+	    && (busy_found		/* Somebody else is already busy */
 	    || !VDI_Healthy(req->director, sp))) {
 					/* Or it is impossible to fetch */
 		o = oc_getobj(&wrk->stats, grace_oc);
@@ -431,7 +431,7 @@ HSH_Lookup(struct sess *sp)
 	oc = wrk->nobjcore;
 	wrk->nobjcore = NULL;
 	AN(oc->flags & OC_F_BUSY);
-	oc->refcnt = 1;
+	oc->refcnt = 1;		/* Owned by busyobj */
 	oc->objhead = oh;
 	VTAILQ_INSERT_TAIL(&oh->objcs, oc, list);
 	/* NB: do not deref objhead the new object inherits our reference */
@@ -440,7 +440,7 @@ HSH_Lookup(struct sess *sp)
 	AZ(req->busyobj);
 	req->busyobj = VBO_GetBusyObj(wrk);
 	req->busyobj->vsl->wid = sp->vsl_id;
-	req->busyobj->refcount = 2;	/* One for headers, one for body*/
+	req->busyobj->refcount = 2;	/* One for req, one for FetchBody */
 
 	VRY_Validate(req->vary_b);
 	if (req->vary_l != NULL)