Replace FTP pool barrier dispatch with per-worker condvar dispatch

Mirror of Phase 5.3 on the FTP side. Replaces broadcast mark_barrier/
done_barrier dispatch in the persistent thread pool with per-worker
(wake_mutex, wake_cond, wake_flag) and pool-level (done_mutex, done_cond,
workers_done_count). Main thread wakes only [1, adaptive_workers) helpers
and waits for that many done signals. Idle workers stay asleep.

Together with the previous controller-state commit (5ef3a61) and
the helper-extraction commit (12a3440), the FTP build now uses the
same random-walk controller logic as the GIL build via the shared
_PyGC_RandomWalkUpdate helper. The adaptive count actually takes effect
on FTP now: _PyGC_GetParallelWorkers returns pool->adaptive_workers, so
FTP collections scale per-collection like GIL.

Component changes:

- Include/internal/pycore_gc_barrier.h: add _PyGCBarrier_Resize. Safe
  only when no one is waiting; used by the FTP dispatch to size
  phase_barrier to adaptive_workers per dispatch (so internal multi-
  phase work like update_refs init/compute only waits for active
  participants).

- _PyGCWorkerState: add wake_mutex, wake_cond, wake_flag.

- _PyGCThreadPool: add done_mutex, done_cond, workers_done_count,
  dispatch_in_progress. Drop mark_barrier and done_barrier (no longer
  used for dispatch). Keep phase_barrier; it's now resized per dispatch.

- _PyGC_FTDispatchAndWait: new helper (mirror of GIL _PyGC_DispatchAndWait).
  Resizes phase_barrier, resets done counter, wakes helpers, runs main's
  share (worker 0), waits for helper completions. Has reentrancy guard
  although FTP STW collections shouldn't ever be reentrant; the four call
  sites assert reentrant == 0.

- thread_pool_worker_func: replace mark_barrier wait with per-worker
  condvar wait; replace done_barrier wait with done_mutex/cond signal.

- thread_pool_do_work: removed the adaptive_workers early-return added
  in the previous commit; no longer needed because idle workers don't
  wake at all under condvar dispatch.

- _PyGC_ThreadPoolInit: initialise the new sync primitives, per-worker
  condvars, drop mark/done barrier Init. Update error-path cleanup.

- _PyGC_ThreadPoolFini: wake all helpers via per-worker condvars (so
  they see shutdown and exit), Fini per-worker mutex/cond and pool-level
  done_mutex/done_cond. Drop mark/done barrier Fini.

- Four pool dispatch call sites (PropagateAliveWithPool,
  ParallelUpdateRefsWithPool, ParallelMarkHeapWithPool,
  ParallelScanHeapWithPool): replace mark_barrier + thread_pool_do_work(0)
  + done_barrier with _PyGC_FTDispatchAndWait(pool, adaptive_workers).
  The ScanHeap merge loop now iterates adaptive_workers (only active
  workers contribute results).

- gcmodule.c get_parallel_config (FTP branch): expose both num_workers
  (configured max) and adaptive_workers (current); mirror of GIL build's
  config API. Previously was inadvertently returning adaptive_workers as
  num_workers via _PyGC_GetParallelWorkers, breaking the historical
  meaning of the num_workers key.

- _PyGC_FTParallelGetStats: same num_workers/adaptive_workers split.

- _PyGC_GetParallelWorkers: now returns pool->adaptive_workers (the
  earlier revert of this is now superseded by the dispatch refactor that
  makes it safe to return the adaptive count).

- Loosen assertions in PropagateAliveWithPool and UpdateRefs/MarkHeap/
  ScanHeap that previously required state->num_workers == pool->num_workers
  to allow state->num_workers <= pool->num_workers.

Verified end-to-end:
- FTP smoke test (20 collections, 4 workers, enable/disable cycle): OK
- FTP parallel-GC suite: 177 tests pass, 7 skips
- FTP full CPython suite: 481 OK, 0 failures
- GIL full CPython suite: 479 OK, 0 failures (no regression)

The barrier mark_barrier and done_barrier struct fields can be cleaned
up in a follow-up (parallel to GIL Phase 5 cleanup that removed the
same fields from _PyParallelGCState).

Author: SonicField

Include/internal/pycore_gc_barrier.h

@@ -97,6 +97,22 @@ _PyGCBarrier_Fini(_PyGCBarrier *barrier)
     _PyGC_MUTEX_FINI(&barrier->lock);
 }
 
+// Re-size a barrier to a new capacity. Safe to call only when no thread
+// is currently inside _PyGCBarrier_Wait for this barrier (typically: at
+// the start of each dispatch, before any worker is woken). Used by the
+// FTP parallel GC to size the phase_barrier to adaptive_workers each
+// dispatch so only active workers participate.
+static inline void
+_PyGCBarrier_Resize(_PyGCBarrier *barrier, unsigned int new_capacity)
+{
+    assert(new_capacity > 0);
+    _PyGC_MUTEX_LOCK(&barrier->lock);
+    barrier->capacity = new_capacity;
+    barrier->num_left = new_capacity;
+    // epoch unchanged; no one should be waiting
+    _PyGC_MUTEX_UNLOCK(&barrier->lock);
+}
+
 // Wait at barrier - blocks until all threads arrive
 static inline void
 _PyGCBarrier_Wait(_PyGCBarrier *barrier)

Include/internal/pycore_gc_ft_parallel.h

@@ -273,6 +273,14 @@ typedef struct _PyGCWorkerState {
     // Per-phase atomic stats - copied from TLS after work completes
     _PyGCAtomicWorkerStats atomic_stats;
 #endif
+
+    // Per-worker condvar for targeted wakeup (adaptive worker count).
+    // Mirrors the GIL parallel GC. Replaces the previous mark_barrier
+    // broadcast dispatch: main thread sets wake_flag and signals wake_cond
+    // for only the active workers; idle workers stay asleep.
+    PyMUTEX_T wake_mutex;
+    PyCOND_T wake_cond;
+    int wake_flag;  // 0=sleeping, 1=wake up
 } _PyGCWorkerState;
 
 // Work descriptor - describes work for a single collection
@@ -328,18 +336,28 @@ typedef struct _PyGCThreadPool {
     // Current work descriptor - set by main thread before signalling workers
     _PyGCWorkDescriptor *current_work;
 
-    // Barrier synchronization (like GIL-based parallel GC)
-    _PyGCBarrier mark_barrier;     // Workers wait here for work
-    _PyGCBarrier done_barrier;     // All workers wait here when done
-    _PyGCBarrier phase_barrier;    // For multi-phase operations
+    // Per-collection done signaling. Workers signal done_cond when they
+    // finish; main waits on done_cond until workers_done_count reaches
+    // the count of woken workers. Mirrors the GIL parallel GC.
+    PyMUTEX_T done_mutex;
+    PyCOND_T done_cond;
+    volatile int workers_done_count;  // Protected by done_mutex
+
+    // Reentrancy guard for the dispatch path.
+    int dispatch_in_progress;
+
+    // Phase barrier — used INSIDE work functions (e.g. update_refs has
+    // init/compute phases). Resized to the active worker count at the
+    // start of each dispatch so only active workers participate.
+    _PyGCBarrier phase_barrier;
 
     // Worker control
     int shutdown;           // 1 = pool is shutting down (use atomics)
 
     // Adaptive worker count — see Include/internal/pycore_gc_random_walk.h.
     // Same controller as the GIL parallel GC for identical behaviour.
     // adaptive_workers ∈ [2, num_workers]; workers with worker_id >=
-    // adaptive_workers no-op for this collection.
+    // adaptive_workers do not wake this dispatch.
     size_t adaptive_workers;
     double prev_cost_per_obj_ns;
     uint32_t explore_rng;
@@ -529,29 +547,28 @@ struct _PyGCScanHeapResult {
 };
 
 // Get number of parallel GC workers for the next collection.
-// Returns 0 if parallel GC is disabled, otherwise the worker count.
+// Returns 0 if parallel GC is disabled, otherwise the adaptive worker
+// count chosen by the random-walk controller (see pycore_gc_random_walk.h).
 //
-// Currently returns pool->num_workers (the configured maximum), NOT the
-// adaptive count. The pool dispatch path
-// (_PyGC_ParallelPropagateAliveWithPool and similar) asserts that the
-// passed num_workers matches pool->num_workers and that the internal
-// phase_barrier (sized num_workers) receives that many participants.
-// Returning adaptive_workers here would break both invariants.
+// Pool dispatch (_PyGC_FTDispatchAndWait) wakes only [0, adaptive_workers)
+// of the pool's helpers via per-worker condvars; idle workers stay asleep
+// and don't enter the work functions. The internal phase_barrier is resized
+// to adaptive_workers per dispatch so it only waits for active participants.
 //
-// Plumbing the adaptive count through requires the pool dispatch refactor:
-// per-worker condvars instead of mark_barrier broadcast, dynamically-sized
-// (or per-dispatch) phase_barriers. Once that lands, change the body to:
-//   _PyGCThreadPool *pool = gc->thread_pool;
-//   return pool ? (int)pool->adaptive_workers : gc->parallel_gc_num_workers;
-// and FTP collections will scale per-collection like the GIL build does.
+// Falls back to configured num_workers only if the pool isn't initialised
+// yet (e.g. enable_parallel hasn't completed). Once pool exists, the
+// adaptive count owned by the pool is authoritative.
 static inline int
 _PyGC_GetParallelWorkers(PyInterpreterState *interp)
 {
     struct _gc_runtime_state *gc = &interp->gc;
     if (!gc->parallel_gc_enabled) {
         return 0;
     }
-    // num_workers is required and validated by gc.enable_parallel()
+    _PyGCThreadPool *pool = gc->thread_pool;
+    if (pool != NULL) {
+        return (int)pool->adaptive_workers;
+    }
     return gc->parallel_gc_num_workers;
 }
 

Modules/gcmodule.c

@@ -730,8 +730,11 @@ gc_get_parallel_config_impl(PyObject *module)
         return NULL;
     }
 
+    // num_workers = the configured maximum (what enable_parallel was called
+    // with). adaptive_workers = the current count chosen by the random-walk
+    // controller; <= num_workers. Mirror of the GIL build's config API.
     int num_workers = interp->gc.parallel_gc_enabled ?
-                      _PyGC_GetParallelWorkers(interp) : 0;
+                      interp->gc.parallel_gc_num_workers : 0;
     PyObject *workers = PyLong_FromLong(num_workers);
     if (workers == NULL || PyDict_SetItemString(result, "num_workers", workers) < 0) {
         Py_XDECREF(workers);
@@ -740,6 +743,16 @@ gc_get_parallel_config_impl(PyObject *module)
     }
     Py_DECREF(workers);
 
+    if (interp->gc.parallel_gc_enabled && interp->gc.thread_pool != NULL) {
+        PyObject *aw = PyLong_FromSize_t(interp->gc.thread_pool->adaptive_workers);
+        if (aw == NULL || PyDict_SetItemString(result, "adaptive_workers", aw) < 0) {
+            Py_XDECREF(aw);
+            Py_DECREF(result);
+            return NULL;
+        }
+        Py_DECREF(aw);
+    }
+
     // Parallel cleanup is available in FTP builds
     if (PyDict_SetItemString(result, "parallel_cleanup", Py_True) < 0) {
         Py_DECREF(result);