From d8a4366122de7789ff4f018be653d68a30fdac9c Mon Sep 17 00:00:00 2001 From: Jeong YunWon Date: Sat, 22 Aug 2026 08:55:11 +0900 Subject: [PATCH 01/12] Add _interpreters and _interpchannels modules Apply the patch baseline for the low-level subinterpreter modules: - crates/vm/src/stdlib/_interpreters.rs: create/destroy/list_all/get_main/ get_current/is_running/exec/call/run_string/run_func/set___main___attrs/ is_shareable/whence and the Interpreter*Error exceptions - crates/vm/src/stdlib/_interpchannels.rs: channel create/destroy/send/recv/ list_all/list_interpreters/release/close and the Channel*Error exceptions - crates/vm/src/vm/crossinterp.rs: cross-interpreter data protocol - crates/vm/src/vm/{interpreter,mod,runtime,vm_new}.rs: interpreter registry and lifecycle plumbing the modules need - Lib/concurrent/interpreters/: high-level PEP 734 package Assisted-by: Claude:claude-opus-5 --- .cspell.dict/cpython.txt | 5 + Lib/concurrent/interpreters/__init__.py | 6 + Lib/concurrent/interpreters/_crossinterp.py | 96 ++ crates/vm/src/builtins/memory.rs | 7 + crates/vm/src/lib.rs | 5 +- crates/vm/src/stdlib/_interpchannels.rs | 960 ++++++++++++++++++++ crates/vm/src/stdlib/_interpreters.rs | 697 ++++++++++++++ crates/vm/src/stdlib/_thread.rs | 5 +- crates/vm/src/stdlib/mod.rs | 4 + crates/vm/src/stdlib/posix.rs | 15 + crates/vm/src/vm/crossinterp.rs | 387 ++++++++ crates/vm/src/vm/interpreter.rs | 128 ++- crates/vm/src/vm/mod.rs | 38 +- crates/vm/src/vm/runtime.rs | 160 +++- crates/vm/src/vm/vm_new.rs | 20 + 15 files changed, 2484 insertions(+), 49 deletions(-) create mode 100644 Lib/concurrent/interpreters/__init__.py create mode 100644 Lib/concurrent/interpreters/_crossinterp.py create mode 100644 crates/vm/src/stdlib/_interpchannels.rs create mode 100644 crates/vm/src/stdlib/_interpreters.rs create mode 100644 crates/vm/src/vm/crossinterp.rs diff --git a/.cspell.dict/cpython.txt b/.cspell.dict/cpython.txt index f9aa440edee..77d028f5b64 100644 --- a/.cspell.dict/cpython.txt +++ b/.cspell.dict/cpython.txt @@ -35,6 +35,8 @@ ceval cfield cfws CFWS +chans +cidx CLASSDEREF classdict cmpop @@ -170,6 +172,7 @@ nseen NSIGNALS numer nvars +obmalloc opname opnames orelse @@ -206,6 +209,7 @@ PYTHONTRACEMALLOC PYTHONUTF8 pythonw PYTHREAD_NAME +recvd releasebuffer repr resinfo @@ -256,6 +260,7 @@ Typeparam typeparams typeslots unaryop +unboundop uncollectable Unhandle unparse diff --git a/Lib/concurrent/interpreters/__init__.py b/Lib/concurrent/interpreters/__init__.py new file mode 100644 index 00000000000..4c3c28395d6 --- /dev/null +++ b/Lib/concurrent/interpreters/__init__.py @@ -0,0 +1,6 @@ +"""Subinterpreters High Level Module.""" + +# The high-level `concurrent.interpreters` API is built on `_interpreters`. +# This package is introduced incrementally; `_crossinterp` is shared with +# the low-level `_interpchannels` tests. + diff --git a/Lib/concurrent/interpreters/_crossinterp.py b/Lib/concurrent/interpreters/_crossinterp.py new file mode 100644 index 00000000000..52bfa601a31 --- /dev/null +++ b/Lib/concurrent/interpreters/_crossinterp.py @@ -0,0 +1,96 @@ +"""Common code between queues and channels.""" + + +class ItemInterpreterDestroyed(Exception): + """Raised when trying to get an item whose interpreter was destroyed.""" + + +class classonly: + """A non-data descriptor that makes a value only visible on the class.""" + + def __init__(self, value): + self.value = value + self.getter = classmethod(value).__get__ + self.name = None + + def __set_name__(self, cls, name): + if self.name is not None: + raise TypeError('already used') + self.name = name + + def __get__(self, obj, cls): + if obj is not None: + raise AttributeError(self.name) + return self.getter(None, cls) + + +class UnboundItem: + """Represents a cross-interpreter item no longer bound to an interpreter.""" + + __slots__ = () + + @classonly + def singleton(cls, kind, module, name='UNBOUND'): + doc = cls.__doc__ + if doc: + doc = doc.replace( + 'cross-interpreter container', kind, + ).replace( + 'cross-interpreter', kind, + ) + subclass = type( + f'Unbound{kind.capitalize()}Item', + (cls,), + { + "_MODULE": module, + "_NAME": name, + "__doc__": doc, + }, + ) + return object.__new__(subclass) + + _MODULE = __name__ + _NAME = 'UNBOUND' + + def __new__(cls): + raise Exception(f'use {cls._MODULE}.{cls._NAME}') + + def __repr__(self): + return f'{self._MODULE}.{self._NAME}' + + +UNBOUND = object.__new__(UnboundItem) +UNBOUND_ERROR = object() +UNBOUND_REMOVE = object() + +_UNBOUND_CONSTANT_TO_FLAG = { + UNBOUND_REMOVE: 1, + UNBOUND_ERROR: 2, + UNBOUND: 3, +} +_UNBOUND_FLAG_TO_CONSTANT = {v: k + for k, v in _UNBOUND_CONSTANT_TO_FLAG.items()} + + +def serialize_unbound(unbound): + op = unbound + try: + flag = _UNBOUND_CONSTANT_TO_FLAG[op] + except KeyError: + raise NotImplementedError(f'unsupported unbound replacement op {op!r}') + return flag, + + +def resolve_unbound(flag, exctype_destroyed): + try: + op = _UNBOUND_FLAG_TO_CONSTANT[flag] + except KeyError: + raise NotImplementedError(f'unsupported unbound replacement op {flag!r}') + if op is UNBOUND_REMOVE: + raise NotImplementedError + elif op is UNBOUND_ERROR: + raise exctype_destroyed("item's original interpreter destroyed") + elif op is UNBOUND: + return UNBOUND + else: + raise NotImplementedError(repr(op)) diff --git a/crates/vm/src/builtins/memory.rs b/crates/vm/src/builtins/memory.rs index 86396825b0c..aa6111372ec 100644 --- a/crates/vm/src/builtins/memory.rs +++ b/crates/vm/src/builtins/memory.rs @@ -118,6 +118,13 @@ impl PyMemoryView { Self::from_object_with_flags(obj, BufferFlags::FULL_RO, vm) } + /// One share of the underlying buffer export. Used for cross-interpreter + /// `send_buffer` so the destination memoryview sees the same memory. + #[must_use] + pub fn clone_buffer(&self) -> PyBuffer { + self.buffer.clone() + } + // PyMemoryView_FromObjectAndFlags pub fn from_object_with_flags( obj: &PyObject, diff --git a/crates/vm/src/lib.rs b/crates/vm/src/lib.rs index a15e30c34a3..9a8c27197c0 100644 --- a/crates/vm/src/lib.rs +++ b/crates/vm/src/lib.rs @@ -111,8 +111,9 @@ pub use self::object::{ }; pub use self::vm::runtime; pub use self::vm::{ - Context, Interpreter, InterpreterBuilder, InterpreterInfo, InterpreterWhence, - MAIN_INTERPRETER_ID, Settings, VirtualMachine, + Context, InterpFeatureFlags, Interpreter, InterpreterBuilder, InterpreterConfig, + InterpreterGil, InterpreterInfo, InterpreterWhence, MAIN_INTERPRETER_ID, Settings, + VirtualMachine, }; pub use rustpython_common as common; diff --git a/crates/vm/src/stdlib/_interpchannels.rs b/crates/vm/src/stdlib/_interpchannels.rs new file mode 100644 index 00000000000..ad812be4283 --- /dev/null +++ b/crates/vm/src/stdlib/_interpchannels.rs @@ -0,0 +1,960 @@ +//! Low-level cross-interpreter channels (`_interpchannels`). +//! +//! Mirrors CPython `Modules/_interpchannelsmodule.c`. + +#[cfg_attr(not(feature = "threading"), allow(unused_imports))] +pub(crate) use _interpchannels::clear_interpreter; +pub(crate) use _interpchannels::module_def; +pub(crate) use _interpchannels::{channel_id_from_parts, channel_id_parts}; + +#[pymodule] +pub(crate) mod _interpchannels { + use crate::{ + AsObject, Py, PyObject, PyObjectRef, PyPayload, PyResult, VirtualMachine, + builtins::{PyFloat, PyInt}, + convert::ToPyObject, + function::FuncArgs, + protocol::PyNumberMethods, + types::{AsNumber, Comparable, Constructor, Hashable, PyComparisonOp, Representable}, + vm::crossinterp::{self, SharedValue, UNBOUND_REPLACE}, + }; + use alloc::{ + collections::{BTreeSet, VecDeque}, + sync::Arc, + }; + use core::time::Duration; + use parking_lot::{Condvar, Mutex}; + use std::{collections::HashMap, sync::OnceLock, time::Instant}; + + const CHANNEL_BOTH: i32 = 0; + const CHANNEL_SEND: i32 = 1; + const CHANNEL_RECV: i32 = 2; + + #[pyattr] + #[pyexception(name = "ChannelError", base = crate::exceptions::types::PyRuntimeError)] + #[derive(Debug)] + #[repr(transparent)] + pub(crate) struct PyChannelError(crate::exceptions::types::PyRuntimeError); + + #[pyexception] + impl PyChannelError {} + + #[pyattr] + #[pyexception(name = "ChannelNotFoundError", base = PyChannelError)] + #[derive(Debug)] + #[repr(transparent)] + pub(crate) struct PyChannelNotFoundError(PyChannelError); + + #[pyexception] + impl PyChannelNotFoundError {} + + #[pyattr] + #[pyexception(name = "ChannelClosedError", base = PyChannelError)] + #[derive(Debug)] + #[repr(transparent)] + pub(crate) struct PyChannelClosedError(PyChannelError); + + #[pyexception] + impl PyChannelClosedError {} + + #[pyattr] + #[pyexception(name = "ChannelEmptyError", base = PyChannelError)] + #[derive(Debug)] + #[repr(transparent)] + pub(crate) struct PyChannelEmptyError(PyChannelError); + + #[pyexception] + impl PyChannelEmptyError {} + + #[pyattr] + #[pyexception(name = "ChannelNotEmptyError", base = PyChannelError)] + #[derive(Debug)] + #[repr(transparent)] + pub(crate) struct PyChannelNotEmptyError(PyChannelError); + + #[pyexception] + impl PyChannelNotEmptyError {} + + fn not_found(vm: &VirtualMachine, cid: i64) -> crate::builtins::PyBaseExceptionRef { + vm.new_exception_msg( + PyChannelNotFoundError::class(&vm.ctx).to_owned(), + format!("channel {cid} not found").into(), + ) + } + + fn closed_err(vm: &VirtualMachine, cid: i64) -> crate::builtins::PyBaseExceptionRef { + vm.new_exception_msg( + PyChannelClosedError::class(&vm.ctx).to_owned(), + format!("channel {cid} is closed").into(), + ) + } + + fn empty_err(vm: &VirtualMachine, cid: i64) -> crate::builtins::PyBaseExceptionRef { + vm.new_exception_msg( + PyChannelEmptyError::class(&vm.ctx).to_owned(), + format!("channel {cid} is empty").into(), + ) + } + + fn not_empty_err(vm: &VirtualMachine) -> crate::builtins::PyBaseExceptionRef { + vm.new_exception_msg( + PyChannelNotEmptyError::class(&vm.ctx).to_owned(), + "channel is not empty".to_owned().into(), + ) + } + + struct ItemWaiter { + received: Mutex, + cond: Condvar, + closed: Mutex, + } + + struct ChannelItem { + #[allow(dead_code)] + interpid: i64, + payload: Option, + unboundop: i32, + waiter: Option>, + } + + struct ChannelInner { + items: VecDeque, + send_assoc: BTreeSet, + recv_assoc: BTreeSet, + send_closed: bool, + recv_closed: bool, + destroyed: bool, + /// `close()` hides the channel from `list_all`; `release()` does not. + hidden_from_list: bool, + unboundop: i32, + fallback: i32, + id_refs: isize, + } + + impl ChannelInner { + fn fully_closed(&self) -> bool { + self.destroyed || (self.send_closed && self.recv_closed) + } + } + + struct Channel { + #[allow(dead_code)] + id: i64, + inner: Mutex, + cond: Condvar, + } + + struct ChannelTable { + next_id: i64, + channels: HashMap>, + } + + fn table() -> &'static Mutex { + static TABLE: OnceLock> = OnceLock::new(); + TABLE.get_or_init(|| { + Mutex::new(ChannelTable { + next_id: 0, + channels: HashMap::new(), + }) + }) + } + + fn lookup(cid: i64) -> Option> { + table().lock().channels.get(&cid).cloned() + } + + fn create_channel(unboundop: i32, fallback: i32) -> i64 { + let mut t = table().lock(); + let id = t.next_id; + t.next_id += 1; + t.channels.insert( + id, + Arc::new(Channel { + id, + inner: Mutex::new(ChannelInner { + items: VecDeque::new(), + send_assoc: BTreeSet::new(), + recv_assoc: BTreeSet::new(), + send_closed: false, + recv_closed: false, + destroyed: false, + hidden_from_list: false, + unboundop, + fallback, + id_refs: 0, + }), + cond: Condvar::new(), + }), + ); + id + } + + fn destroy_channel(cid: i64) -> Result<(), ChannelOpErr> { + let mut t = table().lock(); + match t.channels.remove(&cid) { + Some(ch) => { + let mut inner = ch.inner.lock(); + inner.destroyed = true; + inner.send_closed = true; + inner.recv_closed = true; + for item in inner.items.drain(..) { + if let Some(w) = item.waiter { + *w.closed.lock() = true; + w.cond.notify_all(); + } + } + ch.cond.notify_all(); + Ok(()) + } + None => Err(ChannelOpErr::NotFound), + } + } + + enum ChannelOpErr { + NotFound, + } + + impl ChannelOpErr { + fn into_py(self, vm: &VirtualMachine, cid: i64) -> crate::builtins::PyBaseExceptionRef { + match self { + Self::NotFound => not_found(vm, cid), + } + } + } + + fn parse_cid(obj: &PyObject, vm: &VirtualMachine) -> PyResult<(i64, i32)> { + if let Some(cid) = obj.downcast_ref::() { + return Ok((cid.cid, cid.end)); + } + // Accept any indexable object (`__index__`), matching CPython. + let n = obj.try_index(vm)?; + let id = n.try_to_primitive::(vm)?; + if id < 0 { + return Err( + vm.new_value_error(format!("channel ID must be a non-negative int, got {id}")) + ); + } + Ok((id, CHANNEL_BOTH)) + } + + fn associate(inner: &mut ChannelInner, interpid: i64, send: bool) { + if send { + inner.send_assoc.insert(interpid); + } else { + inner.recv_assoc.insert(interpid); + } + } + + #[allow(dead_code)] + fn apply_unbound(item: &mut ChannelItem) { + match item.unboundop { + crossinterp::UNBOUND_REMOVE => { + item.payload = None; + } + _ => { + item.payload = None; + } + } + } + + #[cfg_attr(not(feature = "threading"), allow(dead_code))] + pub(crate) fn clear_interpreter(interpid: i64) { + let channels: Vec> = table().lock().channels.values().cloned().collect(); + for ch in channels { + let mut inner = ch.inner.lock(); + inner.send_assoc.remove(&interpid); + inner.recv_assoc.remove(&interpid); + let mut i = 0; + while i < inner.items.len() { + if inner.items[i].interpid == interpid { + if inner.items[i].unboundop == crossinterp::UNBOUND_REMOVE { + if let Some(w) = inner.items[i].waiter.take() { + *w.closed.lock() = true; + w.cond.notify_all(); + } + inner.items.remove(i); + continue; + } + apply_unbound(&mut inner.items[i]); + if let Some(w) = inner.items[i].waiter.take() { + *w.received.lock() = true; + w.cond.notify_all(); + } + } + i += 1; + } + if inner.send_assoc.is_empty() && inner.recv_assoc.is_empty() { + for item in inner.items.drain(..) { + if let Some(w) = item.waiter { + *w.closed.lock() = true; + w.cond.notify_all(); + } + } + inner.send_closed = true; + inner.recv_closed = true; + ch.cond.notify_all(); + } + } + } + + fn resolve_unboundop(arg: i32, default: i32) -> i32 { + if arg < 0 { default } else { arg } + } + + #[pyattr] + #[pyclass(name = "ChannelID", module = "_interpchannels")] + #[derive(Debug, PyPayload)] + pub(crate) struct ChannelID { + cid: i64, + end: i32, + resolve: bool, + } + + impl Drop for ChannelID { + fn drop(&mut self) { + if let Some(ch) = lookup(self.cid) { + let remove = { + let mut inner = ch.inner.lock(); + inner.id_refs = inner.id_refs.saturating_sub(1); + inner.id_refs <= 0 && (inner.hidden_from_list || inner.destroyed) + }; + if remove { + table().lock().channels.remove(&self.cid); + } + } + } + } + + impl Constructor for ChannelID { + type Args = FuncArgs; + + fn py_new( + _cls: &crate::Py, + args: Self::Args, + vm: &VirtualMachine, + ) -> PyResult { + channel_id_new(args, vm) + } + } + + impl Representable for ChannelID { + #[inline] + fn repr_str(zelf: &Py, _vm: &VirtualMachine) -> PyResult { + Ok(match zelf.end { + CHANNEL_SEND => format!("ChannelID({}, send=True)", zelf.cid), + CHANNEL_RECV => format!("ChannelID({}, recv=True)", zelf.cid), + _ => format!("ChannelID({})", zelf.cid), + }) + } + } + + impl Hashable for ChannelID { + #[inline] + fn hash(zelf: &Py, vm: &VirtualMachine) -> PyResult { + vm.ctx.new_int(zelf.cid).as_object().hash(vm) + } + } + + impl Comparable for ChannelID { + fn cmp( + zelf: &Py, + other: &PyObject, + op: PyComparisonOp, + vm: &VirtualMachine, + ) -> PyResult { + use crate::function::PyComparisonValue; + if !matches!(op, PyComparisonOp::Eq | PyComparisonOp::Ne) { + return Ok(PyComparisonValue::NotImplemented); + } + let equal = if let Some(o) = other.downcast_ref::() { + zelf.cid == o.cid && zelf.end == o.end + } else if let Some(n) = other.downcast_ref::() { + n.try_to_primitive::(vm).is_ok_and(|v| v == zelf.cid) + } else if let Some(f) = other.downcast_ref::() { + let fv = f.to_f64(); + fv.is_finite() && fv == zelf.cid as f64 && (fv as i64) == zelf.cid + } else { + return Ok(PyComparisonValue::NotImplemented); + }; + Ok(PyComparisonValue::Implemented( + if op == PyComparisonOp::Eq { + equal + } else { + !equal + }, + )) + } + } + + impl AsNumber for ChannelID { + fn as_number() -> &'static PyNumberMethods { + static METHODS: PyNumberMethods = PyNumberMethods { + int: Some(|obj, vm| { + let zelf = obj.downcast_ref::().unwrap(); + Ok(vm.ctx.new_int(zelf.cid).into()) + }), + index: Some(|obj, vm| { + let zelf = obj.downcast_ref::().unwrap(); + Ok(vm.ctx.new_int(zelf.cid).into()) + }), + ..PyNumberMethods::NOT_IMPLEMENTED + }; + &METHODS + } + } + + #[pyclass( + with(Constructor, Representable, Hashable, Comparable, AsNumber), + flags(BASETYPE) + )] + impl ChannelID { + #[pygetset] + fn end(&self) -> String { + match self.end { + CHANNEL_SEND => "send".to_owned(), + CHANNEL_RECV => "recv".to_owned(), + _ => "both".to_owned(), + } + } + + #[pygetset(name = "send")] + fn send_end(&self, vm: &VirtualMachine) -> PyResult { + channel_id_from_parts(self.cid, CHANNEL_SEND, true, self.resolve, vm) + } + + #[pygetset(name = "recv")] + fn recv_end(&self, vm: &VirtualMachine) -> PyResult { + channel_id_from_parts(self.cid, CHANNEL_RECV, true, self.resolve, vm) + } + + #[pymethod] + fn __str__(&self) -> String { + self.cid.to_string() + } + } + + fn channel_id_new(args: FuncArgs, vm: &VirtualMachine) -> PyResult { + let id_obj = args + .args + .first() + .ok_or_else(|| vm.new_type_error("ChannelID() missing required argument: 'id'"))?; + let (cid, mut end) = parse_cid(id_obj, vm)?; + let send = args + .kwargs + .get("send") + .map(|o| o.clone().is_true(vm)) + .transpose()?; + let recv = args + .kwargs + .get("recv") + .map(|o| o.clone().is_true(vm)) + .transpose()?; + let force = args + .kwargs + .get("force") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(false); + let resolve = args + .kwargs + .get("_resolve") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(false); + match (send, recv) { + (Some(false), Some(false)) => { + return Err(vm.new_value_error("'send' and 'recv' cannot both be False")); + } + (Some(true), Some(false) | None) => end = CHANNEL_SEND, + (Some(false) | None, Some(true)) => end = CHANNEL_RECV, + (Some(true), Some(true)) => end = CHANNEL_BOTH, + _ => {} + } + if !force && lookup(cid).is_none() { + return Err(not_found(vm, cid)); + } + if let Some(ch) = lookup(cid) { + ch.inner.lock().id_refs += 1; + } + Ok(ChannelID { cid, end, resolve }) + } + + pub(crate) fn channel_id_from_parts( + cid: i64, + end: i32, + force: bool, + resolve: bool, + vm: &VirtualMachine, + ) -> PyResult { + if !force && lookup(cid).is_none() { + return Err(not_found(vm, cid)); + } + if let Some(ch) = lookup(cid) { + ch.inner.lock().id_refs += 1; + } + Ok(ChannelID { cid, end, resolve }.to_pyobject(vm)) + } + + pub(crate) fn channel_id_parts(obj: &PyObject) -> Option<(i64, i32)> { + obj.downcast_ref::().map(|c| (c.cid, c.end)) + } + + #[pyfunction] + fn create(args: FuncArgs, vm: &VirtualMachine) -> PyResult { + let unbound = match args.args.first() { + Some(o) => o.try_index(vm)?.try_to_primitive::(vm)?, + None => UNBOUND_REPLACE, + }; + let fallback = match args.kwargs.get("fallback") { + Some(o) => o.try_index(vm)?.try_to_primitive::(vm)?, + None => -1, + }; + let cid = create_channel(unbound, fallback); + channel_id_from_parts(cid, CHANNEL_BOTH, false, false, vm) + } + + #[pyfunction] + fn destroy(cid: PyObjectRef, vm: &VirtualMachine) -> PyResult<()> { + let (cid, _) = parse_cid(&cid, vm)?; + destroy_channel(cid).map_err(|e| e.into_py(vm, cid)) + } + + #[pyfunction] + fn list_all(vm: &VirtualMachine) -> PyResult { + let chans: Vec<(i64, i32, i32)> = { + let t = table().lock(); + t.channels + .iter() + .filter_map(|(&id, ch)| { + let inner = ch.inner.lock(); + if inner.hidden_from_list || inner.destroyed { + None + } else { + Some((id, inner.unboundop, inner.fallback)) + } + }) + .collect() + }; + let mut items = Vec::new(); + for (id, unbound, fallback) in chans { + let cid = channel_id_from_parts(id, CHANNEL_BOTH, false, false, vm)?; + items.push( + vm.ctx + .new_tuple(vec![ + cid, + vm.ctx.new_int(unbound).into(), + vm.ctx.new_int(fallback).into(), + ]) + .into(), + ); + } + Ok(vm.ctx.new_list(items).into()) + } + + #[pyfunction] + fn list_interpreters(args: FuncArgs, vm: &VirtualMachine) -> PyResult { + let cid_obj = args + .args + .first() + .ok_or_else(|| vm.new_type_error("list_interpreters() missing argument 1"))?; + let (cid, _) = parse_cid(cid_obj, vm)?; + let send = args + .kwargs + .get("send") + .ok_or_else(|| vm.new_type_error("list_interpreters() missing keyword 'send'"))? + .clone() + .is_true(vm)?; + let ch = lookup(cid).ok_or_else(|| not_found(vm, cid))?; + let inner = ch.inner.lock(); + if inner.destroyed || inner.fully_closed() || (send && inner.send_closed) { + return Err(closed_err(vm, cid)); + } + if !send && inner.recv_closed && inner.items.is_empty() { + return Err(closed_err(vm, cid)); + } + let ids = if send { + &inner.send_assoc + } else { + &inner.recv_assoc + }; + let mut out = Vec::new(); + for &id in ids { + if crate::vm::runtime::lookup_interpreter(id).is_some() { + out.push(vm.ctx.new_int(id).into()); + } + } + Ok(vm.ctx.new_list(out).into()) + } + + fn do_send( + cid: i64, + value: SharedValue, + unboundop: i32, + blocking: bool, + timeout: Option, + vm: &VirtualMachine, + ) -> PyResult<()> { + let ch = lookup(cid).ok_or_else(|| not_found(vm, cid))?; + let waiter = if blocking { + Some(Arc::new(ItemWaiter { + received: Mutex::new(false), + cond: Condvar::new(), + closed: Mutex::new(false), + })) + } else { + None + }; + { + let mut inner = ch.inner.lock(); + if inner.destroyed { + return Err(not_found(vm, cid)); + } + if inner.send_closed || inner.fully_closed() { + return Err(closed_err(vm, cid)); + } + associate(&mut inner, vm.state.interpreter_id, true); + inner.items.push_back(ChannelItem { + interpid: vm.state.interpreter_id, + payload: Some(value), + unboundop, + waiter: waiter.clone(), + }); + ch.cond.notify_all(); + } + if let Some(w) = waiter { + let deadline = timeout.map(|t| Instant::now() + t); + vm.allow_threads(|| { + let mut recvd = w.received.lock(); + loop { + if *recvd { + return Ok(()); + } + if *w.closed.lock() { + return Err(()); + } + if let Some(dl) = deadline { + let now = Instant::now(); + if now >= dl { + return Err(()); + } + let leftover = dl - now; + if w.cond.wait_for(&mut recvd, leftover).timed_out() { + return Err(()); + } + } else { + w.cond.wait(&mut recvd); + } + } + }) + .map_err(|()| { + // timed out or closed: remove item if still queued + let mut inner = ch.inner.lock(); + if let Some(pos) = inner + .items + .iter() + .position(|it| it.waiter.as_ref().is_some_and(|iw| Arc::ptr_eq(iw, &w))) + { + inner.items.remove(pos); + if timeout.is_some() && !*w.closed.lock() { + return vm + .new_os_subtype_error( + vm.ctx.exceptions.timeout_error.to_owned(), + None, + "timed out", + ) + .upcast(); + } + return closed_err(vm, cid); + } + if *w.closed.lock() { + closed_err(vm, cid) + } else { + vm.new_os_subtype_error( + vm.ctx.exceptions.timeout_error.to_owned(), + None, + "timed out", + ) + .upcast() + } + })?; + } + Ok(()) + } + + #[pyfunction] + fn send(args: FuncArgs, vm: &VirtualMachine) -> PyResult<()> { + let cid_obj = args + .args + .first() + .ok_or_else(|| vm.new_type_error("send() missing argument 1"))?; + let obj = args + .args + .get(1) + .ok_or_else(|| vm.new_type_error("send() missing argument 2"))?; + let (cid, _) = parse_cid(cid_obj, vm)?; + let unboundarg = args + .args + .get(2) + .or_else(|| args.kwargs.get("unboundop")) + .map(|o| o.try_index(vm).and_then(|i| i.try_to_primitive::(vm))) + .transpose()? + .unwrap_or(-1); + let blocking = args + .kwargs + .get("blocking") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(true); + let timeout = parse_timeout(&args, blocking, vm)?; + let ch = lookup(cid).ok_or_else(|| not_found(vm, cid))?; + let default_unbound = ch.inner.lock().unboundop; + let unboundop = resolve_unboundop(unboundarg, default_unbound); + let value = SharedValue::from_object(obj, vm)?; + do_send(cid, value, unboundop, blocking, timeout, vm) + } + + #[pyfunction] + fn send_buffer(args: FuncArgs, vm: &VirtualMachine) -> PyResult<()> { + let cid_obj = args + .args + .first() + .ok_or_else(|| vm.new_type_error("send_buffer() missing argument 1"))?; + let obj = args + .args + .get(1) + .ok_or_else(|| vm.new_type_error("send_buffer() missing argument 2"))?; + let (cid, _) = parse_cid(cid_obj, vm)?; + let unboundarg = args + .args + .get(2) + .or_else(|| args.kwargs.get("unboundop")) + .map(|o| o.try_index(vm).and_then(|i| i.try_to_primitive::(vm))) + .transpose()? + .unwrap_or(-1); + let blocking = args + .kwargs + .get("blocking") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(true); + let timeout = parse_timeout(&args, blocking, vm)?; + let ch = lookup(cid).ok_or_else(|| not_found(vm, cid))?; + let default_unbound = ch.inner.lock().unboundop; + let unboundop = resolve_unboundop(unboundarg, default_unbound); + let value = SharedValue::from_buffer_object(obj, vm)?; + do_send(cid, value, unboundop, blocking, timeout, vm) + } + + fn parse_timeout( + args: &FuncArgs, + blocking: bool, + vm: &VirtualMachine, + ) -> PyResult> { + let Some(t) = args.kwargs.get("timeout") else { + return Ok(None); + }; + if vm.is_none(t) { + return Ok(None); + } + if !blocking { + return Err(vm.new_value_error("can't specify a timeout for a non-blocking call")); + } + let secs = t.try_float(vm)?.to_f64(); + if secs < 0.0 { + return Err(vm.new_value_error("timeout value must be non-negative")); + } + Ok(Some(Duration::from_secs_f64(secs))) + } + + #[pyfunction] + fn recv(args: FuncArgs, vm: &VirtualMachine) -> PyResult { + let cid_obj = args + .args + .first() + .ok_or_else(|| vm.new_type_error("recv() missing argument 1"))?; + let (cid, _) = parse_cid(cid_obj, vm)?; + let default = args.args.get(1).cloned(); + let ch = lookup(cid).ok_or_else(|| not_found(vm, cid))?; + let mut inner = ch.inner.lock(); + if inner.destroyed { + return Err(not_found(vm, cid)); + } + associate(&mut inner, vm.state.interpreter_id, false); + if inner.items.is_empty() { + if inner.recv_closed + || inner.fully_closed() + || inner.send_closed && inner.send_assoc.is_empty() + { + return Err(closed_err(vm, cid)); + } + if let Some(d) = default { + return Ok(vm.ctx.new_tuple(vec![d, vm.ctx.none()]).into()); + } + return Err(empty_err(vm, cid)); + } + let item = inner.items.pop_front().unwrap(); + if inner.items.is_empty() && inner.send_closed { + inner.recv_closed = true; + } + drop(inner); + if let Some(w) = &item.waiter { + *w.received.lock() = true; + w.cond.notify_all(); + } + match item.payload { + Some(val) => { + let obj = val.into_object(vm)?; + Ok(vm.ctx.new_tuple(vec![obj, vm.ctx.none()]).into()) + } + None => Ok(vm + .ctx + .new_tuple(vec![vm.ctx.none(), vm.ctx.new_int(item.unboundop).into()]) + .into()), + } + } + + #[pyfunction] + fn close(args: FuncArgs, vm: &VirtualMachine) -> PyResult<()> { + let cid_obj = args + .args + .first() + .ok_or_else(|| vm.new_type_error("close() missing argument 1"))?; + let (cid, _) = parse_cid(cid_obj, vm)?; + let send = args + .kwargs + .get("send") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(false); + let recv = args + .kwargs + .get("recv") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(false); + let force = args + .kwargs + .get("force") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(false); + let ch = lookup(cid).ok_or_else(|| not_found(vm, cid))?; + let mut inner = ch.inner.lock(); + if inner.destroyed { + return Err(not_found(vm, cid)); + } + if inner.fully_closed() { + return Err(closed_err(vm, cid)); + } + let empty = inner.items.is_empty(); + if empty { + inner.send_closed = true; + inner.recv_closed = true; + inner.hidden_from_list = true; + ch.cond.notify_all(); + return Ok(()); + } + if force { + for item in inner.items.drain(..) { + if let Some(w) = item.waiter { + *w.closed.lock() = true; + w.cond.notify_all(); + } + } + inner.send_closed = true; + inner.recv_closed = true; + inner.hidden_from_list = true; + ch.cond.notify_all(); + return Ok(()); + } + // not empty, not force + if recv || (!send && !recv) { + return Err(not_empty_err(vm)); + } + // send only + inner.send_closed = true; + ch.cond.notify_all(); + Ok(()) + } + + #[pyfunction] + fn release(args: FuncArgs, vm: &VirtualMachine) -> PyResult<()> { + let cid_obj = args + .args + .first() + .ok_or_else(|| vm.new_type_error("release() missing argument 1"))?; + let (cid, _) = parse_cid(cid_obj, vm)?; + let mut send = args + .kwargs + .get("send") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(false); + let mut recv = args + .kwargs + .get("recv") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(false); + if !send && !recv { + send = true; + recv = true; + } + let ch = lookup(cid).ok_or_else(|| not_found(vm, cid))?; + let mut inner = ch.inner.lock(); + if inner.destroyed { + return Err(not_found(vm, cid)); + } + if inner.fully_closed() { + return Err(closed_err(vm, cid)); + } + let interpid = vm.state.interpreter_id; + if send { + inner.send_assoc.remove(&interpid); + } + if recv { + inner.recv_assoc.remove(&interpid); + } + if inner.send_assoc.is_empty() && inner.recv_assoc.is_empty() { + inner.send_closed = true; + inner.recv_closed = true; + for item in inner.items.drain(..) { + if let Some(w) = item.waiter { + *w.closed.lock() = true; + w.cond.notify_all(); + } + } + ch.cond.notify_all(); + } + Ok(()) + } + + #[pyfunction] + fn get_count(cid: PyObjectRef, vm: &VirtualMachine) -> PyResult { + let (cid, _) = parse_cid(&cid, vm)?; + let ch = lookup(cid).ok_or_else(|| not_found(vm, cid))?; + Ok(ch.inner.lock().items.len()) + } + + #[pyfunction] + fn get_channel_defaults(cid: PyObjectRef, vm: &VirtualMachine) -> PyResult { + let (cid, _) = parse_cid(&cid, vm)?; + let ch = lookup(cid).ok_or_else(|| not_found(vm, cid))?; + let inner = ch.inner.lock(); + Ok(vm + .ctx + .new_tuple(vec![ + vm.ctx.new_int(inner.unboundop).into(), + vm.ctx.new_int(inner.fallback).into(), + ]) + .into()) + } + + #[pyfunction] + fn _channel_id(args: FuncArgs, vm: &VirtualMachine) -> PyResult { + Ok(channel_id_new(args, vm)?.to_pyobject(vm)) + } + + #[pyfunction] + fn _register_end_types(_send: PyObjectRef, _recv: PyObjectRef) {} +} diff --git a/crates/vm/src/stdlib/_interpreters.rs b/crates/vm/src/stdlib/_interpreters.rs new file mode 100644 index 00000000000..c8ba324cf4a --- /dev/null +++ b/crates/vm/src/stdlib/_interpreters.rs @@ -0,0 +1,697 @@ +//! Low-level multiple-interpreter primitives (`_interpreters`). +//! +//! Mirrors CPython `Modules/_interpretersmodule.c`. + +pub(crate) use _interpreters::module_def; +#[cfg_attr(not(feature = "threading"), allow(unused_imports))] +pub(crate) use _interpreters::{interpreter_error, interpreter_not_found, not_shareable_error}; + +#[pymodule] +pub(crate) mod _interpreters { + use crate::{ + AsObject, PyObject, PyObjectRef, PyPayload, PyRef, PyResult, VirtualMachine, + builtins::{PyCode, PyDict, PyException, PyFunction, PyInt, PyStr}, + function::{FuncArgs, OptionalArg}, + types::Constructor, + vm::{ + InterpreterConfig, InterpreterWhence, + crossinterp::{self, ExcInfo, SharedValue}, + runtime, + }, + }; + use core::sync::atomic::Ordering; + + #[pyattr] + pub(crate) const WHENCE_UNKNOWN: i32 = InterpreterWhence::Unknown as i32; + #[pyattr] + pub(crate) const WHENCE_RUNTIME: i32 = InterpreterWhence::Runtime as i32; + #[pyattr] + pub(crate) const WHENCE_LEGACY_CAPI: i32 = InterpreterWhence::LegacyCapi as i32; + #[pyattr] + pub(crate) const WHENCE_CAPI: i32 = InterpreterWhence::Capi as i32; + #[pyattr] + pub(crate) const WHENCE_XI: i32 = InterpreterWhence::Xi as i32; + #[pyattr] + pub(crate) const WHENCE_STDLIB: i32 = InterpreterWhence::Stdlib as i32; + + #[pyattr] + #[pyexception(name = "InterpreterError", base = PyException)] + #[derive(Debug)] + #[repr(transparent)] + pub(crate) struct PyInterpreterError(PyException); + + #[pyexception] + impl PyInterpreterError {} + + #[pyattr] + #[pyexception(name = "InterpreterNotFoundError", base = PyInterpreterError)] + #[derive(Debug)] + #[repr(transparent)] + pub(crate) struct PyInterpreterNotFoundError(PyInterpreterError); + + #[pyexception] + impl PyInterpreterNotFoundError {} + + #[pyattr] + #[pyexception(name = "NotShareableError", base = crate::exceptions::types::PyTypeError)] + #[derive(Debug)] + #[repr(transparent)] + pub(crate) struct PyNotShareableError(crate::exceptions::types::PyTypeError); + + #[pyexception] + impl PyNotShareableError {} + + pub(crate) fn interpreter_error( + vm: &VirtualMachine, + msg: impl Into, + ) -> crate::builtins::PyBaseExceptionRef { + vm.new_exception_msg( + PyInterpreterError::class(&vm.ctx).to_owned(), + msg.into().into(), + ) + } + + pub(crate) fn interpreter_not_found( + vm: &VirtualMachine, + id: i64, + ) -> crate::builtins::PyBaseExceptionRef { + vm.new_exception_msg( + PyInterpreterNotFoundError::class(&vm.ctx).to_owned(), + format!("interpreter {id} not found").into(), + ) + } + + pub(crate) fn not_shareable_error( + vm: &VirtualMachine, + msg: impl Into, + ) -> crate::builtins::PyBaseExceptionRef { + vm.new_exception_msg( + PyNotShareableError::class(&vm.ctx).to_owned(), + msg.into().into(), + ) + } + + fn parse_id(obj: &PyObject, vm: &VirtualMachine) -> PyResult { + let Some(n) = obj.downcast_ref::() else { + return Err(vm.new_type_error(format!( + "interpreter ID must be an int, got {}", + obj.class() + ))); + }; + let id = n.try_to_primitive::(vm)?; + if id < 0 { + return Err(vm.new_value_error(format!( + "interpreter ID must be a non-negative int, got {id}" + ))); + } + Ok(id) + } + + fn resolve_interp( + id: i64, + restricted: bool, + reqready: bool, + op: &str, + vm: &VirtualMachine, + ) -> PyResult { + let info = runtime::list_interpreters() + .into_iter() + .find(|i| i.id == id) + .ok_or_else(|| interpreter_not_found(vm, id))?; + if reqready { + let state = + runtime::lookup_interpreter(id).ok_or_else(|| interpreter_not_found(vm, id))?; + if !state.ready.load(Ordering::Acquire) { + return Err(interpreter_error( + vm, + format!("cannot {op} interpreter {id} (not ready)"), + )); + } + } + if restricted && info.whence != InterpreterWhence::Stdlib { + return Err(interpreter_error( + vm, + format!("cannot {op} unrecognized interpreter {id}"), + )); + } + Ok(info) + } + + fn require_dict<'a>( + obj: &'a PyObject, + func: &str, + pos: i32, + vm: &'a VirtualMachine, + ) -> PyResult<&'a crate::Py> { + obj.downcast_ref::().ok_or_else(|| { + vm.new_type_error(format!( + "{func} argument {pos} must be dict, not {}", + obj.class() + )) + }) + // downcast_ref already returns Option<&Py> + } + + fn parse_config(obj: Option<&PyObject>, vm: &VirtualMachine) -> PyResult { + let Some(obj) = obj else { + return Ok(InterpreterConfig::ISOLATED); + }; + if vm.is_none(obj) { + return Ok(InterpreterConfig::ISOLATED); + } + if let Some(s) = obj.downcast_ref::() { + let name = s.to_string(); + return InterpreterConfig::named(&name) + .ok_or_else(|| vm.new_value_error(format!("unsupported config name '{name}'"))); + } + // Namespace / object with attributes. + let mut cfg = InterpreterConfig::ISOLATED; + let get_bool = |name: &'static str| -> PyResult> { + match obj.get_attr(name, vm) { + Ok(v) => Ok(Some(v.is_true(vm)?)), + Err(_) => { + let _ = vm.pop_exception(); + Ok(None) + } + } + }; + if let Some(v) = get_bool("use_main_obmalloc")? { + cfg.use_main_obmalloc = v; + } + if let Some(v) = get_bool("allow_fork")? { + cfg.allow_fork = v; + } + if let Some(v) = get_bool("allow_exec")? { + cfg.allow_exec = v; + } + if let Some(v) = get_bool("allow_threads")? { + cfg.allow_threads = v; + } + if let Some(v) = get_bool("allow_daemon_threads")? { + cfg.allow_daemon_threads = v; + } + if let Some(v) = get_bool("check_multi_interp_extensions")? { + cfg.check_multi_interp_extensions = v; + } + Ok(cfg) + } + + fn config_namespace(cfg: InterpreterConfig, vm: &VirtualMachine) -> PyObjectRef { + crate::py_namespace!(vm, { + "use_main_obmalloc" => vm.ctx.new_bool(cfg.use_main_obmalloc), + "allow_fork" => vm.ctx.new_bool(cfg.allow_fork), + "allow_exec" => vm.ctx.new_bool(cfg.allow_exec), + "allow_threads" => vm.ctx.new_bool(cfg.allow_threads), + "allow_daemon_threads" => vm.ctx.new_bool(cfg.allow_daemon_threads), + "check_multi_interp_extensions" => vm.ctx.new_bool(cfg.check_multi_interp_extensions), + "gil" => vm.ctx.new_str(cfg.gil.as_str()), + }) + .into() + } + + fn summary(id: i64, whence: InterpreterWhence, vm: &VirtualMachine) -> PyObjectRef { + vm.ctx + .new_tuple(vec![ + vm.ctx.new_int(id).into(), + vm.ctx.new_int(whence.as_i32()).into(), + ]) + .into() + } + + #[pyfunction] + fn create(args: FuncArgs, vm: &VirtualMachine) -> PyResult { + let config_obj = args.args.first().cloned(); + let reqrefs = args + .kwargs + .get("reqrefs") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(false); + let config = parse_config(config_obj.as_deref(), vm)?; + #[cfg(feature = "threading")] + { + let interp = crate::Interpreter::create_subinterpreter_from_vm(vm, config); + if reqrefs { + interp + .global_state + .require_idref + .store(true, Ordering::Release); + interp + .global_state + .id_refcount + .fetch_add(1, Ordering::AcqRel); + } + let id = runtime::store_owned_interpreter(interp); + return Ok(vm.ctx.new_int(id).into()); + } + #[cfg(not(feature = "threading"))] + { + let _ = (config, reqrefs); + Err(vm.new_runtime_error("isolated interpreters require threading")) + } + } + + #[pyfunction] + fn destroy(args: FuncArgs, vm: &VirtualMachine) -> PyResult<()> { + let id_obj = args + .args + .first() + .ok_or_else(|| vm.new_type_error("destroy() missing required argument: 'id'"))?; + let restricted = args + .kwargs + .get("restrict") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(false); + let id = parse_id(id_obj, vm)?; + resolve_interp(id, restricted, false, "destroy", vm)?; + if id == vm.state.interpreter_id { + return Err(interpreter_error( + vm, + "cannot destroy the current interpreter", + )); + } + if crossinterp::is_running(id) + && !runtime::lookup_interpreter(id).is_some_and(|s| s.is_main) + { + return Err(interpreter_error(vm, "interpreter running")); + } + #[cfg(feature = "threading")] + { + if runtime::is_owned_interpreter(id) { + runtime::destroy_owned_interpreter(id) + .ok_or_else(|| interpreter_not_found(vm, id))?; + return Ok(()); + } + } + Err(interpreter_not_found(vm, id)) + } + + #[pyfunction] + fn list_all(args: FuncArgs, vm: &VirtualMachine) -> PyResult { + let reqready = args + .kwargs + .get("require_ready") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(false); + let mut items = Vec::new(); + for info in runtime::list_interpreters() { + if reqready { + let ready = runtime::lookup_interpreter(info.id) + .is_some_and(|s| s.ready.load(Ordering::Acquire)); + if !ready { + continue; + } + } + items.push(summary(info.id, info.whence, vm)); + } + Ok(vm.ctx.new_list(items).into()) + } + + #[pyfunction] + fn get_current(vm: &VirtualMachine) -> PyObjectRef { + summary(vm.state.interpreter_id, vm.state.whence, vm) + } + + #[pyfunction] + fn get_main(vm: &VirtualMachine) -> PyObjectRef { + let id = runtime::main_interpreter_id().unwrap_or(0); + summary(id, InterpreterWhence::Runtime, vm) + } + + #[pyfunction] + fn is_running(args: FuncArgs, vm: &VirtualMachine) -> PyResult { + let id_obj = args + .args + .first() + .ok_or_else(|| vm.new_type_error("is_running() missing required argument: 'id'"))?; + let restricted = args + .kwargs + .get("restrict") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(false); + let id = parse_id(id_obj, vm)?; + resolve_interp(id, restricted, true, "check if running for", vm)?; + Ok(vm.ctx.new_bool(crossinterp::is_running(id)).into()) + } + + #[pyfunction] + fn whence(id: PyObjectRef, vm: &VirtualMachine) -> PyResult { + let id = parse_id(&id, vm)?; + let info = runtime::list_interpreters() + .into_iter() + .find(|i| i.id == id) + .ok_or_else(|| interpreter_not_found(vm, id))?; + Ok(vm.ctx.new_int(info.whence.as_i32()).into()) + } + + #[pyfunction] + fn get_config(args: FuncArgs, vm: &VirtualMachine) -> PyResult { + let id_obj = args.args.first(); + let restricted = args + .kwargs + .get("restrict") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(false); + let id = match id_obj { + Some(o) if vm.is_none(o) => vm.state.interpreter_id, + Some(o) => parse_id(o, vm)?, + None => vm.state.interpreter_id, + }; + resolve_interp(id, restricted, false, "get the config of", vm)?; + let state = runtime::lookup_interpreter(id).ok_or_else(|| interpreter_not_found(vm, id))?; + let flags = state.feature_flags; + let cfg = InterpreterConfig { + use_main_obmalloc: false, + allow_fork: flags.allow_fork, + allow_exec: flags.allow_exec, + allow_threads: flags.allow_threads, + allow_daemon_threads: flags.allow_daemon_threads, + check_multi_interp_extensions: flags.check_multi_interp_extensions, + gil: if flags.allow_fork { + crate::vm::InterpreterGil::Shared + } else { + crate::vm::InterpreterGil::Own + }, + }; + Ok(config_namespace(cfg, vm)) + } + + #[pyfunction] + fn new_config(args: FuncArgs, vm: &VirtualMachine) -> PyResult { + let name = match args.args.first() { + Some(o) => o + .downcast_ref::() + .map_or_else(|| "isolated".to_owned(), |s| s.to_string()), + None => "isolated".to_owned(), + }; + let mut cfg = InterpreterConfig::named(&name) + .ok_or_else(|| vm.new_value_error(format!("unsupported config name '{name}'")))?; + if let Some(v) = args.kwargs.get("allow_fork") { + cfg.allow_fork = v.clone().is_true(vm)?; + } + if let Some(v) = args.kwargs.get("allow_exec") { + cfg.allow_exec = v.clone().is_true(vm)?; + } + if let Some(v) = args.kwargs.get("allow_threads") { + cfg.allow_threads = v.clone().is_true(vm)?; + } + if let Some(v) = args.kwargs.get("allow_daemon_threads") { + cfg.allow_daemon_threads = v.clone().is_true(vm)?; + } + if let Some(v) = args.kwargs.get("use_main_obmalloc") { + cfg.use_main_obmalloc = v.clone().is_true(vm)?; + } + if let Some(v) = args.kwargs.get("check_multi_interp_extensions") { + cfg.check_multi_interp_extensions = v.clone().is_true(vm)?; + } + Ok(config_namespace(cfg, vm)) + } + + #[pyfunction] + fn is_shareable(obj: PyObjectRef, vm: &VirtualMachine) -> bool { + crossinterp::is_shareable(&obj, vm) + } + + fn run_code_in( + id: i64, + code: PyRef, + shared: Option<&PyObject>, + restrict: bool, + func_name: &str, + vm: &VirtualMachine, + ) -> PyResult { + resolve_interp(id, restrict, true, "exec code for", vm)?; + #[cfg(feature = "threading")] + { + let shared_vals = if let Some(shared) = shared { + let dict = require_dict(shared, func_name, 3, vm)?; + let mut out = Vec::new(); + for (key, value) in dict { + let name = crossinterp::utf8_key(&key, vm)?.to_owned(); + let val = SharedValue::from_object(&value, vm)?; + out.push((name, val)); + } + Some(out) + } else { + None + }; + return match crossinterp::with_interpreter(id, vm, |target| { + if let Some(vals) = &shared_vals { + let ns = target.main_namespace()?; + for (name, val) in vals { + ns.set_item(name.as_str(), val.clone().into_object(target)?, target)?; + } + } + let ns = target.main_namespace()?; + let scope = crate::scope::Scope::with_builtins(None, ns, target); + match target.run_code_obj(code.clone(), scope) { + Ok(_) => Ok(None), + Err(exc) => Ok(Some(ExcInfo::capture(&exc, target))), + } + }) { + Ok(None) => Ok(vm.ctx.none()), + Ok(Some(info)) => Ok(info.into_namespace(vm)), + Err(e) => Err(e), + }; + } + #[cfg(not(feature = "threading"))] + { + let _ = (code, shared, func_name); + Err(vm.new_runtime_error("isolated interpreters require threading")) + } + } + + #[pyfunction] + fn exec(args: FuncArgs, vm: &VirtualMachine) -> PyResult { + let id = parse_id( + args.args + .first() + .ok_or_else(|| vm.new_type_error("_interpreters.exec() missing argument 1"))?, + vm, + )?; + let code_obj = args + .args + .get(1) + .ok_or_else(|| vm.new_type_error("_interpreters.exec() missing argument 2"))?; + let shared = args.args.get(2).or_else(|| args.kwargs.get("shared")); + if let Some(s) = shared { + require_dict(s, "_interpreters.exec()", 3, vm)?; + } + let restrict = args + .kwargs + .get("restrict") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(false); + let code = crossinterp::script_code(code_obj, vm)?; + run_code_in( + id, + code, + shared.map(|o| o.as_ref()), + restrict, + "_interpreters.exec", + vm, + ) + } + + #[pyfunction] + fn run_string(args: FuncArgs, vm: &VirtualMachine) -> PyResult { + let id = parse_id( + args.args.first().ok_or_else(|| { + vm.new_type_error("_interpreters.run_string() missing argument 1") + })?, + vm, + )?; + let script = args + .args + .get(1) + .ok_or_else(|| vm.new_type_error("_interpreters.run_string() missing argument 2"))?; + if script.downcast_ref::().is_none() { + return Err(vm.new_type_error(format!( + "_interpreters.run_string() argument 2 must be a string, not {}", + script.class() + ))); + } + let shared = args.args.get(2).or_else(|| args.kwargs.get("shared")); + if let Some(s) = shared { + require_dict(s, "_interpreters.run_string()", 3, vm)?; + } + let restrict = args + .kwargs + .get("restrict") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(false); + let code = crossinterp::script_code(script, vm)?; + run_code_in( + id, + code, + shared.map(|o| o.as_ref()), + restrict, + "_interpreters.run_string", + vm, + ) + } + + #[pyfunction] + fn run_func(args: FuncArgs, vm: &VirtualMachine) -> PyResult { + let id = parse_id( + args.args + .first() + .ok_or_else(|| vm.new_type_error("_interpreters.run_func() missing argument 1"))?, + vm, + )?; + let func = args + .args + .get(1) + .ok_or_else(|| vm.new_type_error("_interpreters.run_func() missing argument 2"))?; + if func.downcast_ref::().is_none() && func.downcast_ref::().is_none() { + return Err(vm.new_type_error(format!( + "_interpreters.run_func() argument 2 must be a function, not {}", + func.class() + ))); + } + let shared = args.args.get(2).or_else(|| args.kwargs.get("shared")); + if let Some(s) = shared { + require_dict(s, "_interpreters.run_func()", 3, vm)?; + } + let restrict = args + .kwargs + .get("restrict") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(false); + let code = crossinterp::script_code(func, vm)?; + run_code_in( + id, + code, + shared.map(|o| o.as_ref()), + restrict, + "_interpreters.run_func", + vm, + ) + } + + #[pyfunction(name = "set___main___attrs")] + fn set_main_attrs(args: FuncArgs, vm: &VirtualMachine) -> PyResult<()> { + let id = parse_id( + args.args.first().ok_or_else(|| { + vm.new_type_error("_interpreters.set___main___attrs() missing argument 1") + })?, + vm, + )?; + let updates = args.args.get(1).ok_or_else(|| { + vm.new_type_error("_interpreters.set___main___attrs() missing argument 2") + })?; + require_dict(updates, "_interpreters.set___main___attrs()", 2, vm)?; + if updates + .downcast_ref::() + .is_some_and(|d| d.is_empty()) + { + return Err(vm.new_value_error("arg 2 must be a non-empty dict")); + } + let restrict = args + .kwargs + .get("restrict") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(false); + resolve_interp(id, restrict, true, "update __main__ for", vm)?; + let dict = updates.downcast_ref::().unwrap(); + let mut vals = Vec::new(); + for (key, value) in dict { + let name = crossinterp::utf8_key(&key, vm)?.to_owned(); + let val = SharedValue::from_object(&value, vm)?; + vals.push((name, val)); + } + #[cfg(feature = "threading")] + { + return crossinterp::with_interpreter(id, vm, |target| { + let ns = target.main_namespace()?; + for (name, val) in &vals { + ns.set_item(name.as_str(), val.clone().into_object(target)?, target)?; + } + Ok(()) + }); + } + #[cfg(not(feature = "threading"))] + { + let _ = vals; + Err(vm.new_runtime_error("isolated interpreters require threading")) + } + } + + #[pyfunction] + fn incref(args: FuncArgs, vm: &VirtualMachine) -> PyResult<()> { + let id = parse_id( + args.args + .first() + .ok_or_else(|| vm.new_type_error("incref() missing argument 1"))?, + vm, + )?; + let restricted = args + .kwargs + .get("restrict") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(false); + let implieslink = args + .kwargs + .get("implieslink") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(false); + resolve_interp(id, restricted, true, "incref", vm)?; + let state = runtime::lookup_interpreter(id).ok_or_else(|| interpreter_not_found(vm, id))?; + if implieslink { + state.require_idref.store(true, Ordering::Release); + } + state.id_refcount.fetch_add(1, Ordering::AcqRel); + Ok(()) + } + + #[pyfunction] + fn decref(args: FuncArgs, vm: &VirtualMachine) -> PyResult<()> { + let id = parse_id( + args.args + .first() + .ok_or_else(|| vm.new_type_error("decref() missing argument 1"))?, + vm, + )?; + let restricted = args + .kwargs + .get("restrict") + .map(|o| o.clone().is_true(vm)) + .transpose()? + .unwrap_or(false); + resolve_interp(id, restricted, true, "decref", vm)?; + let state = runtime::lookup_interpreter(id).ok_or_else(|| interpreter_not_found(vm, id))?; + let prev = state.id_refcount.fetch_sub(1, Ordering::AcqRel); + if prev <= 1 && state.require_idref.load(Ordering::Acquire) { + #[cfg(feature = "threading")] + { + if runtime::is_owned_interpreter(id) && id != vm.state.interpreter_id { + let _ = runtime::destroy_owned_interpreter(id); + } + } + } + Ok(()) + } + + #[pyfunction] + fn capture_exception(exc: OptionalArg, vm: &VirtualMachine) -> PyResult { + let exc = match exc.into_option() { + Some(e) if !vm.is_none(&e) => e + .downcast::() + .map_err(|_| vm.new_type_error("expected exception"))?, + _ => { + return Ok(vm.ctx.none()); + } + }; + Ok(ExcInfo::capture(&exc, vm).into_namespace(vm)) + } +} diff --git a/crates/vm/src/stdlib/_thread.rs b/crates/vm/src/stdlib/_thread.rs index 45b92ff97e7..87dcbdce452 100644 --- a/crates/vm/src/stdlib/_thread.rs +++ b/crates/vm/src/stdlib/_thread.rs @@ -701,9 +701,8 @@ pub(crate) mod _thread { } #[pyfunction] - fn daemon_threads_allowed() -> bool { - // RustPython always allows daemon threads - true + fn daemon_threads_allowed(vm: &VirtualMachine) -> bool { + vm.state.allow_daemon_threads() } // Registry for non-daemon threads that need to be joined at shutdown diff --git a/crates/vm/src/stdlib/mod.rs b/crates/vm/src/stdlib/mod.rs index 97fb4b372cf..e20bda3878b 100644 --- a/crates/vm/src/stdlib/mod.rs +++ b/crates/vm/src/stdlib/mod.rs @@ -59,6 +59,8 @@ pub(crate) mod msvcrt; ))] mod pwd; +pub(crate) mod _interpchannels; +pub(crate) mod _interpreters; #[cfg(feature = "host_env")] pub(crate) mod _signal; #[cfg(feature = "threading")] @@ -134,6 +136,8 @@ pub fn builtin_module_defs(ctx: &Context) -> Vec<&'static PyModuleDef> { _sysconfig::module_def(ctx), #[cfg(feature = "threading")] _thread::module_def(ctx), + _interpchannels::module_def(ctx), + _interpreters::module_def(ctx), time::module_def(ctx), _typing::module_def(ctx), _warnings::module_def(ctx), diff --git a/crates/vm/src/stdlib/posix.rs b/crates/vm/src/stdlib/posix.rs index 5953f9c874d..ed49ce553a4 100644 --- a/crates/vm/src/stdlib/posix.rs +++ b/crates/vm/src/stdlib/posix.rs @@ -848,6 +848,11 @@ pub mod module { #[pyfunction] fn fork(vm: &VirtualMachine) -> PyResult { + if !vm.state.allow_fork() { + return Err( + vm.new_runtime_error("fork not supported for isolated subinterpreters".to_owned()) + ); + } if vm .state .finalizing @@ -1077,6 +1082,11 @@ pub mod module { argv: Either, vm: &VirtualMachine, ) -> PyResult<()> { + if !vm.state.allow_exec() { + return Err( + vm.new_runtime_error("exec not supported for isolated subinterpreters".to_owned()) + ); + } let path = path.into_cstring(vm)?; let argv = vm.extract_elements_with(argv.as_ref(), |obj| { @@ -1101,6 +1111,11 @@ pub mod module { env: ArgMapping, vm: &VirtualMachine, ) -> PyResult<()> { + if !vm.state.allow_exec() { + return Err( + vm.new_runtime_error("exec not supported for isolated subinterpreters".to_owned()) + ); + } let path = path.into_cstring(vm)?; let argv = vm.extract_elements_with(argv.as_ref(), |obj| { diff --git a/crates/vm/src/vm/crossinterp.rs b/crates/vm/src/vm/crossinterp.rs new file mode 100644 index 00000000000..98bc2eba346 --- /dev/null +++ b/crates/vm/src/vm/crossinterp.rs @@ -0,0 +1,387 @@ +//! Cross-interpreter object sharing and exception snapshots (CPython XI data). +//! +//! Interpreters do not share `PyObject` graphs. Shareable values are converted +//! to an interpreter-neutral payload and rebuilt in the destination. + +use crate::{ + AsObject, PyObject, PyObjectRef, PyPayload, PyRef, PyResult, VirtualMachine, + builtins::{ + PyBaseExceptionRef, PyBytes, PyCode, PyFloat, PyFunction, PyInt, PyMemoryView, PyStr, + PyTuple, + }, + bytecode::{BorrowedConstant, Constant, Instruction, oparg::OpArgState}, + protocol::PyBuffer, +}; +use malachite_bigint::BigInt; +use rustpython_common::wtf8::Wtf8Buf; + +/// CPython unbound-item ops (`UNBOUND_REMOVE` / `UNBOUND_ERROR` / `UNBOUND`). +pub const UNBOUND_REMOVE: i32 = 1; +pub const UNBOUND_ERROR: i32 = 2; +pub const UNBOUND_REPLACE: i32 = 3; + +/// Interpreter-neutral payload that can be materialized in any interpreter +/// sharing the process-wide [`crate::Context`]. +#[derive(Clone)] +pub enum SharedValue { + None, + Bool(bool), + Int(BigInt), + Float(f64), + Bytes(Vec), + Str(Wtf8Buf), + Tuple(Vec), + Channel { + cid: i64, + end: i32, + }, + #[cfg(feature = "threading")] + Buffer(PyBuffer), + #[cfg(not(feature = "threading"))] + Buffer(Vec), +} + +impl SharedValue { + pub fn from_object(obj: &PyObject, vm: &VirtualMachine) -> PyResult { + if vm.is_none(obj) { + return Ok(Self::None); + } + if obj.class().is(vm.ctx.types.bool_type) { + return Ok(Self::Bool(obj.to_owned().is_true(vm)?)); + } + if obj.class().is(vm.ctx.types.int_type) { + let n = obj + .downcast_ref::() + .ok_or_else(|| not_shareable(vm, obj))?; + return Ok(Self::Int(n.as_bigint().clone())); + } + if obj.class().is(vm.ctx.types.float_type) { + let f = obj + .downcast_ref::() + .ok_or_else(|| not_shareable(vm, obj))?; + return Ok(Self::Float(f.to_f64())); + } + if obj.class().is(vm.ctx.types.bytes_type) { + let b = obj + .downcast_ref::() + .ok_or_else(|| not_shareable(vm, obj))?; + return Ok(Self::Bytes(b.as_bytes().to_vec())); + } + if obj.class().is(vm.ctx.types.str_type) { + let s = obj + .downcast_ref::() + .ok_or_else(|| not_shareable(vm, obj))?; + return Ok(Self::Str(s.as_wtf8().to_owned())); + } + if obj.class().is(vm.ctx.types.tuple_type) { + let t = obj + .downcast_ref::() + .ok_or_else(|| not_shareable(vm, obj))?; + let mut items = Vec::with_capacity(t.len()); + for item in t { + items.push(Self::from_object(item, vm)?); + } + return Ok(Self::Tuple(items)); + } + if let Some(ch) = crate::stdlib::_interpchannels::channel_id_parts(obj) { + return Ok(Self::Channel { + cid: ch.0, + end: ch.1, + }); + } + Err(not_shareable(vm, obj)) + } + + pub fn from_buffer_object(obj: &PyObject, vm: &VirtualMachine) -> PyResult { + let view = PyMemoryView::from_object(obj, vm)?; + #[cfg(feature = "threading")] + { + Ok(Self::Buffer(view.clone_buffer())) + } + #[cfg(not(feature = "threading"))] + { + let buf = view.clone_buffer(); + let bytes = buf.as_contiguous().map(|b| b.to_vec()).unwrap_or_default(); + Ok(Self::Buffer(bytes)) + } + } + + pub fn into_object(self, vm: &VirtualMachine) -> PyResult { + match self { + Self::None => Ok(vm.ctx.none()), + Self::Bool(b) => Ok(vm.ctx.new_bool(b).into()), + Self::Int(n) => Ok(vm.ctx.new_int(n).into()), + Self::Float(f) => Ok(vm.ctx.new_float(f).into()), + Self::Bytes(b) => Ok(vm.ctx.new_bytes(b).into()), + Self::Str(s) => Ok(vm.ctx.new_str(s).into()), + Self::Tuple(items) => { + let mut els = Vec::with_capacity(items.len()); + for item in items { + els.push(item.into_object(vm)?); + } + Ok(vm.ctx.new_tuple(els).into()) + } + Self::Channel { cid, end } => { + crate::stdlib::_interpchannels::channel_id_from_parts(cid, end, false, false, vm) + } + Self::Buffer(buffer) => { + #[cfg(feature = "threading")] + { + let mv = PyMemoryView::from_buffer(buffer, vm)?; + Ok(mv.into_pyobject(vm)) + } + #[cfg(not(feature = "threading"))] + { + let mv = PyMemoryView::from_buffer(PyBuffer::from_byte_vector(buffer, vm), vm)?; + Ok(mv.into_pyobject(vm)) + } + } + } + } +} + +fn not_shareable(vm: &VirtualMachine, obj: &PyObject) -> crate::builtins::PyBaseExceptionRef { + let msg = format!("{} is not shareable", obj.class()); + crate::stdlib::_interpreters::not_shareable_error(vm, msg) +} + +/// Exact-type shareable check used by `_interpreters.is_shareable`. +pub fn is_shareable(obj: &PyObject, vm: &VirtualMachine) -> bool { + if vm.is_none(obj) { + return true; + } + let cls = obj.class(); + if cls.is(vm.ctx.types.bool_type) + || cls.is(vm.ctx.types.int_type) + || cls.is(vm.ctx.types.float_type) + || cls.is(vm.ctx.types.bytes_type) + || cls.is(vm.ctx.types.str_type) + { + return true; + } + if cls.is(vm.ctx.types.tuple_type) + && let Some(t) = obj.downcast_ref::() + { + return t.iter().all(|item| is_shareable(item, vm)); + } + crate::stdlib::_interpchannels::channel_id_parts(obj).is_some() +} + +/// Encode a dict key as UTF-8, rejecting lone surrogates like CPython. +pub fn utf8_key<'a>(key: &'a PyObject, vm: &'a VirtualMachine) -> PyResult<&'a str> { + let s = key + .downcast_ref::() + .ok_or_else(|| vm.new_type_error("attribute name must be a string"))?; + s.as_wtf8().as_str().map_err(|_| { + vm.new_unicode_encode_error_real( + vm.ctx.new_str("utf-8"), + s.to_owned(), + 0, + s.char_len(), + vm.ctx.new_str("surrogates not allowed"), + ) + }) +} + +/// Snapshot of an exception that can be rebuilt in another interpreter. +pub struct ExcInfo { + pub type_name: String, + pub type_qualname: String, + pub type_module: String, + pub msg: Option, + pub formatted: String, + pub errdisplay: String, +} + +impl ExcInfo { + pub fn capture(exc: &PyBaseExceptionRef, vm: &VirtualMachine) -> Self { + let cls = exc.class(); + let type_name = cls.name().to_owned(); + let type_qualname = cls + .as_object() + .get_attr("__qualname__", vm) + .ok() + .and_then(|o| o.str(vm).ok()) + .map_or_else(|| type_name.clone(), |s| s.to_string()); + let type_module = cls + .as_object() + .get_attr("__module__", vm) + .ok() + .and_then(|o| o.str(vm).ok()) + .map_or_else(|| "builtins".to_owned(), |s| s.to_string()); + let msg = exc.as_object().str(vm).ok().map(|s| s.to_string()); + let msg = msg.filter(|m| !m.is_empty()); + let formatted = match &msg { + Some(m) => format!("{type_name}: {m}"), + None => type_name.clone(), + }; + let mut errdisplay = String::new(); + let _ = vm.write_exception(&mut errdisplay, exc); + Self { + type_name, + type_qualname, + type_module, + msg, + formatted, + errdisplay, + } + } + + pub fn into_namespace(self, vm: &VirtualMachine) -> PyObjectRef { + let type_ns = crate::py_namespace!(vm, { + "__name__" => vm.ctx.new_str(self.type_name), + "__qualname__" => vm.ctx.new_str(self.type_qualname), + "__module__" => vm.ctx.new_str(self.type_module), + }); + let msg_obj = match self.msg { + Some(m) => vm.ctx.new_str(m).into(), + None => vm.ctx.none(), + }; + crate::py_namespace!(vm, { + "type" => type_ns, + "msg" => msg_obj, + "formatted" => vm.ctx.new_str(self.formatted), + "errdisplay" => vm.ctx.new_str(self.errdisplay), + }) + .into() + } +} + +/// Verify a code object is a valid `_interpreters` script. +pub fn verify_script(code: &PyCode, vm: &VirtualMachine) -> PyResult<()> { + if !code.freevars.is_empty() || !code.cellvars.is_empty() { + return Err(vm.new_value_error("code with a closure is not a script")); + } + if code.arg_count > 0 + || code.posonlyarg_count > 0 + || code.kwonlyarg_count > 0 + || code.flags.contains(crate::bytecode::CodeFlags::VARARGS) + || code.flags.contains(crate::bytecode::CodeFlags::VARKEYWORDS) + { + return Err(vm.new_value_error("code with args not supported")); + } + if !code_returns_only_none(code) { + return Err(vm.new_value_error("code that returns a value is not a script")); + } + Ok(()) +} + +fn code_returns_only_none(code: &PyCode) -> bool { + let mut arg_state = OpArgState::default(); + let mut prev_none = false; + for unit in code.instructions.iter().copied() { + let (op, arg) = arg_state.get(unit); + match op { + Instruction::LoadConst { consti } => { + let cidx = consti.get(arg); + prev_none = matches!( + code.constants[cidx].borrow_constant(), + BorrowedConstant::None + ); + } + Instruction::ReturnValue | Instruction::InstrumentedReturnValue => { + if !prev_none { + return false; + } + } + Instruction::ExtendedArg => {} + _ => prev_none = false, + } + } + true +} + +/// Extract a script code object from a str / function / code object. +pub fn script_code(obj: &PyObject, vm: &VirtualMachine) -> PyResult> { + if let Ok(code) = obj.to_owned().downcast::() { + verify_script(&code, vm)?; + return Ok(code); + } + if let Ok(func) = obj.to_owned().downcast::() { + let code = (*func.code).to_owned(); + if func.closure.is_some() { + return Err(vm.new_value_error("code with a closure is not a script")); + } + verify_script(&code, vm)?; + return Ok(code); + } + if let Ok(s) = obj.to_owned().downcast::() { + // Compile in the caller so SyntaxError is raised here, matching CPython. + let src = s.as_wtf8().as_str().map_err(|_| { + vm.new_exception_msg( + vm.ctx.exceptions.unicode_encode_error.to_owned(), + "surrogates not allowed".to_owned().into(), + ) + })?; + let code = vm + .compile(src, crate::compiler::Mode::Exec, "