Keep automatic GC requests local to each interpreter (#8902)

Assisted-by: Codex:GPT-6

crates/vm/src/gc_state.rs

@@ -135,9 +135,7 @@ impl GcGeneration {
 /// empties its own interpreter's objects; another interpreter's stay behind with
 /// the count already zeroed, and untracking one of those must not wrap.
 fn release_count(count: &AtomicUsize) {
-    if count.load(Ordering::Relaxed) > 0 {
-        count.fetch_sub(1, Ordering::Relaxed);
-    }
+    let _ = count.try_update(Ordering::Relaxed, Ordering::Relaxed, |n| n.checked_sub(1));
 }
 
 /// Whether `owner`'s collections act on `obj`.
@@ -552,7 +550,7 @@ impl GcState {
                 // (e.g. a lazily-initialized frame locals cell) that another
                 // thread is blocked on with no way to reach a safepoint —
                 // a deadlock. At a safepoint no such lock is held.
-                crate::signal::schedule_gc();
+                gc.scheduled.store(true, Ordering::Relaxed);
                 return false;
             }
             // Without threading there is no safepoint to defer to and no other
@@ -574,6 +572,8 @@ impl GcState {
         force: bool,
     ) -> CollectResult {
         if !force && !gc.is_enabled() {
+            #[cfg(feature = "threading")]
+            gc.scheduled.store(false, Ordering::Relaxed);
             return CollectResult::default();
         }
 
@@ -582,6 +582,12 @@ impl GcState {
             return CollectResult::default();
         };
 
+        // A busy collector must not consume another interpreter's request.
+        // Clear only after acquiring the lock, before callbacks can request
+        // a later collection.
+        #[cfg(feature = "threading")]
+        gc.scheduled.store(false, Ordering::Relaxed);
+
         let start_time = cfg_select! {
             target_arch = "wasm32" => (),
             _ => std::time::Instant::now(),
@@ -1268,6 +1274,9 @@ pub struct GcInterpreterState {
     pub generations: [GcGeneration; 3],
     /// GC enabled flag
     enabled: AtomicBool,
+    /// Automatic collection requested at this interpreter's next safepoint.
+    #[cfg(feature = "threading")]
+    scheduled: AtomicBool,
     /// Debug flags
     debug: AtomicU32,
     /// Uncollectable objects saved by this interpreter's collections, drained
@@ -1289,6 +1298,8 @@ impl GcInterpreterState {
                 GcGeneration::new(0),    // old[1]
             ],
             enabled: AtomicBool::new(true),
+            #[cfg(feature = "threading")]
+            scheduled: AtomicBool::new(false),
             debug: AtomicU32::new(0),
             garbage: PyMutex::new(Vec::new()),
             py_garbage: ctx.new_list(Vec::new()),
@@ -1304,6 +1315,14 @@ impl GcInterpreterState {
         self.enabled.load(Ordering::Relaxed)
     }
 
+    /// Leave requests pending while a collector is busy, without repeatedly
+    /// entering the bytecode loop's slow path during its Python callbacks.
+    #[cfg(feature = "threading")]
+    #[inline]
+    pub(crate) fn collection_ready(&self) -> bool {
+        self.scheduled.load(Ordering::Relaxed) && !gc_state().collecting.is_locked()
+    }
+
     /// Enable GC
     pub fn enable(&self) {
         self.enabled.store(true, Ordering::Relaxed);
@@ -1524,4 +1543,76 @@ mod tests {
         assert_ne!(first.owner, GC_NO_OWNER);
         assert_ne!(second.owner, GC_NO_OWNER);
     }
+
+    #[cfg(feature = "threading")]
+    #[test]
+    fn automatic_gc_request_stays_with_allocating_interpreter() {
+        let first = crate::Interpreter::without_stdlib(Default::default());
+        let second = crate::Interpreter::without_stdlib(Default::default());
+        first.enter(|vm| vm.state.gc.scheduled.store(false, Ordering::Relaxed));
+        second.enter(|vm| vm.state.gc.scheduled.store(false, Ordering::Relaxed));
+        // An isolated allocation counter makes crossing the threshold
+        // deterministic without depending on the rest of the test process.
+        let allocations = GcState::new();
+        allocations.counts[0].store(1, Ordering::Relaxed);
+        first.enter(|vm| {
+            vm.state.gc.set_threshold(1, None, None);
+            assert!(!allocations.maybe_collect(&vm.state.gc));
+        });
+        second.enter(|vm| {
+            assert!(!vm.state.gc.scheduled.load(Ordering::Relaxed));
+            vm.run_scheduled_gc();
+        });
+        first.enter(|vm| {
+            assert!(vm.state.gc.scheduled.swap(false, Ordering::Relaxed));
+        });
+    }
+
+    #[cfg(feature = "threading")]
+    #[test]
+    fn automatic_gc_request_survives_busy_collector() {
+        let state = interpreter_state();
+        let _guard = gc_state().collecting.lock();
+        state.scheduled.store(true, Ordering::Relaxed);
+        state.collect(0);
+        assert!(state.scheduled.load(Ordering::Relaxed));
+        assert!(!state.collection_ready());
+
+        state.disable();
+        state.collect(0);
+        assert!(!state.scheduled.load(Ordering::Relaxed));
+    }
+
+    #[test]
+    fn release_count_does_not_wrap_during_reset() {
+        use std::sync::Barrier;
+
+        let count = AtomicUsize::new(1);
+        let start = Barrier::new(3);
+        let finish = Barrier::new(3);
+        let mut underflows = 0;
+        std::thread::scope(|scope| {
+            for reset in [false, true] {
+                let (count, start, finish) = (&count, &start, &finish);
+                scope.spawn(move || {
+                    for _ in 0..10_000 {
+                        start.wait();
+                        if reset {
+                            count.store(0, Ordering::Relaxed);
+                        } else {
+                            release_count(count);
+                        }
+                        finish.wait();
+                    }
+                });
+            }
+            for _ in 0..10_000 {
+                count.store(1, Ordering::Relaxed);
+                start.wait();
+                finish.wait();
+                underflows += usize::from(count.load(Ordering::Relaxed) > 1);
+            }
+        });
+        assert_eq!(underflows, 0);
+    }
 }

crates/vm/src/signal.rs

@@ -31,7 +31,6 @@ bitflagset::bitflag! {
     enum EvalBreakerFlag {
         Signal = 0,
         Qsbr = 1,
-        Gc = 2,
         Stop = 3,
         Finalizing = 4,
     }
@@ -189,20 +188,10 @@ mod mt {
         EVAL_BREAKER.insert(EvalBreakerFlag::Finalizing);
     }
 
-    /// Schedule an automatic collection to run at the next bytecode safepoint.
-    pub(crate) fn schedule_gc() {
-        EVAL_BREAKER.insert(EvalBreakerFlag::Gc);
-    }
-
-    /// Clear the scheduled-GC bit, returning whether it had been set.
-    pub(crate) fn take_gc_scheduled() -> bool {
-        EVAL_BREAKER.remove(EvalBreakerFlag::Gc)
-    }
-
     /// Drop every process-wide eval-breaker bit. Tests that assert a single
     /// thread's `stop_requested` must not trip `eval_breaker_pending` have to
     /// start from a clean word: cargo's Windows runner shares the process
-    /// across `#[test]` functions, so a sibling can leave SIGNAL/QSBR/GC/STOP.
+    /// across `#[test]` functions, so a sibling can leave SIGNAL/QSBR/STOP.
     #[cfg(test)]
     pub(crate) fn clear_eval_breaker_for_test() {
         EVAL_BREAKER.clear();
@@ -211,8 +200,7 @@ mod mt {
 
 #[cfg(feature = "threading")]
 pub(crate) use mt::{
-    clear_qsbr_bit, clear_stop_bit, qsbr_bit_set, schedule_gc, set_finalizing_bit, set_qsbr_bit,
-    set_stop_bit, take_gc_scheduled,
+    clear_qsbr_bit, clear_stop_bit, qsbr_bit_set, set_finalizing_bit, set_qsbr_bit, set_stop_bit,
 };
 
 #[cfg(all(test, feature = "threading"))]

crates/vm/src/vm/mod.rs

@@ -3545,7 +3545,7 @@ impl VirtualMachine {
     #[inline]
     pub(crate) fn eval_breaker_tripped(&self) -> bool {
         #[cfg(feature = "threading")]
-        if thread::stop_requested_for_current_thread() {
+        if thread::stop_requested_for_current_thread() || self.state.gc.collection_ready() {
             return true;
         }
         #[cfg(not(target_arch = "wasm32"))]
@@ -3594,7 +3594,7 @@ impl VirtualMachine {
     /// against a thread blocked on a lock this thread would otherwise hold.
     #[cfg(feature = "threading")]
     pub(crate) fn run_scheduled_gc(&self) {
-        if crate::signal::take_gc_scheduled() {
+        if self.state.gc.collection_ready() {
             self.state.gc.collect(0);
         }
     }