heap.ccm

Go to the documentation of this file.

// SPDX-FileCopyrightText: 2026 Edward Kmett <ekmett@gmail.com>
// SPDX-License-Identifier: BSD-2-Clause OR Apache-2.0

module;
#if defined(__linux__) && !defined(_GNU_SOURCE)
#define _GNU_SOURCE
#endif
#include <native/attributes.h>
#include <cstdio>
#include <cstdlib>
#include <algorithm>
#include <array>
#include <atomic>
#include <cassert>
#include <cerrno>
#include <cstddef>
#include <concepts>
#include <compare>
#include <limits>
#include <memory>
#include <tuple>
#include <type_traits>
#include <variant>
#include <utility>
#include <vector>
#include <deque>
#include <unordered_map>
#if defined(_WIN32)
#define WIN32_LEAN_AND_MEAN
#define NOMINMAX
#include <windows.h>
#else
#include <sys/mman.h>
#include <unistd.h>
#endif
#if defined(__APPLE__)
#include <mach/mach.h>
#include <mach/mach_vm.h>
#elif !defined(__linux__) && !defined(_WIN32)
#error "jam requires macOS, Linux or Windows virtual-memory aliases"
#endif
 
#include <bit>
#include <cstdint>
#include <cstring>
#include <functional>
#include <span>
#include <stdexcept>
 
 
export module jam:heap;
import :work;
import :compact;
export import :units;
 
namespace jam::detail {
 
native_cold native_noreturn
inline void heap_failure(char const * operation, int code = 0) noexcept {
  std::fprintf(stderr, "jam: %s (%d)\n", operation, code);
  std::abort();
}
 
// One reservation gives both generations stable, ordered address ranges. Only
// the active ring views acquire backing; maximum capacities reserve address space.
class generation_reservation {
  std::byte * base_;
  std::size_t bytes_;
public:
  generation_reservation(std::size_t old_maximum, std::size_t young_maximum) noexcept
      : bytes_(2 * (old_maximum + young_maximum)) {
#if defined(_WIN32)
    base_ = static_cast<std::byte *>(::VirtualAlloc2(nullptr, nullptr, bytes_,
      MEM_RESERVE | MEM_RESERVE_PLACEHOLDER, PAGE_NOACCESS, nullptr, 0));
    if (!base_) heap_failure("reserve generation address space", ::GetLastError());
#else
    auto * p = ::mmap(nullptr, bytes_, PROT_NONE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
    if (p == MAP_FAILED) heap_failure("reserve generation address space", errno);
    base_ = static_cast<std::byte *>(p);
#endif
  }
  generation_reservation(generation_reservation const &) = delete;
  ~generation_reservation() noexcept {
#if defined(_WIN32)
    MEMORY_BASIC_INFORMATION region{};
    if (!::VirtualQuery(base_, &region, sizeof(region))) heap_failure("query generation reservation", ::GetLastError());
    if (region.RegionSize != bytes_ &&
        !::VirtualFree(base_, bytes_, MEM_RELEASE | MEM_COALESCE_PLACEHOLDERS))
      heap_failure("coalesce generation reservation", ::GetLastError());
    if (!::VirtualFree(base_, 0, MEM_RELEASE)) heap_failure("release generation reservation", ::GetLastError());
#else
    if (::munmap(base_, bytes_)) heap_failure("release generation reservation", errno);
#endif
  }
  std::byte * data() const noexcept { return base_; }
};

class heap_mapping {
#if defined(_WIN32)
  struct backing {
    HANDLE handle;
    explicit backing(std::size_t bytes) noexcept
      : handle(::CreateFileMappingW(INVALID_HANDLE_VALUE, nullptr, PAGE_READWRITE,
          static_cast<DWORD>(bytes >> 32), static_cast<DWORD>(bytes), nullptr)) {
      if (!handle) system_failure("create heap backing");
    }
    ~backing() noexcept { ::CloseHandle(handle); }
  };
#endif
  struct span {
    std::size_t begin, end;
#if defined(_WIN32)
    std::shared_ptr<backing> section;
    std::size_t offset;
#endif
  };
  struct reservation {};
  std::byte * address = nullptr;
  std::size_t capacity = 0;
  std::size_t fixed_extent = 0;
  // Linux's zero-old-size mremap must remain within one original VMA/object.
  std::vector<span> spans;
 
  [[noreturn]] static void system_failure(char const * operation) noexcept {
#if defined(_WIN32)
    heap_failure(operation, static_cast<int>(::GetLastError()));
#else
    heap_failure(operation, errno);
#endif
  }
  static void validate_size(std::size_t bytes) noexcept {
    if (!bytes || bytes % page_size() || bytes > std::numeric_limits<std::size_t>::max() / 2)
      heap_failure("capacity must be a nonzero whole-page size");
  }
  explicit heap_mapping(reservation, std::size_t bytes) noexcept {
    validate_size(bytes);
#if defined(_WIN32)
    auto const result = ::VirtualAlloc2(nullptr, nullptr, bytes * 2,
      MEM_RESERVE | MEM_RESERVE_PLACEHOLDER, PAGE_NOACCESS, nullptr, 0);
    if (!result) system_failure("reserve heap placeholders");
#else
    auto const result = ::mmap(nullptr, bytes * 2, PROT_NONE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
    if (result == MAP_FAILED) system_failure("jam: reserve virtual memory");
#endif
    address = static_cast<std::byte *>(result);
    capacity = bytes;
  }
#if defined(_WIN32)
  static void alias(span const & source, std::byte * target, std::size_t count,
                    std::size_t offset = 0) noexcept {
    MEMORY_BASIC_INFORMATION region{};
    if (!::VirtualQuery(target, &region, sizeof(region))) system_failure("query heap placeholder");
    auto const begin = static_cast<std::byte *>(region.BaseAddress);
    auto const prefix = static_cast<std::size_t>(target - begin);
    if (prefix && !::VirtualFree(begin, prefix, MEM_RELEASE | MEM_PRESERVE_PLACEHOLDER))
      system_failure("split heap placeholder prefix");
    if (count < region.RegionSize - prefix &&
        !::VirtualFree(target, count, MEM_RELEASE | MEM_PRESERVE_PLACEHOLDER))
      system_failure("split heap placeholder suffix");
    if (!::MapViewOfFile3(source.section->handle, ::GetCurrentProcess(), target,
          source.offset + offset, count, MEM_REPLACE_PLACEHOLDER, PAGE_READWRITE, nullptr, 0))
      system_failure("alias heap backing");
  }
#else
  static void alias(std::byte * source, std::byte * target, std::size_t count) noexcept {
#if defined(__APPLE__)
    auto destination = reinterpret_cast<mach_vm_address_t>(target);
    vm_prot_t current = 0, maximum = 0;
    auto const result = ::mach_vm_remap(::mach_task_self(), &destination, count, 0,
      VM_FLAGS_FIXED | VM_FLAGS_OVERWRITE, ::mach_task_self(),
      reinterpret_cast<mach_vm_address_t>(source), false, &current, &maximum, VM_INHERIT_SHARE);
    if (result != KERN_SUCCESS)
      heap_failure("alias virtual memory", result);
    assert(destination == reinterpret_cast<mach_vm_address_t>(target));
#else
    auto const result = ::mremap(source, 0, count, MREMAP_MAYMOVE | MREMAP_FIXED, target);
    if (result == MAP_FAILED) system_failure("jam: alias virtual memory");
    assert(result == target);
#endif
  }
#endif
  void fresh(std::size_t begin, std::size_t count) noexcept {
    if (!count) return;
#if defined(_WIN32)
    spans.push_back({begin, begin + count, std::make_shared<backing>(count), 0});
    alias(spans.back(), address + begin, count);
#else
    auto const result = ::mmap(address + begin, count, PROT_READ | PROT_WRITE,
      MAP_FIXED | MAP_SHARED | MAP_ANONYMOUS, -1, 0);
    if (result == MAP_FAILED) system_failure("jam: map fresh backing");
    spans.push_back({begin, begin + count});
#endif
  }
  void finish_aliases() noexcept {
    std::sort(spans.begin(), spans.end(), [](auto const & a, auto const & b) noexcept { return a.begin < b.begin; });
    for (auto const & s : spans)
#if defined(_WIN32)
      alias(s, address + capacity + s.begin, s.end - s.begin);
#else
      alias(address + s.begin, address + capacity + s.begin, s.end - s.begin);
#endif
  }
  void discard_gap(std::size_t start, std::size_t live) const noexcept {
#if defined(_WIN32)
    // Advisory discard releases dead physical pages; a partially retained section
    // keeps its commit charge until the last surviving view/handle is released.
    for (auto const & s : spans) {
      auto const discard = [&](std::size_t begin, std::size_t end) noexcept {
        begin = std::max(begin, s.begin);
        end = std::min(end, s.end);
        if (begin < end) static_cast<void>(::VirtualAlloc(address + begin, end - begin, MEM_RESET, PAGE_READWRITE));
      };
      if (start + live <= capacity) {
        discard(0, start);
        discard(start + live, capacity);
      } else discard(start + live - capacity, start);
    }
#else
    auto const begin = (start + live) % capacity;
    auto const count = capacity - live;
    auto const first = std::min(count, capacity - begin);
#if defined(__APPLE__)
    constexpr auto advice = MADV_FREE_REUSABLE;
#else
    constexpr auto advice = MADV_REMOVE;
#endif
    // dev: advisory release; kernels may retain dead pages until memory pressure.
    if (first) static_cast<void>(::madvise(address + begin, first, advice));
    if (count > first) static_cast<void>(::madvise(address, count - first, advice));
#endif
  }
 
  void reset_fixed() noexcept {
    assert(fixed_extent);
#if defined(_WIN32)
    for (auto const & s : spans) {
      if (!::UnmapViewOfFile2(::GetCurrentProcess(), address + s.begin, MEM_PRESERVE_PLACEHOLDER) ||
          !::UnmapViewOfFile2(::GetCurrentProcess(), address + capacity + s.begin, MEM_PRESERVE_PLACEHOLDER))
        system_failure("restore generation placeholders");
    }
    MEMORY_BASIC_INFORMATION region{};
    if (!::VirtualQuery(address, &region, sizeof(region))) system_failure("query generation placeholder");
    if (region.RegionSize < fixed_extent &&
        !::VirtualFree(address, fixed_extent, MEM_RELEASE | MEM_COALESCE_PLACEHOLDERS))
      system_failure("coalesce generation placeholders");
#else
    if (::mmap(address, fixed_extent, PROT_NONE, MAP_FIXED | MAP_PRIVATE | MAP_ANONYMOUS, -1, 0) == MAP_FAILED)
      system_failure("restore generation reservation");
#endif
    spans.clear();
  }
 
public:
  explicit heap_mapping(std::size_t bytes) noexcept : heap_mapping(reservation{}, bytes) {
    spans.reserve(1);
    fresh(0, bytes);
    finish_aliases();
  }
  heap_mapping(std::size_t bytes, std::byte * base, std::size_t maximum) noexcept
      : address(base), capacity(bytes), fixed_extent(2 * maximum) {
    validate_size(bytes);
    if (bytes > maximum) heap_failure("generation exceeds reserved address range");
    spans.reserve(1);
    fresh(0, bytes);
    finish_aliases();
  }
  // Stage surviving aliases before touching the fixed range: source and target
  // rings may overlap during growth, shrinking or view rotation.
  void replace(heap_mapping & staged) noexcept {
    if (!fixed_extent) { swap(staged); return; }
    if (staged.capacity * 2 > fixed_extent) heap_failure("generation growth exceeds maximum");
    reset_fixed();
    capacity = staged.capacity;
    for (auto const & s : staged.spans) {
#if defined(_WIN32)
      alias(s, address + s.begin, s.end - s.begin);
#else
      alias(staged.address + s.begin, address + s.begin, s.end - s.begin);
#endif
    }
    spans = staged.spans;
    finish_aliases();
  }
  heap_mapping(heap_mapping const &) = delete;
  heap_mapping & operator=(heap_mapping const &) = delete;
  heap_mapping(heap_mapping && other) noexcept
    : address(std::exchange(other.address, nullptr)), capacity(std::exchange(other.capacity, 0)),
      fixed_extent(std::exchange(other.fixed_extent, 0)), spans(std::move(other.spans)) {}
  heap_mapping & operator=(heap_mapping && other) noexcept { swap(other); return *this; }
  ~heap_mapping() noexcept {
    if (!address) return;
    if (fixed_extent) { reset_fixed(); return; }
#if defined(_WIN32)
    for (auto const & s : spans) {
      ::UnmapViewOfFile(address + s.begin);
      ::UnmapViewOfFile(address + capacity + s.begin);
    }
#else
    static_cast<void>(::munmap(address, capacity * 2));
#endif
  }
  native_inline native_pure std::byte * data() const noexcept { return address; }
  native_inline native_pure std::size_t bytes() const noexcept { return capacity; }
  static std::size_t page_size() noexcept {
#if defined(_WIN32)
    SYSTEM_INFO info{};
    ::GetSystemInfo(&info);
    return info.dwPageSize;
#else
    return static_cast<std::size_t>(::getpagesize());
#endif
  }
  void publish(std::size_t start, std::byte * target, std::size_t count) const noexcept {
#if defined(_WIN32)
    static_cast<void>(start); static_cast<void>(target); static_cast<void>(count);
    heap_failure("host publication is not supported on Windows");
#else
    if (start >= capacity || start % page_size() || count % page_size() ||
        count > capacity || reinterpret_cast<std::uintptr_t>(target) % page_size())
      heap_failure("invalid published heap arc");
    while (count) {
      auto const where = std::lower_bound(spans.begin(), spans.end(), start,
        [](span const & s, std::size_t value) noexcept { return s.end <= value; });
      assert(where != spans.end() && where->begin <= start);
      auto const part = std::min(count, where->end - start);
      alias(address + start, target, part);
      start = (start + part) % capacity;
      target += part;
      count -= part;
    }
#endif
  }
  void swap(heap_mapping & other) noexcept {
    std::swap(address, other.address);
    std::swap(capacity, other.capacity);
    std::swap(fixed_extent, other.fixed_extent);
    spans.swap(other.spans);
  }
  heap_mapping resized(std::size_t old_start_bytes, std::size_t live_bytes,
                       std::size_t new_bytes, std::size_t new_start_bytes) const noexcept {
    validate_size(new_bytes);
    auto const page = page_size();
    if (!address || old_start_bytes >= capacity || new_start_bytes >= new_bytes ||
        old_start_bytes % page || new_start_bytes % page ||
        live_bytes > capacity || live_bytes > new_bytes)
      heap_failure("invalid circular remap range");
    auto const live = ((live_bytes + page - 1) / page) * page;
    heap_mapping result(reservation{}, new_bytes);
    result.spans.reserve(spans.size() + 4);
    auto source = old_start_bytes;
    auto target = new_start_bytes;
    auto remaining = live;
    while (remaining) {
      auto const where = std::lower_bound(spans.begin(), spans.end(), source,
        [](span const & s, std::size_t value) noexcept { return s.end <= value; });
      assert(where != spans.end() && where->begin <= source);
      auto const count = std::min({remaining, where->end - source, new_bytes - target});
#if defined(_WIN32)
      alias(*where, result.address + target, count, source - where->begin);
      result.spans.push_back({target, target + count, where->section, where->offset + source - where->begin});
#else
      alias(address + source, result.address + target, count);
      result.spans.push_back({target, target + count});
#endif
      source = (source + count) % capacity;
      target = (target + count) % new_bytes;
      remaining -= count;
    }
    if (!live) result.fresh(0, new_bytes);
    else if (new_start_bytes + live <= new_bytes) {
      result.fresh(0, new_start_bytes);
      result.fresh(new_start_bytes + live, new_bytes - new_start_bytes - live);
    } else result.fresh(new_start_bytes + live - new_bytes, new_bytes - live);
    result.finish_aliases();
    discard_gap(old_start_bytes, live);
    return result;
  }
};
}
 
namespace jam {
 
export class heap;
export template<class T> class root;
export template<class T> class ptr;
export template<class T> class weak_ptr;
export template<class T> class weak_root;
export template<class T> class weak;

export template<class V>
concept visitor = std::same_as<typename V::heap_type, heap>;
 
namespace detail {
#if defined(JAM_CONTEXT_X28)
#if !defined(__aarch64__)
#error "JAM_CONTEXT_X28 requires AArch64 and -ffixed-x28"
#endif
register heap * active_heap asm("x28");
#else
inline thread_local heap * active_heap = nullptr;
#endif
template<class T, class H = heap>
concept member_trace = requires(T & value, typename H::visitor & visit) {
  value.trace(visit);
};
template<class T, class H = heap>
concept cooperative_trace = requires(typename H::visitor & visit, ptr<T> const & at) {
  T::trace(visit, at);
};
template<class T, class H = heap>
concept allocation_trace = requires(T const & value, typename H::visitor & visit) {
  { value.claim_and_trace(visit) } noexcept -> std::same_as<void>;
};
template<class T>
concept has_allocation_trace = requires { &T::claim_and_trace; } || allocation_trace<T>;
template<class T>
concept has_trace = requires { &T::trace; } || member_trace<T> || cooperative_trace<T>;
 
struct structural_tracer {};
template<class T, class... Members>
struct member_manifest {
  std::tuple<Members T::*...> members;
  template<class Visitor>
  static constexpr bool accepts = (requires(Visitor & visit, Members const & value) {
    visit(value);
  } && ...);
};
template<class T, class Descriptor> inline constexpr bool manifest_for = false;
template<class T, class... Members>
inline constexpr bool manifest_for<T, member_manifest<T, Members...>> = true;
template<class T>
concept has_manifest = requires {
  requires manifest_for<T, std::remove_cvref_t<decltype(T::manifest)>>;
};
struct invalid_tracer {};
}

export template<class T, class... Members> requires (std::is_object_v<Members> && ...)
[[nodiscard]] consteval auto make_manifest(Members T::*... members) {
  if (((members == nullptr) || ...)) throw "manifest members must not be null";
  return detail::member_manifest<T, Members...>{{members...}};
}

export class heap_scope {
  heap * previous;
public:
  [[nodiscard]] native_inline explicit heap_scope(heap & arena) noexcept;
  native_inline ~heap_scope() noexcept { detail::active_heap = previous; }
  heap_scope(heap_scope const &) = delete;
  heap_scope & operator=(heap_scope const &) = delete;
};

export template<class T>
struct leaf {
  template<class Visitor>
  static constexpr void trace(Visitor &, T const &) noexcept {}
};

export template<class T>
struct tracer : std::conditional_t<requires { T::manifest; } || detail::has_allocation_trace<T>, detail::invalid_tracer, leaf<T>> {};

export template<class T> requires detail::has_trace<T>
struct tracer<T> {
  template<class Visitor>
  static constexpr auto trace(Visitor & visit, T const & value)
      noexcept(noexcept(value.trace(visit))) -> decltype(value.trace(visit)) {
    return value.trace(visit);
  }

  template<class Visitor>
  static constexpr decltype(auto) trace(Visitor & visit, ptr<T> const & at) noexcept
      requires requires { T::trace(visit, at); } {
    return T::trace(visit, at);
  }
};
 
namespace detail {
template<class T, class H>
concept traceable_in = (!has_allocation_trace<T> || allocation_trace<T, H>)
    && (allocation_trace<T, H> || requires(T const & value, typename H::visitor & visit) {
  sizeof(T);
  tracer<T>::trace(visit, value);
} || requires(ptr<T> const & at, typename H::visitor & visit) {
  sizeof(T);
  tracer<T>::trace(visit, at);
});
}

export template<class T>
concept traceable = detail::traceable_in<T, heap>;

export template<class T>
class ptr {
public:
  using element_type = T;
  using offset = std::uint32_t;
 
private:
  mutable offset position = 0;
  native_inline constexpr void update_barrier() noexcept;
 
public:
  constexpr ptr() noexcept = default;
  constexpr ptr(std::same_as<std::nullptr_t> auto) noexcept {}
  native_inline explicit constexpr ptr(offset at) noexcept;
  native_inline constexpr ptr(ptr const & other) noexcept;
  template<class U> requires (!std::same_as<T, U>) && std::derived_from<U, T>
      && detail::allocation_trace<T> && requires(T * p) { static_cast<U *>(p); }
  native_inline constexpr ptr(ptr<U> const & other) noexcept;
 
  native_inline constexpr ptr & operator=(ptr const & other) noexcept;
  native_inline constexpr ptr(ptr && other) noexcept;
  native_inline constexpr ptr & operator=(ptr && other) noexcept;
  native_inline constexpr void swap(ptr & other) noexcept;
  friend native_inline constexpr void swap(ptr & left, ptr & right) noexcept { left.swap(right); }
  native_inline constexpr ~ptr() noexcept;
  constexpr ptr & unsafe_assign(ptr const & other) noexcept { position = other.position; return *this; }
  constexpr ptr & unsafe_assign(offset bits) noexcept { position = bits; return *this; }
  [[nodiscard]] constexpr bool is_young() const noexcept { return (position & 0x80000000u) != 0; }
  [[nodiscard]] constexpr offset local_offset() const noexcept { return position & 0x7fffffffu; }
  [[nodiscard]] native_inline native_pure
  constexpr offset get() const noexcept { return position; }
  [[nodiscard]] native_inline native_pure
  explicit constexpr operator bool() const noexcept {
    return position != 0;
  }

  [[nodiscard]] native_inline T * operator->() const noexcept;
  [[nodiscard]] native_inline T & operator*() const noexcept;
  native_inline void prefetch_marks() const noexcept;
  native_inline void prefetch() const noexcept;
  [[nodiscard]] constexpr auto operator<=>(ptr const &) const noexcept = default;
};

export template<class T>
class weak_ptr {
public:
  using element_type = T;
  using offset = std::uint32_t;
private:
  mutable offset position = 0;
  native_inline constexpr void update_barrier() noexcept;
public:
  constexpr weak_ptr() noexcept = default;
  constexpr weak_ptr(std::same_as<std::nullptr_t> auto) noexcept {}
  native_inline constexpr weak_ptr(ptr<T> const & value) noexcept;
  native_inline weak_ptr(root<T> const & value) noexcept;
  template<class U> requires (!std::same_as<T, U>) && std::convertible_to<ptr<U>, ptr<T>>
  native_inline constexpr weak_ptr(ptr<U> const & value) noexcept : weak_ptr(ptr<T>{value}) {}
  native_inline constexpr weak_ptr(weak_ptr const & other) noexcept;
  native_inline constexpr weak_ptr & operator=(weak_ptr const & other) noexcept;
  native_inline constexpr weak_ptr(weak_ptr && other) noexcept;
  native_inline constexpr weak_ptr & operator=(weak_ptr && other) noexcept;
  native_inline constexpr ~weak_ptr() noexcept;
  native_inline constexpr void swap(weak_ptr & other) noexcept;
  friend native_inline constexpr void swap(weak_ptr & a, weak_ptr & b) noexcept { a.swap(b); }
  constexpr weak_ptr & unsafe_assign(offset bits) noexcept { position = bits; return *this; }
  constexpr weak_ptr & unsafe_assign(weak_ptr const & other) noexcept { return unsafe_assign(other.position); }
  [[nodiscard]] native_inline native_pure constexpr offset get() const noexcept { return position; }
  [[nodiscard]] constexpr bool is_young() const noexcept { return (position & 0x80000000u) != 0; }
  [[nodiscard]] constexpr bool expired() const noexcept { return position == 0; }
  [[nodiscard]] native_inline root<T> lock() const noexcept requires traceable<T>;
  [[nodiscard]] constexpr auto operator<=>(weak_ptr const &) const noexcept = default;
};

export template<class T> weak_ptr(ptr<T> const &) -> weak_ptr<T>;
export template<class T> weak_ptr(root<T> const &) -> weak_ptr<T>;
 
namespace detail {
template<class T> inline constexpr bool is_ptr = false;
template<class T> inline constexpr bool is_ptr<ptr<T>> = true;
template<class T> inline constexpr bool is_weak_ptr = false;
template<class T> inline constexpr bool is_weak_ptr<weak_ptr<T>> = true;
template<class T, class Visitor>
concept visits = requires(Visitor & visit, std::remove_reference_t<T> const & value) {
  tracer<std::remove_cvref_t<T>>::trace(visit, value);
};
}

export template<class T>
struct tracer<ptr<T>> {
  template<visitor Visitor>
  static constexpr void trace(Visitor & visit, ptr<T> const & value)
      noexcept(noexcept(visit(value))) {
    visit(value);
  }
};

export template<class T>
struct tracer<weak_ptr<T>> {
  template<visitor Visitor>
  static constexpr void trace(Visitor & visit, weak_ptr<T> const & value)
      noexcept(noexcept(visit(value))) { visit(value); }
};
 
namespace detail {
template<class Visitor, class... Ts>
constexpr void trace_parts(Visitor & visit, Ts const &... values) noexcept {
  auto part = [&](auto const & value) constexpr noexcept {
    if constexpr (requires { visit(value); }) visit(value);
    else tracer<std::remove_cvref_t<decltype(value)>>::trace(visit, value);
  };
  (part(values), ...);
}
}

export template<class... Ts>
struct tracer<std::tuple<Ts...>> : detail::structural_tracer {
  template<class Visitor>
  static constexpr void trace(Visitor & visit, std::tuple<Ts...> const & value)
      requires (detail::visits<Ts, Visitor> && ...) {
    std::apply([&](auto const &... part) constexpr {
      detail::trace_parts(visit, part...);
    }, value);
  }
};

export template<class T, std::size_t N>
struct tracer<std::array<T, N>> : detail::structural_tracer {
  template<class Visitor>
  static constexpr void trace(Visitor & visit, std::array<T, N> const & value)
      requires detail::visits<T, Visitor> {
    for (auto const & part : value) detail::trace_parts(visit, part);
  }
};

export template<class... Ts>
struct tracer<std::variant<Ts...>> : detail::structural_tracer {
  template<class Visitor>
  static constexpr void trace(Visitor & visit, std::variant<Ts...> const & value)
      requires (detail::visits<Ts, Visitor> && ...) {
    if (value.valueless_by_exception()) return;
    std::visit([&](auto const & part) constexpr {
      detail::trace_parts(visit, part);
    }, value);
  }
};

export template<class T> requires (!detail::has_trace<T>) && detail::has_manifest<T>
struct tracer<T> : detail::structural_tracer {
  template<class Visitor>
  static native_inline constexpr void trace(Visitor & visit, T const & value) noexcept
      requires std::remove_cvref_t<decltype(T::manifest)>::template accepts<Visitor> {
    auto const & [...member] = T::manifest.members;
    visit((value.*member)...);
  }

  template<class Visitor>
  static void trace(Visitor & visit, ptr<T> const & at) noexcept
      requires std::remove_cvref_t<decltype(T::manifest)>::template accepts<Visitor> {
    visit.template trace_manifest<T>(at);
  }
};

export struct generation_options {
  units::pages capacity = 1_MiB;
  units::pages reserve = units::ceil<units::pages>(256_KiB);
  units::pages maximum = 16_GiB;
  unsigned shrink_shift = 2;
};

export struct heap_options {
  generation_options old{};
  generation_options young{};
  std::size_t workers = 1;
  std::size_t minor_collections = 7;
};

export class heap {
  template<class T> friend class ptr;
  template<class T> friend class weak_ptr;
  template<class T> friend class weak;
 
public:
  [[nodiscard]] native_pure native_inline
  static heap * current() noexcept { return detail::active_heap; }

  using word = std::uint64_t;
  using offset = std::uint32_t;
  using bitmap = offset;
  static constexpr offset null = 0;
  static constexpr std::size_t block_words = 32;
  static constexpr unsigned fields_per_word = 2;
  using block = detail::heap_block;
  template<class T> using ptr = jam::ptr<T>;
  template<class T> using weak_ptr = jam::weak_ptr<T>;
  static_assert(sizeof(std::size_t) == 8, "jam requires 64-bit addressing");
  static_assert(sizeof(block) == 16);
  static_assert(std::atomic_ref<bitmap>::is_always_lock_free);
  static_assert(std::atomic_ref<word>::is_always_lock_free);
 
  class visitor;
 
private:
  using trace_function = void (*)(visitor &, offset);
  struct trace_job { offset at; trace_function trace; };
  struct weak_entry {
    trace_job key, value, finalizer;
    void (*invoke)(heap &, weak_entry &) noexcept;
    bool active = true;
    weak_entry * previous_running = nullptr;
  };
  template<class F> struct finalizer_entry : weak_entry {
    void (*runner)(F *) noexcept;
  };
 
public:
  class root_handle {
    friend class heap;
    std::uint32_t slot = 0;
    native_inline void retain() noexcept {
      if (slot) heap::current()->retain_root(slot);
    }
    native_inline void detach() noexcept {
      if (slot) heap::current()->release_root(std::exchange(slot, 0));
    }
    native_inline root_handle(heap & arena, offset at, trace_function fn = nullptr) noexcept
        : slot(arena.add_root(at, fn)) {}
  public:
    root_handle() noexcept = default;
    native_inline root_handle(root_handle const & other) noexcept : slot(other.slot) { retain(); }
    native_inline root_handle(root_handle && other) noexcept : slot(std::exchange(other.slot, 0)) {}
    native_inline root_handle & operator=(root_handle const & other) noexcept {
      if (this != &other) { detach(); slot = other.slot; retain(); }
      return *this;
    }

    native_inline root_handle & operator=(root_handle && other) noexcept {
      if (this != &other) { detach(); slot = std::exchange(other.slot, 0); }
      return *this;
    }

    native_inline ~root_handle() noexcept { detach(); }
    [[nodiscard]] native_inline offset get() const noexcept {
      return slot ? heap::current()->root_slots[slot - 1].position : null;
    }

    [[nodiscard]] native_inline offset operator*() const noexcept { return get(); }
    [[nodiscard]] native_inline explicit operator bool() const noexcept { return get() != null; }
  };

  class visitor {
    friend class heap;
    template<class T> friend struct jam::tracer;
    heap & arena;
    detail::frontier<trace_job> & pending;
    std::size_t worker;
    void const * record = nullptr;
    offset record_offset = 0;
    std::size_t record_bytes = 0;
    visitor(heap & h, detail::frontier<trace_job> & tasks, std::size_t id) noexcept
        : arena(h), pending(tasks), worker(id) {}
 
    native_inline std::size_t pointer_slot(void const * field) const noexcept {
      auto const address = reinterpret_cast<std::uintptr_t>(field);
      auto const begin = reinterpret_cast<std::uintptr_t>(record);
      auto const delta = address - begin;
      if (!record || address < begin || record_bytes < sizeof(offset)
          || delta > record_bytes - sizeof(offset) || delta % sizeof(offset))
        detail::heap_failure("typed visitor requires a field in its tracing value");
      return delta / sizeof(offset);
    }
    template<class T>
    native_inline void enqueue(ptr<T> const & value) noexcept {
      static_assert(traceable<T>, "reference target must support tracing");
      if (value && (!arena.minor_collection || value.is_young())) pending.push(worker, {value.get(), &heap::template trace_record<T>});
    }
 
    // One mask window bounds stack use even for large, sparsely described records.
    template<class T>
    struct manifest_visit {
      using heap_type = heap;
      visitor & visit;
      ptr<T> const * next = nullptr;
      word bits = 0;
      std::size_t first_slot = 0;
 
      native_inline void flush() noexcept {
        visit.pointers(bits, first_slot);
        bits = 0;
      }
      template<class U>
      native_inline void operator()(ptr<U> const & value) noexcept {
        auto const slot = visit.pointer_slot(&value);
        auto const window = slot & ~std::size_t{63};
        if (window != first_slot) { flush(); first_slot = window; }
        bits |= word{1} << (slot - first_slot);
        if constexpr (std::same_as<T, U>) {
          if (value) {
            if (next) visit.enqueue(*next);
            next = std::addressof(value);
          }
        } else visit.enqueue(value);
      }
      template<class U>
      native_inline void operator()(weak_ptr<U> const & value) noexcept { visit(value); }
      template<class U>
      native_inline void operator()(U const & value) noexcept
          requires (!detail::is_ptr<U> && !detail::is_weak_ptr<U>) && detail::visits<U, visitor> {
        if constexpr (std::derived_from<tracer<U>, detail::structural_tracer>)
          tracer<U>::trace(*this, value);
        else visit(value); // Custom hooks keep the actual heap visitor.
      }
      template<class... Us>
      native_inline void operator()(Us const &... values) noexcept
          requires (sizeof...(Us) != 1) && (detail::visits<Us, visitor> && ...) {
        ((*this)(values), ...);
      }
    };
 
    template<class T>
    void trace_manifest(ptr<T> const & at) noexcept {
      // Pointer fields stay put throughout marking; borrow the next edge.
      auto const * current = std::addressof(at);
      while (current) {
        auto const * value = claim_target(*current);
        if (!value) break;
        manifest_visit<T> fields{*this};
        tracer<T>::trace(fields, *value);
        fields.flush();
        current = fields.next;
        poll();
      }
    }
 
  public:
    using heap_type = heap;
    visitor(visitor const &) = delete;
    visitor & operator=(visitor const &) = delete;
    [[nodiscard]] native_inline bool claim(offset start, std::size_t words = 1,
                               std::size_t alignment_bytes = 8) noexcept {
      return arena.claim(start, words, alignment_bytes);
    }

    template<class T>
    [[nodiscard]] native_inline T const * claim_target(ptr<T> const & value) noexcept {
      heap::require_record<T>();
      if (!value || !claim(value.get(), record_words<T>, std::max(alignof(T), sizeof(word))))
        return nullptr;
      auto const * result = arena.template record_address<T>(value.get());
      record = result;
      record_offset = value.get();
      record_bytes = sizeof(T);
      return result;
    }

    template<class T>
    [[nodiscard]] native_inline T const * claim_target(T const * value) noexcept {
      heap::require_record<T>();
      if (!value) return nullptr;
      auto const address = reinterpret_cast<std::uintptr_t>(value);
      auto const old_delta = address - reinterpret_cast<std::uintptr_t>(arena.arena.data());
      auto const young_delta = address - reinterpret_cast<std::uintptr_t>(arena.nursery.data());
      bool const young = young_delta < arena.nursery.used() * sizeof(word);
      auto const delta = young ? young_delta : old_delta;
      if ((!young && old_delta >= arena.arena.used() * sizeof(word)) || delta % sizeof(word))
        detail::heap_failure("claim_target requires a complete object in this heap");
      auto const at = static_cast<offset>(delta / sizeof(word)) | (young ? young_bit : 0);
      if (!claim(at, record_words<T>, std::max(alignof(T), sizeof(word)))) return nullptr;
      record = value;
      record_offset = at;
      record_bytes = sizeof(T);
      return value;
    }

    template<class T>
    native_inline void pointer(ptr<T> const & value) noexcept {
      auto const heap_delta = std::size_t(record_offset & offset_mask) * sizeof(word) + pointer_slot(&value) * sizeof(offset);
      arena.pointer((record_offset & young_bit) | static_cast<offset>(heap_delta / sizeof(word)),
                    static_cast<unsigned>((heap_delta % sizeof(word)) / sizeof(offset)));
    }

    template<class T>
    native_inline void pointer(weak_ptr<T> const & value) noexcept {
      pointers(word{1}, pointer_slot(&value));
    }

    native_inline void pointers(word bits, std::size_t first_slot = 0) noexcept {
      if (!bits) return;
      auto const slots = record_bytes / sizeof(offset);
      if (!record || first_slot > slots
          || static_cast<std::size_t>(std::bit_width(bits)) > slots - first_slot)
        detail::heap_failure("pointer pattern exceeds its tracing value");
      auto & g = arena.containing(record_offset);
      auto const slot = std::size_t(record_offset & offset_mask) * fields_per_word + first_slot;
      constexpr auto width = block_words * fields_per_word;
      auto const shift = slot % width;
      std::atomic_ref<word>{g.metadata[slot / width].pointers}
          .fetch_or(bits << shift, std::memory_order_relaxed);
      if (shift && (bits >> (width - shift)))
        std::atomic_ref<word>{g.metadata[slot / width + 1].pointers}
            .fetch_or(bits >> (width - shift), std::memory_order_relaxed);
    }

    template<class T>
    [[nodiscard]] native_inline T const * claim(ptr<T> const & value) noexcept {
      pointer(value);
      return claim_target(value);
    }

    native_inline offset field(offset cell, unsigned slot = 0) noexcept {
      arena.pointer(cell, slot);
      auto const target = arena.field(cell, slot);
      if (target != null && (!arena.minor_collection || (target & young_bit)))
        pending.push(worker, {target, nullptr});
      return target;
    }

    native_inline void target(offset at) noexcept {
      if (at && (!arena.minor_collection || (at & young_bit))) pending.push(worker, {at, nullptr});
    }

    template<class T>
    native_inline void operator()(T const & value) noexcept
        requires (!detail::is_ptr<T> && !detail::is_weak_ptr<T>) && detail::visits<T, visitor> {
      tracer<T>::trace(*this, value);
    }

    template<class T>
    native_inline ptr<T> operator()(ptr<T> const & value) noexcept {
      // Defer target validation until use: deducing a recursive member hook's
      // return type must not require that hook's own traceable constraint.
      pointer(value);
      enqueue(value);
      return value;
    }

    template<class T>
    native_inline void operator()(weak_ptr<T> const & value) noexcept { pointer(value); }
    template<class... Ts>
    native_inline void operator()(Ts const &... values) noexcept
        requires (sizeof...(Ts) != 1) && (detail::visits<Ts, visitor> && ...) {
      detail::trace_parts(*this, values...);
    }

    native_inline void poll() noexcept { pending.communicate(worker); }
  };
 
private:
  template<class T>
  static constexpr std::size_t record_words = (sizeof(T) + sizeof(word) - 1) / sizeof(word);
 
  template<class T>
  static consteval void require_record() noexcept {
    // Byte relocation is a managed-storage precondition, not something a trace
    // hook or a standard type trait proves. Compaction has no per-record type
    // descriptor and never invokes a move constructor or destructor.
    static_assert(std::is_object_v<T> && !std::is_const_v<T> && !std::is_volatile_v<T>,
                  "jam records must be unqualified objects that permit byte relocation");
    static_assert(alignof(T) <= 64, "record alignment exceeds 64 bytes");
  }
 
  // Marking never changes mappings; no payload copy or stack snapshot is needed.
  template<class T>
  native_inline T * record_address(offset at) noexcept {
    require_record<T>();
    assert(at && (at & offset_mask) < containing(at).cursor
           && record_words<T> <= containing(at).cursor - (at & offset_mask));
    auto * p = decode(at);
    if constexpr (std::is_trivially_copyable_v<T>) {
      return static_cast<T *>(std::memmove(p, p, sizeof(T)));
    } else {
      // Nontrivial records use Jam's byte-relocation contract. This cast does
      // not claim to establish arbitrary nontrivial lifetimes in ISO C++.
      return reinterpret_cast<T *>(p);
    }
  }
 
  template<class T>
  static void trace_record(visitor & visit, offset at) noexcept {
    require_record<T>();
    static_assert(traceable<T>, "tracing hooks must accept the heap visitor");
    if constexpr (detail::allocation_trace<T>) {
      // Dispatch before claiming: only the dynamic override knows the extent.
      visit.arena.template record_address<T>(at)->claim_and_trace(visit);
    } else if constexpr (requires { tracer<T>::trace(visit, ptr<T>{at}); }) {
      // Cooperative hooks receive an unclaimed entry and own the walk.
      tracer<T>::trace(visit, ptr<T>{at});
    } else if (auto const * value = visit.claim_target(ptr<T>{at})) {
      tracer<T>::trace(visit, *value);
    }
  }
 
public:
  struct generation {
    friend class heap;
  private:
    static constexpr offset young_bit = 0x80000000u;
    static constexpr offset offset_mask = 0x7fffffffu;
    struct settings { std::size_t page_words, reserved, minimum, bytes; generation_options policy; };
    generation_options const policy;
    detail::compactor const & compactor;
    detail::work_pool & workers;
    std::size_t const page_word_count;
    std::size_t const reserve_word_count;
    std::size_t const minimum_capacity;
    detail::heap_mapping storage;
    word * view_base;
    std::vector<block> metadata;
    // One alignment nibble per 256 bytes.
    // Low3 bits are a thermometer exponent (0,1,3,7); bit3 joins the next block.
    // Atomic OR therefore implements max(k), even for simultaneous markers.
    std::vector<std::uint8_t> alignment;
    bool has_alignment = false;
    std::size_t cursor = 1;
    std::size_t origin = 0;
    bool collecting = false;
    void require_mutator() const noexcept {
      if (collecting) detail::heap_failure("mutation during collection");
    }
    static constexpr unsigned block_shift = 5;
    static constexpr std::size_t maximum_bytes = std::size_t{1} << 34;
 
    static constexpr bool valid_options(generation_options const & config) noexcept {
      return config.capacity.count() != 0 && config.capacity <= config.maximum && config.maximum <= units::bytes{maximum_bytes}
          && config.reserve.count() != 0 && config.reserve <= units::bytes{maximum_bytes / 2}
          && config.shrink_shift < std::numeric_limits<std::size_t>::digits;
    }
    static settings configure(generation_options config) {
      if (config.maximum > units::bytes{maximum_bytes} || config.capacity > units::bytes{maximum_bytes} || config.reserve > units::bytes{maximum_bytes})
        throw std::length_error("jam generation exceeds 16 GiB");
      constexpr auto page = static_cast<std::size_t>(units::pages::ratio::num);
      if (detail::heap_mapping::page_size() != page)
        detail::heap_failure("platform page size differs from build invariant");
      if (!valid_options(config) || page < 512 || page > maximum_bytes || !std::has_single_bit(page))
        detail::heap_failure("invalid byte sizes, page size or worker count");
      if (config.reserve > config.capacity / 2)
        detail::heap_failure("capacity must hold twice the reserve");
      auto const capacity = units::bytes{config.capacity}.count();
      auto const reserve = units::bytes{config.reserve}.count();
      return {page / 8, reserve / 8, 2 * reserve / 8, capacity, config};
    }
    explicit generation(settings s, detail::compactor const & c, detail::work_pool & w, std::byte * address) noexcept
        : policy(s.policy), compactor(c), workers(w),
          page_word_count(s.page_words), reserve_word_count(s.reserved),
          minimum_capacity(s.minimum), storage(s.bytes, address, units::bytes{s.policy.maximum}.count()),
          view_base(reinterpret_cast<word *>(storage.data())), metadata(s.bytes / (8 * block_words)) {
      alignment.resize(((metadata.size() + 7) / 8) * 4);
    }
    static constexpr unsigned exponent(std::size_t bytes) noexcept {
      if (!std::has_single_bit(bytes) || bytes < 8 || bytes > 64u)
        detail::heap_failure("unsupported record alignment");
      return std::countr_zero(bytes) - 3;
    }
    unsigned flags(std::size_t index) const noexcept {
      return (alignment[index >> 1] >> ((index & 1) * 4)) & 15;
    }
    native_inline void add_flags(std::size_t index, unsigned bits) noexcept {
      std::atomic_ref<std::uint8_t>{alignment[index >> 1]}.fetch_or(
          static_cast<std::uint8_t>(bits << ((index & 1) * 4)), std::memory_order_relaxed);
    }
    native_inline void note_record(offset first, std::size_t words, std::size_t bytes) noexcept {
      assert(std::has_single_bit(bytes) && bytes >= 8 && bytes <= 64);
      assert((static_cast<std::size_t>(first) * 8) % bytes == 0);
      if (!words) return;
      auto const k = std::countr_zero(bytes) - 3;
      if (k) add_flags(first >> block_shift, (1u << k) - 1);
      auto const last = (static_cast<std::size_t>(first) + words - 1) >> block_shift;
      for (auto i = static_cast<std::size_t>(first) >> block_shift; i < last; ++i) add_flags(i, 8);
    }
    native_inline void mark_range(offset first, std::size_t words) noexcept {
      while (words != 0) {
        auto const bit = static_cast<std::size_t>(first & (block_words - 1));
        auto const count = std::min(words, block_words - bit);
        std::atomic_ref<bitmap>{metadata[first >> block_shift].live}.fetch_or(
            static_cast<bitmap>((~bitmap{0} >> (block_words - count)) << bit), std::memory_order_relaxed);
        first += static_cast<offset>(count);
        words -= count;
      }
    }
    static constexpr word pointer_cells(bitmap live) noexcept {
      word x = live;
      x = (x | (x << 16)) & 0x0000ffff0000ffffULL;
      x = (x | (x << 8)) & 0x00ff00ff00ff00ffULL;
      x = (x | (x << 4)) & 0x0f0f0f0f0f0f0f0fULL;
      x = (x | (x << 2)) & 0x3333333333333333ULL;
      x = (x | (x << 1)) & 0x5555555555555555ULL;
      return x | (x << 1);
    }
    std::uint8_t const * alignment_table() const noexcept {
      return has_alignment ? alignment.data() : nullptr;
    }
    [[nodiscard]] offset base(block const & item) const noexcept {
      return item.destination;
    }
    [[nodiscard]] offset forward(offset value) const noexcept {
      if (value == null) return null;
      assert(value < cursor);
      auto const & item = metadata[value >> block_shift];
      auto const bit = static_cast<bitmap>(bitmap{1} << (value & (block_words - 1)));
      if (!(item.live & bit)) return null;
      return base(item) + static_cast<offset>(std::popcount(detail::retained(item, alignment_table(), value >> block_shift) & (bit - 1)));
    }
    void resize(std::size_t words) noexcept {
      assert(words % page_word_count == 0 && words >= minimum_capacity && words <= maximum_bytes / 8);
      assert(cursor <= words - reserve_word_count);
      std::vector<block> next_metadata(words / block_words);
      std::copy_n(metadata.begin(), (cursor + block_words - 1) / block_words, next_metadata.begin());
      auto const next_origin = words - reserve_word_count;
      auto next = storage.resized(origin * 8, cursor * 8, words * 8, next_origin * 8);
      alignment.resize(((next_metadata.size() + 7) / 8) * 4);
      storage.replace(next); metadata.swap(next_metadata);
      origin = next_origin;
      view_base = reinterpret_cast<word *>(storage.data()) + origin;
    }
 
    generation(generation_options config, detail::compactor const & c, detail::work_pool & w, std::byte * address)
        : generation(configure(config), c, w, address) {}
 
  public:
    [[nodiscard]] std::size_t capacity() const noexcept { return storage.bytes() / 8; }
    [[nodiscard]] std::size_t reserved() const noexcept { return reserve_word_count; }
    [[nodiscard]] std::size_t used() const noexcept { return cursor; }
    [[nodiscard]] std::size_t start() const noexcept { return origin; }
    [[nodiscard]] std::size_t page_words() const noexcept { return page_word_count; }
    template<class Self>
    [[nodiscard]] native_inline native_pure
    auto data(this Self & self native_lifetimebound) noexcept
        -> std::conditional_t<std::is_const_v<Self>, word const, word> * {
      using cell = std::conditional_t<std::is_const_v<Self>, word const, word>;
      return static_cast<cell *>(self.view_base);
    }

    template<class Self>
    [[nodiscard]] decltype(auto) operator[](this Self & self native_lifetimebound, offset index) noexcept {
      assert(index < self.cursor); return self.data()[index];
    }

    [[nodiscard]] offset field(offset cell, unsigned slot = 0) const noexcept {
      assert(cell < cursor && slot < fields_per_word);
      offset value;
      auto const address = reinterpret_cast<std::byte const *>(data() + cell) + slot * sizeof(offset);
      std::memcpy(&value, address, sizeof(value));
      return value;
    }

    void set_field(offset cell, offset value, unsigned slot = 0) noexcept {
      require_mutator();
      assert(cell < cursor && slot < fields_per_word);
      auto const address = reinterpret_cast<std::byte *>(data() + cell) + slot * sizeof(offset);
      std::memcpy(address, &value, sizeof(value));
    }

    [[nodiscard]] std::span<block const> blocks() const noexcept { return metadata; }
    [[nodiscard]] offset allocate(std::size_t words, std::size_t alignment_bytes = 8) noexcept
        native_diagnose_if(words == 0, "jam records must contain at least one cell")
        native_diagnose_if(!std::has_single_bit(alignment_bytes) || alignment_bytes < 8 || alignment_bytes > 64u, "unsupported jam record alignment") {
      require_mutator();
      auto const k = exponent(alignment_bytes);
      auto const aligned = (cursor + (std::size_t{1} << k) - 1) & ~((std::size_t{1} << k) - 1);
      if (!words) detail::heap_failure("zero-word allocation");
      auto const limit = units::bytes{policy.maximum}.count() / 8 - reserve_word_count;
      if (aligned > limit || words > limit - aligned)
        detail::heap_failure("capacity overflow");
      while (words > capacity() - reserve_word_count - std::min(aligned, capacity() - reserve_word_count))
        resize(std::min(capacity() * 2, units::bytes{policy.maximum}.count() / 8));
      std::fill(data() + cursor, data() + aligned, word{0});
      cursor = aligned + words;
      return static_cast<offset>(aligned);
    }

    void clear_marks() noexcept {
      for (auto & item : metadata) {
        item.live = 0;
        item.destination = 0;
      }
      std::fill(alignment.begin(), alignment.end(), 0);
      has_alignment = false;
    }

    void mark(offset first, std::size_t words = 1, std::size_t alignment_bytes = 8) noexcept {
      assert(first != null && first <= cursor && words <= cursor - first);
      note_record(first, words, alignment_bytes); mark_range(first, words);
    }

    [[nodiscard]] native_inline bool claim(offset first, std::size_t words = 1,
                             std::size_t alignment_bytes = 8) noexcept {
      assert(first != null && words && first < cursor && words <= cursor - first);
      auto const bit = static_cast<bitmap>(bitmap{1} << (first & (block_words - 1)));
      auto const previous = std::atomic_ref<bitmap>{metadata[first >> block_shift].live}
          .fetch_or(bit, std::memory_order_relaxed);
      if ((previous & bit) != 0) return false;
      note_record(first, words, alignment_bytes);
      mark_range(first + 1, words - 1);
      return true;
    }

    native_inline void pointer(offset where, unsigned slot = 0) noexcept {
      assert(where != null && where < cursor && slot < fields_per_word);
      auto const bit = (where & (block_words - 1)) * fields_per_word + slot;
      std::atomic_ref<word>{metadata[where >> block_shift].pointers}
          .fetch_or(word{1} << bit, std::memory_order_relaxed);
    }

    void compact(std::span<offset> roots = {}) noexcept {
      require_mutator();
      std::atomic_signal_fence(std::memory_order_seq_cst);
      collecting = true;
      compact_impl(roots);
      collecting = false;
      std::atomic_signal_fence(std::memory_order_seq_cst);
    }
 
  private:
    std::size_t prepare(std::size_t start, bool keep_null) noexcept {
      // Keep the null cell in the packed prefix, including aligned collections.
      // It is padding, never a record or a source of outgoing pointers.
      data()[0] = 0;
      if (keep_null) metadata[0].live |= bitmap{1};
      else metadata[0].live &= ~bitmap{1};
      metadata[0].pointers &= ~word{(1u << fields_per_word) - 1};
      auto const old_blocks = (cursor + block_words - 1) / block_words;
      // A parallel prefix scan could compose joined groups' layout functions:
      // F(x) = align_up(x, 1 << k) + retained_cells_after_dilation. Resolve group
      // boundaries/k first; composing (max k, summed counts) loses padding.
      // Eight input-residue displacements suffice (k <= 3), with associative
      // composition c[r] = a[r] + b[(r + a[r]) & 7]. All destinations must be
      // ready before workers forward arbitrary pointers into distant blocks.
      // This whole pass measured only 0.5-1.7% of GC on eight-byte-aligned trees
      // and chains with 1/4 workers, so scan overhead could erase the benefit.
      // Sparse retention and higher alignments were not covered; reprofile those
      // workloads before adding a parallel scan.
      has_alignment = false;
      std::size_t live_words = start;
      for (std::size_t begin = 0; begin < old_blocks;) {
        auto end = begin + 1;
        unsigned k = std::popcount(flags(begin) & 7u);
        while ((flags(end - 1) & 8u) && end < old_blocks) {
          k = std::max(k, static_cast<unsigned>(std::popcount(flags(end) & 7u))); ++end;
        }
        live_words = (live_words + (std::size_t{1} << k) - 1) & ~((std::size_t{1} << k) - 1);
        for (auto i = begin; i != end; ++i) {
          auto & item = metadata[i];
          item.pointers &= pointer_cells(item.live);
          // Publish the joined group's effective k in the existing nibbles.
          // live remains the exact claims; dilation is only a register value.
          auto const shift = (i & 1) * 4;
          alignment[i >> 1] = static_cast<std::uint8_t>((alignment[i >> 1] & ~(7u << shift))
              | (((1u << k) - 1) << shift));
          has_alignment |= k != 0;
          item.destination = static_cast<offset>(live_words);
          live_words += std::popcount(detail::dilate(item.live, k));
        }
        begin = end;
      }
      // Dilation can retain unallocated cells at the final record's rounded end.
      auto const padded = std::min(capacity() - reserve_word_count, (cursor + 7) & ~std::size_t{7});
      std::fill(data() + cursor, data() + padded, word{0});
 
      return live_words;
    }
    void move_to(word * target, detail::forwarding_tables tables) noexcept {
      auto const old_blocks = (cursor + block_words - 1) / block_words;
      auto const source = data();
      auto const source_pages = (cursor + page_word_count - 1) / page_word_count;
      auto const wave_pages = reserve_word_count / page_word_count;
      auto move_page = [&](std::size_t page) noexcept {
        auto const begin = page * (page_word_count / block_words);
        auto const end = std::min(old_blocks, begin + page_word_count / block_words);
        compactor.move(source, target, metadata.data(), begin, end, tables, alignment_table());
      };
      if (workers.available_workers() > 1 && source_pages > 1) {
        auto done = std::make_unique<std::atomic<bool>[]>(source_pages);
        std::atomic<std::size_t> next{0}, front{0};
        auto task = [&](std::size_t) noexcept {
          for (;;) {
            auto const page = next.fetch_add(1, std::memory_order_relaxed);
            if (page >= source_pages) return;
            auto floor = front.load(std::memory_order_acquire);
            while (page - floor >= wave_pages) {
              front.wait(floor, std::memory_order_relaxed);
              floor = front.load(std::memory_order_acquire);
            }
            move_page(page);
            done[page].store(true, std::memory_order_release);
            // Every completion publishes through this RMW chain, even when an
            // older page still blocks the prefix. The last publisher therefore
            // sees all earlier done flags; a store-buffering cycle cannot stall.
            floor = front.fetch_add(0, std::memory_order_acq_rel);
            for (;;) {
              auto end = floor;
              while (end != source_pages && done[end].load(std::memory_order_acquire)) ++end;
              if (end == floor) break;
              if (front.compare_exchange_strong(floor, end, std::memory_order_acq_rel)) {
                front.notify_all(); break;
              }
              floor = front.load(std::memory_order_acquire);
            }
          }
        };
        workers.run(std::min(workers.available_workers(), wave_pages),
          [](void * context, std::size_t i) noexcept {
            (*static_cast<decltype(task) *>(context))(i);
          }, &task);
      } else for (std::size_t page = 0; page < source_pages; ++page) move_page(page);
    }
    void zero_gaps(word * target, std::size_t previous_end) const noexcept {
      auto const old_blocks = (cursor + block_words - 1) / block_words;
      for (std::size_t i = 0; i != old_blocks; ++i) {
        auto const next = std::size_t(metadata[i].destination & offset_mask);
        std::fill(target + previous_end, target + next, word{0});
        previous_end = next + std::popcount(detail::retained(metadata[i], alignment_table(), i));
      }
    }
    void pack_into(std::vector<block> & output) const noexcept {
      auto const old_blocks = (cursor + block_words - 1) / block_words;
      compactor.pack(metadata.data(), output.data(), old_blocks, alignment_table());
    }
    void ensure(std::size_t words) noexcept {
      auto const maximum = units::bytes{policy.maximum}.count() / sizeof(word);
      if (words > maximum - reserve_word_count) detail::heap_failure("generation capacity overflow");
      while (words > capacity() - reserve_word_count) resize(std::min(2 * capacity(), maximum));
    }
    std::size_t rotated_origin() const noexcept {
      return (origin + capacity() - reserve_word_count) % capacity();
    }
    word * at_origin(std::size_t start) const noexcept {
      return reinterpret_cast<word *>(storage.data()) + start;
    }
    void adopt(std::size_t next_origin, std::size_t words) noexcept {
      auto const old_blocks = (cursor + block_words - 1) / block_words;
      cursor = words; origin = next_origin; view_base = at_origin(origin);
      auto const blocks = (cursor + block_words - 1) / block_words;
      for (std::size_t i = 0; i != blocks; ++i) {
        auto const count = std::min(block_words, cursor - i * block_words);
        metadata[i].live = ~bitmap{0} >> (block_words - count);
        metadata[i].destination = 0;
      }
      for (auto i = blocks; i < old_blocks; ++i) metadata[i] = {};
      std::fill(alignment.begin(), alignment.end(), 0);
      has_alignment = false;
      auto const smaller = std::max(minimum_capacity, (capacity() / page_word_count / 2) * page_word_count);
      if (policy.shrink_shift && smaller < capacity()
          && cursor < (capacity() >> policy.shrink_shift)
          && cursor <= smaller - reserve_word_count) resize(smaller);
    }
    void reset() noexcept {
      for (auto & item : metadata) item = {};
      std::fill(alignment.begin(), alignment.end(), 0);
      has_alignment = false;
      cursor = 1; data()[0] = 0;
    }
    void compact_impl(std::span<offset> roots) noexcept {
      auto const live_words = prepare(0, true);
      auto const next_origin = rotated_origin();
      auto * target = at_origin(next_origin);
      move_to(target, {metadata.data(), nullptr, alignment_table(), nullptr});
      for (auto & root : roots) root = forward(root);
      zero_gaps(target, 0);
      pack_into(metadata);
      adopt(next_origin, live_words);
    }
  };
private:
  detail::compactor const compactor = detail::choose_compactor();
  detail::work_pool workers;
  detail::generation_reservation reservation;
  generation arena;
  generation nursery;
  heap_options const policy;
  std::size_t minors_left;
  struct root_slot {
    offset position;
    std::uint32_t copies, previous, next;
    trace_function trace;
  };
  static_assert(sizeof(root_slot) == 24);
  std::vector<root_slot> root_slots;
  std::uint32_t roots_head = 0, free_root = 0;
  native_inline std::uint32_t add_root(offset at, trace_function trace) noexcept {
    assert(heap::current() == this);
    require_mutator();
    if (!at) return 0;
    auto slot = free_root;
    if (slot) free_root = root_slots[slot - 1].next;
    else {
      if (root_slots.size() == std::numeric_limits<std::uint32_t>::max())
        detail::heap_failure("root slot capacity overflow");
      root_slots.push_back({});
      slot = static_cast<std::uint32_t>(root_slots.size());
    }
    root_slots[slot - 1] = {at, 1, 0, roots_head, trace};
    if (roots_head) root_slots[roots_head - 1].previous = slot;
    roots_head = slot;
    return slot;
  }
  native_inline void retain_root(std::uint32_t slot) noexcept {
    require_mutator();
    auto & count = root_slots[slot - 1].copies;
    if (count == std::numeric_limits<std::uint32_t>::max())
      detail::heap_failure("root reference count overflow");
    ++count;
  }
  native_inline void release_root(std::uint32_t slot) noexcept {
    require_mutator();
    auto & entry = root_slots[slot - 1];
    if (--entry.copies) return;
    if (entry.previous) root_slots[entry.previous - 1].next = entry.next;
    else roots_head = entry.next;
    if (entry.next) root_slots[entry.next - 1].previous = entry.previous;
    entry = {0, 0, 0, free_root, nullptr};
    free_root = slot;
  }
  std::vector<std::shared_ptr<weak_entry>> weak_entries;
  std::deque<std::shared_ptr<weak_entry>> pending_finalizers;
  weak_entry * running_finalizer = nullptr;
  bool collecting = false;
  bool minor_collection = false;
  std::unordered_map<std::uint32_t, trace_function> remembered;
  // Callback identity distinguishes weak slots from raw strong slots (nullptr).
  // Weak slots participate in forwarding, never in the initial marking frontier.
  static void weak_slot(visitor &, offset) noexcept {}
  static constexpr unsigned block_shift = 5;
  static constexpr offset young_bit = 0x80000000u;
  static constexpr offset offset_mask = 0x7fffffffu;
  static heap_options configure(heap_options config) {
    static_cast<void>(generation::configure(config.old));
    static_cast<void>(generation::configure(config.young));
    if (!config.workers) detail::heap_failure("worker count must be positive");
    return config;
  }
  struct configured {};
  heap(configured, heap_options config) noexcept
      : workers(config.workers),
        reservation(units::bytes{config.old.maximum}.count(), units::bytes{config.young.maximum}.count()),
        arena(config.old, compactor, workers, reservation.data()),
        nursery(config.young, compactor, workers, reservation.data() + 2 * units::bytes{config.old.maximum}.count()),
        policy(config), minors_left(config.minor_collections) {}
  [[nodiscard]] native_inline generation & containing(offset at) noexcept { return at & young_bit ? nursery : arena; }
  [[nodiscard]] native_inline generation const & containing(offset at) const noexcept { return at & young_bit ? nursery : arena; }
  template<class Self>
  [[nodiscard]] native_inline native_pure
  auto decode(this Self & self native_lifetimebound, offset bits) noexcept
      -> std::conditional_t<std::is_const_v<Self>, std::byte const, std::byte> * {
    using byte = std::conditional_t<std::is_const_v<Self>, std::byte const, std::byte>;
    // Carry selects the generation; the wrapped sum strips its tag and doubles the cell offset.
    std::uint32_t doubled;
    auto * base = (__builtin_add_overflow(bits, bits, &doubled) ? self.nursery : self.arena).data();
    return reinterpret_cast<byte *>(base) + std::size_t(doubled) * 4;
  }
  native_inline bool old_slot(void const * address, std::uint32_t & slot) const noexcept {
    auto const delta = reinterpret_cast<std::uintptr_t>(address) - reinterpret_cast<std::uintptr_t>(arena.storage.data());
    auto const bytes = arena.storage.bytes();
    if (delta >= 2 * bytes) return false;
    auto const physical = delta >= bytes ? delta - bytes : delta;
    auto const start = arena.origin * sizeof(word);
    auto const logical = physical >= start ? physical - start : physical + bytes - start;
    assert(logical % sizeof(offset) == 0 && logical / sizeof(offset) < 2 * arena.cursor);
    slot = static_cast<std::uint32_t>(logical / sizeof(offset));
    return true;
  }
  native_inline void forget(void const * address) noexcept {
    std::uint32_t slot;
    if (old_slot(address, slot)) forget_slot(slot);
  }
  native_noinline void forget_slot(std::uint32_t slot) noexcept {
    remembered.erase(slot);
  }
  native_noinline void remember_slot(std::uint32_t slot, trace_function trace) noexcept {
    require_mutator();
    remembered.insert_or_assign(slot, trace);
  }
  void require_mutator() const noexcept {
    if (collecting) detail::heap_failure("mutation during collection");
  }
  offset forward(offset value, bool minor) const noexcept {
    if (!value || (minor && !(value & young_bit))) return value;
    return containing(value).forward(value & offset_mask);
  }
  // Read only with the marking frontier drained; workers may otherwise own these bits.
  bool marked(offset at) const noexcept {
    if (!at) return false;
    if (minor_collection && !(at & young_bit)) return true;
    auto const local = at & offset_mask;
    return (containing(at).metadata[local >> block_shift].live >> (local % block_words)) & 1;
  }
  void run_finalizer(std::shared_ptr<weak_entry> const & entry) noexcept {
    entry->previous_running = running_finalizer;
    running_finalizer = entry.get();
    {
      heap_scope binding{*this};
      entry->invoke(*this, *entry);
    }
    running_finalizer = entry->previous_running;
  }
  void drain_finalizers() noexcept {
    if (running_finalizer) return;
    while (!pending_finalizers.empty()) {
      auto entry = std::move(pending_finalizers.front());
      pending_finalizers.pop_front();
      run_finalizer(entry);
    }
  }
  void finalize(std::shared_ptr<weak_entry> const & entry) noexcept {
    require_mutator();
    if (!entry->active) return;
    entry->active = false;
    std::erase(weak_entries, entry);
    run_finalizer(entry);
    drain_finalizers();
  }
  void compact_impl(std::span<offset> roots, bool minor = false, bool promote = false) noexcept {
    auto const old_end = minor ? arena.cursor : arena.prepare(0, true);
    auto const young_start = minor && !promote ? std::size_t(young_bit) : old_end;
    auto const end = nursery.prepare(young_start, minor && !promote);
    auto & destination = minor && !promote ? nursery : arena;
    auto const live_words = end - (minor && !promote ? std::size_t(young_bit) : 0);
    destination.ensure(live_words);
    // Growth may rebase metadata, so form the shared forwarding tables afterward.
    detail::forwarding_tables tables{minor ? nullptr : arena.metadata.data(), nursery.metadata.data(),
      minor ? nullptr : arena.alignment_table(), nursery.alignment_table()};
    auto const next_origin = minor && promote ? destination.origin : destination.rotated_origin();
    auto * target = destination.at_origin(next_origin);
    if (!minor) {
      arena.move_to(target, tables);
      arena.zero_gaps(target, 0);
    }
    nursery.move_to(target, tables);
    nursery.zero_gaps(target, minor && !promote ? 0 : old_end);
    for (auto & root : roots) root = forward(root, minor);
    for (auto slot = roots_head; slot; slot = root_slots[slot - 1].next) {
      auto & entry = root_slots[slot - 1];
      entry.position = forward(entry.position, minor);
    }
    if (minor) for (auto entry = remembered.begin(); entry != remembered.end();) {
      auto const slot = entry->first;
      auto * address = reinterpret_cast<std::byte *>(arena.data()) + std::size_t(slot) * sizeof(offset);
      offset value;
      std::memcpy(&value, address, sizeof(value));
      value = forward(value, true);
      std::memcpy(address, &value, sizeof(value)); // Collector writes bypass barriers.
      if (value) ++entry; else entry = remembered.erase(entry);
    }
    auto forward_entry = [&](weak_entry & entry) noexcept {
      entry.key.at = forward(entry.key.at, minor);
      entry.value.at = entry.active ? forward(entry.value.at, minor) : null;
      entry.finalizer.at = forward(entry.finalizer.at, minor);
    };
    for (auto const & entry : weak_entries) forward_entry(*entry);
    for (auto const & entry : pending_finalizers) forward_entry(*entry);
    for (auto * entry = running_finalizer; entry; entry = entry->previous_running) forward_entry(*entry);
    // Both generations and every external/remembered root have consumed the
    // forwarding tables. Packing may now overwrite their pointer masks.
    if (!minor) arena.pack_into(arena.metadata);
    nursery.pack_into(destination.metadata);
    destination.adopt(next_origin, live_words);
    if (!minor || promote) { nursery.reset(); remembered.clear(); }
  }
 
public:
  class host {
    heap & owner;
    std::size_t const prefix;
    enum class phase { idle, marking, retry, prepared };
    phase state = phase::idle;
    std::size_t old_words = 0, young_words = 0;
    bool promoting = false;
  public:
    enum class generation { old, young };
    explicit host(heap & storage, std::size_t prefix) noexcept : owner(storage), prefix(prefix) {
      owner.require_mutator();
      for (auto * g : {&owner.arena, &owner.nursery}) {
        if (g->used() != 1 || g->policy.capacity != g->policy.maximum || g->policy.shrink_shift ||
            !prefix || prefix % g->page_words() || prefix >= g->capacity() - g->reserved())
          detail::heap_failure("host requires fresh fixed generations and a whole-page guard");
      }
      static_cast<void>(owner.arena.allocate(prefix - 1));
      static_cast<void>(owner.nursery.allocate(prefix - 1));
    }
    host(host const &) = delete;
    host & operator=(host const &) = delete;
    ~host() noexcept {
      if (state != phase::idle) detail::heap_failure("unfinished host collection");
    }

    void publish(generation which, void * target, std::size_t skip, std::size_t count) const noexcept {
      if (state != phase::idle) detail::heap_failure("publish during host collection");
      auto const & g = which == generation::young ? owner.nursery : owner.arena;
      auto const & config = which == generation::young ? owner.policy.young : owner.policy.old;
      if (config.capacity != config.maximum || config.shrink_shift ||
          skip > g.capacity() || count > g.capacity() - skip)
        detail::heap_failure("published heap requires a fixed generation");
      g.storage.publish(((g.origin + skip) % g.capacity()) * 8,
        static_cast<std::byte *>(target), count * 8);
    }

    [[nodiscard]] offset allocate(generation which, std::size_t words) noexcept {
      owner.require_mutator();
      return (which == generation::young ? owner.nursery.allocate(words) : owner.arena.allocate(words)) | (which == generation::young ? young_bit : 0);
    }

    void begin(bool minor) noexcept {
      owner.require_mutator();
      if (state != phase::idle) detail::heap_failure("host begin outside idle phase");
      if (owner.roots_head || !owner.weak_entries.empty() || !owner.pending_finalizers.empty() ||
          owner.running_finalizer || !owner.remembered.empty())
        detail::heap_failure("host collection requires exclusively VM-owned roots");
      std::atomic_signal_fence(std::memory_order_seq_cst);
      owner.collecting = true;
      owner.minor_collection = minor;
      state = phase::marking;
      owner.nursery.clear_marks();
      for (auto & item : owner.nursery.metadata) item.pointers = 0;
      if (!minor) {
        owner.arena.clear_marks();
        for (auto & item : owner.arena.metadata) item.pointers = 0;
      }
    }

    template<class Trace>
    void trace(std::span<offset const> roots, Trace && trace,
               std::size_t marker_limit = 1, bool old_owners = false) noexcept
        requires std::is_nothrow_invocable_v<Trace &, visitor &, offset> {
      if (state != phase::marking || !marker_limit ||
          (old_owners && (!owner.minor_collection || marker_limit != 1)))
        detail::heap_failure("host trace outside marking phase");
      std::vector<trace_job> initial;
      initial.reserve(roots.size());
      for (auto at : roots)
        if (at && (!owner.minor_collection || old_owners || (at & young_bit))) initial.push_back({at, nullptr});
      owner.drain_jobs(trace, initial, marker_limit);
    }

    [[nodiscard]] bool marked(offset at) const noexcept {
      if (state != phase::marking && state != phase::retry)
        detail::heap_failure("host liveness outside marking phase");
      return owner.marked(at);
    }

    [[nodiscard]] bool prepare(bool promote = false) noexcept {
      if (state != phase::marking && state != phase::retry)
        detail::heap_failure("host prepare outside marking phase");
      promoting = owner.minor_collection && promote;
      if (!owner.minor_collection) {
        if (prefix > 1) owner.arena.mark(1, prefix - 1);
        old_words = owner.arena.prepare(0, true);
      } else old_words = owner.arena.cursor;
      if (promoting) {
        old_words = owner.nursery.prepare(owner.arena.cursor, false);
        if (old_words > owner.arena.capacity() - owner.arena.reserved()) {
          state = phase::retry;
          return false;
        }
        young_words = prefix;
      } else {
        if (prefix > 1) owner.nursery.mark(1, prefix - 1);
        young_words = owner.nursery.prepare(young_bit, true) - young_bit;
      }
      state = phase::prepared;
      return true;
    }

    [[nodiscard]] offset forward(offset at) const noexcept {
      if (state != phase::prepared)
        detail::heap_failure("host forward outside prepared phase");
      return owner.forward(at, owner.minor_collection);
    }

    void finish() noexcept {
      if (state != phase::prepared)
        detail::heap_failure("host finish outside prepared phase");
      detail::forwarding_tables tables{owner.minor_collection ? nullptr : owner.arena.metadata.data(), owner.nursery.metadata.data(),
        owner.minor_collection ? nullptr : owner.arena.alignment_table(), owner.nursery.alignment_table()};
      auto const old_origin = owner.minor_collection ? owner.arena.origin : owner.arena.rotated_origin();
      auto const young_origin = owner.nursery.rotated_origin();
      if (!owner.minor_collection) {
        auto * target = owner.arena.at_origin(old_origin);
        owner.arena.move_to(target, tables);
        owner.arena.zero_gaps(target, 0);
      }
      auto * target = promoting ? owner.arena.data() : owner.nursery.at_origin(young_origin);
      owner.nursery.move_to(target, tables);
      owner.nursery.zero_gaps(target, promoting ? owner.arena.cursor : 0);
      // Both moves consumed the original tables. Packing may overwrite them now.
      if (!owner.minor_collection) owner.arena.pack_into(owner.arena.metadata);
      owner.nursery.pack_into(promoting ? owner.arena.metadata : owner.nursery.metadata);
      if (!owner.minor_collection || promoting) owner.arena.adopt(old_origin, old_words);
      if (promoting) owner.nursery.reset();
      else owner.nursery.adopt(young_origin, young_words);
      owner.collecting = false;
      owner.minor_collection = false;
      state = phase::idle;
      if (promoting && young_words > 1)
        static_cast<void>(owner.nursery.allocate(young_words - 1));
      std::atomic_signal_fence(std::memory_order_seq_cst);
    }
  };

  [[nodiscard]] char const * compactor_name() const noexcept { return compactor.name; }
  [[nodiscard]] std::size_t remembered_size() const noexcept { return remembered.size(); }
  template<class T> native_inline void remember(ptr<T> const & value) noexcept {
    if (!value.is_young()) return;
    std::uint32_t slot;
    if (!old_slot(&value, slot)) return;
    remember_slot(slot, &heap::template trace_record<T>);
  }

  template<class T> native_inline void remember(std::span<ptr<T>> values) noexcept {
    if (values.empty()) return;
    std::uint32_t first;
    if (!old_slot(values.data(), first)) return;
    require_mutator();
    assert(std::size_t(first) + values.size() <= 2 * arena.cursor);
    for (std::size_t i = 0; i != values.size(); ++i)
      if (values[i].is_young()) remember_slot(first + static_cast<std::uint32_t>(i), &heap::template trace_record<T>);
  }

  template<class T> native_inline void remember(weak_ptr<T> const & value) noexcept {
    std::uint32_t slot;
    if (!old_slot(&value, slot)) return;
    if (value.is_young()) remember_slot(slot, &weak_slot); else forget_slot(slot);
  }

  template<class T> native_inline void remember(std::span<weak_ptr<T>> values) noexcept {
    if (values.empty()) return;
    std::uint32_t first;
    if (!old_slot(values.data(), first)) return;
    require_mutator();
    assert(std::size_t(first) + values.size() <= 2 * arena.cursor);
    for (std::size_t i = 0; i != values.size(); ++i) {
      auto const slot = first + static_cast<std::uint32_t>(i);
      if (values[i].is_young()) remember_slot(slot, &weak_slot); else forget_slot(slot);
    }
  }

  explicit heap(heap_options config = {})
      : heap(configured{}, configure(config)) {}

  [[nodiscard]] generation const & old() const noexcept { return arena; }
  [[nodiscard]] generation const & young() const noexcept { return nursery; }
  [[nodiscard]] heap_options configuration() const noexcept { return policy; }
  heap(heap const &) = delete;
  heap & operator=(heap const &) = delete;
  ~heap() noexcept {
    if (roots_head || running_finalizer) detail::heap_failure("heap destroyed with attached roots or callbacks");
    for (auto const & entry : weak_entries) entry->active = false;
  }

  [[nodiscard]] native_inline root_handle root(offset at) noexcept {
    assert(at == null || (at & offset_mask) < containing(at).cursor);
    return root_handle{*this, at};
  }

  template<class T>
  [[nodiscard]] native_inline jam::root<T> root(ptr<T> const & at) noexcept requires traceable<T> {
    require_record<T>();
    assert(!at || at.local_offset() < containing(at.get()).cursor);
    return jam::root<T>{root_handle{*this, at.get(), &heap::template trace_record<T>}};
  }

  template<class T>
  [[nodiscard]] native_inline jam::weak_root<T> weak_root(ptr<T> const & at) noexcept requires traceable<T> {
    require_record<T>();
    assert(!at || at.local_offset() < containing(at.get()).cursor);
    return jam::weak_root<T>{root_handle{*this, at.get(), &weak_slot}};
  }

  template<traceable K, traceable V, traceable F>
  [[nodiscard]] jam::weak<V> mk_weak(ptr<K> const & key, ptr<V> const & value,
      ptr<F> const & finalizer, std::type_identity_t<void (*)(F *) noexcept> runner) noexcept;
  void collect() noexcept {
    collect([](visitor &, offset) noexcept { detail::heap_failure("untyped target in typed collection"); });
  }

  void collect_major() noexcept {
    collect_major([](visitor &, offset) noexcept { detail::heap_failure("untyped target in typed collection"); });
  }

  void collect_minor(bool promote = false) noexcept {
    collect_minor([](visitor &, offset) noexcept { detail::heap_failure("untyped target in typed collection"); }, promote);
  }

  template<class Trace>
  void collect(Trace && trace) noexcept
      requires std::is_nothrow_invocable_v<Trace &, visitor &, offset> {
    require_mutator();
    if (minors_left) {
      --minors_left;
      collect_minor(std::forward<Trace>(trace));
    } else collect_major(std::forward<Trace>(trace));
  }

  template<class Trace>
  void collect_major(Trace && trace) noexcept
      requires std::is_nothrow_invocable_v<Trace &, visitor &, offset> {
    run_collection(std::forward<Trace>(trace), false, false);
  }

  template<class Trace>
  void collect_minor(Trace && trace, bool promote = false) noexcept
      requires std::is_nothrow_invocable_v<Trace &, visitor &, offset> {
    run_collection(std::forward<Trace>(trace), true, promote);
  }
private:
  template<class Trace>
  void drain_jobs(Trace & trace, std::vector<trace_job> & initial, std::size_t marker_limit) noexcept {
    if (initial.empty()) return;
    auto const marker_count = std::min(marker_limit, workers.available_workers());
    detail::frontier<trace_job> pending{marker_count, initial};
    struct context {
      heap * arena;
      std::remove_reference_t<Trace> * trace;
      detail::frontier<trace_job> * pending;
    } state{this, std::addressof(trace), &pending};
    workers.run(marker_count, [](void * raw, std::size_t worker) noexcept {
      auto & state = *static_cast<context *>(raw);
      heap_scope binding{*state.arena};
      visitor visit{*state.arena, *state.pending, worker};
      trace_job job;
      while (state.pending->pop(worker, job)) {
        if (job.trace) job.trace(visit, job.at);
        else std::invoke(*state.trace, visit, job.at);
      }
    }, &state);
    initial.clear();
  }
  template<class Trace>
  void run_collection(Trace && trace, bool minor, bool promote) noexcept {
    require_mutator();
    // Compiler-only safepoint: publish mutator stores before tracing. Worker
    // synchronization is supplied by the pool; this is not a hardware fence.
    std::atomic_signal_fence(std::memory_order_seq_cst);
    collecting = true;
    minor_collection = minor;
    if (minor) nursery.clear_marks(); else clear_marks();
    // Each collection redeclares fields: conditional tracers may change which
    // slots are references. Raw clear_marks()/compact() retain their old contract.
    for (auto & item : nursery.metadata) item.pointers = 0;
    if (!minor) for (auto & item : arena.metadata) item.pointers = 0;
    std::vector<trace_job> initial;
    for (auto slot = roots_head; slot; slot = root_slots[slot - 1].next) {
      auto const & entry = root_slots[slot - 1];
      if (entry.trace != &weak_slot && entry.position != null && (!minor || (entry.position & young_bit)))
        initial.push_back({entry.position, entry.trace});
    }
    if (minor) for (auto const & [slot, callback] : remembered) {
      offset value;
      std::memcpy(&value, reinterpret_cast<std::byte const *>(arena.data()) + std::size_t(slot) * sizeof(offset), sizeof(value));
      if ((value & young_bit) && callback != &weak_slot) initial.push_back({value, callback});
    }
    auto seed = [&](trace_job job) noexcept {
      if (job.at && !marked(job.at)) initial.push_back(job);
    };
    auto seed_finalizer = [&](weak_entry const & entry) noexcept {
      seed(entry.key); seed(entry.finalizer);
    };
    for (auto const & entry : pending_finalizers) seed_finalizer(*entry);
    for (auto * entry = running_finalizer; entry; entry = entry->previous_running) seed_finalizer(*entry);
    auto drain = [&]() noexcept {
      drain_jobs(trace, initial, workers.available_workers());
    };
    drain();
    std::stable_sort(weak_entries.begin(), weak_entries.end(), [](auto const & a, auto const & b) noexcept {
      return a->key.at < b->key.at;
    });
    std::size_t retained = 0;
    for (auto const & entry : weak_entries) {
      if (marked(entry->key.at)) {
        seed(entry->value); seed(entry->finalizer);
        weak_entries[retained++] = entry;
      } else {
        // Retire before retaining K/F: this entry never rearms, even on resurrection.
        entry->active = false;
        pending_finalizers.push_back(entry);
        seed_finalizer(*entry);
      }
      drain();
    }
    weak_entries.resize(retained);
    compact_impl({}, minor, promote);
    collecting = false; minor_collection = false;
    if (!minor) minors_left = policy.minor_collections;
    std::atomic_signal_fence(std::memory_order_seq_cst);
    drain_finalizers();
  }
public:
  template<class T, class... Args>
  [[nodiscard]] ptr<T> mk(Args &&... args) noexcept requires traceable<T> {
    require_record<T>();
    static_assert(noexcept(T{std::forward<Args>(args)...}), "record construction must be noexcept");
    require_mutator();
    // Arguments can borrow the current nursery view. Construct before allocation
    // can remap it; transport the result under the managed byte-relocation contract.
    T const value{std::forward<Args>(args)...};
    auto const at = nursery.allocate(record_words<T>, std::max(alignof(T), sizeof(word)));
    std::fill_n(nursery.data() + at, record_words<T>, word{0});
    // Transport the complete representation, including a polymorphic vptr.
    std::memcpy(nursery.data() + at, static_cast<void const *>(&value), sizeof(T));
    return ptr<T>{young_bit | at};
  }

  template<class T>
  [[nodiscard]] native_inline T * address(ptr<T> const & at) noexcept requires traceable<T> {
    require_record<T>();
    assert(!collecting);
    return record_address<T>(at.get());
  }

  template<class T>
  [[nodiscard]] T load(ptr<T> const & at) const noexcept requires traceable<T> {
    require_record<T>();
    assert(at && at.local_offset() < containing(at.get()).cursor
           && record_words<T> <= containing(at.get()).cursor - at.local_offset());
    return *std::launder(reinterpret_cast<T const *>(decode(at.get())));
  }

  template<class T>
  void store(ptr<T> const & at, T const & value) noexcept requires traceable<T> {
    require_record<T>();
    require_mutator();
    heap_scope binding{*this};
    auto * destination = record_address<T>(at.get());
    if (destination != &value) *destination = value;
  }
  [[nodiscard]] std::size_t capacity() const noexcept { return arena.capacity(); }
  [[nodiscard]] std::size_t reserved() const noexcept { return arena.reserved(); }
  [[nodiscard]] std::size_t used() const noexcept { return arena.used() + nursery.used() - 1; }
  [[nodiscard]] std::size_t start() const noexcept { return arena.start(); }
  [[nodiscard]] std::size_t page_words() const noexcept { return arena.page_words(); }
  template<class Self>
  [[nodiscard]] native_inline native_pure auto data(this Self & self) noexcept
      -> std::conditional_t<std::is_const_v<Self>, word const, word> * { return self.arena.data(); }
  template<class Self>
  [[nodiscard]] decltype(auto) operator[](this Self & self, offset index) noexcept { return self.containing(index)[index & offset_mask]; }
  [[nodiscard]] offset field(offset cell, unsigned slot = 0) const noexcept { return containing(cell).field(cell & offset_mask, slot); }
  void set_field(offset cell, offset value, unsigned slot = 0) noexcept { require_mutator(); containing(cell).set_field(cell & offset_mask, value, slot);
    if (!(cell & young_bit) && (value & young_bit)) remembered.insert_or_assign(cell * 2 + slot, nullptr); }
  [[nodiscard]] std::span<block const> blocks() const noexcept { return arena.blocks(); }
  [[nodiscard]] offset allocate(std::size_t words, std::size_t alignment_bytes = 8) noexcept {
    require_mutator(); return arena.allocate(words, alignment_bytes);
  }
  void clear_marks() noexcept { arena.clear_marks(); nursery.clear_marks(); }
  void mark(offset first, std::size_t words = 1, std::size_t alignment_bytes = 8) noexcept { containing(first).mark(first & offset_mask, words, alignment_bytes); }
  [[nodiscard]] native_inline bool claim(offset first, std::size_t words = 1, std::size_t alignment_bytes = 8) noexcept {
    if (minor_collection && !(first & young_bit)) return false;
    return containing(first).claim(first & offset_mask, words, alignment_bytes);
  }
  native_inline void pointer(offset where, unsigned slot = 0) noexcept { containing(where).pointer(where & offset_mask, slot); }
  void compact(std::span<offset> roots = {}) noexcept {
    require_mutator();
    if (!weak_entries.empty() || !pending_finalizers.empty() || running_finalizer)
      detail::heap_failure("raw compact requires no weak registrations or pending finalizers");
    std::atomic_signal_fence(std::memory_order_seq_cst);
    collecting = true; compact_impl(roots); collecting = false;
    std::atomic_signal_fence(std::memory_order_seq_cst);
  }
 
};
 
inline heap_scope::heap_scope(heap & arena) noexcept : previous(heap::current()) {
  detail::active_heap = &arena;
}

export template<class T>
class root : private heap::root_handle {
  using base = heap::root_handle;
  friend class heap;
  native_inline explicit root(base value) noexcept : base(std::move(value)) {}
public:
  root() noexcept = default;
  native_inline root(ptr<T> const & value) noexcept requires traceable<T> : root(heap::current()->root(value)) {}
  template<class U> requires (!std::same_as<T, U>) && std::convertible_to<ptr<U>, ptr<T>>
  native_inline root(ptr<U> const & value) noexcept : root(ptr<T>{value}) {}

  [[nodiscard]] native_inline ptr<T> get() const noexcept { return ptr<T>{base::get()}; }
  [[nodiscard]] native_inline operator ptr<T>() const noexcept { return get(); }
  [[nodiscard]] native_inline T * operator->() const noexcept { return get().operator->(); }
  [[nodiscard]] native_inline T & operator*() const noexcept { return *get(); }
  [[nodiscard]] native_inline explicit operator bool() const noexcept { return base::get() != heap::null; }
};

export template<class T>
class weak_root : private heap::root_handle {
  using base = heap::root_handle;
  friend class heap;
  native_inline explicit weak_root(base value) noexcept : base(std::move(value)) {}
public:
  weak_root() noexcept = default;
  native_inline weak_root(ptr<T> const & value) noexcept requires traceable<T>
      : weak_root(heap::current()->weak_root(value)) {}

  native_inline weak_root(root<T> const & value) noexcept requires traceable<T>
      : weak_root(value.get()) {}

  template<class U> requires (!std::same_as<T, U>) && std::convertible_to<ptr<U>, ptr<T>>
  native_inline weak_root(ptr<U> const & value) noexcept : weak_root(ptr<T>{value}) {}
  [[nodiscard]] native_inline bool expired() const noexcept { return base::get() == heap::null; }
  [[nodiscard]] native_inline root<T> lock() const noexcept requires traceable<T> {
    if (expired()) return {};
    return heap::current()->root(ptr<T>{base::get()});
  }

  native_inline void reset() noexcept { base::operator=(base{}); }
  native_inline void swap(weak_root & other) noexcept {
    std::swap(static_cast<base &>(*this), static_cast<base &>(other));
  }

  friend native_inline void swap(weak_root & a, weak_root & b) noexcept { a.swap(b); }
};

export template<class V>
class weak {
  friend class heap;
  std::shared_ptr<heap::weak_entry> entry;
  explicit weak(std::shared_ptr<heap::weak_entry> value) noexcept : entry(std::move(value)) {}
public:
  weak() noexcept = default;
  [[nodiscard]] bool expired() const noexcept { return !entry || !entry->active; }
  [[nodiscard]] native_inline root<V> lock() const noexcept {
    if (expired()) return {};
    return heap::current()->root(ptr<V>{entry->value.at});
  }

  native_inline void finalize() const noexcept {
    // A callback may destroy the very handle used to invoke it.
    auto keep = entry;
    if (keep && keep->active) heap::current()->finalize(keep);
  }
};
 
template<traceable K, traceable V, traceable F>
jam::weak<V> heap::mk_weak(ptr<K> const & key, ptr<V> const & value,
    ptr<F> const & finalizer, std::type_identity_t<void (*)(F *) noexcept> runner) noexcept {
  require_mutator();
  if (!runner) detail::heap_failure("mk_weak requires a runner");
  auto entry = std::make_shared<finalizer_entry<F>>();
  entry->key = {key.get(), &trace_record<K>};
  entry->value = {value.get(), &trace_record<V>};
  entry->finalizer = {finalizer.get(), &trace_record<F>};
  entry->runner = runner;
  entry->invoke = [](heap & owner, weak_entry & base) noexcept {
    auto & entry = static_cast<finalizer_entry<F> &>(base);
    entry.runner(entry.finalizer.at ? owner.template record_address<F>(entry.finalizer.at) : nullptr);
  };
  weak_entries.push_back(entry);
  return jam::weak<V>{std::move(entry)};
}

export template<traceable K, traceable V, traceable F>
[[nodiscard]] native_inline weak<V> mk_weak(ptr<K> const & key, ptr<V> const & value,
    ptr<F> const & finalizer, std::type_identity_t<void (*)(F *) noexcept> runner) noexcept {
  return heap::current()->mk_weak(key, value, finalizer, runner);
}

export template<traceable T, class... Args>
[[nodiscard]] native_inline ptr<T> mk(Args &&... args) noexcept {
  return heap::current()->mk<T>(std::forward<Args>(args)...);
}

export template<class T, std::size_t N>
native_inline constexpr void unsafe_assign(std::array<ptr<T>, N> & target, std::array<ptr<T>, N> const & source) noexcept {
  for (std::size_t i = 0; i != N; ++i) target[i].unsafe_assign(source[i]);
}

export template<class T, std::size_t N>
native_inline constexpr void assign(std::array<ptr<T>, N> & target, std::array<ptr<T>, N> const & source) noexcept {
  unsafe_assign(target, source);
  if !consteval { if (auto * h = heap::current()) h->remember(std::span<ptr<T>>{target}); }
}

export template<class T, std::size_t N>
native_inline constexpr void unsafe_assign(std::array<weak_ptr<T>, N> & target,
                                          std::array<weak_ptr<T>, N> const & source) noexcept {
  for (std::size_t i = 0; i != N; ++i) target[i].unsafe_assign(source[i]);
}

export template<class T, std::size_t N>
native_inline constexpr void assign(std::array<weak_ptr<T>, N> & target,
                                   std::array<weak_ptr<T>, N> const & source) noexcept {
  unsafe_assign(target, source);
  if !consteval { if (auto * h = heap::current()) h->remember(std::span<weak_ptr<T>>{target}); }
}

export native_inline void collect() noexcept { heap::current()->collect(); }
export native_inline void collect_major() noexcept { heap::current()->collect_major(); }
export native_inline void collect_minor(bool promote = false) noexcept { heap::current()->collect_minor(promote); }
 
template<class T>
constexpr void weak_ptr<T>::update_barrier() noexcept {
  if !consteval { if (auto * h = heap::current()) h->remember(*this); }
}
template<class T>
constexpr weak_ptr<T>::weak_ptr(ptr<T> const & value) noexcept : position(value.get()) {
  if (is_young()) update_barrier();
}
template<class T>
weak_ptr<T>::weak_ptr(root<T> const & value) noexcept : weak_ptr(value.get()) {}
template<class T>
constexpr weak_ptr<T>::weak_ptr(weak_ptr const & other) noexcept : position(other.position) {
  if (is_young()) update_barrier();
}
template<class T>
constexpr weak_ptr<T> & weak_ptr<T>::operator=(weak_ptr const & other) noexcept {
  auto const changed = position ^ other.position;
  position = other.position;
  if (changed & 0x80000000u) update_barrier();
  return *this;
}
template<class T>
constexpr weak_ptr<T>::weak_ptr(weak_ptr && other) noexcept : position(std::exchange(other.position, 0)) {
  if (is_young()) { update_barrier(); other.update_barrier(); }
}
template<class T>
constexpr weak_ptr<T> & weak_ptr<T>::operator=(weak_ptr && other) noexcept {
  if (this == &other) return *this;
  *this = static_cast<weak_ptr const &>(other);
  auto const previous = std::exchange(other.position, 0);
  if (previous & 0x80000000u) other.update_barrier();
  return *this;
}
template<class T>
constexpr weak_ptr<T>::~weak_ptr() noexcept {
  if !consteval { if (auto * h = heap::current()) h->forget(this); }
}
template<class T>
constexpr void weak_ptr<T>::swap(weak_ptr & other) noexcept {
  auto const changed = position ^ other.position;
  std::swap(position, other.position);
  if (changed & 0x80000000u) { update_barrier(); other.update_barrier(); }
}
template<class T>
root<T> weak_ptr<T>::lock() const noexcept requires traceable<T> {
  if (!position) return {};
  return root<T>{ptr<T>{position}};
}
 
template<class T>
constexpr void ptr<T>::update_barrier() noexcept {
  if !consteval { if (auto * h = heap::current()) {
    if (is_young()) h->remember(*this); else h->forget(this);
  } }
}
template<class T>
constexpr ptr<T>::ptr(offset at) noexcept : position(at) {
  if (is_young()) update_barrier();
}
template<class T>
constexpr ptr<T>::ptr(ptr const & other) noexcept : ptr(other.position) {}
template<class T>
template<class U> requires (!std::same_as<T, U>) && std::derived_from<U, T>
    && detail::allocation_trace<T> && requires(T * p) { static_cast<U *>(p); }
constexpr ptr<T>::ptr(ptr<U> const & other) noexcept : ptr(other.get()) {
  if !consteval {
    assert(!other || !heap::current() || static_cast<void const *>(static_cast<T const *>(other.operator->()))
                     == static_cast<void const *>(other.operator->()));
  }
}
template<class T>
constexpr ptr<T> & ptr<T>::operator=(ptr const & other) noexcept {
  auto const changed = position ^ other.position;
  position = other.position;
  // Remembered entries name a slot and T, not its current young target.
  if (changed & heap::young_bit) update_barrier();
  return *this;
}
template<class T>
constexpr ptr<T>::ptr(ptr && other) noexcept : position(std::exchange(other.position, 0)) {
  if (is_young()) { update_barrier(); other.update_barrier(); }
}
template<class T>
constexpr ptr<T> & ptr<T>::operator=(ptr && other) noexcept {
  if (this == &other) return *this;
  *this = static_cast<ptr const &>(other);
  auto const previous = std::exchange(other.position, 0);
  if (previous & heap::young_bit) other.update_barrier();
  return *this;
}
template<class T>
constexpr void ptr<T>::swap(ptr & other) noexcept {
  auto const changed = position ^ other.position;
  std::swap(position, other.position);
  if (changed & heap::young_bit) { update_barrier(); other.update_barrier(); }
}
template<class T>
constexpr ptr<T>::~ptr() noexcept {
  if !consteval { if (auto * h = heap::current()) h->forget(this); }
}
template<class T>
T * ptr<T>::operator->() const noexcept {
  return heap::current()->address(*this);
}
template<class T>
T & ptr<T>::operator*() const noexcept { return *operator->(); }
 
template<class T>
native_inline void ptr<T>::prefetch_marks() const noexcept {
  if (!position) return;
  auto const * arena = heap::current();
  assert(arena && local_offset() < arena->containing(position).used());
  __builtin_prefetch(&arena->containing(position).blocks()[local_offset() >> heap::block_shift].live, 0, 3);
}
template<class T>
native_inline void ptr<T>::prefetch() const noexcept {
  if (!position) return;
  auto const * arena = heap::current();
  assert(arena && local_offset() < arena->containing(position).used());
  __builtin_prefetch(arena->decode(position), 0, 3);
}
}