WIP/DRAFT Add a facility for unlocked director iteration

Pondering #4140 made me realize that our central per-vcl director list
constitues a classic case of iterating a mutable list. That besides the already
known fact that running potentially expensive iterator functions while holding
the vcl_mtx is a very bad idea.

So this patch is a suggestion on how to get out of this sitation. It does not
go all the way (it still leaves unsolved a similar problem of iterating over all
backends of _all vcls_), but shows an attempt on how to tackle the "iterate over
one VCL's backends".

We add an fittingly named 'vdure' facility which manages the director list and,
in particular, director retirement. The main change is that, while iterators are
active on the director list, vcl_mtx is _not_ held and any request of a director
to retire only ends up in a resignation, which manifests by this director being
put on a second list.

When all iterators have completed, the resignation list is worked and the actual
retirement carried out.

NOTE: I am particularly _not_ happy about the vdire_(start|end)_iter() interface
being exposed. What I would like to see is a FOREACH macro which does both the
start and end. Having to use an iterator with a callback is clumsy, and having
to directly call start/end with a VTAILQ_FOREACH feels like crossing
interfaces.

bin/varnishd/cache/cache_director.h

@@ -43,7 +43,8 @@ struct vcldir {
 	struct director			*dir;
 	struct vcl			*vcl;
 	const struct vdi_methods	*methods;
-	VTAILQ_ENTRY(vcldir)		list;
+	VTAILQ_ENTRY(vcldir)		directors_list;
+	VTAILQ_ENTRY(vcldir)		resigning_list;
 	const struct vdi_ahealth	*admin_health;
 	vtim_real			health_changed;
 	char				*cli_name;

bin/varnishd/cache/cache_vcl.c

@@ -368,8 +368,121 @@ VCL_Update(struct vcl **vcc, struct vcl *vcl)
 	assert((*vcc)->temp->is_warm);
 }
 
+/*--------------------------------------------------------------------
+ * vdire: Vcl DIrector REsignation Management (born out of a dire situation)
+ * iterators over the director list need to register.
+ * while iterating, directors can not retire immediately,
+ * they get put on a list of resigning directors. The
+ * last iterator executes the retirement.
+ */
+
+static struct vdire *
+vdire_new(struct lock *mtx, const struct vcltemp **tempp)
+{
+	struct vdire *vdire;
+
+	ALLOC_OBJ(vdire, VDIRE_MAGIC);
+	AN(vdire);
+	AN(mtx);
+	VTAILQ_INIT(&vdire->directors);
+	VTAILQ_INIT(&vdire->resigning);
+	vdire->mtx = mtx;
+	vdire->tempp = tempp;
+	return (vdire);
+}
+
+/* starting an interation prevents removals from the directors list */
+void
+vdire_start_iter(struct vdire *vdire)
+{
+
+	CHECK_OBJ_NOTNULL(vdire, VDIRE_MAGIC);
+
+	Lck_Lock(vdire->mtx);
+	vdire->iterating++;
+	Lck_Unlock(vdire->mtx);
+}
+
+void
+vdire_end_iter(struct vdire *vdire)
+{
+	struct vcldir_head resigning = VTAILQ_HEAD_INITIALIZER(resigning);
+	const struct vcltemp *temp = NULL;
+	struct vcldir *vdir;
+	unsigned n;
+
+	CHECK_OBJ_NOTNULL(vdire, VDIRE_MAGIC);
+
+	Lck_Lock(vdire->mtx);
+	AN(vdire->iterating);
+	n = --vdire->iterating;
+
+	if (n == 0) {
+		VTAILQ_SWAP(&vdire->resigning, &resigning, vcldir, resigning_list);
+		VTAILQ_FOREACH(vdir, &resigning, resigning_list)
+			VTAILQ_REMOVE(&vdire->directors, vdir, directors_list);
+		temp = *vdire->tempp;
+	}
+	Lck_Unlock(vdire->mtx);
+
+	VTAILQ_FOREACH(vdir, &resigning, resigning_list)
+		vcldir_retire(vdir, temp);
+}
+
+// if there are no iterators, remove from directors and retire
+// otherwise but on resigning list to work when iterators end
+void
+vdire_resign(struct vdire *vdire, struct vcldir *vdir)
+{
+	const struct vcltemp *temp = NULL;
+
+	CHECK_OBJ_NOTNULL(vdire, VDIRE_MAGIC);
+	AN(vdir);
+
+	Lck_Lock(vdire->mtx);
+	if (vdire->iterating != 0) {
+		VTAILQ_INSERT_TAIL(&vdire->resigning, vdir, resigning_list);
+		vdir = NULL;
+	} else {
+		VTAILQ_REMOVE(&vdire->directors, vdir, directors_list);
+		temp = *vdire->tempp;
+	}
+	Lck_Unlock(vdire->mtx);
+
+	if (vdir != NULL)
+		vcldir_retire(vdir, temp);
+}
+
+// unlocked version of vcl_iterdir
+// pat can also be NULL (to iterate all)
+static int
+vdire_iter(struct cli *cli, const char *pat, const struct vcl *vcl,
+    vcl_be_func *func, void *priv)
+{
+	int i, found = 0;
+	struct vcldir *vdir;
+	struct vdire *vdire = vcl->vdire;
+
+	vdire_start_iter(vdire);
+	VTAILQ_FOREACH(vdir, &vdire->directors, directors_list) {
+		if (pat != NULL && fnmatch(pat, vdir->cli_name, 0))
+			continue;
+		found++;
+		i = func(cli, vdir->dir, priv);
+		if (i < 0) {
+			found = i;
+			break;
+		}
+		found += i;
+	}
+	vdire_end_iter(vdire);
+	return (found);
+}
+
+
 /*--------------------------------------------------------------------*/
 
+// XXX locked case across VCLs - should do without
 static int
 vcl_iterdir(struct cli *cli, const char *pat, const struct vcl *vcl,
     vcl_be_func *func, void *priv)
@@ -378,7 +491,7 @@ vcl_iterdir(struct cli *cli, const char *pat, const struct vcl *vcl,
 	struct vcldir *vdir;
 
 	Lck_AssertHeld(&vcl_mtx);
-	VTAILQ_FOREACH(vdir, &vcl->director_list, list) {
+	VTAILQ_FOREACH(vdir, &vcl->vdire->directors, directors_list) {
 		if (fnmatch(pat, vdir->cli_name, 0))
 			continue;
 		found++;
@@ -416,10 +529,10 @@ VCL_IterDirector(struct cli *cli, const char *pat,
 		vcl = NULL;
 	}
 	AZ(VSB_finish(vsb));
-	Lck_Lock(&vcl_mtx);
 	if (vcl != NULL) {
-		found = vcl_iterdir(cli, VSB_data(vsb), vcl, func, priv);
+		found = vdire_iter(cli, VSB_data(vsb), vcl, func, priv);
 	} else {
+		Lck_Lock(&vcl_mtx);
 		VTAILQ_FOREACH(vcl, &vcl_head, list) {
 			i = vcl_iterdir(cli, VSB_data(vsb), vcl, func, priv);
 			if (i < 0) {
@@ -429,8 +542,8 @@ VCL_IterDirector(struct cli *cli, const char *pat,
 				found += i;
 			}
 		}
+		Lck_Unlock(&vcl_mtx);
 	}
-	Lck_Unlock(&vcl_mtx);
 	VSB_destroy(&vsb);
 	return (found);
 }
@@ -439,31 +552,37 @@ static void
 vcl_BackendEvent(const struct vcl *vcl, enum vcl_event_e e)
 {
 	struct vcldir *vdir;
+	struct vdire *vdire;
 
 	ASSERT_CLI();
 	CHECK_OBJ_NOTNULL(vcl, VCL_MAGIC);
 	AZ(vcl->busy);
+	vdire = vcl->vdire;
 
-	Lck_Lock(&vcl_mtx);
-	VTAILQ_FOREACH(vdir, &vcl->director_list, list)
+	vdire_start_iter(vdire);
+	VTAILQ_FOREACH(vdir, &vdire->directors, directors_list)
 		VDI_Event(vdir->dir, e);
-	Lck_Unlock(&vcl_mtx);
+	vdire_end_iter(vdire);
 }
 
 static void
 vcl_KillBackends(const struct vcl *vcl)
 {
 	struct vcldir *vdir, *vdir2;
+	struct vdire *vdire;
 
 	CHECK_OBJ_NOTNULL(vcl, VCL_MAGIC);
 	assert(vcl->temp == VCL_TEMP_COLD || vcl->temp == VCL_TEMP_INIT);
+	vdire = vcl->vdire;
+	CHECK_OBJ_NOTNULL(vdire, VDIRE_MAGIC);
+
 	/*
-	 * Unlocked because no further directors can be added, and the
+	 * Unlocked and sidelining vdire because no further directors can be added, and the
 	 * remaining ones need to be able to remove themselves.
 	 */
-	VTAILQ_FOREACH_SAFE(vdir, &vcl->director_list, list, vdir2)
+	VTAILQ_FOREACH_SAFE(vdir, &vdire->directors, directors_list, vdir2)
 		VDI_Event(vdir->dir, VCL_EVENT_DISCARD);
-	assert(VTAILQ_EMPTY(&vcl->director_list));
+	assert(VTAILQ_EMPTY(&vdire->directors));
 }
 
 /*--------------------------------------------------------------------*/
@@ -697,7 +816,7 @@ vcl_load(struct cli *cli,
 
 	vcl->loaded_name = strdup(name);
 	XXXAN(vcl->loaded_name);
-	VTAILQ_INIT(&vcl->director_list);
+	vcl->vdire = vdire_new(&vcl_mtx, &vcl->temp);
 	VTAILQ_INIT(&vcl->ref_list);
 	VTAILQ_INIT(&vcl->filters);
 
@@ -747,6 +866,7 @@ VCL_Poll(void)
 			AZ(nomsg);
 			vcl_KillBackends(vcl);
 			free(vcl->loaded_name);
+			FREE_OBJ(vcl->vdire);
 			VCL_Close(&vcl);
 			VSC_C_main->n_vcl--;
 			VSC_C_main->n_vcl_discard--;

bin/varnishd/cache/cache_vcl.h

@@ -37,7 +37,18 @@ struct vfilter;
 struct vcltemp;
 
 VTAILQ_HEAD(vfilter_head, vfilter);
+VTAILQ_HEAD(vcldir_head, vcldir);
 
+struct vdire {
+	unsigned		magic;
+#define VDIRE_MAGIC		0x51748697
+	unsigned		iterating;
+	struct vcldir_head	directors;
+	struct vcldir_head	resigning;
+	// vcl_mtx for now - to be refactored into separate mtx?
+	struct lock		*mtx;
+	const struct vcltemp	**tempp;
+};
 
 struct vcl {
 	unsigned		magic;
@@ -50,10 +61,10 @@ struct vcl {
 	unsigned		busy;
 	unsigned		discard;
 	const struct vcltemp	*temp;
-	VTAILQ_HEAD(,vcldir)	director_list;
 	VTAILQ_HEAD(,vclref)	ref_list;
-	int			nrefs;
+	struct vdire		*vdire;
 	struct vcl		*label;
+	int			nrefs;
 	int			nlabels;
 	struct vfilter_head	filters;
 };
@@ -92,3 +103,13 @@ extern const struct vcltemp VCL_TEMP_COOLING[1];
 		assert(vcl_active == NULL ||			\
 		    vcl_active->temp->is_warm);			\
 	} while (0)
+
+/* cache_vrt_vcl.c used in cache_vcl.c */
+struct vcldir;
+void vcldir_retire(struct vcldir *vdir, const struct vcltemp *temp);
+
+/* cache_vcl.c */
+void vdire_resign(struct vdire *vdire, struct vcldir *vdir);
+
+void vdire_start_iter(struct vdire *vdire);
+void vdire_end_iter(struct vdire *vdire);

bin/varnishd/cache/cache_vrt_vcl.c

@@ -240,7 +240,7 @@ VRT_AddDirector(VRT_CTX, const struct vdi_methods *m, void *priv,
 	Lck_AssertHeld(&vcl_mtx);
 	temp = vcl->temp;
 	if (temp != VCL_TEMP_COOLING)
-		VTAILQ_INSERT_TAIL(&vcl->director_list, vdir, list);
+		VTAILQ_INSERT_TAIL(&vcl->vdire->directors, vdir, directors_list);
 	if (temp->is_warm)
 		VDI_Event(vdir->dir, VCL_EVENT_WARM);
 	Lck_Unlock(&vcl_mtx);
@@ -267,19 +267,15 @@ VRT_StaticDirector(VCL_BACKEND b)
 	vdir->flags |= VDIR_FLG_NOREFCNT;
 }
 
-static void
-vcldir_retire(struct vcldir *vdir)
+// vcldir is already removed from the directors list
+// to be called only from vdire_*
+void
+vcldir_retire(struct vcldir *vdir, const struct vcltemp *temp)
 {
-	const struct vcltemp *temp;
 
 	CHECK_OBJ_NOTNULL(vdir, VCLDIR_MAGIC);
 	assert(vdir->refcnt == 0);
-	CHECK_OBJ_NOTNULL(vdir->vcl, VCL_MAGIC);
-
-	Lck_Lock(&vcl_mtx);
-	temp = vdir->vcl->temp;
-	VTAILQ_REMOVE(&vdir->vcl->director_list, vdir, list);
-	Lck_Unlock(&vcl_mtx);
+	AN(temp);
 
 	if (temp->is_warm)
 		VDI_Event(vdir->dir, VCL_EVENT_COLD);
@@ -302,7 +298,7 @@ vcldir_deref(struct vcldir *vdir)
 	Lck_Unlock(&vdir->dlck);
 
 	if (!busy)
-		vcldir_retire(vdir);
+		vdire_resign(vdir->vcl->vdire, vdir);
 	return (busy);
 }
 
@@ -374,6 +370,7 @@ VRT_LookupDirector(VRT_CTX, VCL_STRING name)
 	struct vcl *vcl;
 	struct vcldir *vdir;
 	VCL_BACKEND dd, d = NULL;
+	struct vdire *vdire;
 
 	CHECK_OBJ_NOTNULL(ctx, VRT_CTX_MAGIC);
 	AN(name);
@@ -384,15 +381,17 @@ VRT_LookupDirector(VRT_CTX, VCL_STRING name)
 	vcl = ctx->vcl;
 	CHECK_OBJ_NOTNULL(vcl, VCL_MAGIC);
 
-	Lck_Lock(&vcl_mtx);
-	VTAILQ_FOREACH(vdir, &vcl->director_list, list) {
+	vdire = vcl->vdire;
+
+	vdire_start_iter(vdire);
+	VTAILQ_FOREACH(vdir, &vdire->directors, directors_list) {
 		dd = vdir->dir;
 		if (strcmp(dd->vcl_name, name))
 			continue;
 		d = dd;
 		break;
 	}
-	Lck_Unlock(&vcl_mtx);
+	vdire_end_iter(vdire);
 
 	return (d);
 }