rustpython-vm 0.6.0

#[cfg(all(not(unix), feature = "threading"))]
use super::FramePtr;
#[cfg(feature = "threading")]
use crate::PyObjectRef;
#[cfg(feature = "threading")]
use crate::builtins::PyBaseExceptionRef;
#[cfg(feature = "threading")]
use alloc::sync::Arc;
#[cfg(feature = "threading")]
use rustpython_common::lock::PyMutex;

use crate::frame::InterpreterFrame;
#[cfg(feature = "threading")]
use crate::vm::PyGlobalState;
use crate::{AsObject, PyObject, VirtualMachine};
#[cfg(all(unix, feature = "threading"))]
use crate::{Py, frame::FrameObject};
#[cfg(all(unix, feature = "threading"))]
use core::sync::atomic::AtomicPtr;
use core::{
    cell::{Cell, RefCell},
    ptr::NonNull,
    sync::atomic::{AtomicUsize, Ordering},
};
use itertools::Itertools;
#[cfg(feature = "threading")]
use std::collections::HashMap;
use std::thread_local;

/// Thread states for stop-the-world support (`_Py_THREAD_*`).
///
/// DETACHED: not executing Python bytecode (in native code, or idle)
/// ATTACHED: actively executing Python bytecode
/// SUSPENDED: parked by a stop-the-world request
/// SHUTTING_DOWN: interpreter is finalizing; the OS thread must hang
/// (`_PyThreadState_HangThread`) and must not look done to `_ThreadHandle`.
#[cfg(feature = "threading")]
#[repr(i32)]
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub enum ThreadState {
    Detached = 0,
    Attached = 1,
    Suspended = 2,
    ShuttingDown = 3,
}

#[cfg(feature = "threading")]
impl ThreadState {
    #[must_use]
    pub const fn from_i32(v: i32) -> Option<Self> {
        match v {
            0 => Some(Self::Detached),
            1 => Some(Self::Attached),
            2 => Some(Self::Suspended),
            3 => Some(Self::ShuttingDown),
            _ => None,
        }
    }
}

/// Per-thread shared state for sys._current_frames() and sys._current_exceptions().
/// The exception field uses atomic operations for lock-free cross-thread reads.
#[cfg(feature = "threading")]
pub struct ThreadSlot {
    /// Top of the owning thread's Python call stack, published for
    /// cross-thread readers (`sys._current_frames`, cross-thread `f_back`).
    /// The rest of the stack is reachable via each frame's `previous` pointer.
    /// Written lock-free on the hot push/pop path with relaxed ordering; every
    /// cross-thread read runs under stop-the-world, which parks the owning
    /// thread at a safepoint and supplies the happens-before edge, so the
    /// pointer and the frames it reaches are quiescent and alive at read time.
    #[cfg(unix)]
    pub top_frame: AtomicPtr<Py<FrameObject>>,
    /// Raw InterpreterFrame pointer, published alongside top_frame so
    /// cross-thread readers (sys._current_frames) can materialize
    /// stack-allocated frames that have no FrameObject.
    pub top_iframe: AtomicUsize,
    /// Raw frame pointers, valid while the owning thread's call stack is active.
    /// Readers must hold the Mutex and convert to FrameObjectRef inside the lock.
    /// Stands in for `top_frame` where that field is not built, so a reader
    /// that finds no `top_iframe` still has the frames to answer from.
    #[cfg(not(unix))]
    pub frames: parking_lot::Mutex<Vec<FramePtr>>,
    pub exception: crate::PyAtomicRef<Option<crate::exceptions::types::PyBaseException>>,
    /// `tstate->c_traceobj`. Cross-thread source of truth for sys.gettrace /
    /// sys._settraceallthreads.
    pub trace_func: PyMutex<PyObjectRef>,
    /// `tstate->c_profileobj`. Cross-thread source of truth for sys.getprofile /
    /// sys._setprofileallthreads.
    pub profile_func: PyMutex<PyObjectRef>,
    /// Thread state for stop-the-world: DETACHED / ATTACHED / SUSPENDED / SHUTTING_DOWN
    pub state: core::sync::atomic::AtomicI32,
    /// Per-thread stop request bit (eval breaker equivalent).
    pub stop_requested: core::sync::atomic::AtomicBool,
    /// Handle for waking this thread from park in stop-the-world paths.
    pub thread: std::thread::Thread,
    /// QSBR state for deferred memory reclamation.
    pub(crate) qsbr: Arc<crate::object::qsbr::QsbrSlot>,
}

#[cfg(feature = "threading")]
pub type CurrentFrameSlot = Arc<ThreadSlot>;

/// Coalesced per-thread frame-publishing state, touched on every
/// `enter_iframe`/`exit_iframe`. Bundling `current_frame` together with
/// the cached `top_frame`/`top_iframe` slot pointers means
/// `set_current_frame` needs a single `thread_local!.with()` call
/// (one `_tlv_get_addr` on macOS) instead of two or three separate
/// ones — each `.with()` on a distinct `thread_local!` is its own TLS
/// lookup even though the bodies are just a cached-pointer store.
struct FrameSlotCache {
    /// Current top frame for signal-safe traceback walking.
    /// Stores a `*const InterpreterFrame` as `usize`.
    /// Read by faulthandler's signal handler to dump tracebacks without
    /// accessing RefCell or locks. Uses AtomicUsize for async-signal-safety.
    current_frame: AtomicUsize,
    /// Cached pointer to this thread's `ThreadSlot::top_frame`, so the hot
    /// push/pop path can publish the top frame with a single relaxed store and
    /// no `CURRENT_THREAD_SLOT` RefCell borrow. Null until the slot is
    /// initialized; the `Arc<ThreadSlot>` in `CURRENT_THREAD_SLOT` keeps the
    /// pointee alive until `cleanup_current_thread_frames` clears this.
    #[cfg(all(unix, feature = "threading"))]
    top_frame: Cell<*const AtomicPtr<Py<FrameObject>>>,
    /// Cached pointer to this thread's `ThreadSlot::top_iframe` for the hot
    /// light-frame push/pop path. The slot's Arc keeps the pointee alive.
    #[cfg(feature = "threading")]
    top_iframe: Cell<*const AtomicUsize>,
}

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

    /// Thread state created through the GILState-style C API.
    ///
    /// This is separate from the current VM stack: it only means "attached now",
    /// while this owns the per-thread VM that may be detached and re-attached.
    /// Despite the historical CPython "GILState" name, this does not model a
    /// GIL; it stores the VM used by that compatibility API.
    ///
    /// The Box keeps the VM address stable while VM_STACK holds a raw pointer to it.
    /// This matters when release_current_thread() moves the owner out of TLS and
    /// drops it while the VM is still current, so object destructors can still find
    /// their VM.
    #[cfg(feature = "threading")]
    static GILSTATE_VM: RefCell<Option<Box<ThreadedVirtualMachine>>> = const { RefCell::new(None) };

    pub(crate) static COROUTINE_ORIGIN_TRACKING_DEPTH: Cell<u32> = const { Cell::new(0) };

    /// Per-interpreter thread slots for this OS thread (PEP 734 multi-interpreter).
    ///
    /// CPython keeps a `PyThreadState` per (thread, interpreter) pair. RustPython
    /// mirrors that: each interpreter's `PyGlobalState.thread_frames` gets its own
    /// [`ThreadSlot`] for this OS thread. `CURRENT_THREAD_SLOT` always points at
    /// the slot for the currently entered interpreter.
    #[cfg(feature = "threading")]
    static INTERP_THREAD_SLOTS: RefCell<HashMap<i64, CurrentFrameSlot>> =
        RefCell::new(HashMap::new());

    /// Current thread's slot for the currently entered interpreter.
    #[cfg(feature = "threading")]
    static CURRENT_THREAD_SLOT: RefCell<Option<CurrentFrameSlot>> = const { RefCell::new(None) };

    pub(crate) static FRAME_SLOT_CACHE: FrameSlotCache = const {
        FrameSlotCache {
            current_frame: AtomicUsize::new(0),
            #[cfg(all(unix, feature = "threading"))]
            top_frame: Cell::new(core::ptr::null()),
            #[cfg(feature = "threading")]
            top_iframe: Cell::new(core::ptr::null()),
        }
    };

    /// Cached pointer to this thread's `ThreadSlot::stop_requested`, for the
    /// safepoint the dispatch loop takes once per instruction. Reading it
    /// through `CURRENT_THREAD_SLOT` costs a `RefCell` borrow — two stores to
    /// thread-local memory — where this costs one relaxed load. The slot's Arc
    /// keeps the pointee alive, as with the frame pointers above.
    #[cfg(feature = "threading")]
    static CURRENT_STOP_REQUESTED: Cell<*const core::sync::atomic::AtomicBool> =
        const { Cell::new(core::ptr::null()) };

}

#[must_use]
pub fn current_vm_is_set() -> bool {
    VM_STACK.with(|vms| !vms.borrow().is_empty())
}

pub fn with_current_vm<R>(f: impl FnOnce(&VirtualMachine) -> R) -> R {
    VM_STACK.with(|vms| {
        let vm = vms
            .borrow()
            .last()
            .copied()
            .expect("call with_current_vm() but no current VM is attached");
        // SAFETY: entries in VM_STACK either borrow a VM for the dynamic
        // scope of a set_current_vm()/enter_vm() call or point at GILSTATE_VM.
        f(unsafe { vm.as_ref() })
    })
}

fn set_current_vm<R>(vm: &VirtualMachine, f: impl FnOnce() -> R) -> R {
    // Attach to this VM's interpreter, detaching the enclosing one if this is a
    // switch between interpreters on the same OS thread.
    #[cfg(feature = "threading")]
    let switched = begin_interpreter_section(vm);

    VM_STACK.with(|vms| {
        vms.borrow_mut().push(vm.into());
        scopeguard::defer! {
            vms.borrow_mut().pop();
            #[cfg(feature = "threading")]
            end_interpreter_section(switched);
        }
        f()
    })
}

/// Pointer to the GC state of the interpreter running on this thread.
///
/// The pointee belongs to the `PyGlobalState` of the VM on top of `VM_STACK`,
/// which is borrowed for the whole `set_current_vm` scope — so the pointer stays
/// valid as long as the caller remains inside that scope.
pub(crate) fn current_gc_state() -> Option<NonNull<crate::gc_state::GcInterpreterState>> {
    // Reached from every tracked allocation, including ones a thread-local
    // destructor makes while the VM stack is being torn down, so neither a
    // destroyed key nor an outstanding borrow may panic here.
    VM_STACK
        .try_with(|vms| {
            let vm = vms.try_borrow().ok()?.last().copied()?;
            // SAFETY: entries in VM_STACK either borrow a VM for the dynamic
            // scope of a set_current_vm()/enter_vm() call or point at GILSTATE_VM.
            Some(NonNull::from(&unsafe { vm.as_ref() }.state.gc))
        })
        .ok()
        .flatten()
}

pub fn try_with_current_vm<R>(f: impl FnOnce(&VirtualMachine) -> R) -> Option<R> {
    VM_STACK.with(|vms| {
        let vm = vms.borrow().last().copied()?;
        // SAFETY: entries in VM_STACK either borrow a VM for the dynamic
        // scope of a set_current_vm()/enter_vm() call or point at GILSTATE_VM.
        Some(f(unsafe { vm.as_ref() }))
    })
}

pub fn enter_vm<R>(vm: &VirtualMachine, f: impl FnOnce() -> R) -> R {
    // Attach/detach is handled by `set_current_vm`, which pairs it with the
    // VM_STACK push so that switching interpreters mid-stack stays consistent.
    set_current_vm(vm, f)
}

/// RAII counterpart to `enter_vm`, for code that runs Python bytecode across
/// several statements interspersed with `&mut VirtualMachine` calls
/// (`VirtualMachine::initialize`), where a single closure-based `enter_vm`
/// scope cannot be expressed because the borrow checker won't let a closure
/// hold `&mut VirtualMachine` at the same time `enter_vm` reborrows it as
/// `&VirtualMachine`. Construction only needs a transient `&VirtualMachine`
/// borrow, so it can be dropped before subsequent `&mut` use.
///
/// Without this, code that runs Python bytecode before any `enter_vm` scope
/// exists would leave the thread not ATTACHED, making lock-free type cache
/// reads unsound.
#[must_use]
pub(crate) struct VmBootstrapGuard {
    #[cfg(feature = "threading")]
    switched: bool,
}

impl VmBootstrapGuard {
    pub(crate) fn new(vm: &VirtualMachine) -> Self {
        #[cfg(feature = "threading")]
        let switched = begin_interpreter_section(vm);

        VM_STACK.with(|vms| vms.borrow_mut().push(vm.into()));

        Self {
            #[cfg(feature = "threading")]
            switched,
        }
    }
}

impl Drop for VmBootstrapGuard {
    fn drop(&mut self) {
        VM_STACK.with(|vms| {
            vms.borrow_mut().pop();
        });

        #[cfg(feature = "threading")]
        end_interpreter_section(self.switched);
    }
}

#[cfg(feature = "threading")]
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum CurrentVmAttachState {
    AlreadyAttached,
    Attached,
}

/// State preserved while the current native thread is detached from its VM.
#[cfg(feature = "threading")]
pub struct SavedThreadState {
    vm_stack: Vec<NonNull<VirtualMachine>>,
    gilstate_vm: Option<Box<ThreadedVirtualMachine>>,
}

/// Detach the current native thread and preserve its VM context for restoration.
#[cfg(feature = "threading")]
#[must_use = "the saved thread state must be restored"]
pub fn save_current_thread() -> SavedThreadState {
    let vm_stack = VM_STACK.with(|vms| core::mem::take(&mut *vms.borrow_mut()));
    assert!(
        !vm_stack.is_empty(),
        "save_current_thread() called without an attached VM"
    );
    let gilstate_vm = GILSTATE_VM.with(|gilstate_vm| gilstate_vm.borrow_mut().take());
    detach_thread();
    SavedThreadState {
        vm_stack,
        gilstate_vm,
    }
}

/// Restore a VM context previously returned by [`save_current_thread`].
#[cfg(feature = "threading")]
pub fn restore_current_thread(state: SavedThreadState) {
    assert!(
        !current_vm_is_set(),
        "restore_current_thread() called with an attached VM"
    );
    let SavedThreadState {
        vm_stack,
        gilstate_vm,
    } = state;
    let vm = vm_stack
        .last()
        .copied()
        .expect("saved thread state has no VM");

    GILSTATE_VM.with(|current| {
        let mut current = current.borrow_mut();
        assert!(
            current.is_none(),
            "restore_current_thread() called with a GILState VM"
        );
        *current = gilstate_vm;
    });

    // SAFETY: borrowed VMs remain alive for the dynamic save/restore scope,
    // while an owned GILState VM was restored above before this dereference.
    let vm = unsafe { vm.as_ref() };
    // Point CURRENT_THREAD_SLOT at the restored interpreter before attach.
    // After subinterpreter bootstrap, CURRENT may still refer to the temporary
    // subinterpreter slot (DETACHED); attaching that would leave the parent
    // slot detached and later confuse outermost detach.
    init_thread_slot_if_needed(vm);
    attach_thread(vm);
    VM_STACK.with(|vms| *vms.borrow_mut() = vm_stack);
}

/// Attach the current native thread to a RustPython VM until
/// `release_current_thread()` is called.
#[cfg(feature = "threading")]
pub fn attach_current_thread(
    make_vm: impl FnOnce() -> ThreadedVirtualMachine,
) -> CurrentVmAttachState {
    if current_vm_is_set() {
        return CurrentVmAttachState::AlreadyAttached;
    }

    GILSTATE_VM.with(|gilstate_vm| {
        let mut gilstate_vm = gilstate_vm.borrow_mut();
        let threaded_vm = gilstate_vm.get_or_insert_with(|| Box::new(make_vm()));
        let vm = &threaded_vm.vm;

        vm.c_stack_soft_limit
            .set(VirtualMachine::calculate_c_stack_soft_limit());

        init_thread_slot_if_needed(vm);

        attach_thread(vm);

        VM_STACK.with(|vms| {
            debug_assert!(vms.borrow().is_empty());
            vms.borrow_mut().push(vm.into());
        });
    });

    CurrentVmAttachState::Attached
}

#[cfg(feature = "threading")]
pub fn release_current_thread(state: CurrentVmAttachState) {
    if state == CurrentVmAttachState::AlreadyAttached {
        return;
    }

    let gilstate_vm = GILSTATE_VM.with(|gilstate_vm| gilstate_vm.borrow_mut().take());
    drop(gilstate_vm);

    VM_STACK.with(|vms| {
        vms.borrow_mut()
            .pop()
            .expect("release_current_thread() called without an attached VM");
    });

    detach_thread();
}

/// Ensure this OS thread has a [`ThreadSlot`] registered with `vm`'s interpreter
/// and make it the current slot.
///
/// Called automatically by `enter_vm()` / `VmBootstrapGuard` whenever a VM
/// becomes current. Switching between interpreters on the same OS thread swaps
/// `CURRENT_THREAD_SLOT` to that interpreter's slot (creating one if needed).
#[cfg(feature = "threading")]
fn init_thread_slot_if_needed(vm: &VirtualMachine) {
    let slot = ensure_thread_slot(vm);
    set_current_thread_slot(slot);
}

/// Look up (creating if needed) this thread's [`ThreadSlot`] for `vm`'s
/// interpreter, without making it the current slot.
#[cfg(feature = "threading")]
fn ensure_thread_slot(vm: &VirtualMachine) -> CurrentFrameSlot {
    let interp_id = vm.state.interpreter_id;
    INTERP_THREAD_SLOTS.with(|slots| {
        let mut slots = slots.borrow_mut();
        if let Some(existing) = slots.get(&interp_id) {
            return existing.clone();
        }

        let thread_id = crate::stdlib::_thread::get_ident();
        let mut registry = vm.state.thread_frames.lock();
        let new_slot = Arc::new(ThreadSlot {
            #[cfg(unix)]
            top_frame: AtomicPtr::new(core::ptr::null_mut()),
            top_iframe: AtomicUsize::new(0),
            #[cfg(not(unix))]
            frames: parking_lot::Mutex::new(Vec::new()),
            exception: crate::PyAtomicRef::from(None::<PyBaseExceptionRef>),
            trace_func: PyMutex::new(
                vm.state
                    .global_trace_func
                    .lock()
                    .clone()
                    .unwrap_or_else(|| vm.ctx.none()),
            ),
            profile_func: PyMutex::new(
                vm.state
                    .global_profile_func
                    .lock()
                    .clone()
                    .unwrap_or_else(|| vm.ctx.none()),
            ),
            state: core::sync::atomic::AtomicI32::new(
                if vm.state.stop_the_world.requested.load(Ordering::Acquire) {
                    // Match init_threadstate(): new thread-state starts
                    // suspended while stop-the-world is active.
                    ThreadState::Suspended as i32
                } else {
                    ThreadState::Detached as i32
                },
            ),
            stop_requested: core::sync::atomic::AtomicBool::new(false),
            thread: std::thread::current(),
            qsbr: crate::object::qsbr::QSBR.register(),
        });
        registry.insert(thread_id, new_slot.clone());
        drop(registry);
        slots.insert(interp_id, new_slot.clone());
        new_slot
    })
}

/// The current thread's `ThreadSlot` for the entered interpreter, if any.
#[cfg(feature = "threading")]
#[must_use]
pub fn current_thread_slot() -> Option<CurrentFrameSlot> {
    CURRENT_THREAD_SLOT.with(|slot| slot.borrow().clone())
}

/// Make `slot` the current thread slot (and the cached top-frame pointer).
#[cfg(feature = "threading")]
fn set_current_thread_slot(slot: CurrentFrameSlot) {
    FRAME_SLOT_CACHE.with(|cache| {
        #[cfg(unix)]
        cache.top_frame.set(&slot.top_frame);
        cache.top_iframe.set(&slot.top_iframe);
    });
    CURRENT_STOP_REQUESTED.with(|c| c.set(&slot.stop_requested));
    CURRENT_THREAD_SLOT.with(|current| {
        *current.borrow_mut() = Some(slot);
    });
}

/// Whether the current thread slot is ATTACHED.
#[cfg(feature = "threading")]
fn current_slot_is_attached() -> bool {
    CURRENT_THREAD_SLOT.with(|slot| {
        slot.borrow()
            .as_ref()
            .is_some_and(|s| s.state.load(Ordering::Acquire) == ThreadState::Attached as i32)
    })
}

/// Attach this thread to `vm`'s interpreter for the duration of a section,
/// detaching whichever interpreter it was attached to (≈ `_PyThreadState_Swap`).
///
/// A thread must never be ATTACHED to two interpreters at once: stop-the-world
/// treats an ATTACHED slot as "running this interpreter's bytecode" and a
/// DETACHED slot as parkable without cooperation, so running interpreter B's
/// code while B's slot is DETACHED would let a collector conclude B is stopped
/// while this thread keeps mutating the (process-global) object graph.
///
/// Returns whether the attachment changed, i.e. whether the matching
/// [`end_interpreter_section`] must undo it.
#[cfg(feature = "threading")]
fn begin_interpreter_section(vm: &VirtualMachine) -> bool {
    let target = ensure_thread_slot(vm);
    let already_current = CURRENT_THREAD_SLOT.with(|slot| {
        slot.borrow()
            .as_ref()
            .is_some_and(|s| Arc::ptr_eq(s, &target))
    });
    if already_current && current_slot_is_attached() {
        // Nested section in the same interpreter: already attached.
        return false;
    }
    if !already_current && current_slot_is_attached() {
        detach_thread();
    }
    set_current_thread_slot(target);
    attach_thread(vm);
    true
}

/// Undo [`begin_interpreter_section`]: detach this interpreter and re-attach the
/// enclosing one, if any. Call after the VM has been popped from `VM_STACK`.
#[cfg(feature = "threading")]
fn end_interpreter_section(switched: bool) {
    if !switched {
        return;
    }
    if current_slot_is_attached() {
        detach_thread();
    }
    // The enclosing section, if any, is the VM now on top of the stack.
    if let Some(vm_ptr) = VM_STACK.with(|vms| vms.borrow().last().copied()) {
        // SAFETY: entries on VM_STACK are valid for their enter/set_current_vm scope.
        let vm = unsafe { vm_ptr.as_ref() };
        set_current_thread_slot(ensure_thread_slot(vm));
        attach_thread(vm);
    }
}

/// Transition DETACHED → ATTACHED. Blocks if the thread was SUSPENDED by
/// a stop-the-world request (like `_PyThreadState_Attach` + `tstate_wait_attach`).
#[cfg(feature = "threading")]
fn wait_while_suspended(slot: &ThreadSlot) -> u64 {
    let mut wait_yields = 0u64;
    while slot.state.load(Ordering::Acquire) == ThreadState::Suspended as i32 {
        wait_yields = wait_yields.saturating_add(1);
        std::thread::park();
    }
    wait_yields
}

/// `PyThread_hang_thread`: park this OS thread forever.
#[cfg(feature = "threading")]
fn hang_thread() -> ! {
    loop {
        std::thread::park();
    }
}

/// `_PyThreadState_HangThread`: this thread may no longer run Python.
///
/// Mark the slot shutting-down (so later stop-the-world requests do not wait
/// for it) and never return. The matching `_ThreadHandle` stays not-done, so
/// `Thread.is_alive()` remains true for a daemon forced off during finalize.
#[cfg(feature = "threading")]
pub fn hang_current_thread(state: &PyGlobalState) -> ! {
    CURRENT_THREAD_SLOT.with(|slot| {
        if let Some(s) = slot.borrow().as_ref() {
            let prev = s
                .state
                .swap(ThreadState::ShuttingDown as i32, Ordering::AcqRel);
            if prev == ThreadState::Attached as i32 {
                crate::object::qsbr::QSBR.offline(&s.qsbr);
            }
            s.stop_requested.store(false, Ordering::Release);
        }
    });
    if state.stop_the_world.requested.load(Ordering::Acquire) {
        state.stop_the_world.notify_thread_gone();
    }
    hang_thread();
}

/// `_PyThreadState_SetShuttingDown` on every non-current thread.
///
/// Call while the world is stopped. Wake parked threads so they observe
/// `SHUTTING_DOWN` and hang on the next attach, instead of resuming Python.
#[cfg(feature = "threading")]
pub fn set_other_threads_shutting_down(state: &PyGlobalState) {
    let current = crate::stdlib::_thread::get_ident();
    let registry = state.thread_frames.lock();

    #[expect(
        clippy::iter_over_hash_type,
        reason = "Iteration order doesn't matter here"
    )]
    for (&id, slot) in registry.iter() {
        if id == current {
            continue;
        }
        slot.stop_requested.store(false, Ordering::Release);
        slot.state
            .store(ThreadState::ShuttingDown as i32, Ordering::Release);
        slot.thread.unpark();
    }
}

#[cfg(feature = "threading")]
fn attach_thread(vm: &VirtualMachine) {
    CURRENT_THREAD_SLOT.with(|slot| {
        if let Some(s) = slot.borrow().as_ref() {
            super::stw_trace(format_args!("attach begin"));
            loop {
                match s.state.compare_exchange(
                    ThreadState::Detached as i32,
                    ThreadState::Attached as i32,
                    Ordering::AcqRel,
                    Ordering::Relaxed,
                ) {
                    Ok(_) => {
                        crate::object::qsbr::QSBR.online(&s.qsbr);
                        super::stw_trace(format_args!("attach DETACHED->ATTACHED"));
                        break;
                    }
                    Err(state) => match ThreadState::from_i32(state) {
                        Some(ThreadState::Suspended) => {
                            // Parked by stop-the-world — wait until released to DETACHED
                            super::stw_trace(format_args!("attach wait-suspended"));
                            let wait_yields = wait_while_suspended(s);
                            vm.state.stop_the_world.add_attach_wait_yields(wait_yields);
                            // Retry CAS
                        }
                        Some(ThreadState::ShuttingDown) => {
                            super::stw_trace(format_args!("attach hang shutting-down"));
                            hang_thread();
                        }
                        _ => {
                            debug_assert!(false, "unexpected thread state in attach: {state}");
                            break;
                        }
                    },
                }
            }
        }
    });
    // A stop-the-world may have been requested while this thread was detached.
    // Honoring it here (rather than only at the next bytecode safepoint) keeps
    // a thread doing rapid allow_threads calls from re-attaching and running
    // past the requester forever, which would stall stop-the-world. Done
    // outside the CURRENT_THREAD_SLOT borrow above because suspend re-borrows
    // it. Safe against a concurrent start_the_world: suspend_if_needed decides
    // whether to park under the registry lock, so it never parks after the
    // request has been withdrawn.
    suspend_if_needed(&vm.state);
}

/// Transition ATTACHED → DETACHED (like `_PyThreadState_Detach`).
#[cfg(feature = "threading")]
fn detach_thread() {
    CURRENT_THREAD_SLOT.with(|slot| {
        if let Some(s) = slot.borrow().as_ref() {
            match s.state.compare_exchange(
                ThreadState::Attached as i32,
                ThreadState::Detached as i32,
                Ordering::AcqRel,
                Ordering::Acquire,
            ) {
                Ok(_) => {
                    crate::object::qsbr::QSBR.offline(&s.qsbr);
                }
                Err(state) => {
                    debug_assert!(
                        matches!(ThreadState::from_i32(state), Some(ThreadState::Detached)),
                        "unexpected thread state in detach: {state}"
                    );
                    return;
                }
            }
            super::stw_trace(format_args!("detach ATTACHED->DETACHED"));
        }
    });
}

/// Temporarily transition the current thread ATTACHED → DETACHED while
/// running `f`, then re-attach afterwards.  This allows `stop_the_world`
/// to park this thread during blocking operations.
///
/// `Py_BEGIN_ALLOW_THREADS` / `Py_END_ALLOW_THREADS` equivalent.
#[cfg(feature = "threading")]
pub fn allow_threads<R>(vm: &VirtualMachine, f: impl FnOnce() -> R) -> R {
    // Preserve save/restore semantics:
    // only detach if this call observed ATTACHED at entry, and always restore
    // on unwind.
    let should_transition = CURRENT_THREAD_SLOT.with(|slot| {
        slot.borrow()
            .as_ref()
            .is_some_and(|s| s.state.load(Ordering::Acquire) == ThreadState::Attached as i32)
    });
    if !should_transition {
        return f();
    }

    detach_thread();
    let reattach_guard = scopeguard::guard(vm, attach_thread);
    let result = f();
    drop(reattach_guard);
    result
}

/// No-op on non-threading builds.
#[cfg(not(feature = "threading"))]
pub fn allow_threads<R>(_vm: &VirtualMachine, f: impl FnOnce() -> R) -> R {
    f()
}

/// Run `f` with this thread attached, then return it to where it was.
///
/// The inverse of [`allow_threads`], for a callback that has to run Python from
/// inside a call the thread detached for — a handshake callback reaching a
/// Python `sni_callback`, say. Running that detached would execute Python on a
/// thread a stop-the-world requester counts as parked. `PyGILState_Ensure` and
/// `PyGILState_Release` bracket such a callback for the same reason.
///
/// A thread already attached, or one with no interpreter to attach to, just
/// runs `f`. A thread a stop-the-world has already moved to SUSPENDED parks
/// here until the world starts again, because [`attach_thread`] treats that
/// state as the wait it is; that is the point of routing through it rather than
/// testing for DETACHED alone.
#[cfg(feature = "threading")]
pub fn attach_for_callback<R>(vm: &VirtualMachine, f: impl FnOnce() -> R) -> R {
    let should_transition = CURRENT_THREAD_SLOT.with(|slot| {
        slot.borrow()
            .as_ref()
            .is_some_and(|s| s.state.load(Ordering::Acquire) != ThreadState::Attached as i32)
    });
    if !should_transition {
        return f();
    }

    attach_thread(vm);
    // Detach again even if `f` unwinds, so the `allow_threads` this is nested
    // inside still finds the state it left behind.
    let redetach_guard = scopeguard::guard((), |()| detach_thread());
    let result = f();
    drop(redetach_guard);
    result
}

/// No-op on non-threading builds.
#[cfg(not(feature = "threading"))]
pub fn attach_for_callback<R>(_vm: &VirtualMachine, f: impl FnOnce() -> R) -> R {
    f()
}

/// Wait for a lock the way a blocking call waits: detached, so a
/// stop-the-world requester never has to wait for this thread to reach a
/// safepoint it cannot reach while blocked.
///
/// Threads with no interpreter to leave — a native thread, or one whose
/// locals are already being destroyed — simply block.
///
/// Detaching cannot park the one thread that can start the world again:
/// [`park_detached_threads`](super::StopTheWorldState) skips the requester's
/// slot outright, by thread id, and [`suspend_if_needed`] keys off a stop bit
/// never set for it. That exemption is wider than the one `_PyEval_StopTheWorld`
/// gives, where only an ATTACHED requester is skipped and a DETACHED one is
/// suspended like any other thread — so this rests on a local invariant rather
/// than on the reference behavior.
#[cfg(feature = "threading")]
fn wait_detached_from_interpreter(wait: &dyn Fn()) {
    // Read the VM out before waiting: attaching afterwards reaches for the
    // same thread locals, which must not still be borrowed here.
    let current = VM_STACK
        .try_with(|vms| vms.try_borrow().ok()?.last().copied())
        .ok()
        .flatten();
    match current {
        // SAFETY: entries in VM_STACK either borrow a VM for the dynamic
        // scope of a set_current_vm()/enter_vm() call or point at GILSTATE_VM.
        Some(vm) => allow_threads(unsafe { vm.as_ref() }, wait),
        None => wait(),
    }
}

/// Teach the lock types how to detach this thread. Idempotent, so every
/// interpreter can call it while initializing.
#[cfg(feature = "threading")]
pub(crate) fn install_blocking_wait_hook() {
    rustpython_common::lock::set_blocking_wait_hook(wait_detached_from_interpreter);
}

/// Called from check_signals when stop-the-world is requested.
/// Transitions ATTACHED → SUSPENDED and waits until released
/// (like `_PyThreadState_Suspend` + `_PyThreadState_Attach`).
#[cfg(feature = "threading")]
pub fn suspend_if_needed(state: &PyGlobalState) {
    let should_suspend = CURRENT_THREAD_SLOT.with(|slot| {
        slot.borrow()
            .as_ref()
            .is_some_and(|s| s.stop_requested.load(Ordering::Relaxed))
    });
    if should_suspend {
        do_suspend(state);
    }
}

#[cfg(feature = "threading")]
#[cold]
fn do_suspend(state: &PyGlobalState) {
    let stw = &state.stop_the_world;
    CURRENT_THREAD_SLOT.with(|slot| {
        let borrowed = slot.borrow();
        let Some(s) = borrowed.as_ref() else {
            return;
        };

        // Decide whether to park while holding the thread registry. Both edges
        // of `requested` are written under that lock: `init_thread_countdown`
        // sets it, and `start_the_world` clears it and then releases every
        // SUSPENDED thread without letting go. Publishing SUSPENDED here is
        // therefore either seen by that release pass or never reached, which
        // leaves the requester the only writer that takes a thread out of
        // SUSPENDED. A completion check that observed this thread parked cannot
        // then be invalidated by the thread resuming on its own.
        let park = {
            let _registry = state.thread_frames.lock();
            if stw.requested.load(Ordering::Acquire) {
                Some(s.state.compare_exchange(
                    ThreadState::Attached as i32,
                    ThreadState::Suspended as i32,
                    Ordering::AcqRel,
                    Ordering::Acquire,
                ))
            } else {
                // The stop already ended; this thread's request bit is stale.
                s.stop_requested.store(false, Ordering::Release);
                None
            }
        };

        match park {
            None => {
                super::stw_trace(format_args!("suspend skip not-requested"));
                return;
            }
            Some(Ok(_)) => {
                // Consumed this thread's stop request bit.
                s.stop_requested.store(false, Ordering::Release);
            }
            Some(Err(state)) => match ThreadState::from_i32(state) {
                Some(ThreadState::Detached) => {
                    // Leaving VM; caller will re-check on next entry.
                    super::stw_trace(format_args!("suspend skip DETACHED"));
                    return;
                }
                Some(ThreadState::Suspended) => {
                    // Already parked by another path.
                    s.stop_requested.store(false, Ordering::Release);
                    super::stw_trace(format_args!("suspend skip already-suspended"));
                    return;
                }
                Some(ThreadState::ShuttingDown) => {
                    s.stop_requested.store(false, Ordering::Release);
                    super::stw_trace(format_args!("suspend hang shutting-down"));
                    hang_thread();
                }
                _ => {
                    debug_assert!(false, "unexpected thread state in suspend: {state}");
                    return;
                }
            },
        }
        super::stw_trace(format_args!("suspend ATTACHED->SUSPENDED"));

        // Notify the stop-the-world requester that we've parked. The registry
        // is released first: the requester's wait loop takes the notify mutex
        // and then the registry, so taking them the other way round here would
        // invert the order.
        stw.notify_suspended();
        super::stw_trace(format_args!("suspend notified-requester"));

        // Wait until start_the_world sets us back to DETACHED
        let wait_yields = wait_while_suspended(s);
        stw.add_suspend_wait_yields(wait_yields);

        // Re-attach (DETACHED → ATTACHED), tstate_wait_attach CAS loop.
        loop {
            match s.state.compare_exchange(
                ThreadState::Detached as i32,
                ThreadState::Attached as i32,
                Ordering::AcqRel,
                Ordering::Acquire,
            ) {
                Ok(_) => break,
                Err(state) => match ThreadState::from_i32(state) {
                    Some(ThreadState::Suspended) => {
                        let extra_wait = wait_while_suspended(s);
                        stw.add_suspend_wait_yields(extra_wait);
                    }
                    Some(ThreadState::Attached) => break,
                    Some(ThreadState::ShuttingDown) => {
                        super::stw_trace(format_args!("suspend resume hang shutting-down"));
                        hang_thread();
                    }
                    _ => {
                        debug_assert!(false, "unexpected post-suspend state: {state}");
                        break;
                    }
                },
            }
        }
        s.stop_requested.store(false, Ordering::Release);
        super::stw_trace(format_args!("suspend resume -> ATTACHED"));
    });
}

#[cfg(feature = "threading")]
#[inline]
#[must_use]
pub fn stop_requested_for_current_thread() -> bool {
    CURRENT_STOP_REQUESTED.with(|cached| {
        let flag = cached.get();
        // SAFETY: the pointer is non-null only while `CURRENT_THREAD_SLOT`
        // holds the `Arc<ThreadSlot>` that owns the flag; both are cleared
        // together in `cleanup_current_thread_frames`.
        !flag.is_null() && unsafe { &*flag }.load(Ordering::Relaxed)
    })
}

#[cfg(all(test, feature = "threading"))]
pub(crate) fn set_stop_requested_for_current_thread(value: bool) -> bool {
    CURRENT_STOP_REQUESTED.with(|cached| {
        let flag = cached.get();
        if flag.is_null() {
            return false;
        }
        // SAFETY: same lifetime as `stop_requested_for_current_thread`.
        unsafe { &*flag }.store(value, Ordering::Release);
        true
    })
}

/// Whether the QSBR subsystem asked this thread to pass a checkpoint.
/// A missed or racing read of this flag is harmless: the pending
/// retirement is still processed at the next checkpoint or by the GC
/// backstop.
#[cfg(feature = "threading")]
pub(crate) fn qsbr_break_requested() -> bool {
    CURRENT_THREAD_SLOT.with(|slot| {
        slot.borrow()
            .as_ref()
            .is_some_and(|s| s.qsbr.requested.load(Ordering::Relaxed))
    })
}

/// Pass a QSBR checkpoint: the calling thread holds no borrowed cache
/// pointers here (instruction boundary), so mark it quiescent and try to
/// free retired allocations.
#[cfg(feature = "threading")]
pub(crate) fn qsbr_checkpoint() {
    use crate::object::qsbr::QSBR;
    CURRENT_THREAD_SLOT.with(|slot| {
        if let Some(s) = slot.borrow().as_ref() {
            s.qsbr.requested.store(false, Ordering::Relaxed);
            QSBR.quiescent_state(&s.qsbr);
        }
    });
    QSBR.process();
}

/// Debug check: lock-free type-cache reads are only sound on threads that
/// are registered with QSBR and currently ATTACHED.
#[cfg(all(feature = "threading", debug_assertions))]
pub(crate) fn debug_assert_current_thread_attached() {
    CURRENT_THREAD_SLOT.with(|slot| {
        if let Some(s) = slot.borrow().as_ref() {
            debug_assert_eq!(
                s.state.load(Ordering::Relaxed),
                ThreadState::Attached as i32,
                "type cache read while thread not ATTACHED"
            );
        }
    });
}

/// Push a frame pointer onto the current thread's shared frame stack.
/// The pointed-to frame must remain alive until the matching pop.
///
/// Only used on non-unix threading builds; unix builds publish the top frame
/// through `set_current_frame` writing `ThreadSlot::top_frame`.
#[cfg(all(not(unix), feature = "threading"))]
pub fn push_thread_frame(fp: FramePtr) {
    CURRENT_THREAD_SLOT.with(|slot| {
        if let Some(s) = slot.borrow().as_ref() {
            s.frames.lock().push(fp);
        } else {
            debug_assert!(
                false,
                "push_thread_frame called without initialized thread slot"
            );
        }
    });
}

/// Pop a frame from the current thread's shared frame stack.
/// Called when a frame is exited.
#[cfg(all(not(unix), feature = "threading"))]
pub fn pop_thread_frame() {
    CURRENT_THREAD_SLOT.with(|slot| {
        if let Some(s) = slot.borrow().as_ref() {
            s.frames.lock().pop();
        } else {
            debug_assert!(
                false,
                "pop_thread_frame called without initialized thread slot"
            );
        }
    });
}

/// Set the current thread's top InterpreterFrame pointer.
/// Returns the previous pointer so it can be restored on pop.
#[must_use]
#[allow(clippy::not_unsafe_ptr_arg_deref)]
pub fn set_current_frame(frame: *const InterpreterFrame) -> *const InterpreterFrame {
    FRAME_SLOT_CACHE.with(|cache| {
        // Publish the top frame for cross-thread readers (faulthandler,
        // sys._current_frames).
        #[cfg(feature = "threading")]
        {
            let slot = cache.top_iframe.get();
            if !slot.is_null() {
                unsafe { &*slot }.store(frame as usize, Ordering::Relaxed);
            }
            #[cfg(unix)]
            {
                let slot = cache.top_frame.get();
                if !slot.is_null() {
                    let fo_ptr = if frame.is_null() {
                        core::ptr::null_mut()
                    } else {
                        let frame_obj = unsafe { (*frame).frame_obj() };
                        frame_obj.map_or(core::ptr::null_mut(), |py| {
                            py as *const Py<FrameObject> as *mut Py<FrameObject>
                        })
                    };
                    unsafe { &*slot }.store(fo_ptr, Ordering::Relaxed);
                }
            }
        }
        cache.current_frame.swap(frame as usize, Ordering::Relaxed)
    }) as *const InterpreterFrame
}

/// Lightweight version that only writes to TLS `current_frame`, returning
/// the previous value. Does not update cross-thread top_frame (that's
/// updated by `set_current_frame` for FrameObject-based calls).
#[inline(always)]
#[must_use]
pub fn set_current_frame_nosave(frame: *const InterpreterFrame) -> *const InterpreterFrame {
    FRAME_SLOT_CACHE.with(|cache| cache.current_frame.swap(frame as usize, Ordering::Relaxed))
        as *const InterpreterFrame
}

/// Get the current thread's top InterpreterFrame pointer.
/// Used by faulthandler's signal handler to start traceback walking.
#[must_use]
pub fn get_current_frame() -> *const InterpreterFrame {
    FRAME_SLOT_CACHE.with(|cache| cache.current_frame.load(Ordering::Relaxed))
        as *const InterpreterFrame
}

/// Update the current thread's exception slot atomically (no locks).
/// Called from push_exception/pop_exception/set_exception.
#[cfg(feature = "threading")]
pub fn update_thread_exception(exc: Option<PyBaseExceptionRef>) {
    CURRENT_THREAD_SLOT.with(|slot| {
        if let Some(s) = slot.borrow().as_ref() {
            // SAFETY: Called only from the owning thread. The old ref is dropped
            // here on the owning thread, which is safe.
            let _old = unsafe { s.exception.swap(exc) };
        }
    });
}

/// Collect all threads' current exceptions for sys._current_exceptions().
/// Acquires the global registry lock briefly, then reads each slot's exception atomically.
#[cfg(feature = "threading")]
pub fn get_all_current_exceptions(vm: &VirtualMachine) -> Vec<(u64, Option<PyBaseExceptionRef>)> {
    let registry = vm.state.thread_frames.lock();
    registry
        .iter()
        .map(|(id, slot)| (*id, slot.exception.load_owned()))
        .collect()
}

/// Cleanup thread slot for the current thread in `vm`'s interpreter.
/// Called at thread exit (or when leaving an interpreter permanently).
#[cfg(feature = "threading")]
pub fn cleanup_current_thread_frames(vm: &VirtualMachine) {
    let thread_id = crate::stdlib::_thread::get_ident();
    let interp_id = vm.state.interpreter_id;

    // Prefer the slot registered for this interpreter; fall back to CURRENT.
    let slot_for_interp = INTERP_THREAD_SLOTS.with(|slots| slots.borrow_mut().remove(&interp_id));
    let current_slot = CURRENT_THREAD_SLOT.with(|slot| slot.borrow().as_ref().cloned());
    let slot_to_clean = slot_for_interp.or(current_slot);

    // A dying thread should not remain logically ATTACHED while its
    // thread-state slot is being removed.
    if let Some(slot) = &slot_to_clean {
        let _ = slot.state.compare_exchange(
            ThreadState::Attached as i32,
            ThreadState::Detached as i32,
            Ordering::AcqRel,
            Ordering::Acquire,
        );
    }

    // Guard against OS thread-id reuse races: only remove the registry entry
    // if it still points at this thread's own slot.
    let _removed = if let Some(slot) = &slot_to_clean {
        let mut registry = vm.state.thread_frames.lock();
        match registry.get(&thread_id) {
            Some(registered) if Arc::ptr_eq(registered, slot) => registry.remove(&thread_id),
            _ => None,
        }
    } else {
        None
    };

    if let Some(slot) = &_removed
        && vm.state.stop_the_world.requested.load(Ordering::Acquire)
        && thread_id != vm.state.stop_the_world.requester_ident()
        && slot.state.load(Ordering::Relaxed) != ThreadState::Suspended as i32
    {
        // A non-requester thread disappeared while stop-the-world is pending.
        // Unblock requester countdown progress.
        vm.state.stop_the_world.notify_thread_gone();
    }

    // If CURRENT pointed at the cleaned slot, clear it (and top-frame cache).
    CURRENT_THREAD_SLOT.with(|s| {
        let clear = match (s.borrow().as_ref(), slot_to_clean.as_ref()) {
            (Some(cur), Some(cleaned)) => Arc::ptr_eq(cur, cleaned),
            (Some(_), None) => false,
            (None, _) => false,
        };
        if clear {
            *s.borrow_mut() = None;
            #[cfg(feature = "threading")]
            FRAME_SLOT_CACHE.with(|cache| {
                #[cfg(unix)]
                cache.top_frame.set(core::ptr::null());
                cache.top_iframe.set(core::ptr::null());
            });
            #[cfg(feature = "threading")]
            CURRENT_STOP_REQUESTED.with(|c| c.set(core::ptr::null()));
        }
    });
}

/// Reinitialize thread slot after fork. Called in child process.
/// Creates a fresh slot and registers it for the current thread,
/// preserving the current thread's frames from the signal-safe frame chain.
///
/// Precondition: `reinit_locks_after_fork()` has already reset all
/// VmState locks to unlocked.
#[cfg(feature = "threading")]
pub fn reinit_frame_slot_after_fork(vm: &VirtualMachine) {
    let current_ident = crate::stdlib::_thread::get_ident();
    // On non-unix, rebuild the shared frame stack (bottom-to-top) from the
    // current thread's frame chain, which walks top-to-bottom via `previous`.
    #[cfg(not(unix))]
    let current_frames: Vec<FramePtr> = {
        let mut current_frames = Vec::new();
        let mut cur = get_current_frame();
        while !cur.is_null() {
            // SAFETY: the forking thread's chain frames are alive.
            let iframe = unsafe { &*cur };
            if let Some(fo) = iframe.frame_obj() {
                current_frames.push(FramePtr(unsafe {
                    NonNull::new_unchecked(fo as *const _ as *mut _)
                }));
            }
            cur = iframe.previous.load(Ordering::Relaxed) as *const InterpreterFrame;
        }
        current_frames.reverse();
        current_frames
    };
    #[cfg(unix)]
    let top_fo_ptr = {
        let top_iframe = get_current_frame();
        if top_iframe.is_null() {
            core::ptr::null_mut()
        } else {
            match unsafe { (*top_iframe).frame_obj() } {
                Some(fo) => fo as *const Py<FrameObject> as *mut Py<FrameObject>,
                None => core::ptr::null_mut(),
            }
        }
    };
    let top_iframe_ptr = get_current_frame() as usize;
    let new_slot = Arc::new(ThreadSlot {
        // The surviving child thread keeps executing its current frame chain.
        // Only publish heavy frames for signal safety.
        #[cfg(unix)]
        top_frame: AtomicPtr::new(top_fo_ptr),
        top_iframe: AtomicUsize::new(top_iframe_ptr),
        #[cfg(not(unix))]
        frames: parking_lot::Mutex::new(current_frames),
        exception: crate::PyAtomicRef::from(vm.topmost_exception()),
        trace_func: PyMutex::new(vm.trace_func.borrow().clone()),
        profile_func: PyMutex::new(vm.profile_func.borrow().clone()),
        state: core::sync::atomic::AtomicI32::new(ThreadState::Attached as i32),
        stop_requested: core::sync::atomic::AtomicBool::new(false),
        thread: std::thread::current(),
        qsbr: crate::object::qsbr::QSBR.register(),
    });
    FRAME_SLOT_CACHE.with(|cache| {
        #[cfg(unix)]
        cache.top_frame.set(&new_slot.top_frame);
        cache.top_iframe.set(&new_slot.top_iframe);
    });
    #[cfg(feature = "threading")]
    CURRENT_STOP_REQUESTED.with(|c| c.set(&new_slot.stop_requested));

    // Lock is safe: reinit_locks_after_fork() already reset it to unlocked.
    let mut registry = vm.state.thread_frames.lock();
    registry.clear();
    registry.insert(current_ident, new_slot.clone());
    drop(registry);

    CURRENT_THREAD_SLOT.with(|s| {
        *s.borrow_mut() = Some(new_slot.clone());
    });
    INTERP_THREAD_SLOTS.with(|slots| {
        slots.borrow_mut().insert(vm.state.interpreter_id, new_slot);
    });
}

/// Drop this thread's cached slots for every interpreter except `keep_id`.
///
/// After `fork()` only the calling thread survives, and the other
/// interpreters' registries are cleared; a cached slot would otherwise stay
/// current for an interpreter that no longer lists it, hiding the thread from
/// that interpreter's stop-the-world. The next enter builds a fresh slot.
#[cfg(feature = "threading")]
pub fn purge_other_interpreter_slots_after_fork(keep_id: i64) {
    INTERP_THREAD_SLOTS.with(|slots| {
        slots.borrow_mut().retain(|&id, _| id == keep_id);
    });
}

/// Whether the interpreter on top of `VM_STACK` is currently ATTACHED on this
/// thread. Without the `threading` feature there is no attach state to check.
#[cfg(feature = "threading")]
fn top_slot_is_attached() -> bool {
    current_slot_is_attached()
}
#[cfg(not(feature = "threading"))]
fn top_slot_is_attached() -> bool {
    true
}

/// Which VM `with_vm` found for `obj`, and whether this thread is already
/// attached to it (see the fast path below).
enum WithVmTarget {
    /// `interp` is on top of `VM_STACK`, so this thread is already ATTACHED
    /// to it (see [`begin_interpreter_section`]'s invariant): no section
    /// switch is needed.
    AlreadyCurrent(NonNull<VirtualMachine>),
    /// `interp` owns `obj` but is not the top of `VM_STACK` (a nested,
    /// currently-detached interpreter), so a real attach/detach section is
    /// required.
    NeedsSwitch(NonNull<VirtualMachine>),
}

pub fn with_vm<F, R>(obj: &PyObject, f: F) -> Option<R>
where
    F: Fn(&VirtualMachine) -> R,
{
    let vm_owns_obj = |interp: NonNull<VirtualMachine>| {
        // SAFETY: all references in VM_STACK should be valid
        let vm = unsafe { interp.as_ref() };
        obj.fast_isinstance(vm.ctx.types.object_type)
    };
    // `with_vm` runs on every teardown of an object with a `__del__` slot or a
    // weakref callback (drop_slow_inner / try_call_finalizer / gc_state), which
    // for `__del__` objects and weakrefs collected during a GC pass means it
    // runs on essentially every such drop. The overwhelming majority of those
    // drops happen from within Python bytecode executing on this very thread,
    // i.e. `obj`'s owning interpreter is already the one on top of `VM_STACK`.
    // `begin_interpreter_section` (via `set_current_vm`) only ever needs to run
    // for the rare case where the object belongs to a *different* interpreter
    // than the one currently attached (a nested subinterpreter scenario) or no
    // interpreter is attached at all (object dropped on a thread outside any
    // VM, e.g. during shutdown or from a plain Rust thread) — in the fast case
    // we can skip it entirely and call `f` directly.
    let target = VM_STACK.with(|vms| {
        let vms = vms.borrow();
        // Fast path: at most one interpreter is ever ATTACHED per OS thread,
        // and it is always the one on top of `VM_STACK` — every push onto
        // VM_STACK (set_current_vm, VmBootstrapGuard) is paired with an attach
        // *before* the push, and every pop is paired with a detach (or a
        // re-attach of the newly-exposed top) in `end_interpreter_section`.
        // So if `obj`'s owning interpreter is the current top, this thread is
        // already attached to it and there is nothing for
        // `begin_interpreter_section` to do: no INTERP_THREAD_SLOTS lookup, no
        // Arc clone, no atomic state transition.
        // The top may still be DETACHED while the thread sits inside an
        // `allow_threads` section (a blocking call that dropped an object
        // with `__del__`); running `f` there would execute Python on a thread
        // a stop-the-world requester counts as parked, so that case takes the
        // full attach path below.
        if let Some(top) = vms.last().copied()
            && vm_owns_obj(top)
            && top_slot_is_attached()
        {
            return Some(WithVmTarget::AlreadyCurrent(top));
        }
        let interp = match vms.iter().copied().exactly_one() {
            Ok(x) => {
                debug_assert!(vm_owns_obj(x));
                x
            }
            Err(mut others) => others.find(|x| vm_owns_obj(*x))?,
        };
        Some(WithVmTarget::NeedsSwitch(interp))
    })?;
    match target {
        WithVmTarget::AlreadyCurrent(interp) => {
            // SAFETY: `interp` is (or was, at the point it was read above) the
            // top of VM_STACK for this thread, so it is valid for at least the
            // dynamic scope of the enclosing set_current_vm()/enter_vm() call,
            // which contains this whole function call.
            let vm = unsafe { interp.as_ref() };
            Some(f(vm))
        }
        WithVmTarget::NeedsSwitch(interp) => {
            // SAFETY: all references in VM_STACK should be valid, and should not be changed or moved
            // at least until this function returns and the stack unwinds to an enter_vm() call
            let vm = unsafe { interp.as_ref() };
            Some(set_current_vm(vm, || f(vm)))
        }
    }
}

#[must_use = "ThreadedVirtualMachine does nothing unless you move it to another thread and call .run()"]
#[cfg(feature = "threading")]
pub struct ThreadedVirtualMachine {
    pub(super) vm: VirtualMachine,
}

#[cfg(feature = "threading")]
impl ThreadedVirtualMachine {
    /// Create a `FnOnce()` that can easily be passed to a function like [`std::thread::Builder::spawn`]
    ///
    /// # Note
    ///
    /// If you return a `PyObjectRef` (or a type that contains one) from `F`, and don't `join()`
    /// on the thread this `FnOnce` runs in, there is a possibility that that thread will panic
    /// as `PyObjectRef`'s `Drop` implementation tries to run the `__del__` destructor of a
    /// Python object but finds that it's not in the context of any vm.
    pub fn make_spawn_func<F, R>(self, f: F) -> impl FnOnce() -> R
    where
        F: FnOnce(&VirtualMachine) -> R,
    {
        move || self.run(f)
    }

    /// Run a function in this thread context
    ///
    /// # Note
    ///
    /// If you return a `PyObjectRef` (or a type that contains one) from `F`, and don't return the object
    /// to the parent thread and then `join()` on the `JoinHandle` (or similar), there is a possibility that
    /// the current thread will panic as `PyObjectRef`'s `Drop` implementation tries to run the `__del__`
    /// destructor of a python object but finds that it's not in the context of any vm.
    pub fn run<F, R>(&self, f: F) -> R
    where
        F: FnOnce(&VirtualMachine) -> R,
    {
        let vm = &self.vm;
        // Each spawned thread has its own native stack bounds. Recompute the
        // soft limit here instead of inheriting the parent thread's value.
        vm.c_stack_soft_limit
            .set(VirtualMachine::calculate_c_stack_soft_limit());
        enter_vm(vm, || f(vm))
    }
}

impl VirtualMachine {
    /// Start a new thread with access to the same interpreter.
    ///
    /// # Note
    ///
    /// If you return a `PyObjectRef` (or a type that contains one) from `F`, and don't `join()`
    /// on the thread, there is a possibility that that thread will panic as `PyObjectRef`'s `Drop`
    /// implementation tries to run the `__del__` destructor of a python object but finds that it's
    /// not in the context of any vm.
    #[cfg(feature = "threading")]
    pub fn start_thread<F, R>(&self, f: F) -> std::thread::JoinHandle<R>
    where
        F: Send + 'static + FnOnce(&Self) -> R,
        R: Send + 'static,
    {
        let func = self.new_thread().make_spawn_func(f);
        std::thread::spawn(func)
    }

    /// Create a new VM thread that can be passed to a function like [`std::thread::spawn`]
    /// to use the same interpreter on a different thread. Note that if you just want to
    /// use this with `thread::spawn`, you can use
    /// [`vm.start_thread()`](`VirtualMachine::start_thread`) as a convenience.
    ///
    /// # Usage
    ///
    /// ```
    /// # rustpython_vm::Interpreter::without_stdlib(Default::default()).enter(|vm| {
    /// use std::thread::Builder;
    /// let handle = Builder::new()
    ///     .name("my thread :)".into())
    ///     .spawn(vm.new_thread().make_spawn_func(|vm| vm.ctx.none()))
    ///     .expect("couldn't spawn thread");
    /// let returned_obj = handle.join().expect("thread panicked");
    /// assert!(vm.is_none(&returned_obj));
    /// # })
    /// ```
    ///
    /// Note: this function is safe, but running the returned ThreadedVirtualMachine in the same
    /// thread context (i.e. with the same thread-local storage) doesn't have any
    /// specific guaranteed behavior.
    #[cfg(feature = "threading")]
    pub fn new_thread(&self) -> ThreadedVirtualMachine {
        let global_trace = self.state.global_trace_func.lock().clone();
        let global_profile = self.state.global_profile_func.lock().clone();
        let use_tracing = global_trace.is_some() || global_profile.is_some();

        let vm = Self {
            builtins: self.builtins.clone(),
            sys_module: self.sys_module.clone(),
            ctx: self.ctx.clone(),
            datastack: core::cell::UnsafeCell::new(crate::datastack::DataStack::new()),
            wasm_id: self.wasm_id.clone(),
            exceptions: RefCell::default(),
            import_func: self.import_func.clone(),
            importlib: self.importlib.clone(),
            profile_func: RefCell::new(global_profile.unwrap_or_else(|| self.ctx.none())),
            trace_func: RefCell::new(global_trace.unwrap_or_else(|| self.ctx.none())),
            use_tracing: Cell::new(use_tracing),
            what_event: Cell::new(None),
            tracing_depth: Cell::new(0),
            recursion_limit: self.recursion_limit.clone(),
            signal_handlers: core::cell::OnceCell::new(),
            signal_rx: None,
            repr_guards: RefCell::default(),
            state: self.state.clone(),
            initialized: self.initialized,
            recursion_depth: Cell::new(0),
            #[cfg(any(miri, target_env = "musl"))]
            native_recursion_depth: Cell::new(0),
            c_stack_soft_limit: Cell::new(Self::calculate_c_stack_soft_limit()),
            async_gen_firstiter: RefCell::new(None),
            async_gen_finalizer: RefCell::new(None),
            asyncio_running_loop: RefCell::new(None),
            asyncio_running_task: RefCell::new(None),
            context_stack: RefCell::default(),
            callable_cache: self.callable_cache.clone(),
            pending_tailcall_frame: Cell::new(None),
            pending_tailcall_owner: core::cell::UnsafeCell::new(None),
            pending_gen_resume: core::cell::UnsafeCell::new(None),
            trampoline_stack: core::cell::UnsafeCell::new(Vec::new()),
        };
        ThreadedVirtualMachine { vm }
    }
}