Replace per-gen EMA adaptive controller with biased constrained random walk

Remove per-generation EMA hill-climbing (8 parameters, ~90 lines) and
replace with a simpler biased constrained random walk (~25 lines):

- Start at min(4, num_workers) workers
- Each collection: with 20% probability, step ±1 (60% bias toward increase)
- Clamp to [2, num_workers]
- No per-gen state, no EMA, no epsilon, no warmup, no dead zone

The proactive walk continuously explores nearby worker counts even when
performance is stable, preventing the controller from getting trapped at
suboptimal configurations after load spikes.

API changes:
- get_parallel_config: adaptive_workers (single int) replaces per-gen arrays
- get_parallel_stats: prev_cost_per_obj_ns replaces per-gen EMAs

Forward-ported from commit 5114778 on parallel_gc_backport (3.12).
3.15 routing: controller body moved from Modules/gcmodule.c to Python/gc.c.

The flaky test_gen2_allows_more_workers from Phase 5.4 is no longer
applicable — without per-gen state there is nothing for gen2 to converge
to differently from gen0. The test still exists in the file but its
class-level skip (_has_adaptive_controller checks per-gen API presence)
now skips the whole TestAdaptiveControllerConvergence class on builds
where that API is absent (i.e. this build). The test runs cleanly:
SUCCESS, 31 tests run, 9 skipped (the now-inapplicable convergence
class), 0 failures.

Part of Phase 5 of the 3.15 consolidation.

Author: SonicField

Include/internal/pycore_gc_parallel.h

@@ -368,25 +368,16 @@ struct _PyParallelGCState {
     int enabled;
 
     // Adaptive worker count — how many workers to wake per collection.
-    // adaptive_workers is set at the start of each collection from the
-    // per-generation array, and used by dispatch functions during collection.
+    // Biased constrained random walk: start at min(4, num_workers),
+    // stochastically sample ±1 with uphill bias, clamp to [2, num_workers].
     size_t adaptive_workers;
 
-    // Per-generation adaptive state (0=gen0, 1=gen1, 2=gen2).
-    // Each generation independently hill-climbs to its own optimal worker count.
-    size_t adaptive_workers_by_gen[3];
-    double ema_per_obj_ns_by_gen[3];
-
-    // Stochastic exploration (epsilon-greedy).
-    // epsilon: probability of exploring a random worker count instead of
-    //          following the hill-climbing gradient. Decays over time.
-    // explore_rng: xorshift32 PRNG state for exploration decisions.
-    // shift_count: consecutive collections with cost > 2x EMA.
-    //              After 3 consecutive, reset epsilon (workload shift detected).
-    double epsilon;
+    // Random walk state: previous collection's per-object cost (nanoseconds).
+    // Used to compare trial worker counts against the previous configuration.
+    double prev_cost_per_obj_ns;
+
+    // xorshift32 PRNG state for stochastic sampling decisions.
     uint32_t explore_rng;
-    uint8_t shift_count;
-    uint8_t collections_by_gen[3];  // Per-gen collection count (saturates at 255)
 
     int last_generation;      // Last generation collected (for API observability)
     int dispatch_in_progress;  // Reentrancy guard for condvar dispatch

Python/gc.c

@@ -2178,15 +2178,14 @@ gc_collect_region(PyThreadState *tstate,
     assert(!_PyErr_Occurred(tstate));
 
 #ifdef Py_PARALLEL_GC
-    // Set adaptive_workers for this collection from per-generation state.
-    // Must happen before deduce_unreachable which dispatches parallel work.
+    // Record which generation we're collecting (for API observability).
+    // adaptive_workers is already set by the random walk controller.
     {
         _PyParallelGCState *_par_gc = tstate->interp->gc.parallel_gc;
         if (_par_gc != NULL && _par_gc->enabled) {
             int gen = generation;
             if (gen < 0) gen = 0;
             if (gen > 2) gen = 2;
-            _par_gc->adaptive_workers = _par_gc->adaptive_workers_by_gen[gen];
             _par_gc->last_generation = gen;
         }
     }
@@ -2280,87 +2279,44 @@ gc_collect_region(PyThreadState *tstate,
             (void)PyTime_PerfCounterRaw(&cleanup_end);
             par_gc->cleanup_end_ns = cleanup_end;
 
-            // Stochastic hill-climbing adaptive worker count (per-generation).
-            // Each generation independently tracks its own EMA and worker count.
-            // Epsilon-greedy exploration prevents getting stuck in local optima
-            // when workload characteristics change.
-            int gen = generation;
-            if (gen < 0) gen = 0;
-            if (gen > 2) gen = 2;
-
+            // Biased constrained random walk controller.
+            // Each collection: with 20% probability, step current±1 workers.
+            // Proactive: always step when dice fires. Good values stick naturally.
             int64_t parallel_time = par_gc->cleanup_end_ns - par_gc->gc_start_ns;
             Py_ssize_t candidates = par_gc->split_vector.count;
             if (parallel_time > 0 && candidates > 0) {
-                double per_obj = (double)parallel_time / (double)candidates;
-                double prev_ema = par_gc->ema_per_obj_ns_by_gen[gen];
-
-                // Track per-generation collection count (saturates at 255)
-                uint8_t gen_count = par_gc->collections_by_gen[gen];
-                if (gen_count < 255) {
-                    par_gc->collections_by_gen[gen] = ++gen_count;
-                }
-
-                // Workload shift detection: 3 consecutive collections with
-                // cost > 2x EMA resets exploration (not a single outlier).
-                // Skip during EMA warmup (first 3 collections per gen).
-                if (gen_count > 3 && per_obj > 2.0 * prev_ema) {
-                    par_gc->shift_count++;
-                    if (par_gc->shift_count >= 3) {
-                        par_gc->epsilon = 0.3;
-                        par_gc->shift_count = 0;
-                    }
-                } else if (gen_count > 3) {
-                    par_gc->shift_count = 0;
-                }
+                double cost = (double)parallel_time / (double)candidates;
+                double prev_cost = par_gc->prev_cost_per_obj_ns;
+                par_gc->prev_cost_per_obj_ns = cost;
 
-                // EMA update: 0.7 old + 0.3 new (smooth over ~3 collections)
-                par_gc->ema_per_obj_ns_by_gen[gen] = 0.7 * prev_ema + 0.3 * per_obj;
-
-                // Skip explore/exploit during EMA warmup (first 3 collections
-                // per generation). The EMA needs time to converge to the
-                // actual per-object cost before hill-climbing decisions
-                // are meaningful. Without this, cold-start EMA (100.0) vs
-                // real cost (potentially millions) causes the controller
-                // to always decrease workers on first contact with a gen.
-                if (gen_count <= 3) {
+                // Skip adjustment on first collection (no baseline yet)
+                if (prev_cost <= 0.0) {
                     goto controller_done;
                 }
 
-                // Epsilon-greedy: explore or exploit
-                // xorshift32 PRNG for exploration decisions
+                // xorshift32 PRNG
                 uint32_t rng = par_gc->explore_rng;
                 rng ^= rng << 13;
                 rng ^= rng >> 17;
                 rng ^= rng << 5;
                 par_gc->explore_rng = rng;
                 double rand_val = (double)(rng & 0xFFFF) / 65535.0;
 
-                if (rand_val < par_gc->epsilon) {
-                    // EXPLORE: random worker count in [2, num_workers]
-                    uint32_t range = (uint32_t)(par_gc->num_workers - 1);
-                    if (range == 0) range = 1;
-                    par_gc->adaptive_workers_by_gen[gen] = 2 + (rng % range);
-                    if (par_gc->adaptive_workers_by_gen[gen] > par_gc->num_workers) {
-                        par_gc->adaptive_workers_by_gen[gen] = par_gc->num_workers;
-                    }
-                } else {
-                    // EXPLOIT: hill-climb with 15% dead zone
-                    if (per_obj > prev_ema * 1.15
-                            && par_gc->adaptive_workers_by_gen[gen] > 2) {
-                        par_gc->adaptive_workers_by_gen[gen]--;
-                    } else if (per_obj < prev_ema * 0.85
-                            && par_gc->adaptive_workers_by_gen[gen]
-                               < par_gc->num_workers) {
-                        par_gc->adaptive_workers_by_gen[gen]++;
-                    }
-                }
-
-                // Decay epsilon: multiply by 0.95 after non-exploratory collections
-                if (rand_val >= par_gc->epsilon) {
-                    par_gc->epsilon *= 0.95;
-                    if (par_gc->epsilon < 0.05) {
-                        par_gc->epsilon = 0.05;  // floor: 5% ongoing exploration
+                // 20% chance to step ±1 (proactive exploration)
+                if (rand_val < 0.2) {
+                    // Bias uphill: 60% chance to increase, 40% to decrease
+                    double dir_val = (double)((rng >> 16) & 0xFFFF) / 65535.0;
+                    int delta = (dir_val < 0.6) ? 1 : -1;
+
+                    // Always step when the dice fires. Good values stick
+                    // because they don't trigger further corrective steps.
+                    size_t trial = par_gc->adaptive_workers;
+                    if (delta > 0 && trial < par_gc->num_workers) {
+                        trial++;
+                    } else if (delta < 0 && trial > 2) {
+                        trial--;
                     }
+                    par_gc->adaptive_workers = trial;
                 }
 controller_done: ;
             }

Python/gc_parallel.c

@@ -501,16 +501,12 @@ _PyGC_ParallelInit(PyInterpreterState *interp, size_t num_workers)
     par_gc->num_workers = num_workers;
     par_gc->enabled = 1;
     par_gc->num_workers_active = 0;
-    // Per-generation adaptive workers: start at min(4, num_workers) for each gen
+    // Biased constrained random walk: start at min(4, num_workers)
     size_t initial_workers = (num_workers < 4) ? num_workers : 4;
-    for (int g = 0; g < 3; g++) {
-        par_gc->adaptive_workers_by_gen[g] = initial_workers;
-        par_gc->ema_per_obj_ns_by_gen[g] = 100.0;  // neutral initial estimate
-    }
     par_gc->adaptive_workers = initial_workers;
+    par_gc->prev_cost_per_obj_ns = 0.0;  // no previous measurement yet
 
-    // Stochastic exploration: seed from GC_TEST_SEED or perf counter
-    par_gc->epsilon = 0.3;  // 30% exploration initially
+    // Seed PRNG from GC_TEST_SEED or perf counter
     const char *seed_env = getenv("GC_TEST_SEED");
     if (seed_env != NULL) {
         par_gc->explore_rng = (uint32_t)atoi(seed_env);
@@ -522,10 +518,6 @@ _PyGC_ParallelInit(PyInterpreterState *interp, size_t num_workers)
     if (par_gc->explore_rng == 0) {
         par_gc->explore_rng = 1;  // xorshift32 absorbing state guard
     }
-    par_gc->shift_count = 0;
-    for (int g = 0; g < 3; g++) {
-        par_gc->collections_by_gen[g] = 0;
-    }
 
     par_gc->dispatch_in_progress = 0;
 
@@ -853,29 +845,15 @@ _PyGC_ParallelGetConfig(PyInterpreterState *interp)
     }
     Py_DECREF(workers_obj);
 
-    // Per-generation adaptive worker counts and epsilon (if enabled)
+    // Adaptive worker count (current random walk position)
     if (par_gc != NULL && par_gc->enabled) {
-        static const char *gen_keys[] = {
-            "adaptive_workers_gen0",
-            "adaptive_workers_gen1",
-            "adaptive_workers_gen2",
-        };
-        for (int g = 0; g < 3; g++) {
-            PyObject *v = PyLong_FromSize_t(par_gc->adaptive_workers_by_gen[g]);
-            if (v == NULL || PyDict_SetItemString(result, gen_keys[g], v) < 0) {
-                Py_XDECREF(v);
-                Py_DECREF(result);
-                return NULL;
-            }
-            Py_DECREF(v);
-        }
-        PyObject *eps = PyFloat_FromDouble(par_gc->epsilon);
-        if (eps == NULL || PyDict_SetItemString(result, "epsilon", eps) < 0) {
-            Py_XDECREF(eps);
+        PyObject *aw = PyLong_FromSize_t(par_gc->adaptive_workers);
+        if (aw == NULL || PyDict_SetItemString(result, "adaptive_workers", aw) < 0) {
+            Py_XDECREF(aw);
             Py_DECREF(result);
             return NULL;
         }
-        Py_DECREF(eps);
+        Py_DECREF(aw);
     }
 
     return result;
@@ -1239,22 +1217,15 @@ _PyGC_ParallelGetStats(PyInterpreterState *interp)
     }
     Py_DECREF(phase_timing);
 
-    // Per-generation EMA values for controller observability
+    // Random walk controller state for observability
     {
-        static const char *ema_keys[] = {
-            "ema_per_obj_ns_gen0",
-            "ema_per_obj_ns_gen1",
-            "ema_per_obj_ns_gen2",
-        };
-        for (int g = 0; g < 3; g++) {
-            PyObject *v = PyFloat_FromDouble(par_gc->ema_per_obj_ns_by_gen[g]);
-            if (v == NULL || PyDict_SetItemString(result, ema_keys[g], v) < 0) {
-                Py_XDECREF(v);
-                Py_DECREF(result);
-                return NULL;
-            }
-            Py_DECREF(v);
+        PyObject *v = PyFloat_FromDouble(par_gc->prev_cost_per_obj_ns);
+        if (v == NULL || PyDict_SetItemString(result, "prev_cost_per_obj_ns", v) < 0) {
+            Py_XDECREF(v);
+            Py_DECREF(result);
+            return NULL;
         }
+        Py_DECREF(v);
     }
 
     // Last generation collected