youknowone · Oct 21, 2026 · Oct 19, 2026 · Oct 19, 2026 · Oct 19, 2026 · Oct 21, 2026
diff --git a/.cspell.dict/rust-more.txt b/.cspell.dict/rust-more.txt
@@ -83,6 +83,8 @@ trai
 ulonglong
 unic
 unistd
+unparked
+unparking
 unraw
 unsync
 wasip1

diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -275,6 +275,7 @@ openssl = "0.10.80"
 openssl-sys = "0.9.110"
 optional = "0.5"
 parking_lot = "0.12.3"
+parking_lot_core = "0.9.12"
 paste = "1.0.15"
 pem-rfc7468 = "1.0"
 pkcs8 = "0.11"

diff --git a/crates/vm/Cargo.toml b/crates/vm/Cargo.toml
@@ -21,7 +21,7 @@ opcode-histogram = []
 flame-it = ["flame", "flamer"]
 freeze-stdlib = ["encodings"]
 jit = ["rustpython-jit"]
-threading = ["rustpython-common/threading"]
+threading = ["rustpython-common/threading", "dep:parking_lot_core"]
 gc = []
 compiler = ["parser", "codegen", "rustpython-compiler"]
 ast = ["ruff_python_ast", "ruff_text_size"]
@@ -68,6 +68,7 @@ num-integer = { workspace = true }
 num-traits = { workspace = true }
 num_enum = { workspace = true }
 parking_lot = { workspace = true }
+parking_lot_core = { workspace = true, optional = true }
 paste = { workspace = true }
 rapidhash = { workspace = true }
 scopeguard = { workspace = true }

diff --git a/crates/vm/src/gc_state.rs b/crates/vm/src/gc_state.rs
@@ -5,6 +5,8 @@
 use crate::common::linked_list::LinkedList;
 use crate::common::lock::{PyMutex, PyRwLock};
 use crate::object::{GC_NO_OWNER, GC_PERMANENT, GC_REACHABLE, GC_UNTRACKED, GcLink, GcOwner};
+#[cfg(feature = "threading")]
+use crate::vm::stop_the_world::StopTheWorldGroup;
 use crate::{AsObject, PyObject, PyObjectRef};
 use core::ptr::NonNull;
 use core::sync::atomic::{AtomicBool, AtomicU16, AtomicU32, AtomicUsize, Ordering};
@@ -191,114 +193,6 @@ type GcBuildHasher = core::hash::BuildHasherDefault<GcPtrHasher>;
 type GcSet<T> = std::collections::HashSet<T, GcBuildHasher>;
 type GcMap<K, V> = std::collections::HashMap<K, V, GcBuildHasher>;
 
-/// RAII barrier that parks every other thread for the pointer-reading phases
-/// of a collection and lets them run again before finalizers execute.
-///
-/// Reference subtraction, the reachability walk and the strong-reference
-/// snapshot dereference the interpreter state of every tracked object,
-/// including the `localsplus` of frames that other threads are actively
-/// executing. Those writes carry no synchronization, so the reads are only
-/// well-defined while all other threads are parked at a safepoint. Restarting
-/// happens explicitly once the snapshot has pinned every object; `Drop` is a
-/// backstop that also restarts on the early-return paths.
-///
-/// A collection acts on one interpreter's objects, but its candidates include
-/// the ones no interpreter owns, which every interpreter can reference and so
-/// incref. Reading a refcount that another interpreter is changing is what
-/// makes an object look unreachable when it is not, so every live interpreter
-/// is stopped, not just the collecting one. Stopping in `runtime` id order
-/// is deterministic. One process-wide STW transaction covers the entire group,
-/// preventing a single-interpreter requester from stopping a collector while
-/// that collector waits for a different interpreter.
-#[cfg(feature = "threading")]
-struct CollectStopTheWorld {
-    /// Entire interpreter snapshot, retained through restart even on unwind.
-    /// Dropping the last state can release Python objects and run finalizers,
-    /// so both stopped and not-yet-stopped states must outlive the transaction
-    /// and the collection's generation locks.
-    states: Vec<crate::common::rc::PyRc<crate::vm::PyGlobalState>>,
-    stopped_count: usize,
-    /// Keeps interpreters from registering between the snapshot below and the
-    /// restart. One registered in that window would be missing from `states`,
-    /// so its bootstrap would keep running — and mutating the shared generation
-    /// lists — while this collection reads them.
-    admission: Option<parking_lot::MutexGuard<'static, ()>>,
-    /// Acquired before admission and retained until every state is restarted.
-    transaction: Option<crate::vm::StopTheWorldTransaction>,
-    restarted: bool,
-}
-
-#[cfg(feature = "threading")]
-impl CollectStopTheWorld {
-    /// Request stop-the-world on every live interpreter when the current thread
-    /// has an attached VM. Falls back to no barrier when no VM is attached (the
-    /// tracked-object reads then run without other threads only if the caller
-    /// guarantees it).
-    fn new() -> Self {
-        // No attached VM means no interpreter is running Python on this thread;
-        // keep the historical no-barrier fallback.
-        if !crate::vm::thread::current_vm_is_set() {
-            return Self {
-                states: Vec::new(),
-                stopped_count: 0,
-                admission: None,
-                transaction: None,
-                restarted: true,
-            };
-        }
-
-        // Accumulate into a live `Self` rather than a bare Vec: if a later
-        // `stop_the_world` unwinds, dropping this guard restarts the
-        // interpreters already stopped, instead of leaving their threads parked
-        // and their exclusion held forever.
-        let transaction = crate::vm::StopTheWorldTransaction::acquire();
-        let mut guard = Self {
-            states: Vec::new(),
-            stopped_count: 0,
-            admission: Some(crate::vm::runtime::lock_admission_for_stop()),
-            transaction: Some(transaction),
-            restarted: false,
-        };
-        guard.states = crate::vm::runtime::live_interpreter_states();
-        for state in &guard.states {
-            state
-                .stop_the_world
-                .stop_with_transaction(state, guard.transaction.as_ref().unwrap());
-            guard.stopped_count += 1;
-        }
-        guard
-    }
-
-    /// Restart the world. Idempotent.
-    fn restart(&mut self) {
-        if self.restarted {
-            return;
-        }
-        self.restarted = true;
-        // Reverse of the stop order. The references stay until this guard is
-        // dropped; see the field comment.
-        for state in self.states[..self.stopped_count].iter().rev() {
-            state.stop_the_world.start_the_world(state);
-        }
-        // Nothing is parked any more, so registration may resume.
-        self.admission = None;
-        self.transaction = None;
-    }
-
-    /// Whether this collection actually stopped the world.
-    #[cfg(all(unix, debug_assertions))]
-    fn is_stopped(&self) -> bool {
-        self.stopped_count != 0
-    }
-}
-
-#[cfg(feature = "threading")]
-impl Drop for CollectStopTheWorld {
-    fn drop(&mut self) {
-        self.restart();
-    }
-}
-
 /// The process-wide object lists every interpreter's collections walk.
 ///
 /// Interpreter-owned policy and results live in [`GcInterpreterState`]; what is
@@ -639,7 +533,7 @@ impl GcState {
         // Closing it fully would require making those locks stop-the-world
         // aware; the exclusion above only serializes the fork/GC requesters.
         #[cfg(feature = "threading")]
-        let mut stw = CollectStopTheWorld::new();
+        let mut stw = StopTheWorldGroup::new();
 
         // Step 1: Gather objects from generations 0..=generation
         // Hold read locks for the entire scan to prevent concurrent modifications.
@@ -1048,7 +942,7 @@ impl GcState {
             // liveness check and weakref closure in the same stopped phase.
             // No callbacks, finalizers or tp_clear may run under this guard.
             #[cfg(feature = "threading")]
-            let mut clear_stw = CollectStopTheWorld::new();
+            let mut clear_stw = StopTheWorldGroup::new();
 
             // Untrack dead objects BEFORE clearing them, mirroring the
             // untrack-then-clear ordering of the refcount dealloc path.

diff --git a/crates/vm/src/stdlib/posix.rs b/crates/vm/src/stdlib/posix.rs
@@ -681,7 +681,12 @@ pub mod module {
         }
     }
 
-    fn py_os_before_fork(vm: &VirtualMachine) {
+    #[cfg(feature = "threading")]
+    use crate::vm::stop_the_world::ForkStopTheWorld;
+    #[cfg(not(feature = "threading"))]
+    struct ForkStopTheWorld;
+
+    fn py_os_before_fork(vm: &VirtualMachine) -> ForkStopTheWorld {
         let before_forkers: Vec<PyObjectRef> = vm.state.before_forkers.lock().clone();
         // functions must be executed in reversed order as they are registered
         // only for before_forkers, refer: test_register_at_fork in test_posix
@@ -692,10 +697,21 @@ pub mod module {
         crate::stdlib::_imp::acquire_imp_lock_for_fork(vm);
 
         #[cfg(feature = "threading")]
-        vm.state.stop_the_world.stop_the_world(&vm.state);
+        {
+            ForkStopTheWorld::new()
+        }
+        #[cfg(not(feature = "threading"))]
+        {
+            ForkStopTheWorld
+        }
     }
 
-    fn py_os_after_fork_child(vm: &VirtualMachine) {
+    fn py_os_after_fork_child(vm: &VirtualMachine, _stopped: &mut ForkStopTheWorld) {
+        #[cfg(feature = "threading")]
+        unsafe {
+            _stopped.begin_child_repair();
+        }
+
         // The interpreter registry is reachable from every thread, so repair it
         // before anything enumerates interpreters.
         #[cfg(all(unix, feature = "threading"))]
@@ -711,20 +727,14 @@ pub mod module {
         // if we try to acquire them. This must happen before anything else.
         #[cfg(feature = "threading")]
         reinit_locks_after_fork(vm);
+        #[cfg(feature = "threading")]
+        _stopped.retain_thread_slots(&vm.state);
 
         // The collector stops every interpreter, so interpreters other than the
         // forking one must be repaired too; otherwise the child's first
         // collection waits for threads that did not survive the fork.
         #[cfg(all(unix, feature = "threading"))]
-        reinit_other_interpreters_after_fork(vm);
-
-        // Reinit per-object IO buffer locks on std streams.
-        // BufferedReader/Writer/TextIOWrapper use PyThreadMutex which can be
-        // held by dead parent threads, causing deadlocks on any IO in the child.
-        #[cfg(feature = "threading")]
-        unsafe {
-            crate::stdlib::_io::reinit_std_streams_after_fork(vm)
-        };
+        reinit_other_interpreters_after_fork(vm, _stopped);
 
         // Phase 2: Reset low-level atomic state (no locks needed).
         crate::signal::clear_after_fork();
@@ -774,6 +784,20 @@ pub mod module {
         vm.signal_handlers
             .get_or_init(crate::signal::SignalHandlers::default);
 
+        #[cfg(feature = "threading")]
+        _stopped.finish_child_repair();
+
+        // Reinit per-object IO buffer locks on std streams.
+        // BufferedReader/Writer/TextIOWrapper use PyThreadMutex which can be
+        // held by dead parent threads, causing deadlocks on any IO in the child.
+        // Lookup can call Python (for example sys.__getattr__), so core repair
+        // and transaction release must precede it. Such hooks must still avoid
+        // using inherited IO locks before this existing repair completes.
+        #[cfg(feature = "threading")]
+        unsafe {
+            crate::stdlib::_io::reinit_std_streams_after_fork(vm)
+        };
+
         // Phase 4: Run Python-level at-fork callbacks.
         let after_forkers_child: Vec<PyObjectRef> = vm.state.after_forkers_child.lock().clone();
         run_at_forkers(after_forkers_child, false, vm);
@@ -831,7 +855,7 @@ pub mod module {
     /// # Safety
     /// Must only be called after `fork()` in the child, when no other threads exist.
     #[cfg(all(unix, feature = "threading"))]
-    fn reinit_other_interpreters_after_fork(vm: &VirtualMachine) {
+    fn reinit_other_interpreters_after_fork(vm: &VirtualMachine, stopped: &mut ForkStopTheWorld) {
         use rustpython_common::lock::reinit_mutex_after_fork;
 
         for state in crate::vm::runtime::live_interpreter_states() {
@@ -857,6 +881,7 @@ pub mod module {
             }
 
             state.stop_the_world.reset_after_fork();
+            stopped.retain_thread_slots(&state);
 
             // Every thread registered here belongs to the parent, including any
             // slot the forking thread itself registered before the fork.
@@ -868,9 +893,9 @@ pub mod module {
         crate::vm::thread::purge_other_interpreter_slots_after_fork(vm.state.interpreter_id);
     }
 
-    fn py_os_after_fork_parent(vm: &VirtualMachine) {
+    fn py_os_after_fork_parent(vm: &VirtualMachine, _stopped: &mut ForkStopTheWorld) {
         #[cfg(feature = "threading")]
-        vm.state.stop_the_world.start_the_world(&vm.state);
+        _stopped.restart();
 
         #[cfg(feature = "threading")]
         crate::stdlib::_imp::release_imp_lock_after_fork_parent();
@@ -955,25 +980,25 @@ pub mod module {
 
         vm.audit("os.fork", || ())?;
 
-        py_os_before_fork(vm);
+        let mut stopped = py_os_before_fork(vm);
         let pid = rustpython_host_env::posix::fork();
 
         match pid {
             Ok(0) => {
-                py_os_after_fork_child(vm);
+                py_os_after_fork_child(vm, &mut stopped);
                 Ok(0)
             }
             Ok(pid) => {
                 // Capture this before parent after-fork hooks in case those
                 // hooks start threads.
                 let num_os_threads = get_number_of_os_threads();
-                py_os_after_fork_parent(vm);
+                py_os_after_fork_parent(vm, &mut stopped);
                 // Warn only after parent callback path resumes the world.
                 warn_if_multi_threaded("fork", num_os_threads, vm)?;
                 Ok(pid)
             }
             Err(err) => {
-                py_os_after_fork_parent(vm);
+                py_os_after_fork_parent(vm, &mut stopped);
                 Err(err.into_pyexception(vm))
             }
         }
@@ -1000,22 +1025,22 @@ pub mod module {
 
         vm.audit("os.forkpty", || ())?;
 
-        py_os_before_fork(vm);
+        let mut stopped = py_os_before_fork(vm);
         let result = rustpython_host_env::posix::forkpty();
 
         match result {
             Ok((0, master)) => {
-                py_os_after_fork_child(vm);
+                py_os_after_fork_child(vm, &mut stopped);
                 Ok((0, master))
             }
             Ok((pid, master)) => {
                 let num_os_threads = get_number_of_os_threads();
-                py_os_after_fork_parent(vm);
+                py_os_after_fork_parent(vm, &mut stopped);
                 warn_if_multi_threaded("forkpty", num_os_threads, vm)?;
                 Ok((pid, master))
             }
             Err(err) => {
-                py_os_after_fork_parent(vm);
+                py_os_after_fork_parent(vm, &mut stopped);
                 Err(err.into_pyexception(vm))
             }
         }

diff --git a/crates/vm/src/vm/mod.rs b/crates/vm/src/vm/mod.rs
@@ -18,6 +18,8 @@ mod python_run;
 pub use python_run::InteractiveLine;
 pub mod runtime;
 mod setting;
+#[cfg(feature = "threading")]
+pub(crate) mod stop_the_world;
 pub mod thread;
 mod vm_new;
 mod vm_object;
@@ -644,9 +646,9 @@ impl StopTheWorldState {
         // Only one thread survives fork; any stop-the-world bit inherited
         // from the parent is stale.
         crate::signal::clear_stop_bit();
-        // before_fork owns the sole transaction. Release it exactly once via
-        // that state's ownership flag; resetting other interpreters must not
-        // release a later transaction. No unconditional global reset is needed.
+        // A single-state stop owns the transaction through this flag. A fork
+        // group instead retains its shared transaction through child repair;
+        // its member states do not own exclusion and must not release it here.
         self.release_exclusion();
         stw_trace(format_args!("reset-after-fork"));
     }
-Original file line number
+Diff line change
@@ Expand Up / @@ -83,6 +83,8 @@ trai @@
     ulonglong
     unic
     unistd
+    unparked
+    unparking
     unraw
     unsync
     wasip1
@@ Expand Down @@