From c422c9a72ffc981afb9147c5bf287a345b0e6df9 Mon Sep 17 00:00:00 2001 From: Jeong YunWon Date: Wed, 19 Aug 2026 10:22:19 +0900 Subject: [PATCH 01/15] lock: add an opt-in rwlock that detaches before it blocks A thread blocked acquiring a lock reaches no safepoint, so stop-the-world cannot stop it, and the lock it waits for is routinely one a thread the requester already suspended is holding. `RawDetachingRwLock` wraps the raw rwlock and hands the wait for a contended acquire to a hook that leaves the interpreter first; an acquire that takes the lock on its first try does not reach the hook. The vm installs the hook during interpreter init and implements it with `allow_threads`. The wait ends with the lock acquired while detached, so re-attaching can park the thread holding it. That is only safe where nothing reachable from a stop-the-world section takes the same lock, so it is opt-in per lock: `PyDetachingRwLock` is a separate type from `PyRwLock`, and `Traverse` is not implemented for it, so a payload holding one cannot derive `Traverse`. `PyByteArray::inner` takes it. `BorrowedValue`/`BorrowedValueMut` gain the matching mapped-guard variants. `a_thread_blocked_on_a_lock_does_not_stall_stop_the_world` blocks an interpreter thread on a `PyDetachingRwLock` and asserts stop-the-world still completes, running the stop on its own thread with a timeout so a stop that never completes fails rather than hangs. Without the hook installed it fails on the 10 s timeout; with it, it passes in 0.07 s. Assisted-by: Claude --- crates/common/src/borrow.rs | 16 +- crates/common/src/lock.rs | 16 ++ crates/common/src/lock/detaching.rs | 258 ++++++++++++++++++++++++++++ crates/vm/src/builtins/bytearray.rs | 29 ++-- crates/vm/src/vm/interpreter.rs | 72 ++++++++ crates/vm/src/vm/thread.rs | 34 ++++ 6 files changed, 410 insertions(+), 15 deletions(-) create mode 100644 crates/common/src/lock/detaching.rs diff --git a/crates/common/src/borrow.rs b/crates/common/src/borrow.rs index 70d755ff155..ebf69fde71d 100644 --- a/crates/common/src/borrow.rs +++ b/crates/common/src/borrow.rs @@ -1,5 +1,6 @@ use crate::lock::{ - MapImmutable, PyImmutableMappedMutexGuard, PyMappedMutexGuard, PyMappedRwLockReadGuard, + MapImmutable, PyImmutableMappedMutexGuard, PyMappedDetachingRwLockReadGuard, + PyMappedDetachingRwLockWriteGuard, PyMappedMutexGuard, PyMappedRwLockReadGuard, PyMappedRwLockWriteGuard, PyMutexGuard, PyRwLockReadGuard, PyRwLockWriteGuard, }; use alloc::fmt; @@ -24,6 +25,7 @@ pub enum BorrowedValue<'a, T: ?Sized> { MappedMuLock(PyImmutableMappedMutexGuard<'a, T>), ReadLock(PyRwLockReadGuard<'a, T>), MappedReadLock(PyMappedRwLockReadGuard<'a, T>), + MappedDetachingReadLock(PyMappedDetachingRwLockReadGuard<'a, T>), } impl_from!('a, T, BorrowedValue<'a, T>, Ref(&'a T), @@ -31,6 +33,7 @@ impl_from!('a, T, BorrowedValue<'a, T>, MappedMuLock(PyImmutableMappedMutexGuard<'a, T>), ReadLock(PyRwLockReadGuard<'a, T>), MappedReadLock(PyMappedRwLockReadGuard<'a, T>), + MappedDetachingReadLock(PyMappedDetachingRwLockReadGuard<'a, T>), ); impl<'a, T: ?Sized> BorrowedValue<'a, T> { @@ -59,6 +62,9 @@ impl<'a, T: ?Sized> BorrowedValue<'a, T> { Self::MappedReadLock(m) => { BorrowedValue::MappedReadLock(PyMappedRwLockReadGuard::map(m, f)) } + Self::MappedDetachingReadLock(m) => { + BorrowedValue::MappedDetachingReadLock(PyMappedDetachingRwLockReadGuard::map(m, f)) + } } } } @@ -73,6 +79,7 @@ impl Deref for BorrowedValue<'_, T> { Self::MappedMuLock(m) => m, Self::ReadLock(r) => r, Self::MappedReadLock(m) => m, + Self::MappedDetachingReadLock(m) => m, } } } @@ -90,6 +97,7 @@ pub enum BorrowedValueMut<'a, T: ?Sized> { MappedMuLock(PyMappedMutexGuard<'a, T>), WriteLock(PyRwLockWriteGuard<'a, T>), MappedWriteLock(PyMappedRwLockWriteGuard<'a, T>), + MappedDetachingWriteLock(PyMappedDetachingRwLockWriteGuard<'a, T>), } impl_from!('a, T, BorrowedValueMut<'a, T>, @@ -98,6 +106,7 @@ impl_from!('a, T, BorrowedValueMut<'a, T>, MappedMuLock(PyMappedMutexGuard<'a, T>), WriteLock(PyRwLockWriteGuard<'a, T>), MappedWriteLock(PyMappedRwLockWriteGuard<'a, T>), + MappedDetachingWriteLock(PyMappedDetachingRwLockWriteGuard<'a, T>), ); impl<'a, T: ?Sized> BorrowedValueMut<'a, T> { @@ -113,6 +122,9 @@ impl<'a, T: ?Sized> BorrowedValueMut<'a, T> { Self::MappedWriteLock(m) => { BorrowedValueMut::MappedWriteLock(PyMappedRwLockWriteGuard::map(m, f)) } + Self::MappedDetachingWriteLock(m) => BorrowedValueMut::MappedDetachingWriteLock( + PyMappedDetachingRwLockWriteGuard::map(m, f), + ), } } } @@ -127,6 +139,7 @@ impl Deref for BorrowedValueMut<'_, T> { Self::MappedMuLock(m) => m, Self::WriteLock(w) => w, Self::MappedWriteLock(w) => w, + Self::MappedDetachingWriteLock(w) => w, } } } @@ -139,6 +152,7 @@ impl DerefMut for BorrowedValueMut<'_, T> { Self::MappedMuLock(m) => &mut *m, Self::WriteLock(w) => &mut *w, Self::MappedWriteLock(w) => &mut *w, + Self::MappedDetachingWriteLock(w) => &mut *w, } } } diff --git a/crates/common/src/lock.rs b/crates/common/src/lock.rs index 08fbc316599..c8bfc4cef8c 100644 --- a/crates/common/src/lock.rs +++ b/crates/common/src/lock.rs @@ -8,6 +8,7 @@ use lock_api::{ cfg_select! { feature = "threading" => { + pub use detaching::{BlockingWaitHook, set_blocking_wait_hook}; pub use parking_lot::{RawMutex, RawRwLock, RawThreadId}; pub use std::sync::OnceLock as OnceCell; pub use core::cell::LazyCell; @@ -47,6 +48,8 @@ cfg_select! { } } +mod detaching; +pub use detaching::RawDetachingRwLock; mod immutable_mutex; pub use immutable_mutex::*; mod thread_mutex; @@ -60,6 +63,19 @@ pub type PyThreadMutex = ThreadMutex; pub type PyThreadMutexGuard<'a, T> = ThreadMutexGuard<'a, RawMutex, RawThreadId, T>; pub type PyMappedThreadMutexGuard<'a, T> = MappedThreadMutexGuard<'a, RawMutex, RawThreadId, T>; +/// A `PyRwLock` for data a thread may hold locked across a blocking call. +/// +/// Waiting for one of these leaves the interpreter first, so a thread blocked +/// on it is a thread stop-the-world can park. That is only safe where a +/// collection never takes the same lock — see [`RawDetachingRwLock`] — so this +/// is opt-in per lock rather than what every `PyRwLock` does. +pub type PyDetachingRwLock = RwLock; +pub type PyDetachingRwLockReadGuard<'a, T> = RwLockReadGuard<'a, RawDetachingRwLock, T>; +pub type PyDetachingRwLockWriteGuard<'a, T> = RwLockWriteGuard<'a, RawDetachingRwLock, T>; +pub type PyMappedDetachingRwLockReadGuard<'a, T> = MappedRwLockReadGuard<'a, RawDetachingRwLock, T>; +pub type PyMappedDetachingRwLockWriteGuard<'a, T> = + MappedRwLockWriteGuard<'a, RawDetachingRwLock, T>; + pub type PyRwLock = RwLock; pub type PyRwLockUpgradableReadGuard<'a, T> = RwLockUpgradableReadGuard<'a, RawRwLock, T>; pub type PyRwLockReadGuard<'a, T> = RwLockReadGuard<'a, RawRwLock, T>; diff --git a/crates/common/src/lock/detaching.rs b/crates/common/src/lock/detaching.rs new file mode 100644 index 00000000000..5952de27a23 --- /dev/null +++ b/crates/common/src/lock/detaching.rs @@ -0,0 +1,258 @@ +//! A reader-writer lock that lets a thread leave its interpreter before it +//! blocks. +//! +//! Stopping the world means waiting for every running thread to reach a +//! safepoint. A thread blocked on a lock reaches none, so if the thread holding +//! that lock has already been stopped, the two wait on each other forever. The +//! holder is not the one who can avoid this — a lock is held across a blocking +//! call precisely because that is what the call needs — so the waiter gives up +//! its interpreter for the duration of the wait instead, which is what a +//! blocking call does anyway. +//! +//! Doing so is safe only for locks nothing reachable from a stop-the-world +//! section takes, so it is opt-in per lock — see [`RawDetachingRwLock`] for the +//! rule and why it is needed. +//! +//! Only the contended path pays for any of this: an acquire that takes the lock +//! on the first try is the same atomic exchange it was, and never reaches the +//! hook. The hook is installed by whoever knows how to detach a thread +//! ([`set_blocking_wait_hook`]); until then, and on any thread that is not +//! running an interpreter, a blocked acquire just blocks. + +use super::RawRwLock; +#[cfg(feature = "threading")] +use core::cell::Cell; +use lock_api::{ + RawRwLock as RawRwLockTrait, RawRwLockDowngrade, RawRwLockRecursive as RawRwLockRecursiveTrait, + RawRwLockUpgrade as RawRwLockUpgradeTrait, RawRwLockUpgradeDowngrade, +}; +#[cfg(feature = "threading")] +use std::sync::OnceLock; + +/// Runs `wait` with the calling thread detached from its interpreter. +#[cfg(feature = "threading")] +pub type BlockingWaitHook = fn(wait: &dyn Fn()); + +#[cfg(feature = "threading")] +static BLOCKING_WAIT: OnceLock = OnceLock::new(); + +/// Install the hook that detaches a thread around a blocked lock acquire. +/// +/// Later calls are ignored, so every interpreter in a process can call this +/// during its own initialization. +#[cfg(feature = "threading")] +pub fn set_blocking_wait_hook(hook: BlockingWaitHook) { + let _ = BLOCKING_WAIT.set(hook); +} + +#[cfg(feature = "threading")] +std::thread_local! { + /// Set while this thread is inside the hook, so that a lock taken by the + /// hook itself — or by anything detaching and re-attaching runs — waits + /// plainly instead of recursing back into it. + static IN_HOOK: Cell = const { Cell::new(false) }; +} + +/// Clears [`IN_HOOK`] even if the hook unwinds. +#[cfg(feature = "threading")] +struct HookGuard; + +#[cfg(feature = "threading")] +impl Drop for HookGuard { + fn drop(&mut self) { + let _ = IN_HOOK.try_with(|in_hook| in_hook.set(false)); + } +} + +/// Block on `wait`, detached from this thread's interpreter if there is one. +/// +/// Nothing spins on the way here. The lock underneath already spins before it +/// parks, and skips that spin once a waiter has parked — the same condition +/// `_PyMutex_LockTimed` spins under. A spin layered on top cannot read that +/// condition, and would go on retrying a `try_lock` that reports failure for as +/// long as a writer holds the writer bit, which it takes before it waits for +/// readers to drain: a yield per retry for the whole of exactly the wait this +/// exists to survive. +#[cfg(feature = "threading")] +#[cold] +#[inline(never)] +fn wait_detached(wait: impl Fn()) { + let Some(hook) = BLOCKING_WAIT.get() else { + wait(); + return; + }; + // `try_with` fails once the thread's locals are being destroyed, which is + // also a point at which there is no interpreter left to detach from. + let entered = IN_HOOK + .try_with(|in_hook| !in_hook.replace(true)) + .unwrap_or(false); + if !entered { + wait(); + return; + } + let _guard = HookGuard; + hook(&wait); +} + +/// Without threads there is no interpreter to leave and nothing to stop. +#[cfg(not(feature = "threading"))] +#[inline] +fn wait_detached(wait: impl Fn()) { + wait(); +} + +/// A reader-writer lock whose blocking acquires detach first, and which is the +/// raw lock it wraps in every other respect. +/// +/// Use through [`PyDetachingRwLock`](super::PyDetachingRwLock). +/// +/// # Only for locks a collection never takes +/// +/// The wait acquires the lock while detached, so the thread comes back holding +/// it, and re-attaching is a point at which a stop-the-world in flight will +/// park the thread. It is therefore parked *holding the lock*. Everything that +/// stops the world must be able to finish without that lock: if a collection +/// were to take it, the collection would block on a thread only the collection +/// can release, and neither would move again. +/// +/// So this is opt-in per lock, and the rule for opting in is that nothing +/// reachable from a stop-the-world section takes the same lock. An object whose +/// payload holds no references — nothing for the collector to traverse into — +/// satisfies that; most do not. +/// +/// Not implementing the vm's `Traverse` for this lock is what keeps that from +/// being only a convention: a payload holding one cannot derive `Traverse`, so +/// it cannot become something a collection walks into. +#[repr(transparent)] +pub struct RawDetachingRwLock(RawRwLock); + +// SAFETY: every method forwards to the wrapped raw lock, which upholds the +// contract; the blocking acquires only add a wait that ends with the same lock +// acquired. +unsafe impl RawRwLockTrait for RawDetachingRwLock { + #[allow( + clippy::declare_interior_mutable_const, + reason = "raw lock initializer, as in the type it wraps" + )] + const INIT: Self = Self(::INIT); + + type GuardMarker = ::GuardMarker; + + #[inline] + fn lock_shared(&self) { + if !self.0.try_lock_shared() { + wait_detached(|| self.0.lock_shared()); + } + } + + #[inline] + fn try_lock_shared(&self) -> bool { + self.0.try_lock_shared() + } + + #[inline] + unsafe fn unlock_shared(&self) { + unsafe { self.0.unlock_shared() } + } + + #[inline] + fn lock_exclusive(&self) { + if !self.0.try_lock_exclusive() { + wait_detached(|| self.0.lock_exclusive()); + } + } + + #[inline] + fn try_lock_exclusive(&self) -> bool { + self.0.try_lock_exclusive() + } + + #[inline] + unsafe fn unlock_exclusive(&self) { + unsafe { self.0.unlock_exclusive() } + } + + #[inline] + fn is_locked(&self) -> bool { + self.0.is_locked() + } + + #[inline] + fn is_locked_exclusive(&self) -> bool { + self.0.is_locked_exclusive() + } +} + +// SAFETY: forwards to the wrapped raw lock. +unsafe impl RawRwLockDowngrade for RawDetachingRwLock { + #[inline] + unsafe fn downgrade(&self) { + unsafe { self.0.downgrade() } + } +} + +// SAFETY: forwards to the wrapped raw lock; the blocking acquires only add +// a wait that ends with the same lock acquired. +unsafe impl RawRwLockUpgradeTrait for RawDetachingRwLock { + #[inline] + fn lock_upgradable(&self) { + if !self.0.try_lock_upgradable() { + wait_detached(|| self.0.lock_upgradable()); + } + } + + #[inline] + fn try_lock_upgradable(&self) -> bool { + self.0.try_lock_upgradable() + } + + #[inline] + unsafe fn unlock_upgradable(&self) { + unsafe { self.0.unlock_upgradable() } + } + + #[inline] + unsafe fn upgrade(&self) { + // SAFETY: the caller holds the upgradable lock, as `upgrade` requires, + // and it stays held for both the failed attempt and the wait. + unsafe { + if !self.0.try_upgrade() { + wait_detached(|| self.0.upgrade()); + } + } + } + + #[inline] + unsafe fn try_upgrade(&self) -> bool { + unsafe { self.0.try_upgrade() } + } +} + +// SAFETY: forwards to the wrapped raw lock. +unsafe impl RawRwLockUpgradeDowngrade for RawDetachingRwLock { + #[inline] + unsafe fn downgrade_upgradable(&self) { + unsafe { self.0.downgrade_upgradable() } + } + + #[inline] + unsafe fn downgrade_to_upgradable(&self) { + unsafe { self.0.downgrade_to_upgradable() } + } +} + +// SAFETY: forwards to the wrapped raw lock; the blocking acquire only adds +// a wait that ends with the same lock acquired. +unsafe impl RawRwLockRecursiveTrait for RawDetachingRwLock { + #[inline] + fn lock_shared_recursive(&self) { + if !self.0.try_lock_shared_recursive() { + wait_detached(|| self.0.lock_shared_recursive()); + } + } + + #[inline] + fn try_lock_shared_recursive(&self) -> bool { + self.0.try_lock_shared_recursive() + } +} diff --git a/crates/vm/src/builtins/bytearray.rs b/crates/vm/src/builtins/bytearray.rs index 594ecc569d8..f063a1d08c9 100644 --- a/crates/vm/src/builtins/bytearray.rs +++ b/crates/vm/src/builtins/bytearray.rs @@ -18,8 +18,8 @@ use crate::{ common::{ atomic::{AtomicUsize, Ordering}, lock::{ - PyMappedRwLockReadGuard, PyMappedRwLockWriteGuard, PyMutex, PyRwLock, - PyRwLockReadGuard, PyRwLockWriteGuard, + PyDetachingRwLock, PyDetachingRwLockReadGuard, PyDetachingRwLockWriteGuard, + PyMappedDetachingRwLockReadGuard, PyMappedDetachingRwLockWriteGuard, PyMutex, }, }, convert::{ToPyObject, ToPyResult}, @@ -43,7 +43,7 @@ use core::mem::size_of; #[pyclass(module = false, name = "bytearray", unhashable = true)] #[derive(Debug, Default)] pub struct PyByteArray { - inner: PyRwLock, + inner: PyDetachingRwLock, exports: AtomicUsize, } @@ -81,17 +81,17 @@ impl PyByteArray { const fn from_inner(inner: PyBytesInner) -> Self { Self { - inner: PyRwLock::new(inner), + inner: PyDetachingRwLock::new(inner), exports: AtomicUsize::new(0), } } - pub fn borrow_buf(&self) -> PyMappedRwLockReadGuard<'_, [u8]> { - PyRwLockReadGuard::map(self.inner.read(), |inner| &*inner.elements) + pub fn borrow_buf(&self) -> PyMappedDetachingRwLockReadGuard<'_, [u8]> { + PyDetachingRwLockReadGuard::map(self.inner.read(), |inner| &*inner.elements) } - pub fn borrow_buf_mut(&self) -> PyMappedRwLockWriteGuard<'_, Vec> { - PyRwLockWriteGuard::map(self.inner.write(), |inner| &mut inner.elements) + pub fn borrow_buf_mut(&self) -> PyMappedDetachingRwLockWriteGuard<'_, Vec> { + PyDetachingRwLockWriteGuard::map(self.inner.write(), |inner| &mut inner.elements) } fn repeat(&self, value: isize, vm: &VirtualMachine) -> PyResult { @@ -194,11 +194,11 @@ impl PyByteArray { } #[inline] - fn inner(&self) -> PyRwLockReadGuard<'_, PyBytesInner> { + fn inner(&self) -> PyDetachingRwLockReadGuard<'_, PyBytesInner> { self.inner.read() } #[inline] - fn inner_mut(&self) -> PyRwLockWriteGuard<'_, PyBytesInner> { + fn inner_mut(&self) -> PyDetachingRwLockWriteGuard<'_, PyBytesInner> { self.inner.write() } @@ -739,9 +739,10 @@ impl Comparable for PyByteArray { static BUFFER_METHODS: BufferMethods = BufferMethods { obj_bytes: |buffer| buffer.obj_as::().borrow_buf().into(), obj_bytes_mut: |buffer| { - PyMappedRwLockWriteGuard::map(buffer.obj_as::().borrow_buf_mut(), |x| { - x.as_mut_slice() - }) + PyMappedDetachingRwLockWriteGuard::map( + buffer.obj_as::().borrow_buf_mut(), + |x| x.as_mut_slice(), + ) .into() }, release: |buffer| { @@ -783,7 +784,7 @@ impl AsBuffer for PyByteArray { } impl BufferResizeGuard for PyByteArray { - type Resizable<'a> = PyRwLockWriteGuard<'a, PyBytesInner>; + type Resizable<'a> = PyDetachingRwLockWriteGuard<'a, PyBytesInner>; fn try_resizable_opt(&self) -> Option> { // An export is a borrow someone else still holds, so it is answered diff --git a/crates/vm/src/vm/interpreter.rs b/crates/vm/src/vm/interpreter.rs index 6d4e1f75a22..7f3dd8ff814 100644 --- a/crates/vm/src/vm/interpreter.rs +++ b/crates/vm/src/vm/interpreter.rs @@ -76,6 +76,10 @@ where use core::sync::atomic::{AtomicBool, AtomicU64}; use crossbeam_utils::atomic::AtomicCell; + // Before any lock this interpreter's threads can contend on exists. + #[cfg(feature = "threading")] + thread::install_blocking_wait_hook(); + let (config, all_module_defs, frozen, hash_secret, int_max_str_digits) = if let Some(parent) = parent_state { // Subinterpreter: clone config and module tables from parent, fresh runtime state. @@ -1656,6 +1660,74 @@ for _ in range(40): worker.join().expect("nested worker panicked"); } + /// A thread blocked on a detaching lock must not stall stop-the-world. + /// + /// Blocking on a lock reaches no safepoint, so an interpreter thread that + /// waits while attached is a thread the world can never stop — and the + /// lock it waits for is routinely one a stopped thread holds, which is the + /// deadlock. The waiter therefore leaves its interpreter for the wait. + #[cfg(feature = "threading")] + #[test] + fn a_thread_blocked_on_a_lock_does_not_stall_stop_the_world() { + use crate::common::lock::PyDetachingRwLock; + use alloc::sync::Arc; + use core::{ + sync::atomic::{AtomicBool, Ordering}, + time::Duration, + }; + + let interp = Interpreter::without_stdlib(Default::default()); + let state = interp.enter(|vm| vm.state.clone()); + + let lock: Arc> = Arc::new(PyDetachingRwLock::new(())); + let at_lock = Arc::new(AtomicBool::new(false)); + + // Held for the whole test, so the worker below blocks and stays blocked. + let held = lock.write(); + + let worker_lock = Arc::clone(&lock); + let worker_at_lock = Arc::clone(&at_lock); + let worker = interp.enter(|vm| { + let thread_vm = vm.new_thread(); + std::thread::spawn(move || { + thread_vm.run(|_vm| { + worker_at_lock.store(true, Ordering::Release); + let _read = worker_lock.read(); + }); + }) + }); + + while !at_lock.load(Ordering::Acquire) { + std::thread::yield_now(); + } + // The store above only says the worker is about to block, not that it + // has; give it the moment it needs to get there. + std::thread::sleep(Duration::from_millis(50)); + + // Stop from a thread of its own so that a stop that never completes + // fails the test instead of hanging it. + let (tx, rx) = std::sync::mpsc::channel(); + let stop_state = state; + let stopper = std::thread::spawn(move || { + stop_state.stop_the_world.stop_the_world(&stop_state); + let stopped = tx.send(()); + stop_state.stop_the_world.start_the_world(&stop_state); + stopped + }); + + let stopped = rx.recv_timeout(Duration::from_secs(10)); + + // Release before any assertion: the worker has to finish for the + // stopper to be joinable, and for the test to end at all. + drop(held); + assert!( + stopped.is_ok(), + "stop-the-world did not complete while a thread was blocked on a lock" + ); + stopper.join().expect("stopper panicked").expect("send"); + worker.join().expect("worker panicked"); + } + /// The process main id is recorded once and is stable across later creates. #[test] fn process_main_id_recorded_and_stable() { diff --git a/crates/vm/src/vm/thread.rs b/crates/vm/src/vm/thread.rs index 4ba0d7ffada..ea0a79a8b1a 100644 --- a/crates/vm/src/vm/thread.rs +++ b/crates/vm/src/vm/thread.rs @@ -605,6 +605,40 @@ pub fn allow_threads(_vm: &VirtualMachine, f: impl FnOnce() -> R) -> R { f() } +/// Wait for a lock the way a blocking call waits: detached, so a +/// stop-the-world requester never has to wait for this thread to reach a +/// safepoint it cannot reach while blocked. +/// +/// Threads with no interpreter to leave — a native thread, or one whose +/// locals are already being destroyed — simply block. +/// +/// The requester of a stop is exempt from being parked by it +/// ([`StopTheWorldState::park_detached_threads`](super::StopTheWorldState) and +/// [`suspend_if_needed`] both skip it), so detaching here does not risk parking +/// the one thread that can start the world again. +#[cfg(feature = "threading")] +fn wait_detached_from_interpreter(wait: &dyn Fn()) { + // Read the VM out before waiting: attaching afterwards reaches for the + // same thread locals, which must not still be borrowed here. + let current = VM_STACK + .try_with(|vms| vms.try_borrow().ok()?.last().copied()) + .ok() + .flatten(); + match current { + // SAFETY: entries in VM_STACK either borrow a VM for the dynamic + // scope of a set_current_vm()/enter_vm() call or point at GILSTATE_VM. + Some(vm) => allow_threads(unsafe { vm.as_ref() }, wait), + None => wait(), + } +} + +/// Teach the lock types how to detach this thread. Idempotent, so every +/// interpreter can call it while initializing. +#[cfg(feature = "threading")] +pub(crate) fn install_blocking_wait_hook() { + rustpython_common::lock::set_blocking_wait_hook(wait_detached_from_interpreter); +} + /// Called from check_signals when stop-the-world is requested. /// Transitions ATTACHED → SUSPENDED and waits until released /// (like `_PyThreadState_Suspend` + `_PyThreadState_Attach`). From 25639de8582646d80a904a9206488dd939668a61 Mon Sep 17 00:00:00 2001 From: Jeong YunWon Date: Thu, 20 Aug 2026 00:12:31 +0900 Subject: [PATCH 02/15] lock: stop detaching where the lock may already be held `upgrade` runs with the upgradable lock held, and `lock_shared_recursive` may be the re-entrant take of a lock the calling thread holds; detaching there parks a thread holding the lock, which is what this type documents it must not do. They forward to the wrapped lock instead. `lock_upgradable` starts from holding nothing, but nothing takes an upgradable read of one of these, so it forwards too. `lock_shared` and `lock_exclusive` still detach. Also narrow two claims the comments overstated. Not implementing `Traverse` enforces the opt-in rule only against collections, not against the other things that stop the world. And the requester exemption the hook relies on is wider than `_PyEval_StopTheWorld` gives, so it is a local invariant. Assisted-by: Claude --- crates/common/src/lock/detaching.rs | 40 ++++++++++++++--------------- crates/vm/src/vm/thread.rs | 11 +++++--- 2 files changed, 27 insertions(+), 24 deletions(-) diff --git a/crates/common/src/lock/detaching.rs b/crates/common/src/lock/detaching.rs index 5952de27a23..e5700ce8897 100644 --- a/crates/common/src/lock/detaching.rs +++ b/crates/common/src/lock/detaching.rs @@ -120,9 +120,12 @@ fn wait_detached(wait: impl Fn()) { /// payload holds no references — nothing for the collector to traverse into — /// satisfies that; most do not. /// -/// Not implementing the vm's `Traverse` for this lock is what keeps that from -/// being only a convention: a payload holding one cannot derive `Traverse`, so -/// it cannot become something a collection walks into. +/// Not implementing the vm's `Traverse` for this lock enforces part of that: a +/// payload holding one cannot derive `Traverse`, so it cannot become something +/// a collection walks into. Only that part. A collection is not the only thing +/// that stops the world — dumping tracebacks, enumerating thread frames and +/// forking all do — and nothing checks what those reach. For them the rule is +/// still a convention. #[repr(transparent)] pub struct RawDetachingRwLock(RawRwLock); @@ -191,14 +194,18 @@ unsafe impl RawRwLockDowngrade for RawDetachingRwLock { } } -// SAFETY: forwards to the wrapped raw lock; the blocking acquires only add -// a wait that ends with the same lock acquired. +// SAFETY: forwards to the wrapped raw lock. +// +// None of these detach. `upgrade` runs with the upgradable lock already held, +// and `lock_shared_recursive` may be the re-entrant take of a lock this thread +// holds; detaching there would park a thread *holding* the lock, the one thing +// this type must not do. `lock_upgradable` starts from holding nothing and +// could detach as safely as `lock_shared` does, but nothing takes an upgradable +// read of one of these, so it does not. unsafe impl RawRwLockUpgradeTrait for RawDetachingRwLock { #[inline] fn lock_upgradable(&self) { - if !self.0.try_lock_upgradable() { - wait_detached(|| self.0.lock_upgradable()); - } + self.0.lock_upgradable() } #[inline] @@ -213,13 +220,8 @@ unsafe impl RawRwLockUpgradeTrait for RawDetachingRwLock { #[inline] unsafe fn upgrade(&self) { - // SAFETY: the caller holds the upgradable lock, as `upgrade` requires, - // and it stays held for both the failed attempt and the wait. - unsafe { - if !self.0.try_upgrade() { - wait_detached(|| self.0.upgrade()); - } - } + // SAFETY: the caller holds the upgradable lock, as `upgrade` requires. + unsafe { self.0.upgrade() } } #[inline] @@ -241,14 +243,12 @@ unsafe impl RawRwLockUpgradeDowngrade for RawDetachingRwLock { } } -// SAFETY: forwards to the wrapped raw lock; the blocking acquire only adds -// a wait that ends with the same lock acquired. +// SAFETY: forwards to the wrapped raw lock. Does not detach; see the upgrade +// impl above. unsafe impl RawRwLockRecursiveTrait for RawDetachingRwLock { #[inline] fn lock_shared_recursive(&self) { - if !self.0.try_lock_shared_recursive() { - wait_detached(|| self.0.lock_shared_recursive()); - } + self.0.lock_shared_recursive() } #[inline] diff --git a/crates/vm/src/vm/thread.rs b/crates/vm/src/vm/thread.rs index ea0a79a8b1a..5e0aa076865 100644 --- a/crates/vm/src/vm/thread.rs +++ b/crates/vm/src/vm/thread.rs @@ -612,10 +612,13 @@ pub fn allow_threads(_vm: &VirtualMachine, f: impl FnOnce() -> R) -> R { /// Threads with no interpreter to leave — a native thread, or one whose /// locals are already being destroyed — simply block. /// -/// The requester of a stop is exempt from being parked by it -/// ([`StopTheWorldState::park_detached_threads`](super::StopTheWorldState) and -/// [`suspend_if_needed`] both skip it), so detaching here does not risk parking -/// the one thread that can start the world again. +/// Detaching cannot park the one thread that can start the world again: +/// [`park_detached_threads`](super::StopTheWorldState) skips the requester's +/// slot outright, by thread id, and [`suspend_if_needed`] keys off a stop bit +/// never set for it. That exemption is wider than the one `_PyEval_StopTheWorld` +/// gives, where only an ATTACHED requester is skipped and a DETACHED one is +/// suspended like any other thread — so this rests on a local invariant rather +/// than on the reference behavior. #[cfg(feature = "threading")] fn wait_detached_from_interpreter(wait: &dyn Fn()) { // Read the VM out before waiting: attaching afterwards reaches for the From 72289fbd29241bcde7ce0f6306cf5d5623462945 Mon Sep 17 00:00:00 2001 From: Jeong YunWon Date: Thu, 20 Aug 2026 01:03:23 +0900 Subject: [PATCH 03/15] os.readinto: read aside when the fd can wait `readinto` held the destination's write lock for the whole call, including the `read(2)` inside `allow_threads`. On a pipe, socket or terminal that read returns only when the other end writes, so a thread reaching the same object waited on a lock for an unbounded time, reaching no safepoint while it did. Take the fd that answers without waiting directly, as before, and otherwise read into scratch and take the lock only for the copy. This is what `FileIO.readinto`, `socket.recv_into` and `socket.recvfrom_into` already do; `os.readinto` was the site left over. The EINTR retry moves to `read_into_slice`, unchanged. Assisted-by: Claude --- crates/vm/src/stdlib/os.rs | 47 ++++++++++++++++++++++++++++---------- 1 file changed, 35 insertions(+), 12 deletions(-) diff --git a/crates/vm/src/stdlib/os.rs b/crates/vm/src/stdlib/os.rs index 70afdf827b8..523097ad971 100644 --- a/crates/vm/src/stdlib/os.rs +++ b/crates/vm/src/stdlib/os.rs @@ -344,24 +344,47 @@ pub(super) mod _os { } } + /// `read(2)` into `buf`, retrying on EINTR (PEP 475). + fn read_into_slice( + fd: crt_fd::Borrowed<'_>, + buf: &mut [u8], + vm: &VirtualMachine, + ) -> PyResult { + loop { + match vm.allow_threads(|| crt_fd::read(fd, buf)) { + Ok(n) => return Ok(n), + Err(e) if e.raw_os_error() == Some(libc::EINTR) => { + vm.check_signals()?; + continue; + } + Err(e) => return Err(e.into_pyexception(vm)), + } + } + } + #[pyfunction] fn readinto( fd: crt_fd::Borrowed<'_>, buffer: ArgMemoryBuffer, vm: &VirtualMachine, ) -> PyResult { - buffer.with_ref(|buf| { - loop { - match vm.allow_threads(|| crt_fd::read(fd, buf)) { - Ok(n) => return Ok(n), - Err(e) if e.raw_os_error() == Some(libc::EINTR) => { - vm.check_signals()?; - continue; - } - Err(e) => return Err(e.into_pyexception(vm)), - } - } - }) + if rustpython_host_env::io::reads_without_waiting(fd) { + // The read answers from the file itself, so it returns without + // waiting on anyone; write where the caller asked directly. + return buffer.with_ref(|buf| read_into_slice(fd, buf, vm)); + } + + // A pipe, socket or terminal answers only when the other end writes, + // which may be never. Holding the export for the whole call is what + // keeps the target from being resized meanwhile; but reaching its + // bytes takes a lock that every other thread touching the same object + // waits on, and a thread waiting on a lock never reaches a safepoint, + // so holding that one across the wait stops the world from being + // stopped at all. Read aside and take the lock for the copy. + let mut scratch = vm.new_zeroed_bytes(buffer.len())?; + let n = read_into_slice(fd, &mut scratch, vm)?; + buffer.borrow_buf_mut()[..n].copy_from_slice(&scratch[..n]); + Ok(n) } #[pyfunction] From 14148310a89ab3bdfab4ca7c25d08a9791c078db Mon Sep 17 00:00:00 2001 From: Jeong YunWon Date: Thu, 20 Aug 2026 07:39:03 +0900 Subject: [PATCH 04/15] fcntl: detach around the calls that wait Every call in this module ran with the thread attached, so a thread inside one reached no safepoint until it returned. `flock(LOCK_EX)` and `lockf(F_LOCK)` return when whoever holds the lock gives it up, which may be never, and an ioctl on a terminal or socket answers when the device is ready to; the world could not be stopped for that long. `fcntl_fcntl_impl`, `fcntl_ioctl_impl`, `fcntl_flock_impl` and `fcntl_lockf_impl` all release around the call. `ioctl` with `mutate_flag` additionally held the target's write lock for the whole call, so a thread reaching the same object waited on a lock for as long as the device took. Its bytes now go in and come back through a buffer of our own, as `fcntl_ioctl_impl` copies through one of its own for anything up to IOCTL_BUFSZ. The export the argument holds is what keeps the length from changing in between. test_fcntl and test_ioctl pass. Assisted-by: Claude --- crates/stdlib/src/fcntl.rs | 50 ++++++++++++++++++++++++++------------ 1 file changed, 34 insertions(+), 16 deletions(-) diff --git a/crates/stdlib/src/fcntl.rs b/crates/stdlib/src/fcntl.rs index 8e24f2b6e4a..d482bf256b6 100644 --- a/crates/stdlib/src/fcntl.rs +++ b/crates/stdlib/src/fcntl.rs @@ -78,15 +78,16 @@ mod fcntl { .ok_or_else(|| vm.new_value_error("fcntl string arg too long"))? .copy_from_slice(&s) } - host_fcntl::fcntl_with_bytes(fd, cmd, &mut buf[..arg_len]) + vm.allow_threads(|| host_fcntl::fcntl_with_bytes(fd, cmd, &mut buf[..arg_len])) .map_err(|_| vm.new_last_errno_error())?; return Ok(vm.ctx.new_bytes(buf[..arg_len].to_vec()).into()); } OptionalArg::Present(Either::B(i)) => i.as_u32_mask(), OptionalArg::Missing => 0, }; - let ret = - host_fcntl::fcntl_int(fd, cmd, int as i32).map_err(|_| vm.new_last_errno_error())?; + let ret = vm + .allow_threads(|| host_fcntl::fcntl_int(fd, cmd, int as i32)) + .map_err(|_| vm.new_last_errno_error())?; Ok(vm.new_pyobj(ret)) } @@ -114,26 +115,37 @@ mod fcntl { let buf_len = match buf_kind { Either::A(rw_arg) => { let mutate_flag = mutate_flag.unwrap_or(true); - let mut arg_buf = rw_arg.borrow_buf_mut(); if mutate_flag { - let ret = unsafe { - host_fcntl::ioctl_ptr(fd, request, arg_buf.as_mut_ptr().cast()) - } - .map_err(|_| vm.new_last_errno_error())?; + // A terminal or a socket answers an ioctl when it is + // ready to, so the call runs detached, and the target's + // bytes go in and come back through a buffer of our own + // rather than stay locked meanwhile -- `fcntl_ioctl_impl` + // copies through one the same way. + let mut scratch = vm.new_zeroed_bytes(rw_arg.len())?; + scratch.copy_from_slice(&rw_arg.borrow_buf_mut()); + let ret = vm + .allow_threads(|| unsafe { + host_fcntl::ioctl_ptr(fd, request, scratch.as_mut_ptr().cast()) + }) + .map_err(|_| vm.new_last_errno_error())?; + rw_arg.borrow_buf_mut().copy_from_slice(&scratch); return Ok(vm.ctx.new_int(ret).into()); } // treat like an immutable buffer - fill_buf(&arg_buf)? + fill_buf(&rw_arg.borrow_buf_mut())? } Either::B(ro_buf) => fill_buf(&ro_buf.borrow_bytes())?, }; - unsafe { host_fcntl::ioctl_ptr(fd, request, buf.as_mut_ptr().cast()) } - .map_err(|_| vm.new_last_errno_error())?; + vm.allow_threads(|| unsafe { + host_fcntl::ioctl_ptr(fd, request, buf.as_mut_ptr().cast()) + }) + .map_err(|_| vm.new_last_errno_error())?; Ok(vm.ctx.new_bytes(buf[..buf_len].to_vec()).into()) } Either::B(i) => { - let ret = - host_fcntl::ioctl_int(fd, request, i).map_err(|_| vm.new_last_errno_error())?; + let ret = vm + .allow_threads(|| host_fcntl::ioctl_int(fd, request, i)) + .map_err(|_| vm.new_last_errno_error())?; Ok(vm.ctx.new_int(ret).into()) } } @@ -143,7 +155,11 @@ mod fcntl { #[cfg(not(any(target_os = "wasi", target_os = "redox")))] #[pyfunction] fn flock(_io::Fildes(fd): _io::Fildes, operation: i32, vm: &VirtualMachine) -> PyResult { - let ret = host_fcntl::flock(fd, operation).map_err(|_| vm.new_last_errno_error())?; + // LOCK_EX without LOCK_NB waits for whoever holds the lock, which may + // be for good. + let ret = vm + .allow_threads(|| host_fcntl::flock(fd, operation)) + .map_err(|_| vm.new_last_errno_error())?; Ok(vm.ctx.new_int(ret).into()) } @@ -170,8 +186,10 @@ mod fcntl { OptionalArg::Present(w) => w, OptionalArg::Missing => 0, }; - let ret = - host_fcntl::lockf(fd, cmd, len, start, whence).map_err(|err| err.to_pyexception(vm))?; + // F_LOCK and F_TLOCK differ in exactly this: the first one waits. + let ret = vm + .allow_threads(|| host_fcntl::lockf(fd, cmd, len, start, whence)) + .map_err(|err| err.to_pyexception(vm))?; Ok(vm.ctx.new_int(ret).into()) } } From 11b53f1a1614da0d5e7d87700e0932d846979629 Mon Sep 17 00:00:00 2001 From: Jeong YunWon Date: Thu, 20 Aug 2026 07:39:16 +0900 Subject: [PATCH 05/15] openssl: read aside instead of holding the caller's buffer locked `SSLSocket.read` wrote straight into the destination buffer, holding the lock that reaching its bytes takes for the whole call. That read returns when the peer writes, which may be never, so a thread touching the same object waited on that lock for as long as the peer took, reaching no safepoint while it did. Read into a buffer of our own and take the destination's lock only for the copy. The rustls backend already reads this way, and `_ssl__SSLSocket_read_impl` works from a `Py_buffer` whose critical section ended before the read. `test_ssl` on this backend fails the same 16 tests before and after. Assisted-by: Claude --- crates/stdlib/src/openssl.rs | 34 +++++++++++++++------------------- 1 file changed, 15 insertions(+), 19 deletions(-) diff --git a/crates/stdlib/src/openssl.rs b/crates/stdlib/src/openssl.rs index ee9d9ae84e0..0859eed6246 100644 --- a/crates/stdlib/src/openssl.rs +++ b/crates/stdlib/src/openssl.rs @@ -3240,19 +3240,12 @@ mod _ssl { } let mut stream = self.connection.write(); - let mut inner_buffer = if let OptionalArg::Present(buffer) = &buffer { - Either::A(buffer.borrow_buf_mut()) - } else { - Either::B(vec![0u8; read_len]) - }; - let buf = match &mut inner_buffer { - Either::A(b) => &mut **b, - Either::B(b) => b.as_mut_slice(), - }; - let buf = match buf.get_mut(..read_len) { - Some(b) => b, - None => buf, - }; + // The read below answers when the peer writes, which may be never, + // and reaching the caller's buffer takes a lock that every other + // thread touching the same object waits on. Read aside and take + // that lock only for the copy. + let mut scratch = vec![0u8; read_len]; + let buf = scratch.as_mut_slice(); // BIO mode: no timeout/select logic let count = if stream.is_bio() { @@ -3312,12 +3305,15 @@ mod _ssl { return Err(convert_ssl_error(vm, err)); } }; - let ret = match inner_buffer { - Either::A(_buf) => vm.ctx.new_int(count).into(), - Either::B(mut buf) => { - buf.truncate(count); - buf.shrink_to_fit(); - vm.ctx.new_bytes(buf).into() + let ret = match &buffer { + OptionalArg::Present(buffer) => { + buffer.borrow_buf_mut()[..count].copy_from_slice(&scratch[..count]); + vm.ctx.new_int(count).into() + } + OptionalArg::Missing => { + scratch.truncate(count); + scratch.shrink_to_fit(); + vm.ctx.new_bytes(scratch).into() } }; Ok(ret) From 02a4b488164c265cef12d2637c344bffefc5c080 Mon Sep 17 00:00:00 2001 From: Jeong YunWon Date: Fri, 21 Aug 2026 10:20:43 +0900 Subject: [PATCH 06/15] vm: add the inverse of allow_threads for callbacks A call that detaches hands the thread to stop-the-world, which counts it as parked. A callback that reaches Python from inside such a call would then run on a thread the requester believes is stopped, and nothing in the vm could stop it: `attach_thread` and `detach_thread` are private to `thread.rs`, and `allow_threads` only goes the one way. `attach_for_callback` attaches for the duration of the closure and returns the thread to where it was, the way `PyGILState_Ensure` and `PyGILState_Release` bracket `_servername_callback`. It tests for "not ATTACHED" rather than for DETACHED, so a thread a stop-the-world has already moved to SUSPENDED routes through `attach_thread` and parks there until the world starts again. `a_callback_inside_a_detached_call_waits_for_the_world` stops the world with a thread detached, then turns that thread loose at a callback and asserts it does not run until the world starts. With the transition disabled it fails. Assisted-by: Claude --- crates/vm/src/vm/interpreter.rs | 88 +++++++++++++++++++++++++++++++++ crates/vm/src/vm/mod.rs | 11 +++++ crates/vm/src/vm/thread.rs | 39 +++++++++++++++ 3 files changed, 138 insertions(+) diff --git a/crates/vm/src/vm/interpreter.rs b/crates/vm/src/vm/interpreter.rs index 7f3dd8ff814..7c39427ffd1 100644 --- a/crates/vm/src/vm/interpreter.rs +++ b/crates/vm/src/vm/interpreter.rs @@ -1728,6 +1728,94 @@ for _ in range(40): worker.join().expect("worker panicked"); } + /// A callback reaching Python from inside a detached call waits for the + /// world to start again. + /// + /// Detaching for a blocking call is what lets stop-the-world count this + /// thread as parked. A callback that runs Python from in there — an SSL + /// handshake reaching a Python `sni_callback`, say — would run on a thread + /// the requester believes is stopped, so it has to attach first, and + /// attaching while the world is stopped means waiting. + #[cfg(feature = "threading")] + #[test] + fn a_callback_inside_a_detached_call_waits_for_the_world() { + use alloc::sync::Arc; + use core::{ + sync::atomic::{AtomicBool, Ordering}, + time::Duration, + }; + + let interp = Interpreter::without_stdlib(Default::default()); + let state = interp.enter(|vm| vm.state.clone()); + + let detached = Arc::new(AtomicBool::new(false)); + let ran = Arc::new(AtomicBool::new(false)); + let go = Arc::new(AtomicBool::new(false)); + + let worker_detached = Arc::clone(&detached); + let worker_ran = Arc::clone(&ran); + let worker_go = Arc::clone(&go); + let worker = interp.enter(|vm| { + let thread_vm = vm.new_thread(); + std::thread::spawn(move || { + thread_vm.run(|vm| { + vm.allow_threads(|| { + worker_detached.store(true, Ordering::Release); + // Spinning here is spinning *detached*, which is what a + // blocking call looks like to the requester: it marks + // this thread SUSPENDED and the stop completes. + while !worker_go.load(Ordering::Acquire) { + std::thread::yield_now(); + } + vm.attach_for_callback(|| worker_ran.store(true, Ordering::Release)); + }); + }); + }) + }); + + while !detached.load(Ordering::Acquire) { + std::thread::yield_now(); + } + + // Stop from a thread of its own so that a stop that never completes + // fails the test instead of hanging it. + let (stopped_tx, stopped_rx) = std::sync::mpsc::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + let stop_state = state; + let stopper = std::thread::spawn(move || { + stop_state.stop_the_world.stop_the_world(&stop_state); + stopped_tx.send(()).expect("send"); + release_rx.recv().expect("recv"); + stop_state.stop_the_world.start_the_world(&stop_state); + }); + + stopped_rx + .recv_timeout(Duration::from_secs(10)) + .expect("stop-the-world did not complete"); + + // The world is stopped; turn the worker loose at its callback. It has + // to park instead of running it, so the flag stays clear — give it the + // time it needs to get there and fail to run. + go.store(true, Ordering::Release); + std::thread::sleep(Duration::from_millis(200)); + let ran_while_stopped = ran.load(Ordering::Acquire); + + // Release before asserting: the worker has to finish for the stopper to + // be joinable, and for the test to end at all. + release_tx.send(()).expect("send"); + stopper.join().expect("stopper panicked"); + worker.join().expect("worker panicked"); + + assert!( + !ran_while_stopped, + "a callback ran Python while the world was stopped" + ); + assert!( + ran.load(Ordering::Acquire), + "the callback never ran once the world started again" + ); + } + /// The process main id is recorded once and is stable across later creates. #[test] fn process_main_id_recorded_and_stable() { diff --git a/crates/vm/src/vm/mod.rs b/crates/vm/src/vm/mod.rs index 28e59a2f477..10e0dbde626 100644 --- a/crates/vm/src/vm/mod.rs +++ b/crates/vm/src/vm/mod.rs @@ -957,6 +957,17 @@ impl VirtualMachine { thread::allow_threads(self, f) } + /// Re-attach the current thread for the duration of `f`, then return it to + /// where it was. The inverse of [`allow_threads`](Self::allow_threads), for + /// a callback that runs Python from inside a call this thread detached for. + /// + /// Equivalent to `PyGILState_Ensure` / `PyGILState_Release` around such a + /// callback. + #[inline] + pub fn attach_for_callback(&self, f: impl FnOnce() -> R) -> R { + thread::attach_for_callback(self, f) + } + /// Check whether the current thread is the main thread. /// Mirrors `_Py_ThreadCanHandleSignals`. #[allow(dead_code)] diff --git a/crates/vm/src/vm/thread.rs b/crates/vm/src/vm/thread.rs index 5e0aa076865..7b9f5102c5b 100644 --- a/crates/vm/src/vm/thread.rs +++ b/crates/vm/src/vm/thread.rs @@ -605,6 +605,45 @@ pub fn allow_threads(_vm: &VirtualMachine, f: impl FnOnce() -> R) -> R { f() } +/// Run `f` with this thread attached, then return it to where it was. +/// +/// The inverse of [`allow_threads`], for a callback that has to run Python from +/// inside a call the thread detached for — a handshake callback reaching a +/// Python `sni_callback`, say. Running that detached would execute Python on a +/// thread a stop-the-world requester counts as parked. `PyGILState_Ensure` and +/// `PyGILState_Release` bracket such a callback for the same reason. +/// +/// A thread already attached, or one with no interpreter to attach to, just +/// runs `f`. A thread a stop-the-world has already moved to SUSPENDED parks +/// here until the world starts again, because [`attach_thread`] treats that +/// state as the wait it is; that is the point of routing through it rather than +/// testing for DETACHED alone. +#[cfg(feature = "threading")] +pub fn attach_for_callback(vm: &VirtualMachine, f: impl FnOnce() -> R) -> R { + let should_transition = CURRENT_THREAD_SLOT.with(|slot| { + slot.borrow() + .as_ref() + .is_some_and(|s| s.state.load(Ordering::Acquire) != THREAD_ATTACHED) + }); + if !should_transition { + return f(); + } + + attach_thread(vm); + // Detach again even if `f` unwinds, so the `allow_threads` this is nested + // inside still finds the state it left behind. + let redetach_guard = scopeguard::guard((), |()| detach_thread()); + let result = f(); + drop(redetach_guard); + result +} + +/// No-op on non-threading builds. +#[cfg(not(feature = "threading"))] +pub fn attach_for_callback(_vm: &VirtualMachine, f: impl FnOnce() -> R) -> R { + f() +} + /// Wait for a lock the way a blocking call waits: detached, so a /// stop-the-world requester never has to wait for this thread to reach a /// safepoint it cannot reach while blocked. From 707b7c420125e8504a7a424216e0d8730af88656 Mon Sep 17 00:00:00 2001 From: Jeong YunWon Date: Fri, 21 Aug 2026 10:24:45 +0900 Subject: [PATCH 07/15] openssl: attach around the callbacks that run Python `_servername_callback` and `_msg_callback` reach the interpreter from inside an SSL call -- they take a reference to the Python callback, build arguments and call it. That call is about to detach, and running Python from a detached thread runs it on a thread a stop-the-world requester counts as parked. `_servername_callback` opens with `PyGILState_Ensure()` for the same reason. Both now rejoin the interpreter for the duration of the callback and give the thread back afterwards. The reference to the callback moves inside that section, since taking it is itself an interpreter operation; the check for whether a callback is set at all stays outside, so a socket with none set never reaches the interpreter. No behavior change yet: nothing detaches around these calls, so `attach_for_callback` finds the thread already attached and just runs. Assisted-by: Claude --- crates/stdlib/src/openssl.rs | 248 +++++++++++++++++++---------------- 1 file changed, 134 insertions(+), 114 deletions(-) diff --git a/crates/stdlib/src/openssl.rs b/crates/stdlib/src/openssl.rs index 0859eed6246..4349f23db42 100644 --- a/crates/stdlib/src/openssl.rs +++ b/crates/stdlib/src/openssl.rs @@ -672,11 +672,10 @@ mod _ssl { unsafe { let ctx = &*(arg as *const PySslContext); - // Get the callback - let callback_opt = ctx.sni_callback.lock().clone(); - let Some(callback) = callback_opt else { + // Nothing to call: leave without reaching the interpreter at all. + if ctx.sni_callback.lock().is_none() { return SSL_TLSEXT_ERR_OK; - }; + } // Get callback data from SSL ex_data let idx = get_sni_ex_data_index(); @@ -695,66 +694,77 @@ mod _ssl { }; let vm = &*vm_ptr; - // Get server name - let servername = sys::SSL_get_servername(ssl_ptr, TLSEXT_NAMETYPE_host_name); - let server_name_arg = if servername.is_null() { - vm.ctx.none() - } else { - let name_cstr = core::ffi::CStr::from_ptr(servername); - match name_cstr.to_str() { - Ok(name_str) => vm.ctx.new_str(name_str).into(), - Err(_) => vm.ctx.none(), - } - }; + // The handshake this runs inside has left the interpreter, so + // everything below rejoins it first — taking a reference to the + // callback already counts — and gives the thread back after. + vm.attach_for_callback(|| { + // Get the callback + let callback_opt = ctx.sni_callback.lock().clone(); + let Some(callback) = callback_opt else { + return SSL_TLSEXT_ERR_OK; + }; - // Get SSL socket from callback data via weak reference - let ssl_socket_obj = callback_data - .ssl_socket_weak - .upgrade() - .unwrap_or_else(|| vm.ctx.none()); + // Get server name + let servername = sys::SSL_get_servername(ssl_ptr, TLSEXT_NAMETYPE_host_name); + let server_name_arg = if servername.is_null() { + vm.ctx.none() + } else { + let name_cstr = core::ffi::CStr::from_ptr(servername); + match name_cstr.to_str() { + Ok(name_str) => vm.ctx.new_str(name_str).into(), + Err(_) => vm.ctx.none(), + } + }; - // Call the Python callback - match callback.call( - ( - ssl_socket_obj, - server_name_arg, - callback_data.ssl_context.to_owned(), - ), - vm, - ) { - Ok(result) => { - // Check return value type (must be None or integer) - if vm.is_none(&result) { - // None is OK - SSL_TLSEXT_ERR_OK - } else { - // Try to convert to integer - match result.try_to_value::(vm) { - Ok(alert_code) => { - // Valid integer - use as alert code - *al = alert_code; - SSL_TLSEXT_ERR_ALERT_FATAL - } - Err(_) => { - // Type conversion failed - raise TypeError - let type_error = vm.new_type_error(format!( + // Get SSL socket from callback data via weak reference + let ssl_socket_obj = callback_data + .ssl_socket_weak + .upgrade() + .unwrap_or_else(|| vm.ctx.none()); + + // Call the Python callback + match callback.call( + ( + ssl_socket_obj, + server_name_arg, + callback_data.ssl_context.to_owned(), + ), + vm, + ) { + Ok(result) => { + // Check return value type (must be None or integer) + if vm.is_none(&result) { + // None is OK + SSL_TLSEXT_ERR_OK + } else { + // Try to convert to integer + match result.try_to_value::(vm) { + Ok(alert_code) => { + // Valid integer - use as alert code + *al = alert_code; + SSL_TLSEXT_ERR_ALERT_FATAL + } + Err(_) => { + // Type conversion failed - raise TypeError + let type_error = vm.new_type_error(format!( "servername callback must return None or an integer, not '{}'", result.class().name() )); - vm.run_unraisable(type_error, None, result); - *al = SSL_AD_INTERNAL_ERROR; - SSL_TLSEXT_ERR_ALERT_FATAL + vm.run_unraisable(type_error, None, result); + *al = SSL_AD_INTERNAL_ERROR; + SSL_TLSEXT_ERR_ALERT_FATAL + } } } } + Err(exc) => { + // Log the exception but don't propagate it + vm.run_unraisable(exc, None, vm.ctx.none()); + *al = SSL_AD_INTERNAL_ERROR; + SSL_TLSEXT_ERR_ALERT_FATAL + } } - Err(exc) => { - // Log the exception but don't propagate it - vm.run_unraisable(exc, None, vm.ctx.none()); - *al = SSL_AD_INTERNAL_ERROR; - SSL_TLSEXT_ERR_ALERT_FATAL - } - } + }) } } @@ -794,11 +804,10 @@ mod _ssl { // ssl_socket_ptr is a pointer to Box>, set in _wrap_socket/_wrap_bio let ssl_socket: &Py = &*(ssl_socket_ptr as *const Py); - // Get the callback from the context - let callback_opt = ssl_socket.ctx.read().msg_callback.lock().clone(); - let Some(callback) = callback_opt else { + // Nothing to call: leave without reaching the interpreter at all. + if ssl_socket.ctx.read().msg_callback.lock().is_none() { return; - }; + } // Get VM from thread-local storage (set by HandshakeVmGuard in do_handshake) let Some(vm_ptr) = HANDSHAKE_VM.with(|cell| cell.get()) else { @@ -807,63 +816,74 @@ mod _ssl { }; let vm = &*vm_ptr; - // Get SSL socket owner object - let ssl_socket_obj = ssl_socket - .owner - .read() - .as_ref() - .and_then(|weak| weak.upgrade()) - .unwrap_or_else(|| vm.ctx.none()); - - // Create the message bytes - let buf_slice = core::slice::from_raw_parts(buf as *const u8, len); - let msg_bytes = vm.ctx.new_bytes(buf_slice.to_vec()); - - // Determine direction string - let direction_str = if write_p != 0 { "write" } else { "read" }; - - // Calculate msg_type based on content_type (debughelpers.c behavior) - let msg_type = match content_type { - SSL3_RT_CHANGE_CIPHER_SPEC => SSL3_MT_CHANGE_CIPHER_SPEC, - SSL3_RT_ALERT if len >= 2 => { - // byte 1 is alert type - buf_slice[1] as i32 - } - SSL3_RT_HANDSHAKE if !buf_slice.is_empty() => { - // byte 0 is handshake type - buf_slice[0] as i32 - } - SSL3_RT_HEADER if len >= 3 => { - // Frame header: version in bytes 1..2, type in byte 0 - version = ((buf_slice[1] as i32) << 8) | (buf_slice[2] as i32); - buf_slice[0] as i32 - } - SSL3_RT_INNER_CONTENT_TYPE if !buf_slice.is_empty() => { - // Inner content type in byte 0 - buf_slice[0] as i32 - } - _ => -1, - }; + // The SSL call this reports from has left the interpreter; rejoin + // it for the duration of the callback, as `_servername_callback` + // does above. + vm.attach_for_callback(|| { + // Get the callback from the context + let callback_opt = ssl_socket.ctx.read().msg_callback.lock().clone(); + let Some(callback) = callback_opt else { + return; + }; - // Call the Python callback - // Signature: callback(conn, direction, version, content_type, msg_type, data) - match callback.call( - ( - ssl_socket_obj, - vm.ctx.new_str(direction_str), - vm.ctx.new_int(version), - vm.ctx.new_int(content_type), - vm.ctx.new_int(msg_type), - msg_bytes, - ), - vm, - ) { - Ok(_) => {} - Err(exc) => { - // Log the exception but don't propagate it - vm.run_unraisable(exc, None, vm.ctx.none()); + // Get SSL socket owner object + let ssl_socket_obj = ssl_socket + .owner + .read() + .as_ref() + .and_then(|weak| weak.upgrade()) + .unwrap_or_else(|| vm.ctx.none()); + + // Create the message bytes + let buf_slice = core::slice::from_raw_parts(buf as *const u8, len); + let msg_bytes = vm.ctx.new_bytes(buf_slice.to_vec()); + + // Determine direction string + let direction_str = if write_p != 0 { "write" } else { "read" }; + + // Calculate msg_type based on content_type (debughelpers.c behavior) + let msg_type = match content_type { + SSL3_RT_CHANGE_CIPHER_SPEC => SSL3_MT_CHANGE_CIPHER_SPEC, + SSL3_RT_ALERT if len >= 2 => { + // byte 1 is alert type + buf_slice[1] as i32 + } + SSL3_RT_HANDSHAKE if !buf_slice.is_empty() => { + // byte 0 is handshake type + buf_slice[0] as i32 + } + SSL3_RT_HEADER if len >= 3 => { + // Frame header: version in bytes 1..2, type in byte 0 + version = ((buf_slice[1] as i32) << 8) | (buf_slice[2] as i32); + buf_slice[0] as i32 + } + SSL3_RT_INNER_CONTENT_TYPE if !buf_slice.is_empty() => { + // Inner content type in byte 0 + buf_slice[0] as i32 + } + _ => -1, + }; + + // Call the Python callback + // Signature: callback(conn, direction, version, content_type, msg_type, data) + match callback.call( + ( + ssl_socket_obj, + vm.ctx.new_str(direction_str), + vm.ctx.new_int(version), + vm.ctx.new_int(content_type), + vm.ctx.new_int(msg_type), + msg_bytes, + ), + vm, + ) { + Ok(_) => {} + Err(exc) => { + // Log the exception but don't propagate it + vm.run_unraisable(exc, None, vm.ctx.none()); + } } - } + }) } } From ce8a00c64f1d700c2a7ea7c0c1a55e888fb0bda8 Mon Sep 17 00:00:00 2001 From: Jeong YunWon Date: Fri, 21 Aug 2026 10:53:28 +0900 Subject: [PATCH 08/15] openssl: detach around the SSL calls that wait On a socket with a timeout the wait lands in `select` -> `sock_wait`, which already detaches. On a blocking socket there is no such return: `SSL_read` blocks in `recv(2)` through `impl Read for &PySocket`, with the thread attached and the connection's write lock held, so the world could not be stopped for as long as the peer stayed silent. `SSL_do_handshake`, `SSL_read_ex`, `SSL_write_ex` and `SSL_shutdown` all run between `Py_BEGIN_ALLOW_THREADS` and `Py_END_ALLOW_THREADS`; these now do too, in both socket and BIO mode as there. The connection lock stays held across the call and so becomes a detaching lock: a thread reaching the same socket gives up its interpreter rather than wait for it attached. `connection` is `#[pytraverse(skip)]`, so a collection does not walk into it and never takes that lock -- which is the rule for opting in, though the skip is what supplies it here rather than the missing `Traverse`. A server that completes a handshake and then says nothing used to deadlock the whole process: the collector suspended the main thread at a safepoint and then waited forever for the reader, so even the test's own timeout could not fire. It now collects in 3 ms. Verified separately that a Python `_msg_callback` still runs from inside the handshake -- 920 invocations across 40 handshakes with a collector looping, five runs clean. test_ssl on this backend fails the same 8 tests before and after, by name, and openssl.rs draws no clippy warning it did not draw before. Assisted-by: Claude --- crates/stdlib/src/openssl.rs | 47 +++++++++++++++++++++++++----------- 1 file changed, 33 insertions(+), 14 deletions(-) diff --git a/crates/stdlib/src/openssl.rs b/crates/stdlib/src/openssl.rs index 4349f23db42..7bc27e97c8e 100644 --- a/crates/stdlib/src/openssl.rs +++ b/crates/stdlib/src/openssl.rs @@ -64,8 +64,8 @@ mod _ssl { }; use crate::{ common::lock::{ - LazyLock, PyMappedRwLockReadGuard, PyMutex, PyRwLock, PyRwLockReadGuard, - PyRwLockWriteGuard, + LazyLock, PyDetachingRwLock, PyMappedRwLockReadGuard, PyMutex, PyRwLock, + PyRwLockReadGuard, PyRwLockWriteGuard, }, socket::{self, PySocket, SockWaitKind, sock_wait}, vm::{ @@ -568,7 +568,9 @@ mod _ssl { } // Get SSL pointer - either from thread-local (during handshake) or from connection - fn get_ssl_ptr_for_context_change(connection: &PyRwLock) -> *mut sys::SSL { + fn get_ssl_ptr_for_context_change( + connection: &PyDetachingRwLock, + ) -> *mut sys::SSL { // First check if we're in a handshake callback (lock already held) if let Some(ptr) = HANDSHAKE_SSL_PTR.with(|cell| cell.get()) { return ptr; @@ -2177,7 +2179,7 @@ mod _ssl { let py_ssl_socket = PySslSocket { ctx: PyRwLock::new(zelf.clone()), - connection: PyRwLock::new(SslConnection::Socket(stream)), + connection: PyDetachingRwLock::new(SslConnection::Socket(stream)), socket_type, server_hostname, owner: PyRwLock::new(args.owner.map(|o| o.downgrade(None, vm)).transpose()?), @@ -2246,7 +2248,7 @@ mod _ssl { let py_ssl_socket = PySslSocket { ctx: PyRwLock::new(zelf.clone()), - connection: PyRwLock::new(SslConnection::Bio(stream)), + connection: PyDetachingRwLock::new(SslConnection::Bio(stream)), socket_type, server_hostname, owner: PyRwLock::new(args.owner.map(|o| o.downgrade(None, vm)).transpose()?), @@ -2547,7 +2549,7 @@ mod _ssl { struct PySslSocket { ctx: PyRwLock>, #[pytraverse(skip)] - connection: PyRwLock, + connection: PyDetachingRwLock, #[pytraverse(skip)] socket_type: SslServerOrClient, server_hostname: Option, @@ -2888,7 +2890,7 @@ mod _ssl { // BIO mode: just try shutdown once and raise SSLWantReadError if needed if stream.is_bio() { - let ret = unsafe { sys::SSL_shutdown(ssl_ptr) }; + let ret = vm.allow_threads(|| unsafe { sys::SSL_shutdown(ssl_ptr) }); if ret < 0 { let err = unsafe { sys::SSL_get_error(ssl_ptr, ret) }; if err == sys::SSL_ERROR_WANT_READ { @@ -2916,7 +2918,10 @@ mod _ssl { let mut zeros = 0; loop { - let ret = unsafe { sys::SSL_shutdown(ssl_ptr) }; + // Shutting down sends close-notify and waits for the peer's, + // which a peer that has gone away never sends. `SSL_shutdown` + // is released around for the same reason. + let ret = vm.allow_threads(|| unsafe { sys::SSL_shutdown(ssl_ptr) }); // ret > 0: complete shutdown if ret > 0 { @@ -3033,7 +3038,7 @@ mod _ssl { // BIO mode: no timeout/select logic, just do handshake if stream.is_bio() { - let result = stream.do_handshake().map_err(|e| { + let result = vm.allow_threads(|| stream.do_handshake()).map_err(|e| { let exc = convert_ssl_error(vm, e); // If it's a cert verification error, set verify info if exc.class().is(PySSLCertVerificationError::class(&vm.ctx)) { @@ -3053,7 +3058,13 @@ mod _ssl { .expect("handshake called in bio mode; should only be called in socket mode") .timeout_deadline(); loop { - let err = match stream.do_handshake() { + // On a blocking socket this waits for the peer, which may never + // answer. `SSL_do_handshake` runs between + // `Py_BEGIN_ALLOW_THREADS` and `Py_END_ALLOW_THREADS` for the + // same reason. The connection lock stays held across it, which + // is why it is a detaching lock: a thread reaching the same + // socket gives up its interpreter rather than wait attached. + let err = match vm.allow_threads(|| stream.do_handshake()) { Ok(()) => { // Clean up SNI ex_data after successful handshake // SAFETY: ssl_ptr is valid for the lifetime of stream @@ -3111,7 +3122,9 @@ mod _ssl { // BIO mode: no timeout/select logic if stream.is_bio() { - return stream.ssl_write(data).map_err(|e| convert_ssl_error(vm, e)); + return vm + .allow_threads(|| stream.ssl_write(data)) + .map_err(|e| convert_ssl_error(vm, e)); } // Socket mode: handle timeout and blocking @@ -3132,7 +3145,10 @@ mod _ssl { _ => {} } loop { - let err = match stream.ssl_write(data) { + // Sending waits for the peer to make room, which it need not + // ever do; `SSL_write_ex` is released around for the same + // reason. + let err = match vm.allow_threads(|| stream.ssl_write(data)) { Ok(len) => return Ok(len), Err(e) => e, }; @@ -3269,7 +3285,7 @@ mod _ssl { // BIO mode: no timeout/select logic let count = if stream.is_bio() { - match stream.ssl_read(buf) { + match vm.allow_threads(|| stream.ssl_read(buf)) { Ok(count) => count, Err(e) => { // Handle ZERO_RETURN (EOF) - raise SSLEOFError @@ -3291,7 +3307,10 @@ mod _ssl { .expect("read called in bio mode; should only be called in socket mode") .timeout_deadline(); loop { - let err = match stream.ssl_read(buf) { + // This is the wait the whole method is shaped around: it + // ends when the peer writes. `SSL_read_ex` is released + // around for the same reason. + let err = match vm.allow_threads(|| stream.ssl_read(buf)) { Ok(count) => break count, Err(e) => e, }; From 5f2e1ea2d94fc9c4b3652b003bf2e58df544b967 Mon Sep 17 00:00:00 2001 From: Jeong YunWon Date: Fri, 21 Aug 2026 11:09:04 +0900 Subject: [PATCH 09/15] lock: assert the opt-in rule against every stop-the-world section Not implementing `Traverse` for `PyDetachingRwLock` states the rule to a collection: a payload holding one cannot derive `Traverse`, so a collection cannot walk into it. It says nothing to the other sections that stop the world -- fork, traceback dumps, frame enumeration -- and `#[pytraverse(skip)]` steps around it besides, which is how `_SSLSocket.connection` holds one. For those the rule was a comment. `set_world_stopped` records, on the one thread still running inside a stopped world, that it is that thread; `lock_shared` and `lock_exclusive` assert it is not set. A section that took one of these could block on a lock a parked thread holds and only that section can release, which is the deadlock the rule exists to prevent. Debug builds only; release builds track nothing. Nothing in the tree trips it: 180 rounds of collect, `sys._current_frames`, `faulthandler.dump_traceback` and 18 forks with four threads churning bytearrays, plus test_gc/test_bytes/test_threading/test_memoryview/test_buffer on a debug build, all clean. `taking_one_while_stopping_the_world_is_caught` takes one with the flag set and asserts the panic, so the guard is not dead code. Assisted-by: Claude --- crates/common/src/lock.rs | 2 +- crates/common/src/lock/detaching.rs | 79 +++++++++++++++++++++++++++++ crates/vm/src/vm/mod.rs | 4 ++ 3 files changed, 84 insertions(+), 1 deletion(-) diff --git a/crates/common/src/lock.rs b/crates/common/src/lock.rs index c8bfc4cef8c..134b7f2a4fa 100644 --- a/crates/common/src/lock.rs +++ b/crates/common/src/lock.rs @@ -8,7 +8,7 @@ use lock_api::{ cfg_select! { feature = "threading" => { - pub use detaching::{BlockingWaitHook, set_blocking_wait_hook}; + pub use detaching::{BlockingWaitHook, set_blocking_wait_hook, set_world_stopped}; pub use parking_lot::{RawMutex, RawRwLock, RawThreadId}; pub use std::sync::OnceLock as OnceCell; pub use core::cell::LazyCell; diff --git a/crates/common/src/lock/detaching.rs b/crates/common/src/lock/detaching.rs index e5700ce8897..92d9d639aa3 100644 --- a/crates/common/src/lock/detaching.rs +++ b/crates/common/src/lock/detaching.rs @@ -64,6 +64,47 @@ impl Drop for HookGuard { } } +#[cfg(all(feature = "threading", debug_assertions))] +std::thread_local! { + /// Set on the one thread still running while the world is stopped. + static WORLD_STOPPED: Cell = const { Cell::new(false) }; +} + +/// Record whether this thread is the one running inside a stopped world. +/// +/// The rule for opting a lock into detaching is that nothing reachable from a +/// stop-the-world section takes it — a section that did could block on a lock +/// only that same section can release. Not implementing `Traverse` states the +/// rule to a collection; this states it to every other section, which is +/// otherwise unchecked. Debug builds only; release builds track nothing and +/// pay nothing. +#[cfg(feature = "threading")] +#[inline] +pub fn set_world_stopped(stopped: bool) { + #[cfg(debug_assertions)] + let _ = WORLD_STOPPED.try_with(|flag| flag.set(stopped)); + #[cfg(not(debug_assertions))] + let _ = stopped; +} + +/// Panics if a stop-the-world section is taking one of these locks. +#[cfg(all(feature = "threading", debug_assertions))] +#[track_caller] +fn assert_not_stopping_the_world() { + // `try_with` fails only once thread locals are being destroyed, which is + // not a point at which this thread is driving a stop. + let stopped = WORLD_STOPPED.try_with(Cell::get).unwrap_or(false); + assert!( + !stopped, + "a stop-the-world section took a detaching lock, which a parked thread \ + may be holding and only this section can release" + ); +} + +#[cfg(not(all(feature = "threading", debug_assertions)))] +#[inline(always)] +fn assert_not_stopping_the_world() {} + /// Block on `wait`, detached from this thread's interpreter if there is one. /// /// Nothing spins on the way here. The lock underneath already spins before it @@ -143,6 +184,7 @@ unsafe impl RawRwLockTrait for RawDetachingRwLock { #[inline] fn lock_shared(&self) { + assert_not_stopping_the_world(); if !self.0.try_lock_shared() { wait_detached(|| self.0.lock_shared()); } @@ -160,6 +202,7 @@ unsafe impl RawRwLockTrait for RawDetachingRwLock { #[inline] fn lock_exclusive(&self) { + assert_not_stopping_the_world(); if !self.0.try_lock_exclusive() { wait_detached(|| self.0.lock_exclusive()); } @@ -256,3 +299,39 @@ unsafe impl RawRwLockRecursiveTrait for RawDetachingRwLock { self.0.try_lock_shared_recursive() } } + +#[cfg(test)] +mod tests { + #[cfg(all(feature = "threading", debug_assertions))] + use super::set_world_stopped; + #[cfg(all(feature = "threading", debug_assertions))] + use crate::lock::PyDetachingRwLock; + + /// The opt-in rule holds for every stop-the-world section, not only the + /// collector that not implementing `Traverse` speaks to. + #[cfg(all(feature = "threading", debug_assertions))] + #[test] + fn taking_one_while_stopping_the_world_is_caught() { + let lock = PyDetachingRwLock::new(()); + + // Ordinary use, for contrast. + drop(lock.write()); + + set_world_stopped(true); + let hook = std::panic::take_hook(); + std::panic::set_hook(Box::new(|_| {})); + let taken = std::panic::catch_unwind(core::panic::AssertUnwindSafe(|| { + let _guard = lock.read(); + })); + std::panic::set_hook(hook); + set_world_stopped(false); + + assert!( + taken.is_err(), + "a stop-the-world section took a detaching lock and nothing complained" + ); + + // The flag is per-thread and back to clear, so the lock still works. + drop(lock.write()); + } +} diff --git a/crates/vm/src/vm/mod.rs b/crates/vm/src/vm/mod.rs index 10e0dbde626..9f2d6b25013 100644 --- a/crates/vm/src/vm/mod.rs +++ b/crates/vm/src/vm/mod.rs @@ -441,6 +441,7 @@ impl StopTheWorldState { self.park_detached_threads(state); if initial_countdown == 0 || self.all_non_requester_suspended(state) { self.world_stopped.store(true, Ordering::Release); + crate::common::lock::set_world_stopped(true); #[cfg(debug_assertions)] self.debug_assert_all_non_requester_suspended(state); stw_trace(format_args!( @@ -488,6 +489,7 @@ impl StopTheWorldState { } } self.world_stopped.store(true, Ordering::Release); + crate::common::lock::set_world_stopped(true); #[cfg(debug_assertions)] self.debug_assert_all_non_requester_suspended(state); stw_trace(format_args!( @@ -508,6 +510,7 @@ impl StopTheWorldState { // thread-slot initialization. self.requested.store(false, Ordering::Release); self.world_stopped.store(false, Ordering::Release); + crate::common::lock::set_world_stopped(false); #[expect( clippy::iter_over_hash_type, @@ -545,6 +548,7 @@ impl StopTheWorldState { pub fn reset_after_fork(&self) { self.requested.store(false, Ordering::Relaxed); self.world_stopped.store(false, Ordering::Relaxed); + crate::common::lock::set_world_stopped(false); self.requester.store(0, Ordering::Relaxed); self.thread_countdown.store(0, Ordering::Relaxed); // The surviving child thread inherited the exclusion taken by the From f6c27e7b4c0cda82e88ee35c8e266c0a9f8462f0 Mon Sep 17 00:00:00 2001 From: Jeong YunWon Date: Fri, 21 Aug 2026 13:59:39 +0900 Subject: [PATCH 10/15] clippy: drop std_instead_of_core suppressions 1.98 no longer reports `std::io` items for the lint, so the six `expect` attributes for it are unfulfilled. Also drops `from_iter_instead_of_collect` from the workspace lint table, which 1.98 removed. Assisted-by: Claude --- Cargo.toml | 1 - 1 file changed, 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index 3b489a88687..74bc202e2b2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -398,7 +398,6 @@ explicit_iter_loop = "warn" filter_map_next = "warn" flat_map_option = "warn" format_collect = "warn" -from_iter_instead_of_collect = "warn" inconsistent_struct_constructor = "warn" index_refutable_slice = "warn" inefficient_to_string = "warn" From c2af8b69ff708f658a222891ff86b2637ffd12e1 Mon Sep 17 00:00:00 2001 From: Jeong YunWon Date: Fri, 21 Aug 2026 23:12:14 +0900 Subject: [PATCH 11/15] lock: spell out why the detaching rwlock exists States the motivation as the three-thread cycle it avoids: how a stop reaches a DETACHED thread but not an ATTACHED one, why a thread blocked on a lock reaches no safepoint, and why the waiter rather than the holder gives way. Puts it on `RawDetachingRwLock`, which is public and so rendered; the module doc is private and keeps only the hook description. Assisted-by: Claude --- crates/common/src/lock/detaching.rs | 66 +++++++++++++++++++++-------- 1 file changed, 49 insertions(+), 17 deletions(-) diff --git a/crates/common/src/lock/detaching.rs b/crates/common/src/lock/detaching.rs index 92d9d639aa3..14987a5709f 100644 --- a/crates/common/src/lock/detaching.rs +++ b/crates/common/src/lock/detaching.rs @@ -1,23 +1,16 @@ //! A reader-writer lock that lets a thread leave its interpreter before it //! blocks. //! -//! Stopping the world means waiting for every running thread to reach a -//! safepoint. A thread blocked on a lock reaches none, so if the thread holding -//! that lock has already been stopped, the two wait on each other forever. The -//! holder is not the one who can avoid this — a lock is held across a blocking -//! call precisely because that is what the call needs — so the waiter gives up -//! its interpreter for the duration of the wait instead, which is what a -//! blocking call does anyway. +//! [`RawDetachingRwLock`] carries the reasoning: what goes wrong when a thread +//! waits for a lock while attached, why the waiter rather than the holder is +//! the one that has to give way, and the rule that comes with fixing it. //! -//! Doing so is safe only for locks nothing reachable from a stop-the-world -//! section takes, so it is opt-in per lock — see [`RawDetachingRwLock`] for the -//! rule and why it is needed. -//! -//! Only the contended path pays for any of this: an acquire that takes the lock -//! on the first try is the same atomic exchange it was, and never reaches the -//! hook. The hook is installed by whoever knows how to detach a thread -//! ([`set_blocking_wait_hook`]); until then, and on any thread that is not -//! running an interpreter, a blocked acquire just blocks. +//! The wait itself is handed to a hook, because this crate cannot depend on the +//! vm and so cannot detach a thread by itself. Whoever can installs it through +//! [`set_blocking_wait_hook`]; until then, and on any thread that is not +//! running an interpreter, a blocked acquire just blocks. Only the contended +//! path reaches any of this — an acquire that takes the lock on its first try +//! is the same atomic exchange it was. use super::RawRwLock; #[cfg(feature = "threading")] @@ -147,7 +140,46 @@ fn wait_detached(wait: impl Fn()) { /// /// Use through [`PyDetachingRwLock`](super::PyDetachingRwLock). /// -/// # Only for locks a collection never takes +/// # Why this exists +/// +/// Stopping the world means waiting until every other thread sits at +/// SUSPENDED, and there are two ways a thread gets there: +/// +/// - A DETACHED thread is not running interpreter code, so the requester moves +/// it to SUSPENDED itself. The thread never finds out. +/// - An ATTACHED thread can only suspend itself, at a safepoint — the check +/// `check_signals` makes between bytecodes. +/// +/// A thread blocked acquiring a lock runs no bytecode, so it reaches no +/// safepoint. While ATTACHED it is a thread the world cannot stop for as long +/// as it waits, and the requester waits without a bound. +/// +/// On its own that is a pause. It becomes a deadlock as soon as the lock being +/// waited for is held by a thread the same stop has already parked: +/// +/// ```text +/// A holds the lock, blocks inside allow_threads -> DETACHED +/// B requests a stop, and parks A -> A is SUSPENDED, holding the lock +/// C wants the same lock, and waits for it -> ATTACHED, blocked +/// +/// B waits for C to suspend C reaches no safepoint +/// C waits for A to release A is parked +/// A waits for B to start the world B is still waiting for C +/// ``` +/// +/// No thread in that cycle can break it, because none of them is running. It is +/// not hypothetical: an `SSLSocket.read` against a peer that completed a +/// handshake and then went quiet froze whole processes this way, the main +/// thread included, so not even a Python-level timeout could fire. +/// +/// The holder cannot be the one to give way. A lock is held across a blocking +/// call precisely because that is what the call needs. So the waiter gives way +/// instead: it leaves its interpreter for the duration of the wait, which is +/// what a blocking call does anyway, and a waiter that has left is a waiter the +/// requester can park. C detaches before it blocks, the stop completes, B +/// finishes, A resumes and releases, and C takes the lock and attaches again. +/// +/// # Only for locks a stop-the-world section never takes /// /// The wait acquires the lock while detached, so the thread comes back holding /// it, and re-attaching is a point at which a stop-the-world in flight will From 657964b9eff05ed7c138e79e3ab563fb55e811aa Mon Sep 17 00:00:00 2001 From: Jeong YunWon Date: Fri, 28 Aug 2026 00:14:02 +0900 Subject: [PATCH 12/15] fcntl: raise the error the call returned The six detached calls discarded the `io::Error` and read `errno` again after `allow_threads`, which re-attaches in between and can park on the way. `lockf` already converts the returned error; these now do too. Assisted-by: Claude --- crates/stdlib/src/fcntl.rs | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/crates/stdlib/src/fcntl.rs b/crates/stdlib/src/fcntl.rs index d482bf256b6..dca53c104dd 100644 --- a/crates/stdlib/src/fcntl.rs +++ b/crates/stdlib/src/fcntl.rs @@ -79,7 +79,7 @@ mod fcntl { .copy_from_slice(&s) } vm.allow_threads(|| host_fcntl::fcntl_with_bytes(fd, cmd, &mut buf[..arg_len])) - .map_err(|_| vm.new_last_errno_error())?; + .map_err(|err| err.to_pyexception(vm))?; return Ok(vm.ctx.new_bytes(buf[..arg_len].to_vec()).into()); } OptionalArg::Present(Either::B(i)) => i.as_u32_mask(), @@ -87,7 +87,7 @@ mod fcntl { }; let ret = vm .allow_threads(|| host_fcntl::fcntl_int(fd, cmd, int as i32)) - .map_err(|_| vm.new_last_errno_error())?; + .map_err(|err| err.to_pyexception(vm))?; Ok(vm.new_pyobj(ret)) } @@ -127,7 +127,7 @@ mod fcntl { .allow_threads(|| unsafe { host_fcntl::ioctl_ptr(fd, request, scratch.as_mut_ptr().cast()) }) - .map_err(|_| vm.new_last_errno_error())?; + .map_err(|err| err.to_pyexception(vm))?; rw_arg.borrow_buf_mut().copy_from_slice(&scratch); return Ok(vm.ctx.new_int(ret).into()); } @@ -139,13 +139,13 @@ mod fcntl { vm.allow_threads(|| unsafe { host_fcntl::ioctl_ptr(fd, request, buf.as_mut_ptr().cast()) }) - .map_err(|_| vm.new_last_errno_error())?; + .map_err(|err| err.to_pyexception(vm))?; Ok(vm.ctx.new_bytes(buf[..buf_len].to_vec()).into()) } Either::B(i) => { let ret = vm .allow_threads(|| host_fcntl::ioctl_int(fd, request, i)) - .map_err(|_| vm.new_last_errno_error())?; + .map_err(|err| err.to_pyexception(vm))?; Ok(vm.ctx.new_int(ret).into()) } } @@ -159,7 +159,7 @@ mod fcntl { // be for good. let ret = vm .allow_threads(|| host_fcntl::flock(fd, operation)) - .map_err(|_| vm.new_last_errno_error())?; + .map_err(|err| err.to_pyexception(vm))?; Ok(vm.ctx.new_int(ret).into()) } From 45331ef151fcdc3b731a8f6f91edca54c2fd73ed Mon Sep 17 00:00:00 2001 From: Jeong YunWon Date: Fri, 28 Aug 2026 00:14:03 +0900 Subject: [PATCH 13/15] openssl: allocate the read buffer through the vm Without a `buffer` argument `read_len` is whatever non-negative size the caller passed, so `vec![0u8; read_len]` aborts on allocation failure. `new_zeroed_bytes` raises `MemoryError` instead, as the two other reads in this file already do. Assisted-by: Claude --- crates/stdlib/src/openssl.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/stdlib/src/openssl.rs b/crates/stdlib/src/openssl.rs index 7bc27e97c8e..6a35a30dc89 100644 --- a/crates/stdlib/src/openssl.rs +++ b/crates/stdlib/src/openssl.rs @@ -3280,7 +3280,7 @@ mod _ssl { // and reaching the caller's buffer takes a lock that every other // thread touching the same object waits on. Read aside and take // that lock only for the copy. - let mut scratch = vec![0u8; read_len]; + let mut scratch = vm.new_zeroed_bytes(read_len)?; let buf = scratch.as_mut_slice(); // BIO mode: no timeout/select logic From 809b7fd89dd9c8ecd32a17aa60047ae55db961e9 Mon Sep 17 00:00:00 2001 From: Jeong YunWon Date: Fri, 28 Aug 2026 00:14:13 +0900 Subject: [PATCH 14/15] lock: leave no attached blocking acquire on a detaching lock `lock_upgradable` starts from holding nothing, so it detaches like `lock_shared`. A recursive read cannot: it may be the re-entrant take of a lock this thread holds, so detaching there parks a thread holding it, and staying attached stalls stop-the-world. `RawRwLockRecursive` is no longer implemented, which removes `read_recursive` from these locks; nothing took one. The assertion test no longer replaces the panic hook, which is process-wide and was suppressing panic output from whatever else ran beside it. Assisted-by: Claude --- crates/common/src/lock/detaching.rs | 48 +++++++++++++---------------- 1 file changed, 22 insertions(+), 26 deletions(-) diff --git a/crates/common/src/lock/detaching.rs b/crates/common/src/lock/detaching.rs index 14987a5709f..b662e5d85a1 100644 --- a/crates/common/src/lock/detaching.rs +++ b/crates/common/src/lock/detaching.rs @@ -16,8 +16,8 @@ use super::RawRwLock; #[cfg(feature = "threading")] use core::cell::Cell; use lock_api::{ - RawRwLock as RawRwLockTrait, RawRwLockDowngrade, RawRwLockRecursive as RawRwLockRecursiveTrait, - RawRwLockUpgrade as RawRwLockUpgradeTrait, RawRwLockUpgradeDowngrade, + RawRwLock as RawRwLockTrait, RawRwLockDowngrade, RawRwLockUpgrade as RawRwLockUpgradeTrait, + RawRwLockUpgradeDowngrade, }; #[cfg(feature = "threading")] use std::sync::OnceLock; @@ -269,18 +269,19 @@ unsafe impl RawRwLockDowngrade for RawDetachingRwLock { } } -// SAFETY: forwards to the wrapped raw lock. +// SAFETY: forwards to the wrapped raw lock; `lock_upgradable` only adds a wait +// that ends with the same lock acquired. // -// None of these detach. `upgrade` runs with the upgradable lock already held, -// and `lock_shared_recursive` may be the re-entrant take of a lock this thread -// holds; detaching there would park a thread *holding* the lock, the one thing -// this type must not do. `lock_upgradable` starts from holding nothing and -// could detach as safely as `lock_shared` does, but nothing takes an upgradable -// read of one of these, so it does not. +// `lock_upgradable` detaches for the same reason `lock_shared` does: it starts +// from holding nothing, so the wait cannot park a thread that holds the lock. +// `upgrade` does not, because it runs with the upgradable lock already held. unsafe impl RawRwLockUpgradeTrait for RawDetachingRwLock { #[inline] fn lock_upgradable(&self) { - self.0.lock_upgradable() + assert_not_stopping_the_world(); + if !self.0.try_lock_upgradable() { + wait_detached(|| self.0.lock_upgradable()); + } } #[inline] @@ -318,19 +319,13 @@ unsafe impl RawRwLockUpgradeDowngrade for RawDetachingRwLock { } } -// SAFETY: forwards to the wrapped raw lock. Does not detach; see the upgrade -// impl above. -unsafe impl RawRwLockRecursiveTrait for RawDetachingRwLock { - #[inline] - fn lock_shared_recursive(&self) { - self.0.lock_shared_recursive() - } - - #[inline] - fn try_lock_shared_recursive(&self) -> bool { - self.0.try_lock_shared_recursive() - } -} +// `RawRwLockRecursive` is deliberately not implemented, so that `read_recursive` +// does not exist on these locks. It is the one blocking acquire that cannot +// detach: a recursive read may be the re-entrant take of a lock this thread +// already holds, and detaching there parks a thread *holding* the lock, which is +// the deadlock this type exists to avoid. Leaving it implemented but attached +// would instead leave an acquire that stalls stop-the-world, so neither form of +// it belongs here. #[cfg(test)] mod tests { @@ -349,13 +344,14 @@ mod tests { // Ordinary use, for contrast. drop(lock.write()); + // The panic below is the expected result, so it prints where an + // unexpected one would. Silencing it would mean replacing the panic hook, + // which is process-wide and would swallow the output of whatever else the + // test binary is running at the same time. set_world_stopped(true); - let hook = std::panic::take_hook(); - std::panic::set_hook(Box::new(|_| {})); let taken = std::panic::catch_unwind(core::panic::AssertUnwindSafe(|| { let _guard = lock.read(); })); - std::panic::set_hook(hook); set_world_stopped(false); assert!( From b0f546754d93ef13e7c7120b624ed5bffc460a6a Mon Sep 17 00:00:00 2001 From: Jeong YunWon Date: Fri, 28 Aug 2026 00:14:41 +0900 Subject: [PATCH 15/15] vm: wait for the actual block in the stop-the-world test The worker signalled before `read()` and the test slept 50ms, so a stop could complete with nothing blocked on the lock and the test would pass having checked nothing. It now publishes its thread id from inside the interpreter and the test waits for that slot to reach DETACHED, bounded so an acquire that never detaches fails rather than hangs. Assisted-by: Claude --- crates/vm/src/vm/interpreter.rs | 39 ++++++++++++++++++++++++++------- 1 file changed, 31 insertions(+), 8 deletions(-) diff --git a/crates/vm/src/vm/interpreter.rs b/crates/vm/src/vm/interpreter.rs index 7c39427ffd1..8545d6152df 100644 --- a/crates/vm/src/vm/interpreter.rs +++ b/crates/vm/src/vm/interpreter.rs @@ -1669,10 +1669,11 @@ for _ in range(40): #[cfg(feature = "threading")] #[test] fn a_thread_blocked_on_a_lock_does_not_stall_stop_the_world() { + use super::super::thread::THREAD_DETACHED; use crate::common::lock::PyDetachingRwLock; use alloc::sync::Arc; use core::{ - sync::atomic::{AtomicBool, Ordering}, + sync::atomic::{AtomicU64, Ordering}, time::Duration, }; @@ -1680,29 +1681,51 @@ for _ in range(40): let state = interp.enter(|vm| vm.state.clone()); let lock: Arc> = Arc::new(PyDetachingRwLock::new(())); - let at_lock = Arc::new(AtomicBool::new(false)); + // The worker's thread id, published from inside the interpreter. No + // thread has id 0, so it doubles as "not registered yet". + let worker_ident = Arc::new(AtomicU64::new(0)); // Held for the whole test, so the worker below blocks and stays blocked. let held = lock.write(); let worker_lock = Arc::clone(&lock); - let worker_at_lock = Arc::clone(&at_lock); + let published_ident = Arc::clone(&worker_ident); let worker = interp.enter(|vm| { let thread_vm = vm.new_thread(); std::thread::spawn(move || { thread_vm.run(|_vm| { - worker_at_lock.store(true, Ordering::Release); + published_ident.store(crate::stdlib::_thread::get_ident(), Ordering::Release); let _read = worker_lock.read(); }); }) }); - while !at_lock.load(Ordering::Acquire) { + // Wait for the worker to have blocked, not merely to have been scheduled + // to. It publishes its id while attached, so that slot reaching DETACHED + // is the contended acquire leaving the interpreter — the state this test + // is about. A sleep here would let the stop below complete with no + // blocked waiter at all, and pass without testing anything. + // + // Bounded, so an acquire that never detaches fails the test instead of + // hanging it, as the timeout on the stop below does. + let deadline = std::time::Instant::now() + Duration::from_secs(10); + let blocked_detached = |ident| { + state + .thread_frames + .lock() + .get(&ident) + .is_some_and(|slot| slot.state.load(Ordering::Acquire) == THREAD_DETACHED) + }; + loop { + match worker_ident.load(Ordering::Acquire) { + ident if ident != 0 && blocked_detached(ident) => break, + _ => assert!( + std::time::Instant::now() < deadline, + "the worker never detached for the contended acquire" + ), + } std::thread::yield_now(); } - // The store above only says the worker is about to block, not that it - // has; give it the moment it needs to get there. - std::thread::sleep(Duration::from_millis(50)); // Stop from a thread of its own so that a stop that never completes // fails the test instead of hanging it.