Fix problem with purging and the n_obj_purged counter

When the do..while loop in HSH_Purge executes on a oh with many
popular variants, there is a potential problem with the "array" of oc
pointers, allocated in the thread workspace. If many of the oc's have
positive refcounts, they will fill up the array and

	EXP_Rearm(oc, now, ttl, grace, keep);
	(void)HSH_DerefObjCore(wrk, &oc, 0);

will be called several on the same oc's. At the same time, the counter
n_obj_purged will be updated with a too low number.

The test case demonstrates how we get a too low value for this counter,
but it is not able to force varnishd to use a siginificant amount of CPU.

bin/varnishd/cache/cache_hash.c

@@ -590,7 +590,7 @@ HSH_Purge(struct worker *wrk, struct objhead *oh, double ttl, double grace,
 double keep)
 {
 	struct objcore *oc, **ocp;
-	unsigned spc, ospc, nobj, n;
+	unsigned spc, ospc, nobj, n, n_tot = 0;
 	int more = 0;
 	double now;
 
@@ -598,6 +598,20 @@ double keep)
 	CHECK_OBJ_NOTNULL(oh, OBJHEAD_MAGIC);
 	ospc = WS_Reserve(wrk->aws, 0);
 	assert(ospc >= sizeof *ocp);
+	/*
+	 * Because of "soft" purges, there might be oc's in the list that has
+	 * the OC_F_PURGED flag set. We do not want to let these slip through,
+	 * so we need to clear the flag before entering the do..while loop.
+	 */
+	Lck_Lock(&oh->mtx);
+	assert(oh->refcnt > 0);
+	VTAILQ_FOREACH(oc, &oh->objcs, hsh_list) {
+		CHECK_OBJ_NOTNULL(oc, OBJCORE_MAGIC);
+		assert(oc->objhead == oh);
+		oc->flags &= ~OC_F_PURGED;
+	}
+	Lck_Unlock(&oh->mtx);
+
 	do {
 		more = 0;
 		spc = ospc;
@@ -621,6 +635,15 @@ double keep)
 			}
 			if (oc->flags & OC_F_DYING)
 				continue;
+			if (oc->flags & OC_F_PURGED) {
+				/*
+				 * We have already called EXP_Rearm on this
+				 * object, and we do not want to do it
+				 * again. Plus the space in the ocp array may
+				 * be limited.
+				 */
+				continue;
+			}
 			if (spc < sizeof *ocp) {
 				/* Iterate if aws is not big enough */
 				more = 1;
@@ -629,6 +652,7 @@ double keep)
 			oc->refcnt++;
 			spc -= sizeof *ocp;
 			ocp[nobj++] = oc;
+			oc->flags |= OC_F_PURGED;
 		}
 		Lck_Unlock(&oh->mtx);
 
@@ -638,9 +662,10 @@ double keep)
 			EXP_Rearm(oc, now, ttl, grace, keep);
 			(void)HSH_DerefObjCore(wrk, &oc, 0);
 		}
+		n_tot += nobj;
 	} while (more);
 	WS_Release(wrk->aws, 0);
-	Pool_PurgeStat(nobj);
+	Pool_PurgeStat(n_tot);
 }
 
 /*---------------------------------------------------------------------

bin/varnishtest/tests/r02372.vtc

+varnishtest "Count purges when there are many variants"
+
+server s1 -repeat 40 {
+	rxreq
+	txresp -hdr "Vary: foo"
+} -start
+
+varnish v1 -arg "-p workspace_thread=256" -vcl+backend {
+	import std;
+
+	sub vcl_recv {
+		if (req.method == "PURGE") {
+			return (purge);
+		}
+		set req.http.foo = "x" + std.random(1,10) * 1000000000;
+	}
+} -start
+
+client c1 -repeat 40 {
+	txreq
+	rxresp
+} -run
+
+client c2 {
+	txreq -req "PURGE"
+	rxresp
+} -run
+
+varnish v1 -expect n_lru_nuked == 0
+varnish v1 -expect cache_hit == 0
+varnish v1 -expect n_purges == 1
+varnish v1 -expect n_obj_purged == 40

include/tbl/oc_flags.h

@@ -28,6 +28,7 @@
 
 /*lint -save -e525 -e539 */
 
+OC_FLAG(PURGED,		purged,		(1<<0))
 OC_FLAG(BUSY,		busy,		(1<<1))
 OC_FLAG(PASS,		pass,		(1<<2))
 OC_FLAG(HFP,		hfp,		(1<<3))