Split InterpreterFrame from FrameObject for stack-allocated execution - #8354

unsafe impl Traverse for PyBaseException {
    fn traverse(&self, tracer_fn: &mut TraverseFn<'_>) {
        self.traceback.traverse(tracer_fn);
        self.cause.traverse(tracer_fn);
        self.context.traverse(tracer_fn);
        self.args.traverse(tracer_fn);
    }
}

pub enum ExceptionCtor {
    Class(PyTypeRef),
    Instance(PyBaseExceptionRef),
}

impl TryFromObject for ExceptionCtor {
    fn try_from_object(vm: &VirtualMachine, obj: PyObjectRef) -> PyResult<Self> {
        obj.downcast::<PyType>()
            .and_then(|cls| {
                if cls.fast_issubclass(vm.ctx.exceptions.base_exception_type) {
                    Ok(Self::Class(cls))
                } else {
                    Err(cls.into())
                }
            })
            .or_else(|obj| obj.downcast::<PyBaseException>().map(Self::Instance))
            .map_err(|obj| {
                vm.new_type_error(format!(
                    "exceptions must be classes or instances deriving from BaseException, not {}",
                    obj.class().name()
                ))
            })
    }
}

impl ExceptionCtor {
    pub fn instantiate(self, vm: &VirtualMachine) -> PyResult<PyBaseExceptionRef> {
        match self {
            Self::Class(cls) => vm.invoke_exception(&cls, vec![]),
            Self::Instance(exc) => Ok(exc),
        }
    }

    pub fn instantiate_value(
        self,
        value: PyObjectRef,
        vm: &VirtualMachine,
    ) -> PyResult<PyBaseExceptionRef> {
        let exc_inst = value.clone().downcast::<PyBaseException>().ok();
        match (self, exc_inst) {
            // both are instances; which would we choose?
            (Self::Instance(_exc_a), Some(_exc_b)) => {
                Err(vm.new_type_error("instance exception may not have a separate value"))
            }
            // if the "type" is an instance and the value isn't, use the "type"
            (Self::Instance(exc), None) => Ok(exc),
            // if the value is an instance of the type, use the instance value
            (Self::Class(cls), Some(exc)) if exc.fast_isinstance(&cls) => Ok(exc),
            // otherwise; construct an exception of the type using the value as args
            (Self::Class(cls), _) => {
                let args = match_class!(match value {
                    PyNone => vec![],
                    tup @ PyTuple => tup.to_vec(),
                    exc @ PyBaseException => exc.args().to_vec(),
                    obj => vec![obj],
                });
                vm.invoke_exception(&cls, args)
            }
        }
    }
}

pub struct PyBaseException {
        pub(super) traceback: PyRwLock<Option<PyTracebackRef>>,
        pub(super) cause: PyRwLock<Option<PyRef<Self>>>,
        pub(super) context: PyRwLock<Option<PyRef<Self>>>,
        pub(super) suppress_context: AtomicCell<bool>,
        pub(super) args: PyRwLock<PyTupleRef>,
    }

impl PyBaseException {
    pub fn get_arg(&self, idx: usize) -> Option<PyObjectRef> {
        self.args.read().get(idx).cloned()
    }
}

pub(crate) fn exception_group_match(
    exc_value: &PyObjectRef,
    match_type: &PyObjectRef,
    vm: &VirtualMachine,
) -> PyResult<(PyObjectRef, PyObjectRef)> {
    // Implements _PyEval_ExceptionGroupMatch

    // If exc_value is None, return (None, None)
    if vm.is_none(exc_value) {
        return Ok((vm.ctx.none(), vm.ctx.none()));
    }

    // Validate match_type and reject ExceptionGroup/BaseExceptionGroup
    check_except_star_type_valid(match_type, vm)?;

    // Check if exc_value matches match_type
    if exc_value.is_instance(match_type, vm)? {
        // Full match of exc itself
        let is_eg = exc_value.fast_isinstance(vm.ctx.exceptions.base_exception_group);
        let matched = if is_eg {
            exc_value.clone()
        } else {
            // Naked exception - wrap it in ExceptionGroup
            let excs = vm.ctx.new_tuple(vec![exc_value.clone()]);
            let eg_type: PyObjectRef = crate::exception_group::exception_group().to_owned().into();
            let wrapped = eg_type.call((vm.ctx.new_str(""), excs), vm)?;
            // Copy traceback from original exception
            if let Ok(exc) = exc_value.clone().downcast::<types::PyBaseException>()
                && let Some(tb) = exc.__traceback__()
                && let Ok(wrapped_exc) = wrapped.clone().downcast::<types::PyBaseException>()
            {
                let _ = wrapped_exc.set___traceback__(tb.into(), vm);
            }
            wrapped
        };
        return Ok((vm.ctx.none(), matched));
    }

    // Check for partial match if it's an exception group
    if exc_value.fast_isinstance(vm.ctx.exceptions.base_exception_group) {
        let pair = vm.call_method(exc_value, "split", (match_type.clone(),))?;
        if !pair.class().is(vm.ctx.types.tuple_type) {
            return Err(vm.new_type_error(format!(
                "{}.split must return a tuple, not {}",
                exc_value.class().name(),
                pair.class().name()
            )));
        }
        let pair_tuple: PyTupleRef = pair.try_into_value(vm)?;
        if pair_tuple.len() < 2 {
            return Err(vm.new_type_error(format!(
                "{}.split must return a 2-tuple, got tuple of size {}",
                exc_value.class().name(),
                pair_tuple.len()
            )));
        }
        let matched = pair_tuple[0].clone();
        let rest = pair_tuple[1].clone();
        return Ok((rest, matched));
    }

    // No match
    Ok((exc_value.clone(), vm.ctx.none()))
}

pub fn prep_reraise_star(orig: PyObjectRef, excs: PyObjectRef, vm: &VirtualMachine) -> PyResult {
    use crate::builtins::PyList;

    let excs_list = excs
        .downcast::<PyList>()
        .map_err(|_| vm.new_type_error("expected list for prep_reraise_star"))?;

    let excs_vec: Vec<PyObjectRef> = excs_list.borrow_vec().to_vec();

    // If no exceptions to process, return None
    if excs_vec.is_empty() {
        return Ok(vm.ctx.none());
    }

    // Special case: naked exception (not an ExceptionGroup)
    // Only one except* clause could have executed, so there's at most one exception to raise
    if !orig.fast_isinstance(vm.ctx.exceptions.base_exception_group) {
        // Find first non-None exception
        let first = excs_vec.into_iter().find(|e| !vm.is_none(e));
        return Ok(first.unwrap_or_else(|| vm.ctx.none()));
    }

    // Split excs into raised (new) and reraised (from original) by comparing metadata
    let mut raised: Vec<PyObjectRef> = Vec::new();
    let mut reraised: Vec<PyObjectRef> = Vec::new();

    for exc in excs_vec {
        if vm.is_none(&exc) {
            continue;
        }
        // Check if this exception came from the original group
        if is_exception_from_orig(&exc, &orig, vm) {
            reraised.push(exc);
        } else {
            raised.push(exc);
        }
    }

    // If no exceptions to reraise, return None
    if raised.is_empty() && reraised.is_empty() {
        return Ok(vm.ctx.none());
    }

    // Project reraised exceptions onto original structure to preserve nesting
    let reraised_eg = exception_group_projection(&orig, &reraised, vm)?;

    // If no new raised exceptions, just return the reraised projection
    if raised.is_empty() {
        return Ok(reraised_eg);
    }

    // Combine raised with reraised_eg
    if !vm.is_none(&reraised_eg) {
        raised.push(reraised_eg);
    }

    // If only one exception, return it directly
    if raised.len() == 1 {
        return Ok(raised.into_iter().next().unwrap());
    }

    // Create new ExceptionGroup for multiple exceptions
    let excs_tuple = vm.ctx.new_tuple(raised);
    let eg_type: PyObjectRef = crate::exception_group::exception_group().to_owned().into();
    eg_type.call((vm.ctx.new_str(""), excs_tuple), vm)
}

impl VirtualMachine {
    /// Bump-allocate `size` bytes from the thread data stack.
    ///
    /// # Safety
    /// The returned pointer must be freed by calling `datastack_pop` in LIFO order.
    #[inline(always)]
    pub(crate) fn datastack_push(&self, size: usize) -> *mut u8 {
        unsafe { (*self.datastack.get()).push(size) }
    }

    /// Check whether the thread data stack currently has room for `size` bytes.
    #[inline(always)]
    pub(crate) fn datastack_has_space(&self, size: usize) -> bool {
        unsafe { (*self.datastack.get()).has_space(size) }
    }

    /// Pop a previous data stack allocation.
    ///
    /// # Safety
    /// `base` must be a pointer returned by `datastack_push` on this VM,
    /// and all allocations made after it must already have been popped.
    #[inline(always)]
    pub(crate) unsafe fn datastack_pop(&self, base: *mut u8) {
        unsafe { (*self.datastack.get()).pop(base) }
    }

    /// Temporarily detach the current thread (ATTACHED → DETACHED) while
    /// running `f`, then re-attach afterwards.  Allows `stop_the_world` to
    /// park this thread during blocking syscalls.
    #[inline]
    pub fn allow_threads<R>(&self, f: impl FnOnce() -> R) -> R {
        thread::allow_threads(self, f)
    }
}