youknowone · Sep 22, 2026 · Sep 22, 2026
diff --git a/crates/vm/src/vm/interpreter.rs b/crates/vm/src/vm/interpreter.rs
@@ -1733,6 +1733,125 @@ for _ in range(40):
         worker.join().expect("worker panicked");
     }
 
+    /// `STOP_BIT` is process-wide and `start_the_world` in any interpreter
+    /// clears it. A worker must still park from its own `stop_requested`.
+    #[cfg(all(feature = "threading", feature = "rustpython-compiler"))]
+    #[test]
+    fn stop_the_world_parks_when_global_stop_bit_is_cleared() {
+        use crate::compiler::Mode;
+        use alloc::sync::Arc;
+        use core::{
+            sync::atomic::{AtomicBool, AtomicU64, Ordering},
+            time::Duration,
+        };
+
+        let main = Interpreter::without_stdlib(Default::default());
+        let sub = main.create_subinterpreter();
+        let sub_state = sub.enter(|vm| vm.state.clone());
+
+        let progress = Arc::new(AtomicU64::new(0));
+        let stop = Arc::new(AtomicBool::new(false));
+        let flip = Arc::new(AtomicBool::new(true));
+
+        let progress_worker = Arc::clone(&progress);
+        let stop_worker = Arc::clone(&stop);
+        let worker = sub.enter(|vm| {
+            let thread_vm = vm.new_thread();
+            std::thread::spawn(move || {
+                thread_vm.run(|vm| {
+                    let source = "x = 1 + 1\n";
+                    let code = vm
+                        .compile(source, Mode::Exec, "<spin>")
+                        .map_err(|err| err.into_pyexception(vm, Some(source)))
+                        .unwrap();
+                    while !stop_worker.load(Ordering::Acquire) {
+                        let scope = vm.new_scope_with_builtins();
+                        vm.run_code_obj(code.clone(), scope).unwrap();
+                        progress_worker.fetch_add(1, Ordering::Release);
+                    }
+                });
+            })
+        });
+
+        let deadline = std::time::Instant::now() + Duration::from_secs(10);
+        while progress.load(Ordering::Acquire) == 0 {
+            assert!(
+                std::time::Instant::now() < deadline,
+                "worker never started making progress"
+            );
+            std::thread::yield_now();
+        }
+
+        let flip_flag = Arc::clone(&flip);
+        let flicker_cycles = Arc::new(AtomicU64::new(0));
+        let flicker_cycles_thread = Arc::clone(&flicker_cycles);
+        let flicker = std::thread::spawn(move || {
+            while flip_flag.load(Ordering::Acquire) {
+                crate::signal::set_stop_bit();
+                crate::signal::clear_stop_bit();
+                flicker_cycles_thread.fetch_add(1, Ordering::Release);
+            }
+        });
+
+        while flicker_cycles.load(Ordering::Acquire) == 0 {
+            assert!(
+                std::time::Instant::now() < deadline,
+                "STOP_BIT flicker never ran"
+            );
+            std::thread::yield_now();
+        }
+
+        // Stop from a thread of its own so a hang fails the test instead of
+        // stalling the suite. Keep flickering until after the stop completes.
+        let (tx, rx) = std::sync::mpsc::channel();
+        let stop_state = sub_state.clone();
+        let stopper = std::thread::spawn(move || {
+            stop_state.stop_the_world.stop_the_world(&stop_state);
+            let stopped = tx.send(());
+            (stop_state, stopped)
+        });
+
+        let stopped = rx.recv_timeout(Duration::from_secs(10));
+
+        let parked_at = progress.load(Ordering::Acquire);
+        std::thread::sleep(Duration::from_millis(50));
+        let parked_after = progress.load(Ordering::Acquire);
+
+        // Release before joining: if the stop hung, stopping the worker lets
+        // its slot go away so the stopper can finish.
+        flip.store(false, Ordering::Release);
+        if stopped.is_err() {
+            stop.store(true, Ordering::Release);
+        }
+        flicker.join().expect("stop-bit flicker panicked");
+        let (stop_state, sent) = stopper.join().expect("stopper panicked");
+        if stopped.is_ok() {
+            stop_state.stop_the_world.start_the_world(&stop_state);
+        }
+        sent.expect("send");
+        assert!(
+            stopped.is_ok(),
+            "stop-the-world hung while STOP_BIT was flickering"
+        );
+        assert_eq!(
+            parked_after, parked_at,
+            "subinterpreter thread kept running after STOP_BIT was cleared"
+        );
+
+        let resumed_from = progress.load(Ordering::Acquire);
+        let resume_deadline = std::time::Instant::now() + Duration::from_secs(10);
+        while progress.load(Ordering::Acquire) == resumed_from {
+            assert!(
+                std::time::Instant::now() < resume_deadline,
+                "worker never resumed after start_the_world"
+            );
+            std::thread::yield_now();
+        }
+
+        stop.store(true, Ordering::Release);
+        worker.join().expect("worker panicked");
+    }
+
     /// Entering a subinterpreter from inside the parent's `enter` must attach
     /// the subinterpreter's thread slot (and detach the parent's). Otherwise the
     /// thread runs the sub's bytecode with a DETACHED slot, and a collector

diff --git a/crates/vm/src/vm/mod.rs b/crates/vm/src/vm/mod.rs
@@ -567,11 +567,9 @@ impl StopTheWorldState {
         drop(registry);
         self.thread_countdown.store(0, Ordering::Release);
         self.requester.store(0, Ordering::Relaxed);
-        // Every non-requester thread's `stop_requested` was just cleared
-        // above, so the shared fast-path bit can be cleared too. Safe here
-        // (and only here / `reset_after_fork`) because both run under the
-        // single stop-the-world exclusion, released below, so no new
-        // requester can be setting the bit concurrently.
+        // Drop the process-wide stop hint. Another interpreter may still
+        // have `stop_requested` threads; those keep parking via the
+        // per-thread check in `eval_breaker_tripped`.
         crate::signal::clear_stop_bit();
         #[cfg(debug_assertions)]
         self.debug_assert_all_non_requester_detached(state);
@@ -3500,20 +3498,20 @@ impl VirtualMachine {
         self.get_method(obj, method_name)
     }
 
-    /// Single relaxed load of the shared eval-breaker word, folding in every
-    /// condition that used to require its own check: pending signals, QSBR
-    /// reclamation, scheduled GC, and (under `threading`) finalization and
-    /// stop-the-world. Finalizing sets `FINALIZING_BIT` and any per-thread
-    /// `stop_requested` store also sets `STOP_BIT` (see signal.rs), so this
-    /// no longer needs to read `state.finalizing` or the
-    /// `CURRENT_STOP_REQUESTED` thread-local itself — both are folded into
-    /// the same atomic word `eval_breaker_pending` already loads for
-    /// signals/QSBR/GC. The slow path (`check_signals`) re-derives the exact
-    /// per-thread answer (this thread's own `stop_requested`, `finalizing` +
-    /// `is_main_thread`) once it is taken, so semantics are unchanged; only
-    /// the common (nothing pending) case gets cheaper.
+    /// Fast path for the bytecode loop: pending signals, QSBR, scheduled GC,
+    /// finalization, and stop-the-world.
+    ///
+    /// `STOP_BIT` is a process-wide hint set when any interpreter asks a
+    /// thread to park. Each interpreter's `start_the_world` clears that hint,
+    /// even if another interpreter still has `stop_requested` threads, so the
+    /// per-thread flag is checked first. Missing it lets a worker skip
+    /// `check_signals` and never park, so `stop_the_world` waits forever.
     #[inline]
     pub(crate) fn eval_breaker_tripped(&self) -> bool {
+        #[cfg(feature = "threading")]
+        if thread::stop_requested_for_current_thread() {
+            return true;
+        }
         #[cfg(not(target_arch = "wasm32"))]
         {
             crate::signal::eval_breaker_pending()