Modify the ban_lurker so we lift oc's as far up the ban-list as we

can when we check a batch of lurkable bans.

For obj.* only bans, this should concentrate the entire list at the
very top.

If there are req.* bans, the oc's end up on the bans right below
the req.* bans.

Fixes	#1635

(As much as we can fix it)

bin/varnishd/cache/cache_ban.c

@@ -159,14 +159,16 @@ ban_mark_completed(struct ban *b)
 	Lck_AssertHeld(&ban_mtx);
 
 	AN(b->spec);
-	AZ(b->flags & BANS_FLAG_COMPLETED);
-	ln = ban_len(b->spec);
-	b->flags |= BANS_FLAG_COMPLETED;
-	b->spec[BANS_FLAGS] |= BANS_FLAG_COMPLETED;
-	VWMB();
-	vbe32enc(b->spec + BANS_LENGTH, BANS_HEAD_LEN);
-	VSC_C_main->bans_completed++;
-	VSC_C_main->bans_persisted_fragmentation += ln - ban_len(b->spec);
+	if (!(b->flags & BANS_FLAG_COMPLETED)) {
+		ln = ban_len(b->spec);
+		b->flags |= BANS_FLAG_COMPLETED;
+		b->spec[BANS_FLAGS] |= BANS_FLAG_COMPLETED;
+		VWMB();
+		vbe32enc(b->spec + BANS_LENGTH, BANS_HEAD_LEN);
+		VSC_C_main->bans_completed++;
+		VSC_C_main->bans_persisted_fragmentation +=
+		    ln - ban_len(b->spec);
+	}
 }
 
 /*--------------------------------------------------------------------
@@ -683,7 +685,7 @@ ccf_ban_list(struct cli *cli, const char * const *av, void *priv)
 		o = bl == b ? 1 : 0;
 		VCLI_Out(cli, "%10.6f %5ju %s", ban_time(b->spec),
 		    (intmax_t)b->refcount - o,
-		    b->flags & BANS_FLAG_COMPLETED ? "C" : " ");
+		    b->flags & BANS_FLAG_COMPLETED ? "C" : "-");
 		if (DO_DEBUG(DBG_LURKER)) {
 			VCLI_Out(cli, "%s%s %p ",
 			    b->flags & BANS_FLAG_REQ ? "R" : "-",

bin/varnishd/cache/cache_ban_lurker.c

@@ -112,7 +112,9 @@ ban_lurker_getfirst(struct vsl_log *vsl, struct ban *bt)
 			} else {
 				/*
 				 * We got the lock, and the oc is not being
-				 * dismantled under our feet, run with it...
+				 * dismantled under our feet.
+				 * Take it off the ban and (optimistically)
+				 * put it on the * destination ban
 				 */
 				AZ(oc->flags & OC_F_BUSY);
 				oc->refcnt += 1;
@@ -135,7 +137,7 @@ ban_lurker_getfirst(struct vsl_log *vsl, struct ban *bt)
 
 static void
 ban_lurker_test_ban(struct worker *wrk, struct vsl_log *vsl, struct ban *bt,
-    struct banhead_s *obans)
+    struct banhead_s *obans, struct ban *bd)
 {
 	struct ban *bl, *bln;
 	struct objcore *oc;
@@ -182,6 +184,17 @@ ban_lurker_test_ban(struct worker *wrk, struct vsl_log *vsl, struct ban *bt,
 			EXP_Rearm(oc, oc->exp.t_origin, 0, 0, 0);
 					// XXX ^ fake now
 			VSC_C_main->bans_lurker_obj_killed++;
+		} else {
+			if (oc->ban != bd) {
+				Lck_Lock(&ban_mtx);
+				oc->ban->refcount--;
+				VTAILQ_REMOVE(&oc->ban->objcore, oc, ban_list);
+				oc->ban = bd;
+				bd->refcount++;
+				VTAILQ_INSERT_TAIL(&bd->objcore, oc, ban_list);
+				Lck_Unlock(&ban_mtx);
+				ObjUpdateMeta(wrk, oc);
+			}
 		}
 		(void)HSH_DerefObjCore(wrk, &oc);
 	}
@@ -194,63 +207,43 @@ ban_lurker_test_ban(struct worker *wrk, struct vsl_log *vsl, struct ban *bt,
 static double
 ban_lurker_work(struct worker *wrk, struct vsl_log *vsl)
 {
-	struct ban *b, *bt;
+	struct ban *b, *bd;
 	struct banhead_s obans;
 	double d, dt, n;
-	int i;
 
 	dt = 49.62;		// Random, non-magic
 	if (cache_param->ban_lurker_sleep == 0)
 		return (dt);
-	/* Make a list of the bans we can do something about */
-	VTAILQ_INIT(&obans);
+
 	Lck_Lock(&ban_mtx);
 	b = ban_start;
 	Lck_Unlock(&ban_mtx);
-	i = 0;
 	d = VTIM_real() - cache_param->ban_lurker_age;
+	bd = NULL;
+	VTAILQ_INIT(&obans);
 	for (; b != NULL; b = VTAILQ_NEXT(b, list)) {
+		if (bd != NULL)
+			ban_lurker_test_ban(wrk, vsl, b, &obans, bd);
 		if (b->flags & BANS_FLAG_COMPLETED)
 			continue;
-		if (b->flags & BANS_FLAG_REQ)
+		if (b->flags & BANS_FLAG_REQ) {
+			bd = VTAILQ_NEXT(b, list);
 			continue;
-		if (b == VTAILQ_LAST(&ban_head, banhead_s))
-			continue;	// XXX: why ?
+		}
 		n = ban_time(b->spec) - d;
 		if (n < 0) {
 			VTAILQ_INSERT_TAIL(&obans, b, l_list);
-			i++;
+			if (bd == NULL)
+				bd = b;
 		} else if (n < dt) {
 			dt = n;
 		}
 	}
-	if (DO_DEBUG(DBG_LURKER))
-		VSLb(vsl, SLT_Debug, "lurker: %d actionable bans, dt = %lf", i, dt);
-	if (i == 0)
-		return (dt);
 
-	dt = cache_param->ban_lurker_sleep;
-	/* Go though all the bans to test the objects */
-	VTAILQ_FOREACH_REVERSE(bt, &ban_head, banhead_s, list) {
-		if (bt == VTAILQ_LAST(&obans, banhead_s)) {
-			if (DO_DEBUG(DBG_LURKER))
-				VSLb(vsl, SLT_Debug,
-				    "Lurk bt completed %p", bt);
-			Lck_Lock(&ban_mtx);
-			/* We can be raced by a new ban */
-			if (!(bt->flags & BANS_FLAG_COMPLETED))
-				ban_mark_completed(bt);
-			Lck_Unlock(&ban_mtx);
-			VTAILQ_REMOVE(&obans, bt, l_list);
-			if (VTAILQ_EMPTY(&obans))
-				break;
-		}
-		if (DO_DEBUG(DBG_LURKER))
-			VSLb(vsl, SLT_Debug, "Lurk bt %p", bt);
-		ban_lurker_test_ban(wrk, vsl, bt, &obans);
-		if (VTAILQ_EMPTY(&obans))
-			break;
-	}
+	Lck_Lock(&ban_mtx);
+	VTAILQ_FOREACH(b, &obans, l_list)
+		ban_mark_completed(b);
+	Lck_Unlock(&ban_mtx);
 	return (dt);
 }
 
@@ -268,9 +261,9 @@ ban_lurker(struct worker *wrk, void *priv)
 
 	while (!ban_shutdown) {
 		d = ban_lurker_work(wrk, &vsl);
+		ban_cleantail();
 		if (DO_DEBUG(DBG_LURKER))
 			VSLb(&vsl, SLT_Debug, "lurker: sleep = %lf", d);
-		ban_cleantail();
 		d += VTIM_real();
 		Lck_Lock(&ban_mtx);
 		if (gen == ban_generation) {

bin/varnishtest/tests/c00049.vtc

@@ -94,7 +94,7 @@ client c1 {
 } -run
 
 # Get the VSL out of the way
-delay .1
+delay 1
 
 varnish v1 -cliok "ban.list"
 
@@ -115,18 +115,20 @@ varnish v1 -expect bans_dups == 0
 
 varnish v1 -cliok "param.set ban_lurker_sleep .01"
 
+delay 1
+
 varnish v1 -cliok "ban.list"
 
-delay 2
+delay 1
 
 varnish v1 -cliok "ban.list"
 
-varnish v1 -expect bans == 5
-varnish v1 -expect bans_completed == 4
+varnish v1 -expect bans == 4
+varnish v1 -expect bans_completed == 3
 varnish v1 -expect bans_req == 1
-varnish v1 -expect bans_obj == 4
+varnish v1 -expect bans_obj == 3
 varnish v1 -expect bans_added == 6
-varnish v1 -expect bans_deleted == 1
+varnish v1 -expect bans_deleted == 2
 varnish v1 -expect bans_tested == 0
 varnish v1 -expect bans_tests_tested == 0
 varnish v1 -expect bans_obj_killed == 0