mirrors/RustPython
0
0 Fork 0
mirror of https://github.com/RustPython/RustPython.git synced 2026-10-29 11:14:38 +08:00
Code Activity

lock: add an opt-in rwlock that detaches before it blocks (#8556)

Browse Source 
* 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

* 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

* 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

* 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

* 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

* 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

* 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

* 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

* 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

* 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

* 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

* 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

* 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

* 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

* 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
This commit is contained in:
Jeong, YunWon authored and GitHub committed 2026-08-28 16:20:39 +09:00
1 parent 7931cb062b
commit b0b4d2f8a9
11 files changed
+934 -189
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Cargo.toml
-1
View File
@@ -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"
crates/common/src/borrow.rs
+15 -1
View File
@@ -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<T: ?Sized> 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<T: ?Sized> 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<T: ?Sized> DerefMut for BorrowedValueMut<'_, T> {
Self::MappedMuLock(m) => &mut *m,
Self::WriteLock(w) => &mut *w,
Self::MappedWriteLock(w) => &mut *w,
Self::MappedDetachingWriteLock(w) => &mut *w,
}
}
}
crates/common/src/lock.rs
+16
View File
@@ -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<T> = ThreadMutex<RawMutex, RawThreadId, T>;
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<T> = RwLock<RawDetachingRwLock, T>;
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<T> = RwLock<RawRwLock, T>;
pub type PyRwLockUpgradableReadGuard<'a, T> = RwLockUpgradableReadGuard<'a, RawRwLock, T>;
pub type PyRwLockReadGuard<'a, T> = RwLockReadGuard<'a, RawRwLock, T>;
crates/common/src/lock/detaching.rs
+365
View File
@@ -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<BlockingWaitHook> = 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<bool> = 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<bool> = 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(<RawRwLock as RawRwLockTrait>::INIT);
type GuardMarker = <RawRwLock as RawRwLockTrait>::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());
}
}
crates/stdlib/src/fcntl.rs
+35 -17
View File
@@ -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())
}
}
crates/stdlib/src/openssl.rs
+179 -144
View File
@@ -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<SslConnection>) -> *mut sys::SSL {
fn get_ssl_ptr_for_context_change(
connection: &PyDetachingRwLock<SslConnection>,
) -> *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::<i32>(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::<i32>(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<Py<PySslSocket>>, set in _wrap_socket/_wrap_bio
let ssl_socket: &Py<PySslSocket> = &*(ssl_socket_ptr as *const Py<PySslSocket>);
// 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());
// 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;
};
// 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());
// 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());
// Determine direction string
let direction_str = if write_p != 0 { "write" } else { "read" };
// 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());
// 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,
};
// Determine direction string
let direction_str = if write_p != 0 { "write" } else { "read" };
// 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());
// 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<PyRef<PySslContext>>,
#[pytraverse(skip)]
connection: PyRwLock<SslConnection>,
connection: PyDetachingRwLock<SslConnection>,
#[pytraverse(skip)]
socket_type: SslServerOrClient,
server_hostname: Option<PyStrRef>,
@@ -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)
crates/vm/src/builtins/bytearray.rs
+15 -14
View File
@@ -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<PyBytesInner>,
inner: PyDetachingRwLock<PyBytesInner>,
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<u8>> {
PyRwLockWriteGuard::map(self.inner.write(), |inner| &mut inner.elements)
pub fn borrow_buf_mut(&self) -> PyMappedDetachingRwLockWriteGuard<'_, Vec<u8>> {
PyDetachingRwLockWriteGuard::map(self.inner.write(), |inner| &mut inner.elements)
}
fn repeat(&self, value: isize, vm: &VirtualMachine) -> PyResult<Self> {
@@ -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::<PyByteArray>().borrow_buf().into(),
obj_bytes_mut: |buffer| {
PyMappedRwLockWriteGuard::map(buffer.obj_as::<PyByteArray>().borrow_buf_mut(), |x| {
x.as_mut_slice()
})
PyMappedDetachingRwLockWriteGuard::map(
buffer.obj_as::<PyByteArray>().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<Self::Resizable<'_>> {
// An export is a borrow someone else still holds, so it is answered
crates/vm/src/stdlib/os.rs
+35 -12
View File
@@ -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<usize> {
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<usize> {
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]
crates/vm/src/vm/interpreter.rs
+183
View File
@@ -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<PyDetachingRwLock<()>> = 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() {
crates/vm/src/vm/mod.rs
+15
View File
@@ -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<R>(&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)]
crates/vm/src/vm/thread.rs
+76
View File
@@ -605,6 +605,82 @@ pub fn allow_threads<R>(_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<R>(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<R>(_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`).