Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion crates/vm/src/builtins/memory.rs
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,9 @@ impl PyMemoryView {
/// `send_buffer` so the destination memoryview sees the same memory.
#[must_use]
pub fn clone_buffer(&self) -> PyBuffer {
self.buffer.clone()
let mut buffer = self.buffer.clone();
buffer.desc = self.desc.clone();
buffer
}

// PyMemoryView_FromObjectAndFlags
Expand Down
27 changes: 27 additions & 0 deletions crates/vm/src/stdlib/_interpchannels.rs
Original file line number Diff line number Diff line change
Expand Up @@ -230,6 +230,9 @@ pub(crate) mod _interpchannels {
}

fn release_end(&mut self, index: usize, send: bool) {
if !self.list(send)[index].1 {
return;
}
self.list(send)[index].1 = false;
if send {
self.numsendopen -= 1;
Expand Down Expand Up @@ -1295,4 +1298,28 @@ pub(crate) mod _interpchannels {
}
Ok(())
}

#[cfg(test)]
mod tests {
use super::ChannelEnds;

#[test]
fn releasing_an_end_twice_is_a_noop() {
let mut ends = ChannelEnds::default();
ends.add(1, true);
ends.add(1, false);

let send = ends.find(1, true).unwrap();
ends.release_end(send, true);
ends.release_end(send, true);
assert_eq!(ends.numsendopen, 0);
assert_eq!(ends.numrecvopen, 1);
assert!(ends.is_open());

let recv = ends.find(1, false).unwrap();
ends.release_end(recv, false);
assert_eq!(ends.numrecvopen, 0);
assert!(!ends.is_open());
}
}
}
32 changes: 29 additions & 3 deletions crates/vm/src/vm/interpreter.rs
Original file line number Diff line number Diff line change
Expand Up @@ -148,6 +148,7 @@ where
let warnings = WarningsState::init_state(&ctx);

let interpreter_id = runtime::alloc_interpreter_id();
let runtime_root_id = parent_state.map_or(interpreter_id, |parent| parent.runtime_root_id);

// Process main OS thread identity is process-global; subinterpreters inherit
// it from the parent so `is_main_thread()` stays correct when running on the
Expand All @@ -162,6 +163,7 @@ where
let global_state = PyRc::new(PyGlobalState {
gc: crate::gc_state::GcInterpreterState::new(&ctx),
interpreter_id,
runtime_root_id,
whence,
is_main,
config,
Expand Down Expand Up @@ -735,9 +737,9 @@ fn finalize_subinterpreters(vm: &VirtualMachine) {
if !vm.state.is_main {
return;
}
let ids = runtime::owned_interpreter_ids();
let root_id = vm.state.runtime_root_id;
// Bail out if there are no subinterpreters left.
if ids.is_empty() {
if runtime::owned_interpreter_ids_for(root_id).is_empty() {
return;
}
// Warn the user if they forgot to clean up subinterpreters.
Expand All @@ -751,7 +753,12 @@ fn finalize_subinterpreters(vm: &VirtualMachine) {
None,
vm,
);
for id in ids {
// A subinterpreter's finalizers may create another subinterpreter, so
// re-read the owner table after each destruction like CPython does.
while let Some(id) = runtime::owned_interpreter_ids_for(root_id)
.into_iter()
.next()
{
let _ = runtime::destroy_owned_interpreter(id);
}
}
Expand Down Expand Up @@ -1557,6 +1564,25 @@ mod tests {
assert!(!sub.is_main());
}

/// Finalizing one embedded runtime must leave another runtime's owned
/// subinterpreters alone.
#[cfg(feature = "threading")]
#[test]
fn owned_subinterpreter_cleanup_is_scoped_to_its_runtime() {
let main1 = Interpreter::without_stdlib(Default::default());
let main2 = Interpreter::without_stdlib(Default::default());
let sub1 = main1.create_owned_subinterpreter();
let sub2 = main2.create_owned_subinterpreter();

main1.enter(finalize_subinterpreters);

assert!(!runtime::is_owned_interpreter(sub1));
assert!(runtime::is_owned_interpreter(sub2));

let sub2 = runtime::take_owned_interpreter(sub2).expect("owned by second runtime");
let _ = sub2.finalize(None);
}

/// A collection must stop every interpreter, not just the collecting one:
/// the generation lists are process-global, so the reachability walk reads
/// objects owned by other interpreters while their threads would otherwise
Expand Down
2 changes: 2 additions & 0 deletions crates/vm/src/vm/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -759,6 +759,8 @@ pub(crate) struct CallableCache {
pub struct PyGlobalState {
/// Unique process-global interpreter id (main is [`MAIN_INTERPRETER_ID`]).
pub interpreter_id: i64,
/// Top-level interpreter whose runtime owns this interpreter.
pub runtime_root_id: i64,
/// How this interpreter was created.
pub whence: runtime::InterpreterWhence,
/// True for every top-level (non-sub) interpreter, each of which keeps its
Expand Down
39 changes: 34 additions & 5 deletions crates/vm/src/vm/runtime.rs
Original file line number Diff line number Diff line change
Expand Up @@ -452,19 +452,32 @@ pub fn live_interpreter_states() -> Vec<PyRc<PyGlobalState>> {
/// via [`owned_new_thread`], which only clones the shared interpreter fields
/// (`sys`, builtins, `PyGlobalState`).
#[cfg(feature = "threading")]
fn owned_interpreters() -> &'static Mutex<HashMap<i64, crate::Interpreter>> {
struct OwnedInterpreter {
root_id: i64,
interpreter: crate::Interpreter,
}

#[cfg(feature = "threading")]
fn owned_interpreters() -> &'static Mutex<HashMap<i64, OwnedInterpreter>> {
use std::sync::OnceLock;
static OWNED: OnceLock<Mutex<HashMap<i64, crate::Interpreter>>> = OnceLock::new();
static OWNED: OnceLock<Mutex<HashMap<i64, OwnedInterpreter>>> = OnceLock::new();
OWNED.get_or_init(|| Mutex::new(HashMap::new()))
}

/// Transfer ownership of `interp` to the runtime, returning its id.
#[cfg(feature = "threading")]
pub fn store_owned_interpreter(interp: crate::Interpreter) -> i64 {
let id = interp.id();
let root_id = interp.global_state.runtime_root_id;
// Ids are strictly monotonic, so this never displaces (and drops) an
// existing entry under the lock.
owned_interpreters().lock().insert(id, interp);
owned_interpreters().lock().insert(
id,
OwnedInterpreter {
root_id,
interpreter: interp,
},
);
id
}

Expand All @@ -475,7 +488,10 @@ pub fn store_owned_interpreter(interp: crate::Interpreter) -> i64 {
#[cfg(feature = "threading")]
#[must_use]
pub fn take_owned_interpreter(id: i64) -> Option<crate::Interpreter> {
owned_interpreters().lock().remove(&id)
owned_interpreters()
.lock()
.remove(&id)
.map(|owned| owned.interpreter)
}

/// Create a thread-state VM for a runtime-owned interpreter.
Expand All @@ -487,7 +503,7 @@ pub fn owned_new_thread(id: i64) -> Option<crate::vm::thread::ThreadedVirtualMac
owned_interpreters()
.lock()
.get(&id)
.map(|interp| interp.new_thread())
.map(|owned| owned.interpreter.new_thread())
}

/// Whether `id` refers to a runtime-owned interpreter.
Expand All @@ -513,6 +529,19 @@ pub fn owned_interpreter_ids() -> Vec<i64> {
ids
}

/// Ids of runtime-owned interpreters belonging to `root_id`, oldest first.
#[cfg(feature = "threading")]
#[must_use]
pub fn owned_interpreter_ids_for(root_id: i64) -> Vec<i64> {
let mut ids: Vec<i64> = owned_interpreters()
.lock()
.iter()
.filter_map(|(&id, owned)| (owned.root_id == root_id).then_some(id))
.collect();
ids.sort_unstable();
ids
}

/// Finalize and drop a runtime-owned interpreter.
///
/// The caller must already have checked that the interpreter is not the
Expand Down
23 changes: 23 additions & 0 deletions extra_tests/snippets/stdlib_subinterpreters.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,23 @@
import _interpchannels
import _interpreters


def roundtrip_buffer(view):
cid = _interpchannels.create(3)
try:
_interpchannels.send_buffer(cid, view, blocking=False)
received, unbound = _interpchannels.recv(cid)
assert unbound is None
return received
finally:
_interpchannels.destroy(cid)


data = bytearray(range(12))
for view in (memoryview(data)[2:10:2], memoryview(data).cast("I")):
received = roundtrip_buffer(view)
assert received.tolist() == view.tolist()
assert received.shape == view.shape
assert received.strides == view.strides
assert received.format == view.format
assert received.itemsize == view.itemsize
Loading