//! Implement virtual machine to run instructions.

//!

//! See also:

//! https://github.com/ProgVal/pythonvm-rust/blob/master/src/processor/mod.rs

//!

use std::cell::{Cell, Ref, RefCell};

use std::collections::HashMap;

use std::collections::HashSet;

use std::fmt;

use crossbeam_utils::atomic::AtomicCell;

use num_traits::{Signed, ToPrimitive};

use crate::builtins::code::{self, PyCode, PyCodeRef};

use crate::builtins::dict::PyDictRef;

use crate::builtins::int::{PyInt, PyIntRef};

use crate::builtins::list::PyList;

use crate::builtins::module::{self, PyModule};

use crate::builtins::object;

use crate::builtins::pybool;

use crate::builtins::pystr::{PyStr, PyStrRef};

use crate::builtins::pytype::PyTypeRef;

use crate::builtins::tuple::{PyTuple, PyTupleTyped};

use crate::common::{hash::HashSecret, lock::PyMutex, rc::PyRc};

#[cfg(feature = "rustpython-compiler")]

use crate::compile::{self, CompileError, CompileErrorType, CompileOpts};

use crate::exceptions::{self, PyBaseException, PyBaseExceptionRef};

use crate::frame::{ExecutionResult, Frame, FrameRef};

use crate::function::{FuncArgs, IntoFuncArgs};

use crate::pyobject::{

BorrowValue, Either, IdProtocol, IntoPyObject, ItemProtocol, PyArithmaticValue, PyContext,

PyObject, PyObjectRef, PyRef, PyRefExact, PyResult, PyValue, TryFromObject, TryIntoRef,

TypeProtocol,

};

use crate::scope::Scope;

use crate::slots::PyComparisonOp;

use crate::{builtins, bytecode, frozen, import, iterator, stdlib, sysmodule};

// use objects::ects;

// Objects are live when they are on stack, or referenced by a name (for now)

/// Top level container of a python virtual machine. In theory you could

/// create more instances of this struct and have them operate fully isolated.

pub struct VirtualMachine {

pub builtins: PyObjectRef,

pub sys_module: PyObjectRef,

pub ctx: PyRc<PyContext>,

pub frames: RefCell<Vec<FrameRef>>,

pub wasm_id: Option<String>,

pub exceptions: RefCell<Vec<PyBaseExceptionRef>>,

pub import_func: PyObjectRef,

pub profile_func: RefCell<PyObjectRef>,

pub trace_func: RefCell<PyObjectRef>,

pub use_tracing: Cell<bool>,

pub recursion_limit: Cell<usize>,

pub signal_handlers: Option<Box<RefCell<[Option<PyObjectRef>; NSIG]>>>,

pub repr_guards: RefCell<HashSet<usize>>,

pub state: PyRc<PyGlobalState>,

pub initialized: bool,

}

pub(crate) mod thread {

use super::{PyObjectRef, TypeProtocol, VirtualMachine};

use itertools::Itertools;

use std::cell::RefCell;

use std::ptr::NonNull;

use std::thread_local;

thread_local! {

pub(super) static VM_STACK: RefCell<Vec<NonNull<VirtualMachine>>> = Vec::with_capacity(1).into();

}

pub fn enter_vm<R>(vm: &VirtualMachine, f: impl FnOnce() -> R) -> R {

VM_STACK.with(|vms| {

vms.borrow_mut().push(vm.into());

let ret = std::panic::catch_unwind(std::panic::AssertUnwindSafe(f));

vms.borrow_mut().pop();

ret.unwrap_or_else(|e| std::panic::resume_unwind(e))

})

}

pub fn with_vm<F, R>(obj: &PyObjectRef, f: F) -> R

where

F: Fn(&VirtualMachine) -> R,

{

let vm_owns_obj = |intp: NonNull<VirtualMachine>| {

// SAFETY: all references in VM_STACK should be valid

let vm = unsafe { intp.as_ref() };

obj.isinstance(&vm.ctx.types.object_type)

};

VM_STACK.with(|vms| {

let intp = match vms.borrow().iter().copied().exactly_one() {

Ok(x) => {

debug_assert!(vm_owns_obj(x));

x

}

Err(mut others) => others

.find(|x| vm_owns_obj(*x))

.unwrap_or_else(|| panic!("can't get a vm for {:?}; none on stack", obj)),

};

// SAFETY: all references in VM_STACK should be valid, and should not be changed or moved

// at least until this function returns and the stack unwinds to an enter_vm() call

let vm = unsafe { intp.as_ref() };

f(vm)

})

}

}

pub struct PyGlobalState {

pub settings: PySettings,

pub stdlib_inits: HashMap<String, stdlib::StdlibInitFunc, ahash::RandomState>,

pub frozen: HashMap<String, code::FrozenModule, ahash::RandomState>,

pub stacksize: AtomicCell<usize>,

pub thread_count: AtomicCell<usize>,

pub hash_secret: HashSecret,

pub atexit_funcs: PyMutex<Vec<(PyObjectRef, FuncArgs)>>,

}

pub const NSIG: usize = 64;

#[derive(Copy, Clone, PartialEq, Eq)]

pub enum InitParameter {

Internal,

External,

}

/// Struct containing all kind of settings for the python vm.

pub struct PySettings {

/// -d command line switch

pub debug: bool,

/// -i

pub inspect: bool,

/// -i, with no script

pub interactive: bool,

/// -O optimization switch counter

pub optimize: u8,

/// -s

pub no_user_site: bool,

/// -S

pub no_site: bool,

/// -E

pub ignore_environment: bool,

/// verbosity level (-v switch)

pub verbose: u8,

/// -q

pub quiet: bool,

/// -B

pub dont_write_bytecode: bool,

/// -b

pub bytes_warning: u64,

/// -Xfoo[=bar]

pub xopts: Vec<(String, Option<String>)>,

/// -I

pub isolated: bool,

/// -Xdev

pub dev_mode: bool,

/// -Wfoo

pub warnopts: Vec<String>,

/// Environment PYTHONPATH and RUSTPYTHONPATH:

pub path_list: Vec<String>,

/// sys.argv

pub argv: Vec<String>,

/// PYTHONHASHSEED=x

pub hash_seed: Option<u32>,

}

/// Trace events for sys.settrace and sys.setprofile.

enum TraceEvent {

Call,

Return,

}

impl fmt::Display for TraceEvent {

fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {

use TraceEvent::*;

match self {

Call => write!(f, "call"),

Return => write!(f, "return"),

}

}

}

/// Sensible default settings.

impl Default for PySettings {

fn default() -> Self {

PySettings {

debug: false,

inspect: false,

interactive: false,

optimize: 0,

no_user_site: false,

no_site: false,

ignore_environment: false,

verbose: 0,

quiet: false,

dont_write_bytecode: false,

bytes_warning: 0,

xopts: vec![],

isolated: false,

dev_mode: false,

warnopts: vec![],

path_list: vec![],

argv: vec![],

hash_seed: None,

}

}

}

impl VirtualMachine {

/// Create a new `VirtualMachine` structure.

fn new(settings: PySettings) -> VirtualMachine {

flame_guard!("new VirtualMachine");

let ctx = PyContext::new();

// make a new module without access to the vm; doesn't

// set __spec__, __loader__, etc. attributes

let new_module =

|dict| PyObject::new(PyModule {}, ctx.types.module_type.clone(), Some(dict));

// Hard-core modules:

let builtins_dict = ctx.new_dict();

let builtins = new_module(builtins_dict.clone());

let sysmod_dict = ctx.new_dict();

let sysmod = new_module(sysmod_dict.clone());

let import_func = ctx.none();

let profile_func = RefCell::new(ctx.none());

let trace_func = RefCell::new(ctx.none());

// hack to get around const array repeat expressions, rust issue #79270

const NONE: Option<PyObjectRef> = None;

let signal_handlers = RefCell::new([NONE; NSIG]);

let stdlib_inits = stdlib::get_module_inits();

let hash_secret = match settings.hash_seed {

Some(seed) => HashSecret::new(seed),

None => rand::random(),

};

let mut vm = VirtualMachine {

builtins,

sys_module: sysmod,

ctx: PyRc::new(ctx),

frames: RefCell::new(vec![]),

wasm_id: None,

exceptions: RefCell::new(vec![]),

import_func,

profile_func,

trace_func,

use_tracing: Cell::new(false),

recursion_limit: Cell::new(if cfg!(debug_assertions) { 256 } else { 512 }),

signal_handlers: Some(Box::new(signal_handlers)),

repr_guards: RefCell::default(),

state: PyRc::new(PyGlobalState {

settings,

stdlib_inits,

frozen: HashMap::default(),

stacksize: AtomicCell::new(0),

thread_count: AtomicCell::new(0),

hash_secret,

atexit_funcs: PyMutex::default(),

}),

initialized: false,

};

let frozen = frozen::get_module_inits(&vm);

PyRc::get_mut(&mut vm.state).unwrap().frozen = frozen;

module::init_module_dict(

&vm,

&builtins_dict,

vm.ctx.new_str("builtins"),

vm.ctx.none(),

);

module::init_module_dict(&vm, &sysmod_dict, vm.ctx.new_str("sys"), vm.ctx.none());

vm

}

fn initialize(&mut self, initialize_parameter: InitParameter) {

flame_guard!("init VirtualMachine");

if self.initialized {

panic!("Double Initialize Error");

}

builtins::make_module(self, self.builtins.clone());

sysmodule::make_module(self, self.sys_module.clone(), self.builtins.clone());

let mut inner_init = || -> PyResult<()> {

#[cfg(not(target_arch = "wasm32"))]

import::import_builtin(self, "_signal")?;

#[cfg(any(not(target_arch = "wasm32"), target_os = "wasi"))]

{

// this isn't fully compatible with CPython; it imports "io" and sets

// builtins.open to io.OpenWrapper, but this is easier, since it doesn't

// require the Python stdlib to be present

let io = import::import_builtin(self, "_io")?;

let set_stdio = |name, fd, mode: &str| {

let stdio = crate::stdlib::io::open(

self.ctx.new_int(fd),

Some(mode),

Default::default(),

self,

)?;

self.set_attr(

&self.sys_module,

format!("__{}__", name), // e.g. __stdin__

stdio.clone(),

)?;

self.set_attr(&self.sys_module, name, stdio)?;

Ok(())

};

set_stdio("stdin", 0, "r")?;

set_stdio("stdout", 1, "w")?;

set_stdio("stderr", 2, "w")?;

let io_open = self.get_attribute(io, "open")?;

self.set_attr(&self.builtins, "open", io_open)?;

}

import::init_importlib(self, initialize_parameter)?;

Ok(())

};

let res = inner_init();

self.expect_pyresult(res, "initializiation failed");

self.initialized = true;

}

/// Can only be used in the initialization closure passed to [`Interpreter::new_with_init`]

pub fn add_native_module(&mut self, name: String, module: stdlib::StdlibInitFunc) {

let state = PyRc::get_mut(&mut self.state)

.expect("can't add_native_module when there are multiple threads");

state.stdlib_inits.insert(name, module);

}

/// Can only be used in the initialization closure passed to [`Interpreter::new_with_init`]

pub fn add_frozen<I>(&mut self, frozen: I)

where

I: IntoIterator<Item = (String, bytecode::FrozenModule)>,

{

let frozen = frozen::map_frozen(self, frozen).collect::<Vec<_>>();

let state = PyRc::get_mut(&mut self.state)

.expect("can't add_frozen when there are multiple threads");

state.frozen.extend(frozen);

}

/// Start a new thread with access to the same interpreter.

///

/// # Note

///

/// If you return a `PyObjectRef` (or a type that contains one) from `F`, and don't `join()`

/// on the thread, there is a possibility that that thread will panic as `PyObjectRef`'s `Drop`

/// implementation tries to run the `__del__` destructor of a python object but finds that it's

/// not in the context of any vm.

#[cfg(feature = "threading")]

pub fn start_thread<F, R>(&self, f: F) -> std::thread::JoinHandle<R>

where

F: FnOnce(&VirtualMachine) -> R,

F: Send + 'static,

R: Send + 'static,

{

let thread = self.new_thread();

std::thread::spawn(|| thread.run(f))

}

/// Create a new VM thread that can be passed to a function like [`std::thread::spawn`]

/// to use the same interpreter on a different thread. Note that if you just want to

/// use this with `thread::spawn`, you can use

/// [`vm.start_thread()`](`VirtualMachine::start_thread`) as a convenience.

///

/// # Usage

///

/// ```

/// # rustpython_vm::Interpreter::default().enter(|vm| {

/// use std::thread::Builder;

/// let handle = Builder::new()

/// .name("my thread :)".into())

/// .spawn(vm.new_thread().make_spawn_func(|vm| vm.ctx.none()))

/// .expect("couldn't spawn thread");

/// let returned_obj = handle.join().expect("thread panicked");

/// assert!(vm.is_none(&returned_obj));

/// # })

/// ```

///

/// Note: this function is safe, but running the returned PyThread in the same

/// thread context (i.e. with the same thread-local storage) doesn't have any

/// specific guaranteed behavior.

#[cfg(feature = "threading")]

pub fn new_thread(&self) -> PyThread {

let thread_vm = VirtualMachine {

builtins: self.builtins.clone(),

sys_module: self.sys_module.clone(),

ctx: self.ctx.clone(),

frames: RefCell::new(vec![]),

wasm_id: self.wasm_id.clone(),

exceptions: RefCell::new(vec![]),

import_func: self.import_func.clone(),

profile_func: RefCell::new(self.ctx.none()),

trace_func: RefCell::new(self.ctx.none()),

use_tracing: Cell::new(false),

recursion_limit: self.recursion_limit.clone(),

signal_handlers: None,

repr_guards: RefCell::default(),

state: self.state.clone(),

initialized: self.initialized,

};

PyThread { thread_vm }

}

pub fn run_atexit_funcs(&self) -> PyResult<()> {

let mut last_exc = None;

for (func, args) in self.state.atexit_funcs.lock().drain(..).rev() {

if let Err(e) = self.invoke(&func, args) {

last_exc = Some(e.clone());

if !e.isinstance(&self.ctx.exceptions.system_exit) {

writeln!(sysmodule::PyStderr(self), "Error in atexit._run_exitfuncs:");

exceptions::print_exception(self, e);

}

}

}

match last_exc {

None => Ok(()),

Some(e) => Err(e),

}

}

pub fn run_code_obj(&self, code: PyCodeRef, scope: Scope) -> PyResult {

let frame =

Frame::new(code, scope, self.builtins.dict().unwrap(), &[], self).into_ref(self);

self.run_frame_full(frame)

}

pub fn run_frame_full(&self, frame: FrameRef) -> PyResult {

match self.run_frame(frame)? {

ExecutionResult::Return(value) => Ok(value),

_ => panic!("Got unexpected result from function"),

}

}

pub fn with_frame<R, F: FnOnce(FrameRef) -> PyResult<R>>(

&self,

frame: FrameRef,

f: F,

) -> PyResult<R> {

self.check_recursive_call("")?;

self.frames.borrow_mut().push(frame.clone());

let result = f(frame);

// defer dec frame

let _popped = self.frames.borrow_mut().pop();

result

}

pub fn run_frame(&self, frame: FrameRef) -> PyResult<ExecutionResult> {

self.with_frame(frame, |f| f.run(self))

}

fn check_recursive_call(&self, _where: &str) -> PyResult<()> {

if self.frames.borrow().len() > self.recursion_limit.get() {

Err(self.new_recursion_error(format!("maximum recursion depth exceeded {}", _where)))

} else {

Ok(())

}

}

pub fn current_frame(&self) -> Option<Ref<FrameRef>> {

let frames = self.frames.borrow();

if frames.is_empty() {

None

} else {

Some(Ref::map(self.frames.borrow(), |frames| {

frames.last().unwrap()

}))

}

}

pub fn current_locals(&self) -> PyResult<PyDictRef> {

self.current_frame()

.expect("called current_locals but no frames on the stack")

.locals(self)

}

pub fn current_globals(&self) -> Ref<PyDictRef> {

let frame = self

.current_frame()

.expect("called current_globals but no frames on the stack");

Ref::map(frame, |f| &f.globals)

}

pub fn try_class(&self, module: &str, class: &str) -> PyResult<PyTypeRef> {

let class = self

.get_attribute(self.import(module, None, 0)?, class)?

.downcast()

.expect("not a class");

Ok(class)

}

pub fn class(&self, module: &str, class: &str) -> PyTypeRef {

let module = self

.import(module, None, 0)

.unwrap_or_else(|_| panic!("unable to import {}", module));

let class = self

.get_attribute(module.clone(), class)

.unwrap_or_else(|_| panic!("module {} has no class {}", module, class));

class.downcast().expect("not a class")

}

/// Create a new python object

pub fn new_pyobj<T: IntoPyObject>(&self, value: T) -> PyObjectRef {

value.into_pyobject(self)

}

pub fn new_code_object(&self, code: impl code::IntoCodeObject) -> PyCodeRef {

self.ctx.new_code_object(code.into_codeobj(self))

}

pub fn new_module(&self, name: &str, dict: PyDictRef) -> PyObjectRef {

module::init_module_dict(

self,

&dict,

self.new_pyobj(name.to_owned()),

self.ctx.none(),

);

PyObject::new(PyModule {}, self.ctx.types.module_type.clone(), Some(dict))

}

/// Instantiate an exception with arguments.

/// This function should only be used with builtin exception types; if a user-defined exception

/// type is passed in, it may not be fully initialized; try using [`exceptions::invoke`](invoke)

/// or [`exceptions::ExceptionCtor`](ctor) instead.

///

/// [invoke]: rustpython_vm::exceptions::invoke

/// [ctor]: rustpython_vm::exceptions::ExceptionCtor

pub fn new_exception(&self, exc_type: PyTypeRef, args: Vec<PyObjectRef>) -> PyBaseExceptionRef {

// TODO: add repr of args into logging?

vm_trace!("New exception created: {}", exc_type.name);

PyRef::new_ref(

PyBaseException::new(args, self),

exc_type,

Some(self.ctx.new_dict()),

)

}

/// Instantiate an exception with no arguments.

/// This function should only be used with builtin exception types; if a user-defined exception

/// type is passed in, it may not be fully initialized; try using [`exceptions::invoke`](invoke)

/// or [`exceptions::ExceptionCtor`](ctor) instead.

///

/// [invoke]: rustpython_vm::exceptions::invoke

/// [ctor]: rustpython_vm::exceptions::ExceptionCtor

pub fn new_exception_empty(&self, exc_type: PyTypeRef) -> PyBaseExceptionRef {

self.new_exception(exc_type, vec![])

}

/// Instantiate an exception with `msg` as the only argument.

/// This function should only be used with builtin exception types; if a user-defined exception

/// type is passed in, it may not be fully initialized; try using [`exceptions::invoke`](invoke)

/// or [`exceptions::ExceptionCtor`](ctor) instead.

///

/// [invoke]: rustpython_vm::exceptions::invoke

/// [ctor]: rustpython_vm::exceptions::ExceptionCtor

pub fn new_exception_msg(&self, exc_type: PyTypeRef, msg: String) -> PyBaseExceptionRef {

self.new_exception(exc_type, vec![self.ctx.new_str(msg)])

}

pub fn new_lookup_error(&self, msg: String) -> PyBaseExceptionRef {

let lookup_error = self.ctx.exceptions.lookup_error.clone();

self.new_exception_msg(lookup_error, msg)

}

pub fn new_attribute_error(&self, msg: String) -> PyBaseExceptionRef {

let attribute_error = self.ctx.exceptions.attribute_error.clone();

self.new_exception_msg(attribute_error, msg)

}

pub fn new_type_error(&self, msg: String) -> PyBaseExceptionRef {

let type_error = self.ctx.exceptions.type_error.clone();

self.new_exception_msg(type_error, msg)

}

pub fn new_name_error(&self, msg: String) -> PyBaseExceptionRef {

let name_error = self.ctx.exceptions.name_error.clone();

self.new_exception_msg(name_error, msg)

}

pub fn new_unsupported_binop_error(

&self,

a: &PyObjectRef,

b: &PyObjectRef,

op: &str,

) -> PyBaseExceptionRef {

self.new_type_error(format!(

"Unsupported operand types for '{}': '{}' and '{}'",

op,

a.class().name,

b.class().name

))

}

pub fn new_unsupported_ternop_error(

&self,

a: &PyObjectRef,

b: &PyObjectRef,

c: &PyObjectRef,

op: &str,

) -> PyBaseExceptionRef {

self.new_type_error(format!(

"Unsupported operand types for '{}': '{}', '{}', and '{}'",

op,

a.class().name,

b.class().name,

c.class().name

))

}

pub fn new_os_error(&self, msg: String) -> PyBaseExceptionRef {

let os_error = self.ctx.exceptions.os_error.clone();

self.new_exception_msg(os_error, msg)

}

pub fn new_unicode_decode_error(&self, msg: String) -> PyBaseExceptionRef {

let unicode_decode_error = self.ctx.exceptions.unicode_decode_error.clone();

self.new_exception_msg(unicode_decode_error, msg)

}

pub fn new_unicode_encode_error(&self, msg: String) -> PyBaseExceptionRef {

let unicode_encode_error = self.ctx.exceptions.unicode_encode_error.clone();

self.new_exception_msg(unicode_encode_error, msg)

}

/// Create a new python ValueError object. Useful for raising errors from

/// python functions implemented in rust.

pub fn new_value_error(&self, msg: String) -> PyBaseExceptionRef {

let value_error = self.ctx.exceptions.value_error.clone();

self.new_exception_msg(value_error, msg)

}

pub fn new_buffer_error(&self, msg: String) -> PyBaseExceptionRef {

let buffer_error = self.ctx.exceptions.buffer_error.clone();

self.new_exception_msg(buffer_error, msg)

}

pub fn new_key_error(&self, obj: PyObjectRef) -> PyBaseExceptionRef {

let key_error = self.ctx.exceptions.key_error.clone();

self.new_exception(key_error, vec![obj])

}

pub fn new_index_error(&self, msg: String) -> PyBaseExceptionRef {

let index_error = self.ctx.exceptions.index_error.clone();

self.new_exception_msg(index_error, msg)

}

pub fn new_not_implemented_error(&self, msg: String) -> PyBaseExceptionRef {

let not_implemented_error = self.ctx.exceptions.not_implemented_error.clone();

self.new_exception_msg(not_implemented_error, msg)

}

pub fn new_recursion_error(&self, msg: String) -> PyBaseExceptionRef {

let recursion_error = self.ctx.exceptions.recursion_error.clone();

self.new_exception_msg(recursion_error, msg)

}

pub fn new_zero_division_error(&self, msg: String) -> PyBaseExceptionRef {

let zero_division_error = self.ctx.exceptions.zero_division_error.clone();

self.new_exception_msg(zero_division_error, msg)

}

pub fn new_overflow_error(&self, msg: String) -> PyBaseExceptionRef {

let overflow_error = self.ctx.exceptions.overflow_error.clone();

self.new_exception_msg(overflow_error, msg)

}

#[cfg(feature = "rustpython-compiler")]

pub fn new_syntax_error(&self, error: &CompileError) -> PyBaseExceptionRef {

let syntax_error_type = match &error.error {

CompileErrorType::Parse(p) if p.is_indentation_error() => {

self.ctx.exceptions.indentation_error.clone()

}

CompileErrorType::Parse(p) if p.is_tab_error() => self.ctx.exceptions.tab_error.clone(),

_ => self.ctx.exceptions.syntax_error.clone(),

};

let syntax_error = self.new_exception_msg(syntax_error_type, error.to_string());

let lineno = self.ctx.new_int(error.location.row());

let offset = self.ctx.new_int(error.location.column());

self.set_attr(syntax_error.as_object(), "lineno", lineno)

.unwrap();

self.set_attr(syntax_error.as_object(), "offset", offset)

.unwrap();

self.set_attr(

syntax_error.as_object(),

"text",

error.statement.clone().into_pyobject(self),

)

.unwrap();

self.set_attr(

syntax_error.as_object(),

"filename",

self.ctx.new_str(error.source_path.clone()),

)

.unwrap();

syntax_error

}

pub fn new_import_error(

&self,

msg: String,

name: impl TryIntoRef<PyStr>,

) -> PyBaseExceptionRef {

let import_error = self.ctx.exceptions.import_error.clone();

let exc = self.new_exception_msg(import_error, msg);

self.set_attr(exc.as_object(), "name", name.try_into_ref(self).unwrap())

.unwrap();

exc

}

pub fn new_runtime_error(&self, msg: String) -> PyBaseExceptionRef {

let runtime_error = self.ctx.exceptions.runtime_error.clone();

self.new_exception_msg(runtime_error, msg)

}

pub fn new_stop_iteration(&self) -> PyBaseExceptionRef {

let stop_iteration_type = self.ctx.exceptions.stop_iteration.clone();

self.new_exception_empty(stop_iteration_type)

}

// TODO: #[track_caller] when stabilized

fn _py_panic_failed(&self, exc: PyBaseExceptionRef, msg: &str) -> ! {

#[cfg(not(all(target_arch = "wasm32", not(target_os = "wasi"))))]

{

let show_backtrace = std::env::var_os("RUST_BACKTRACE").map_or(false, |v| &v != "0");

let after = if show_backtrace {

exceptions::print_exception(self, exc);

"exception backtrace above"

} else {

"run with RUST_BACKTRACE=1 to see Python backtrace"

};

panic!("{}; {}", msg, after)

}

#[cfg(all(target_arch = "wasm32", not(target_os = "wasi")))]

{

use wasm_bindgen::prelude::*;

#[wasm_bindgen]

extern "C" {

#[wasm_bindgen(js_namespace = console)]

fn error(s: &str);

}

let mut s = Vec::<u8>::new();

exceptions::write_exception(&mut s, self, &exc).unwrap();

error(std::str::from_utf8(&s).unwrap());

panic!("{}; exception backtrace above", msg)

}

}

pub fn unwrap_pyresult<T>(&self, result: PyResult<T>) -> T {

result.unwrap_or_else(|exc| {

self._py_panic_failed(exc, "called `vm.unwrap_pyresult()` on an `Err` value")

})

}

pub fn expect_pyresult<T>(&self, result: PyResult<T>, msg: &str) -> T {

result.unwrap_or_else(|exc| self._py_panic_failed(exc, msg))

}

pub fn new_scope_with_builtins(&self) -> Scope {

Scope::with_builtins(None, self.ctx.new_dict(), self)

}

/// Test whether a python object is `None`.

pub fn is_none(&self, obj: &PyObjectRef) -> bool {

obj.is(&self.ctx.none)

}

pub fn option_if_none(&self, obj: PyObjectRef) -> Option<PyObjectRef> {

if self.is_none(&obj) {

None

} else {

Some(obj)

}

}

pub fn unwrap_or_none(&self, obj: Option<PyObjectRef>) -> PyObjectRef {

obj.unwrap_or_else(|| self.ctx.none())

}

// Container of the virtual machine state:

pub fn to_str(&self, obj: &PyObjectRef) -> PyResult<PyStrRef> {

if obj.class().is(&self.ctx.types.str_type) {

Ok(obj.clone().downcast().unwrap())

} else {

let s = self.call_method(&obj, "__str__", ())?;

PyStrRef::try_from_object(self, s)

}

}

pub fn to_pystr<'a, T: Into<&'a PyObjectRef>>(&'a self, obj: T) -> PyResult<String> {

let py_str_obj = self.to_str(obj.into())?;

Ok(py_str_obj.borrow_value().to_owned())

}

pub fn to_repr(&self, obj: &PyObjectRef) -> PyResult<PyStrRef> {

let repr = self.call_method(obj, "__repr__", ())?;

TryFromObject::try_from_object(self, repr)

}

pub fn to_index_opt(&self, obj: PyObjectRef) -> Option<PyResult<PyIntRef>> {

match obj.downcast() {

Ok(val) => Some(Ok(val)),

Err(obj) => self.get_method(obj, "__index__").map(|index| {

// TODO: returning strict subclasses of int in __index__ is deprecated

self.invoke(&index?, ())?.downcast().map_err(|bad| {

self.new_type_error(format!(

"__index__ returned non-int (type {})",

bad.class().name

))

})

}),

}

}

pub fn to_index(&self, obj: &PyObjectRef) -> PyResult<PyIntRef> {

self.to_index_opt(obj.clone()).unwrap_or_else(|| {

Err(self.new_type_error(format!(

"'{}' object cannot be interpreted as an integer",

obj.class().name

)))

})

}

#[inline]

pub fn import(

&self,

module: impl TryIntoRef<PyStr>,

from_list: Option<PyTupleTyped<PyStrRef>>,

level: usize,

) -> PyResult {

self._import_inner(module.try_into_ref(self)?, from_list, level)

}

fn _import_inner(

&self,

module: PyStrRef,

from_list: Option<PyTupleTyped<PyStrRef>>,

level: usize,

) -> PyResult {

// if the import inputs seem weird, e.g a package import or something, rather than just

// a straight `import ident`

let weird = module.borrow_value().contains('.')

|| level != 0

|| from_list

.as_ref()

.map_or(false, |x| !x.borrow_value().is_empty());

let cached_module = if weird {

None

} else {

let sys_modules = self.get_attribute(self.sys_module.clone(), "modules")?;

sys_modules.get_item(module.clone(), self).ok()

};

match cached_module {

Some(cached_module) => {

if self.is_none(&cached_module) {

Err(self.new_import_error(

format!("import of {} halted; None in sys.modules", module),

module,

))

} else {

Ok(cached_module)

}

}

None => {

let import_func = self

.get_attribute(self.builtins.clone(), "__import__")

.map_err(|_| {

self.new_import_error("__import__ not found".to_owned(), module.clone())

})?;

let (locals, globals) = if let Some(frame) = self.current_frame() {

(Some(frame.locals.clone()), Some(frame.globals.clone()))

} else {

(None, None)

};

let from_list = match from_list {

Some(tup) => tup.into_pyobject(self),

None => self.ctx.new_tuple(vec![]),

};

self.invoke(&import_func, (module, globals, locals, from_list, level))

.map_err(|exc| import::remove_importlib_frames(self, &exc))

}

}

}

/// Determines if `obj` is an instance of `cls`, either directly, indirectly or virtually via

/// the __instancecheck__ magic method.

pub fn isinstance(&self, obj: &PyObjectRef, cls: &PyTypeRef) -> PyResult<bool> {

// cpython first does an exact check on the type, although documentation doesn't state that

// https://github.com/python/cpython/blob/a24107b04c1277e3c1105f98aff5bfa3a98b33a0/Objects/abstract.c#L2408

if obj.class().is(cls) {

Ok(true)

} else {

let ret = self.call_method(cls.as_object(), "__instancecheck__", (obj.clone(),))?;

pybool::boolval(self, ret)

}

}

/// Determines if `subclass` is a subclass of `cls`, either directly, indirectly or virtually

/// via the __subclasscheck__ magic method.

pub fn issubclass(&self, subclass: &PyTypeRef, cls: &PyTypeRef) -> PyResult<bool> {

let ret = self.call_method(cls.as_object(), "__subclasscheck__", (subclass.clone(),))?;

pybool::boolval(self, ret)

}

pub fn call_get_descriptor_specific(

&self,

descr: PyObjectRef,

obj: Option<PyObjectRef>,

cls: Option<PyObjectRef>,

) -> Option<PyResult> {

let descr_get = descr.class().mro_find_map(|cls| cls.slots.descr_get.load());

descr_get.map(|descr_get| descr_get(descr, obj, cls, self))

}

pub fn call_get_descriptor(&self, descr: PyObjectRef, obj: PyObjectRef) -> Option<PyResult> {

let cls = obj.clone_class().into_object();

self.call_get_descriptor_specific(descr, Some(obj), Some(cls))

}

pub fn call_if_get_descriptor(&self, attr: PyObjectRef, obj: PyObjectRef) -> PyResult {

self.call_get_descriptor(attr.clone(), obj)

.unwrap_or(Ok(attr))

}

pub fn call_method<T>(&self, obj: &PyObjectRef, method_name: &str, args: T) -> PyResult

where

T: IntoFuncArgs,

{

flame_guard!(format!("call_method({:?})", method_name));

// This is only used in the vm for magic methods, which use a greatly simplified attribute lookup.

match obj.get_class_attr(method_name) {

Some(func) => {

vm_trace!("vm.call_method {:?} {:?} -> {:?}", obj, method_name, func);

let wrapped = self.call_if_get_descriptor(func, obj.clone())?;

self.invoke(&wrapped, args)

}

None => Err(self.new_type_error(format!("Unsupported method: {}", method_name))),

}

}

fn _invoke(&self, callable: &PyObjectRef, args: FuncArgs) -> PyResult {

vm_trace!("Invoke: {:?} {:?}", callable, args);

let slot_call = callable.class().mro_find_map(|cls| cls.slots.call.load());

match slot_call {

Some(slot_call) => {

self.trace_event(TraceEvent::Call)?;

let result = slot_call(callable, args, self);

self.trace_event(TraceEvent::Return)?;

result

}

None => Err(self.new_type_error(format!(

"'{}' object is not callable",

callable.class().name

))),

}

}

#[inline]

pub fn invoke<T>(&self, func_ref: &PyObjectRef, args: T) -> PyResult

where

T: IntoFuncArgs,

{

self._invoke(func_ref, args.into_args(self))

}

/// Call registered trace function.

#[inline]

fn trace_event(&self, event: TraceEvent) -> PyResult<()> {

if self.use_tracing.get() {

self._trace_event_inner(event)

} else {

Ok(())