lib: Improve test & fix race condition in thread pool

It could sometimes happen that we enqueue a small amount of work, and
then manage to lock the pool mutex in a subsequent call to
thread_pool_wait() before any threads managed to grab any work. Since we
didn't check if the pool has any queued work, it could happen that we
continue executing on the main thread assuming the pool is done working,
when in reality it actually just started working.
Simple fix, just check if there is work to be done, and block as long as
we do.

src/common/platform/thread_pool.c

@@ -11,6 +11,15 @@
 #include "thread.h"
 #include "../logging.h"
 
+// Mostly based on John Schember's excellent blog post:
+// https://nachtimwald.com/2019/04/12/thread-pool-in-c/
+//
+// I just added an extra pool->first check to thread_pool_wait()
+// since I discovered a race condition in my torture tests for this
+// implementation. Basically, sometimes we could blow through a
+// call to thread_pool_wait() if we enqueue a small amount of work
+// and call it before threads had a chance to fetch work.
+
 struct cr_task {
 	void (*fn)(void *arg);
 	void *arg;
@@ -148,7 +157,7 @@ void thread_pool_wait(struct cr_thread_pool *pool) {
 	if (!pool) return;
 	mutex_lock(pool->mutex);
 	while (true) {
-		if ((!pool->stop_flag && pool->active_workers != 0) || (pool->stop_flag && pool->thread_count != 0)) {
+		if (pool->first || (!pool->stop_flag && pool->active_workers != 0) || (pool->stop_flag && pool->thread_count != 0)) {
 			thread_cond_wait(&pool->work_ongoing, pool->mutex);
 		} else {
 			break;

tests/test_thread_pool.h

@@ -14,31 +14,32 @@
 
 void test_task(void *arg) {
 	int *input = arg;
-	int old = *input;
-
 	*input += 1000;
-
 	if (*input % 2) {
 		timer_sleep_ms(10);
 	}
 }
 
 bool test_thread_pool(void) {
 
-	const size_t threads = 4;
-	const int iterations = 100;
+	const size_t threads = 8;
+	const int iterations = 10;
+	int loops = 100;
 
 	struct cr_thread_pool *pool = thread_pool_create(threads);
 	int *values = calloc(iterations, sizeof(*values));
-	for (int i = 0; i < iterations; ++i) {
-		values[i] = i;
-		thread_pool_enqueue(pool, test_task, values + i);
-	}
-
-	thread_pool_wait(pool);
-
-	for (int i = 0; i < iterations; ++i) {
-		test_assert(values[i] == i + 1000);
+	while (loops--) {
+		for (int i = 0; i < iterations; ++i) {
+			values[i] = i;
+			thread_pool_enqueue(pool, test_task, &values[i]);
+		}
+
+		thread_pool_wait(pool);
+
+		for (int i = 0; i < iterations; ++i) {
+			test_assert(values[i] == i + 1000);
+		}
+		memset(values, 0, iterations * sizeof(*values));
 	}
 
 	thread_pool_destroy(pool);