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..e5700ce8897 --- /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 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) { + 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. +// +// 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) { + 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() } + } +} + +// 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() + } +} 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()) } } 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) 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 3e5e2393ee3..765fe6e6e31 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..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..5e0aa076865 100644 --- a/crates/vm/src/vm/thread.rs +++ b/crates/vm/src/vm/thread.rs @@ -605,6 +605,43 @@ 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. +/// +/// 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`).