Add random-walk controller state to FTP parallel GC (tracking only)

Phase 6 step 2 of bringing the free-threaded parallel GC to feature
parity with the GIL parallel GC. The GIL build adopted the random-walk
adaptive worker controller in Phase 5; this commit adds the same
controller state and the same shared update call to the FTP build, so
both run identical worker-allocation logic.

Changes:

- _PyGCThreadPool gains adaptive_workers, prev_cost_per_obj_ns,
  explore_rng fields (mirroring _PyParallelGCState on the GIL side).
- _PyGC_ThreadPoolInit initialises them: adaptive_workers = min(4,
  num_workers), prev_cost = 0, rng = _PyGC_RandomWalkSeed().
- gc_free_threading.c::gc_collect_internal calls
  _PyGC_RandomWalkUpdate after cleanup_end_ns is recorded — same
  algorithm as the GIL build via the shared helper introduced in
  77c1247.

What this commit does NOT yet do (and why):

  adaptive_workers is tracked but NOT yet used to skip work on idle
  workers. The natural place — an early return in thread_pool_do_work
  for worker_id >= adaptive_workers — deadlocks because FTP's work
  functions (update_refs, mark_heap, scan_heap, propagate) use
  internal phase_barriers that require ALL num_workers participants
  to arrive. Skipping the work skips the phase_barriers and the active
  workers wait forever for the missing ones. The barrier-vs-condvar
  dispatch mismatch is what blocks this; the next commit replaces
  FTP's barrier dispatch with per-worker condvars (mirror of Phase 5.3
  on the GIL side), and at that point adaptive_workers naturally takes
  effect because idle workers don't wake at all.

So: this commit is "controller state and call wiring", not "dispatch
behaviour change". The state evolves identically to the GIL build; the
observed worker count is still num_workers until the dispatch refactor
lands.

Verified: FTP build, all 5 parallel-GC test files pass (177 tests,
7 platform skips). Smoke test (20 collections with 4 workers) completes
without hang.

Author: SonicField

Include/internal/pycore_gc_ft_parallel.h

@@ -336,6 +336,14 @@ typedef struct _PyGCThreadPool {
     // 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.
+    size_t adaptive_workers;
+    double prev_cost_per_obj_ns;
+    uint32_t explore_rng;
+
     // Debug/testing counters (for assertions)
     size_t threads_created;          // Total threads ever created (should equal num_workers-1)
     size_t collections_completed;    // Number of GC collections processed

Python/gc_free_threading.c

@@ -7,6 +7,7 @@
 #include "pycore_freelist.h"      // _PyObject_ClearFreeLists()
 #ifdef Py_PARALLEL_GC
 #include "pycore_gc_ft_parallel.h" // Parallel GC support
+#include "pycore_gc_random_walk.h"  // _PyGC_RandomWalkUpdate()
 #endif
 #include "pycore_genobject.h"     // _PyGen_GetGeneratorFromFrame()
 #include "pycore_initconfig.h"    // _PyStatus_NO_MEMORY()
@@ -2726,6 +2727,22 @@ gc_collect_internal(PyInterpreterState *interp, struct collection_state *state,
     (void)PyTime_PerfCounterRaw(&cleanup_end);
     interp->gc.cleanup_end_ns = cleanup_end;
 
+    // Update adaptive worker count via shared random-walk controller.
+    // Mirrors the GIL parallel GC body in Python/gc.c — both builds use the
+    // same logic to converge on a worker count for this workload.
+    {
+        _PyGCThreadPool *_pool = interp->gc.thread_pool;
+        if (_pool != NULL) {
+            _PyGC_RandomWalkUpdate(
+                interp->gc.cleanup_end_ns - interp->gc.gc_start_ns,
+                state->candidates,
+                &_pool->prev_cost_per_obj_ns,
+                &_pool->explore_rng,
+                &_pool->adaptive_workers,
+                (size_t)_pool->num_workers);
+        }
+    }
+
     // Store the current memory usage, can be smaller now if breaking cycles
     // freed some memory.
     Py_ssize_t last_mem = get_process_mem_usage();

Python/gc_free_threading_parallel.c

@@ -11,6 +11,7 @@
 
 #include "pycore_gc.h"
 #include "pycore_gc_ft_parallel.h"
+#include "pycore_gc_random_walk.h"  // _PyGC_RandomWalkUpdate, _PyGC_RandomWalkSeed
 #include "pycore_interp.h"
 #include "pycore_lock.h"                // PyMutex_Lock/Unlock
 #include "pycore_pystate.h"
@@ -1657,6 +1658,21 @@ thread_pool_do_work(_PyGCThreadPool *pool, int worker_id)
         return;  // No work to do
     }
 
+    // NOTE: pool->adaptive_workers is tracked (see _PyGC_RandomWalkUpdate
+    // call after each collection) but is NOT YET used here to skip work
+    // for idle workers. Reason: FTP work functions (update_refs, mark_heap,
+    // scan_heap, propagate) use internal phase_barriers that require ALL
+    // num_workers participants to arrive. Skipping work in workers >=
+    // adaptive_workers would leave them not participating in those
+    // phase_barriers, causing deadlock.
+    //
+    // The next commit replaces barrier dispatch with per-worker condvars,
+    // at which point idle workers stay asleep and don't enter the work
+    // functions at all. Then adaptive_workers will actually skip work.
+    // For now, the controller state is updated so both builds use the same
+    // logic; the dispatch refactor enables it to take effect on FTP.
+    (void)pool->adaptive_workers;  // intentionally unused this commit
+
 #if GC_DEBUG_ATOMICS
     // Reset TLS stats at start of work
     _PyGC_ATOMIC_RESET_STATS();
@@ -1791,6 +1807,12 @@ _PyGC_ThreadPoolInit(PyInterpreterState *interp, int num_workers)
     pool->threads_created = 0;
     pool->collections_completed = 0;
 
+    // Adaptive worker controller — same as GIL build, see
+    // Include/internal/pycore_gc_random_walk.h.
+    pool->adaptive_workers = (num_workers < 4) ? (size_t)num_workers : 4;
+    pool->prev_cost_per_obj_ns = 0.0;
+    pool->explore_rng = _PyGC_RandomWalkSeed();
+
     // Initialize barriers for synchronization
     // All barriers include all workers (main thread as worker 0)
     _PyGCBarrier_Init(&pool->mark_barrier, num_workers);