diff --git a/crates/vm/src/builtins/memory.rs b/crates/vm/src/builtins/memory.rs index 2e3403dcd..b67bfbe9f 100644 --- a/crates/vm/src/builtins/memory.rs +++ b/crates/vm/src/builtins/memory.rs @@ -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 diff --git a/crates/vm/src/stdlib/_interpchannels.rs b/crates/vm/src/stdlib/_interpchannels.rs index 5d0a5af24..87bb9e060 100644 --- a/crates/vm/src/stdlib/_interpchannels.rs +++ b/crates/vm/src/stdlib/_interpchannels.rs @@ -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; @@ -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()); + } + } } diff --git a/crates/vm/src/vm/interpreter.rs b/crates/vm/src/vm/interpreter.rs index 26c302b00..db806e067 100644 --- a/crates/vm/src/vm/interpreter.rs +++ b/crates/vm/src/vm/interpreter.rs @@ -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 @@ -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, @@ -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. @@ -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); } } @@ -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 diff --git a/crates/vm/src/vm/mod.rs b/crates/vm/src/vm/mod.rs index b37e69f01..8a47514c3 100644 --- a/crates/vm/src/vm/mod.rs +++ b/crates/vm/src/vm/mod.rs @@ -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 diff --git a/crates/vm/src/vm/runtime.rs b/crates/vm/src/vm/runtime.rs index 0c34b7087..3f447fcbd 100644 --- a/crates/vm/src/vm/runtime.rs +++ b/crates/vm/src/vm/runtime.rs @@ -452,9 +452,15 @@ pub fn live_interpreter_states() -> Vec> { /// via [`owned_new_thread`], which only clones the shared interpreter fields /// (`sys`, builtins, `PyGlobalState`). #[cfg(feature = "threading")] -fn owned_interpreters() -> &'static Mutex> { +struct OwnedInterpreter { + root_id: i64, + interpreter: crate::Interpreter, +} + +#[cfg(feature = "threading")] +fn owned_interpreters() -> &'static Mutex> { use std::sync::OnceLock; - static OWNED: OnceLock>> = OnceLock::new(); + static OWNED: OnceLock>> = OnceLock::new(); OWNED.get_or_init(|| Mutex::new(HashMap::new())) } @@ -462,9 +468,16 @@ fn owned_interpreters() -> &'static Mutex> { #[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 } @@ -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 { - owned_interpreters().lock().remove(&id) + owned_interpreters() + .lock() + .remove(&id) + .map(|owned| owned.interpreter) } /// Create a thread-state VM for a runtime-owned interpreter. @@ -487,7 +503,7 @@ pub fn owned_new_thread(id: i64) -> Option Vec { 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 { + let mut ids: Vec = 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 diff --git a/extra_tests/snippets/stdlib_subinterpreters.py b/extra_tests/snippets/stdlib_subinterpreters.py new file mode 100644 index 000000000..d30ae9aaa --- /dev/null +++ b/extra_tests/snippets/stdlib_subinterpreters.py @@ -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