BUG/MEDIUM: init/threads: provide per-thread alloc/free function callbacks

We currently have the ability to register functions to be called early
on thread creation and at thread deinitialization. It turns out this is
not sufficient because certain such functions may use resources that are
being allocated by the other ones, thus creating a race condition depending
only on the linking order. For example the mworker needs to register a
file descriptor while the pollers will reallocate the fd_updt[] array.
Similarly logs and trashes may be used by some init functions while it's
unclear whether they have been deduplicated.

The same issue happens on deinit, if the fd_updt[] or trash is released
before some functions finish to use them, we'll get into trouble.

This patch creates a couple of early and late callbacks for per-thread
allocation/freeing of resources. A few init functions were moved there,
and the fd init code was split between the two (since it used to both
allocate and initialize at once). This way the init/deinit sequence is
expected to be safe now.

This patch should be backported to 1.9 as at least the trash/log issue
seems to be present. The run_thread_poll_loop() code is a bit different
there as the mworker is not a callback, but it will have no effect and
it's enough to drop the mworker changes.

This bug was reported by Ilya Shipitsin in github issue #104.

include/types/global.h

@@ -283,8 +283,10 @@ void hap_register_build_opts(const char *str, int must_free);
 void hap_register_post_check(int (*fct)());
 void hap_register_post_deinit(void (*fct)());
 
+void hap_register_per_thread_alloc(int (*fct)());
 void hap_register_per_thread_init(int (*fct)());
 void hap_register_per_thread_deinit(void (*fct)());
+void hap_register_per_thread_free(int (*fct)());
 
 void mworker_accept_wrapper(int fd);
 void mworker_reload();
@@ -301,6 +303,10 @@ void mworker_reload();
 #define REGISTER_POST_DEINIT(fct) \
 	INITCALL1(STG_REGISTER, hap_register_post_deinit, (fct))
 
+/* simplified way to declare a per-thread allocation callback in a file */
+#define REGISTER_PER_THREAD_ALLOC(fct) \
+	INITCALL1(STG_REGISTER, hap_register_per_thread_alloc, (fct))
+
 /* simplified way to declare a per-thread init callback in a file */
 #define REGISTER_PER_THREAD_INIT(fct) \
 	INITCALL1(STG_REGISTER, hap_register_per_thread_init, (fct))
@@ -309,6 +315,10 @@ void mworker_reload();
 #define REGISTER_PER_THREAD_DEINIT(fct) \
 	INITCALL1(STG_REGISTER, hap_register_per_thread_deinit, (fct))
 
+/* simplified way to declare a per-thread free callback in a file */
+#define REGISTER_PER_THREAD_FREE(fct) \
+	INITCALL1(STG_REGISTER, hap_register_per_thread_free, (fct))
+
 #endif /* _TYPES_GLOBAL_H */
 
 /*

src/chunk.c

@@ -75,12 +75,12 @@ static int alloc_trash_buffers(int bufsize)
 	return trash.area && trash_buf1 && trash_buf2;
 }
 
-static int init_trash_buffers_per_thread()
+static int alloc_trash_buffers_per_thread()
 {
 	return alloc_trash_buffers(global.tune.bufsize);
 }
 
-static void deinit_trash_buffers_per_thread()
+static void free_trash_buffers_per_thread()
 {
 	chunk_destroy(&trash);
 	free(trash_buf2);
@@ -308,8 +308,8 @@ int chunk_strcasecmp(const struct buffer *chk, const char *str)
 	return diff;
 }
 
-REGISTER_PER_THREAD_INIT(init_trash_buffers_per_thread);
-REGISTER_PER_THREAD_DEINIT(deinit_trash_buffers_per_thread);
+REGISTER_PER_THREAD_ALLOC(alloc_trash_buffers_per_thread);
+REGISTER_PER_THREAD_FREE(free_trash_buffers_per_thread);
 
 /*
  * Local variables:

src/fd.c

@@ -576,17 +576,23 @@ void poller_pipe_io_handler(int fd)
 	fd_cant_recv(fd);
 }
 
-/* Initialize the pollers per thread */
+/* allocate the per-thread fd_updt thus needs to be called early after
+ * thread creation.
+ */
+static int alloc_pollers_per_thread()
+{
+	fd_updt = calloc(global.maxsock, sizeof(*fd_updt));
+	return fd_updt != NULL;
+}
+
+/* Initialize the pollers per thread.*/
 static int init_pollers_per_thread()
 {
 	int mypipe[2];
-	if ((fd_updt = calloc(global.maxsock, sizeof(*fd_updt))) == NULL)
+
+	if (pipe(mypipe) < 0)
 		return 0;
-	if (pipe(mypipe) < 0) {
-		free(fd_updt);
-		fd_updt = NULL;
-		return 0;
-	}
+
 	poller_rd_pipe = mypipe[0];
 	poller_wr_pipe[tid] = mypipe[1];
 	fcntl(poller_rd_pipe, F_SETFL, O_NONBLOCK);
@@ -599,9 +605,6 @@ static int init_pollers_per_thread()
 /* Deinitialize the pollers per thread */
 static void deinit_pollers_per_thread()
 {
-	free(fd_updt);
-	fd_updt = NULL;
-
 	/* rd and wr are init at the same place, but only rd is init to -1, so
 	  we rely to rd to close.   */
 	if (poller_rd_pipe > -1) {
@@ -612,6 +615,13 @@ static void deinit_pollers_per_thread()
 	}
 }
 
+/* Release the pollers per thread, to be called late */
+static void free_pollers_per_thread()
+{
+	free(fd_updt);
+	fd_updt = NULL;
+}
+
 /*
  * Initialize the pollers till the best one is found.
  * If none works, returns 0, otherwise 1.
@@ -769,8 +779,10 @@ int fork_poller()
 	return 1;
 }
 
+REGISTER_PER_THREAD_ALLOC(alloc_pollers_per_thread);
 REGISTER_PER_THREAD_INIT(init_pollers_per_thread);
 REGISTER_PER_THREAD_DEINIT(deinit_pollers_per_thread);
+REGISTER_PER_THREAD_FREE(free_pollers_per_thread);
 
 /*
  * Local variables:

src/haproxy.c

@@ -268,15 +268,14 @@ struct post_check_fct {
 	int (*fct)();
 };
 
-/* These functions are called when freeing the global sections at the end
- * of deinit, after everything is stopped. They don't return anything, and
- * they work in best effort mode as their sole goal is to make valgrind
- * mostly happy.
- */
-struct list post_deinit_list = LIST_HEAD_INIT(post_deinit_list);
-struct post_deinit_fct {
+/* These functions are called for each thread just after the thread creation
+ * and before running the init functions. They should be used to do per-thread
+ * (re-)allocations that are needed by subsequent functoins. They must return 0
+ * if an error occurred. */
+struct list per_thread_alloc_list = LIST_HEAD_INIT(per_thread_alloc_list);
+struct per_thread_alloc_fct {
 	struct list list;
-	void (*fct)();
+	int (*fct)();
 };
 
 /* These functions are called for each thread just after the thread creation
@@ -288,6 +287,29 @@ struct per_thread_init_fct {
 	int (*fct)();
 };
 
+/* These functions are called when freeing the global sections at the end of
+ * deinit, after everything is stopped. They don't return anything. They should
+ * not release shared resources that are possibly used by other deinit
+ * functions, only close/release what is private. Use the per_thread_free_list
+ * to release shared resources.
+ */
+struct list post_deinit_list = LIST_HEAD_INIT(post_deinit_list);
+struct post_deinit_fct {
+	struct list list;
+	void (*fct)();
+};
+
+/* These functions are called when freeing the global sections at the end of
+ * deinit, after the thread deinit functions, to release unneeded memory
+ * allocations. They don't return anything, and they work in best effort mode
+ * as their sole goal is to make valgrind mostly happy.
+ */
+struct list per_thread_free_list = LIST_HEAD_INIT(per_thread_free_list);
+struct per_thread_free_fct {
+	struct list list;
+	int (*fct)();
+};
+
 /* These functions are called for each thread just after the scheduler loop and
  * before exiting the thread. They don't return anything and, as for post-deinit
  * functions, they work in best effort mode as their sole goal is to make
@@ -349,6 +371,20 @@ void hap_register_post_deinit(void (*fct)())
 	LIST_ADDQ(&post_deinit_list, &b->list);
 }
 
+/* used to register some allocation functions to call for each thread. */
+void hap_register_per_thread_alloc(int (*fct)())
+{
+	struct per_thread_alloc_fct *b;
+
+	b = calloc(1, sizeof(*b));
+	if (!b) {
+		fprintf(stderr, "out of memory\n");
+		exit(1);
+	}
+	b->fct = fct;
+	LIST_ADDQ(&per_thread_alloc_list, &b->list);
+}
+
 /* used to register some initialization functions to call for each thread. */
 void hap_register_per_thread_init(int (*fct)())
 {
@@ -377,6 +413,20 @@ void hap_register_per_thread_deinit(void (*fct)())
 	LIST_ADDQ(&per_thread_deinit_list, &b->list);
 }
 
+/* used to register some free functions to call for each thread. */
+void hap_register_per_thread_free(int (*fct)())
+{
+	struct per_thread_free_fct *b;
+
+	b = calloc(1, sizeof(*b));
+	if (!b) {
+		fprintf(stderr, "out of memory\n");
+		exit(1);
+	}
+	b->fct = fct;
+	LIST_ADDQ(&per_thread_free_list, &b->list);
+}
+
 static void display_version()
 {
 	printf("HA-Proxy version %s %s - https://haproxy.org/\n", haproxy_version, haproxy_date);
@@ -2504,8 +2554,10 @@ static void run_poll_loop()
 
 static void *run_thread_poll_loop(void *data)
 {
+	struct per_thread_alloc_fct  *ptaf;
 	struct per_thread_init_fct   *ptif;
 	struct per_thread_deinit_fct *ptdf;
+	struct per_thread_free_fct   *ptff;
 
 	ha_set_tid((unsigned long)data);
 
@@ -2519,6 +2571,18 @@ static void *run_thread_poll_loop(void *data)
 
 	tv_update_date(-1,-1);
 
+	/* per-thread alloc calls performed here are not allowed to snoop on
+	 * other threads, so they are free to initialize at their own rhythm
+	 * as long as they act as if they were alone. None of them may rely
+	 * on resources initialized by the other ones.
+	 */
+	list_for_each_entry(ptaf, &per_thread_alloc_list, list) {
+		if (!ptaf->fct()) {
+			ha_alert("failed to allocate resources for thread %u.\n", tid);
+			exit(1);
+		}
+	}
+
 	/* per-thread init calls performed here are not allowed to snoop on
 	 * other threads, so they are free to initialize at their own rhythm
 	 * as long as they act as if they were alone.
@@ -2541,6 +2605,9 @@ static void *run_thread_poll_loop(void *data)
 	list_for_each_entry(ptdf, &per_thread_deinit_list, list)
 		ptdf->fct();
 
+	list_for_each_entry(ptff, &per_thread_free_list, list)
+		ptff->fct();
+
 #ifdef USE_THREAD
 	_HA_ATOMIC_AND(&all_threads_mask, ~tid_bit);
 	if (tid > 0)

src/log.c

@@ -1863,16 +1863,6 @@ static void init_log()
 
 INITCALL0(STG_PREPARE, init_log);
 
-static int init_log_buffers_per_thread()
-{
-	return init_log_buffers();
-}
-
-static void deinit_log_buffers_per_thread()
-{
-	deinit_log_buffers();
-}
-
 /* Initialize log buffers used for syslog messages */
 int init_log_buffers()
 {
@@ -3021,8 +3011,8 @@ static struct cli_kw_list cli_kws = {{ },{
 
 INITCALL1(STG_REGISTER, cli_register_kw, &cli_kws);
 
-REGISTER_PER_THREAD_INIT(init_log_buffers_per_thread);
-REGISTER_PER_THREAD_DEINIT(deinit_log_buffers_per_thread);
+REGISTER_PER_THREAD_ALLOC(init_log_buffers);
+REGISTER_PER_THREAD_FREE(deinit_log_buffers);
 
 /*
  * Local variables: