rusty_alloc 0.1.0-alpha.2

//! Thread + heap lifecycle (mirrors upstream `init.c` + the abandonment side
//! of `arena-abandon.c`).
//!
//! Every thread gets a lazily-created [`HeapBox`] (one OS page from the prim
//! layer — never from the allocator itself, so bootstrap can't recurse). The
//! fast path reaches it through a const-init `thread_local!` `Cell` pointer,
//! which the R1 spike measured at atomic-load parity (0.3 ns) — no dtor is
//! registered on that key (Cell is !Drop), so accessing it never allocates.
//!
//! Thread EXIT runs through the prim TLS destructor (`FlsAlloc` /
//! `pthread_key_create`, built in M1 for exactly this): drain delayed frees,
//! retire what died, transition every surviving page to `NEVER` (spinning out
//! in-flight `FREEING` remotes — the loom-modeled teardown order), zero the
//! segments' thread id, and publish them on the global abandoned list. Any
//! other thread's allocation slow path adopts them later.

use core::cell::{Cell, UnsafeCell};
use core::ffi::c_void;
use core::ptr;
use core::sync::atomic::{AtomicBool, AtomicPtr, AtomicUsize, Ordering};

use crate::heap::Heap;
use crate::page::{DelayedList, XFLAG_NEVER, page_collect, page_set_flag};
use crate::segment::{self, Segment};
use crate::{os, prim};

/// A heap plus the cross-thread fields remote threads may touch. Remote
/// threads only ever dereference `delayed` (via page `xheap` addresses) —
/// never the heap itself — which is what makes `&mut` on the inner heap sound.
#[repr(C)]
pub struct HeapBox {
    /// Cross-thread delayed-free list (remote-reachable). MUST stay the first
    /// field: pages store `&delayed` in `xheap`, and owner routing recovers
    /// the box by container-of (delayed at offset 0 ⇒ the addresses coincide).
    pub delayed: DelayedList,
    /// Next box in the global registry (guarded by the registry lock).
    /// AtomicPtr, not usize: reachability (miri's leak check and honesty
    /// about what is cached vs leaked) follows pointers, not integers.
    pub next_box: AtomicPtr<HeapBox>,
    /// Owning thread id (heaps allocate only from their creating thread).
    pub owner_tid: usize,
    /// Heap tag (`mi_heap_new_ex`; reported by the block visitor).
    pub tag: i32,
    /// Whether `mi_heap_destroy` may drop live blocks (else it delegates to
    /// delete semantics).
    pub allow_destroy: bool,
    /// The owner-only heap state.
    pub heap: UnsafeCell<Heap>,
}

const _: () = assert!(core::mem::offset_of!(HeapBox, delayed) == 0);

/// Recover the owning HeapBox from a page's `xheap` value (the container-of
/// trick over the offset-0 delayed list).
///
/// # Safety
/// `xheap` must be a live HeapBox's delayed-list address (nonzero).
#[inline]
pub unsafe fn box_of_xheap(xheap: usize) -> *mut HeapBox {
    debug_assert!(xheap != 0);
    xheap as *mut HeapBox
}

// Global registry of live heap boxes: keeps every heap (and through it every
// segment) reachable-from-static — the process-wide stats walk of M7, and
// what makes miri's reachability-based leak check reflect reality (an
// allocator's arenas at process exit are cached state, not leaks).
static HEAPS_LOCK: AtomicBool = AtomicBool::new(false);
static HEAPS_HEAD: AtomicPtr<HeapBox> = AtomicPtr::new(ptr::null_mut());

fn heaps_lock() {
    while HEAPS_LOCK
        .compare_exchange_weak(false, true, Ordering::Acquire, Ordering::Relaxed)
        .is_err()
    {
        core::hint::spin_loop();
    }
}

fn heaps_unlock() {
    HEAPS_LOCK.store(false, Ordering::Release);
}

fn heaps_register(hb: *mut HeapBox) {
    heaps_lock();
    // SAFETY: hb is fully initialized and not yet published.
    unsafe {
        (*hb)
            .next_box
            .store(HEAPS_HEAD.load(Ordering::Relaxed), Ordering::Relaxed);
    }
    HEAPS_HEAD.store(hb, Ordering::Relaxed);
    heaps_unlock();
}

fn heaps_unregister(hb: *mut HeapBox) {
    heaps_lock();
    let mut cur = HEAPS_HEAD.load(Ordering::Relaxed);
    if cur == hb {
        // SAFETY: links guarded by the lock.
        unsafe {
            HEAPS_HEAD.store((*hb).next_box.load(Ordering::Relaxed), Ordering::Relaxed);
        }
    } else {
        while !cur.is_null() {
            // SAFETY: links guarded by the lock; nodes are live boxes.
            unsafe {
                let nxt = (*cur).next_box.load(Ordering::Relaxed);
                if nxt == hb {
                    (*cur)
                        .next_box
                        .store((*nxt).next_box.load(Ordering::Relaxed), Ordering::Relaxed);
                    break;
                }
                cur = nxt;
            }
        }
    }
    heaps_unlock();
}

/// Storage for the calling thread's default-heap pointer.
///
/// On x86-64 Linux this is a RAW ELF TLS symbol reached with the
/// **initial-exec** model rather than `thread_local!`'s general-dynamic. That
/// was the last `__tls_get_addr` call on the malloc fast path — 7.3 M
/// instructions on perl, a third of everything still separating us from
/// mimalloc after the M10b free-path bricks. Stable Rust has no
/// `-Z tls-model`, so the symbol is declared in `global_asm!` and the linker
/// resolves its offset.
///
/// **Why this is sound where a thread-pointer-keyed side table is not.** The
/// storage IS the thread's own TLS block. Every thread gets a fresh block
/// initialised from the (all-zero) `.tbss` image when it is created, so a
/// recycled TCB cannot hand a new thread a dead thread's heap — the staleness
/// question that made the keyed-cache design a P0 risk simply does not arise
/// here. What initial-exec costs instead is a LOAD-TIME constraint: it needs a
/// slot in the static TLS block, so a `dlopen` long after startup could fail
/// to load us. That fails loudly at load time rather than corrupting memory,
/// and it is the same trade upstream mimalloc ships
/// (`__attribute__((tls_model("initial-exec")))` on `_mi_heap_default`).
#[cfg(all(target_arch = "x86_64", target_os = "linux", not(miri)))]
mod heap_tls {
    use super::HeapBox;

    // An 8-byte thread-local slot. `.tbss` is the zero-initialised TLS
    // section, so this reads as null on every new thread — which is exactly
    // the "not yet created" state `heap_box` tests for. `.hidden` makes the
    // symbol non-preemptible, which is what permits the initial-exec
    // (GOTTPOFF) relocation instead of a general-dynamic call.
    core::arch::global_asm!(
        ".section .tbss,\"awT\",@nobits",
        ".globl __ra_tls_heap",
        ".hidden __ra_tls_heap",
        ".type __ra_tls_heap,@object",
        ".p2align 3",
        "__ra_tls_heap:",
        ".zero 8",
        ".size __ra_tls_heap,8",
        ".text",
    );

    /// Address of THIS thread's slot: the thread pointer plus the offset the
    /// dynamic linker wrote into the GOT.
    #[inline(always)]
    fn slot() -> *mut *mut HeapBox {
        let off: usize;
        // SAFETY: reads the GOT entry the dynamic linker relocated at load
        // time, before any of our code can run, and never writes. It is
        // constant for the life of the process, hence `pure` + `readonly` —
        // that is what lets LLVM hoist and CSE the address computation
        // instead of treating the block as an optimisation barrier.
        unsafe {
            core::arch::asm!(
                "mov {o}, qword ptr [rip + __ra_tls_heap@GOTTPOFF]",
                o = out(reg) off,
                options(nostack, preserves_flags, readonly, pure),
            );
        }
        // `thread_id()` IS the fs base on this target. TLS offsets are
        // negative there (the block grows down from the thread pointer), so
        // this adds a two's-complement offset and must wrap.
        core::ptr::with_exposed_provenance_mut(super::thread_id().wrapping_add(off))
    }

    /// Read this thread's heap pointer in TWO instructions.
    ///
    /// The obvious form — `slot().read()` — costs FOUR: load the offset from
    /// the GOT, read the thread pointer from `fs:0`, add them, dereference.
    /// But x86 performs that addition IN THE ADDRESSING MODE: `fs:[reg]` is a
    /// segment-relative load, so the explicit `fs:0` read and the add both
    /// disappear. Measured at 4.00 Ir per malloc before this change — a
    /// quarter of our entire malloc-side deficit spent just locating the heap.
    #[inline(always)]
    pub fn get() -> *mut HeapBox {
        let v: *mut HeapBox;
        // SAFETY: reads the GOT entry the loader relocated at startup, then
        // loads 8 aligned bytes of this thread's own TLS block through the fs
        // segment. Both are reads; nothing is written.
        unsafe {
            core::arch::asm!(
                "mov {t}, qword ptr [rip + __ra_tls_heap@GOTTPOFF]",
                "mov {o}, qword ptr fs:[{t}]",
                t = out(reg) _,
                o = out(reg) v,
                options(nostack, preserves_flags, readonly),
            );
        }
        v
    }

    #[inline(always)]
    pub fn set(hb: *mut HeapBox) {
        // SAFETY: as `get` — the slot is owned exclusively by this thread.
        unsafe { slot().write(hb) }
    }
}

/// Every other target keeps the `thread_local!`: Windows TLS is an index into
/// a per-thread array (no `__tls_get_addr` to remove), and miri models the
/// macro but not a hand-rolled TLS symbol.
#[cfg(not(all(target_arch = "x86_64", target_os = "linux", not(miri))))]
mod heap_tls {
    use super::HeapBox;
    use core::cell::Cell;
    use core::ptr;

    std::thread_local! {
        /// Fast-path heap pointer. Const-init + !Drop ⇒ plain TLS access, no
        /// lazy-init branch beyond the null check, no allocation ever.
        static HEAP_PTR: Cell<*mut HeapBox> = const { Cell::new(ptr::null_mut()) };
    }

    #[inline(always)]
    pub fn get() -> *mut HeapBox {
        HEAP_PTR.with(|c| c.get())
    }

    #[inline(always)]
    pub fn set(hb: *mut HeapBox) {
        HEAP_PTR.with(|c| c.set(hb));
    }
}

std::thread_local! {
    /// Cached OS thread id. `free` needs the calling thread's id on EVERY
    /// call to route local-vs-remote; the raw `prim::thread_id()` is a libc
    /// call (`pthread_self` through the PLT from a cdylib) measured at
    /// 1.41 ns vs 0.25 ns for this cache — ~18-20% of a whole malloc+free
    /// pair (M9 probe, both platforms). 0 = not yet cached; no real thread id
    /// is 0 on either platform.
    static TID: Cell<usize> = const { Cell::new(0) };
}

/// The calling thread's id — read STRAIGHT FROM THE THREAD-POINTER REGISTER
/// where the target allows it.
///
/// Why this and not `thread_local!`: the shipping artifact is a **cdylib**
/// (LD_PRELOAD). Rust's `thread_local!` in a shared library compiles to the
/// general-dynamic TLS model, so every access is a CALL into `ld.so`'s
/// `__tls_get_addr` — the per-function instruction profile put that call at
/// 12.97 M Ir, 1.96% of a whole lua run, comparable to our entire `free`.
/// (The earlier ns-level probe missed it because it measured TLS inside an
/// EXECUTABLE, where the model is local-exec: a register offset, no call.
/// Measure the artifact you ship.)
///
/// A ceiling probe with `-Z tls-model=initial-exec` valued removing those
/// calls at 2.00% of our instructions and 33% of the gap to mimalloc; this
/// gets the same effect on stable, for the id, with no linker involvement —
/// the thread pointer is one register read. It is exactly what mimalloc's
/// `_mi_thread_id()` does.
///
/// Uniqueness: the value is the thread-control-block address, unique among
/// LIVE threads. It can be recycled after a thread dies, which is sound here
/// for the same reason it is in mimalloc: a dying thread abandons its
/// segments (`thread_id` stored as 0) before its TCB can be reused, so a
/// recycled id can never match a stale segment.
#[inline(always)]
pub fn thread_id() -> usize {
    #[cfg(all(target_arch = "x86_64", target_os = "linux", not(miri)))]
    {
        let tp: usize;
        // SAFETY: reads the TCB self-pointer at fs:0, which the x86-64 Linux
        // ABI guarantees is present and thread-unique. No memory is written.
        unsafe {
            core::arch::asm!("mov {}, fs:0", out(reg) tp,
                             options(nostack, preserves_flags, readonly));
        }
        tp
    }
    #[cfg(all(target_arch = "x86_64", target_os = "windows", not(miri)))]
    {
        let teb: usize;
        // SAFETY: gs:0x30 is the TEB self-pointer on x86-64 Windows —
        // present in every thread, unique, read-only here.
        unsafe {
            core::arch::asm!("mov {}, gs:0x30", out(reg) teb,
                             options(nostack, preserves_flags, readonly));
        }
        teb
    }
    #[cfg(all(target_arch = "aarch64", not(miri)))]
    {
        let tp: usize;
        // SAFETY: tpidr_el0 is the AArch64 thread pointer; read-only.
        unsafe {
            core::arch::asm!("mrs {}, tpidr_el0", out(reg) tp,
                             options(nostack, preserves_flags));
        }
        tp
    }
    // Every other target: the cached-TLS path (still far cheaper than asking
    // the OS on each call).
    #[cfg(not(any(
        all(target_arch = "x86_64", target_os = "linux", not(miri)),
        all(target_arch = "x86_64", target_os = "windows", not(miri)),
        all(target_arch = "aarch64", not(miri))
    )))]
    {
        let t = TID.with(|c| c.get());
        if t != 0 { t } else { init_tid() }
    }
}

#[cold]
#[cfg_attr(
    any(
        all(target_arch = "x86_64", target_os = "linux", not(miri)),
        all(target_arch = "x86_64", target_os = "windows", not(miri)),
        all(target_arch = "aarch64", not(miri))
    ),
    allow(dead_code)
)]
fn init_tid() -> usize {
    let t = prim::thread_id();
    TID.with(|c| c.set(t));
    t
}

/// Snapshot-visit every registered heap (stats aggregation). The visitor
/// receives a COPY of the heap's counters: owners mutate them concurrently, so
/// this is an approximate racy snapshot by design (volatile whole-struct read;
/// converting the counters to atomics is a ledgered follow-up).
pub fn for_each_heap(f: &mut dyn FnMut(&crate::heap::Heap)) {
    heaps_lock();
    let mut hb = HEAPS_HEAD.load(Ordering::Acquire);
    while !hb.is_null() {
        // SAFETY: registry entries are live boxes; read_volatile snapshots the
        // owner-mutated struct without asserting exclusive access via &mut.
        unsafe {
            let snapshot = core::ptr::read_volatile((*hb).heap.get());
            f(&snapshot);
            hb = (*hb).next_box.load(Ordering::Acquire);
        }
    }
    heaps_unlock();
}

/// Get (or create) the calling thread's heap box.
#[inline]
pub fn heap_box() -> *mut HeapBox {
    let hb = heap_tls::get();
    if !hb.is_null() {
        hb
    } else {
        init_thread_heap()
    }
}

/// Allocate + initialize a HeapBox (shared by thread bootstrap and
/// `mi_heap_new*`). Registered in the global registry; NOT installed in TLS.
pub fn create_heap(tag: i32, allow_destroy: bool, arena_id: i32) -> *mut HeapBox {
    let size = core::mem::size_of::<HeapBox>();
    let block = os::alloc_aligned(size, os::page_size(), true, false)
        .expect("rusty_alloc: cannot allocate heap");
    let hb: *mut HeapBox = block.ptr.cast();
    // SAFETY: fresh committed mapping large enough for HeapBox; we initialize
    // every field before publishing the pointer.
    unsafe {
        ptr::write(
            hb,
            HeapBox {
                delayed: DelayedList::new(),
                next_box: AtomicPtr::new(ptr::null_mut()),
                owner_tid: thread_id(),
                tag,
                allow_destroy,
                heap: UnsafeCell::new(Heap::new()),
            },
        );
        (*(*hb).heap.get()).delayed = &raw const (*hb).delayed;
        (*(*hb).heap.get()).arena_id = arena_id;
        (*(*hb).heap.get()).tag = tag;
        // Per-heap CSPRNG stream (free-list keys, guarded sampling).
        (*(*hb).heap.get()).rng.reseed();
        // Guarded objects follow the option table (MI_GUARDED-equivalent).
        let rate = crate::options::get(33).max(0) as usize; // guarded_sample_rate
        let gmin = crate::options::get(30).max(0) as usize; // guarded_min
        let gmax = crate::options::get(31).max(0) as usize; // guarded_max
        if gmax > 0 {
            (*(*hb).heap.get()).guarded_set_size_bound(gmin, gmax);
            (*(*hb).heap.get())
                .guarded_set_sample_rate(rate, crate::options::get(34).max(0) as usize);
        }
    }
    heaps_register(hb);
    hb
}

#[cold]
fn init_thread_heap() -> *mut HeapBox {
    let hb = create_heap(0, false, -1);
    heap_tls::set(hb);
    BACKING_PTR.with(|c| c.set(hb));
    done_slot().set(hb.cast::<c_void>());
    hb
}

std::thread_local! {
    /// The thread's original (backing) heap — what `mi_heap_get_backing`
    /// returns regardless of `mi_heap_set_default` swaps.
    static BACKING_PTR: Cell<*mut HeapBox> = const { Cell::new(ptr::null_mut()) };
}

/// `mi_heap_get_backing`.
pub fn backing_heap() -> *mut HeapBox {
    let b = BACKING_PTR.with(|c| c.get());
    if !b.is_null() {
        b
    } else {
        let _ = heap_box(); // bootstraps and sets BACKING_PTR
        BACKING_PTR.with(|c| c.get())
    }
}

/// `mi_heap_set_default`: install `hb` as this thread's default; returns the
/// previous default.
///
/// # Safety
/// `hb` must be a live heap owned by the calling thread.
pub unsafe fn set_default_heap(hb: *mut HeapBox) -> *mut HeapBox {
    let prev = heap_box();
    heap_tls::set(hb);
    prev
}

/// The process-wide TLS slot whose destructor abandons a dying thread's heap.
fn done_slot() -> prim::TlsSlot {
    static RAW: AtomicUsize = AtomicUsize::new(0);
    static INIT: AtomicBool = AtomicBool::new(false);
    let raw = RAW.load(Ordering::Acquire);
    if raw != 0 {
        // SAFETY: RAW holds an into_raw of a slot created below.
        return unsafe { prim::TlsSlot::from_raw(raw - 1) };
    }
    if INIT
        .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
        .is_ok()
    {
        // A panic here would be wrong twice over. An allocator must not unwind
        // into its C callers (which is why the release profile is
        // panic=abort), AND the winner of this race is the only thread that
        // ever publishes RAW — so unwinding past the store below would strand
        // every other thread in the spin loop FOREVER, turning a rare
        // resource failure into a silent process-wide hang. Fail loudly and
        // deterministically instead, identically in debug and release.
        let Some(slot) = prim::TlsSlot::new(Some(thread_done_cb)) else {
            std::process::abort();
        };
        RAW.store(slot.into_raw() + 1, Ordering::Release); // +1: 0 is the unset sentinel
    }
    loop {
        let raw = RAW.load(Ordering::Acquire);
        if raw != 0 {
            // SAFETY: as above.
            return unsafe { prim::TlsSlot::from_raw(raw - 1) };
        }
        core::hint::spin_loop();
    }
}

#[cfg(all(windows, not(miri)))]
unsafe extern "system" fn thread_done_cb(v: *const c_void) {
    if !v.is_null() {
        // SAFETY: v is the HeapBox this thread stored in the done slot.
        unsafe { thread_done(v.cast_mut().cast()) };
    }
}

#[cfg(any(not(windows), miri))]
unsafe extern "C" fn thread_done_cb(v: *mut c_void) {
    if !v.is_null() {
        // SAFETY: v is the HeapBox this thread stored in the done slot.
        unsafe { thread_done(v.cast()) };
    }
}

/// Abandon the dying thread's heap. Teardown order is the loom-modeled one:
/// (1) every surviving page → NEVER (spins out FREEING remotes), (2) THEN
/// drain the delayed list, (3) THEN release the heap storage.
///
/// # Safety
/// Must run on the dying thread with `hb` its live heap box, exactly once.
pub unsafe fn thread_done(hb: *mut HeapBox) {
    // SAFETY: per contract we are the owner; sequencing per the model.
    unsafe {
        let h = &mut *(*hb).heap.get();
        // Retire everything that is already dead (also drains delayed once).
        h.collect(true);
        // Walk remaining segments: transition pages to NEVER, disown, publish.
        let mut seg = h.segments;
        h.segments = ptr::null_mut();
        while !seg.is_null() {
            let next = (*seg).next;
            let end = (*seg).next_free_slice as usize;
            let mut idx = segment::HEADER_SLICES;
            while idx < end {
                let slot = &raw mut (*seg).pages[idx];
                let len = ((*slot).slice_count as usize).max(1);
                if (*slot).block_size > 0 {
                    page_collect(slot);
                    page_set_flag(slot, XFLAG_NEVER);
                    (*slot).xheap.store(0, Ordering::Release);
                }
                idx += len;
            }
            if (*seg).used_pages == 0 {
                let _ = segment::segment_free(seg);
                h.stats.segments_freed += 1;
            } else {
                (*seg).thread_id.store(0, Ordering::Release);
                abandoned_push(seg);
            }
            seg = next;
        }
        // Huge segments: same teardown shape (collect NEVER-bound frees,
        // release dead ones, abandon the rest).
        let mut hseg = h.huge_segments;
        h.huge_segments = ptr::null_mut();
        while !hseg.is_null() {
            let next = (*hseg).next;
            let pg = &raw mut (*hseg).pages[1];
            page_collect(pg);
            if (*pg).used == 0 {
                let _ = segment::huge_free(hseg);
                h.stats.segments_freed += 1;
            } else {
                page_set_flag(pg, XFLAG_NEVER);
                (*pg).xheap.store(0, Ordering::Release);
                (*hseg).thread_id.store(0, Ordering::Release);
                abandoned_push(hseg);
            }
            hseg = next;
        }
        // Post-NEVER drain: blocks a FREEING remote landed while we walked.
        // Their pages are abandoned now, so free_local would be wrong — but
        // these blocks belong to pages we JUST abandoned; push them back on
        // their pages' own lists (flag NEVER routes them there).
        let mut b = (*hb).delayed.head.swap(0, Ordering::AcqRel) as *mut crate::page::Block;
        while !b.is_null() {
            let next = (*b).next;
            let seg = segment::segment_of(b.cast());
            let pg = segment::page_of(seg, b.cast());
            crate::page::remote_free(pg, b);
            b = next;
        }
        // Release the heap storage and clear the fast-path pointer.
        heaps_unregister(hb);
        let size = core::mem::size_of::<HeapBox>();
        let blockdesc = os::OsBlock {
            ptr: hb.cast(),
            size: os::page_align_up(size),
            is_large: false,
            is_zero: false,
        };
        let _ = os::free(blockdesc);
    }
    heap_tls::set(ptr::null_mut());
}

/// `mi_heap_delete`: free the heap structure, MIGRATING its live pages and
/// segments into the thread's backing heap (blocks stay valid).
///
/// # Safety
/// `hb` must be a live first-class heap owned by the calling thread, not the
/// backing heap, used never again.
pub unsafe fn heap_delete(hb: *mut HeapBox) {
    let backing = backing_heap();
    if hb == backing {
        return; // deleting the backing heap is a no-op (upstream-compatible)
    }
    // SAFETY: owner thread per contract; absorb everything into backing.
    unsafe {
        debug_assert_eq!((*hb).owner_tid, thread_id());
        let h = &mut *(*hb).heap.get();
        let bh = &mut *(*backing).heap.get();
        h.collect(true);
        let mut seg = h.segments;
        h.segments = ptr::null_mut();
        while !seg.is_null() {
            let next = (*seg).next;
            bh.adopt_segment(seg); // re-homes queues + xheap (tid already ours)
            seg = next;
        }
        let mut hseg = h.huge_segments;
        h.huge_segments = ptr::null_mut();
        while !hseg.is_null() {
            let next = (*hseg).next;
            let pg = &raw mut (*hseg).pages[1];
            (*pg).xheap.store(bh.delayed as usize, Ordering::Release);
            (*hseg).next = bh.huge_segments;
            bh.huge_segments = hseg;
            hseg = next;
        }
        // Fold the dying heap's counters into backing so process totals hold.
        merge_stats(&mut bh.stats, &h.stats);
        release_heap_box(hb);
    }
}

/// `mi_heap_destroy`: drop EVERY block of the heap at once and release its
/// memory. Falls back to delete semantics when the heap was created with
/// `allow_destroy = false`.
///
/// # Safety
/// As [`heap_delete`], and additionally no block of this heap may be used
/// afterwards (the C contract).
pub unsafe fn heap_destroy(hb: *mut HeapBox) {
    // SAFETY: owner thread per contract.
    unsafe {
        if !(*hb).allow_destroy {
            heap_delete(hb);
            return;
        }
        debug_assert_eq!((*hb).owner_tid, thread_id());
        let h = &mut *(*hb).heap.get();
        // Spin out any in-flight FREEING remotes before the box dies; blocks
        // they queued are dying with the heap, so the delayed list is simply
        // discarded.
        let mut seg = h.segments;
        while !seg.is_null() {
            let end = (*seg).next_free_slice as usize;
            let mut idx = segment::HEADER_SLICES;
            while idx < end {
                let slot = &raw mut (*seg).pages[idx];
                let len = ((*slot).slice_count as usize).max(1);
                if (*slot).block_size > 0 {
                    page_set_flag(slot, XFLAG_NEVER);
                    (*slot).xheap.store(0, Ordering::Release);
                }
                idx += len;
            }
            let next = (*seg).next;
            let _ = segment::segment_free(seg);
            seg = next;
        }
        let mut hseg = h.huge_segments;
        while !hseg.is_null() {
            let pg = &raw mut (*hseg).pages[1];
            page_set_flag(pg, XFLAG_NEVER);
            (*pg).xheap.store(0, Ordering::Release);
            let next = (*hseg).next;
            let _ = segment::huge_free(hseg);
            hseg = next;
        }
        release_heap_box(hb);
    }
}

fn merge_stats(into: &mut crate::heap::Stats, from: &crate::heap::Stats) {
    into.allocs += from.allocs;
    into.frees += from.frees;
    into.generic += from.generic;
    into.pages_fresh += from.pages_fresh;
    into.segments += from.segments;
    into.huge_allocs += from.huge_allocs;
    into.extends += from.extends;
    into.large_allocs += from.large_allocs;
    into.pages_retired += from.pages_retired;
    into.segments_freed += from.segments_freed;
    into.realloc_in_place += from.realloc_in_place;
    into.realloc_moved += from.realloc_moved;
    into.delayed_frees += from.delayed_frees;
    into.reclaims += from.reclaims;
}

/// Unregister and release a HeapBox, repairing the TLS default if needed.
///
/// # Safety
/// `hb` live, owned by the calling thread, all contents already migrated or
/// released.
unsafe fn release_heap_box(hb: *mut HeapBox) {
    heaps_unregister(hb);
    if heap_tls::get() == hb {
        heap_tls::set(backing_heap());
    }
    let size = core::mem::size_of::<HeapBox>();
    let blockdesc = os::OsBlock {
        ptr: hb.cast(),
        size: os::page_align_up(size),
        is_large: false,
        is_zero: false,
    };
    // SAFETY: hb's mapping came from create_heap's os::alloc_aligned with this
    // exact size; unregistered above, no references remain.
    unsafe {
        let _ = os::free(blockdesc);
    }
}

// ---------------------------------------------------------------------------
// Global abandoned-segment list (spin-locked; cold path).
// ---------------------------------------------------------------------------

/// Subprocess isolation (M6, plan §5.9): threads tagged with a subproc id
/// abandon/reclaim only within it — separate interpreters never exchange
/// segments. Id 0 is the main subprocess.
const MAX_SUBPROCS: usize = 64;

static ABANDONED_LOCK: AtomicBool = AtomicBool::new(false);
static ABANDONED_HEADS: [AtomicPtr<Segment>; MAX_SUBPROCS] =
    [const { AtomicPtr::new(ptr::null_mut()) }; MAX_SUBPROCS];
static SUBPROC_NEXT: AtomicUsize = AtomicUsize::new(1);
/// Diagnostic: segments currently abandoned (all subprocs).
pub static ABANDONED_COUNT: AtomicUsize = AtomicUsize::new(0);

std::thread_local! {
    static SUBPROC: Cell<usize> = const { Cell::new(0) };
}

/// `mi_subproc_main` → 0.
pub fn subproc_main() -> usize {
    0
}

/// `mi_subproc_new`: allocate a fresh subprocess id.
pub fn subproc_new() -> usize {
    let id = SUBPROC_NEXT.fetch_add(1, Ordering::AcqRel);
    assert!(id < MAX_SUBPROCS, "out of subprocess ids");
    id
}

/// `mi_subproc_delete`: ids are not recycled in v1; abandoned segments of the
/// subproc are drained into the main subproc so they stay reclaimable.
pub fn subproc_delete(id: usize) {
    if id == 0 || id >= MAX_SUBPROCS {
        return;
    }
    abandoned_lock();
    // SAFETY: list surgery under the lock.
    unsafe {
        let mut seg = ABANDONED_HEADS[id].swap(ptr::null_mut(), Ordering::AcqRel);
        while !seg.is_null() {
            let next = (*seg).next;
            (*seg).next = ABANDONED_HEADS[0].load(Ordering::Relaxed);
            ABANDONED_HEADS[0].store(seg, Ordering::Relaxed);
            seg = next;
        }
    }
    abandoned_unlock();
}

/// `mi_subproc_add_current_thread`.
pub fn subproc_add_current_thread(id: usize) {
    assert!(id < MAX_SUBPROCS);
    SUBPROC.with(|c| c.set(id));
}

fn my_subproc() -> usize {
    SUBPROC.with(|c| c.get())
}

/// `mi_abandoned_visit_blocks`: visit live pages/blocks of a subprocess's
/// abandoned segments, holding the list lock (segments are pinned meanwhile).
pub fn abandoned_visit_blocks(
    subproc_id: usize,
    heap_tag: i32,
    visit_blocks: bool,
    f: &mut dyn FnMut(&crate::heap::AreaInfo, *mut u8, usize) -> bool,
) -> bool {
    if subproc_id >= MAX_SUBPROCS {
        return false;
    }
    abandoned_lock();
    let mut ok = true;
    let mut seg = ABANDONED_HEADS[subproc_id].load(Ordering::Acquire);
    while !seg.is_null() && ok {
        // SAFETY: the lock pins abandoned segments (adopt also takes it).
        unsafe {
            ok = crate::heap::visit_segment_blocks(seg, heap_tag, visit_blocks, false, f);
            seg = (*seg).next;
        }
    }
    abandoned_unlock();
    ok
}

fn abandoned_lock() {
    while ABANDONED_LOCK
        .compare_exchange_weak(false, true, Ordering::Acquire, Ordering::Relaxed)
        .is_err()
    {
        core::hint::spin_loop();
    }
}

fn abandoned_unlock() {
    ABANDONED_LOCK.store(false, Ordering::Release);
}

fn abandoned_push(seg: *mut Segment) {
    let sp = my_subproc();
    abandoned_lock();
    // SAFETY: seg is disowned; its `next` is ours to use as the list link.
    unsafe {
        (*seg).next = ABANDONED_HEADS[sp].load(Ordering::Relaxed);
    }
    ABANDONED_HEADS[sp].store(seg, Ordering::Relaxed);
    ABANDONED_COUNT.fetch_add(1, Ordering::Relaxed);
    abandoned_unlock();
}

/// Pop one abandoned segment from the CALLER's subprocess and take ownership.
/// Returns null when none are available.
pub fn abandoned_pop() -> *mut Segment {
    let sp = my_subproc();
    if ABANDONED_HEADS[sp].load(Ordering::Acquire).is_null() {
        return ptr::null_mut(); // fast no-lock exit for the common case
    }
    abandoned_lock();
    let seg = ABANDONED_HEADS[sp].load(Ordering::Relaxed);
    if !seg.is_null() {
        // SAFETY: list links are protected by the lock.
        unsafe {
            ABANDONED_HEADS[sp].store((*seg).next, Ordering::Relaxed);
            (*seg).next = ptr::null_mut();
            (*seg).thread_id.store(thread_id(), Ordering::Release);
        }
        ABANDONED_COUNT.fetch_sub(1, Ordering::Relaxed);
    }
    abandoned_unlock();
    seg
}