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" 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..134b7f2a4fa 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, set_world_stopped}; 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..b662e5d85a1 --- /dev/null +++ b/crates/common/src/lock/detaching.rs @@ -0,0 +1,365 @@ +//! A reader-writer lock that lets a thread leave its interpreter before it +//! blocks. +//! +//! [`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. +//! +//! 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")] +use core::cell::Cell; +use lock_api::{ + RawRwLock as RawRwLockTrait, RawRwLockDowngrade, 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)); + } +} + +#[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 +/// 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). +/// +/// # 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 +/// 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 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); + +// 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) { + assert_not_stopping_the_world(); + 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) { + assert_not_stopping_the_world(); + 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; `lock_upgradable` only adds a wait +// that ends with the same lock acquired. +// +// `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) { + assert_not_stopping_the_world(); + 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. + unsafe { 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() } + } +} + +// `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 { + #[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()); + + // 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 taken = std::panic::catch_unwind(core::panic::AssertUnwindSafe(|| { + let _guard = lock.read(); + })); + 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/stdlib/src/fcntl.rs b/crates/stdlib/src/fcntl.rs index 8e24f2b6e4a..dca53c104dd 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]) - .map_err(|_| vm.new_last_errno_error())?; + vm.allow_threads(|| host_fcntl::fcntl_with_bytes(fd, cmd, &mut buf[..arg_len])) + .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(), 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(|err| err.to_pyexception(vm))?; 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(|err| err.to_pyexception(vm))?; + 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(|err| err.to_pyexception(vm))?; 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(|err| err.to_pyexception(vm))?; 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(|err| err.to_pyexception(vm))?; 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()) } } diff --git a/crates/stdlib/src/openssl.rs b/crates/stdlib/src/openssl.rs index ee9d9ae84e0..6a35a30dc89 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; @@ -672,11 +674,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 +696,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 +806,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 +818,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()); + } } - } + }) } } @@ -2157,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()?), @@ -2226,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()?), @@ -2527,7 +2549,7 @@ mod _ssl { struct PySslSocket { ctx: PyRwLock>, #[pytraverse(skip)] - connection: PyRwLock, + connection: PyDetachingRwLock, #[pytraverse(skip)] socket_type: SslServerOrClient, server_hostname: Option, @@ -2868,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 { @@ -2896,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 { @@ -3013,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)) { @@ -3033,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 @@ -3091,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 @@ -3112,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, }; @@ -3240,23 +3276,16 @@ 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 = vm.new_zeroed_bytes(read_len)?; + let buf = scratch.as_mut_slice(); // 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 @@ -3278,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, }; @@ -3312,12 +3344,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) 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/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] diff --git a/crates/vm/src/vm/interpreter.rs b/crates/vm/src/vm/interpreter.rs index 6d4e1f75a22..8545d6152df 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,185 @@ 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 super::super::thread::THREAD_DETACHED; + use crate::common::lock::PyDetachingRwLock; + use alloc::sync::Arc; + use core::{ + sync::atomic::{AtomicU64, 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(())); + // 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 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| { + published_ident.store(crate::stdlib::_thread::get_ident(), Ordering::Release); + let _read = worker_lock.read(); + }); + }) + }); + + // 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(); + } + + // 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"); + } + + /// 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..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 @@ -957,6 +961,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 4ba0d7ffada..7b9f5102c5b 100644 --- a/crates/vm/src/vm/thread.rs +++ b/crates/vm/src/vm/thread.rs @@ -605,6 +605,82 @@ 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. +/// +/// Threads with no interpreter to leave — a native thread, or one whose +/// locals are already being destroyed — simply block. +/// +/// 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 + // 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`).