mirror of
https://github.com/RustPython/RustPython.git
synced 2026-10-28 18:59:04 +08:00
Fix post-merge subinterpreter lifecycle issues (#8632)
Preserve logical memoryview descriptors across interpreter transfers, make channel-end release idempotent, and scope runtime-owned interpreter cleanup to the creating top-level runtime. Re-read owned interpreters during shutdown so finalizers cannot leave newly-created children behind. Add regression coverage for sliced and cast memoryviews, repeated channel release, and multiple embedded runtimes. Assisted-by: Codex:GPT-5
This commit is contained in:
1 parent
516a116f9a
commit
6aac4653e6
6 files changed
+118
-9
No files matched your search
@@ -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
|
||||
|
||||
@@ -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());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -452,9 +452,15 @@ 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()))
|
||||
}
|
||||
|
||||
@@ -462,9 +468,16 @@ fn owned_interpreters() -> &'static Mutex<HashMap<i64, crate::Interpreter>> {
|
||||
#[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<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.
|
||||
@@ -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.
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
Reference in new issue
Block a user