tpool: tpool_destroy joins its threads

The lock-free pool detached its worker threads and tpool_destroy()
returned once each had counted itself out under the mutex. A thread
does that before it has ended: it still has the mutex's unlock to
finish and its own exit to run. So tpool_destroy() could return, free
the pool and hand control back - to a caller that frees what the jobs
used, or leaves the process - with pool threads still on their way
out.

The win32-waits-check lane showed a crash of that shape after the
lock-free handoff went in: tpool_wait_test as a Windows binary under
Wine passed every test, printed its last line and then took a
segmentation fault as the process exited. That crash did not reproduce
here in over a hundred runs on a one-CPU box, so this is the fix for
the window the crash sits in, not a confirmed fix for the crash.

The threads are kept and tpool_destroy() joins each one. With that the
exit count, its condition and the lock a worker took on its way out
are gone, and tpool_wait() takes no lock at all; the mutex is left for
the overflow list alone. The lock budget for tpool_worker and
tpool_wait drops from 1 to 0.

The pool for backends without lock-free atomics is unchanged.

tpool_wait_test gains a case for POSIX threads: each worker is marked
with a thread-specific value whose destructor runs as the thread ends,
and every destructor must have run when tpool_destroy() is back, over
300 pools. Against the pool as it was, it fails within the first
hundred with "3 of 4 workers had ended when tpool_destroy returned".

libretro-common/rthreads/tpool.c

@@ -85,6 +85,7 @@ typedef struct tpool_slot
 typedef struct tpool_thread
 {
    struct tpool      *tp;
+   sthread_t         *thread;  /* Joined by tpool_destroy(). */
    retro_eventcount_t wake;    /* Where this thread parks. */
    retro_atomic_int_t parked;  /* 1 from saying so until woken or back. */
    bool               wake_ok;
@@ -93,12 +94,10 @@ typedef struct tpool_thread
 struct tpool
 {
    retro_atomic_int_t enq;        /* Next position to fill. */
-   slock_t           *work_mutex; /* thread_cnt and the exit; the overflow. */
+   slock_t           *work_mutex; /* The overflow. */
    tpool_work_t      *over_first; /* Jobs posted with the ring full, */
    tpool_work_t      *over_last;  /* under work_mutex. */
    retro_atomic_int_t over;       /* How many: read without the mutex. */
-   scond_t           *exit_cond;  /* Signalled when the last thread leaves. */
-   size_t             thread_cnt; /* Threads still running. */
    size_t             thread_num; /* Entries in threads. */
    tpool_thread_t    *threads;
    /* tpool_wait() sleeps here until outstanding reaches zero. */
@@ -284,13 +283,8 @@ static void tpool_worker(void *arg)
       (void)retro_atomic_cas_int(&self->parked, 1, 0);
    }
 
-   /* Unlocking is this thread's last touch of tp: tpool_destroy()
-    * frees it once thread_cnt reads zero under the same lock. */
-   slock_lock(tp->work_mutex);
-   tp->thread_cnt--;
-   if (tp->thread_cnt == 0)
-      scond_signal(tp->exit_cond);
-   slock_unlock(tp->work_mutex);
+   /* Nothing to say on the way out: tpool_destroy() joins this thread,
+    * so tp outlives it without it having to count itself out. */
 }
 
 static void tpool_free(tpool_t *tp)
@@ -301,8 +295,6 @@ static void tpool_free(tpool_t *tp)
          retro_eventcount_free(&tp->threads[i].wake);
    if (tp->work_mutex)
       slock_free(tp->work_mutex);
-   if (tp->exit_cond)
-      scond_free(tp->exit_cond);
    retro_eventcount_free(&tp->idle);
    free(tp->threads);
    free(tp);
@@ -312,6 +304,7 @@ tpool_t *tpool_create_with_stack_size(size_t num, size_t stack_size)
 {
    tpool_t *tp;
    size_t   i;
+   size_t   made = 0;
 
    if (num == 0)
       num = 2;
@@ -321,9 +314,8 @@ tpool_t *tpool_create_with_stack_size(size_t num, size_t stack_size)
 
    tp->threads    = (tpool_thread_t*)calloc(num, sizeof(*tp->threads));
    tp->work_mutex = slock_new();
-   tp->exit_cond  = scond_new();
 
-   if (     !tp->threads || !tp->work_mutex || !tp->exit_cond
+   if (     !tp->threads || !tp->work_mutex
          || !retro_eventcount_init(&tp->idle))
    {
       tpool_free(tp);
@@ -334,30 +326,25 @@ tpool_t *tpool_create_with_stack_size(size_t num, size_t stack_size)
    for (i = 0; i < TPOOL_QUEUE_SIZE; i++)
       retro_atomic_int_init(&tp->slots[i].seq, (int)i);
 
-   /* Create the requested number of threads and detach them. The
-    * count is theirs to lower under the mutex from here on. */
-   slock_lock(tp->work_mutex);
+   /* Create the requested number of threads. They are kept, not
+    * detached: tpool_destroy() joins each one, so none of them is
+    * still running when it returns. */
    for (i = 0; i < num; i++)
    {
-      sthread_t      *thread;
       tpool_thread_t *t = &tp->threads[i];
 
       t->tp = tp;
       if (!(t->wake_ok = retro_eventcount_init(&t->wake)))
          continue;
-      thread = stack_size
+      t->thread = stack_size
             ? sthread_create_with_stack_size(tpool_worker, t, stack_size)
             : sthread_create(tpool_worker, t);
-      if (!thread)
-         continue;
-      tp->thread_cnt++;
-      sthread_detach(thread);
+      if (t->thread)
+         made++;
    }
-   i = tp->thread_cnt;
-   slock_unlock(tp->work_mutex);
 
    /* If no threads were created, clean up and fail. */
-   if (i == 0)
+   if (made == 0)
    {
       tpool_free(tp);
       return NULL;
@@ -395,8 +382,13 @@ void tpool_destroy(tpool_t *tp)
    /* The queue just emptied without passing through tpool_retire(). */
    retro_eventcount_notify(&tp->idle);
 
-   /* Wait for all threads to stop. */
-   tpool_wait(tp);
+   /* Wait for every thread to end - the thread itself, not a count it
+    * lowers on its way out, which left it running past this point and
+    * into whatever the caller did next, the end of the process
+    * included. */
+   for (i = 0; i < tp->thread_num; i++)
+      if (tp->threads[i].thread)
+         sthread_join(tp->threads[i].thread);
 
    tpool_free(tp);
 }
@@ -442,16 +434,9 @@ void tpool_wait(tpool_t *tp)
    if (!tp)
       return;
 
-   /* Teardown: wait for every thread to leave. Under the lock - see
-    * the end of tpool_worker(). */
+   /* Teardown is tpool_destroy()'s to wait out: it joins the threads. */
    if (retro_atomic_load_acquire_int(&tp->stop))
-   {
-      slock_lock(tp->work_mutex);
-      while (tp->thread_cnt != 0)
-         scond_wait(tp->exit_cond, tp->work_mutex);
-      slock_unlock(tp->work_mutex);
       return;
-   }
 
    /* Nothing queued and nothing in progress. outstanding counts both,
     * from the moment tpool_add_work() returns until the item's

libretro-common/samples/rthreads/tpool_wait_test/tpool_wait_test.c

@@ -45,6 +45,14 @@
  *    not crash and does not corrupt the heap.  tpool_destroy is
  *    documented to discard outstanding queued work, so we do NOT
  *    assert on the counter -- we only verify clean teardown.
+ * 5. No worker thread is still running when tpool_destroy returns.
+ *    The pool used to detach its threads and return once each had
+ *    counted itself out, which a thread does before it has ended:
+ *    the caller went on - to free what the jobs used, or to leave
+ *    the process - with pool threads still on their way out.  Each
+ *    worker is marked with a thread-specific value whose destructor
+ *    runs as the thread ends; every destructor must have run by the
+ *    time tpool_destroy is back.  POSIX threads only.
  *
  * What this test does NOT assert
  * ------------------------------
@@ -75,11 +83,18 @@
 #include <rthreads/rthreads.h>
 #include <rthreads/tpool.h>
 
+#if !defined(_WIN32)
+#include <pthread.h>
+#include <retro_atomic.h>
+#define HAVE_EXIT_PROBE 1
+#endif
+
 #define POOL_THREADS         4
 #define WORK_JOBS         1000
 #define ROUNDTRIP_ITERS   2000
 #define STRESS_CYCLES      200
 #define STRESS_JOBS         32
+#define EXIT_CYCLES        300
 
 struct work_ctx
 {
@@ -292,6 +307,73 @@ static int test_stress_destroy_with_pending(void)
    return 0;
 }
 
+/* -----------------------------------------------------------------
+ * Test 5: no worker outlives tpool_destroy.
+ * ----------------------------------------------------------------- */
+#ifdef HAVE_EXIT_PROBE
+static pthread_key_t      exit_key;
+static retro_atomic_int_t exit_marked; /* workers holding a marking job */
+static retro_atomic_int_t exit_ended;  /* marked workers that have ended */
+
+static void exit_probe_dtor(void *p)
+{
+   (void)p;
+   retro_atomic_fetch_add_int(&exit_ended, 1);
+}
+
+/* One per worker: marks the thread, then holds it until every worker
+ * has one, so that no worker takes two. */
+static void exit_mark_job(void *arg)
+{
+   pthread_setspecific(exit_key, arg);
+   retro_atomic_fetch_add_int(&exit_marked, 1);
+   while (retro_atomic_load_acquire_int(&exit_marked) < POOL_THREADS)
+      sthread_yield();
+}
+
+static int test_no_worker_outlives_destroy(void)
+{
+   int i, j;
+
+   if (pthread_key_create(&exit_key, exit_probe_dtor))
+   {
+      printf("[FAIL] test_no_worker_outlives_destroy: pthread_key_create\n");
+      return 1;
+   }
+
+   for (i = 0; i < EXIT_CYCLES; i++)
+   {
+      int ended;
+      tpool_t *tp = tpool_create(POOL_THREADS);
+      if (!tp)
+      {
+         printf("[FAIL] test_no_worker_outlives_destroy: tpool_create returned NULL at i=%d\n", i);
+         return 1;
+      }
+      retro_atomic_int_init(&exit_marked, 0);
+      retro_atomic_int_init(&exit_ended, 0);
+      for (j = 0; j < POOL_THREADS; j++)
+         tpool_add_work(tp, exit_mark_job, &exit_key);
+      tpool_wait(tp);
+      tpool_destroy(tp);
+
+      ended = retro_atomic_load_acquire_int(&exit_ended);
+      if (ended != POOL_THREADS)
+      {
+         printf("[FAIL] test_no_worker_outlives_destroy: cycle %d, %d of %d"
+               " workers had ended when tpool_destroy returned\n",
+               i, ended, POOL_THREADS);
+         return 1;
+      }
+   }
+
+   pthread_key_delete(exit_key);
+   printf("[PASS] test_no_worker_outlives_destroy (%d cycles x %d threads)\n",
+         EXIT_CYCLES, POOL_THREADS);
+   return 0;
+}
+#endif
+
 int main(void)
 {
    int failures = 0;
@@ -300,6 +382,9 @@ int main(void)
    failures += test_zero_threads_default();
    failures += test_create_destroy_no_work();
    failures += test_stress_destroy_with_pending();
+#ifdef HAVE_EXIT_PROBE
+   failures += test_no_worker_outlives_destroy();
+#endif
 
    if (failures)
    {

tools/lock_acquire_budget.ref

@@ -30,15 +30,15 @@ audio_thread_loop                0
 # here) takes the lock only when such a task is due, and settling it
 # reads the task's flags, which are atomic.
 retro_task_threaded_gather       6
-# A job is handed over through a ring of slots and takes no lock; the
-# one a worker takes here is at its exit, at teardown. Jobs posted
-# past the ring's size wait on a list under the lock (tpool_take, a
-# call of its own). Finishing an item retires it on the outstanding
-# count and wakes tpool_wait() through an eventcount, without the lock.
-tpool_worker                     1
+# A job is handed over through a ring of slots and takes no lock, and
+# a worker takes none at its exit either: tpool_destroy() joins it.
+# Jobs posted past the ring's size wait on a list under the lock
+# (tpool_take, a call of its own). Finishing an item retires it on the
+# outstanding count and wakes tpool_wait() through an eventcount,
+# without the lock.
+tpool_worker                     0
 tpool_help                       0
-# Only at teardown, waiting for the threads to leave.
-tpool_wait                       1
+tpool_wait                       0
 # The recorder's pushes and its encoder hand frames over through
 # retro_spsc rings and sleep on eventcounts; nothing here.
 ffmpeg_push_video                0