use std::cell::{Ref, RefCell};
use std::collections::hash_map::HashMap;
use std::collections::hash_set::HashSet;
use std::fmt;
use std::rc::Rc;
use std::sync::{Mutex, MutexGuard};
use crate::builtins::{self, to_ascii};
use crate::bytecode;
use crate::frame::{ExecutionResult, Frame, FrameRef};
use crate::frozen;
use crate::function::PyFuncArgs;
use crate::import;
use crate::obj::objbool;
use crate::obj::objbuiltinfunc::PyBuiltinFunction;
use crate::obj::objcode::{PyCode, PyCodeRef};
use crate::obj::objdict::PyDictRef;
use crate::obj::objfunction::{PyFunction, PyMethod};
use crate::obj::objgenerator::PyGenerator;
use crate::obj::objint::PyInt;
use crate::obj::objiter;
use crate::obj::objmodule::{self, PyModule};
use crate::obj::objsequence;
use crate::obj::objstr::{PyString, PyStringRef};
use crate::obj::objtuple::PyTupleRef;
use crate::obj::objtype;
use crate::obj::objtype::PyClassRef;
use crate::pyhash;
use crate::pyobject::{
IdProtocol, ItemProtocol, PyContext, PyObject, PyObjectRef, PyResult, PyValue, TryFromObject,
TryIntoRef, TypeProtocol,
};
use crate::scope::Scope;
use crate::stdlib;
use crate::sysmodule;
use arr_macro::arr;
use num_bigint::BigInt;
#[cfg(feature = "rustpython-compiler")]
use rustpython_compiler::{compile, error::CompileError};
pub struct VirtualMachine {
pub builtins: PyObjectRef,
pub sys_module: PyObjectRef,
pub stdlib_inits: RefCell<HashMap<String, stdlib::StdlibInitFunc>>,
pub ctx: PyContext,
pub frames: RefCell<Vec<FrameRef>>,
pub wasm_id: Option<String>,
pub exceptions: RefCell<Vec<PyObjectRef>>,
pub frozen: RefCell<HashMap<String, bytecode::FrozenModule>>,
pub import_func: RefCell<PyObjectRef>,
pub profile_func: RefCell<PyObjectRef>,
pub trace_func: RefCell<PyObjectRef>,
pub use_tracing: RefCell<bool>,
pub signal_handlers: RefCell<[PyObjectRef; NSIG]>,
pub settings: PySettings,
}
pub const NSIG: usize = 64;
pub struct PySettings {
pub debug: bool,
pub inspect: bool,
pub optimize: u8,
pub no_user_site: bool,
pub no_site: bool,
pub ignore_environment: bool,
pub verbose: u8,
pub quiet: bool,
pub dont_write_bytecode: bool,
pub path_list: Vec<String>,
pub argv: Vec<String>,
}
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"),
}
}
}
impl Default for PySettings {
fn default() -> Self {
PySettings {
debug: false,
inspect: false,
optimize: 0,
no_user_site: false,
no_site: false,
ignore_environment: false,
verbose: 0,
quiet: false,
dont_write_bytecode: false,
path_list: vec![],
argv: vec![],
}
}
}
impl VirtualMachine {
pub fn new(settings: PySettings) -> VirtualMachine {
flame_guard!("init VirtualMachine");
let ctx = PyContext::new();
let new_module = |name: &str, dict| {
PyObject::new(
PyModule {
name: name.to_owned(),
},
ctx.types.module_type.clone(),
Some(dict),
)
};
let builtins_dict = ctx.new_dict();
let builtins = new_module("builtins", builtins_dict.clone());
let sysmod_dict = ctx.new_dict();
let sysmod = new_module("sys", sysmod_dict.clone());
let stdlib_inits = RefCell::new(stdlib::get_module_inits());
let frozen = RefCell::new(frozen::get_module_inits());
let import_func = RefCell::new(ctx.none());
let profile_func = RefCell::new(ctx.none());
let trace_func = RefCell::new(ctx.none());
let signal_handlers = RefCell::new(arr![ctx.none(); 64]);
let vm = VirtualMachine {
builtins: builtins.clone(),
sys_module: sysmod.clone(),
stdlib_inits,
ctx,
frames: RefCell::new(vec![]),
wasm_id: None,
exceptions: RefCell::new(vec![]),
frozen,
import_func,
profile_func,
trace_func,
use_tracing: RefCell::new(false),
signal_handlers,
settings,
};
objmodule::init_module_dict(
&vm,
&builtins_dict,
vm.new_str("builtins".to_owned()),
vm.get_none(),
);
objmodule::init_module_dict(
&vm,
&sysmod_dict,
vm.new_str("sys".to_owned()),
vm.get_none(),
);
builtins::make_module(&vm, builtins.clone());
sysmodule::make_module(&vm, sysmod, builtins);
#[cfg(not(target_arch = "wasm32"))]
import::import_builtin(&vm, "signal").expect("Couldn't initialize signal module");
vm
}
pub fn run_code_obj(&self, code: PyCodeRef, scope: Scope) -> PyResult {
let frame = Frame::new(code, scope).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 run_frame(&self, frame: FrameRef) -> PyResult<ExecutionResult> {
self.frames.borrow_mut().push(frame.clone());
let result = frame.run(self);
self.frames.borrow_mut().pop();
result
}
pub fn frame_throw(
&self,
frame: FrameRef,
exception: PyObjectRef,
) -> PyResult<ExecutionResult> {
self.frames.borrow_mut().push(frame.clone());
let result = frame.throw(self, exception);
self.frames.borrow_mut().pop();
result
}
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_scope(&self) -> Ref<Scope> {
let frame = self
.current_frame()
.expect("called current_scope but no frames on the stack");
Ref::map(frame, |f| &f.scope)
}
pub fn try_class(&self, module: &str, class: &str) -> PyResult<PyClassRef> {
let class = self
.get_attribute(self.import(module, &[], 0)?, class)?
.downcast()
.expect("not a class");
Ok(class)
}
pub fn class(&self, module: &str, class: &str) -> PyClassRef {
let module = self
.import(module, &[], 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")
}
pub fn new_str(&self, s: String) -> PyObjectRef {
self.ctx.new_str(s)
}
pub fn new_int<T: Into<BigInt>>(&self, i: T) -> PyObjectRef {
self.ctx.new_int(i)
}
pub fn new_bool(&self, b: bool) -> PyObjectRef {
self.ctx.new_bool(b)
}
pub fn new_module(&self, name: &str, dict: PyDictRef) -> PyObjectRef {
objmodule::init_module_dict(self, &dict, self.new_str(name.to_owned()), self.get_none());
PyObject::new(
PyModule {
name: name.to_owned(),
},
self.ctx.types.module_type.clone(),
Some(dict),
)
}
#[cfg_attr(feature = "flame-it", flame("VirtualMachine"))]
fn new_exception_obj(&self, exc_type: PyClassRef, args: Vec<PyObjectRef>) -> PyResult {
vm_trace!("New exception created: {}", exc_type.name);
self.invoke(&exc_type.into_object(), args)
}
pub fn new_empty_exception(&self, exc_type: PyClassRef) -> PyResult {
self.new_exception_obj(exc_type, vec![])
}
pub fn new_exception(&self, exc_type: PyClassRef, msg: String) -> PyObjectRef {
let pystr_msg = self.new_str(msg);
self.new_exception_obj(exc_type, vec![pystr_msg]).unwrap()
}
pub fn new_lookup_error(&self, msg: String) -> PyObjectRef {
let lookup_error = self.ctx.exceptions.lookup_error.clone();
self.new_exception(lookup_error, msg)
}
pub fn new_attribute_error(&self, msg: String) -> PyObjectRef {
let attribute_error = self.ctx.exceptions.attribute_error.clone();
self.new_exception(attribute_error, msg)
}
pub fn new_type_error(&self, msg: String) -> PyObjectRef {
let type_error = self.ctx.exceptions.type_error.clone();
self.new_exception(type_error, msg)
}
pub fn new_name_error(&self, msg: String) -> PyObjectRef {
let name_error = self.ctx.exceptions.name_error.clone();
self.new_exception(name_error, msg)
}
pub fn new_unsupported_operand_error(
&self,
a: PyObjectRef,
b: PyObjectRef,
op: &str,
) -> PyObjectRef {
self.new_type_error(format!(
"Unsupported operand types for '{}': '{}' and '{}'",
op,
a.class().name,
b.class().name
))
}
pub fn new_os_error(&self, msg: String) -> PyObjectRef {
let os_error = self.ctx.exceptions.os_error.clone();
self.new_exception(os_error, msg)
}
pub fn new_unicode_decode_error(&self, msg: String) -> PyObjectRef {
let unicode_decode_error = self.ctx.exceptions.unicode_decode_error.clone();
self.new_exception(unicode_decode_error, msg)
}
pub fn new_value_error(&self, msg: String) -> PyObjectRef {
let value_error = self.ctx.exceptions.value_error.clone();
self.new_exception(value_error, msg)
}
pub fn new_key_error(&self, obj: PyObjectRef) -> PyObjectRef {
let key_error = self.ctx.exceptions.key_error.clone();
self.new_exception_obj(key_error, vec![obj]).unwrap()
}
pub fn new_index_error(&self, msg: String) -> PyObjectRef {
let index_error = self.ctx.exceptions.index_error.clone();
self.new_exception(index_error, msg)
}
pub fn new_not_implemented_error(&self, msg: String) -> PyObjectRef {
let not_implemented_error = self.ctx.exceptions.not_implemented_error.clone();
self.new_exception(not_implemented_error, msg)
}
pub fn new_zero_division_error(&self, msg: String) -> PyObjectRef {
let zero_division_error = self.ctx.exceptions.zero_division_error.clone();
self.new_exception(zero_division_error, msg)
}
pub fn new_overflow_error(&self, msg: String) -> PyObjectRef {
let overflow_error = self.ctx.exceptions.overflow_error.clone();
self.new_exception(overflow_error, msg)
}
#[cfg(feature = "rustpython-compiler")]
pub fn new_syntax_error(&self, error: &CompileError) -> PyObjectRef {
let syntax_error_type = self.ctx.exceptions.syntax_error.clone();
let syntax_error = self.new_exception(syntax_error_type, error.to_string());
let lineno = self.new_int(error.location.row());
self.set_attr(&syntax_error, "lineno", lineno).unwrap();
syntax_error
}
pub fn new_import_error(&self, msg: String) -> PyObjectRef {
let import_error = self.ctx.exceptions.import_error.clone();
self.new_exception(import_error, msg)
}
pub fn new_scope_with_builtins(&self) -> Scope {
Scope::with_builtins(None, self.ctx.new_dict(), self)
}
pub fn get_none(&self) -> PyObjectRef {
self.ctx.none()
}
pub fn is_none(&self, obj: &PyObjectRef) -> bool {
obj.is(&self.get_none())
}
pub fn get_type(&self) -> PyClassRef {
self.ctx.type_type()
}
pub fn get_object(&self) -> PyClassRef {
self.ctx.object()
}
pub fn get_locals(&self) -> PyDictRef {
self.current_scope().get_locals().clone()
}
pub fn context(&self) -> &PyContext {
&self.ctx
}
pub fn to_str(&self, obj: &PyObjectRef) -> PyResult<PyStringRef> {
let str = self.call_method(&obj, "__str__", vec![])?;
TryFromObject::try_from_object(self, str)
}
pub fn to_pystr<'a, T: Into<&'a PyObjectRef>>(&'a self, obj: T) -> Result<String, PyObjectRef> {
let py_str_obj = self.to_str(obj.into())?;
Ok(py_str_obj.as_str().to_owned())
}
pub fn to_repr(&self, obj: &PyObjectRef) -> PyResult<PyStringRef> {
let repr = self.call_method(obj, "__repr__", vec![])?;
TryFromObject::try_from_object(self, repr)
}
pub fn to_ascii(&self, obj: &PyObjectRef) -> PyResult {
let repr = self.call_method(obj, "__repr__", vec![])?;
let repr: PyStringRef = TryFromObject::try_from_object(self, repr)?;
let ascii = to_ascii(repr.as_str());
Ok(self.new_str(ascii))
}
pub fn import(&self, module: &str, from_list: &[String], level: usize) -> PyResult {
let weird = module.contains('.') || level != 0 || !from_list.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, self).ok()
};
match cached_module {
Some(module) => Ok(module),
None => {
let import_func = self
.get_attribute(self.builtins.clone(), "__import__")
.map_err(|_| self.new_import_error("__import__ not found".to_string()))?;
let (locals, globals) = if let Some(frame) = self.current_frame() {
(
frame.scope.get_locals().into_object(),
frame.scope.globals.clone().into_object(),
)
} else {
(self.get_none(), self.get_none())
};
let from_list = self.ctx.new_tuple(
from_list
.iter()
.map(|name| self.new_str(name.to_string()))
.collect(),
);
self.invoke(
&import_func,
vec![
self.new_str(module.to_owned()),
globals,
locals,
from_list,
self.ctx.new_int(level),
],
)
.map_err(|exc| import::remove_importlib_frames(self, &exc))
}
}
}
pub fn isinstance(&self, obj: &PyObjectRef, cls: &PyClassRef) -> PyResult<bool> {
if Rc::ptr_eq(&obj.class().into_object(), cls.as_object()) {
Ok(true)
} else {
let ret = self.call_method(cls.as_object(), "__instancecheck__", vec![obj.clone()])?;
objbool::boolval(self, ret)
}
}
pub fn issubclass(&self, subclass: &PyClassRef, cls: &PyClassRef) -> PyResult<bool> {
let ret = self.call_method(
cls.as_object(),
"__subclasscheck__",
vec![subclass.clone().into_object()],
)?;
objbool::boolval(self, ret)
}
pub fn call_get_descriptor(&self, attr: PyObjectRef, obj: PyObjectRef) -> PyResult {
let attr_class = attr.class();
if let Some(ref descriptor) = objtype::class_get_attr(&attr_class, "__get__") {
let cls = obj.class();
self.invoke(descriptor, vec![attr, obj.clone(), cls.into_object()])
} else {
Ok(attr)
}
}
pub fn call_method<T>(&self, obj: &PyObjectRef, method_name: &str, args: T) -> PyResult
where
T: Into<PyFuncArgs>,
{
flame_guard!(format!("call_method({:?})", method_name));
let cls = obj.class();
match objtype::class_get_attr(&cls, method_name) {
Some(func) => {
vm_trace!(
"vm.call_method {:?} {:?} {:?} -> {:?}",
obj,
cls,
method_name,
func
);
let wrapped = self.call_get_descriptor(func, obj.clone())?;
self.invoke(&wrapped, args)
}
None => Err(self.new_type_error(format!("Unsupported method: {}", method_name))),
}
}
fn _invoke(&self, func_ref: &PyObjectRef, args: PyFuncArgs) -> PyResult {
vm_trace!("Invoke: {:?} {:?}", func_ref, args);
if let Some(PyFunction {
ref code,
ref scope,
ref defaults,
ref kw_only_defaults,
}) = func_ref.payload()
{
self.trace_event(TraceEvent::Call)?;
let res = self.invoke_python_function(code, scope, defaults, kw_only_defaults, args);
self.trace_event(TraceEvent::Return)?;
res
} else if let Some(PyMethod {
ref function,
ref object,
}) = func_ref.payload()
{
self.invoke(&function, args.insert(object.clone()))
} else if let Some(PyBuiltinFunction { ref value }) = func_ref.payload() {
value(self, args)
} else {
vm_trace!("invoke __call__ for: {:?}", &func_ref.payload);
self.call_method(&func_ref, "__call__", args)
}
}
#[inline]
pub fn invoke<T>(&self, func_ref: &PyObjectRef, args: T) -> PyResult
where
T: Into<PyFuncArgs>,
{
let res = self._invoke(func_ref, args.into());
res
}
fn trace_event(&self, event: TraceEvent) -> PyResult<()> {
if *self.use_tracing.borrow() {
let frame = self.get_none();
let event = self.new_str(event.to_string());
let arg = self.get_none();
let args = vec![frame, event, arg];
let trace_func = self.trace_func.borrow().clone();
if !self.is_none(&trace_func) {
self.use_tracing.replace(false);
let res = self.invoke(&trace_func, args.clone());
self.use_tracing.replace(true);
res?;
}
let profile_func = self.profile_func.borrow().clone();
if !self.is_none(&profile_func) {
self.use_tracing.replace(false);
let res = self.invoke(&profile_func, args);
self.use_tracing.replace(true);
res?;
}
}
Ok(())
}
fn invoke_python_function(
&self,
code: &PyCodeRef,
scope: &Scope,
defaults: &Option<PyTupleRef>,
kw_only_defaults: &Option<PyDictRef>,
func_args: PyFuncArgs,
) -> PyResult {
let scope = scope.new_child_scope(&self.ctx);
self.fill_locals_from_args(
&code.code,
&scope.get_locals(),
func_args,
defaults,
kw_only_defaults,
)?;
let frame = Frame::new(code.clone(), scope).into_ref(self);
if code.code.is_generator {
Ok(PyGenerator::new(frame, self).into_object())
} else {
self.run_frame_full(frame)
}
}
pub fn invoke_with_locals(
&self,
function: &PyObjectRef,
cells: PyDictRef,
locals: PyDictRef,
) -> PyResult {
if let Some(PyFunction { code, scope, .. }) = &function.payload() {
let scope = scope
.new_child_scope_with_locals(cells)
.new_child_scope_with_locals(locals);
let frame = Frame::new(code.clone(), scope).into_ref(self);
return self.run_frame_full(frame);
}
panic!(
"invoke_with_locals: expected python function, got: {:?}",
*function
);
}
fn fill_locals_from_args(
&self,
code_object: &bytecode::CodeObject,
locals: &PyDictRef,
func_args: PyFuncArgs,
defaults: &Option<PyTupleRef>,
kw_only_defaults: &Option<PyDictRef>,
) -> PyResult<()> {
let nargs = func_args.args.len();
let nexpected_args = code_object.arg_names.len();
let n = if nargs > nexpected_args {
nexpected_args
} else {
nargs
};
for i in 0..n {
let arg_name = &code_object.arg_names[i];
let arg = &func_args.args[i];
locals.set_item(arg_name, arg.clone(), self)?;
}
match code_object.varargs {
bytecode::Varargs::Named(ref vararg_name) => {
let mut last_args = vec![];
for i in n..nargs {
let arg = &func_args.args[i];
last_args.push(arg.clone());
}
let vararg_value = self.ctx.new_tuple(last_args);
locals.set_item(vararg_name, vararg_value, self)?;
}
bytecode::Varargs::Unnamed | bytecode::Varargs::None => {
if nargs > nexpected_args {
return Err(self.new_type_error(format!(
"Expected {} arguments (got: {})",
nexpected_args, nargs
)));
}
}
}
let kwargs = match code_object.varkeywords {
bytecode::Varargs::Named(ref kwargs_name) => {
let d = self.ctx.new_dict();
locals.set_item(kwargs_name, d.as_object().clone(), self)?;
Some(d)
}
bytecode::Varargs::Unnamed => Some(self.ctx.new_dict()),
bytecode::Varargs::None => None,
};
for (name, value) in func_args.kwargs {
if code_object.arg_names.contains(&name) || code_object.kwonlyarg_names.contains(&name)
{
if locals.contains_key(&name, self) {
return Err(
self.new_type_error(format!("Got multiple values for argument '{}'", name))
);
}
locals.set_item(&name, value, self)?;
} else if let Some(d) = &kwargs {
d.set_item(&name, value, self)?;
} else {
return Err(
self.new_type_error(format!("Got an unexpected keyword argument '{}'", name))
);
}
}
if nargs < nexpected_args {
let num_defaults_available = defaults.as_ref().map_or(0, |d| d.elements.len());
let required_args = nexpected_args - num_defaults_available;
let mut missing = vec![];
for i in 0..required_args {
let variable_name = &code_object.arg_names[i];
if !locals.contains_key(variable_name, self) {
missing.push(variable_name)
}
}
if !missing.is_empty() {
return Err(self.new_type_error(format!(
"Missing {} required positional arguments: {:?}",
missing.len(),
missing
)));
}
if let Some(defaults) = defaults {
let defaults = &defaults.elements;
for (default_index, i) in (required_args..nexpected_args).enumerate() {
let arg_name = &code_object.arg_names[i];
if !locals.contains_key(arg_name, self) {
locals.set_item(arg_name, defaults[default_index].clone(), self)?;
}
}
}
};
for arg_name in &code_object.kwonlyarg_names {
if !locals.contains_key(arg_name, self) {
if let Some(kw_only_defaults) = kw_only_defaults {
if let Some(default) = kw_only_defaults.get_item_option(arg_name, self)? {
locals.set_item(arg_name, default, self)?;
continue;
}
}
return Err(self
.new_type_error(format!("Missing required kw only argument: '{}'", arg_name)));
}
}
Ok(())
}
pub fn extract_elements(&self, value: &PyObjectRef) -> PyResult<Vec<PyObjectRef>> {
let elements = if objtype::isinstance(value, &self.ctx.tuple_type()) {
objsequence::get_elements_tuple(value).to_vec()
} else if objtype::isinstance(value, &self.ctx.list_type()) {
objsequence::get_elements_list(value).to_vec()
} else {
let iter = objiter::get_iter(self, value)?;
objiter::get_all(self, &iter)?
};
Ok(elements)
}
#[cfg_attr(feature = "flame-it", flame("VirtualMachine"))]
pub fn get_attribute<T>(&self, obj: PyObjectRef, attr_name: T) -> PyResult
where
T: TryIntoRef<PyString>,
{
let attr_name = attr_name.try_into_ref(self)?;
vm_trace!("vm.__getattribute__: {:?} {:?}", obj, attr_name);
self.call_method(&obj, "__getattribute__", vec![attr_name.into_object()])
}
pub fn set_attr<K, V>(&self, obj: &PyObjectRef, attr_name: K, attr_value: V) -> PyResult
where
K: TryIntoRef<PyString>,
V: Into<PyObjectRef>,
{
let attr_name = attr_name.try_into_ref(self)?;
self.call_method(
obj,
"__setattr__",
vec![attr_name.into_object(), attr_value.into()],
)
}
pub fn del_attr(&self, obj: &PyObjectRef, attr_name: PyObjectRef) -> PyResult<()> {
self.call_method(&obj, "__delattr__", vec![attr_name])?;
Ok(())
}
pub fn get_method_or_type_error<F>(
&self,
obj: PyObjectRef,
method_name: &str,
err_msg: F,
) -> PyResult
where
F: FnOnce() -> String,
{
let cls = obj.class();
match objtype::class_get_attr(&cls, method_name) {
Some(method) => self.call_get_descriptor(method, obj.clone()),
None => Err(self.new_type_error(err_msg())),
}
}
pub fn get_method(&self, obj: PyObjectRef, method_name: &str) -> Option<PyResult> {
let cls = obj.class();
let method = objtype::class_get_attr(&cls, method_name)?;
Some(self.call_get_descriptor(method, obj.clone()))
}
pub fn call_or_unsupported<F>(
&self,
obj: PyObjectRef,
arg: PyObjectRef,
method: &str,
unsupported: F,
) -> PyResult
where
F: Fn(&VirtualMachine, PyObjectRef, PyObjectRef) -> PyResult,
{
if let Some(method_or_err) = self.get_method(obj.clone(), method) {
let method = method_or_err?;
let result = self.invoke(&method, vec![arg.clone()])?;
if !result.is(&self.ctx.not_implemented()) {
return Ok(result);
}
}
unsupported(self, obj, arg)
}
pub fn call_or_reflection(
&self,
lhs: PyObjectRef,
rhs: PyObjectRef,
default: &str,
reflection: &str,
unsupported: fn(&VirtualMachine, PyObjectRef, PyObjectRef) -> PyResult,
) -> PyResult {
self.call_or_unsupported(lhs, rhs, default, move |vm, lhs, rhs| {
vm.call_or_unsupported(rhs, lhs, reflection, unsupported)
})
}
pub fn generic_getattribute(
&self,
obj: PyObjectRef,
name_str: PyStringRef,
) -> PyResult<Option<PyObjectRef>> {
let name = name_str.as_str();
let cls = obj.class();
if let Some(attr) = objtype::class_get_attr(&cls, &name) {
let attr_class = attr.class();
if objtype::class_has_attr(&attr_class, "__set__") {
if let Some(descriptor) = objtype::class_get_attr(&attr_class, "__get__") {
return self
.invoke(&descriptor, vec![attr, obj, cls.into_object()])
.map(Some);
}
}
}
let attr = if let Some(ref dict) = obj.dict {
dict.get_item_option(name_str.as_str(), self)?
} else {
None
};
if let Some(obj_attr) = attr {
Ok(Some(obj_attr))
} else if let Some(attr) = objtype::class_get_attr(&cls, &name) {
self.call_get_descriptor(attr, obj).map(Some)
} else if let Some(getter) = objtype::class_get_attr(&cls, "__getattr__") {
self.invoke(&getter, vec![obj, name_str.into_object()])
.map(Some)
} else {
Ok(None)
}
}
pub fn is_callable(&self, obj: &PyObjectRef) -> bool {
match_class!(match obj {
PyFunction => true,
PyMethod => true,
PyBuiltinFunction => true,
obj => objtype::class_has_attr(&obj.class(), "__call__"),
})
}
#[cfg(feature = "rustpython-compiler")]
pub fn compile(
&self,
source: &str,
mode: compile::Mode,
source_path: String,
) -> Result<PyCodeRef, CompileError> {
compile::compile(source, mode, source_path, self.settings.optimize)
.map(|codeobj| PyCode::new(codeobj).into_ref(self))
}
pub fn _sub(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_reflection(a, b, "__sub__", "__rsub__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "-"))
})
}
pub fn _isub(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_unsupported(a, b, "__isub__", |vm, a, b| {
vm.call_or_reflection(a, b, "__sub__", "__rsub__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "-="))
})
})
}
pub fn _add(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_reflection(a, b, "__add__", "__radd__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "+"))
})
}
pub fn _iadd(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_unsupported(a, b, "__iadd__", |vm, a, b| {
vm.call_or_reflection(a, b, "__add__", "__radd__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "+="))
})
})
}
pub fn _mul(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_reflection(a, b, "__mul__", "__rmul__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "*"))
})
}
pub fn _imul(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_unsupported(a, b, "__imul__", |vm, a, b| {
vm.call_or_reflection(a, b, "__mul__", "__rmul__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "*="))
})
})
}
pub fn _matmul(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_reflection(a, b, "__matmul__", "__rmatmul__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "@"))
})
}
pub fn _imatmul(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_unsupported(a, b, "__imatmul__", |vm, a, b| {
vm.call_or_reflection(a, b, "__matmul__", "__rmatmul__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "@="))
})
})
}
pub fn _truediv(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_reflection(a, b, "__truediv__", "__rtruediv__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "/"))
})
}
pub fn _itruediv(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_unsupported(a, b, "__itruediv__", |vm, a, b| {
vm.call_or_reflection(a, b, "__truediv__", "__rtruediv__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "/="))
})
})
}
pub fn _floordiv(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_reflection(a, b, "__floordiv__", "__rfloordiv__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "//"))
})
}
pub fn _ifloordiv(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_unsupported(a, b, "__ifloordiv__", |vm, a, b| {
vm.call_or_reflection(a, b, "__floordiv__", "__rfloordiv__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "//="))
})
})
}
pub fn _pow(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_reflection(a, b, "__pow__", "__rpow__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "**"))
})
}
pub fn _ipow(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_unsupported(a, b, "__ipow__", |vm, a, b| {
vm.call_or_reflection(a, b, "__pow__", "__rpow__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "**="))
})
})
}
pub fn _mod(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_reflection(a, b, "__mod__", "__rmod__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "%"))
})
}
pub fn _imod(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_unsupported(a, b, "__imod__", |vm, a, b| {
vm.call_or_reflection(a, b, "__mod__", "__rmod__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "%="))
})
})
}
pub fn _lshift(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_reflection(a, b, "__lshift__", "__rlshift__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "<<"))
})
}
pub fn _ilshift(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_unsupported(a, b, "__ilshift__", |vm, a, b| {
vm.call_or_reflection(a, b, "__lshift__", "__rlshift__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "<<="))
})
})
}
pub fn _rshift(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_reflection(a, b, "__rshift__", "__rrshift__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, ">>"))
})
}
pub fn _irshift(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_unsupported(a, b, "__irshift__", |vm, a, b| {
vm.call_or_reflection(a, b, "__rshift__", "__rrshift__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, ">>="))
})
})
}
pub fn _xor(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_reflection(a, b, "__xor__", "__rxor__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "^"))
})
}
pub fn _ixor(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_unsupported(a, b, "__ixor__", |vm, a, b| {
vm.call_or_reflection(a, b, "__xor__", "__rxor__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "^="))
})
})
}
pub fn _or(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_reflection(a, b, "__or__", "__ror__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "|"))
})
}
pub fn _ior(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_unsupported(a, b, "__ior__", |vm, a, b| {
vm.call_or_reflection(a, b, "__or__", "__ror__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "|="))
})
})
}
pub fn _and(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_reflection(a, b, "__and__", "__rand__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "&"))
})
}
pub fn _iand(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_unsupported(a, b, "__iand__", |vm, a, b| {
vm.call_or_reflection(a, b, "__and__", "__rand__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "&="))
})
})
}
pub fn _eq(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_reflection(a, b, "__eq__", "__eq__", |vm, a, b| {
Ok(vm.new_bool(a.is(&b)))
})
}
pub fn _ne(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_reflection(a, b, "__ne__", "__ne__", |vm, a, b| {
let eq = vm._eq(a, b)?;
objbool::not(vm, &eq)
})
}
pub fn _lt(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_reflection(a, b, "__lt__", "__gt__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "<"))
})
}
pub fn _le(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_reflection(a, b, "__le__", "__ge__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, "<="))
})
}
pub fn _gt(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_reflection(a, b, "__gt__", "__lt__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, ">"))
})
}
pub fn _ge(&self, a: PyObjectRef, b: PyObjectRef) -> PyResult {
self.call_or_reflection(a, b, "__ge__", "__le__", |vm, a, b| {
Err(vm.new_unsupported_operand_error(a, b, ">="))
})
}
pub fn _hash(&self, obj: &PyObjectRef) -> PyResult<pyhash::PyHash> {
let hash_obj = self.call_method(obj, "__hash__", vec![])?;
if objtype::isinstance(&hash_obj, &self.ctx.int_type()) {
Ok(hash_obj.payload::<PyInt>().unwrap().hash(self))
} else {
Err(self.new_type_error("__hash__ method should return an integer".to_string()))
}
}
fn _membership_iter_search(&self, haystack: PyObjectRef, needle: PyObjectRef) -> PyResult {
let iter = objiter::get_iter(self, &haystack)?;
loop {
if let Some(element) = objiter::get_next_object(self, &iter)? {
let equal = self._eq(needle.clone(), element.clone())?;
if objbool::get_value(&equal) {
return Ok(self.new_bool(true));
} else {
continue;
}
} else {
return Ok(self.new_bool(false));
}
}
}
pub fn _membership(&self, haystack: PyObjectRef, needle: PyObjectRef) -> PyResult {
if let Some(method_or_err) = self.get_method(haystack.clone(), "__contains__") {
let method = method_or_err?;
self.invoke(&method, vec![needle])
} else {
self._membership_iter_search(haystack, needle)
}
}
pub fn push_exception(&self, exc: PyObjectRef) {
self.exceptions.borrow_mut().push(exc)
}
pub fn pop_exception(&self) -> Option<PyObjectRef> {
self.exceptions.borrow_mut().pop()
}
pub fn current_exception(&self) -> Option<PyObjectRef> {
self.exceptions.borrow().last().cloned()
}
}
impl Default for VirtualMachine {
fn default() -> Self {
VirtualMachine::new(Default::default())
}
}
lazy_static! {
static ref REPR_GUARDS: Mutex<HashSet<usize>> = { Mutex::new(HashSet::new()) };
}
pub struct ReprGuard {
id: usize,
}
impl ReprGuard {
fn get_guards<'a>() -> MutexGuard<'a, HashSet<usize>> {
REPR_GUARDS.lock().expect("ReprGuard lock poisoned")
}
pub fn enter(obj: &PyObjectRef) -> Option<ReprGuard> {
let mut guards = ReprGuard::get_guards();
let id = obj.get_id();
if guards.contains(&id) {
return None;
}
guards.insert(id);
Some(ReprGuard { id })
}
}
impl Drop for ReprGuard {
fn drop(&mut self) {
ReprGuard::get_guards().remove(&self.id);
}
}
#[cfg(test)]
mod tests {
use super::VirtualMachine;
use crate::obj::{objint, objstr};
use num_bigint::ToBigInt;
#[test]
fn test_add_py_integers() {
let vm: VirtualMachine = Default::default();
let a = vm.ctx.new_int(33_i32);
let b = vm.ctx.new_int(12_i32);
let res = vm._add(a, b).unwrap();
let value = objint::get_value(&res);
assert_eq!(*value, 45_i32.to_bigint().unwrap());
}
#[test]
fn test_multiply_str() {
let vm: VirtualMachine = Default::default();
let a = vm.ctx.new_str(String::from("Hello "));
let b = vm.ctx.new_int(4_i32);
let res = vm._mul(a, b).unwrap();
let value = objstr::get_value(&res);
assert_eq!(value, String::from("Hello Hello Hello Hello "))
}
}