From e5f332fda45b80888fac0cd565078dc78154a589 Mon Sep 17 00:00:00 2001 From: 1ndahous3 <9205230+1ndahous3@users.noreply.github.com> Date: Mon, 5 Oct 2026 03:43:04 +0300 Subject: [PATCH] Release interpreter channel payloads outside internal locks (#8964) Assisted-by: Codex:gpt-6 --- crates/vm/src/stdlib/_interpchannels.rs | 78 ++++++++++++------- .../snippets/stdlib_subinterpreters.py | 47 +++++++++++ 2 files changed, 98 insertions(+), 27 deletions(-) diff --git a/crates/vm/src/stdlib/_interpchannels.rs b/crates/vm/src/stdlib/_interpchannels.rs index d9fde089f..6878ac425 100644 --- a/crates/vm/src/stdlib/_interpchannels.rs +++ b/crates/vm/src/stdlib/_interpchannels.rs @@ -366,18 +366,23 @@ pub(crate) mod _interpchannels { /// `channel_destroy`: forget the channel entirely. fn channel_destroy(cid: i64) -> ChanResult, bool)>> { - let mut table = channels().lock(); - let entry = table.refs.remove(&cid).ok_or(ChanErr::NotFound)?; - Ok(match entry.chan { - Some(chan) => drain_queue(&mut chan.state.lock()), - None => Vec::new(), - }) + let entry = channels() + .lock() + .refs + .remove(&cid) + .ok_or(ChanErr::NotFound)?; + let items = match entry.chan { + Some(chan) => core::mem::take(&mut chan.state.lock().queue), + None => VecDeque::new(), + }; + Ok(retire_items(items)) } - fn drain_queue(state: &mut ChannelState) -> Vec<(Arc, bool)> { - state - .queue - .drain(..) + // Shared buffers can run owner callbacks when released. Call only after + // unlocking both the directory and channel state. + fn retire_items(items: VecDeque) -> Vec<(Arc, bool)> { + items + .into_iter() .filter_map(|item| item.waiting.map(|w| (w, false))) .collect() } @@ -394,10 +399,12 @@ pub(crate) mod _interpchannels { return Vec::new(); } let entry = table.refs.remove(&cid).expect("just looked it up"); - match entry.chan { - Some(chan) => drain_queue(&mut chan.state.lock()), - None => Vec::new(), - } + drop(table); + let items = match entry.chan { + Some(chan) => core::mem::take(&mut chan.state.lock().queue), + None => VecDeque::new(), + }; + retire_items(items) } /// `_channels_add_id_object`. @@ -456,12 +463,13 @@ pub(crate) mod _interpchannels { state.closing = true; return Ok(Vec::new()); } - let waiters = drain_queue(&mut state); + let items = core::mem::take(&mut state.queue); state.open = false; state.ends.release_all(); drop(state); entry.chan = None; - Ok(waiters) + drop(table); + Ok(retire_items(items)) } /// `channel_release`: close one or both ends for the current interpreter. @@ -486,6 +494,7 @@ pub(crate) mod _interpchannels { .filter_map(|r| r.chan.clone()) .collect(); let mut waiters = Vec::new(); + let mut dropped = Vec::new(); for chan in chans { let mut state = chan.state.lock(); let mut i = 0; @@ -500,17 +509,19 @@ pub(crate) mod _interpchannels { if let Some(w) = item.waiting { waiters.push((w, false)); } + dropped.push(item.data); continue; } // UNBOUND_ERROR / UNBOUND_REPLACE keep the slot but throw the // data away; the item is now "unbound". - item.data = None; + dropped.push(item.data.take()); i += 1; } state.ends.clear_interpreter(interpid); state.open = state.ends.is_open(); } release_waiters_detached(waiters); + drop(dropped); } /// `resolve_unboundop`. @@ -894,12 +905,14 @@ pub(crate) mod _interpchannels { let cid = parse_cid(parsed[0].as_deref().unwrap(), vm)?.0; let send = flag(parsed[1].as_deref(), vm)?; let chan = channels_lookup(cid).map_err(|e| e.into_py(cid, vm))?; + let interpreters = crate::vm::runtime::list_interpreters(); let state = chan.state.lock(); if send && state.closing { + drop(state); return Err(ChanErr::Closed.into_py(cid, vm)); } let mut out = Vec::new(); - for info in crate::vm::runtime::list_interpreters() { + for info in interpreters { if state.ends.find(info.id, send).is_some_and(|i| { if send { state.ends.send[i].1 @@ -907,10 +920,18 @@ pub(crate) mod _interpchannels { state.ends.recv[i].1 } }) { - out.push(vm.ctx.new_int(info.id).into()); + out.push(info.id); } } - Ok(vm.ctx.new_list(out).into()) + drop(state); + Ok(vm + .ctx + .new_list( + out.into_iter() + .map(|id| vm.ctx.new_int(id).into()) + .collect(), + ) + .into()) } /// `channel_send` up to its `closing` check, which precedes converting the @@ -982,15 +1003,18 @@ pub(crate) mod _interpchannels { if timed_out { // The send is failing now, so make sure the item won't be received. let mut state = chan.state.lock(); - if let Some(pos) = state.queue.iter().position(|it| { - it.waiting - .as_ref() - .is_some_and(|w| Arc::ptr_eq(w, &waiting)) - }) { - state.queue.remove(pos); - } + let removed = state + .queue + .iter() + .position(|it| { + it.waiting + .as_ref() + .is_some_and(|w| Arc::ptr_eq(w, &waiting)) + }) + .and_then(|pos| state.queue.remove(pos)); drop(state); finish_closing(cid); + drop(removed); if !waiting.state.lock().received { return Err(vm .new_os_subtype_error( diff --git a/extra_tests/snippets/stdlib_subinterpreters.py b/extra_tests/snippets/stdlib_subinterpreters.py index 3ca0be274..e2ad9d76b 100644 --- a/extra_tests/snippets/stdlib_subinterpreters.py +++ b/extra_tests/snippets/stdlib_subinterpreters.py @@ -117,3 +117,50 @@ if __name__ == '__main__': encoding="utf-8", ) subprocess.run([sys.executable, str(script)], check=True, timeout=30) + + +# Releasing queued buffers can run a finalizer which reenters the channel API. +subprocess.run( + [ + sys.executable, + "-c", + """ +import _interpchannels as channels +import _interpreters +import sys + +released = [] + +class Exporter(bytearray): + def __del__(self): + channels.list_all() + released.append(True) + +for operation in ("destroy", "close", "drop", "timeout"): + # CPython 3.14.6 also deadlocks when close/drop releases a reentrant exporter. + if sys.implementation.name == "cpython" and operation in ("close", "drop"): + continue + channel = channels.create(3) + released.clear() + if operation == "timeout": + try: + channels.send_buffer(channel, Exporter(b"data"), timeout=0) + except TimeoutError: + pass + else: + raise AssertionError("send must time out without a receiver") + else: + channels.send_buffer(channel, Exporter(b"data"), blocking=False) + assert not released + if operation == "destroy": + channels.destroy(channel) + elif operation == "close": + channels.close(channel, force=True) + else: + del channel + assert released == [True], operation +""", + ], + check=True, + timeout=30, +)