frame_alloc/implementations/

summary_buddy.rs

use crate::allocator::PhysRange;
use crate::{AllocError, InitError, PageSize, PhysicalAllocator, Provenance, RegionInit};
use core::marker::PhantomData;
use core::num::NonZeroUsize;
use core::ptr;
use core::sync::atomic::Ordering::{AcqRel, Acquire, Relaxed, Release};
use core::sync::atomic::{AtomicPtr, AtomicUsize};

/// Number of times the allocator [`SummaryBuddyAllocator`] retries an
/// allocation that failed with a *spurious* `OutOfMemory` before giving up.
const SPURIOUS_OOM_RETRIES: usize = 8;

/// Compile-time tuning of the two summary-scan gates.
///
/// By default, the allocator uses [`DefaultGates`].
pub trait GateConfig {
    /// Minimum L1 words an order must have to be given a summary segment.
    const SUMMARY_MIN_L1_WORDS: usize;
    /// Minimum summary words an order must have before the frontier cursor is used.
    const CURSOR_MIN_SUMMARY_WORDS: usize;
}

/// Default, production gate values.
pub struct DefaultGates;
impl GateConfig for DefaultGates {
    const SUMMARY_MIN_L1_WORDS: usize = 128;
    const CURSOR_MIN_SUMMARY_WORDS: usize = 32;
}

/// `ceil(total_frames / 2^k)` - number of blocks at order `k`. Overflow-safe.
#[inline(always)]
const fn blocks_at_order(total_frames: usize, k: usize) -> usize {
    total_frames.div_ceil(1 << k)
}

/// `ceil(bits / usize::BITS)`.
#[inline(always)]
const fn words_for_bits(bits: usize) -> usize {
    const BPW: usize = usize::BITS as usize;
    bits.div_ceil(BPW)
}

/// Compute the number of `usize` words needed for the bitmap. `summary_min_l1_words`
/// is `G::SUMMARY_MIN_L1_WORDS` threaded in so layout matches the gate in effect.
const fn alloc_bitmap_words_for(
    total_frames: usize,
    orders: usize,
    summary_min_l1_words: usize,
) -> usize {
    let mut total = 0usize;
    let mut k = 0;
    while k < orders {
        let l1 = words_for_bits(blocks_at_order(total_frames, k));
        let summary = if l1 >= summary_min_l1_words {
            words_for_bits(l1)
        } else {
            0
        };
        total += l1 + summary;
        k += 1;
    }
    total
}

/// Validated initialisation parameters produced by [`SummaryBuddyAllocator::validate`]
/// and consumed by [`SummaryBuddyAllocator::commit`]. Carrying it separates the
/// pure validation from the infallible mutation so that a rejected `try_init`
/// never touches allocator state.
struct InitPlan {
    /// Coordinate origin: `phys_base` rounded **down** to a `max_page` boundary.
    /// The sub-`max_page` gap `[base_phys, phys_base)` is a phantom prefix - never
    /// registered as usable, so it stays reserved and costs only bitmap bits.
    base_phys: usize,
    /// Total base frames spanned by `[base_phys, phys_base + span_len)`, i.e.
    /// including the phantom prefix.
    total_frames: usize,
    /// Index into `usable` of the range that hosts the bitmap.
    host_idx: usize,
    /// Frames carved from the host range for the bitmap.
    reserved_frames: usize,
}

/// A lock-free, bitmap-only buddy allocator with a two-level summary bitmap
/// that carves its bitmap out of the managed physical memory region at init time.
///
/// `ORDERS` is the number of size classes: order `0` corresponds to the base
/// frame size (`base.bytes()`), order `ORDERS - 1` to the largest block
/// (`base.bytes() << (ORDERS - 1)`).
///
/// `P` is the [`Provenance`] strategy. Unlike the intrusive backends, this
/// allocator only touches the managed memory at init - to obtain a pointer to the
/// carved bitmap - and never reaches into the frames it hands out; so `P::create`
/// is called exactly once, for the bitmap, and `P::destroy` is never needed (the
/// bitmap is permanent).
pub struct SummaryBuddyAllocator<const ORDERS: usize, P: Provenance, G: GateConfig = DefaultGates> {
    base_frame: PageSize,
    /// Largest page size a caller may request. Caps the allocation size and,
    /// equivalently, the boundary `init` rounds `phys_base` down to.
    max_page: PageSize,
    /// Physical base of the entire managed region (including bitmap frames).
    base_phys: AtomicUsize,
    /// Total base-frame count (including bitmap frames). Zero before init.
    total_frames: AtomicUsize,
    /// Virtual pointer to the flat bitmap stored in its hosting usable range.
    /// Null before `init` runs.
    bitmap: AtomicPtr<u8>,
    /// Word offset into the flat bitmap where order k's segment begins.
    order_word_offsets: [AtomicUsize; ORDERS],
    /// Number of `usize` words in order k's bitmap segment (precomputed).
    bitmap_lens: [AtomicUsize; ORDERS],
    /// Word offset into the flat bitmap where order k's **summary** segment begins.
    summary_word_offsets: [AtomicUsize; ORDERS],
    /// Number of `usize` words in order k's summary segment (precomputed).
    summary_lens: [AtomicUsize; ORDERS],
    /// Per-order summary-scan start hint.
    summary_cursor: [AtomicUsize; ORDERS],
    /// Free block count per order.
    /// Invariant: `free_counts[k] >= actual free blocks at order k`.
    free_counts: [AtomicUsize; ORDERS],
    /// Total base-frames ever registered, for [`AllocatorStats::total_bytes`].
    /// Written single-threaded at init; read for diagnostics. Stats-only.
    #[cfg(any(feature = "stats", test))]
    capacity_frames: AtomicUsize,
    _gates: PhantomData<fn() -> G>,
    _provenance: PhantomData<fn() -> P>,
}

impl<const ORDERS: usize, P: Provenance, G: GateConfig> SummaryBuddyAllocator<ORDERS, P, G> {
    /// Create a new, empty allocator with the given base frame size.
    ///
    /// All runtime state is zeroed. Call `init` / `init_region` to finish
    /// initialisation before allocating. `max_page` defaults to the max block
    /// (`base_frame << (ORDERS-1)`), the most conservative choice; use
    /// [`with_max_page`](Self::with_max_page) to shrink the boundary `init` rounds
    /// `phys_base` down to (a smaller `max_page` shrinks the phantom prefix).
    pub const fn new(base_frame: PageSize) -> Self {
        let max_block = PageSize::from_log2(base_frame.log2() + (ORDERS as u8) - 1);
        Self::with_max_page(base_frame, max_block)
    }

    /// Like [`new`](Self::new) but pins `max_page`, the largest page a caller may
    /// request. It caps allocations (`ps > max_page` -> `InvalidPageSize`) and,
    /// equivalently, sets the boundary `init` rounds `phys_base` down to - the two
    /// are the same bound. Must lie in `base_frame ..= base_frame << (ORDERS-1)`.
    pub const fn with_max_page(base_frame: PageSize, max_page: PageSize) -> Self {
        assert!(ORDERS > 0, "ORDERS must be > 0");
        assert!(
            ORDERS <= usize::BITS as usize,
            "ORDERS exceeds usize bit width"
        );
        assert!(
            base_frame.bytes() >= align_of::<AtomicUsize>(),
            "base_frame must be at least word-aligned so the in-pool bitmap is AtomicUsize-aligned"
        );
        assert!(
            (base_frame.log2() as usize) + ORDERS - 1 < usize::BITS as usize,
            "base_frame.bytes() << (ORDERS-1) overflows usize; reduce ORDERS or base_frame"
        );
        assert!(
            base_frame.log2() <= max_page.log2()
                && (max_page.log2() as usize) < base_frame.log2() as usize + ORDERS,
            "max_page must be in base_frame ..= base_frame << (ORDERS-1)"
        );
        Self {
            base_frame,
            max_page,
            base_phys: AtomicUsize::new(0),
            total_frames: AtomicUsize::new(0),
            bitmap: AtomicPtr::new(ptr::null_mut()),
            order_word_offsets: [const { AtomicUsize::new(0) }; ORDERS],
            bitmap_lens: [const { AtomicUsize::new(0) }; ORDERS],
            summary_word_offsets: [const { AtomicUsize::new(0) }; ORDERS],
            summary_lens: [const { AtomicUsize::new(0) }; ORDERS],
            summary_cursor: [const { AtomicUsize::new(0) }; ORDERS],
            free_counts: [const { AtomicUsize::new(0) }; ORDERS],
            #[cfg(any(feature = "stats", test))]
            capacity_frames: AtomicUsize::new(0),
            _gates: PhantomData,
            _provenance: PhantomData,
        }
    }

    /// Pure validation behind [`RegionInit::try_init`](RegionInit::try_init):
    /// check every argument and locate the bitmap host without mutating any
    /// allocator state. On success it returns an [`InitPlan`] for
    /// [`commit`](Self::commit); on failure the allocator is untouched.
    ///
    /// The bitmap is carved from the first usable range large enough to hold it;
    /// the remainder of that range stays allocatable. Holes between usable ranges
    /// stay reserved. `phys_base` is a pure coordinate origin and need not itself
    /// be usable RAM; it must be base-frame aligned but need **not** be
    /// `max_page`-aligned - init rounds it down to a `max_page` boundary internally
    /// (so a handed-out page is still naturally aligned) and reserves the
    /// sub-`max_page` phantom prefix, which costs only bitmap bits.
    ///
    /// # Errors
    ///
    /// See [`RegionInit::try_init`](RegionInit::try_init). This implementation
    /// reports base-frame misalignment of `phys_base` via [`InitError::Misaligned`].
    fn validate(
        &self,
        phys_base: usize,
        span_len: usize,
        usable: &[PhysRange],
    ) -> Result<InitPlan, InitError> {
        let frame_bytes = self.base_frame.bytes();

        if !self.bitmap.load(Relaxed).is_null() {
            return Err(InitError::AlreadyInitialized);
        }

        if span_len == 0 || !span_len.is_multiple_of(frame_bytes) {
            return Err(InitError::InvalidSpan);
        }

        if !phys_base.is_multiple_of(frame_bytes) {
            return Err(InitError::Misaligned {
                required: frame_bytes,
            });
        }

        // Round the coordinate origin down to a `max_page` boundary so a returned
        // page (≤ max_page) is naturally aligned per the PhysicalAllocator
        // contract. The gap `[base_phys, phys_base)` is a phantom prefix: it is
        // never registered as usable, so it stays reserved and only enlarges the
        // bitmap by the frames it spans. `base_phys` ≤ `phys_base`, both
        // frame-aligned, so `prefix_frames` is exact.
        let base_phys = self.max_page.align_down(phys_base);
        let prefix_frames = (phys_base - base_phys) / frame_bytes;
        let total_frames = prefix_frames + span_len / frame_bytes;

        // Usable ranges live at or above the real `phys_base`; the phantom prefix
        // below it is never usable.
        let span_end = phys_base
            .checked_add(span_len)
            .ok_or(InitError::InvalidSpan)?;
        let mut prev_end = phys_base;
        for (index, r) in usable.iter().enumerate() {
            if r.len == 0
                || !r.base.is_multiple_of(frame_bytes)
                || !r.len.is_multiple_of(frame_bytes)
                || r.base < prev_end
            {
                return Err(InitError::InvalidUsable { index });
            }
            let r_end = r
                .base
                .checked_add(r.len)
                .ok_or(InitError::InvalidUsable { index })?;
            if r_end > span_end {
                return Err(InitError::InvalidUsable { index });
            }
            prev_end = r_end;
        }

        let bitmap_words = alloc_bitmap_words_for(total_frames, ORDERS, G::SUMMARY_MIN_L1_WORDS);
        let bitmap_bytes = bitmap_words * size_of::<usize>();
        let reserved = bitmap_bytes.div_ceil(frame_bytes);
        let required_bytes = reserved * frame_bytes;

        // Find the first usable range that can host the bitmap.
        let host_idx = usable
            .iter()
            .position(|r| r.len >= required_bytes)
            .ok_or(InitError::MetadataWontFit { required_bytes })?;

        Ok(InitPlan {
            base_phys,
            total_frames,
            host_idx,
            reserved_frames: reserved,
        })
    }

    /// Infallible commit behind [`RegionInit::try_init`](RegionInit::try_init):
    /// publish the metadata and register the usable frames from a validated
    /// [`InitPlan`]. This is the first and only mutation of allocator state.
    ///
    /// # Safety
    ///
    /// `plan` must have come from [`validate`](Self::validate) with the same
    /// `usable`. The host usable range must be exclusively owned and reachable
    /// through `P::create`; call once, single-threaded. The allocator must be
    /// published to other threads with a happens-before edge (thread spawn, mutex,
    /// or a Release store / Acquire load of a ready flag) before any concurrent use.
    unsafe fn commit(&self, usable: &[PhysRange], plan: InitPlan) {
        let frame_bytes = self.base_frame.bytes();
        let InitPlan {
            base_phys,
            total_frames,
            host_idx,
            reserved_frames: reserved,
        } = plan;
        let reserved_bytes = reserved * frame_bytes;

        let bitmap_phys = usable[host_idx].base;
        // Create provenance for the carved bitmap.
        // SAFETY: `bitmap_phys` is the base of the validated host range, which the
        // caller guarantees is exclusively owned and reachable through `P::create`.
        let bitmap_virt = unsafe { P::create(bitmap_phys) }.as_ptr();

        // Zero the bitmap:
        // SAFETY: `bitmap_virt` carries provenance over the whole host run, which
        // holds ≥ `reserved` frames.
        unsafe { ptr::write_bytes(bitmap_virt, 0, reserved_bytes) };

        // Initialise lookup metadata.
        let mut off = 0usize;
        for k in 0..ORDERS {
            self.order_word_offsets[k].store(off, Relaxed);
            let words = words_for_bits(blocks_at_order(total_frames, k));
            self.bitmap_lens[k].store(words, Relaxed);
            off += words;
        }
        for k in 0..ORDERS {
            self.summary_word_offsets[k].store(off, Relaxed);
            let l1_words = self.bitmap_lens[k].load(Relaxed);
            let words = if l1_words >= G::SUMMARY_MIN_L1_WORDS {
                words_for_bits(l1_words)
            } else {
                0
            };
            self.summary_lens[k].store(words, Relaxed);
            off += words;
        }
        debug_assert_eq!(
            off,
            alloc_bitmap_words_for(total_frames, ORDERS, G::SUMMARY_MIN_L1_WORDS),
            "bitmap layout size mismatch"
        );

        self.base_phys.store(base_phys, Relaxed);
        self.total_frames.store(total_frames, Relaxed);
        self.bitmap.store(bitmap_virt, Relaxed);

        // Register usable frames.
        let host_alloc_base = bitmap_phys + reserved_bytes;
        for (i, r) in usable.iter().enumerate() {
            let (base, len) = if i == host_idx {
                (host_alloc_base, r.len - reserved_bytes)
            } else {
                (r.base, r.len)
            };
            if len > 0 {
                // SAFETY: `[base, base + len)` is an in-span, frame-aligned,
                // exclusively-owned sub-range; metadata is fully published above.
                unsafe { self.add_region(base, len) };
            }
        }
    }

    /// Register `[base, base + len)` as free memory.
    ///
    /// `init` calls this internally for each usable range. It is also the
    /// primitive behind [`add_usable`](RegionInit::add_usable), so it may
    /// be called after `init` to register a currently-reserved in-span sub-range,
    /// provided it lies within the bitmap's address range `[base_phys, base_phys +
    /// total_frames * base_frame)` and is not already free.
    ///
    /// # Safety
    ///
    /// * `init` must have been called before this method.
    /// * `base` must be aligned to `self.base_frame.bytes()`.
    /// * `len` must be a non-zero multiple of `self.base_frame.bytes()`.
    /// * The memory must be exclusively owned and not accessed through any
    ///   other alias while registered with this allocator.
    unsafe fn add_region(&self, base: usize, len: usize) {
        let base_bytes = self.base_frame.bytes();
        let base_phys = self.base_phys.load(Relaxed);
        let total_frames = self.total_frames.load(Relaxed);

        // Uninitialised allocator -> null deref (UB).
        assert!(
            !self.bitmap.load(Relaxed).is_null(),
            "add_usable/add_region called before init"
        );
        // An out-of-span/overflowing range -> out-of-bounds atomic write (UB).
        let region_end = base
            .checked_add(len)
            .expect("add_region: base + len overflows usize");
        let span_end = base_phys
            .checked_add(total_frames * base_bytes)
            .expect("add_region: span end overflows usize");
        assert!(
            base >= base_phys && region_end <= span_end,
            "add_region: [{base:#x}, {region_end:#x}) falls outside the initialised span [{base_phys:#x}, {span_end:#x})",
        );
        debug_assert_eq!(base % base_bytes, 0, "base not aligned to base frame size");
        debug_assert_eq!(len % base_bytes, 0, "len not a multiple of base frame size");
        debug_assert!(len > 0, "empty region");

        #[cfg(any(feature = "stats", test))]
        self.capacity_frames.fetch_add(len / base_bytes, Relaxed);

        let mut addr = base;
        while addr < region_end {
            let remaining = region_end - addr;
            let order = (0..ORDERS).rev().find(|&k| {
                let block = base_bytes << k;
                block <= remaining && (addr - base_phys).is_multiple_of(block)
            });
            let Some(order) = order else { break };
            let block_size = base_bytes << order;
            unsafe { self.dealloc_order(order, addr) };
            addr += block_size;
        }
    }

    /// Allocate `count` contiguous frames of size `ps`.
    ///
    /// The total `count * ps.bytes()` is rounded up to the next power-of-two
    /// multiple of `base.bytes()` to select the buddy order. If the requested
    /// total is not already such a multiple, the allocated block is larger than
    /// requested and the excess bytes are wasted until the corresponding
    /// `deallocate_physical` call (which must use the same `ps` and `count`).
    #[inline]
    fn alloc(&self, ps: PageSize, count: NonZeroUsize) -> Result<usize, AllocError> {
        let order = Self::order_for(self.base_frame, self.max_page, ps, count)?;
        if order >= ORDERS {
            return Err(AllocError::RequestTooLarge);
        }
        let mut result = self.alloc_order(order);
        let mut attempts = 0;
        while result == Err(AllocError::OutOfMemory) && attempts < SPURIOUS_OOM_RETRIES {
            core::hint::spin_loop();
            result = self.alloc_order(order);
            attempts += 1;
        }
        #[cfg(audit)]
        self.audit();
        result
    }

    /// Return a block of `order` starting at `phys`.
    ///
    /// # Safety
    ///
    /// `phys` must have been returned by [`alloc`](Self::alloc) with the same
    /// `order`, and must not be used after this call.
    #[inline]
    unsafe fn dealloc(&self, ps: PageSize, count: NonZeroUsize, phys: usize) {
        let order = Self::order_for(self.base_frame, self.max_page, ps, count);
        debug_assert!(
            order.is_ok(),
            "deallocate_physical: invalid page size or count"
        );
        if let Ok(order) = order
            && order < ORDERS
        {
            unsafe { self.dealloc_order(order, phys) };
        }
        #[cfg(audit)]
        self.audit();
    }

    #[inline]
    fn alloc_order(&self, order: usize) -> Result<usize, AllocError> {
        let mut k = order;
        // On an *uninitialised* allocator this is null, but every `free_counts[k]`
        // is also zero, so the scan below short-circuits every order and returns
        // `OutOfMemory` before the pointer is ever dereferenced.
        let bitmap_base = self.bitmap.load(Relaxed) as *const AtomicUsize;

        let (found_block, found_order) = 'find: loop {
            if k >= ORDERS {
                return Err(AllocError::OutOfMemory);
            }

            if self.free_counts[k].load(Relaxed) == 0 {
                k += 1;
                continue;
            }

            let n_words = self.bitmap_lens[k].load(Relaxed);
            let word_off = self.order_word_offsets[k].load(Relaxed);
            let n_sum = self.summary_lens[k].load(Relaxed);

            // Fast path: Summary scan.
            if n_sum != 0 {
                let sum_off = self.summary_word_offsets[k].load(Relaxed);
                let use_cursor = n_sum >= G::CURSOR_MIN_SUMMARY_WORDS;
                let cur = if use_cursor {
                    self.summary_cursor[k].load(Relaxed)
                } else {
                    0
                };
                let start = if cur < n_sum { cur } else { 0 };
                for off in 0..n_sum {
                    let sw = {
                        let t = start + off;
                        if t >= n_sum { t - n_sum } else { t }
                    };
                    let s_cell = unsafe { &*bitmap_base.add(sum_off + sw) };
                    let mut s = s_cell.load(Relaxed);
                    while s != 0 {
                        let s_bit = s.trailing_zeros() as usize;
                        let wi = sw * usize::BITS as usize + s_bit;
                        if wi < n_words {
                            let l1_cell = unsafe { &*bitmap_base.add(word_off + wi) };
                            if let Some(bit) =
                                unsafe { self.try_grab_word(bitmap_base, word_off + wi, k, wi) }
                            {
                                if use_cursor && sw != cur {
                                    self.summary_cursor[k].store(sw, Relaxed);
                                }
                                break 'find (wi * usize::BITS as usize + bit, k);
                            }
                            self.clear_summary_bit_then_recheck(s_cell, s_bit, l1_cell);
                        }
                        // This summary bit didn't yield a block. For valid L1 words,
                        // repair the shared stale-positive bit above; then drop it
                        // locally and try the next.
                        s &= !(1usize << s_bit);
                    }
                }
            }

            // Fallback: Direct scan.
            if self.free_counts[k].load(Relaxed) != 0 {
                for wi in 0..n_words {
                    if let Some(bit) =
                        unsafe { self.try_grab_word(bitmap_base, word_off + wi, k, wi) }
                    {
                        break 'find (wi * usize::BITS as usize + bit, k);
                    }
                }
            }
            k += 1;
        };

        let phys = self.block_to_phys(found_block, found_order);

        // Split down to the requested order, freeing the upper buddy at each level.
        let mut block_at_m = found_block;
        let mut m = found_order;
        while m > order {
            m -= 1;
            unsafe {
                self.free_block_no_merge(bitmap_base, m, block_at_m * 2 + 1);
            }
            block_at_m *= 2;
        }

        Ok(phys)
    }

    #[inline]
    unsafe fn dealloc_order(&self, order: usize, phys: usize) {
        let mut k = order;
        let mut block_i = self.phys_to_block(phys, k);
        let bitmap_base = self.bitmap.load(Relaxed) as *const AtomicUsize;
        // dealloc *writes* through `bitmap_base`, so an uninitialised allocator
        // would be a null deref (UB), but covered by the unsafe contract.
        debug_assert!(
            !bitmap_base.is_null(),
            "allocator not initialised; call init before deallocate"
        );

        loop {
            let wi = block_i / usize::BITS as usize;
            let (word_idx, i_bit) = self.bit_addr(k, block_i);
            let j_bit = i_bit ^ 1;
            // SAFETY: word_idx is within the bitmap; bitmap_base is the
            // init-set, non-null base of that allocation.
            let cell = unsafe { &*bitmap_base.add(word_idx) };

            if k + 1 == ORDERS {
                // Top order - no further merge possible.
                self.free_counts[k].fetch_add(1, Relaxed);
                let prev = cell.fetch_or(1usize << i_bit, Release);
                debug_assert!(
                    prev & (1usize << i_bit) == 0,
                    "double-free: order-{k} block already marked free"
                );
                unsafe {
                    self.sync_summary(bitmap_base, cell, k, wi, prev, prev | (1usize << i_bit))
                };
                return;
            }

            loop {
                let old = cell.load(Acquire);
                if (old >> j_bit) & 1 == 1 {
                    // Buddy appears free - consume it.
                    let new = old & !(1usize << j_bit);
                    if cell
                        .compare_exchange_weak(old, new, AcqRel, Acquire)
                        .is_ok()
                    {
                        self.free_counts[k].fetch_sub(1, Relaxed);
                        unsafe { self.sync_summary(bitmap_base, cell, k, wi, old, new) };
                        break; // ascend
                    }
                } else {
                    // Buddy allocated - mark this block free.
                    debug_assert!(
                        (old >> i_bit) & 1 == 0,
                        "double-free: order-{k} block already marked free"
                    );
                    self.free_counts[k].fetch_add(1, Relaxed);
                    let new = old | (1usize << i_bit);
                    match cell.compare_exchange_weak(old, new, AcqRel, Acquire) {
                        Ok(_) => {
                            unsafe { self.sync_summary(bitmap_base, cell, k, wi, old, new) };
                            return;
                        }
                        Err(_) => {
                            self.free_counts[k].fetch_sub(1, Relaxed);
                        }
                    }
                }
            }

            // Buddy consumed; ascend to the parent.
            k += 1;
            block_i = block_i.min(block_i ^ 1) >> 1;
        }
    }

    /// Compute the buddy order for `count` frames of size `ps`.
    ///
    /// `ps` must be a power-of-two multiple of `base_bytes`, which is guaranteed
    /// by construction because [`PageSize`] only represents powers of two and the
    /// caller rejects `ps < base_bytes` below.
    #[inline]
    fn order_for(
        base: PageSize,
        max_page: PageSize,
        ps: PageSize,
        count: NonZeroUsize,
    ) -> Result<usize, AllocError> {
        if ps.bytes() < base.bytes() || ps.bytes() > max_page.bytes() {
            return Err(AllocError::InvalidPageSize);
        }
        let total = ps
            .bytes()
            .checked_mul(count.get())
            .ok_or(AllocError::RequestTooLarge)?;
        // `total >= 1`, so `total - 1` never underflows.
        let frames = ((total - 1) >> base.log2()) + 1;
        let blocks = frames
            .checked_next_power_of_two()
            .ok_or(AllocError::RequestTooLarge)?;
        Ok(blocks.trailing_zeros() as usize)
    }

    #[inline(always)]
    fn phys_to_block(&self, phys: usize, order: usize) -> usize {
        (phys - self.base_phys.load(Relaxed)) >> (self.base_frame.log2() as usize + order)
    }

    #[inline(always)]
    fn block_to_phys(&self, block_i: usize, order: usize) -> usize {
        self.base_phys.load(Relaxed) + (block_i << (self.base_frame.log2() as usize + order))
    }

    /// `(absolute_word_index, bit_position)` for `block_i` at `order`.
    #[inline(always)]
    fn bit_addr(&self, order: usize, block_i: usize) -> (usize, usize) {
        (
            self.order_word_offsets[order].load(Relaxed) + block_i / usize::BITS as usize,
            block_i % usize::BITS as usize,
        )
    }

    /// `(absolute_summary_word_index, bit_within_word)` for L1 word `wi`
    /// (counted within order `order`'s L1 segment).
    #[inline(always)]
    fn summary_addr(&self, order: usize, wi: usize) -> (usize, usize) {
        let bpw = usize::BITS as usize;
        (
            self.summary_word_offsets[order].load(Relaxed) + wi / bpw,
            wi % bpw,
        )
    }

    /// Set the free bit for `block_i` at `order` without checking for a merge.
    /// Used during split, where the buddy is known to be allocated.
    ///
    /// # Safety
    ///
    /// `base` must be the live bitmap base.
    #[inline(always)]
    unsafe fn free_block_no_merge(&self, base: *const AtomicUsize, order: usize, block_i: usize) {
        let wi = block_i / usize::BITS as usize;
        let (word_idx, bit) = self.bit_addr(order, block_i);
        let cell = unsafe { &*base.add(word_idx) };
        self.free_counts[order].fetch_add(1, Relaxed);
        let old = cell.fetch_or(1usize << bit, Release);
        debug_assert!(
            old & (1usize << bit) == 0,
            "double-free: order-{order} block already marked free"
        );
        unsafe { self.sync_summary(base, cell, order, wi, old, old | (1usize << bit)) };
    }

    /// Keep the summary bit for L1 word `wi` consistent with that word's
    /// zero/non-zero state, given an L1 CAS that moved it from `old` to `new`.
    ///
    /// # Safety
    ///
    /// `base` must be the live bitmap base; `l1_cell` must be the L1 word for
    /// `(order, wi)`.
    #[inline(always)]
    unsafe fn sync_summary(
        &self,
        base: *const AtomicUsize,
        l1_cell: &AtomicUsize,
        order: usize,
        wi: usize,
        old: usize,
        new: usize,
    ) {
        if (old == 0) == (new == 0) {
            return;
        }
        if self.summary_lens[order].load(Relaxed) == 0 {
            return;
        }
        let (s_word, s_bit) = self.summary_addr(order, wi);
        let s_cell = unsafe { &*base.add(s_word) };
        if new == 0 {
            self.clear_summary_bit_then_recheck(s_cell, s_bit, l1_cell);
        } else {
            s_cell.fetch_or(1usize << s_bit, Release);
        }
    }

    /// Clear one summary bit, then re-set it if the L1 word is or became
    /// non-empty. The AcqRel clear synchronizes with a concurrent Release set of
    /// the same summary bit, making that freer's preceding L1 write visible to
    /// the recheck.
    #[inline(always)]
    fn clear_summary_bit_then_recheck(
        &self,
        s_cell: &AtomicUsize,
        s_bit: usize,
        l1_cell: &AtomicUsize,
    ) {
        let mask = 1usize << s_bit;
        s_cell.fetch_and(!mask, AcqRel);
        if l1_cell.load(Acquire) != 0 {
            s_cell.fetch_or(mask, Release);
        }
    }

    /// Try to grab one free block from the L1 word at absolute index `word_idx`
    /// (L1 word `wi` within order `order`). Returns the bit on success.
    ///
    /// # Safety
    ///
    /// `base` must be the live bitmap base and `word_idx` in bounds.
    #[inline(always)]
    unsafe fn try_grab_word(
        &self,
        base: *const AtomicUsize,
        word_idx: usize,
        order: usize,
        wi: usize,
    ) -> Option<usize> {
        let cell = unsafe { &*base.add(word_idx) };
        loop {
            let old = cell.load(Relaxed);
            if old == 0 {
                return None;
            }
            let bit = old.trailing_zeros() as usize;
            let new = old & !(1usize << bit);
            if cell
                .compare_exchange_weak(old, new, AcqRel, Acquire)
                .is_ok()
            {
                self.free_counts[order].fetch_sub(1, Relaxed);
                unsafe { self.sync_summary(base, cell, order, wi, old, new) };
                return Some(bit);
            }
        }
    }

    /// Total free bytes. Non-linearizable under concurrent use.
    ///
    /// Internal: the public path is [`AllocatorStats::free_bytes`].
    #[cfg(any(feature = "stats", test))]
    fn free_bytes(&self) -> usize {
        let base = self.bitmap.load(Relaxed) as *const AtomicUsize;
        let mut total = 0usize;
        for k in 0..ORDERS {
            let block_size = self.base_frame.bytes() << k;
            let n_words = self.bitmap_lens[k].load(Relaxed);
            let off = self.order_word_offsets[k].load(Relaxed);
            for wi in 0..n_words {
                let w = unsafe { &*base.add(off + wi) }.load(Relaxed);
                total = total.saturating_add(w.count_ones() as usize * block_size);
            }
        }
        total
    }

    /// Free block count per order. Non-linearizable under concurrent use.
    #[cfg(any(feature = "stats", test))]
    pub fn free_stats(&self) -> [usize; ORDERS] {
        let base = self.bitmap.load(Relaxed) as *const AtomicUsize;
        let mut counts = [0usize; ORDERS];
        for (k, count) in counts.iter_mut().enumerate() {
            let n_words = self.bitmap_lens[k].load(Relaxed);
            let off = self.order_word_offsets[k].load(Relaxed);
            for wi in 0..n_words {
                let w = unsafe { &*base.add(off + wi) }.load(Relaxed);
                *count += w.count_ones() as usize;
            }
        }
        counts
    }

    /// Number of base frames the bitmap occupies inside its hosting usable range
    /// (carved from the first usable range large enough to hold it; for a
    /// whole-span init that is the span start). Returns 0 before `init`.
    #[cfg(any(feature = "stats", test))]
    pub fn reserved_frames(&self) -> usize {
        let total = self.total_frames.load(Relaxed);
        if total == 0 {
            return 0;
        }
        let bitmap_words = alloc_bitmap_words_for(total, ORDERS, G::SUMMARY_MIN_L1_WORDS);
        let bitmap_bytes = bitmap_words * size_of::<usize>();
        bitmap_bytes.div_ceil(self.base_frame.bytes())
    }

    /// Full structural-invariant check, run after every op under `--cfg audit`
    /// (test builds only). The bitmap is lock-free, so this reads `Relaxed` and is
    /// only meaningful for the sequential audit tests.
    ///
    /// Asserts the invariants the split/merge machinery is responsible for:
    ///
    /// * **valid bits** - every set bit is a real block index, never a stray bit
    ///   in the padding tail of an order's last word;
    /// * **merge completeness** - no block and its buddy are both free at the same
    ///   order; they must have coalesced into the parent.
    /// * **no overlap** - expanded to byte intervals, free blocks across all
    ///   orders are pairwise disjoint (catches a sub-block aliased inside a larger
    ///   free block);
    /// * the free total never exceeds the allocatable capacity.
    /// * `free_counts[k]` equals the L1 popcount at every order (in the quiescent
    ///   single-threaded state these runs use) and the summary has no false
    ///   negatives. False-positive summary bits are allowed under concurrent
    ///   maintenance and are repaired lazily by the scanner.
    #[cfg(audit)]
    fn audit(&self) {
        let base = self.bitmap.load(Relaxed) as *const AtomicUsize;
        let base_bytes = self.base_frame.bytes();
        let total_frames = self.total_frames.load(Relaxed);
        if total_frames == 0 {
            return; // not initialised yet
        }

        let mut blocks: alloc::vec::Vec<(usize, usize)> = alloc::vec::Vec::new();

        for k in 0..ORDERS {
            let n_blocks = blocks_at_order(total_frames, k);
            let off = self.order_word_offsets[k].load(Relaxed);
            let n_words = self.bitmap_lens[k].load(Relaxed);
            let block_size = base_bytes << k;
            let mut pop = 0usize;
            for wi in 0..n_words {
                pop += unsafe { &*base.add(off + wi) }.load(Relaxed).count_ones() as usize;

                let mut bits = unsafe { &*base.add(off + wi) }.load(Relaxed);
                while bits != 0 {
                    let block_i = wi * usize::BITS as usize + bits.trailing_zeros() as usize;
                    bits &= bits - 1; // clear lowest set bit

                    assert!(
                        block_i < n_blocks,
                        "audit: stray free bit at order {k}, block {block_i} \
                         ({n_blocks} blocks at this order)"
                    );

                    // Merge completeness: below the top order, a free block's
                    // buddy must not also be free.
                    if k + 1 < ORDERS {
                        let buddy = block_i ^ 1;
                        if buddy < n_blocks {
                            let (bw, bb) = self.bit_addr(k, buddy);
                            let buddy_free =
                                (unsafe { &*base.add(bw) }.load(Relaxed) >> bb) & 1 == 1;
                            assert!(
                                !buddy_free,
                                "audit: order-{k} buddies {block_i} and {buddy} both free \
                                 (should have merged)"
                            );
                        }
                    }

                    let phys = self.block_to_phys(block_i, k);
                    blocks.push((phys, phys + block_size));
                }
            }

            assert_eq!(
                pop,
                self.free_counts[k].load(Relaxed),
                "audit: free_counts[{k}] disagrees with the order-{k} L1 popcount"
            );
        }

        assert!(
            self.debug_summary_consistent(),
            "audit: summary inconsistent"
        );

        // No two free blocks may overlap (adjacency is fine).
        blocks.sort_unstable_by_key(|&(start, _)| start);
        for w in blocks.windows(2) {
            assert!(
                w[0].1 <= w[1].0,
                "audit: overlapping free blocks [{:#x},{:#x}) and [{:#x},{:#x})",
                w[0].0,
                w[0].1,
                w[1].0,
                w[1].1
            );
        }

        // The free total can never exceed the allocatable (non-bitmap) capacity.
        let bitmap_words = alloc_bitmap_words_for(total_frames, ORDERS, G::SUMMARY_MIN_L1_WORDS);
        let reserved = (bitmap_words * size_of::<usize>()).div_ceil(base_bytes);
        let free: usize = blocks.iter().map(|&(s, e)| e - s).sum();
        let allocatable = (total_frames - reserved) * base_bytes;
        assert!(
            free <= allocatable,
            "audit: free bytes {free:#x} exceed allocatable capacity {allocatable:#x}"
        );
    }

    /// Test-only corruption hooks used by the audit non-vacuity tests.
    #[cfg(all(test, audit))]
    pub(crate) fn corrupt_set_free_bit(&self, order: usize, block_i: usize) {
        let base = self.bitmap.load(Relaxed) as *const AtomicUsize;
        let (w, b) = self.bit_addr(order, block_i);
        unsafe { &*base.add(w) }.fetch_or(1usize << b, Relaxed);
    }

    /// Test-only corruption hook: set the summary bit for L1 word `wi` without
    /// changing the L1 word. This models a stale-positive summary bit.
    #[cfg(test)]
    pub(crate) fn corrupt_set_summary_bit(&self, order: usize, wi: usize) {
        assert_ne!(
            self.summary_lens[order].load(Relaxed),
            0,
            "order-{order} has no summary segment"
        );
        let base = self.bitmap.load(Relaxed) as *const AtomicUsize;
        let (w, b) = self.summary_addr(order, wi);
        unsafe { &*base.add(w) }.fetch_or(1usize << b, Relaxed);
    }

    /// Test-only: run the structural audit on demand (see [`Self::audit`]).
    #[cfg(all(test, audit))]
    pub(crate) fn run_audit(&self) {
        self.audit();
    }

    /// Test-only invariant: every non-empty L1 word must have its summary bit
    /// set. Under concurrent maintenance the reverse is intentionally weaker:
    /// stale-positive summary bits may point at empty L1 words and are repaired
    /// lazily by the scanner.
    #[cfg(any(test, audit))]
    pub(crate) fn debug_summary_consistent(&self) -> bool {
        let base = self.bitmap.load(Relaxed) as *const AtomicUsize;
        if base.is_null() {
            return false;
        }
        let bpw = usize::BITS as usize;
        for k in 0..ORDERS {
            if self.summary_lens[k].load(Relaxed) == 0 {
                continue;
            }
            let n_words = self.bitmap_lens[k].load(Relaxed);
            let word_off = self.order_word_offsets[k].load(Relaxed);
            let sum_off = self.summary_word_offsets[k].load(Relaxed);
            for wi in 0..n_words {
                let l1 = unsafe { &*base.add(word_off + wi) }.load(Relaxed);
                let s = unsafe { &*base.add(sum_off + wi / bpw) }.load(Relaxed);
                let bit = (s >> (wi % bpw)) & 1;
                if l1 != 0 && bit == 0 {
                    return false;
                }
            }
        }
        true
    }

    /// Test-only exact summary check for sequential/quiescent tests. Concurrent
    /// operation only guarantees [`Self::debug_summary_consistent`].
    #[cfg(test)]
    pub(crate) fn debug_summary_exact(&self) -> bool {
        let base = self.bitmap.load(Relaxed) as *const AtomicUsize;
        if base.is_null() {
            return false;
        }
        let bpw = usize::BITS as usize;
        for k in 0..ORDERS {
            if self.summary_lens[k].load(Relaxed) == 0 {
                continue;
            }
            let n_words = self.bitmap_lens[k].load(Relaxed);
            let word_off = self.order_word_offsets[k].load(Relaxed);
            let sum_off = self.summary_word_offsets[k].load(Relaxed);
            for wi in 0..n_words {
                let l1 = unsafe { &*base.add(word_off + wi) }.load(Relaxed);
                let s = unsafe { &*base.add(sum_off + wi / bpw) }.load(Relaxed);
                let bit = (s >> (wi % bpw)) & 1;
                let expect = usize::from(l1 != 0);
                if bit != expect {
                    return false;
                }
            }
        }
        true
    }
}

unsafe impl<const ORDERS: usize, P: Provenance, G: GateConfig> RegionInit
    for SummaryBuddyAllocator<ORDERS, P, G>
{
    unsafe fn try_init(
        &self,
        phys_base: usize,
        span_len: usize,
        usable: &[PhysRange],
    ) -> Result<(), InitError> {
        let plan = self.validate(phys_base, span_len, usable)?;
        // SAFETY: `plan` came from `validate` with these same arguments; the
        // caller upholds the trait-level provenance/ownership/once contract.
        unsafe { self.commit(usable, plan) };
        Ok(())
    }

    unsafe fn add_usable(&self, base: usize, len: usize) {
        unsafe { self.add_region(base, len) };
    }
}

unsafe impl<const ORDERS: usize, P: Provenance, G: GateConfig> PhysicalAllocator
    for SummaryBuddyAllocator<ORDERS, P, G>
{
    #[inline]
    fn allocate_physical(&self, ps: PageSize, count: NonZeroUsize) -> Result<usize, AllocError> {
        self.alloc(ps, count)
    }

    #[inline]
    unsafe fn deallocate_physical(&self, ps: PageSize, count: NonZeroUsize, phys: usize) {
        unsafe { self.dealloc(ps, count, phys) }
    }
}

#[cfg(any(feature = "stats", test))]
impl<const ORDERS: usize, P: Provenance, G: GateConfig> crate::AllocatorStats
    for SummaryBuddyAllocator<ORDERS, P, G>
{
    fn total_bytes(&self) -> usize {
        self.capacity_frames.load(Relaxed) * self.base_frame.bytes()
    }

    fn free_bytes(&self) -> usize {
        // Inherent method (same name) wins method-call resolution - no recursion.
        self.free_bytes()
    }

    fn largest_free_bytes(&self) -> usize {
        let counts = self.free_stats();
        // Highest order with a free block is the largest run that can be served.
        for k in (0..ORDERS).rev() {
            if counts[k] > 0 {
                return self.base_frame.bytes() << k;
            }
        }
        0
    }
}