vcp: Poll dying connection pools asynchronously

Before, we would wait in a blocking loop for
connection pools to be freed before deleting
them for good.

This commit transforms this blocking loop into a
CLI hook that will delete the freed connection
pools at each trigger.

Refs #4064

Better diff with the --ignore-all-space option.

bin/varnishd/cache/cache_cli.c

@@ -78,6 +78,7 @@ cli_cb_before(const struct cli *cli)
 	VSL(SLT_CLI, NO_VXID, "Rd %s", VSB_data(cli->cmd));
 	Lck_Lock(&cli_mtx);
 	VCL_Poll();
+	VCP_RelPoll();
 }
 
 static void

bin/varnishd/cache/cache_conn_pool.c

@@ -106,14 +106,20 @@ struct conn_pool {
 };
 
 static struct lock conn_pools_mtx;
+static struct lock dead_pools_mtx;
 
-static VRBT_HEAD(vrb, conn_pool) conn_pools = VRBT_INITIALIZER(&conn_pools);
+VRBT_HEAD(vrb, conn_pool);
 VRBT_GENERATE_REMOVE_COLOR(vrb, conn_pool, entry, static)
 VRBT_GENERATE_REMOVE(vrb, conn_pool, entry, static)
 VRBT_GENERATE_FIND(vrb, conn_pool, entry, vcp_cmp, static)
 VRBT_GENERATE_INSERT_COLOR(vrb, conn_pool, entry, static)
 VRBT_GENERATE_INSERT_FINISH(vrb, conn_pool, entry, static)
 VRBT_GENERATE_INSERT(vrb, conn_pool, entry, vcp_cmp, static)
+VRBT_GENERATE_NEXT(vrb, conn_pool, entry, static);
+VRBT_GENERATE_MINMAX(vrb, conn_pool, entry, static);
+
+static struct vrb conn_pools = VRBT_INITIALIZER(&conn_pools);
+static struct vrb dead_pools = VRBT_INITIALIZER(&dying_cps);
 
 /*--------------------------------------------------------------------
  */
@@ -218,6 +224,22 @@ VCP_AddRef(struct conn_pool *cp)
 	Lck_Unlock(&conn_pools_mtx);
 }
 
+/*--------------------------------------------------------------------
+ */
+
+static void
+vcp_destroy(struct conn_pool **cpp)
+{
+	struct conn_pool *cp;
+
+	TAKE_OBJ_NOTNULL(cp, cpp, CONN_POOL_MAGIC);
+	AZ(cp->n_conn);
+	AZ(cp->n_kill);
+	Lck_Delete(&cp->mtx);
+	free(cp->endpoint);
+	FREE_OBJ(cp);
+}
+
 /*--------------------------------------------------------------------
  * Release Conn pool, destroy if last reference.
  */
@@ -249,17 +271,56 @@ VCP_Rel(struct conn_pool **cpp)
 		(void)shutdown(pfd->fd, SHUT_RDWR);
 		cp->n_kill++;
 	}
-	while (cp->n_kill) {
-		Lck_Unlock(&cp->mtx);
-		(void)usleep(20000);
-		Lck_Lock(&cp->mtx);
-	}
 	Lck_Unlock(&cp->mtx);
-	Lck_Delete(&cp->mtx);
-	AZ(cp->n_conn);
-	AZ(cp->n_kill);
-	free(cp->endpoint);
-	FREE_OBJ(cp);
+	if (cp->n_kill == 0) {
+		vcp_destroy(&cp);
+		return;
+	}
+	Lck_Lock(&dead_pools_mtx);
+	/*
+	 * Here we reuse cp's entry but it will probably not be correctly
+	 * indexed because of the hack in VCP_RelPoll
+	 */
+	VRBT_INSERT(vrb, &dead_pools, cp);
+	Lck_Unlock(&dead_pools_mtx);
+}
+
+void
+VCP_RelPoll(void)
+{
+	struct vrb dead;
+	struct conn_pool *cp, *cp2;
+
+	ASSERT_CLI();
+
+	Lck_Lock(&dead_pools_mtx);
+	if (VRBT_EMPTY(&dead_pools)) {
+		Lck_Unlock(&dead_pools_mtx);
+		return;
+	}
+	dead = dead_pools;
+	VRBT_INIT(&dead_pools);
+	Lck_Unlock(&dead_pools_mtx);
+
+	VRBT_FOREACH_SAFE(cp, vrb, &dead, cp2) {
+		CHECK_OBJ_NOTNULL(cp, CONN_POOL_MAGIC);
+		if (cp->n_kill > 0)
+			continue;
+		VRBT_REMOVE(vrb, &dead, cp);
+		vcp_destroy(&cp);
+	}
+
+	if (!VRBT_EMPTY(&dead)) {
+		Lck_Lock(&dead_pools_mtx);
+		/*
+		 * The following insertion will most likely result in an
+		 * unordered tree, but in this case it does not matter
+		 * as we just want to iterate over all the elements
+		 * in the tree in order to delete them.
+		 */
+		VRBT_INSERT(vrb, &dead_pools, dead.rbh_root);
+		Lck_Unlock(&dead_pools_mtx);
+	}
 }
 
 /*--------------------------------------------------------------------
@@ -583,6 +644,7 @@ void
 VCP_Init(void)
 {
 	Lck_New(&conn_pools_mtx, lck_conn_pool);
+	Lck_New(&dead_pools_mtx, lck_dead_pool);
 }
 
 /**********************************************************************/

bin/varnishd/cache/cache_varnishd.h

@@ -475,6 +475,7 @@ void VSL_Flush(struct vsl_log *, int overflow);
 struct conn_pool;
 void VCP_Init(void);
 void VCP_Panic(struct vsb *, struct conn_pool *);
+void VCP_RelPoll(void);
 
 /* cache_backend_probe.c */
 void VBP_Init(void);

include/tbl/locks.h

@@ -45,6 +45,7 @@ LOCK(pipestat)
 LOCK(probe)
 LOCK(sess)
 LOCK(conn_pool)
+LOCK(dead_pool)
 LOCK(vbe)
 LOCK(vcapace)
 LOCK(vcl)