use std::{
borrow::Cow,
cmp::Ordering,
collections::hash_map::DefaultHasher,
fmt::{self, Write},
hash::{Hash, Hasher},
mem::{self, discriminant},
str::FromStr,
};
use num_bigint::BigInt;
use num_integer::Integer;
use num_traits::{FromPrimitive, ToPrimitive, Zero};
use smallvec::SmallVec;
use crate::{
builtins::Builtins,
bytecode::{CallResult, VM},
defer_drop,
exception_private::{ExcType, RunError, RunResult, SimpleException},
fstring::FormatFloat,
hash::{HashValue, hash_one, hash_python_long_int, hash_python_str},
heap::{ContainsHeap, DropWithHeap, Heap, HeapData, HeapGuard, HeapId, HeapReadOutput},
intern::{BytesId, FunctionId, Interns, LongIntId, StaticStrings, StringId},
modules::ModuleFunctions,
resource::{
ResourceError, ResourceTracker, check_div_size, check_lshift_size, check_mult_size, check_pow_size,
check_repeat_size,
},
types::{
Bytes, CmpOrder, LazyHeapSet, List, LongInt, Property, PyTrait, Type, allocate_tuple,
bytes::{bytes_repr_fmt, get_byte_at_index},
instance::{instance_getattr, instance_repr, instance_str},
long_int::{bigint_cmp_f64, check_bits_str_digits_limit, i64_cmp_f64},
path,
slice::slice_collect_iterator,
str::{allocate_char, allocate_string, concat_allocate_str, get_char_at_index, string_repr_fmt},
timedelta,
},
};
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub(crate) enum Value {
Undefined,
Ellipsis,
None,
Bool(bool),
Int(i64),
Float(f64),
InternString(StringId),
InternBytes(BytesId),
InternLongInt(LongIntId),
Builtin(Builtins),
ModuleFunction(ModuleFunctions),
DefFunction(FunctionId),
ExtFunction(StringId),
Marker(Marker),
Property(Property),
Ref(HeapId),
#[cfg(feature = "memory-model-checks")]
Dereferenced,
}
pub(crate) const VALUE_SIZE: usize = mem::size_of::<Value>();
#[cfg(feature = "memory-model-checks")]
impl Drop for Value {
fn drop(&mut self) {
if let Self::Ref(id) = self {
panic!("Value::Ref({id:?}) dropped without calling drop_with_heap() - this is a reference counting bug");
}
}
}
impl From<bool> for Value {
fn from(v: bool) -> Self {
Self::Bool(v)
}
}
impl<'h> PyTrait<'h> for Value {
fn py_type(&self, vm: &VM<'_, impl ResourceTracker>) -> Type {
match self {
Self::Undefined => panic!("Cannot get type of undefined value"),
Self::Ellipsis => Type::Ellipsis,
Self::None => Type::NoneType,
Self::Bool(_) => Type::Bool,
Self::Int(_) | Self::InternLongInt(_) => Type::Int,
Self::Float(_) => Type::Float,
Self::InternString(_) => Type::Str,
Self::InternBytes(_) => Type::Bytes,
Self::Builtin(c) => c.py_type(),
Self::ModuleFunction(_) => Type::BuiltinFunction,
Self::DefFunction(_) | Self::ExtFunction(_) => Type::Function,
Self::Marker(m) => m.py_type(),
Self::Property(_) => Type::Property,
Self::Ref(id) => vm.heap.read(*id).py_type(vm),
#[cfg(feature = "memory-model-checks")]
Self::Dereferenced => panic!("Cannot access Dereferenced object"),
}
}
fn py_len(&self, vm: &VM<'_, impl ResourceTracker>) -> Option<usize> {
match self {
Self::InternString(string_id) => Some(vm.interns.get_str(*string_id).chars().count()),
Self::InternBytes(bytes_id) => Some(vm.interns.get_bytes(*bytes_id).len()),
Self::Ref(id) => vm.heap.read(*id).py_len(vm),
_ => None,
}
}
fn py_eq_impl(&self, other: &Value, vm: &mut VM<'_, impl ResourceTracker>) -> RunResult<Option<bool>> {
match self {
Self::Undefined => Ok(Some(false)),
Self::None => Ok(matches!(other, Self::None).then_some(true)),
Self::Ellipsis => Ok(matches!(other, Self::Ellipsis).then_some(true)),
Self::Bool(b) => Ok(eq_i64(i64::from(*b), other, vm)),
Self::Int(a) => Ok(eq_i64(*a, other, vm)),
Self::Float(f) => Ok(eq_f64(*f, other, vm)),
Self::InternLongInt(id) => Ok(eq_bigint(vm.interns.get_long_int(*id), other, vm)),
Self::InternString(id) => Ok(match other {
Self::InternString(o) => Some(id == o),
_ => eq_str(vm.interns.get_str(*id), other, vm),
}),
Self::InternBytes(id) => Ok(match other {
Self::InternBytes(o) if id == o => Some(true),
_ => eq_bytes(vm.interns.get_bytes(*id), other, vm),
}),
Self::Builtin(b) => Ok(match other {
Self::Builtin(o) => Some(b == o),
_ => None,
}),
Self::ModuleFunction(mf) => Ok(match other {
Self::ModuleFunction(o) => Some(mf == o),
_ => None,
}),
Self::DefFunction(f) => Ok(match other {
Self::DefFunction(o) => Some(f == o),
_ => None,
}),
Self::ExtFunction(name_id) => Ok(eq_ext_function(vm.interns.get_str(*name_id), other, vm)),
Self::Marker(m) => Ok(match other {
Self::Marker(o) => Some(m == o),
_ => None,
}),
Self::Property(p) => Ok(match other {
Self::Property(o) => Some(p == o),
_ => None,
}),
Self::Ref(id) => {
if let Self::Ref(other_id) = other
&& id == other_id
{
Ok(Some(true))
} else {
vm.heap.read(*id).py_eq_impl(other, vm)
}
}
#[cfg(feature = "memory-model-checks")]
Self::Dereferenced => panic!("Cannot access Dereferenced object"),
}
}
fn py_cmp(&self, other: &Self, vm: &mut VM<'_, impl ResourceTracker>) -> RunResult<CmpOrder> {
let interns = vm.interns;
match (self, other) {
(Self::Int(s), Self::Int(o)) => Ok(CmpOrder::Ordered(s.cmp(o))),
(Self::Float(s), Self::Float(o)) => Ok(CmpOrder::from_numeric(s.partial_cmp(o))),
(Self::Int(s), Self::Float(o)) => Ok(CmpOrder::from_numeric(i64_cmp_f64(*s, *o))),
(Self::Float(s), Self::Int(o)) => Ok(CmpOrder::from_numeric(i64_cmp_f64(*o, *s).map(Ordering::reverse))),
(Self::Bool(s), _) => Self::Int(i64::from(*s)).py_cmp(other, vm),
(_, Self::Bool(s)) => self.py_cmp(&Self::Int(i64::from(*s)), vm),
(Self::Int(a), Self::Ref(id)) if let HeapData::LongInt(li) = vm.heap.get(*id) => {
Ok(CmpOrder::Ordered(BigInt::from(*a).cmp(li.inner())))
}
(Self::Ref(id), Self::Int(b)) if let HeapData::LongInt(li) = vm.heap.get(*id) => {
Ok(CmpOrder::Ordered(li.inner().cmp(&BigInt::from(*b))))
}
(Self::Float(s), Self::Ref(id)) if let HeapData::LongInt(li) = vm.heap.get(*id) => Ok(
CmpOrder::from_numeric(bigint_cmp_f64(li.inner(), *s).map(Ordering::reverse)),
),
(Self::Ref(id), Self::Float(o)) if let HeapData::LongInt(li) = vm.heap.get(*id) => {
Ok(CmpOrder::from_numeric(li.partial_cmp_f64(*o)))
}
(Self::Ref(id1), Self::Ref(id2)) => match (vm.heap.read(*id1), vm.heap.read(*id2)) {
(HeapReadOutput::LongInt(a), HeapReadOutput::LongInt(b)) => {
Ok(CmpOrder::Ordered(a.get(vm.heap).inner().cmp(b.get(vm.heap).inner())))
}
(HeapReadOutput::Str(a), HeapReadOutput::Str(b)) => {
Ok(CmpOrder::Ordered(a.get(vm.heap).as_str().cmp(b.get(vm.heap).as_str())))
}
(HeapReadOutput::Tuple(a), HeapReadOutput::Tuple(b)) => a.py_cmp(&b, vm),
(HeapReadOutput::List(a), HeapReadOutput::List(b)) => a.py_cmp(&b, vm),
(HeapReadOutput::Date(a), HeapReadOutput::Date(b)) => {
Ok(CmpOrder::from_total(a.get(vm.heap).partial_cmp(b.get(vm.heap))))
}
(HeapReadOutput::DateTime(a), HeapReadOutput::DateTime(b)) => a.py_cmp(&b, vm),
(HeapReadOutput::TimeDelta(a), HeapReadOutput::TimeDelta(b)) => {
Ok(CmpOrder::from_total(a.get(vm.heap).partial_cmp(b.get(vm.heap))))
}
_ => Ok(CmpOrder::Incomparable),
},
(Self::InternString(s1), Self::InternString(s2)) => {
Ok(CmpOrder::Ordered(interns.get_str(*s1).cmp(interns.get_str(*s2))))
}
(Self::InternString(s1), Self::Ref(id2)) if let HeapData::Str(s2) = vm.heap.get(*id2) => {
Ok(CmpOrder::Ordered(interns.get_str(*s1).cmp(s2.as_str())))
}
(Self::Ref(id1), Self::InternString(s2)) if let HeapData::Str(s1) = vm.heap.get(*id1) => {
Ok(CmpOrder::Ordered(s1.as_str().cmp(interns.get_str(*s2))))
}
(Self::InternBytes(b1), Self::InternBytes(b2)) => {
Ok(CmpOrder::Ordered(interns.get_bytes(*b1).cmp(interns.get_bytes(*b2))))
}
_ => Ok(CmpOrder::Incomparable),
}
}
fn py_bool(&self, vm: &mut VM<'_, impl ResourceTracker>) -> bool {
match self {
Self::Undefined => false,
Self::Ellipsis => true,
Self::None => false,
Self::Bool(b) => *b,
Self::Int(v) => *v != 0,
Self::Float(f) => *f != 0.0,
Self::InternLongInt(_) => true,
Self::Builtin(_) | Self::ModuleFunction(_) => true, Self::DefFunction(_) | Self::ExtFunction(_) => true, Self::Marker(_) => true, Self::Property(_) => true, Self::InternString(string_id) => !vm.interns.get_str(*string_id).is_empty(),
Self::InternBytes(bytes_id) => !vm.interns.get_bytes(*bytes_id).is_empty(),
Self::Ref(id) => vm.heap.read(*id).py_bool(vm),
#[cfg(feature = "memory-model-checks")]
Self::Dereferenced => panic!("Cannot access Dereferenced object"),
}
}
fn py_repr_fmt(
&self,
f: &mut impl Write,
vm: &mut VM<'_, impl ResourceTracker>,
heap_ids: &mut LazyHeapSet,
) -> RunResult<()> {
let interns = vm.interns;
match self {
Self::Undefined => Ok(f.write_str("Undefined")?),
Self::Ellipsis => Ok(f.write_str("Ellipsis")?),
Self::None => Ok(f.write_str("None")?),
Self::Bool(true) => Ok(f.write_str("True")?),
Self::Bool(false) => Ok(f.write_str("False")?),
Self::Int(v) => Ok(f.write_str(itoa::Buffer::new().format(*v))?),
Self::InternLongInt(long_int_id) => {
let bi = interns.get_long_int(*long_int_id);
check_bits_str_digits_limit(bi.bits())?;
Ok(write!(f, "{bi}")?)
}
Self::Float(v) => Ok(write!(f, "{}", FormatFloat(*v))?),
Self::Builtin(b) => Ok(b.py_repr_fmt(f)?),
Self::ModuleFunction(mf) => Ok(mf.py_repr_fmt(f, self.id(vm))?),
Self::DefFunction(f_id) => Ok(interns.get_function(*f_id).py_repr_fmt(f, interns, self.id(vm))?),
Self::ExtFunction(name_id) => Ok(write!(f, "<function '{}' external>", interns.get_str(*name_id))?),
Self::InternString(string_id) => Ok(string_repr_fmt(interns.get_str(*string_id), f)?),
Self::InternBytes(bytes_id) => Ok(bytes_repr_fmt(interns.get_bytes(*bytes_id), f)?),
Self::Marker(m) => Ok(m.py_repr_fmt(f)?),
Self::Property(p) => Ok(write!(f, "<property {p:?}>")?),
Self::Ref(id) => {
if heap_ids.contains(id) {
match vm.heap.get(*id) {
HeapData::List(_) => Ok(f.write_str("[...]")?),
HeapData::Tuple(_) => Ok(f.write_str("(...)")?),
HeapData::Dict(_) => Ok(f.write_str("{...}")?),
_ => Ok(f.write_str("...")?),
}
} else if matches!(vm.heap.get(*id), HeapData::Instance(_)) {
let str_value = instance_repr(*id, vm)?;
defer_drop!(str_value, vm);
Ok(f.write_str(str_value.to_str(vm)?)?)
} else {
heap_ids.insert(*id);
let result = vm.heap.read(*id).py_repr_fmt(f, vm, heap_ids);
heap_ids.remove(id);
result
}
}
#[cfg(feature = "memory-model-checks")]
Self::Dereferenced => panic!("Cannot access Dereferenced object"),
}
}
fn py_repr(&self, vm: &mut VM<'h, impl ResourceTracker>) -> RunResult<Value> {
match self {
Self::None => Ok(Self::InternString(StaticStrings::NoneRepr.into())),
Self::Bool(true) => Ok(Self::InternString(StaticStrings::TrueRepr.into())),
Self::Bool(false) => Ok(Self::InternString(StaticStrings::FalseRepr.into())),
Self::Ellipsis => Ok(Self::InternString(StaticStrings::EllipsisRepr.into())),
Self::Int(i) => Ok(allocate_string(itoa::Buffer::new().format(*i), vm.heap)?),
_ => {
let mut s = String::new();
let mut heap_ids = LazyHeapSet::default();
self.py_repr_fmt(&mut s, vm, &mut heap_ids)?;
Ok(allocate_string(s, vm.heap)?)
}
}
}
fn py_str(&self, vm: &mut VM<'h, impl ResourceTracker>) -> RunResult<Value> {
match self {
Self::InternString(string_id) => Ok(Self::InternString(*string_id)),
Self::Ref(id) if matches!(vm.heap.get(*id), HeapData::Str(_)) => Ok(self.clone_with_heap(vm.heap)),
Self::Ref(id) if matches!(vm.heap.get(*id), HeapData::Instance(_)) => instance_str(*id, vm),
Self::Ref(id) => vm.heap.read(*id).py_str(vm),
_ => self.py_repr(vm),
}
}
fn py_add(&self, other: &Self, vm: &mut VM<'_, impl ResourceTracker>) -> Result<Option<Value>, ResourceError> {
let interns = vm.interns;
match (self, other) {
(Self::Int(a), Self::Int(b)) => {
if let Some(result) = a.checked_add(*b) {
Ok(Some(Self::Int(result)))
} else {
let li = LongInt::from(*a) + LongInt::from(*b);
li.into_value(vm.heap).map(Some)
}
}
(Self::Int(i), Self::Ref(id)) | (Self::Ref(id), Self::Int(i))
if let HeapData::LongInt(li) = vm.heap.get(*id) =>
{
let result = LongInt::new(li.inner() + i);
result.into_value(vm.heap).map(Some)
}
(Self::Float(v1), Self::Float(v2)) => Ok(Some(Self::Float(v1 + v2))),
(Self::Int(a), Self::Float(b)) => Ok(Some(Self::Float(*a as f64 + b))),
(Self::Float(a), Self::Int(b)) => Ok(Some(Self::Float(a + *b as f64))),
(Self::Ref(id1), Self::Ref(id2)) => {
let left = vm.heap.read(*id1);
let right = vm.heap.read(*id2);
left.py_add(&right, vm)
}
(Self::InternString(s1), Self::InternString(s2)) => Ok(Some(concat_allocate_str(
interns.get_str(*s1),
interns.get_str(*s2),
vm.heap,
)?)),
(Self::InternString(string_id), Self::Ref(id2)) if let HeapData::Str(s2) = vm.heap.get(*id2) => Ok(Some(
concat_allocate_str(interns.get_str(*string_id), s2.as_str(), vm.heap)?,
)),
(Self::Ref(id1), Self::InternString(string_id)) if let HeapData::Str(s1) = vm.heap.get(*id1) => Ok(Some(
concat_allocate_str(s1.as_str(), interns.get_str(*string_id), vm.heap)?,
)),
(Self::InternBytes(b1), Self::InternBytes(b2)) => {
let bytes1 = interns.get_bytes(*b1);
let bytes2 = interns.get_bytes(*b2);
let mut b = Vec::with_capacity(bytes1.len() + bytes2.len());
b.extend_from_slice(bytes1);
b.extend_from_slice(bytes2);
Ok(Some(Self::Ref(vm.heap.allocate(HeapData::Bytes(b.into()))?)))
}
(Self::InternBytes(bytes_id), Self::Ref(id2)) if let HeapData::Bytes(b2) = vm.heap.get(*id2) => {
let bytes1 = interns.get_bytes(*bytes_id);
let mut b = Vec::with_capacity(bytes1.len() + b2.len());
b.extend_from_slice(bytes1);
b.extend_from_slice(b2);
Ok(Some(Self::Ref(vm.heap.allocate(HeapData::Bytes(b.into()))?)))
}
(Self::Ref(id1), Self::InternBytes(bytes_id)) if let HeapData::Bytes(b1) = vm.heap.get(*id1) => {
let bytes2 = interns.get_bytes(*bytes_id);
let mut b = Vec::with_capacity(b1.len() + bytes2.len());
b.extend_from_slice(b1);
b.extend_from_slice(bytes2);
Ok(Some(Self::Ref(vm.heap.allocate(HeapData::Bytes(b.into()))?)))
}
_ => Ok(None),
}
}
fn py_sub(&self, other: &Self, vm: &mut VM<'_, impl ResourceTracker>) -> Result<Option<Self>, ResourceError> {
match (self, other) {
(Self::Int(a), Self::Int(b)) => {
if let Some(result) = a.checked_sub(*b) {
Ok(Some(Self::Int(result)))
} else {
let li = LongInt::from(*a) - LongInt::from(*b);
li.into_value(vm.heap).map(Some)
}
}
(Self::Int(a), Self::Ref(id)) if let HeapData::LongInt(li) = vm.heap.get(*id) => {
let result = LongInt::from(*a) - LongInt::new(li.inner().clone());
result.into_value(vm.heap).map(Some)
}
(Self::Ref(id), Self::Int(b)) if let HeapData::LongInt(li) = vm.heap.get(*id) => {
let result = LongInt::new(li.inner().clone()) - LongInt::from(*b);
result.into_value(vm.heap).map(Some)
}
(Self::Ref(id1), Self::Ref(id2)) => {
let left = vm.heap.read(*id1);
let right = vm.heap.read(*id2);
left.py_sub(&right, vm)
}
(Self::Float(a), Self::Float(b)) => Ok(Some(Self::Float(a - b))),
(Self::Int(a), Self::Float(b)) => Ok(Some(Self::Float(*a as f64 - b))),
(Self::Float(a), Self::Int(b)) => Ok(Some(Self::Float(a - *b as f64))),
_ => Ok(None),
}
}
fn py_mod(&self, other: &Self, vm: &mut VM<'_, impl ResourceTracker>) -> RunResult<Option<Self>> {
match (self, other) {
(Self::Int(a), Self::Int(b)) => {
if *b == 0 {
Err(ExcType::zero_division().into())
} else if let Some(r) = a.checked_rem(*b) {
let result = if r != 0 && (*a < 0) != (*b < 0) { r + *b } else { r };
Ok(Some(Self::Int(result)))
} else {
Ok(Some(Self::Int(0)))
}
}
(Self::Int(a), Self::Ref(id)) if let HeapData::LongInt(li) = vm.heap.get(*id) => {
if li.is_zero() {
return Err(ExcType::zero_division().into());
}
let bi = BigInt::from(*a).mod_floor(li.inner());
Ok(Some(LongInt::new(bi).into_value(vm.heap)?))
}
(Self::Ref(id), Self::Int(b)) if let HeapData::LongInt(li) = vm.heap.get(*id) => {
if *b == 0 {
return Err(ExcType::zero_division().into());
}
let bi = li.inner().mod_floor(&BigInt::from(*b));
Ok(Some(LongInt::new(bi).into_value(vm.heap)?))
}
(Self::Ref(id1), Self::Ref(id2)) => {
let left = vm.heap.read(*id1);
let right = vm.heap.read(*id2);
left.py_mod(&right, vm)
}
(Self::Float(v1), Self::Float(v2)) => {
if *v2 == 0.0 {
Err(ExcType::zero_division().into())
} else {
Ok(Some(Self::Float(v1 % v2)))
}
}
(Self::Float(v1), Self::Int(v2)) => {
if *v2 == 0 {
Err(ExcType::zero_division().into())
} else {
Ok(Some(Self::Float(v1 % (*v2 as f64))))
}
}
(Self::Int(v1), Self::Float(v2)) => {
if *v2 == 0.0 {
Err(ExcType::zero_division().into())
} else {
Ok(Some(Self::Float((*v1 as f64) % v2)))
}
}
_ => Ok(None),
}
}
fn py_mod_eq(&self, other: &Self, right_value: i64) -> Option<bool> {
match (self, other) {
(Self::Int(v1), Self::Int(v2)) => {
if let Some(r) = v1.checked_rem(*v2) {
let result = if r != 0 && (*v1 < 0) != (*v2 < 0) { r + *v2 } else { r };
Some(result == right_value)
} else {
(*v2 != 0).then_some(0 == right_value)
}
}
(Self::Float(v1), Self::Float(v2)) => Some(v1 % v2 == right_value as f64),
(Self::Float(v1), Self::Int(v2)) => Some(v1 % (*v2 as f64) == right_value as f64),
(Self::Int(v1), Self::Float(v2)) => Some((*v1 as f64) % v2 == right_value as f64),
_ => None,
}
}
fn py_iadd(
&mut self,
other: &Self,
vm: &mut VM<'_, impl ResourceTracker>,
_self_id: Option<HeapId>,
) -> Result<bool, ResourceError> {
let interns = vm.interns;
match (&self, other) {
(Self::Int(v1), Self::Int(v2)) => {
if let Some(result) = v1.checked_add(*v2) {
*self = Self::Int(result);
} else {
let li = LongInt::from(*v1) + LongInt::from(*v2);
*self = li.into_value(vm.heap)?;
}
Ok(true)
}
(Self::Float(v1), Self::Float(v2)) => {
*self = Self::Float(*v1 + *v2);
Ok(true)
}
(Self::InternString(s1), Self::InternString(s2)) => {
let concat = format!("{}{}", interns.get_str(*s1), interns.get_str(*s2));
*self = allocate_string(concat, vm.heap)?;
Ok(true)
}
(Self::InternString(string_id), Self::Ref(id2)) => {
let result = if let HeapData::Str(s2) = vm.heap.get(*id2) {
let concat = format!("{}{}", interns.get_str(*string_id), s2.as_str());
*self = allocate_string(concat, vm.heap)?;
true
} else {
false
};
Ok(result)
}
(Self::InternBytes(b1), Self::InternBytes(b2)) => {
let bytes1 = interns.get_bytes(*b1);
let bytes2 = interns.get_bytes(*b2);
let mut b = Vec::with_capacity(bytes1.len() + bytes2.len());
b.extend_from_slice(bytes1);
b.extend_from_slice(bytes2);
*self = Self::Ref(vm.heap.allocate(HeapData::Bytes(b.into()))?);
Ok(true)
}
(Self::InternBytes(bytes_id), Self::Ref(id2)) => {
let result = if let HeapData::Bytes(b2) = vm.heap.get(*id2) {
let bytes1 = interns.get_bytes(*bytes_id);
let mut b = Vec::with_capacity(bytes1.len() + b2.len());
b.extend_from_slice(bytes1);
b.extend_from_slice(b2);
*self = Self::Ref(vm.heap.allocate(HeapData::Bytes(b.into()))?);
true
} else {
false
};
Ok(result)
}
(Self::Ref(id), Self::Ref(_)) => vm.heap.read(*id).py_iadd(other, vm, Some(*id)),
_ => Ok(false),
}
}
fn py_mult(&self, other: &Self, vm: &mut VM<'_, impl ResourceTracker>) -> RunResult<Option<Value>> {
let interns = vm.interns;
match (self, other) {
(Self::Int(a), Self::Int(b)) => {
if let Some(result) = a.checked_mul(*b) {
Ok(Some(Self::Int(result)))
} else {
let li = LongInt::from(*a) * LongInt::from(*b);
Ok(Some(li.into_value(vm.heap)?))
}
}
(Self::Int(n), Self::Ref(id)) | (Self::Ref(id), Self::Int(n)) => match vm.heap.get(*id) {
HeapData::LongInt(li) => {
check_mult_size(li.bits(), i64_bits(*n), vm.heap.tracker())?;
let result = LongInt::new(li.inner().clone()) * LongInt::from(*n);
Ok(Some(result.into_value(vm.heap)?))
}
HeapData::TimeDelta(td) => {
let total = timedelta::total_microseconds(td)
.checked_mul(i128::from(*n))
.ok_or_else(|| {
SimpleException::new_msg(ExcType::OverflowError, "timedelta multiplication overflow")
})?;
let delta = timedelta::from_total_microseconds(total)?;
Ok(Some(Self::Ref(vm.heap.allocate(HeapData::TimeDelta(delta))?)))
}
HeapData::Str(s) => {
let count = i64_to_repeat_count(*n)?;
check_repeat_size(s.len(), count, vm.heap.tracker())?;
let repeated = s.as_str().repeat(count);
Ok(Some(allocate_string(repeated, vm.heap)?))
}
HeapData::Bytes(b) => {
let count = i64_to_repeat_count(*n)?;
check_repeat_size(b.len(), count, vm.heap.tracker())?;
Ok(Some(Self::Ref(
vm.heap.allocate(HeapData::Bytes(b.as_slice().repeat(count).into()))?,
)))
}
HeapData::List(list) => {
let count = i64_to_repeat_count(*n)?;
check_repeat_size(
list.len().saturating_mul(mem::size_of::<Self>()),
count,
vm.heap.tracker(),
)?;
let mut result = Vec::with_capacity(list.as_slice().len() * count);
for _ in 0..count {
result.extend(list.as_slice().iter().map(|v| v.clone_with_heap(vm.heap)));
vm.heap.check_time()?;
}
Ok(Some(Self::Ref(vm.heap.allocate(HeapData::List(List::new(result)))?)))
}
HeapData::Tuple(tuple) => {
let count = i64_to_repeat_count(*n)?;
if count == 0 {
Ok(Some(vm.heap.get_empty_tuple()))
} else {
check_repeat_size(
tuple.as_slice().len().saturating_mul(mem::size_of::<Self>()),
count,
vm.heap.tracker(),
)?;
let mut result = SmallVec::with_capacity(tuple.as_slice().len() * count);
for _ in 0..count {
result.extend(tuple.as_slice().iter().map(|v| v.clone_with_heap(vm.heap)));
vm.heap.check_time()?;
}
Ok(Some(allocate_tuple(result, vm.heap)?))
}
}
_ => Ok(None),
},
(Self::Ref(id1), Self::Ref(id2)) => {
let (seq_id, count) = match (vm.heap.get(*id1), vm.heap.get(*id2)) {
(HeapData::LongInt(a), HeapData::LongInt(b)) => {
check_mult_size(a.bits(), b.bits(), vm.heap.tracker())?;
let result = LongInt::new(a.inner() * b.inner());
return Ok(Some(result.into_value(vm.heap)?));
}
(HeapData::LongInt(li), _) => (*id2, longint_to_repeat_count(li)?),
(_, HeapData::LongInt(li)) => (*id1, longint_to_repeat_count(li)?),
_ => return Ok(None),
};
match vm.heap.get(seq_id) {
HeapData::Str(s) => {
check_repeat_size(s.len(), count, vm.heap.tracker())?;
let repeated = s.as_str().repeat(count);
Ok(Some(allocate_string(repeated, vm.heap)?))
}
HeapData::Bytes(b) => {
check_repeat_size(b.len(), count, vm.heap.tracker())?;
Ok(Some(Self::Ref(
vm.heap.allocate(HeapData::Bytes(b.as_slice().repeat(count).into()))?,
)))
}
HeapData::List(list) => {
check_repeat_size(
list.len().saturating_mul(mem::size_of::<Self>()),
count,
vm.heap.tracker(),
)?;
let mut result = Vec::with_capacity(list.as_slice().len() * count);
for _ in 0..count {
result.extend(list.as_slice().iter().map(|v| v.clone_with_heap(vm.heap)));
vm.heap.check_time()?;
}
Ok(Some(Self::Ref(vm.heap.allocate(HeapData::List(List::new(result)))?)))
}
HeapData::Tuple(tuple) => {
if count == 0 {
Ok(Some(vm.heap.get_empty_tuple()))
} else {
check_repeat_size(
tuple.as_slice().len().saturating_mul(mem::size_of::<Self>()),
count,
vm.heap.tracker(),
)?;
let mut result = SmallVec::with_capacity(tuple.as_slice().len() * count);
for _ in 0..count {
result.extend(tuple.as_slice().iter().map(|v| v.clone_with_heap(vm.heap)));
vm.heap.check_time()?;
}
Ok(Some(allocate_tuple(result, vm.heap)?))
}
}
_ => Ok(None),
}
}
(Self::Float(a), Self::Float(b)) => Ok(Some(Self::Float(a * b))),
(Self::Int(a), Self::Float(b)) => Ok(Some(Self::Float(*a as f64 * b))),
(Self::Float(a), Self::Int(b)) => Ok(Some(Self::Float(a * *b as f64))),
(Self::Bool(a), Self::Int(b)) => {
let a_int = i64::from(*a);
Ok(Some(Self::Int(a_int * b)))
}
(Self::Int(a), Self::Bool(b)) => {
let b_int = i64::from(*b);
Ok(Some(Self::Int(a * b_int)))
}
(Self::Bool(a), Self::Float(b)) => {
let a_float = if *a { 1.0 } else { 0.0 };
Ok(Some(Self::Float(a_float * b)))
}
(Self::Float(a), Self::Bool(b)) => {
let b_float = if *b { 1.0 } else { 0.0 };
Ok(Some(Self::Float(a * b_float)))
}
(Self::Bool(a), Self::Bool(b)) => {
let result = i64::from(*a) * i64::from(*b);
Ok(Some(Self::Int(result)))
}
(Self::InternString(s), Self::Int(n)) | (Self::Int(n), Self::InternString(s)) => {
let count = i64_to_repeat_count(*n)?;
let str_ref = interns.get_str(*s);
check_repeat_size(str_ref.len(), count, vm.heap.tracker())?;
let result = str_ref.repeat(count);
Ok(Some(allocate_string(result, vm.heap)?))
}
(Self::InternBytes(b), Self::Int(n)) | (Self::Int(n), Self::InternBytes(b)) => {
let count = i64_to_repeat_count(*n)?;
let bytes_ref = interns.get_bytes(*b);
check_repeat_size(bytes_ref.len(), count, vm.heap.tracker())?;
let result: Vec<u8> = bytes_ref.repeat(count);
Ok(Some(Self::Ref(vm.heap.allocate(HeapData::Bytes(result.into()))?)))
}
(Self::InternString(s), Self::Ref(id)) | (Self::Ref(id), Self::InternString(s))
if let HeapData::LongInt(li) = vm.heap.get(*id) =>
{
let count = longint_to_repeat_count(li)?;
let str_ref = interns.get_str(*s);
check_repeat_size(str_ref.len(), count, vm.heap.tracker())?;
let result = str_ref.repeat(count);
Ok(Some(allocate_string(result, vm.heap)?))
}
(Self::InternBytes(b), Self::Ref(id)) | (Self::Ref(id), Self::InternBytes(b))
if let HeapData::LongInt(li) = vm.heap.get(*id) =>
{
let count = longint_to_repeat_count(li)?;
let bytes_ref = interns.get_bytes(*b);
check_repeat_size(bytes_ref.len(), count, vm.heap.tracker())?;
let result: Vec<u8> = bytes_ref.repeat(count);
Ok(Some(Self::Ref(vm.heap.allocate(HeapData::Bytes(result.into()))?)))
}
_ => Ok(None),
}
}
fn py_div(&self, other: &Self, vm: &mut VM<'_, impl ResourceTracker>) -> RunResult<Option<Value>> {
let interns = vm.interns;
match (self, other) {
(Self::Int(a), Self::Int(b)) => {
if *b == 0 {
Err(ExcType::zero_division().into())
} else {
Ok(Some(Self::Float(*a as f64 / *b as f64)))
}
}
(Self::Int(a), Self::Ref(id)) => {
if let HeapData::LongInt(li) = vm.heap.get(*id) {
if li.is_zero() {
Err(ExcType::zero_division().into())
} else {
let a_f64 = *a as f64;
let b_f64 = li.to_f64().unwrap_or(f64::INFINITY);
Ok(Some(Self::Float(a_f64 / b_f64)))
}
} else {
Ok(None)
}
}
(Self::Ref(id), Self::Int(b)) => match vm.heap.get(*id) {
HeapData::LongInt(li) => {
if *b == 0 {
Err(ExcType::zero_division().into())
} else {
let a_f64 = li.to_f64().unwrap_or(f64::INFINITY);
let b_f64 = *b as f64;
Ok(Some(Self::Float(a_f64 / b_f64)))
}
}
HeapData::TimeDelta(td) => {
if *b == 0 {
Err(ExcType::zero_division().into())
} else {
let total = timedelta::total_microseconds(td);
let result = timedelta::div_microseconds_round_ties_even(total, i128::from(*b));
let delta = timedelta::from_total_microseconds(result)?;
Ok(Some(Self::Ref(vm.heap.allocate(HeapData::TimeDelta(delta))?)))
}
}
_ => Ok(None),
},
(Self::Ref(id1), Self::Ref(id2)) => match (vm.heap.get(*id1), vm.heap.get(*id2)) {
(HeapData::LongInt(li1), HeapData::LongInt(li2)) => {
if li2.is_zero() {
Err(ExcType::zero_division().into())
} else {
let a_f64 = li1.to_f64().unwrap_or(f64::INFINITY);
let b_f64 = li2.to_f64().unwrap_or(f64::INFINITY);
Ok(Some(Self::Float(a_f64 / b_f64)))
}
}
_ => Ok(None),
},
(Self::Ref(id), Self::Float(b)) => {
if let HeapData::LongInt(li) = vm.heap.get(*id) {
if *b == 0.0 {
Err(ExcType::zero_division().into())
} else {
let a_f64 = li.to_f64().unwrap_or(f64::INFINITY);
Ok(Some(Self::Float(a_f64 / b)))
}
} else {
Ok(None)
}
}
(Self::Float(a), Self::Ref(id)) => {
if let HeapData::LongInt(li) = vm.heap.get(*id) {
if li.is_zero() {
Err(ExcType::zero_division().into())
} else {
let b_f64 = li.to_f64().unwrap_or(f64::INFINITY);
Ok(Some(Self::Float(a / b_f64)))
}
} else {
Ok(None)
}
}
(Self::Float(a), Self::Float(b)) => {
if *b == 0.0 {
Err(ExcType::zero_division().into())
} else {
Ok(Some(Self::Float(a / b)))
}
}
(Self::Int(a), Self::Float(b)) => {
if *b == 0.0 {
Err(ExcType::zero_division().into())
} else {
Ok(Some(Self::Float(*a as f64 / b)))
}
}
(Self::Float(a), Self::Int(b)) => {
if *b == 0 {
Err(ExcType::zero_division().into())
} else {
Ok(Some(Self::Float(a / *b as f64)))
}
}
(Self::Bool(a), Self::Int(b)) => {
if *b == 0 {
Err(ExcType::zero_division().into())
} else {
Ok(Some(Self::Float(f64::from(*a) / *b as f64)))
}
}
(Self::Int(a), Self::Bool(b)) => {
if *b {
Ok(Some(Self::Float(*a as f64))) } else {
Err(ExcType::zero_division().into())
}
}
(Self::Bool(a), Self::Float(b)) => {
if *b == 0.0 {
Err(ExcType::zero_division().into())
} else {
Ok(Some(Self::Float(f64::from(*a) / b)))
}
}
(Self::Float(a), Self::Bool(b)) => {
if *b {
Ok(Some(Self::Float(*a))) } else {
Err(ExcType::zero_division().into())
}
}
(Self::Bool(a), Self::Bool(b)) => {
if *b {
Ok(Some(Self::Float(f64::from(*a)))) } else {
Err(ExcType::zero_division().into())
}
}
_ => {
if let Self::Ref(id) = self
&& matches!(vm.heap.get(*id), HeapData::Path(_))
{
return path::path_div(*id, other, vm.heap, interns);
}
Ok(None)
}
}
}
fn py_floordiv(&self, other: &Self, vm: &mut VM<'_, impl ResourceTracker>) -> RunResult<Option<Value>> {
match (self, other) {
(Self::Int(a), Self::Int(b)) => {
if *b == 0 {
Err(ExcType::zero_division().into())
} else if let Some((d, _)) = floor_divmod(*a, *b) {
Ok(Some(Self::Int(d)))
} else {
check_div_size(i64_bits(*a), vm.heap.tracker())?;
let bi = BigInt::from(*a).div_floor(&BigInt::from(*b));
Ok(Some(LongInt::new(bi).into_value(vm.heap)?))
}
}
(Self::Int(a), Self::Ref(id)) => {
if let HeapData::LongInt(li) = vm.heap.get(*id) {
if li.is_zero() {
Err(ExcType::zero_division().into())
} else {
let bi = BigInt::from(*a).div_floor(li.inner());
Ok(Some(LongInt::new(bi).into_value(vm.heap)?))
}
} else {
Ok(None)
}
}
(Self::Ref(id), Self::Int(b)) => match vm.heap.get(*id) {
HeapData::LongInt(li) => {
if *b == 0 {
Err(ExcType::zero_division().into())
} else {
let bi = li.inner().div_floor(&BigInt::from(*b));
Ok(Some(LongInt::new(bi).into_value(vm.heap)?))
}
}
HeapData::TimeDelta(td) => {
if *b == 0 {
Err(ExcType::zero_division().into())
} else {
let total = timedelta::total_microseconds(td);
let result = total.div_euclid(i128::from(*b));
let delta = timedelta::from_total_microseconds(result)?;
Ok(Some(Self::Ref(vm.heap.allocate(HeapData::TimeDelta(delta))?)))
}
}
_ => Ok(None),
},
(Self::Ref(id1), Self::Ref(id2)) => match (vm.heap.get(*id1), vm.heap.get(*id2)) {
(HeapData::LongInt(li1), HeapData::LongInt(li2)) => {
if li2.is_zero() {
Err(ExcType::zero_division().into())
} else {
let bi = li1.inner().div_floor(li2.inner());
Ok(Some(LongInt::new(bi).into_value(vm.heap)?))
}
}
_ => Ok(None),
},
(Self::Float(a), Self::Float(b)) => {
if *b == 0.0 {
Err(ExcType::zero_division().into())
} else {
Ok(Some(Self::Float((a / b).floor())))
}
}
(Self::Int(a), Self::Float(b)) => {
if *b == 0.0 {
Err(ExcType::zero_division().into())
} else {
Ok(Some(Self::Float((*a as f64 / b).floor())))
}
}
(Self::Float(a), Self::Int(b)) => {
if *b == 0 {
Err(ExcType::zero_division().into())
} else {
Ok(Some(Self::Float((a / *b as f64).floor())))
}
}
(Self::Bool(a), Self::Int(b)) => {
if *b == 0 {
Err(ExcType::zero_division().into())
} else {
let a_int = i64::from(*a);
let d = a_int / b;
let r = a_int % b;
let result = if r != 0 && (a_int < 0) != (*b < 0) { d - 1 } else { d };
Ok(Some(Self::Int(result)))
}
}
(Self::Int(a), Self::Bool(b)) => {
if *b {
Ok(Some(Self::Int(*a))) } else {
Err(ExcType::zero_division().into())
}
}
(Self::Bool(a), Self::Float(b)) => {
if *b == 0.0 {
Err(ExcType::zero_division().into())
} else {
Ok(Some(Self::Float((f64::from(*a) / b).floor())))
}
}
(Self::Float(a), Self::Bool(b)) => {
if *b {
Ok(Some(Self::Float(a.floor()))) } else {
Err(ExcType::zero_division().into())
}
}
(Self::Bool(a), Self::Bool(b)) => {
if *b {
Ok(Some(Self::Int(i64::from(*a)))) } else {
Err(ExcType::zero_division().into())
}
}
_ => Ok(None),
}
}
fn py_pow(&self, other: &Self, vm: &mut VM<'_, impl ResourceTracker>) -> RunResult<Option<Value>> {
match (self, other) {
(Self::Int(base), Self::Int(exp)) => {
if *base == 0 && *exp < 0 {
Err(ExcType::zero_negative_power())
} else if *exp >= 0 {
if let Ok(exp_u32) = u32::try_from(*exp) {
if let Some(result) = base.checked_pow(exp_u32) {
Ok(Some(Self::Int(result)))
} else {
check_pow_size(i64_bits(*base), u64::from(exp_u32), vm.heap.tracker())?;
let bi = BigInt::from(*base).pow(exp_u32);
Ok(Some(LongInt::new(bi).into_value(vm.heap)?))
}
} else {
#[expect(clippy::cast_sign_loss)]
let exp_u64 = *exp as u64;
check_pow_size(i64_bits(*base), exp_u64, vm.heap.tracker())?;
let bi = bigint_pow(BigInt::from(*base), exp_u64);
Ok(Some(LongInt::new(bi).into_value(vm.heap)?))
}
} else {
if let Ok(exp_i32) = i32::try_from(*exp) {
Ok(Some(Self::Float((*base as f64).powi(exp_i32))))
} else {
Ok(Some(Self::Float((*base as f64).powf(*exp as f64))))
}
}
}
(Self::Ref(id), Self::Int(exp)) => {
if let HeapData::LongInt(li) = vm.heap.get(*id) {
if li.is_zero() && *exp < 0 {
Err(ExcType::zero_negative_power())
} else if *exp >= 0 {
if let Ok(exp_u32) = u32::try_from(*exp) {
check_pow_size(li.bits(), u64::from(exp_u32), vm.heap.tracker())?;
let bi = li.inner().pow(exp_u32);
Ok(Some(LongInt::new(bi).into_value(vm.heap)?))
} else {
#[expect(clippy::cast_sign_loss)]
let exp_u64 = *exp as u64;
check_pow_size(li.bits(), exp_u64, vm.heap.tracker())?;
let bi = bigint_pow(li.inner().clone(), exp_u64);
Ok(Some(LongInt::new(bi).into_value(vm.heap)?))
}
} else {
if let Some(base_f64) = li.to_f64() {
if let Ok(exp_i32) = i32::try_from(*exp) {
Ok(Some(Self::Float(base_f64.powi(exp_i32))))
} else {
Ok(Some(Self::Float(base_f64.powf(*exp as f64))))
}
} else {
Ok(Some(Self::Float(0.0)))
}
}
} else {
Ok(None)
}
}
(Self::Int(base), Self::Ref(id)) => {
if let HeapData::LongInt(li) = vm.heap.get(*id) {
if *base == 0 && li.is_negative() {
Err(ExcType::zero_negative_power())
} else if !li.is_negative() {
if li.is_zero() {
Ok(Some(Self::Int(1)))
} else if *base == 0 {
Ok(Some(Self::Int(0)))
} else if *base == 1 {
Ok(Some(Self::Int(1)))
} else if *base == -1 {
let is_even = (li.inner() % 2i32).is_zero();
Ok(Some(Self::Int(if is_even { 1 } else { -1 })))
} else if let Some(exp_u32) = li.to_u32() {
if let Some(result) = base.checked_pow(exp_u32) {
Ok(Some(Self::Int(result)))
} else {
check_pow_size(i64_bits(*base), u64::from(exp_u32), vm.heap.tracker())?;
let bi = BigInt::from(*base).pow(exp_u32);
Ok(Some(LongInt::new(bi).into_value(vm.heap)?))
}
} else {
Err(SimpleException::new_msg(ExcType::OverflowError, "exponent too large").into())
}
} else {
if let (Some(base_f64), Some(exp_f64)) = (Some(*base as f64), li.to_f64()) {
Ok(Some(Self::Float(base_f64.powf(exp_f64))))
} else {
Ok(Some(Self::Float(0.0)))
}
}
} else {
Ok(None)
}
}
(Self::Float(base), Self::Float(exp)) => {
if *base == 0.0 && *exp < 0.0 {
Err(ExcType::zero_negative_power())
} else {
Ok(Some(Self::Float(base.powf(*exp))))
}
}
(Self::Int(base), Self::Float(exp)) => {
if *base == 0 && *exp < 0.0 {
Err(ExcType::zero_negative_power())
} else {
Ok(Some(Self::Float((*base as f64).powf(*exp))))
}
}
(Self::Float(base), Self::Int(exp)) => {
if *base == 0.0 && *exp < 0 {
Err(ExcType::zero_negative_power())
} else if let Ok(exp_i32) = i32::try_from(*exp) {
Ok(Some(Self::Float(base.powi(exp_i32))))
} else {
Ok(Some(Self::Float(base.powf(*exp as f64))))
}
}
(Self::Bool(base), Self::Int(exp)) => {
let base_int = i64::from(*base);
if base_int == 0 && *exp < 0 {
Err(ExcType::zero_negative_power())
} else if *exp >= 0 {
if let Ok(exp_u32) = u32::try_from(*exp) {
match base_int.checked_pow(exp_u32) {
Some(result) => Ok(Some(Self::Int(result))),
None => Ok(Some(Self::Float((base_int as f64).powf(*exp as f64)))),
}
} else {
Ok(Some(Self::Float((base_int as f64).powf(*exp as f64))))
}
} else {
if let Ok(exp_i32) = i32::try_from(*exp) {
Ok(Some(Self::Float((base_int as f64).powi(exp_i32))))
} else {
Ok(Some(Self::Float((base_int as f64).powf(*exp as f64))))
}
}
}
(Self::Int(base), Self::Bool(exp)) => {
if *exp {
Ok(Some(Self::Int(*base)))
} else {
Ok(Some(Self::Int(1)))
}
}
(Self::Bool(base), Self::Float(exp)) => {
let base_float = f64::from(*base);
if base_float == 0.0 && *exp < 0.0 {
Err(ExcType::zero_negative_power())
} else {
Ok(Some(Self::Float(base_float.powf(*exp))))
}
}
(Self::Float(base), Self::Bool(exp)) => {
if *exp {
Ok(Some(Self::Float(*base)))
} else {
Ok(Some(Self::Float(1.0)))
}
}
(Self::Bool(base), Self::Bool(exp)) => {
let base_int = i64::from(*base);
let exp_int = i64::from(*exp);
if exp_int == 0 {
Ok(Some(Self::Int(1))) } else {
Ok(Some(Self::Int(base_int))) }
}
_ => Ok(None),
}
}
fn py_getitem(&self, key: &Self, vm: &mut VM<'_, impl ResourceTracker>) -> RunResult<Self> {
let interns = vm.interns;
match self {
Self::Ref(id) => vm.heap.read(*id).py_getitem(key, vm),
Self::InternString(string_id) => {
if let Self::Ref(key_id) = key
&& let HeapData::Slice(slice_obj) = vm.heap.get(*key_id)
{
let s = interns.get_str(*string_id);
let result_str: Box<str> = slice_collect_iterator(vm, slice_obj, s.chars(), |c| c)?;
return Ok(allocate_string(result_str, vm.heap)?);
}
let index = match key {
Self::Int(i) => *i,
Self::Bool(b) => i64::from(*b),
_ => return Err(ExcType::type_error_indices(Type::Str, &key.py_type_name(vm))),
};
let s = interns.get_str(*string_id);
let c = get_char_at_index(s, index).ok_or_else(ExcType::str_index_error)?;
Ok(allocate_char(c, vm.heap)?)
}
Self::InternBytes(bytes_id) => {
if let Self::Ref(key_id) = key
&& let HeapData::Slice(slice_obj) = vm.heap.get(*key_id)
{
let bytes = interns.get_bytes(*bytes_id);
let result_bytes = slice_collect_iterator(vm, slice_obj, bytes.iter(), |b| *b)?;
let heap_id = vm.heap.allocate(HeapData::Bytes(Bytes::new(result_bytes)))?;
return Ok(Self::Ref(heap_id));
}
let index = match key {
Self::Int(i) => *i,
Self::Bool(b) => i64::from(*b),
_ => return Err(ExcType::type_error_indices(Type::Bytes, &key.py_type_name(vm))),
};
let bytes = interns.get_bytes(*bytes_id);
let byte = get_byte_at_index(bytes, index).ok_or_else(ExcType::bytes_index_error)?;
Ok(Self::Int(i64::from(byte)))
}
_ => Err(ExcType::type_error_not_sub(&self.py_type_name(vm))),
}
}
fn py_setitem(&mut self, key: Self, value: Self, vm: &mut VM<'_, impl ResourceTracker>) -> RunResult<()> {
match self {
Self::Ref(id) => vm.heap.read(*id).py_setitem(key, value, vm),
_ => Err(ExcType::type_error(format!(
"'{}' object does not support item assignment",
self.py_type_name(vm)
))),
}
}
}
impl Value {
#[must_use]
pub(crate) fn py_type_heap(&self, heap: &Heap<impl ResourceTracker>) -> Type {
match self {
Self::Ref(id) => heap.get(*id).py_type(),
_ => self.py_type_shallow(),
}
}
#[must_use]
pub(crate) fn py_type_name<'h>(&self, vm: &VM<'h, impl ResourceTracker>) -> Cow<'h, str> {
self.py_type(vm).name(vm.heap, vm.interns)
}
#[must_use]
pub(crate) fn py_type_name_heap<'i>(
&self,
heap: &Heap<impl ResourceTracker>,
interns: &'i Interns,
) -> Cow<'i, str> {
self.py_type_heap(heap).name(heap, interns)
}
#[must_use]
pub(crate) fn py_type_shallow(&self) -> Type {
match self {
Self::Undefined | Self::None => Type::NoneType,
Self::Ellipsis => Type::Ellipsis,
Self::Bool(_) => Type::Bool,
Self::Int(_) | Self::InternLongInt(_) => Type::Int,
Self::Float(_) => Type::Float,
Self::InternString(_) => Type::Str,
Self::InternBytes(_) => Type::Bytes,
Self::Builtin(_) => Type::BuiltinFunction,
Self::ModuleFunction(_) | Self::DefFunction(_) | Self::ExtFunction(_) => Type::Function,
Self::Marker(_) => Type::SpecialForm,
Self::Property(_) => Type::Property,
Self::Ref(_) => Type::NoneType, #[cfg(feature = "memory-model-checks")]
Self::Dereferenced => Type::NoneType,
}
}
pub fn id(&self, vm: &VM<'_, impl ResourceTracker>) -> usize {
match self {
Self::ExtFunction(name_id) => ext_function_value_id(vm.interns.get_str(*name_id)),
Self::Ref(id) if let HeapData::ExtFunction(name) = vm.heap.get(*id) => ext_function_value_id(name.as_str()),
Self::Undefined => singleton_id(SingletonSlot::Undefined),
Self::Ellipsis => singleton_id(SingletonSlot::Ellipsis),
Self::None => singleton_id(SingletonSlot::None),
Self::Bool(b) => {
if *b {
singleton_id(SingletonSlot::True)
} else {
singleton_id(SingletonSlot::False)
}
}
Self::InternString(string_id) => INTERN_STR_ID_TAG | (string_id.index() & INTERN_STR_ID_MASK),
Self::InternBytes(bytes_id) => INTERN_BYTES_ID_TAG | (bytes_id.index() & INTERN_BYTES_ID_MASK),
Self::InternLongInt(long_int_id) => {
INTERN_LONG_INT_ID_TAG | (long_int_id.index() & INTERN_LONG_INT_ID_MASK)
}
Self::Ref(id) => heap_tagged_id(*id),
Self::Int(v) => int_value_id(*v),
Self::Float(v) => float_value_id(*v),
Self::Builtin(c) => builtin_value_id(*c),
Self::ModuleFunction(mf) => module_function_value_id(*mf),
Self::DefFunction(f_id) => function_value_id(*f_id),
Self::Marker(m) => marker_value_id(*m),
Self::Property(p) => property_value_id(*p),
#[cfg(feature = "memory-model-checks")]
Self::Dereferenced => panic!("Cannot get id of Dereferenced object"),
}
}
pub fn ref_id(&self) -> Option<HeapId> {
match self {
Self::Ref(id) => Some(*id),
_ => None,
}
}
pub fn module_name(&self, vm: &mut VM<'_, impl ResourceTracker>) -> String {
match self {
Self::Ref(id) => match vm.heap.get(*id) {
HeapData::Module(module) => vm.interns.get_str(module.name()).to_string(),
_ => "<unknown>".to_string(),
},
_ => "<unknown>".to_string(),
}
}
pub fn is(&self, other: &Self, vm: &VM<'_, impl ResourceTracker>) -> bool {
self.id(vm) == other.id(vm)
}
pub fn py_eq(&self, other: &Self, vm: &mut VM<'_, impl ResourceTracker>) -> RunResult<bool> {
if let Some(result) = self.py_eq_impl(other, vm)? {
Ok(result)
} else if let Some(result) = other.py_eq_impl(self, vm)? {
Ok(result)
} else {
Ok(false)
}
}
pub(crate) fn read_heap<'a>(&self, vm: &VM<'a, impl ResourceTracker>) -> Option<HeapReadOutput<'a>> {
match self {
Self::Ref(id) => Some(vm.heap.read(*id)),
_ => None,
}
}
pub fn py_hash(&self, vm: &mut VM<'_, impl ResourceTracker>) -> RunResult<Option<HashValue>> {
match self {
Self::InternString(string_id) => Ok(Some(vm.interns.str_hash(*string_id))),
Self::InternBytes(bytes_id) => Ok(Some(vm.interns.bytes_hash(*bytes_id))),
Self::InternLongInt(long_int_id) => Ok(Some(vm.interns.long_int_hash(*long_int_id))),
Self::Bool(b) => Ok(Some(HashValue::new((*b).into()))),
Self::Int(i) => Ok(Some(HashValue::new(i.cast_unsigned()))),
Self::Float(f) => {
const TWO_POW_63: f64 = 9_223_372_036_854_775_808.0;
if f.fract() != 0.0 || !f.is_finite() {
Ok(Some(HashValue::new(f.to_bits())))
} else if *f >= -TWO_POW_63 && *f < TWO_POW_63 {
#[expect(clippy::cast_possible_truncation)]
Ok(Some(HashValue::new((*f as i64).cast_unsigned())))
} else {
Ok(Some(hash_python_long_int(
&BigInt::from_f64(*f).expect("finite f64 converts to BigInt"),
)))
}
}
Self::Ref(id) => vm.heap.read(*id).py_hash(*id, vm),
Self::Undefined | Self::Ellipsis | Self::None => Ok(Some(hash_one(discriminant(self)))),
Self::Builtin(b) => Ok(Some(hash_one(b))),
Self::ModuleFunction(mf) => Ok(Some(hash_one(mf))),
Self::DefFunction(f_id) => Ok(Some(hash_one(f_id))),
Self::ExtFunction(name_id) => Ok(Some(hash_python_str(vm.interns.get_str(*name_id)))),
Self::Marker(m) => Ok(Some(hash_one(m))),
Self::Property(p) => Ok(Some(hash_one(p))),
#[cfg(feature = "memory-model-checks")]
Self::Dereferenced => panic!("Cannot access Dereferenced object"),
}
}
pub fn py_contains(&self, item: &Self, vm: &mut VM<'_, impl ResourceTracker>) -> RunResult<bool> {
match self {
Self::Ref(heap_id) => {
let output = vm.heap.read(*heap_id);
match output {
HeapReadOutput::List(list) => {
let len = list.get(vm.heap).len();
for i in 0..len {
let el = list.clone_item(i, vm);
let eq = item.py_eq(&el, vm);
el.drop_with_heap(vm);
if eq? {
return Ok(true);
}
}
Ok(false)
}
HeapReadOutput::Tuple(tuple) => {
let len = tuple.get(vm.heap).as_slice().len();
for i in 0..len {
let el = tuple.clone_item(i, vm);
let eq = item.py_eq(&el, vm);
el.drop_with_heap(vm);
if eq? {
return Ok(true);
}
}
Ok(false)
}
HeapReadOutput::Dict(dict) => dict.contains_key(item, vm),
HeapReadOutput::DictKeysView(view) => {
let dict_id = view.get(vm.heap).dict_id();
let HeapReadOutput::Dict(dict) = vm.heap.read(dict_id) else {
panic!("dict_keys view must reference a dict");
};
dict.contains_key(item, vm)
}
HeapReadOutput::DictItemsView(view) => {
let dict_id = view.get(vm.heap).dict_id();
let Some((key, value)) = cloned_items_view_candidate(item, vm) else {
return Ok(false);
};
let mut key_guard = HeapGuard::new(key, vm);
let (key, vm) = key_guard.as_parts_mut();
let mut value_guard = HeapGuard::new(value, vm);
let (value, vm) = value_guard.as_parts_mut();
let HeapReadOutput::Dict(dict) = vm.heap.read(dict_id) else {
panic!("dict_items view must reference a dict");
};
match dict.dict_get(key, vm) {
Ok(Some(existing_value)) => {
let result = value.py_eq(&existing_value, vm);
existing_value.drop_with_heap(vm);
result
}
Ok(None) => Ok(false),
Err(e) => Err(e),
}
}
HeapReadOutput::DictValuesView(view) => {
let dict_id = view.get(vm.heap).dict_id();
let HeapReadOutput::Dict(dict) = vm.heap.read(dict_id) else {
panic!("dict_values view must reference a dict");
};
let len = dict.get(vm.heap).len();
for i in 0..len {
let ref_id = match dict.get(vm.heap).value_at(i) {
Some(Self::Ref(id)) => Some(*id),
_ => None,
};
let el = if let Some(id) = ref_id {
vm.heap.inc_ref(id);
Self::Ref(id)
} else {
dict.get(vm.heap).value_at(i).expect("index valid").clone_immediate()
};
let eq = item.py_eq(&el, vm);
el.drop_with_heap(vm);
if eq? {
return Ok(true);
}
}
Ok(false)
}
HeapReadOutput::Set(set) => set.contains(item, vm),
HeapReadOutput::FrozenSet(fset) => fset.contains(item, vm),
HeapReadOutput::Str(s) => {
let s_str = s.get(vm.heap).as_str();
str_contains(s_str, item, vm.heap, vm.interns)
}
HeapReadOutput::Range(range) => {
let range = range.get(vm.heap);
let n = match item {
Self::Int(i) => *i,
Self::Bool(b) => i64::from(*b),
Self::Float(f) => {
if f.fract() != 0.0 {
return Ok(false);
}
let int_val = f.trunc();
if int_val < i64::MIN as f64 || int_val > i64::MAX as f64 {
return Ok(false);
}
#[expect(clippy::cast_possible_truncation)]
let n = int_val as i64;
n
}
_ => return Ok(false),
};
Ok(range.contains(n))
}
_ => {
let type_name = self.py_type_name(vm);
Err(ExcType::type_error(format!(
"argument of type '{type_name}' is not a container or iterable"
)))
}
}
}
Self::InternString(string_id) => {
let container_str = vm.interns.get_str(*string_id);
str_contains(container_str, item, vm.heap, vm.interns)
}
_ => {
let type_name = self.py_type_name(vm);
Err(ExcType::type_error(format!(
"argument of type '{type_name}' is not a container or iterable"
)))
}
}
}
pub fn py_getattr(&self, attr: &EitherStr, vm: &mut VM<'_, impl ResourceTracker>) -> RunResult<CallResult> {
match self {
Self::Ref(heap_id) if matches!(vm.heap.get(*heap_id), HeapData::Instance(_)) => {
return instance_getattr(*heap_id, attr, vm);
}
Self::Ref(heap_id) => {
if let Some(call_result) = vm.heap.read(*heap_id).py_getattr(attr, vm)? {
return Ok(call_result);
}
}
Self::Builtin(Builtins::Type(t)) => {
let is_dunder_name = attr.static_string().map_or_else(
|| attr.as_str(vm.interns) == "__name__",
|ss| ss == StaticStrings::DunderName,
);
if is_dunder_name {
return Ok(CallResult::Value(allocate_string(
t.name(vm.heap, vm.interns),
vm.heap,
)?));
}
if *t == Type::TimeZone && attr.as_str(vm.interns) == "utc" {
return Ok(CallResult::Value(vm.heap.get_timezone_utc()?));
}
}
_ => {}
}
let type_name = self.py_type_name(vm);
Err(ExcType::attribute_error(type_name, attr.as_str(vm.interns)))
}
pub fn py_set_attr(&self, name: &EitherStr, value: Self, vm: &mut VM<'_, impl ResourceTracker>) -> RunResult<()> {
if let Self::Ref(heap_id) = self {
let old_value = match vm.heap.read(*heap_id) {
HeapReadOutput::Dataclass(mut dc) => dc.set_attr(Self::attr_name_value(name, vm)?, value, vm)?,
HeapReadOutput::Instance(mut instance) => {
instance.set_attr(Self::attr_name_value(name, vm)?, value, vm)?
}
HeapReadOutput::Class(mut class) => class.set_attr(Self::attr_name_value(name, vm)?, value, vm)?,
other => {
let type_name = other.py_type(vm).name(vm.heap, vm.interns);
value.drop_with_heap(vm);
return Err(ExcType::attribute_error_no_setattr(&type_name, name.as_str(vm.interns)));
}
};
old_value.drop_with_heap(vm);
Ok(())
} else {
let type_name = self.py_type_name(vm);
value.drop_with_heap(vm);
Err(ExcType::attribute_error_no_setattr(&type_name, name.as_str(vm.interns)))
}
}
fn attr_name_value(name: &EitherStr, vm: &VM<'_, impl ResourceTracker>) -> RunResult<Self> {
Ok(match name {
EitherStr::Interned(string_id) => Self::InternString(*string_id),
EitherStr::Heap(s) => allocate_string(s.as_str(), vm.heap)?,
})
}
pub fn as_int(&self, vm: &VM<'_, impl ResourceTracker>) -> RunResult<i64> {
match self {
Self::Int(i) => Ok(*i),
Self::Ref(heap_id) => {
if let HeapData::LongInt(li) = vm.heap.get(*heap_id) {
li.to_i64().ok_or_else(ExcType::overflow_c_ssize_t)
} else {
let msg = format!("'{}' object cannot be interpreted as an integer", self.py_type_name(vm));
Err(SimpleException::new_msg(ExcType::TypeError, msg).into())
}
}
_ => {
let msg = format!("'{}' object cannot be interpreted as an integer", self.py_type_name(vm));
Err(SimpleException::new_msg(ExcType::TypeError, msg).into())
}
}
}
pub fn as_index(&self, vm: &VM<'_, impl ResourceTracker>, container_type: Type) -> RunResult<i64> {
match self {
Self::Int(i) => Ok(*i),
Self::Bool(b) => Ok(i64::from(*b)),
Self::Ref(heap_id) => {
if let HeapData::LongInt(li) = vm.heap.get(*heap_id) {
li.to_i64().ok_or_else(ExcType::index_error_int_too_large)
} else {
Err(ExcType::type_error_indices(container_type, &self.py_type_name(vm)))
}
}
_ => Err(ExcType::type_error_indices(container_type, &self.py_type_name(vm))),
}
}
pub fn py_bitwise(
&self,
other: &Self,
op: BitwiseOp,
vm: &mut VM<'_, impl ResourceTracker>,
) -> Result<Self, RunError> {
let lhs_type = self.py_type(vm);
let lhs_name = self.py_type_name(vm);
let rhs_name = other.py_type_name(vm);
let lhs_bigint = extract_bigint(self, vm.heap);
let rhs_bigint = extract_bigint(other, vm.heap);
if let (Some(l), Some(r)) = (lhs_bigint, rhs_bigint) {
let result = match op {
BitwiseOp::And => l & r,
BitwiseOp::Or => l | r,
BitwiseOp::Xor => l ^ r,
BitwiseOp::LShift => {
let shift_amount = r.to_i64();
if let Some(shift) = shift_amount {
if shift < 0 {
return Err(ExcType::value_error_negative_shift_count());
}
#[expect(clippy::cast_sign_loss)]
let shift_u64 = shift as u64;
check_lshift_size(l.bits(), shift_u64, vm.heap.tracker())?;
l << shift_u64
} else if r.sign() == num_bigint::Sign::Minus {
return Err(ExcType::value_error_negative_shift_count());
} else {
return Err(ExcType::overflow_c_ssize_t());
}
}
BitwiseOp::RShift => {
let shift_amount = r.to_i64();
if let Some(shift) = shift_amount {
if shift < 0 {
return Err(ExcType::value_error_negative_shift_count());
}
#[expect(clippy::cast_sign_loss)]
let shift_u64 = shift as u64;
l >> shift_u64
} else if r.sign() == num_bigint::Sign::Minus {
return Err(ExcType::value_error_negative_shift_count());
} else {
if l.sign() == num_bigint::Sign::Minus {
BigInt::from(-1)
} else {
BigInt::from(0)
}
}
}
};
LongInt::new(result).into_value(vm.heap).map_err(Into::into)
} else {
Err(ExcType::binary_type_error(op.as_str(), lhs_type, lhs_name, rhs_name))
}
}
#[must_use]
pub fn clone_with_heap(&self, heap: &impl ContainsHeap) -> Self {
match self {
Self::Ref(id) => {
heap.heap().inc_ref(*id);
Self::Ref(*id)
}
other => other.clone_immediate(),
}
}
#[cfg(not(feature = "memory-model-checks"))]
#[inline]
pub fn drop_with_heap(self, heap: &mut impl ContainsHeap) {
if let Self::Ref(id) = self {
heap.heap_mut().dec_ref(id);
}
}
#[cfg(feature = "memory-model-checks")]
pub fn drop_with_heap(mut self, heap: &mut impl ContainsHeap) {
let old = mem::replace(&mut self, Self::Dereferenced);
if let Self::Ref(id) = &old {
heap.heap_mut().dec_ref(*id);
mem::forget(old);
}
}
pub fn clone_immediate(&self) -> Self {
match self {
Self::Undefined => Self::Undefined,
Self::Ellipsis => Self::Ellipsis,
Self::None => Self::None,
Self::Bool(b) => Self::Bool(*b),
Self::Int(v) => Self::Int(*v),
Self::Float(v) => Self::Float(*v),
Self::Builtin(b) => Self::Builtin(*b),
Self::ModuleFunction(mf) => Self::ModuleFunction(*mf),
Self::DefFunction(f) => Self::DefFunction(*f),
Self::ExtFunction(f) => Self::ExtFunction(*f),
Self::InternString(s) => Self::InternString(*s),
Self::InternBytes(b) => Self::InternBytes(*b),
Self::InternLongInt(bi) => Self::InternLongInt(*bi),
Self::Marker(m) => Self::Marker(*m),
Self::Property(p) => Self::Property(*p),
Self::Ref(_) => panic!("Ref clones must go through clone_with_heap to maintain refcounts"),
#[cfg(feature = "memory-model-checks")]
Self::Dereferenced => panic!("Cannot copy Dereferenced object"),
}
}
#[cfg(feature = "memory-model-checks")]
pub fn dec_ref_forget(&mut self) {
let old = mem::replace(self, Self::Dereferenced);
mem::forget(old);
}
pub fn py_dec_ref_ids(&mut self, stack: &mut Vec<HeapId>) {
if let Self::Ref(id) = self {
stack.push(*id);
#[cfg(feature = "memory-model-checks")]
self.dec_ref_forget();
}
}
pub fn as_either_str(&self, heap: &Heap<impl ResourceTracker>) -> Option<EitherStr> {
match self {
Self::InternString(id) => Some(EitherStr::Interned(*id)),
Self::Ref(heap_id) => match heap.get(*heap_id) {
HeapData::Str(s) => Some(EitherStr::Heap(s.as_str().to_owned())),
_ => None,
},
_ => None,
}
}
pub(crate) fn to_str<'a>(&'a self, vm: &'a VM<'_, impl ResourceTracker>) -> RunResult<&'a str> {
self.to_str_heap(vm.heap, vm.interns)
}
pub(crate) fn to_str_heap<'a>(
&'a self,
heap: &'a Heap<impl ResourceTracker>,
interns: &'a Interns,
) -> RunResult<&'a str> {
match self {
Self::InternString(string_id) => return Ok(interns.get_str(*string_id)),
Self::Ref(heap_id) => {
if let HeapData::Str(s) = heap.get(*heap_id) {
return Ok(s.as_str());
}
}
_ => {}
}
Err(ExcType::type_error(format!(
"expected string, not {}",
self.py_type_name_heap(heap, interns)
)))
}
pub fn is_str(&self, heap: &Heap<impl ResourceTracker>) -> bool {
match self {
Self::InternString(_) => true,
Self::Ref(heap_id) => matches!(heap.get(*heap_id), HeapData::Str(_)),
_ => false,
}
}
}
pub(crate) fn eq_i64(a: i64, other: &Value, vm: &VM<'_, impl ResourceTracker>) -> Option<bool> {
match other {
Value::Int(b) => Some(a == *b),
Value::Bool(b) => Some(a == i64::from(*b)),
Value::Float(f) => Some(i64_cmp_f64(a, *f) == Some(Ordering::Equal)),
Value::Ref(id) if let HeapData::LongInt(li) = vm.heap.get(*id) => Some(*li.inner() == BigInt::from(a)),
_ => None,
}
}
pub(crate) fn eq_f64(f: f64, other: &Value, vm: &VM<'_, impl ResourceTracker>) -> Option<bool> {
match other {
Value::Float(o) => Some(f == *o),
Value::Int(o) => Some(i64_cmp_f64(*o, f) == Some(Ordering::Equal)),
Value::Bool(o) => Some(i64_cmp_f64(i64::from(*o), f) == Some(Ordering::Equal)),
Value::Ref(id) if let HeapData::LongInt(li) = vm.heap.get(*id) => {
Some(li.partial_cmp_f64(f) == Some(Ordering::Equal))
}
_ => None,
}
}
pub(crate) fn eq_bigint(b: &BigInt, other: &Value, vm: &VM<'_, impl ResourceTracker>) -> Option<bool> {
match other {
Value::Int(o) => Some(*b == BigInt::from(*o)),
Value::Bool(o) => Some(*b == BigInt::from(i64::from(*o))),
Value::Float(f) => Some(bigint_cmp_f64(b, *f) == Some(Ordering::Equal)),
Value::Ref(id) if let HeapData::LongInt(li) = vm.heap.get(*id) => Some(b == li.inner()),
_ => None,
}
}
pub(crate) fn eq_str(s: &str, other: &Value, vm: &VM<'_, impl ResourceTracker>) -> Option<bool> {
match other {
Value::InternString(id) => Some(s == vm.interns.get_str(*id)),
Value::Ref(id) if let HeapData::Str(o) = vm.heap.get(*id) => Some(s == o.as_str()),
_ => None,
}
}
pub(crate) fn eq_bytes(b: &[u8], other: &Value, vm: &VM<'_, impl ResourceTracker>) -> Option<bool> {
match other {
Value::InternBytes(id) => Some(b == vm.interns.get_bytes(*id)),
Value::Ref(id) if let HeapData::Bytes(o) = vm.heap.get(*id) => Some(b == o.as_slice()),
_ => None,
}
}
pub(crate) fn eq_ext_function(name: &str, other: &Value, vm: &VM<'_, impl ResourceTracker>) -> Option<bool> {
match other {
Value::ExtFunction(id) => Some(name == vm.interns.get_str(*id)),
Value::Ref(id) if let HeapData::ExtFunction(o) = vm.heap.get(*id) => Some(name == o.as_str()),
_ => None,
}
}
#[derive(Debug, Clone, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
pub(crate) enum EitherStr {
Interned(StringId),
Heap(String),
}
impl From<StringId> for EitherStr {
fn from(id: StringId) -> Self {
Self::Interned(id)
}
}
impl From<StaticStrings> for EitherStr {
fn from(s: StaticStrings) -> Self {
Self::Interned(s.into())
}
}
impl From<String> for EitherStr {
fn from(s: String) -> Self {
match StaticStrings::from_str(&s) {
Ok(s) => s.into(),
Err(_) => Self::Heap(s),
}
}
}
impl EitherStr {
pub fn as_str<'a>(&'a self, interns: &'a Interns) -> &'a str {
match self {
Self::Interned(id) => interns.get_str(*id),
Self::Heap(s) => s.as_str(),
}
}
pub fn matches(&self, target: StringId, interns: &Interns) -> bool {
match self {
Self::Interned(id) => *id == target,
Self::Heap(s) => s == interns.get_str(target),
}
}
#[inline]
pub fn string_id(&self) -> Option<StringId> {
match self {
Self::Interned(id) => Some(*id),
Self::Heap(_) => None,
}
}
#[inline]
pub fn static_string(&self) -> Option<StaticStrings> {
match self {
Self::Interned(id) => StaticStrings::from_string_id(*id),
Self::Heap(_) => None,
}
}
pub fn into_string(self, interns: &Interns) -> String {
match self {
Self::Interned(id) => interns.get_str(id).to_owned(),
Self::Heap(s) => s,
}
}
pub fn py_estimate_size(&self) -> usize {
match self {
Self::Interned(_) => 0,
Self::Heap(s) => s.capacity(),
}
}
}
#[derive(Debug, Clone, Copy)]
pub enum BitwiseOp {
And,
Or,
Xor,
LShift,
RShift,
}
impl BitwiseOp {
pub fn as_str(self) -> &'static str {
match self {
Self::And => "&",
Self::Or => "|",
Self::Xor => "^",
Self::LShift => "<<",
Self::RShift => ">>",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub(crate) struct Marker(pub StaticStrings);
impl Marker {
pub(crate) fn py_type(self) -> Type {
match self.0 {
StaticStrings::Stdout | StaticStrings::Stderr => Type::TextIOWrapper,
StaticStrings::UnionType => Type::Type,
_ => Type::SpecialForm,
}
}
pub(crate) fn py_repr_fmt(self, f: &mut impl Write) -> fmt::Result {
let s: &'static str = self.0.into();
match self.0 {
StaticStrings::Stdout => f.write_str("<stdout>")?,
StaticStrings::Stderr => f.write_str("<stderr>")?,
StaticStrings::UnionType => f.write_str("<class 'typing.Union'>")?,
_ => write!(f, "typing.{s}")?,
}
Ok(())
}
}
const SINGLETON_ID_TAG: usize = 1usize << (usize::BITS - 1);
const INTERN_STR_ID_TAG: usize = 1usize << (usize::BITS - 2);
const INTERN_BYTES_ID_TAG: usize = 1usize << (usize::BITS - 3);
const HEAP_ID_TAG: usize = 1usize << (usize::BITS - 4);
const INTERN_BYTES_ID_MASK: usize = INTERN_BYTES_ID_TAG - 1;
const INTERN_STR_ID_MASK: usize = INTERN_STR_ID_TAG - 1;
const SINGLETON_ID_MASK: usize = SINGLETON_ID_TAG - 1;
const HEAP_ID_MASK: usize = HEAP_ID_TAG - 1;
const INT_ID_TAG: usize = 1usize << (usize::BITS - 5);
const FLOAT_ID_TAG: usize = 1usize << (usize::BITS - 6);
const BUILTIN_ID_TAG: usize = 1usize << (usize::BITS - 7);
const FUNCTION_ID_TAG: usize = 1usize << (usize::BITS - 8);
const EXTFUNCTION_ID_TAG: usize = 1usize << (usize::BITS - 9);
const MARKER_ID_TAG: usize = 1usize << (usize::BITS - 10);
const MODULE_FUNCTION_ID_TAG: usize = 1usize << (usize::BITS - 12);
const INTERN_LONG_INT_ID_TAG: usize = 1usize << (usize::BITS - 13);
const PROPERTY_ID_TAG: usize = 1usize << (usize::BITS - 14);
const INT_ID_MASK: usize = INT_ID_TAG - 1;
const FLOAT_ID_MASK: usize = FLOAT_ID_TAG - 1;
const BUILTIN_ID_MASK: usize = BUILTIN_ID_TAG - 1;
const FUNCTION_ID_MASK: usize = FUNCTION_ID_TAG - 1;
const EXTFUNCTION_ID_MASK: usize = EXTFUNCTION_ID_TAG - 1;
const MARKER_ID_MASK: usize = MARKER_ID_TAG - 1;
const MODULE_FUNCTION_ID_MASK: usize = MODULE_FUNCTION_ID_TAG - 1;
const INTERN_LONG_INT_ID_MASK: usize = INTERN_LONG_INT_ID_TAG - 1;
const PROPERTY_ID_MASK: usize = PROPERTY_ID_TAG - 1;
#[repr(usize)]
#[derive(Copy, Clone)]
enum SingletonSlot {
Undefined = 0,
Ellipsis = 1,
None = 2,
False = 3,
True = 4,
}
#[inline]
const fn singleton_id(slot: SingletonSlot) -> usize {
SINGLETON_ID_TAG | ((slot as usize) & SINGLETON_ID_MASK)
}
pub(crate) fn floor_divmod(a: i64, b: i64) -> Option<(i64, i64)> {
let quot = a.checked_div(b)?;
let rem = a.checked_rem(b)?;
if rem != 0 && (rem < 0) != (b < 0) {
Some((quot - 1, rem + b))
} else {
Some((quot, rem))
}
}
#[inline]
pub fn heap_tagged_id(heap_id: HeapId) -> usize {
HEAP_ID_TAG | (heap_id.index() & HEAP_ID_MASK)
}
#[inline]
fn int_value_id(value: i64) -> usize {
let mut hasher = DefaultHasher::new();
value.hash(&mut hasher);
let hash_u64 = hasher.finish();
let masked = hash_u64 & (usize::MAX as u64);
let hash_usize = usize::try_from(masked).expect("masked value fits in usize");
INT_ID_TAG | (hash_usize & INT_ID_MASK)
}
#[inline]
fn float_value_id(value: f64) -> usize {
let mut hasher = DefaultHasher::new();
value.to_bits().hash(&mut hasher);
let hash_u64 = hasher.finish();
let masked = hash_u64 & (usize::MAX as u64);
let hash_usize = usize::try_from(masked).expect("masked value fits in usize");
FLOAT_ID_TAG | (hash_usize & FLOAT_ID_MASK)
}
#[inline]
fn builtin_value_id(b: Builtins) -> usize {
let mut hasher = DefaultHasher::new();
b.hash(&mut hasher);
let hash_u64 = hasher.finish();
#[expect(clippy::cast_possible_truncation)]
let hash_usize = hash_u64 as usize;
BUILTIN_ID_TAG | (hash_usize & BUILTIN_ID_MASK)
}
#[inline]
fn function_value_id(f_id: FunctionId) -> usize {
FUNCTION_ID_TAG | (f_id.index() & FUNCTION_ID_MASK)
}
#[inline]
fn ext_function_value_id(name: &str) -> usize {
let hash_u64 = hash_python_str(name).raw();
let masked = hash_u64 & (usize::MAX as u64);
let hash_usize = usize::try_from(masked).expect("masked value fits in usize");
EXTFUNCTION_ID_TAG | (hash_usize & EXTFUNCTION_ID_MASK)
}
#[inline]
fn marker_value_id(m: Marker) -> usize {
MARKER_ID_TAG | ((m.0 as usize) & MARKER_ID_MASK)
}
#[inline]
fn property_value_id(p: Property) -> usize {
let discriminant = match p {
Property::Os(os_fn) => os_fn as usize,
};
PROPERTY_ID_TAG | (discriminant & PROPERTY_ID_MASK)
}
#[inline]
fn module_function_value_id(mf: ModuleFunctions) -> usize {
let mut hasher = DefaultHasher::new();
mf.hash(&mut hasher);
let hash_u64 = hasher.finish();
#[expect(clippy::cast_possible_truncation)]
let hash_usize = hash_u64 as usize;
MODULE_FUNCTION_ID_TAG | (hash_usize & MODULE_FUNCTION_ID_MASK)
}
#[inline]
fn i64_to_repeat_count(n: i64) -> RunResult<usize> {
if n <= 0 {
Ok(0)
} else {
usize::try_from(n).map_err(|_| ExcType::overflow_repeat_count().into())
}
}
#[inline]
fn longint_to_repeat_count(li: &LongInt) -> RunResult<usize> {
if li.is_negative() {
Ok(0)
} else if let Some(count) = li.to_usize() {
Ok(count)
} else {
Err(ExcType::overflow_repeat_count().into())
}
}
fn extract_bigint(value: &Value, heap: &Heap<impl ResourceTracker>) -> Option<BigInt> {
match value {
Value::Int(i) => Some(BigInt::from(*i)),
Value::Bool(b) => Some(BigInt::from(i64::from(*b))),
Value::Ref(id) => {
if let HeapData::LongInt(li) = heap.get(*id) {
Some(li.inner().clone())
} else {
None
}
}
_ => None,
}
}
fn cloned_items_view_candidate(item: &Value, heap: &impl ContainsHeap) -> Option<(Value, Value)> {
let Value::Ref(heap_id) = item else {
return None;
};
match heap.heap().get(*heap_id) {
HeapData::Tuple(tuple) => {
let items = tuple.as_slice();
if items.len() == 2 {
Some((items[0].clone_with_heap(heap), items[1].clone_with_heap(heap)))
} else {
None
}
}
HeapData::NamedTuple(namedtuple) => {
let items = namedtuple.as_vec();
if items.len() == 2 {
Some((items[0].clone_with_heap(heap), items[1].clone_with_heap(heap)))
} else {
None
}
}
_ => None,
}
}
fn str_contains(
container_str: &str,
item: &Value,
heap: &Heap<impl ResourceTracker>,
interns: &Interns,
) -> RunResult<bool> {
match item {
Value::InternString(item_id) => {
let item_str = interns.get_str(*item_id);
Ok(container_str.contains(item_str))
}
Value::Ref(item_heap_id) => {
if let HeapData::Str(item_str) = heap.get(*item_heap_id) {
Ok(container_str.contains(item_str.as_str()))
} else {
Err(ExcType::type_error("'in <str>' requires string as left operand"))
}
}
_ => Err(ExcType::type_error("'in <str>' requires string as left operand")),
}
}
fn i64_bits(value: i64) -> u64 {
if value == 0 {
0
} else {
u64::from(64 - value.unsigned_abs().leading_zeros())
}
}
fn bigint_pow(base: BigInt, exp: u64) -> BigInt {
if exp == 0 {
return BigInt::from(1);
}
if exp == 1 {
return base;
}
let mut result = BigInt::from(1);
let mut b = base;
let mut e = exp;
while e > 0 {
if e & 1 == 1 {
result *= &b;
}
b = &b * &b;
e >>= 1;
}
result
}
#[cfg(test)]
mod tests {
use num_bigint::BigInt;
use super::*;
use crate::{PrintWriter, heap::HeapReader, intern::InternerBuilder, resource::NoLimitTracker};
fn create_heap_with_longint(value: BigInt) -> (Heap<NoLimitTracker>, HeapId) {
let heap = Heap::new(16, NoLimitTracker);
let long_int = LongInt::new(value);
let heap_id = heap.allocate(HeapData::LongInt(long_int)).unwrap();
(heap, heap_id)
}
fn create_test_interns() -> Interns {
let interner = InternerBuilder::new("");
Interns::new(interner, vec![])
}
#[test]
fn as_index_longint_fits_in_i64() {
let (mut heap, heap_id) = create_heap_with_longint(BigInt::from(42));
let value = Value::Ref(heap_id);
let mut interns = create_test_interns();
let result = HeapReader::with(&mut heap, &mut interns, |reader, interns| {
let vm = VM::new(Vec::new(), reader, interns, PrintWriter::Disabled);
value.as_index(&vm, Type::List)
});
assert_eq!(result.unwrap(), 42);
value.drop_with_heap(&mut heap);
}
#[test]
fn as_index_longint_negative_fits_in_i64() {
let (mut heap, heap_id) = create_heap_with_longint(BigInt::from(-100));
let value = Value::Ref(heap_id);
let mut interns = create_test_interns();
let result = HeapReader::with(&mut heap, &mut interns, |reader, interns| {
let vm = VM::new(Vec::new(), reader, interns, PrintWriter::Disabled);
value.as_index(&vm, Type::List)
});
assert_eq!(result.unwrap(), -100);
value.drop_with_heap(&mut heap);
}
#[test]
fn as_index_longint_too_large() {
let big_value = BigInt::from(2).pow(100);
let (mut heap, heap_id) = create_heap_with_longint(big_value);
let value = Value::Ref(heap_id);
let mut interns = create_test_interns();
let result = HeapReader::with(&mut heap, &mut interns, |reader, interns| {
let vm = VM::new(Vec::new(), reader, interns, PrintWriter::Disabled);
value.as_index(&vm, Type::List)
});
assert!(result.is_err());
value.drop_with_heap(&mut heap);
}
#[test]
fn as_int_longint_fits_in_i64() {
let (mut heap, heap_id) = create_heap_with_longint(BigInt::from(12345));
let value = Value::Ref(heap_id);
let mut interns = create_test_interns();
let result = HeapReader::with(&mut heap, &mut interns, |reader, interns| {
let vm = VM::new(Vec::new(), reader, interns, PrintWriter::Disabled);
value.as_int(&vm)
});
assert_eq!(result.unwrap(), 12345);
value.drop_with_heap(&mut heap);
}
#[test]
fn as_int_longint_too_large() {
let big_value = BigInt::from(2).pow(100);
let (mut heap, heap_id) = create_heap_with_longint(big_value);
let value = Value::Ref(heap_id);
let mut interns = create_test_interns();
let result = HeapReader::with(&mut heap, &mut interns, |reader, interns| {
let vm = VM::new(Vec::new(), reader, interns, PrintWriter::Disabled);
value.as_int(&vm)
});
assert!(result.is_err());
value.drop_with_heap(&mut heap);
}
#[test]
fn as_index_longint_at_i64_max() {
let (mut heap, heap_id) = create_heap_with_longint(BigInt::from(i64::MAX));
let value = Value::Ref(heap_id);
let mut interns = create_test_interns();
let result = HeapReader::with(&mut heap, &mut interns, |reader, interns| {
let vm = VM::new(Vec::new(), reader, interns, PrintWriter::Disabled);
value.as_index(&vm, Type::List)
});
assert_eq!(result.unwrap(), i64::MAX);
value.drop_with_heap(&mut heap);
}
#[test]
fn as_index_longint_at_i64_min() {
let (mut heap, heap_id) = create_heap_with_longint(BigInt::from(i64::MIN));
let value = Value::Ref(heap_id);
let mut interns = create_test_interns();
let result = HeapReader::with(&mut heap, &mut interns, |reader, interns| {
let vm = VM::new(Vec::new(), reader, interns, PrintWriter::Disabled);
value.as_index(&vm, Type::List)
});
assert_eq!(result.unwrap(), i64::MIN);
value.drop_with_heap(&mut heap);
}
#[test]
fn as_index_longint_just_over_i64_max() {
let big_value = BigInt::from(i64::MAX) + BigInt::from(1);
let (mut heap, heap_id) = create_heap_with_longint(big_value);
let value = Value::Ref(heap_id);
let mut interns = create_test_interns();
let result = HeapReader::with(&mut heap, &mut interns, |reader, interns| {
let vm = VM::new(Vec::new(), reader, interns, PrintWriter::Disabled);
value.as_index(&vm, Type::List)
});
assert!(result.is_err());
value.drop_with_heap(&mut heap);
}
#[test]
fn as_index_longint_just_under_i64_min() {
let big_value = BigInt::from(i64::MIN) - BigInt::from(1);
let (mut heap, heap_id) = create_heap_with_longint(big_value);
let value = Value::Ref(heap_id);
let mut interns = create_test_interns();
let result = HeapReader::with(&mut heap, &mut interns, |reader, interns| {
let vm = VM::new(Vec::new(), reader, interns, PrintWriter::Disabled);
value.as_index(&vm, Type::List)
});
assert!(result.is_err());
value.drop_with_heap(&mut heap);
}
}