use std::{borrow::Cow, fmt::Write, mem};
use super::{Dict, LazyHeapSet, PyTrait, Type};
use crate::{
args::{ArgValues, KwargsValues},
builtins::Builtins,
bytecode::{CallResult, VM},
defer_drop,
exception_private::{ExcType, RunResult},
hash::{HashValue, identity_hash},
heap::{
BorrowedHeapReadMut, DropWithHeap, Heap, HeapData, HeapId, HeapItem, HeapRead, HeapReadOutput,
heap_read_ref_as_field_mut,
},
intern::Interns,
resource::ResourceTracker,
types::allocate_string,
value::{EitherStr, Value},
};
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub(crate) struct Instance {
class: HeapId,
attrs: Dict,
}
impl Instance {
#[must_use]
pub fn new(class: HeapId, attrs: Dict) -> Self {
Self { class, attrs }
}
#[must_use]
pub fn class(&self) -> HeapId {
self.class
}
#[must_use]
pub fn attrs(&self) -> &Dict {
&self.attrs
}
}
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub(crate) struct BoundMethod {
pub instance: Value,
pub func: Value,
}
impl<'h> HeapRead<'h, Instance> {
fn attrs_mut(&mut self) -> BorrowedHeapReadMut<'_, 'h, Dict> {
heap_read_ref_as_field_mut!(self, Instance, attrs)
}
pub fn set_attr(
&mut self,
name: Value,
value: Value,
vm: &mut VM<'h, impl ResourceTracker>,
) -> RunResult<Option<Value>> {
self.attrs_mut().set(name, value, vm)
}
}
impl<'h> PyTrait<'h> for HeapRead<'h, Instance> {
fn py_type(&self, vm: &VM<'h, impl ResourceTracker>) -> Type {
Type::Instance(self.get(vm.heap).class)
}
fn py_len(&self, _vm: &VM<'h, impl ResourceTracker>) -> Option<usize> {
None
}
fn py_eq_impl(&self, _other: &Value, _vm: &mut VM<'h, impl ResourceTracker>) -> RunResult<Option<bool>> {
Ok(None)
}
fn py_hash(&self, self_id: HeapId, _vm: &mut VM<'h, impl ResourceTracker>) -> RunResult<Option<HashValue>> {
Ok(Some(identity_hash(self_id)))
}
fn py_repr_fmt(
&self,
f: &mut impl Write,
vm: &mut VM<'h, impl ResourceTracker>,
_heap_ids: &mut LazyHeapSet,
) -> RunResult<()> {
let class_id = self.get(vm.heap).class;
Ok(write!(f, "<{} object>", class_name(class_id, vm.heap, vm.interns))?)
}
fn py_call_attr(
&mut self,
self_id: HeapId,
vm: &mut VM<'h, impl ResourceTracker>,
attr: &EitherStr,
args: ArgValues,
) -> RunResult<CallResult> {
let attr_str = attr.as_str(vm.interns);
if let Some(callable) = self
.get(vm.heap)
.attrs
.get_by_str(attr_str, vm.heap, vm.interns)
.map(|v| v.clone_with_heap(vm.heap))
{
defer_drop!(callable, vm);
return vm.call_function(callable, args);
}
let class_id = self.get(vm.heap).class;
if let Some(member) = class_member(class_id, attr_str, vm) {
defer_drop!(member, vm);
return call_member_bound(member, self_id, args, vm);
}
if attr_str == "__class__" {
vm.heap.inc_ref(class_id);
let class_val = Value::Ref(class_id);
defer_drop!(class_val, vm);
return vm.call_function(class_val, args);
}
args.drop_with_heap(vm);
Err(ExcType::attribute_error(
class_name(class_id, vm.heap, vm.interns),
attr_str,
))
}
fn py_is_context_manager(&self, vm: &VM<'h, impl ResourceTracker>) -> bool {
let class_id = self.get(vm.heap).class;
match vm.heap.get(class_id) {
HeapData::Class(class) => class.namespace().get_by_str("__exit__", vm.heap, vm.interns).is_some(),
_ => false,
}
}
fn py_enter(&mut self, self_id: HeapId, vm: &mut VM<'h, impl ResourceTracker>) -> RunResult<CallResult> {
let class_id = self.get(vm.heap).class;
let Some(enter) = class_member(class_id, "__enter__", vm) else {
return Err(ExcType::type_error_not_context_manager(
class_name(class_id, vm.heap, vm.interns),
"__enter__",
));
};
defer_drop!(enter, vm);
call_member_bound(enter, self_id, ArgValues::Empty, vm)
}
fn py_exit(
&mut self,
self_id: HeapId,
vm: &mut VM<'h, impl ResourceTracker>,
exc: Option<HeapId>,
) -> RunResult<CallResult> {
let class_id = self.get(vm.heap).class;
let Some(exit) = class_member(class_id, "__exit__", vm) else {
return Err(ExcType::attribute_error(
class_name(class_id, vm.heap, vm.interns),
"__exit__",
));
};
defer_drop!(exit, vm);
let (typ, val) = match exc {
Some(exc_id) => {
let HeapData::Exception(e) = vm.heap.get(exc_id) else {
unreachable!("Instance py_exit called with a non-exception heap id");
};
vm.heap.inc_ref(exc_id);
(Value::Builtin(Builtins::ExcType(e.exc_type())), Value::Ref(exc_id))
}
None => (Value::None, Value::None),
};
let args = ArgValues::ArgsKargs {
args: vec![typ, val, Value::None],
kwargs: KwargsValues::Empty,
};
call_member_bound(exit, self_id, args, vm)
}
}
impl HeapItem for Instance {
fn py_estimate_size(&self) -> usize {
mem::size_of::<Self>() + self.attrs.py_estimate_size()
}
fn py_dec_ref_ids(&mut self, stack: &mut Vec<HeapId>) {
stack.push(self.class);
self.attrs.py_dec_ref_ids(stack);
}
}
impl<'h> PyTrait<'h> for HeapRead<'h, BoundMethod> {
fn py_type(&self, _vm: &VM<'h, impl ResourceTracker>) -> Type {
Type::Function
}
fn py_len(&self, _vm: &VM<'h, impl ResourceTracker>) -> Option<usize> {
None
}
fn py_eq_impl(&self, _other: &Value, _vm: &mut VM<'h, impl ResourceTracker>) -> RunResult<Option<bool>> {
Ok(None)
}
fn py_hash(&self, self_id: HeapId, _vm: &mut VM<'h, impl ResourceTracker>) -> RunResult<Option<HashValue>> {
Ok(Some(identity_hash(self_id)))
}
fn py_repr_fmt(
&self,
f: &mut impl Write,
_vm: &mut VM<'h, impl ResourceTracker>,
_heap_ids: &mut LazyHeapSet,
) -> RunResult<()> {
Ok(write!(f, "<bound method>")?)
}
}
impl HeapItem for BoundMethod {
fn py_estimate_size(&self) -> usize {
mem::size_of::<Self>()
}
fn py_dec_ref_ids(&mut self, stack: &mut Vec<HeapId>) {
self.instance.py_dec_ref_ids(stack);
self.func.py_dec_ref_ids(stack);
}
}
pub(crate) fn instance_getattr(
self_id: HeapId,
attr: &EitherStr,
vm: &mut VM<'_, impl ResourceTracker>,
) -> RunResult<CallResult> {
let attr_str = attr.as_str(vm.interns);
if let HeapReadOutput::Instance(inst) = vm.heap.read(self_id)
&& let Some(value) = inst
.get(vm.heap)
.attrs
.get_by_str(attr_str, vm.heap, vm.interns)
.map(|v| v.clone_with_heap(vm.heap))
{
return Ok(CallResult::Value(value));
}
let class_id = instance_class(self_id, vm);
if let Some(member) = class_member(class_id, attr_str, vm) {
if is_method_value(&member, vm) {
vm.heap.inc_ref(self_id);
let bound = BoundMethod {
instance: Value::Ref(self_id),
func: member,
};
let id = vm.heap.allocate(HeapData::BoundMethod(bound))?;
Ok(CallResult::Value(Value::Ref(id)))
} else {
Ok(CallResult::Value(member))
}
} else if attr_str == "__class__" {
vm.heap.inc_ref(class_id);
Ok(CallResult::Value(Value::Ref(class_id)))
} else {
Err(ExcType::attribute_error(
class_name(class_id, vm.heap, vm.interns),
attr_str,
))
}
}
pub(crate) fn instance_repr(self_id: HeapId, vm: &mut VM<'_, impl ResourceTracker>) -> RunResult<Value> {
match instance_call_str_dunder(self_id, "__repr__", vm)? {
Some(s) => Ok(s),
None => Ok(allocate_string(default_repr(self_id, vm), vm.heap)?),
}
}
pub(crate) fn instance_str(self_id: HeapId, vm: &mut VM<'_, impl ResourceTracker>) -> RunResult<Value> {
match instance_call_str_dunder(self_id, "__str__", vm)? {
Some(s) => Ok(s),
None => instance_repr(self_id, vm),
}
}
fn instance_call_str_dunder(
self_id: HeapId,
dunder: &'static str,
vm: &mut VM<'_, impl ResourceTracker>,
) -> RunResult<Option<Value>> {
let class_id = instance_class(self_id, vm);
let Some(func) = class_member(class_id, dunder, vm) else {
return Ok(None);
};
defer_drop!(func, vm);
let args = if is_method_value(func, vm) {
vm.heap.inc_ref(self_id);
ArgValues::One(Value::Ref(self_id))
} else {
ArgValues::Empty
};
let result = vm.evaluate_function(dunder, func, args)?;
if result.is_str(vm.heap) {
Ok(Some(result))
} else {
let exc = ExcType::type_error(format!(
"{dunder} returned non-string (type {})",
result.py_type_name(vm)
));
result.drop_with_heap(vm);
Err(exc)
}
}
fn default_repr(self_id: HeapId, vm: &mut VM<'_, impl ResourceTracker>) -> String {
let class_id = instance_class(self_id, vm);
format!(
"<{} object at 0x{:x}>",
class_name(class_id, vm.heap, vm.interns),
self_id.index()
)
}
fn instance_class(self_id: HeapId, vm: &VM<'_, impl ResourceTracker>) -> HeapId {
match vm.heap.get(self_id) {
HeapData::Instance(inst) => inst.class,
_ => unreachable!("instance_class called on non-instance heap value"),
}
}
fn class_member(class_id: HeapId, name: &str, vm: &VM<'_, impl ResourceTracker>) -> Option<Value> {
match vm.heap.get(class_id) {
HeapData::Class(class) => class
.namespace()
.get_by_str(name, vm.heap, vm.interns)
.map(|v| v.clone_with_heap(vm.heap)),
_ => None,
}
}
pub(crate) fn class_name<'i>(
class_id: HeapId,
heap: &Heap<impl ResourceTracker>,
interns: &'i Interns,
) -> Cow<'i, str> {
match heap.get(class_id) {
HeapData::Class(class) => match class.name() {
EitherStr::Interned(id) => Cow::Borrowed(interns.get_str(*id)),
EitherStr::Heap(s) => Cow::Owned(s.clone()),
},
_ => unreachable!("class_name called with a non-class heap id"),
}
}
fn call_member_bound(
member: &Value,
self_id: HeapId,
args: ArgValues,
vm: &mut VM<'_, impl ResourceTracker>,
) -> RunResult<CallResult> {
if is_method_value(member, vm) {
vm.heap.inc_ref(self_id);
vm.call_function(member, args.prepend(Value::Ref(self_id)))
} else {
vm.call_function(member, args)
}
}
fn is_method_value(value: &Value, vm: &VM<'_, impl ResourceTracker>) -> bool {
match value {
Value::DefFunction(_) => true,
Value::Ref(id) => matches!(vm.heap.get(*id), HeapData::Closure(_) | HeapData::FunctionDefaults(_)),
_ => false,
}
}