Merge branch 'wip/pwithnall/2456-thread-pool-fix' into 'main'

tests: Rewrite thread-pool test for freeing queued items

Closes #2456

See merge request GNOME/glib!2229

glib/gthreadpool.c

@@ -964,6 +964,11 @@ g_thread_pool_free_internal (GRealThreadPool* pool)
   g_return_if_fail (pool->running == FALSE);
   g_return_if_fail (pool->num_threads == 0);
 
+  /* Ensure the dummy item pushed on by g_thread_pool_wakeup_and_stop_all() is
+   * removed, before it’s potentially passed to the user-provided
+   * @item_free_func. */
+  g_async_queue_remove (pool->queue, GUINT_TO_POINTER (1));
+
   g_async_queue_unref (pool->queue);
   g_cond_clear (&pool->cond);
 

glib/tests/thread-pool.c

@@ -87,12 +87,6 @@ dummy_pool_func (gpointer data, gpointer user_data)
   g_assert_true (data == GUINT_TO_POINTER (123));
 }
 
-static void
-dummy_pool_func_full (gpointer data, gpointer user_data)
-{
-  g_assert_true (data == user_data);
-}
-
 static void
 test_create_first_pool (gconstpointer shared_first)
 {
@@ -148,53 +142,115 @@ test_create_first_pool (gconstpointer shared_first)
   g_thread_pool_free (pool, TRUE, TRUE);
 }
 
+typedef struct
+{
+  GMutex mutex;  /* (owned) */
+  GCond cond;  /* (owned) */
+  gboolean threads_should_block;  /* protected by mutex, cond */
+
+  guint n_jobs_started;  /* (atomic) */
+  guint n_jobs_completed;  /* (atomic) */
+  guint n_free_func_calls;  /* (atomic) */
+} TestThreadPoolFullData;
+
+static void
+full_thread_func (gpointer data,
+                  gpointer user_data)
+{
+  TestThreadPoolFullData *test_data = data;
+
+  g_atomic_int_inc (&test_data->n_jobs_started);
+
+  /* Make the thread block until told to stop blocking. */
+  g_mutex_lock (&test_data->mutex);
+  while (test_data->threads_should_block)
+    g_cond_wait (&test_data->cond, &test_data->mutex);
+  g_mutex_unlock (&test_data->mutex);
+
+  g_atomic_int_inc (&test_data->n_jobs_completed);
+}
+
 static void
 free_func (gpointer user_data)
 {
-  gboolean *free_func_called = user_data;
+  TestThreadPoolFullData *test_data = user_data;
-  *free_func_called = TRUE;
+
+  g_atomic_int_inc (&test_data->n_free_func_calls);
 }
 
 static void
 test_thread_pool_full (gconstpointer shared_first)
 {
-  GThreadPool *pool;
+  guint i;
-  gboolean free_func_called = FALSE;
-  GError *err = NULL;
-  gboolean success;
 
   g_test_bug ("https://gitlab.gnome.org/GNOME/glib/-/issues/121");
 
   g_thread_pool_set_max_unused_threads (0);
 
-  if (GPOINTER_TO_INT (shared_first))
+  /* Run the test twice, once with a shared pool and once with an exclusive one. */
-    pool = g_thread_pool_new_full (dummy_pool_func_full, &free_func_called, free_func, -1, FALSE, &err);
+  for (i = 0; i < 2; i++)
-  else
+    {
-    pool = g_thread_pool_new_full (dummy_pool_func_full, &free_func_called, free_func, 2, TRUE, &err);
+      GThreadPool *pool;
-  g_assert_no_error (err);
+      TestThreadPoolFullData test_data;
-  g_assert_nonnull (pool);
+      GError *local_error = NULL;
+      gboolean success;
+      guint j;
 
-  success = g_thread_pool_push (pool, &free_func_called, &err);
+      g_mutex_init (&test_data.mutex);
-  g_assert_no_error (err);
+      g_cond_init (&test_data.cond);
-  g_assert_true (success);
+      test_data.threads_should_block = TRUE;
+      test_data.n_jobs_started = 0;
+      test_data.n_jobs_completed = 0;
+      test_data.n_free_func_calls = 0;
 
-  g_thread_pool_free (pool, TRUE, TRUE);
+      /* Create a thread pool with only one worker thread. The pool can be
-  g_assert_true (free_func_called);
+       * created in shared or exclusive mode. */
+      pool = g_thread_pool_new_full (full_thread_func, &test_data, free_func,
+                                     1, (i == 0),
+                                     &local_error);
+      g_assert_no_error (local_error);
+      g_assert_nonnull (pool);
 
-  free_func_called = FALSE;
+      /* Push two jobs into the pool. The first one will start executing and
-  if (GPOINTER_TO_INT (shared_first))
+       * will block, the second one will wait in the queue as there’s only one
-    pool = g_thread_pool_new_full (dummy_pool_func_full, &free_func_called, free_func, 2, TRUE, &err);
+       * worker thread. */
-  else
+      for (j = 0; j < 2; j++)
-    pool = g_thread_pool_new_full (dummy_pool_func_full, &free_func_called, free_func, -1, FALSE, &err);
+        {
-  g_assert_no_error (err);
+          success = g_thread_pool_push (pool, &test_data, &local_error);
-  g_assert_nonnull (pool);
+          g_assert_no_error (local_error);
+          g_assert_true (success);
+        }
 
-  success = g_thread_pool_push (pool, &free_func_called, &err);
+      /* Wait for the first job to start. */
-  g_assert_no_error (err);
+      while (g_atomic_int_get (&test_data.n_jobs_started) == 0);
-  g_assert_true (success);
 
-  g_thread_pool_free (pool, TRUE, TRUE);
+      /* Free the pool. This won’t actually free the queued second job yet, as
-  g_assert_true (free_func_called);
+       * the thread pool hangs around until the executing first job has
+       * completed. The first job will complete only once @threads_should_block
+       * is unset. */
+      g_thread_pool_free (pool, TRUE, FALSE);
+
+      g_assert_cmpuint (g_atomic_int_get (&test_data.n_jobs_started), ==, 1);
+      g_assert_cmpuint (g_atomic_int_get (&test_data.n_jobs_completed), ==, 0);
+      g_assert_cmpuint (g_atomic_int_get (&test_data.n_free_func_calls), ==, 0);
+
+      /* Unblock the job and allow the pool to be freed. */
+      g_mutex_lock (&test_data.mutex);
+      test_data.threads_should_block = FALSE;
+      g_cond_signal (&test_data.cond);
+      g_mutex_unlock (&test_data.mutex);
+
+      /* Wait for the first job to complete before freeing the mutex and cond. */
+      while (g_atomic_int_get (&test_data.n_jobs_completed) != 1 ||
+             g_atomic_int_get (&test_data.n_free_func_calls) != 1);
+
+      g_assert_cmpuint (g_atomic_int_get (&test_data.n_jobs_started), ==, 1);
+      g_assert_cmpuint (g_atomic_int_get (&test_data.n_jobs_completed), ==, 1);
+      g_assert_cmpuint (g_atomic_int_get (&test_data.n_free_func_calls), ==, 1);
+
+      g_cond_clear (&test_data.cond);
+      g_mutex_clear (&test_data.mutex);
+    }
 }
 
 int
@@ -205,7 +261,7 @@ main (int argc, char *argv[])
   g_test_add_data_func ("/thread_pool/shared", GINT_TO_POINTER (TRUE), test_simple);
   g_test_add_data_func ("/thread_pool/exclusive", GINT_TO_POINTER (FALSE), test_simple);
   g_test_add_data_func ("/thread_pool/create_shared_after_exclusive", GINT_TO_POINTER (FALSE), test_create_first_pool);
-  g_test_add_data_func ("/thread_pool/create_full", GINT_TO_POINTER (FALSE), test_thread_pool_full);
+  g_test_add_data_func ("/thread_pool/create_full", NULL, test_thread_pool_full);
   g_test_add_data_func ("/thread_pool/create_exclusive_after_shared", GINT_TO_POINTER (TRUE), test_create_first_pool);
 
   return g_test_run ();