From 5525592037ad1bb7063920c235ca66dc48cfb286 Mon Sep 17 00:00:00 2001 From: Eric Voskuil Date: Tue, 22 Sep 2026 14:17:51 -0400 Subject: [PATCH 1/5] Mark head writes on the serialization failure path. --- include/bitcoin/database/impl/primitives/headmap.ipp | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/include/bitcoin/database/impl/primitives/headmap.ipp b/include/bitcoin/database/impl/primitives/headmap.ipp index ef8f42000..5d55cd0e6 100644 --- a/include/bitcoin/database/impl/primitives/headmap.ipp +++ b/include/bitcoin/database/impl/primitives/headmap.ipp @@ -269,11 +269,9 @@ bool CLASS::put(const Link& link, const Element& element) NOEXCEPT iostream stream{ ptr.data(), size }; flipper sink{ stream }; - if (!element.to_data(sink)) - return false; - + const auto written = element.to_data(sink); file_.mark(link.value, bytes); - return true; + return written; } // NOT WRITER-WRITER THREAD SAFE (the logical top is read-write). From 9d3bc5c6673b789ebb7fd84d157be6451918a7a9 Mon Sep 17 00:00:00 2001 From: Eric Voskuil Date: Tue, 22 Sep 2026 14:18:20 -0400 Subject: [PATCH 2/5] Take the remap lock before draining head writers. --- .../database/impl/memory/mmap_staging.ipp | 29 ++++++++++--------- 1 file changed, 15 insertions(+), 14 deletions(-) diff --git a/include/bitcoin/database/impl/memory/mmap_staging.ipp b/include/bitcoin/database/impl/memory/mmap_staging.ipp index b5fa73482..cf0885cf7 100644 --- a/include/bitcoin/database/impl/memory/mmap_staging.ipp +++ b/include/bitcoin/database/impl/memory/mmap_staging.ipp @@ -1399,18 +1399,6 @@ void CLASS::head_run_() NOEXCEPT } } -// Settle a quiescent managed head to a writable mapping of its file over the -// committed span (as a shared head loads): pages drop under pressure and -// writes dirty page cache for kernel writeback, so page tracking idles (it -// remains allocated, as writers read it unlocked). Exclusive remap excludes -// accessors and the transition excludes counted head writers (raw pointer -// writes hold no lock), so a mark count still at the drained count proves -// the file current (a write landing after the drain snapshot would otherwise -// be lost to the remap). -// Exclude head writers across a settle transition: they write through raw -// pointers under no lock, so only the writer count can exclude them. The -// drain precedes the remap lock, as a writer never takes it (and a transition -// that waited under it would deadlock the first one that did). TEMPLATE std::atomic& CLASS::writer_slot_() NOEXCEPT { @@ -1445,11 +1433,24 @@ void CLASS::quiesce_() NOEXCEPT std::this_thread::yield(); } +// Settle a quiescent managed head to a writable mapping of its file over the +// committed span (as a shared head loads): pages drop under pressure and +// writes dirty page cache for kernel writeback, so page tracking idles (it +// remains allocated, as writers read it unlocked). Exclusive remap excludes +// accessors and the transition excludes counted head writers (raw pointer +// writes hold no lock), so a mark count still at the drained count proves +// the file current (a write landing after the drain snapshot would otherwise +// be lost to the remap). +// The remap lock precedes the drain: a writer holding an accessor waits +// uncounted at the transition, so a drain that preceded the lock would return +// on its own and the lock would then wait on that writer forever. Every +// counted writer either holds no lock or took its accessor before its count, +// so under the lock the count drains. TEMPLATE bool CLASS::share_(size_t transferred) NOEXCEPT { - quiesce_(); std::unique_lock map_lock(remap_mutex_); + quiesce_(); auto shared = false; if (loaded_.load() && !fault_.load() && (marks_.load() == transferred)) @@ -1488,8 +1489,8 @@ bool CLASS::share_(size_t transferred) NOEXCEPT TEMPLATE void CLASS::unshare_() NOEXCEPT { - quiesce_(); std::unique_lock map_lock(remap_mutex_); + quiesce_(); const auto floor = page_floor(to_width(logical_.load())); const auto ceiling = page_ceiling(to_width(capacity_.load())); From b60555f6760d2e62568c90a7fe189081c0453171 Mon Sep 17 00:00:00 2001 From: Eric Voskuil Date: Tue, 22 Sep 2026 14:20:48 -0400 Subject: [PATCH 3/5] Grow filled head allocation through the growth ladder. --- include/bitcoin/database/impl/memory/mmap_dispatch.ipp | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/include/bitcoin/database/impl/memory/mmap_dispatch.ipp b/include/bitcoin/database/impl/memory/mmap_dispatch.ipp index d5f86c052..c57a459a6 100644 --- a/include/bitcoin/database/impl/memory/mmap_dispatch.ipp +++ b/include/bitcoin/database/impl/memory/mmap_dispatch.ipp @@ -45,13 +45,11 @@ memory CLASS::get_filled(size_t offset, size_t size, const auto end = std::max(logical_.load(), offset + size); if (end > capacity_.load()) { - const auto extended = to_growth(end); - // TODO: Could loop over a try lock here and log deadlock warning. std::unique_lock remap_lock(remap_mutex_); // Disk full condition leaves store in valid state despite null. - if (!remap_all_(extended, sequence{})) + if (!grow_(end)) return {}; // Fill new capacity as offset may not be at end due to expansion. From afa4043de8ddc8c60b9515a5b6fe6acf72f70aa2 Mon Sep 17 00:00:00 2001 From: Eric Voskuil Date: Tue, 22 Sep 2026 14:39:44 -0400 Subject: [PATCH 4/5] Split the staging backend by mechanism and unify the worker thread. --- builds/gnu/Makefile.am | 3 + .../libbitcoin-database.vcxproj | 3 + .../libbitcoin-database.vcxproj.filters | 9 + include/bitcoin/database/impl/memory/mmap.ipp | 23 +- .../database/impl/memory/mmap_body.ipp | 381 +++++ .../database/impl/memory/mmap_extent.ipp | 298 ++++ .../database/impl/memory/mmap_head.ipp | 824 +++++++++ .../database/impl/memory/mmap_native.ipp | 172 +- .../database/impl/memory/mmap_private.ipp | 343 +--- .../database/impl/memory/mmap_staging.ipp | 1470 ++--------------- .../database/impl/memory/mmap_storage.ipp | 207 +-- include/bitcoin/database/memory/mmap.hpp | 224 ++- include/bitcoin/database/memory/mstage.hpp | 6 + src/memory/mstage.cpp | 57 +- test/memory/mmap.cpp | 87 +- 15 files changed, 2093 insertions(+), 2014 deletions(-) create mode 100644 include/bitcoin/database/impl/memory/mmap_body.ipp create mode 100644 include/bitcoin/database/impl/memory/mmap_extent.ipp create mode 100644 include/bitcoin/database/impl/memory/mmap_head.ipp diff --git a/builds/gnu/Makefile.am b/builds/gnu/Makefile.am index 9886056b9..4a9e0b34c 100644 --- a/builds/gnu/Makefile.am +++ b/builds/gnu/Makefile.am @@ -95,7 +95,10 @@ include_bitcoin_database_impl_memorydir = \ include_bitcoin_database_impl_memory_HEADERS = \ ${srcdir}/../../include/bitcoin/database/impl/memory/mmap.ipp \ + ${srcdir}/../../include/bitcoin/database/impl/memory/mmap_body.ipp \ ${srcdir}/../../include/bitcoin/database/impl/memory/mmap_dispatch.ipp \ + ${srcdir}/../../include/bitcoin/database/impl/memory/mmap_extent.ipp \ + ${srcdir}/../../include/bitcoin/database/impl/memory/mmap_head.ipp \ ${srcdir}/../../include/bitcoin/database/impl/memory/mmap_native.ipp \ ${srcdir}/../../include/bitcoin/database/impl/memory/mmap_private.ipp \ ${srcdir}/../../include/bitcoin/database/impl/memory/mmap_staging.ipp \ diff --git a/builds/msvc/vs2026/libbitcoin-database/libbitcoin-database.vcxproj b/builds/msvc/vs2026/libbitcoin-database/libbitcoin-database.vcxproj index caf773cdc..cfb5daf05 100644 --- a/builds/msvc/vs2026/libbitcoin-database/libbitcoin-database.vcxproj +++ b/builds/msvc/vs2026/libbitcoin-database/libbitcoin-database.vcxproj @@ -245,7 +245,10 @@ + + + diff --git a/builds/msvc/vs2026/libbitcoin-database/libbitcoin-database.vcxproj.filters b/builds/msvc/vs2026/libbitcoin-database/libbitcoin-database.vcxproj.filters index c9e9e5dd4..61558e599 100644 --- a/builds/msvc/vs2026/libbitcoin-database/libbitcoin-database.vcxproj.filters +++ b/builds/msvc/vs2026/libbitcoin-database/libbitcoin-database.vcxproj.filters @@ -457,9 +457,18 @@ include\bitcoin\database\impl\memory + + include\bitcoin\database\impl\memory + include\bitcoin\database\impl\memory + + include\bitcoin\database\impl\memory + + + include\bitcoin\database\impl\memory + include\bitcoin\database\impl\memory diff --git a/include/bitcoin/database/impl/memory/mmap.ipp b/include/bitcoin/database/impl/memory/mmap.ipp index dae8d968d..057942357 100644 --- a/include/bitcoin/database/impl/memory/mmap.ipp +++ b/include/bitcoin/database/impl/memory/mmap.ipp @@ -31,6 +31,9 @@ namespace database { // Constructors. // ---------------------------------------------------------------------------- +// A managed head is an unstaged scalar instance under the anonymous model: +// page-tracked, transferred by its worker, released and shared by currency. +// Unstaged aggregates transfer in full, and shared heads map their files. TEMPLATE CLASS::mmap(const path& filename, const storage_settings& settings, bool random, bool staged) NOEXCEPT @@ -42,6 +45,7 @@ CLASS::mmap(const path& filename, const storage_settings& settings, access_(settings.access), random_(random), staged_(staged), + managed_(!staged && !head_shared), opened_{ file::invalid } { } @@ -57,6 +61,7 @@ CLASS::mmap(const paths& filenames, const storage_settings& settings, access_(settings.access), random_(random), staged_(staged), + managed_(false), opened_{} { opened_.fill(file::invalid); @@ -65,12 +70,8 @@ CLASS::mmap(const paths& filenames, const storage_settings& settings, TEMPLATE CLASS::~mmap() NOEXCEPT { -#if defined(MANAGE_STAGING) - // Join a settler left running by an unload bypass (thread safety). - settler_stop_(); -#elif defined(HAVE_MSC) - scanner_stop_(); -#endif + // Join a worker left running by an unload bypass (thread safety). + worker_stop_(); BC_ASSERT(!loaded_.load()); BC_ASSERT(is_zero(logical_.load())); @@ -98,6 +99,16 @@ bool CLASS::is_loaded() const NOEXCEPT return loaded_.load(); } +TEMPLATE +bool CLASS::shared() const NOEXCEPT +{ +#if defined(MANAGE_STAGING) + return shared_.load(); +#else + return false; +#endif +} + // protected // ---------------------------------------------------------------------------- diff --git a/include/bitcoin/database/impl/memory/mmap_body.ipp b/include/bitcoin/database/impl/memory/mmap_body.ipp new file mode 100644 index 000000000..4ec199b63 --- /dev/null +++ b/include/bitcoin/database/impl/memory/mmap_body.ipp @@ -0,0 +1,381 @@ +/** + * Copyright (c) 2011-2026 libbitcoin developers (see AUTHORS) + * + * This file is part of libbitcoin. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +#ifndef LIBBITCOIN_DATABASE_MEMORY_MMAP_BODY_IPP +#define LIBBITCOIN_DATABASE_MEMORY_MMAP_BODY_IPP + +#include +#include +#include +#include +#include + +#if defined(MANAGE_STAGING) + +namespace libbitcoin { +namespace database { + +// staged body dispatch, not thread safe. +// ---------------------------------------------------------------------------- +// private + +TEMPLATE +template +bool CLASS::settle_all_(size_t rows, std::index_sequence) NOEXCEPT +{ + const auto from = settled_.load(); + if (!(settle_(from, rows) && ...)) + return false; + + settled_.store(rows); + signal_(); + check_invariants_(); + return true; +} + +TEMPLATE +template +bool CLASS::unsettle_all_(size_t rows, std::index_sequence) NOEXCEPT +{ + if (!(unsettle_(rows) && ...)) + return false; + + settled_.store(rows); + return true; +} + +TEMPLATE +template +bool CLASS::evict_all_(size_t from, size_t to, + std::index_sequence) NOEXCEPT +{ + return (evict_(from, to) && ...); +} + +TEMPLATE +template +bool CLASS::settle_write_(size_t from, size_t to, + std::index_sequence) NOEXCEPT +{ + return (pwrite_all(opened_[Index], + std::next(memory_map_[Index], to_width(from)), + to_width(to) - to_width(from), + to_width(from)) && ...); +} + +// staged body wrappers, not thread safe. +// ---------------------------------------------------------------------------- +// private + +// Convert flushed rows [from, to) to a read-only shared file mapping, page +// floored so the settle boundary page remains anonymous with its settled +// bytes retained. Releases the covered anonymous pages. Failure results in +// unmapped. +TEMPLATE +template +bool CLASS::settle_(size_t from, size_t to) NOEXCEPT +{ + if (!staged_) + return true; + + const auto begin = page_floor(to_width(from)); + const auto end = page_floor(to_width(to)); + if (begin == end) + return true; + + const auto address = std::next(memory_map_[Column], begin); + if (mmap_settle(address, end - begin, opened_[Column], begin) == fail) + { + teardown_(error::mmap_failure); + return false; + } + +#if !defined(WITHOUT_MADVISE) + if (!advise_(address, end - begin)) + { + teardown_(error::madvise_failure); + return false; + } + + // Demote the settled extent to reclaim-first order: cached for imminent + // re-read (validation follows archival) but reclaimed under pressure + // before active pages and anonymous heads (head residency priority). + // Ample hosts skip demotion: with nothing contesting memory it only + // converts warm cache into re-read faults (see demote_memory). + static const auto ample = system_memory() > demote_memory; + if (!ample && (mmap_cold(address, end - begin) == fail)) + { + teardown_(error::madvise_failure); + return false; + } +#endif + + return true; +} + +// Revert settled pages at/above rows to committed anonymous memory and +// restore the retained bytes below rows from the file (truncation below the +// settle boundary). Failure results in unmapped. +TEMPLATE +template +bool CLASS::unsettle_(size_t rows) NOEXCEPT +{ + const auto bytes = to_width(rows); + const auto begin = page_floor(bytes); + const auto end = page_floor(to_width(settled_.load())); + if (begin == end) + return true; + + const auto address = std::next(memory_map_[Column], begin); + if (mmap_unsettle(address, end - begin) == fail) + { + teardown_(error::mmap_failure); + return false; + } + + if ((begin < bytes) && !pread_all(opened_[Column], address, bytes - begin, + begin)) + { + teardown_(error::fsync_failure); + return false; + } + + return true; +} + +// Release cached pages of settled rows [from, to), page floored within the +// converted read-only mapping (its pages cannot be dirtied, and any pending +// settle write-back is synced by the release). Cost follows residency, so +// re-eviction is idempotent and free. The mapping is unaffected (a later +// read faults the page back from the file). Failure faults the store +// (advisory-class failures have no benign modes) but leaves it mapped, as +// the shared-lock caller cannot tear down under live accessors. +TEMPLATE +template +bool CLASS::evict_(size_t from, size_t to) NOEXCEPT +{ + const auto begin = page_floor(to_width(from)); + const auto end = page_floor(to_width(to)); + if (begin == end) + return true; + + if (mmap_evict(std::next(memory_map_[Column], begin), end - begin) == fail) + { + set_first_code(error::fsync_failure); + return false; + } + + return true; +} + +// write throttle. +// ---------------------------------------------------------------------------- +// private + +// Delay the caller while staging exceeds its memory bound (write throttle). +// Signaled as the worker drains; a fault or worker stop releases the caller +// into the normal allocation path (which fails on fault/unload). The relieved +// state is lock-free for the common (unparked) case; notifiers synchronize on +// throttle_mutex_ so a parked caller cannot miss its signal. +TEMPLATE +void CLASS::throttle_() NOEXCEPT +{ + const auto relieved = [this]() NOEXCEPT + { + using namespace system; + const auto rows = floored_subtract(logical_.load(), settled_.load()); + const auto debt = ceilinged_multiply(rows, stride); + return (debt <= limit_) || !working_.load() || fault_.load(); + }; + + if (!staged_ || relieved()) + return; + + std::unique_lock throttle_lock(throttle_mutex_); + + throttle_cv_.wait(throttle_lock, relieved); +} + +// Wake throttled callers, synchronized so a parking caller cannot miss the +// signal between its relieved check and its wait. +TEMPLATE +void CLASS::signal_() NOEXCEPT +{ + std::unique_lock throttle_lock(throttle_mutex_); + + throttle_cv_.notify_all(); +} + +// settle scheduler (worker thread, drains completed writes to clean cache). +// ---------------------------------------------------------------------------- +// private + +// Pressure-paced draining (the windows model): drain intensity follows memory +// conditions, settled pages remain cached, writers delay only at the staging +// memory bound (write throttle), sustained stillness drains the residual (the +// lazy writer) so a quiescent map converges to fully settled. +TEMPLATE +void CLASS::body_run_() NOEXCEPT +{ + // Tiers derive from physical memory, pressure and compression occupancy + // (a pressure precursor: the kernel compresses without raising level). + const auto memory = system_memory(); + const auto urgent = memory / urgent_factor; + const auto active = memory / active_factor; + const auto squeeze = memory / compress_factor; + const auto scarce = memory / sweep_factor; + const auto chunk = std::max(one, settle_chunk / stride); + const auto sweep = std::max(one, evict_chunk / stride); + + // Ticks without allocation before idle draining (settling is not writing, + // so draining does not hold its own clock). + auto mark = logical_.load(); + size_t still{}; + size_t evicted{}; + + const auto backlog = [this]() NOEXCEPT + { + using namespace system; + return ceilinged_multiply(floored_subtract(frontier_.load(), + settled_.load()), stride); + }; + + while (tick_()) + { + const auto top = logical_.load(); + still = (top == mark) ? std::min(add1(still), idle_seconds) : zero; + mark = top; + + // Scarcity is read directly: clean cache is reclaimable, so the + // kernel pressure level does not raise while free memory exhausts. + if (system_free() < scarce) + evict_next_(sweep, evicted); + + auto bytes = backlog(); + if (is_zero(bytes)) + continue; + + // Urgency drains continuously, activity/stillness one chunk per tick. + const auto driven = + (system_pressure() > one) || + (system_compressed() > squeeze) || + (bytes > urgent); + + if (!driven && + (bytes <= active) && + (still < idle_seconds)) + continue; + + do + { + if (!settle_next_(chunk)) + break; + + bytes = backlog(); + } + while (working_.load() && driven && (bytes > active)); + } +} + +// Settle up to chunk completed rows: write under the shared remap lock +// (completed extents are immutable, writers proceed), convert under a brief +// exclusive. Durability remains a snapshot property (no sync here). +TEMPLATE +bool CLASS::settle_next_(size_t chunk) NOEXCEPT +{ + size_t from{}; + size_t target{}; + + { + std::shared_lock map_lock(remap_mutex_); + + if (!loaded_.load() || fault_.load()) + return false; + + from = settled_.load(); + target = std::min(frontier_.load(), + system::ceilinged_add(from, chunk)); + + if (target <= from) + return false; + + if (!settle_write_(from, target, sequence{})) + { + set_first_code(error::fsync_failure); + return false; + } + } + + std::unique_lock map_lock(remap_mutex_); + + if (!loaded_.load()) + return false; + + // A raced settle boundary (flush) invalidates only this conversion. + if (settled_.load() != from) + return true; + + return settle_all_(target, sequence{}); +} + +// Evict up to chunk settled rows under memory scarcity: a wrapping sweep +// from the oldest offset (offset is write age in an append-only body, and +// re-read probability decays with it). Scarcity-driven release keeps the +// machine out of the exhausted-cache regime, where the kernel periodic +// sync walk (cost follows resident pages, not dirty) otherwise degrades +// fault service. A wrong eviction costs one cold read, so the sweep needs +// no precision. Runs under the shared remap lock (readers proceed). +TEMPLATE +bool CLASS::evict_next_(size_t chunk, size_t& cursor) NOEXCEPT +{ + std::shared_lock map_lock(remap_mutex_); + + if (!loaded_.load() || fault_.load()) + return false; + + // A truncated boundary self-heals here (the cursor clamps to a lap). + const auto end = settled_.load(); + + // Confine the lap to the recently settled tail: cache holds what was + // just written (and what validation just read), while the body below + // is long evicted, so a full-body lap sweeps unresident rows for most + // of its length and the tail refills faster than the cursor returns. + // A lap can hold no more than physical memory, as nothing else is + // resident to evict. + const auto span = system::greater(chunk, system_memory() / stride); + const auto floor = (end > span) ? (end - span) : zero; + const auto from = ((cursor > floor) && (cursor < end)) ? cursor : floor; + const auto target = std::min(end, system::ceilinged_add(from, chunk)); + + if (target <= from) + return false; + + if (!evict_all_(from, target, sequence{})) + return false; + + // Wrap at the lap end so the next sweep resumes at the tail floor. + cursor = (target == end) ? zero : target; + return true; +} + +} // namespace database +} // namespace libbitcoin + +#endif // MANAGE_STAGING + +#endif diff --git a/include/bitcoin/database/impl/memory/mmap_extent.ipp b/include/bitcoin/database/impl/memory/mmap_extent.ipp new file mode 100644 index 000000000..0c6415ea2 --- /dev/null +++ b/include/bitcoin/database/impl/memory/mmap_extent.ipp @@ -0,0 +1,298 @@ +/** + * Copyright (c) 2011-2026 libbitcoin developers (see AUTHORS) + * + * This file is part of libbitcoin. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +#ifndef LIBBITCOIN_DATABASE_MEMORY_MMAP_EXTENT_IPP +#define LIBBITCOIN_DATABASE_MEMORY_MMAP_EXTENT_IPP + +#include +#include +#include + +namespace libbitcoin { +namespace database { + +// Write-completion accounting (staged instances). Extents chase completions, +// closing the gaps, settling the completed prefix. +// ---------------------------------------------------------------------------- + +TEMPLATE +void CLASS::complete(size_t STAGING_ONLY(offset), + size_t STAGING_ONLY(count)) NOEXCEPT +{ +#if defined(MANAGE_STAGING) + if (!staged_ || is_zero(count)) + return; + + // The covering extent is immobile while this completion is pending and + // cannot refuse a consistent claim, so a failed pass raced a recycle + // (torn navigation or stale snapshot): rescan against a fresh window. + for (;;) + { + // Acquire-ordered window snapshot (published entries are immobile; + // the packed word precludes observing a torn head/size pair). + const auto window = window_.load(std::memory_order_acquire); + const auto [head, size] = system::unpack_word(window); + + // Lock-free binary search of the sorted window (relaxed navigation + // is heuristic only; claim_ verifies cover under the generation). + auto high = size; + for (size_t low{}; low < high;) + { + const auto middle = to_half(low + high); + auto& record = ring_.at((head + middle) % extents); + const auto start = record.start.load(relaxed); + + if (offset < start) + { + high = middle; + continue; + } + + if (offset >= (start + record.count.load(relaxed))) + { + low = add1(middle); + continue; + } + + if (claim_(record, offset, count)) + return; + + // The matched slot recycled under the search: scan below. + break; + } + + // Contention can record extents out of start order (allocation claim + // and recording are not one atomic step), transiently breaking search + // order. Contiguity guarantees a unique cover: scan linearly. + for (size_t index{}; index < size; ++index) + if (claim_(ring_.at((head + index) % extents), offset, count)) + return; + } +#endif +} + +TEMPLATE +size_t CLASS::frontier() const NOEXCEPT +{ +#if defined(MANAGE_STAGING) + if (staged_) + return frontier_.load(); +#endif + + return size(); +} + +#if defined(MANAGE_STAGING) + +// extent ring, locked except claim_ (lock-free). +// ---------------------------------------------------------------------------- +// private + +// Claim completion of count rows against the extent, by cas decrement of the +// packed state under a cover check ordered by that state's own acquire: the +// publishing release sequences start/count before state, so both read after +// it belong to the observed generation (a live outstanding precludes a +// recycle in progress, as slots re-record only after retirement). A range +// match read before the acquire can be torn across a recycle and debit the +// slot's successor extent, pinning the true cover's frontier forever. False +// implies the slot does not (or no longer does) cover the offset, or is +// drained (recycle in flight): the caller rescans. The true covering extent +// cannot refuse: its generation is stable and its outstanding covers every +// pending completion while any remains. +TEMPLATE +bool CLASS::claim_(extent& record, size_t offset, size_t count) NOEXCEPT +{ + using namespace system; + auto state = record.state.load(std::memory_order_acquire); + + for (;;) + { + const auto generation = shift_right(state, generation_shift); + const auto outstanding = bit_and(state, outstanding_mask); + const auto start = record.start.load(relaxed); + + // An outstanding below the claim is a recycle in flight or a drained + // slot (never the covering extent), refused rather than wrapped. + if ((outstanding < count) || (offset < start) || + (offset >= (start + record.count.load(relaxed)))) + return false; + + if (record.state.compare_exchange_weak(state, + pack_extent_(generation, outstanding - count), + std::memory_order_acq_rel, std::memory_order_acquire)) + { + // Extent completion is allocation-coarse, so maintenance is + // cheap here and the element write fast path takes no lock. + if (outstanding == count) + { + std::unique_lock extent_lock(extent_mutex_, std::try_to_lock); + + if (extent_lock.owns_lock()) + maintain_(); + } + + return true; + } + + // The failed cas reloaded state (acquire): recheck under it. + } +} + +// Claim and record an extent under one lock: a claim never exists outside +// the ring and the ring is start-ordered, so the frontier can never pass an +// unwritten extent (claim-then-record raced the frontier past the claim). +// Returns eof (unclaimed) on insufficient capacity, fault, or disk full. +TEMPLATE +size_t CLASS::record_(size_t count) NOEXCEPT +{ + std::unique_lock extent_lock(extent_mutex_); + + if (is_zero(count)) + return logical_.load(); + + maintain_(); + + using namespace system; + auto [head, size] = unpack_word(window_.load(relaxed)); + + // A full ring waits on completions (extents are allocation-coarse, so + // saturation implies extreme concurrency). Completions are lock-free, so + // waiting needs only this thread's own maintenance; fault or disk full + // releases the wait unclaimed (the write then fails fast). + while (size == extents) + { + if (fault_.load() || !is_zero(space_.load())) + return storage::eof; + + std::this_thread::yield(); + maintain_(); + std::tie(head, size) = unpack_word(window_.load(relaxed)); + } + + const auto start = logical_.load(); + if (is_add_overflow(start, count) || + ((start + count) > capacity_.load())) + return storage::eof; + + auto& record = ring_.at((head + size) % extents); + const auto generation = bit_and(add1(shift_right( + record.state.load(relaxed), generation_shift)), generation_mask); + BC_ASSERT((count * columns) <= outstanding_mask); + record.start.store(start, relaxed); + record.count.store(count, relaxed); + + // Publish the state (pairs with the acquire in claim_): a claim against + // the stale generation now fails its cas, and one against this + // generation observes the new start and count through the release. + record.state.store(pack_extent_(generation, count * columns), + std::memory_order_release); + + // Publish the extent (pairs with the acquire window snapshot). + window_.store(pack_word(head, add1(size)), release); + if (is_zero(size)) + frontier_.store(start); + + logical_.store(start + count); + check_invariants_(); + return start; +} + +// Pop completed extents from the head, advancing the frontier (locked). +TEMPLATE +void CLASS::maintain_() NOEXCEPT +{ + using namespace system; + auto [head, size] = unpack_word(window_.load(relaxed)); + while (!is_zero(size) && is_zero(bit_and( + ring_.at(head).state.load(relaxed), outstanding_mask))) + { + head = add1(head) % extents; + --size; + } + + window_.store(pack_word(head, size), release); + frontier_.store(is_zero(size) ? logical_.load() : + ring_.at(head).start.load(relaxed)); + check_invariants_(); +} + +// Discard all extents, requires quiescent writers (locked). Any extent then +// outstanding is an abandoned write (unreferenced), safe to settle as is. +TEMPLATE +void CLASS::discard_() NOEXCEPT +{ + if (!staged_) + return; + + std::unique_lock extent_lock(extent_mutex_); + + window_.store(zero, release); + frontier_.store(logical_.load()); + check_invariants_(); +} + +// Discard extents above the truncation and clamp any overlap, requires +// quiescent writers (locked). +TEMPLATE +void CLASS::trim_(size_t count) NOEXCEPT +{ + if (!staged_) + return; + + std::unique_lock extent_lock(extent_mutex_); + + using namespace system; + auto [head, size] = unpack_word(window_.load(relaxed)); + while (!is_zero(size)) + { + auto& tail = ring_.at((head + sub1(size)) % extents); + const auto start = tail.start.load(relaxed); + if (start >= count) + { + --size; + continue; + } + + if (ceilinged_add(start, tail.count.load(relaxed)) > count) + { + const auto trimmed = count - start; + tail.count.store(trimmed, relaxed); + + // Clamp outstanding within the state, generation unchanged + // (the extent is trimmed, not recycled; writers quiescent). + const auto limit = trimmed * columns; + const auto state = tail.state.load(relaxed); + if (bit_and(state, outstanding_mask) > limit) + tail.state.store(pack_extent_(shift_right( + state, generation_shift), limit), relaxed); + } + + break; + } + + window_.store(pack_word(head, size), release); + frontier_.store(is_zero(size) ? count : + ring_.at(head).start.load(relaxed)); +} + +#endif // MANAGE_STAGING + +} // namespace database +} // namespace libbitcoin + +#endif diff --git a/include/bitcoin/database/impl/memory/mmap_head.ipp b/include/bitcoin/database/impl/memory/mmap_head.ipp new file mode 100644 index 000000000..36b2e6bfe --- /dev/null +++ b/include/bitcoin/database/impl/memory/mmap_head.ipp @@ -0,0 +1,824 @@ +/** + * Copyright (c) 2011-2026 libbitcoin developers (see AUTHORS) + * + * This file is part of libbitcoin. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +#ifndef LIBBITCOIN_DATABASE_MEMORY_MMAP_HEAD_IPP +#define LIBBITCOIN_DATABASE_MEMORY_MMAP_HEAD_IPP + +#include +#include +#include +#include +#include +#include + +namespace libbitcoin { +namespace database { + +// Write declaration and marking (managed head instances). Marks drive the +// dirty transfer, declarations restore released pages ahead of the write. +// ---------------------------------------------------------------------------- + +TEMPLATE +void CLASS::prepare(size_t STAGING_ONLY(offset), + size_t STAGING_ONLY(size)) NOEXCEPT +{ +#if defined(MANAGE_STAGING) + if (is_zero(size) || !managed_) + return; + + // Count the writer before loading released below (sequentially + // consistent), pairing with the release protocol: release either observes + // the count (and aborts) or this writer observes released (and restores). + // Intent bits age (hot sampling), so they cannot protect a write held + // in flight across passes; the count persists until mark. + // A settle transition pairs the same way, but the writer waits UNCOUNTED + // (it retires its count and does not retake one until the transition + // clears), so the count drains monotonically and the transition is + // guaranteed to observe zero rather than merely likely to. + auto& writers = writer_slot_(); + for (;;) + { + while (transition_.load()) + std::this_thread::yield(); + + writers.fetch_add(one); + if (!transition_.load()) + break; + + writers.fetch_sub(one); + } + + // A settled head writes through its mapping. + if (shared_.load()) + return; + + if (!engaged_.load(relaxed) && !lazy_.load(relaxed)) + return; + + // Declare intent before the write (sequentially consistent, pairing with + // the release protocol), then restore any released page in the range. + auto restore = false; + auto page = offset / page_; + const auto end = (offset + sub1(size)) / page_; + while ((page <= end) && ((page / page_bound) < words_)) + { + const auto word = page / page_bound; + const auto flag = system::bit_right(page % page_bound); + intent_[word].fetch_or(flag); + restore |= !is_zero(system::bit_and(released_[word].load(), flag)); + ++page; + } + + if (restore) + restore_(offset, size); +#endif +} + +TEMPLATE +void CLASS::mark(size_t STAGING_ONLY(offset), + size_t STAGING_ONLY(size)) NOEXCEPT +{ +#if defined(MANAGE_STAGING) + if (is_zero(size) || !managed_) + return; + + // Marks follow content writes; transfer clears before reading, so pages + // remarked during a transfer are simply rewritten by the next pass. A + // settled head writes through its mapping, so its marks count only (the + // worker reads the rate); no transition intervenes (the writer is + // counted), so the path matches prepare. + if (shared_.load()) + marks_.fetch_add(one, relaxed); + else + remark_(offset, size); + + // Uncount the writer after its marks (sequentially consistent), so a + // release pass loading a drained count observes the dirty bits. Only + // prepare() counts, so only mark() may uncount (transfer failure restores + // marks by remark_, as an unpaired uncount here corrupts the count). + writer_slot_().fetch_sub(one); +#endif +} + +TEMPLATE +void CLASS::current(bool STAGING_ONLY(state)) NOEXCEPT +{ +#if defined(MANAGE_STAGING) + current_.store(state); +#endif +} + +#if defined(MANAGE_STAGING) + +// managed head installation, not thread safe. +// ---------------------------------------------------------------------------- +// private + +// Head-creation fill: the fill is file content, written sequentially to the +// file (page cache, no mapping involvement) and installed released, so +// creation contributes no memory residency (an in-memory backfill of the +// head set is the store's largest allocation transient). +TEMPLATE +size_t CLASS::allocate_filled_(size_t count, uint8_t backfill) NOEXCEPT +{ + BC_ASSERT(managed_ && !shared_.load()); + std::unique_lock field_lock(field_mutex_); + std::unique_lock map_lock(remap_mutex_); + + using namespace system; + const auto start = logical_.load(); + if (!loaded_.load() || fault_.load() || is_add_overflow(start, count)) + return storage::eof; + + // Provision the file physically (disk full detected here). + const auto end = start + count; + if (!resize_(end)) + return storage::eof; + + // Write the fill. + const auto from = to_width(start); + const auto to = to_width(end); + std::vector chunk(std::min(to - from, release_chunk), backfill); + for (auto at = from; at < to; ) + { + const auto size = std::min(to - at, chunk.size()); + if (!pwrite_all(opened_[zero], chunk.data(), size, at)) + { + set_first_code(error::fsync_failure); + return storage::eof; + } + + at += size; + } + + // The fill bounds capacity (installation populates only the fill). + BC_ASSERT(capacity_.load() <= end); + logical_.store(end); + file_.store(std::max(file_.load(), end)); + capacity_.store(end); + if (!lazy_install_()) + return storage::eof; + + check_invariants_(); + return start; +} + +// Install the managed head lazily over its current logical span: a released +// (read-only file) full-page prefix restored to anonymous per segment on +// first write, an anonymous populated tail, and clean page tracking sized to +// a replacement reservation. Caller holds exclusive remap (or is loading). +TEMPLATE +bool CLASS::lazy_install_() NOEXCEPT +{ + using namespace system; + const auto rows = logical_.load(); + const auto logical = to_width(rows); + + // Replace any standing reservation (sized as stage_). + if (!is_null(memory_map_[zero])) + mmap_unreserve(memory_map_[zero], reserved_[zero]); + + const auto reserved = page_ceiling(to_width( + to_reservation(to_provision()))); + const auto base = mmap_reserve(reserved); + if (base == MAP_FAILED) + { + set_first_code(error::mmap_failure); + return false; + } + + memory_map_[zero] = pointer_cast(base); + reserved_[zero] = reserved; + + // Rebuild page tracking for the new reservation (value-initialized). + const auto pages = ceilinged_divide(reserved, page_); + words_ = ceilinged_divide(pages, page_bound); + dirty_ = std::make_unique(words_); + intent_ = std::make_unique(words_); + released_ = std::make_unique(words_); + sweep_ = std::make_unique(words_); + writers_reset_(); + + // Released file prefix (full pages below logical). + const auto floor = page_floor(logical); + if (is_nonzero(floor) && (mmap_settle(memory_map_[zero], floor, + opened_[zero], zero) == fail)) + { + set_first_code(error::mmap_failure); + return false; + } + + // Commit the remainder to the commitment target and populate the tail. + const auto target = std::max(page_ceiling(to_width( + to_commitment())), page_ceiling(logical)); + if ((target > floor) && (mmap_commit(std::next(memory_map_[zero], floor), + target - floor, headroom_) == fail)) + { + set_first_code(error::mmap_failure); + return false; + } + + if ((logical > floor) && !pread_all(opened_[zero], + std::next(memory_map_[zero], floor), logical - floor, floor)) + { + set_first_code(error::fsync_failure); + return false; + } + + if (target > floor) + mmap_wire(std::next(memory_map_[zero], floor), target - floor); + + declare_released_(); + + // Attribute the anonymous span for diagnostics (smaps decomposition). + mmap_name(std::next(memory_map_[zero], floor), reserved - floor, + filenames_.front().filename().string().c_str()); + + return true; +} + +// dirty page transfer, lock-free with writers. +// ---------------------------------------------------------------------------- +// private + +// Dirty marking without the writer uncount (no paired prepare), for internal +// mark restoration (mark() pairs with prepare() and uncounts its writer). +TEMPLATE +void CLASS::remark_(size_t offset, size_t size) NOEXCEPT +{ + auto page = offset / page_; + const auto end = (offset + sub1(size)) / page_; + while ((page <= end) && ((page / page_bound) < words_)) + { + const auto bit = system::bit_right(page % page_bound); + dirty_[page++ / page_bound].fetch_or(bit, relaxed); + } + + marks_.fetch_add(one, relaxed); +} + +// Transfer dirty pages within [0, bytes), clearing marks before content +// reads so that concurrently remarked pages transfer on the next pass. +// Marks beyond bytes are retained (backfill above logical transfers when +// logical grows over it). Adjacent dirty pages coalesce into single writes. +TEMPLATE +template +bool CLASS::transfer_(size_t bytes) NOEXCEPT +{ + using namespace system; + if (is_zero(bytes)) + return true; + + // One pass at a time: concurrent passes split the claimed dirty set. + std::unique_lock transfer_lock(transfer_mutex_); + + // Untracked (multi-column unstaged) instances transfer in full. + if (!managed_) + return pwrite_all(opened_[Column], memory_map_[Column], bytes, zero); + + size_t from{}; + size_t to{}; + const auto write = [&]() NOEXCEPT + { + return (from >= to) || pwrite_all(opened_[Column], + std::next(memory_map_[Column], from), to - from, from); + }; + + const auto pages = ceilinged_divide(bytes, page_); + const auto bound = std::min(words_, ceilinged_divide(pages, page_bound)); + for (size_t word{}; word < bound; ++word) + { + auto bits = dirty_[word].exchange(zero, relaxed); + + // Retain marks at and above the page bound (boundary word only). + const auto first = word * page_bound; + if (pages < (first + page_bound)) + { + const auto keep = mask_right(pages - first); + dirty_[word].fetch_or(bit_and(bits, keep), relaxed); + bits = bit_and(bits, bit_not(keep)); + } + + for (size_t bit{}; !is_zero(bits) && (bit < page_bound); ++bit) + { + if (!get_right(bits, bit)) + continue; + + bits = set_right(bits, bit, false); + const auto start = (first + bit) * page_; + const auto end = std::min(start + page_, bytes); + if (start == to) + { + to = end; + continue; + } + + if (!write()) + { + // Restore claimed marks (the failed range, the page that + // ended it, and this word's unwritten remainder) so failure + // is retryable, not lossy. + remark_(from, to - from); + remark_(start, end - start); + dirty_[word].fetch_or(bits, relaxed); + return false; + } + + from = start; + to = end; + } + } + + if (!write()) + { + remark_(from, to - from); + return false; + } + + return true; +} + +// head scheduler (worker thread). +// ---------------------------------------------------------------------------- +// private + +// Idle transfer for unstaged (rewrite-in-place head) instances: sustained +// mark stillness drains dirty pages (the lazy writer) so a quiescent map +// converges to persisted and snapshot/close transfer approximately nothing. +// Requires no write exclusion: marks follow writes and transfer clears +// before reading, so racing writes remark and transfer on the next pass +// (torn disk pages are unreachable, as live heads are only trusted +// following a clean close). +// +// Memory scarcity additionally pages the head to its own file (the windows +// model): transfer, then release cold clean pages to read-only file mappings +// (reclaimable cache), restored to anonymous by prepare() before any write. +// Release waits one pass after engagement so all writers declare intent. +TEMPLATE +void CLASS::head_run_() NOEXCEPT +{ + const auto scarce = system_memory() / evict_factor; + + // Ticks without a mark before idle draining. + auto mark = marks_.load(); + auto transferred = mark; + size_t still{}; + size_t hot{}; + size_t touched{}; + + while (tick_()) + { + const auto top = marks_.load(); + const auto writes = top - mark; + still = (top == mark) ? std::min(add1(still), idle_seconds) : zero; + mark = top; + + // A settled head under sustained writes reinstalls lazily. + if (shared_.load()) + { + hot = (writes >= unshare_writes) ? add1(hot) : zero; + if (hot >= unshare_seconds) + { + unshare_(); + transferred = top; + hot = zero; + } + + continue; + } + + // A drained head settles to its file mapping while the store is + // current (settled pages are droppable, anonymous pages are not). + if (current_.load() && (transferred == top) && share_(transferred)) + continue; + + // Touch pass: assert working-set residency at a bounded rate. + // Hash-uniform probing is per-page sparse in every phase, so the + // kernel ages head pages cold and swaps them under cache pressure, + // though the set is the process working set (and head misses are + // unshieldable serial faults on the probe path). A read sets the + // page-table accessed bit without dirtying the page; volatile + // prevents elision. The unprivileged equivalent of mlock, and soft: + // under true extremity the kernel can still take the pages. The + // per-tick budget revisits every page each touch_seconds regardless + // of instance size (a whole-instance lap gave the largest head a + // proportionally longer revisit, which lost the aging race first). + // Residency-guarded so the touch never faults: a swapped page that + // nothing probes rests in swap, one the workload needs returns by + // its own fault and is defended thereafter. + // Residency is only contested under scarcity, so at plenty the tick + // costs a counter test (the aging race has no other runner). + if (system_available() < (system_memory() / sweep_factor)) + { + using namespace system; + std::shared_lock touch_lock(remap_mutex_); + if (loaded_.load() && !fault_.load()) + { + const auto pages = to_width(logical_.load()) / page_; + auto budget = ceilinged_divide(pages, touch_seconds); +#if !defined(HAVE_APPLE) + // Probe buffer and map alias, declared with the touch they + // serve (excluded on darwin below), as they are otherwise + // unused there. + unsigned char resident[touch_span]; + const volatile auto* map = memory_map_[zero]; +#endif + while (!is_zero(pages) && !is_zero(budget)) + { + if (touched >= pages) + touched = zero; + + const auto count = std::min( + { touch_span, pages - touched, budget }); + +#if !defined(HAVE_APPLE) + // Darwin excluded: mincore hides compressed pages from + // the guard, and unguarded touching measured as pure + // decompression churn; release is the darwin mechanism. + const auto at = touched * page_; + if (mmap_resident(std::next(memory_map_[zero], at), + count * page_, resident) == 0) + for (size_t page{}; page < count; ++page) + if (is_odd(resident[page])) + (void)map[at + page * page_]; +#endif + + touched += count; + budget = floored_subtract(budget, count); + } + } + } + + // Available includes reclaimable file cache, which a loaded store + // keeps large while the kernel swaps cold anonymous pages, so free + // exhaustion also signals scarcity (anon is being displaced). + // A sync writes every head hot (a converted run restores at the next + // burst), so release engages only while the store is current. + const auto scarcity = head_release && managed_ && current_.load() && + ((system_available() < scarce) || (system_free() < scarce)); + + // Once engaged, a quiet instance converts independent of momentary + // scarcity: the signal clears as swap absorbs the hot set, but the + // swapped pages remain anonymous, so reads fault them back one at a + // time (a serial swap-in per probe). Conversion instead settles + // clean pages without read-back (the file holds their content), + // freeing swap and routing reads through the file mapping. + const auto engaged = engaged_.load(); + const auto quiet = writes < release_quiet; + const auto draining = engaged && quiet; + if (!scarcity && !draining && + ((still < idle_seconds) || (transferred == top))) + continue; + + // A write-hot instance neither transfers nor releases under + // scarcity: transferred pages re-dirty immediately (write + // amplification without release payoff, as hash-scattered writes + // into released pages each cost a segment restore, and the sweep + // otherwise re-releases restored segments every pass). Its + // anonymous set is left to swap (dirty-exempt) until quiescence, + // typically the phase change. +#if !defined(HAVE_APPLE) + // EXPERIMENT: darwin releases while write-hot (page-level intent and + // dirty filters remain) to price segment restores against the + // measured compressed-hot crawl. + if (scarcity && !quiet) + continue; +#endif + + { + std::shared_lock map_lock(remap_mutex_); + + if (!loaded_.load() || fault_.load()) + continue; + + if constexpr (head_release) + if (scarcity && !engaged) + engaged_.store(true); + + // Scarcity passes pace writeback for release: settle maps page + // cache content, and durability remains a clean close property, + // so sync applies only to idle draining. + if (!transfer_(to_width(logical_.load())) || + (!scarcity && !sync_())) + { + set_first_code(error::fsync_failure); + continue; + } + + // Discard the page cache copy of the transfer: the anonymous + // head is the live copy, so caching the file doubles it. + file_discard(opened_[zero]); + + // Quiet is assured here (hot scarcity skipped above, and idle + // draining implies sixty still seconds). + // Lazy head load engages restore on all platforms; the release + // sweep remains a darwin response (see head_release). + if constexpr (head_release) + if (engaged && !release_pages_()) + continue; + } + + transferred = top; + still = zero; + } +} + +// head page release, synchronized with writers by the bit protocol. +// ---------------------------------------------------------------------------- +// private + +// Declare the full-page prefix below logical released and engage the restore +// protocol (prepare restores a released segment before its write). +TEMPLATE +void CLASS::declare_released_() NOEXCEPT +{ + using namespace system; + const auto flags = page_floor(to_width(logical_.load())) / page_; + for (size_t word{}; word < ceilinged_divide(flags, page_bound); ++word) + { + const auto first = word * page_bound; + released_[word].store((flags >= (first + page_bound)) ? + bit_all : unmask_right(flags - first)); + } + + lazy_.store(true); +} + +// Release cold clean head page runs to read-only file mappings (reclaimable), +// full pages below logical only. Conversion is run-granular (release_chunk +// minimum) as each conversion splits a mapping: page granularity fragments +// the address space beyond what host memory management tolerates. Writer +// synchronization is a per-page bit protocol: prepare() declares intent then +// loads released; release stores released then loads intent (both +// sequentially consistent), so a run converts only when no write can land on +// it unrestored. A wrong release costs one restore. Conversion and restore +// serialize on restore_mutex_. +TEMPLATE +bool CLASS::release_pages_() NOEXCEPT +{ + using namespace system; + const auto bytes = to_width(logical_.load()); + const auto pages = bytes / page_; + const auto bound = std::min(words_, ceilinged_divide(pages, page_bound)); + const auto chunk = std::max(one, release_chunk / page_); + + std::unique_lock restore_lock(restore_mutex_); + + // Materialize candidacy (hot aging clears only the snapshot bits, so a + // concurrent declaration on a candidate page is retained for the live + // rechecks below). + for (size_t word{}; word < bound; ++word) + { + const auto hot = intent_[word].load(); + intent_[word].fetch_and(bit_not(hot)); + sweep_[word] = release_below(release_candidates( + dirty_[word].load(relaxed), hot, released_[word].load(relaxed)), + pages, word * page_bound); + } + + // Convert maximal candidate runs of at least chunk pages, clamped to + // chunk bounds. Restore is chunk-aligned and chunk-granular, so a + // partially released chunk places unreleased pages inside a restored + // span. The bit protocol guards released pages only: a writer to an + // unreleased page observes released clear, so it neither restores nor + // takes restore_mutex_, and its write is dropped by the copy-install + // window. Whole chunks make the restored span exactly the released set. + for (auto run = next_run(sweep_.get(), pages, zero); run.first < pages; + run = next_run(sweep_.get(), pages, run.second)) + { + const auto first = ceilinged_divide(run.first, chunk) * chunk; + const auto second = (run.second / chunk) * chunk; + + if ((second <= first) || ((second - first) < chunk)) + continue; + + const auto begin = first / page_bound; + const auto end = sub1(second) / page_bound; + const auto mask = [&](size_t word) NOEXCEPT + { + return page_mask(first, second, word * page_bound); + }; + + // Declare the release (prepare() restores from this point). + for (auto word = begin; word <= end; ++word) + released_[word].fetch_or(mask(word)); + + // An in-flight writer (counted, unaged) or raced intent or mark + // invalidates the conversion (whole run). The count loads first: a + // writer counted later observes released and restores, one drained + // earlier has published its marks (both sequentially consistent). + auto raced = is_nonzero(writers_count_()); + for (auto word = begin; (word <= end) && !raced; ++word) + raced = !is_zero(bit_and(mask(word), + bit_or(intent_[word].load(), dirty_[word].load()))); + + if (!raced) + mmap_unwire(std::next(memory_map_[zero], first * page_), + (second - first) * page_); + + if (raced || (mmap_settle( + std::next(memory_map_[zero], first * page_), + (second - first) * page_, opened_[zero], + first * page_) == fail)) + { + for (auto word = begin; word <= end; ++word) + released_[word].fetch_and(bit_not(mask(word))); + + if (!raced) + { + set_first_code(error::mmap_failure); + return false; + } + } + } + + return true; +} + +// Restore released pages overlapping [offset, offset+size) to writable +// anonymous memory (content preserved by atomic installation), before a +// declared write. Restoration is segmented at release_chunk alignment: the +// containing segment restores whole (unifying any fragmentation within it) +// but never more, as released runs consolidate without bound and restoring +// a maximal run copies gigabytes per scattered write (a restore convoy). +TEMPLATE +void CLASS::restore_(size_t offset, size_t size) NOEXCEPT +{ + using namespace system; + std::unique_lock restore_lock(restore_mutex_); + + // Segments clamp to full pages below logical (as does release candidacy): + // the reservation above commitment is inaccessible (installation reads). + const auto pages = to_width(logical_.load()) / page_; + const auto span = std::max(one, release_chunk / page_); + const auto last = (offset + sub1(size)) / page_; + auto page = offset / page_; + page -= (page % span); + + for (; (page <= last) && (page < pages); page += span) + { + const auto stop = std::min(page + span, pages); + const auto mask = [&](size_t word) NOEXCEPT + { + return page_mask(page, stop, word * page_bound); + }; + + const auto begin = page / page_bound; + const auto end = sub1(stop) / page_bound; + auto any = false; + for (auto word = begin; (word <= end) && !any; ++word) + any = !is_zero(bit_and(released_[word].load(), mask(word))); + + if (!any) + continue; + + if (mmap_restore(std::next(memory_map_[zero], page * page_), + (stop - page) * page_) == fail) + { + set_first_code(error::mmap_failure); + return; + } + + for (auto word = begin; word <= end; ++word) + released_[word].fetch_and(bit_not(mask(word))); + } +} + +// head share transitions, synchronized with writers by the count. +// ---------------------------------------------------------------------------- +// private + +TEMPLATE +std::atomic& CLASS::writer_slot_() NOEXCEPT +{ + static std::atomic threads{}; + static const thread_local size_t slot = threads.fetch_add(one) % + writer_shards; + return writers_.at(slot).count; +} + +TEMPLATE +size_t CLASS::writers_count_() const NOEXCEPT +{ + size_t count{}; + for (const auto& shard: writers_) + count += shard.count.load(); + + return count; +} + +TEMPLATE +void CLASS::writers_reset_() NOEXCEPT +{ + for (auto& shard: writers_) + shard.count.store(zero); +} + +// Exclude head writers across a share transition: they write through raw +// pointers under no lock, so only the writer count can exclude them. +TEMPLATE +void CLASS::quiesce_() NOEXCEPT +{ + transition_.store(true); + while (!is_zero(writers_count_())) + std::this_thread::yield(); +} + +// Settle a quiescent managed head to a writable mapping of its file over the +// committed span (as a shared head loads): pages drop under pressure and +// writes dirty page cache for kernel writeback, so page tracking idles (it +// remains allocated, as writers read it unlocked). Exclusive remap excludes +// accessors and the transition excludes counted head writers (raw pointer +// writes hold no lock), so a mark count still at the drained count proves +// the file current (a write landing after the drain snapshot would otherwise +// be lost to the remap). +// The remap lock precedes the drain: a writer holding an accessor waits +// uncounted at the transition, so a drain that preceded the lock would return +// on its own and the lock would then wait on that writer forever. Every +// counted writer either holds no lock or took its accessor before its count, +// so under the lock the count drains. +TEMPLATE +bool CLASS::share_(size_t transferred) NOEXCEPT +{ + std::unique_lock map_lock(remap_mutex_); + quiesce_(); + + auto shared = false; + if (loaded_.load() && !fault_.load() && (marks_.load() == transferred)) + { + // The drain transfers below logical; the committed fill above it is + // content (a raised logical exposes it unwritten). + const auto logical = to_width(logical_.load()); + const auto span = to_width(capacity_.load()); + const auto persisted = (span <= logical) || pwrite_all(opened_[zero], + std::next(memory_map_[zero], logical), span - logical, logical); + + shared = persisted && (mmap_share(memory_map_[zero], span, + opened_[zero], zero) != fail); + + if (!persisted) + set_first_code(error::fsync_failure); + else if (!shared) + set_first_code(error::mmap_failure); +#if !defined(WITHOUT_MADVISE) + else if (!advise_(memory_map_[zero], span)) + set_first_code(error::madvise_failure); +#endif + } + + shared_.store(shared); + transition_.store(false); + return shared; +} + +// Return a settled head to the anonymous model without a bulk remap: every +// full page declares released (the shared mapping is droppable file content), +// so tracked writers restore segments to anonymous before writing, and the +// tail above the full-page floor restores here (as the load commits it), so +// no dirty page is ever transferred from the file mapping (a self-copy, which +// darwin serves as an uninterruptible wait). Cold pages remain mapped. +TEMPLATE +void CLASS::unshare_() NOEXCEPT +{ + std::unique_lock map_lock(remap_mutex_); + quiesce_(); + + const auto floor = page_floor(to_width(logical_.load())); + const auto ceiling = page_ceiling(to_width(capacity_.load())); + const auto restored = (ceiling <= floor) || (mmap_restore( + std::next(memory_map_[zero], floor), ceiling - floor) != fail); + + if (restored) + { + declare_released_(); + shared_.store(false); + } + else + { + set_first_code(error::mmap_failure); + } + + transition_.store(false); +} + +#endif // MANAGE_STAGING + +} // namespace database +} // namespace libbitcoin + +#endif diff --git a/include/bitcoin/database/impl/memory/mmap_native.ipp b/include/bitcoin/database/impl/memory/mmap_native.ipp index d1c35e358..15f0b5fbf 100644 --- a/include/bitcoin/database/impl/memory/mmap_native.ipp +++ b/include/bitcoin/database/impl/memory/mmap_native.ipp @@ -19,7 +19,6 @@ #ifndef LIBBITCOIN_DATABASE_MEMORY_MMAP_NATIVE_IPP #define LIBBITCOIN_DATABASE_MEMORY_MMAP_NATIVE_IPP -#include #include #include #include @@ -30,55 +29,160 @@ namespace libbitcoin { namespace database { +// native backend (file-backed mapping), not thread safe. +// ---------------------------------------------------------------------------- +// private + +// Never results in unmapped. TEMPLATE -void CLASS::scanner_start_() NOEXCEPT +template +bool CLASS::flush_(size_t rows) NOEXCEPT { - // Idempotent across instances (process-wide, same values each call). - const auto memory = system_memory(); - /* int */ ::working_floor((memory / 4) * 3, memory); + // unmap (and therefore msync) must be called before ftruncate. + // "To flush all the dirty pages plus the metadata for the file and ensure + // that they are physically written to disk..." + const auto size = to_width(rows); + const auto success = + (::msync(memory_map_[Column], size, MS_SYNC) != fail) + && (::fsync(opened_[Column]) != fail); + + if (!success) + set_first_code(error::fsync_failure); + + return success; +} - touched_ = zero; - unlocked_ = zero; - scanning_.store(true); - scanner_ = std::thread([this]() NOEXCEPT - { - scanner_run_(); - }); +// Mapping failure results in unmapped. +// Mapping has no effect on logical size, always maps max(logical, min) size. +TEMPLATE +template +bool CLASS::map_() NOEXCEPT +{ + // Cannot map empty file, and want minimum capacity, so expand as required. + // The classic mapping is file-backed, so commitment is provisioning. + // disk_full: space is set but no code is set with false return. + const auto size = to_provision(); + if (!resize_(size)) + return false; + + memory_map_[Column] = system::pointer_cast( + ::mmap(nullptr, to_width(size), PROT_READ | PROT_WRITE, + MAP_SHARED, opened_[Column], 0)); + + return finalize_(); } +// Always results in unmapped, trims to logical (can be zero). TEMPLATE -void CLASS::scanner_stop_() NOEXCEPT +template +bool CLASS::unmap_(size_t size) NOEXCEPT { - if (!scanning_.exchange(false)) - return; + const auto logical = to_width(logical_.load()); + + // Windows cannot resize a mapped file. + // msync requires the live mapping, ftruncate requires it gone. + const auto synced = + (::msync(memory_map_[Column], logical, MS_SYNC) != fail); + + // Order ensures release in case of sync failure. + const auto success = release_(size) && synced + && (::ftruncate(opened_[Column], logical) != fail) + && (::fsync(opened_[Column]) != fail); + + loaded_.store(false); + + if (!success) + set_first_code(error::munmap_failure); + + return success; +} + +// Remap failure results in unmapped. +// Remapping has no effect on logical size, sets map_/capacity_. +TEMPLATE +template +bool CLASS::remap_(size_t size, bool final) NOEXCEPT +{ + BC_ASSERT(size >= logical_.load()); + + // Cannot remap empty file, so expand to minimum capacity if zero. + if (is_zero(size)) + size = minimum_; + + if (!resize_(size, final)) + return false; + + // mman-win32 mremap hack (umap/map) requires flags and file descriptor. + memory_map_[Column] = system::pointer_cast( + ::mremap_(memory_map_[Column], to_width(capacity_.load()), + to_width(size), PROT_READ | PROT_WRITE, MAP_SHARED, + opened_[Column])); - scanner_cv_.notify_all(); - if (scanner_.joinable()) - scanner_.join(); + return finalize_(); } +// Always results in unmapped, file is unchanged. +TEMPLATE +template +bool CLASS::release_(size_t size) NOEXCEPT +{ + const auto success = + ::munmap(memory_map_[Column], to_width(size)) != fail; + + if (!success) + set_first_code(error::munmap_failure); + + // loaded_ is caller-owned: unmap_ publishes unloaded, remap_ remains + // loaded across replacement (lock-free allocate guards must not observe + // a transient unload). + memory_map_[Column] = {}; + return success; +} + +// Finalize failure results in unmapped. +TEMPLATE +template +bool CLASS::finalize_() NOEXCEPT +{ + if (memory_map_[Column] == MAP_FAILED) + { + loaded_.store(false); + memory_map_[Column] = {}; + + // mmap or mremap failure (not mapped). + set_first_code(error::mmap_failure); + return false; + } + + loaded_.store(true); + return true; +} + +// working set scan (worker thread). +// ---------------------------------------------------------------------------- +// private + // Heads assert residency, bodies lead the trim. Both are advisory: failure // alters nothing and is not a fault (unlike the staged eviction primitives, // which own persistence). TEMPLATE -void CLASS::scanner_run_() NOEXCEPT +void CLASS::scan_run_() NOEXCEPT { using namespace system; const auto page = page_size(); if (is_zero(page)) return; + // Idempotent across instances (process-wide, same values each call). + const auto memory = system_memory(); + /* int */ ::working_floor((memory / 4) * 3, memory); + const auto span = std::max(one, evict_chunk / stride); + size_t touched{}; + size_t unlocked{}; - while (scanning_.load()) + while (tick_()) { - std::unique_lock scanner_lock(scanner_mutex_); - scanner_cv_.wait_for(scanner_lock, std::chrono::seconds(1)); - scanner_lock.unlock(); - - if (!scanning_.load()) - return; - // Steering is only useful against a trim, and a trim only happens // under contention, so at plenty the tick is a counter test. This // also bounds the idle cost to the wait itself. @@ -95,7 +199,7 @@ void CLASS::scanner_run_() NOEXCEPT // in an append-only body, and re-read probability decays with // it), so the trim takes these before the head set. const auto rows = logical_.load(); - const auto from = (unlocked_ < rows) ? unlocked_ : zero; + const auto from = (unlocked < rows) ? unlocked : zero; const auto to = std::min(rows, ceilinged_add(from, span)); if (to <= from) continue; @@ -104,7 +208,7 @@ void CLASS::scanner_run_() NOEXCEPT /* bool */ ::munlock(std::next(memory_map_[zero], at), to_width(to) - at); - unlocked_ = (to == rows) ? zero : to; + unlocked = (to == rows) ? zero : to; } else { @@ -117,17 +221,17 @@ void CLASS::scanner_run_() NOEXCEPT while (!is_zero(pages) && !is_zero(budget)) { - if (touched_ >= pages) - touched_ = zero; + if (touched >= pages) + touched = zero; - const auto at = touched_ * page; + const auto at = touched * page; const auto count = std::min( - { touch_span, pages - touched_, budget }); + { touch_span, pages - touched, budget }); for (size_t index{}; index < count; ++index) (void)map[at + index * page]; - touched_ += count; + touched += count; budget = floored_subtract(budget, count); } } diff --git a/include/bitcoin/database/impl/memory/mmap_private.ipp b/include/bitcoin/database/impl/memory/mmap_private.ipp index a3627e275..70f6bb1bc 100644 --- a/include/bitcoin/database/impl/memory/mmap_private.ipp +++ b/include/bitcoin/database/impl/memory/mmap_private.ipp @@ -19,7 +19,9 @@ #ifndef LIBBITCOIN_DATABASE_MEMORY_MMAP_PRIVATE_IPP #define LIBBITCOIN_DATABASE_MEMORY_MMAP_PRIVATE_IPP +#include #include +#include #include #include #include @@ -29,7 +31,7 @@ namespace libbitcoin { namespace database { -// mman dispatch, not thread safe. +// column dispatch, not thread safe. // ---------------------------------------------------------------------------- // private @@ -99,6 +101,7 @@ bool CLASS::unmap_all_(std::index_sequence) NOEXCEPT sweep_.reset(); words_ = zero; engaged_.store(false); + lazy_.store(false); shared_.store(false); #endif @@ -144,196 +147,61 @@ bool CLASS::remap_all_(size_t capacity, std::index_sequence, return true; } -// mman wrappers, not thread safe. -// ---------------------------------------------------------------------------- -// private - -// Never results in unmapped. -TEMPLATE -template -bool CLASS::flush_(size_t - #if defined(MANAGE_STAGING) || defined(HAVE_MSC) - rows - #endif -) NOEXCEPT -{ -#if defined(MANAGE_STAGING) - const auto success = persist_(to_width(rows)) - && sync_(); -#elif defined(HAVE_MSC) - // unmap (and therefore msync) must be called before ftruncate. - // "To flush all the dirty pages plus the metadata for the file and ensure - // that they are physically written to disk..." - const auto size = to_width(rows); - const auto success = - (::msync(memory_map_[Column], size, MS_SYNC) != fail) - && (::fsync(opened_[Column]) != fail); -#else - // msync should not be required on modern linux, see linus et al. - // stackoverflow.com/questions/5902629/mmap-msync-and-linux-process-termination - // Linux: fsync "transfers ("flushes") all modified in-core data of - // (i.e., modified buffer cache pages for) the file referred to by the - // file descriptor fd to the disk device so all changed information - // can be retrieved even if the system crashes or is rebooted. This - // includes writing through or flushing a disk cache if present. The - // call blocks until the device reports that transfer has completed." - const auto success = ::fsync(opened_[Column]) != fail; -#endif - - if (!success) - set_first_code(error::fsync_failure); - - return success; -} - -#if defined(MANAGE_STAGING) -// Persist rows below to: settled rows are already on disk (staged appends -// the remainder), a shared head synchronizes its mapping (writes through), -// an anonymous head transfers its dirty pages. -TEMPLATE -template -bool CLASS::persist_(size_t to) NOEXCEPT -{ - const auto from = to_width(settled_.load()); - return staged_ ? ((from >= to) || pwrite_all(opened_[Column], - std::next(memory_map_[Column], from), to - from, from)) : - (head_shared || shared_.load()) ? - (::msync(memory_map_[Column], to, MS_SYNC) != fail) : - transfer_(to); -} -#endif - -// Always results in unmapped, file is unchanged. -TEMPLATE -template -bool CLASS::release_(size_t size) NOEXCEPT -{ - const auto success = - ::munmap(memory_map_[Column], to_width(size)) != fail; - - if (!success) - set_first_code(error::munmap_failure); - - // loaded_ is caller-owned: unmap_ publishes unloaded, remap_ remains - // loaded across replacement (lock-free allocate guards must not observe - // a transient unload). - memory_map_[Column] = {}; - return success; -} - -// Always results in unmapped, trims to logical (can be zero). -TEMPLATE -template -bool CLASS::unmap_(size_t - #if !defined(MANAGE_STAGING) - size - #endif -) NOEXCEPT -{ - const auto logical = to_width(logical_.load()); - -#if defined(MANAGE_STAGING) - // Persist unflushed rows, trim preallocation to logical, sync to disk. - const auto transferred = persist_(logical) - && (::ftruncate(opened_[Column], logical) != fail) - && sync_(); - - // Order ensures release of the reservation in case of transfer failure. - const auto success = (::munmap(memory_map_[Column], - reserved_[Column]) != fail) && transferred; - - memory_map_[Column] = {}; - reserved_[Column] = zero; -#elif defined(HAVE_MSC) - // Windows cannot resize a mapped file. - // msync requires the live mapping, ftruncate requires it gone. - const auto synced = - (::msync(memory_map_[Column], logical, MS_SYNC) != fail); - - // Order ensures release in case of sync failure. - const auto success = release_(size) && synced - && (::ftruncate(opened_[Column], logical) != fail) - && (::fsync(opened_[Column]) != fail); -#else - // POSIX permits resizing a mapped file. - const auto truncated = - (::ftruncate(opened_[Column], logical) != fail) - && (::fsync(opened_[Column]) != fail); - - // Order ensures release in case of truncate failure. - const auto success = release_(size) && truncated; -#endif - - loaded_.store(false); - - if (!success) - set_first_code(error::munmap_failure); - - return success; -} - -// Mapping failure results in unmapped. -// Mapping has no effect on logical size, always maps max(logical, min) size. +// Iterated growth (callers hold the remap lock). Growth asks are amortized +// (rate surplus over the necessity), and each is admitted only while it +// leaves the configured headroom of the backing resource unclaimed (probed +// with the ask, released on grant), so exhaustion never consumes the +// system's final bytes. A large amortization step can be refused while the +// necessity fits, so iterate: halve the refused surplus toward the +// necessity. Refusal of the necessity is exhaustion, not store damage: +// published as disk full (space set, store intact, writes fail fast until +// cleared), it clears by settle drainage or operator relief, where teardown +// would convert a shortage into a restore. TEMPLATE -template -bool CLASS::map_() NOEXCEPT +bool CLASS::grow_(size_t end) NOEXCEPT { -#if defined(MANAGE_STAGING) - return stage_(); -#else - // Cannot map empty file, and want minimum capacity, so expand as required. - // The classic mapping is file-backed, so commitment is provisioning. - // disk_full: space is set but no code is set with false return. - const auto size = to_provision(); - if (!resize_(size)) + if (!is_zero(space_.load())) return false; - memory_map_[Column] = system::pointer_cast( - ::mmap(nullptr, to_width(size), PROT_READ | PROT_WRITE, - MAP_SHARED, opened_[Column], 0)); + using namespace system; + for (auto extended = to_growth(end); + !remap_all_(extended, sequence{}, false); + extended = ceilinged_add(end, + to_half(floored_subtract(extended, end)))) + { + if (fault_.load()) + return false; + + if (extended <= end) + { + set_disk_space(ceilinged_add(headroom_, ceilinged_multiply( + floored_subtract(end, capacity_.load()), stride))); + return false; + } + } - return finalize_(size); -#endif + return true; } -// Remap failure results in unmapped. -// Remapping has no effect on logical size, sets map_/capacity_. +// The wave probe reserves the whole extension plus headroom on the store +// volume (column widths sum to the stride) leaves the headroom unclaimed. +// An unmeasurable volume admits the wave: growth failure is the store's own +// disk full detection, and a query failure is not an exhaustion signal. TEMPLATE -template -bool CLASS::remap_(size_t size, bool final) NOEXCEPT +bool CLASS::probe_(size_t capacity) NOEXCEPT { - BC_ASSERT(size >= logical_.load()); - - // Cannot remap empty file, so expand to minimum capacity if zero. - if (is_zero(size)) - size = minimum_; - -#if defined(MANAGE_STAGING) - // The file is preallocated to capacity, preserving disk full detection at - // allocation, and growth commits reserved anonymous pages in place, so no - // mapping is released and the map base is stable within the reservation. - if (!resize_(size, final)) - return false; + using namespace system; + const auto bytes = ceilinged_multiply( + floored_subtract(capacity, file_.load()), stride); - return commit_(size, final); -#else - if (!resize_(size, final)) - return false; + if (is_zero(bytes) || is_zero(headroom_)) + return true; -#if defined(HAVE_MSC) - // mman-win32 mremap hack (umap/map) requires flags and file descriptor. - memory_map_[Column] = system::pointer_cast( - ::mremap_(memory_map_[Column], to_width(capacity_.load()), - to_width(size), PROT_READ | PROT_WRITE, MAP_SHARED, - opened_[Column])); -#else - memory_map_[Column] = system::pointer_cast( - ::mremap(memory_map_[Column], to_width(capacity_.load()), - to_width(size), MREMAP_MAYMOVE)); -#endif + size_t available{}; + if (!file::space(available, filenames_.front())) + return true; - return finalize_(size); -#endif // MANAGE_STAGING + return available >= ceilinged_add(bytes, headroom_); } // disk_full: space is set but no code is set with false return. @@ -379,95 +247,66 @@ bool CLASS::resize_(size_t size, bool final) NOEXCEPT return true; } -// The wave probe reserves the whole extension plus headroom on the store -// volume (column widths sum to the stride) leaves the headroom unclaimed. -// An unmeasurable volume admits the wave: growth failure is the store's own -// disk full detection, and a query failure is not an exhaustion signal. +// worker, instance-owned (load/unload lifecycle). +// ---------------------------------------------------------------------------- +// private + TEMPLATE -bool CLASS::probe_(size_t capacity) NOEXCEPT +void CLASS::worker_start_() NOEXCEPT { - using namespace system; - const auto bytes = ceilinged_multiply( - floored_subtract(capacity, file_.load()), stride); - - if (is_zero(bytes) || is_zero(headroom_)) - return true; +#if defined(MANAGE_STAGING) + // A shared head has no lazy writer (the kernel writes its mapping back). + if (!staged_ && head_shared) + return; - size_t available{}; - if (!file::space(available, filenames_.front())) - return true; + limit_ = system_memory() / throttle_factor; +#endif - return available >= ceilinged_add(bytes, headroom_); + working_.store(true); + worker_ = std::thread([this]() NOEXCEPT + { + worker_run_(); + }); } -// Finalize failure results in unmapped. TEMPLATE -template -bool CLASS::finalize_(size_t - #if !defined(HAVE_MSC) && !defined(WITHOUT_MADVISE) - size - #endif -) NOEXCEPT +void CLASS::worker_stop_() NOEXCEPT { - if (memory_map_[Column] == MAP_FAILED) - { - loaded_.store(false); - memory_map_[Column] = {}; - - // mmap or mremap failure (not mapped). - set_first_code(error::mmap_failure); - return false; - } + if (!working_.exchange(false)) + return; -#if !defined(HAVE_MSC) && !defined(WITHOUT_MADVISE) - // Get page size (usually 4KB). - using namespace system; - const int page_size = ::sysconf(_SC_PAGESIZE); - const auto page = possible_narrow_sign_cast(page_size); +#if defined(MANAGE_STAGING) + signal_(); +#endif - // If not one bit then page size is not a power of two as required. - if (page_size == fail || !is_one(ones_count(page))) { - set_first_code(error::sysconf_failure); - unmap_(size); - return false; + std::unique_lock worker_lock(worker_mutex_); + worker_cv_.notify_all(); } - // Align mapped bytes up to page boundary. - const auto max = sub1(page); - const auto target = to_width(size); - const auto align = bit_and(ceilinged_add(target, max), bit_not(max)); - - // Advice is elective (normal is the kernel default) and configured from - // the read pattern (see database::advice), as advising from the write - // pattern (structural) invites fault read amplification on random reads. - // Random access preloads (small heads, avoiding initial fault stalls). - if (access_ != advice::normal) - { - const auto random = (access_ != advice::sequential); - const auto preload = (access_ == advice::random); - const auto behavior = random ? MADV_RANDOM : MADV_SEQUENTIAL; - - for (size_t offset{}; offset < align; offset += advise_chunk) - { - const auto length = std::min(advise_chunk, align - offset); - const auto start = std::next(memory_map_[Column], offset); - - if (::madvise(start, length, behavior) == fail || (preload && - ::madvise(start, length, MADV_WILLNEED) == fail)) - { - set_first_code(error::madvise_failure); - unmap_(size); - return false; - } - } - } -#endif // !HAVE_MSC && !WITHOUT_MADVISE + if (worker_.joinable()) + worker_.join(); +} - loaded_.store(true); - return true; +TEMPLATE +void CLASS::worker_run_() NOEXCEPT +{ +#if defined(MANAGE_STAGING) + staged_ ? body_run_() : head_run_(); +#else + scan_run_(); +#endif } +// One second tick, false when stopped. +TEMPLATE +bool CLASS::tick_() NOEXCEPT +{ + std::unique_lock worker_lock(worker_mutex_); + worker_cv_.wait_for(worker_lock, std::chrono::seconds(1)); + worker_lock.unlock(); + return working_.load(); +} } // namespace database } // namespace libbitcoin diff --git a/include/bitcoin/database/impl/memory/mmap_staging.ipp b/include/bitcoin/database/impl/memory/mmap_staging.ipp index cf0885cf7..acb9905bb 100644 --- a/include/bitcoin/database/impl/memory/mmap_staging.ipp +++ b/include/bitcoin/database/impl/memory/mmap_staging.ipp @@ -19,314 +19,119 @@ #ifndef LIBBITCOIN_DATABASE_MEMORY_MMAP_STAGING_IPP #define LIBBITCOIN_DATABASE_MEMORY_MMAP_STAGING_IPP -#include #include -#if defined(STAGING_TELEMETRY) - #include - #include -#endif #include +#include #include -#include #include +#if defined(MANAGE_STAGING) + namespace libbitcoin { namespace database { -// Write-completion accounting (staged instances). Extents chase completions, -// closing the gaps, settling the completed prefix. +// staging backend (anonymous reservation), not thread safe. +// ---------------------------------------------------------------------------- +// private +// Never results in unmapped. TEMPLATE -void CLASS::complete(size_t STAGING_ONLY(offset), - size_t STAGING_ONLY(count)) NOEXCEPT +template +bool CLASS::flush_(size_t rows) NOEXCEPT { -#if defined(MANAGE_STAGING) - if (!staged_ || is_zero(count)) - return; - - // The covering extent is immobile while this completion is pending and - // cannot refuse a consistent claim, so a failed pass raced a recycle - // (torn navigation or stale snapshot): rescan against a fresh window. - for (;;) - { - // Acquire-ordered window snapshot (published entries are immobile; - // the packed word precludes observing a torn head/size pair). - const auto window = window_.load(std::memory_order_acquire); - const auto [head, size] = system::unpack_word(window); - - // Lock-free binary search of the sorted window (relaxed navigation - // is heuristic only; claim_ verifies cover under the generation). - auto high = size; - for (size_t low{}; low < high;) - { - const auto middle = to_half(low + high); - auto& record = ring_.at((head + middle) % extents); - const auto start = record.start.load(relaxed); - - if (offset < start) - { - high = middle; - continue; - } - - if (offset >= (start + record.count.load(relaxed))) - { - low = add1(middle); - continue; - } + const auto success = persist_(to_width(rows)) + && sync_(); - if (claim_(record, offset, count)) - return; - - // The matched slot recycled under the search: scan below. - break; - } + if (!success) + set_first_code(error::fsync_failure); - // Contention can record extents out of start order (allocation claim - // and recording are not one atomic step), transiently breaking search - // order. Contiguity guarantees a unique cover: scan linearly. - for (size_t index{}; index < size; ++index) - if (claim_(ring_.at((head + index) % extents), offset, count)) - return; - } -#endif + return success; } -#if defined(MANAGE_STAGING) - -// Claim completion of count rows against the extent, by cas decrement of the -// packed state under a cover check ordered by that state's own acquire: the -// publishing release sequences start/count before state, so both read after -// it belong to the observed generation (a live outstanding precludes a -// recycle in progress, as slots re-record only after retirement). A range -// match read before the acquire can be torn across a recycle and debit the -// slot's successor extent, pinning the true cover's frontier forever. False -// implies the slot does not (or no longer does) cover the offset, or is -// drained (recycle in flight): the caller rescans. The true covering extent -// cannot refuse: its generation is stable and its outstanding covers every -// pending completion while any remains. +// Persist rows below to: settled rows are already on disk (staged appends +// the remainder), a shared head synchronizes its mapping (writes through), +// an anonymous head transfers its dirty pages. TEMPLATE -bool CLASS::claim_(extent& record, size_t offset, size_t count) NOEXCEPT +template +bool CLASS::persist_(size_t to) NOEXCEPT { - using namespace system; - auto state = record.state.load(std::memory_order_acquire); - - for (;;) - { - const auto generation = shift_right(state, generation_shift); - const auto outstanding = bit_and(state, outstanding_mask); - const auto start = record.start.load(relaxed); - - // An outstanding below the claim is a recycle in flight or a drained - // slot (never the covering extent), refused rather than wrapped. - if ((outstanding < count) || (offset < start) || - (offset >= (start + record.count.load(relaxed)))) - return false; - - if (record.state.compare_exchange_weak(state, - pack_extent_(generation, outstanding - count), - std::memory_order_acq_rel, std::memory_order_acquire)) - { - // Extent completion is allocation-coarse, so maintenance is - // cheap here and the element write fast path takes no lock. - if (outstanding == count) - { - std::unique_lock extent_lock(extent_mutex_, std::try_to_lock); - - if (extent_lock.owns_lock()) - maintain_(); - } - - return true; - } - - // The failed cas reloaded state (acquire): recheck under it. - } + const auto from = to_width(settled_.load()); + return staged_ ? ((from >= to) || pwrite_all(opened_[Column], + std::next(memory_map_[Column], from), to - from, from)) : + (head_shared || shared_.load()) ? + (::msync(memory_map_[Column], to, MS_SYNC) != fail) : + transfer_(to); } -#endif // MANAGE_STAGING - +// Durability barrier for the column file. TEMPLATE -size_t CLASS::frontier() const NOEXCEPT +template +bool CLASS::sync_() NOEXCEPT { -#if defined(MANAGE_STAGING) - if (staged_) - return frontier_.load(); +#if defined(F_FULLFSYNC) + // non-standard macOS behavior: news.ycombinator.com/item?id=30372218 + return ::fcntl(opened_[Column], F_FULLFSYNC, 0) != fail; +#else + return ::fsync(opened_[Column]) != fail; #endif - - return size(); } -#if defined(MANAGE_STAGING) - -// Claim and record an extent under one lock: a claim never exists outside -// the ring and the ring is start-ordered, so the frontier can never pass an -// unwritten extent (claim-then-record raced the frontier past the claim). -// Returns eof (unclaimed) on insufficient capacity, fault, or disk full. +// Mapping failure results in unmapped. +// Mapping has no effect on logical size, always maps max(logical, min) size. TEMPLATE -size_t CLASS::record_(size_t count) NOEXCEPT -{ - std::unique_lock extent_lock(extent_mutex_); - - if (is_zero(count)) - return logical_.load(); - - maintain_(); - - using namespace system; - auto [head, size] = unpack_word(window_.load(relaxed)); - - // A full ring waits on completions (extents are allocation-coarse, so - // saturation implies extreme concurrency). Completions are lock-free, so - // waiting needs only this thread's own maintenance; fault or disk full - // releases the wait unclaimed (the write then fails fast). - while (size == extents) - { - if (fault_.load() || !is_zero(space_.load())) - return storage::eof; - - std::this_thread::yield(); - maintain_(); - std::tie(head, size) = unpack_word(window_.load(relaxed)); - } - - const auto start = logical_.load(); - if (is_add_overflow(start, count) || - ((start + count) > capacity_.load())) - return storage::eof; - - auto& record = ring_.at((head + size) % extents); - const auto generation = bit_and(add1(shift_right( - record.state.load(relaxed), generation_shift)), generation_mask); - BC_ASSERT((count * columns) <= outstanding_mask); - record.start.store(start, relaxed); - record.count.store(count, relaxed); - - // Publish the state (pairs with the acquire in claim_): a claim against - // the stale generation now fails its cas, and one against this - // generation observes the new start and count through the release. - record.state.store(pack_extent_(generation, count * columns), - std::memory_order_release); - - // Publish the extent (pairs with the acquire window snapshot). - window_.store(pack_word(head, add1(size)), release); - if (is_zero(size)) - frontier_.store(start); - - logical_.store(start + count); - check_invariants_(); - return start; -} - -// Pop completed extents from the head, advancing the frontier (locked). -TEMPLATE -void CLASS::maintain_() NOEXCEPT -{ - using namespace system; - auto [head, size] = unpack_word(window_.load(relaxed)); - while (!is_zero(size) && is_zero(bit_and( - ring_.at(head).state.load(relaxed), outstanding_mask))) - { - head = add1(head) % extents; - --size; - } - - window_.store(pack_word(head, size), release); - frontier_.store(is_zero(size) ? logical_.load() : - ring_.at(head).start.load(relaxed)); - check_invariants_(); -} - -// Discard all extents, requires quiescent writers (locked). Any extent then -// outstanding is an abandoned write (unreferenced), safe to settle as is. -TEMPLATE -void CLASS::discard_() NOEXCEPT +template +bool CLASS::map_() NOEXCEPT { - if (!staged_) - return; - - std::unique_lock extent_lock(extent_mutex_); - - window_.store(zero, release); - frontier_.store(logical_.load()); - check_invariants_(); + return stage_(); } -// Delay the caller while staging exceeds its memory bound (write throttle). -// Signaled as the settler drains; a fault or settler stop releases the caller -// into the normal allocation path (which fails on fault/unload). The relieved -// state is lock-free for the common (unparked) case; notifiers synchronize on -// throttle_mutex_ so a parked caller cannot miss its signal. +// Always results in unmapped, trims to logical (can be zero). TEMPLATE -void CLASS::throttle_() NOEXCEPT +template +bool CLASS::unmap_(size_t) NOEXCEPT { - const auto relieved = [this]() NOEXCEPT - { - using namespace system; - const auto rows = floored_subtract(logical_.load(), settled_.load()); - const auto debt = ceilinged_multiply(rows, stride); - return (debt <= limit_) || !settling_.load() || fault_.load(); - }; + const auto logical = to_width(logical_.load()); - if (!staged_ || relieved()) - return; + // Persist unflushed rows, trim preallocation to logical, sync to disk. + const auto transferred = persist_(logical) + && (::ftruncate(opened_[Column], logical) != fail) + && sync_(); - std::unique_lock throttle_lock(throttle_mutex_); + // Order ensures release of the reservation in case of transfer failure. + const auto success = (::munmap(memory_map_[Column], + reserved_[Column]) != fail) && transferred; - throttle_cv_.wait(throttle_lock, relieved); -} + memory_map_[Column] = {}; + reserved_[Column] = zero; + loaded_.store(false); -// Wake throttled callers, synchronized so a parking caller cannot miss the -// signal between its relieved check and its wait. -TEMPLATE -void CLASS::signal_() NOEXCEPT -{ - std::unique_lock throttle_lock(throttle_mutex_); + if (!success) + set_first_code(error::munmap_failure); - throttle_cv_.notify_all(); + return success; } -// staging dispatch, not thread safe. -// ---------------------------------------------------------------------------- -// private - +// Remap failure results in unmapped. +// Remapping has no effect on logical size, sets map_/capacity_. TEMPLATE -template -bool CLASS::settle_all_(size_t rows, std::index_sequence) NOEXCEPT +template +bool CLASS::remap_(size_t size, bool final) NOEXCEPT { - const auto from = settled_.load(); - if (!(settle_(from, rows) && ...)) - return false; + BC_ASSERT(size >= logical_.load()); - settled_.store(rows); - signal_(); - check_invariants_(); - return true; -} + // Cannot remap empty file, so expand to minimum capacity if zero. + if (is_zero(size)) + size = minimum_; -TEMPLATE -template -bool CLASS::unsettle_all_(size_t rows, std::index_sequence) NOEXCEPT -{ - if (!(unsettle_(rows) && ...)) + // The file is preallocated to capacity, preserving disk full detection at + // allocation, and growth commits reserved anonymous pages in place, so no + // mapping is released and the map base is stable within the reservation. + if (!resize_(size, final)) return false; - settled_.store(rows); - return true; -} - -TEMPLATE -template -bool CLASS::evict_all_(size_t from, size_t to, - std::index_sequence) NOEXCEPT -{ - return (evict_(from, to) && ...); + return commit_(size, final); } -// staging wrappers, not thread safe. -// ---------------------------------------------------------------------------- -// private - // Stage failure results in unmapped. // Staging has no effect on logical size, commits max(logical, min) capacity. TEMPLATE @@ -341,7 +146,21 @@ bool CLASS::stage_() NOEXCEPT if (!resize_(provision)) return false; - const auto size = to_commitment(); + // Managed heads load released: the file holds the content and pages + // restore to anonymous per segment on first write, so load contributes + // no residency (an eager head population is the store's largest + // memory transient, the full head set at open). + if (managed_) + { + if (!lazy_install_()) + { + teardown_(error::mmap_failure); + return false; + } + + loaded_.store(true); + return true; + } // Reserve address space with generous multiple of capacity (costless), so // that commitment growth never migrates the mapping (base is stable). @@ -359,24 +178,7 @@ bool CLASS::stage_() NOEXCEPT memory_map_[Column] = pointer_cast(base); reserved_[Column] = reserved; - // Page-dirty tracking for unstaged (rewrite-in-place head) instances, - // sized to the reservation (value-initialized to clean). Single column - // only (byte-offset marks); aggregates transfer in full. Shared heads - // write through their mapping, so they track and transfer nothing. - if constexpr (is_zero(Column) && is_one(columns)) - { - if (!staged_ && !head_shared) - { - const auto pages = ceilinged_divide(reserved, page_); - words_ = ceilinged_divide(pages, page_bound); - dirty_ = std::make_unique(words_); - intent_ = std::make_unique(words_); - released_ = std::make_unique(words_); - sweep_ = std::make_unique(words_); - writers_reset_(); - } - } - + const auto size = to_commitment(); const auto target = to_width(size); // A shared head maps its file in place of commitment (content is the @@ -403,22 +205,6 @@ bool CLASS::stage_() NOEXCEPT return true; } - // Managed heads load released: the file holds the content and pages - // restore to anonymous per segment on first write, so load contributes - // no residency (an eager head population is the store's largest - // memory transient, the full head set at open). - if (!staged_ && dirty_) - { - if (!lazy_install_()) - { - teardown_(error::mmap_failure); - return false; - } - - loaded_.store(true); - return true; - } - // Commit anonymous pages above the settle boundary page floor. const auto settled = page_floor(to_width(settled_.load())); @@ -466,129 +252,6 @@ bool CLASS::stage_() NOEXCEPT return true; } -// Install the managed head lazily over its current logical span: a released -// (read-only file) full-page prefix restored to anonymous per segment on -// first write, an anonymous populated tail, and clean page tracking sized to -// a replacement reservation. Caller holds exclusive remap (or is loading). -TEMPLATE -bool CLASS::lazy_install_() NOEXCEPT -{ - using namespace system; - const auto rows = logical_.load(); - const auto logical = to_width(rows); - - // Replace any standing reservation (sized as stage_). - if (!is_null(memory_map_[zero])) - mmap_unreserve(memory_map_[zero], reserved_[zero]); - - const auto reserved = page_ceiling(to_width( - to_reservation(to_provision()))); - const auto base = mmap_reserve(reserved); - if (base == MAP_FAILED) - { - set_first_code(error::mmap_failure); - return false; - } - - memory_map_[zero] = pointer_cast(base); - reserved_[zero] = reserved; - - // Rebuild page tracking for the new reservation (value-initialized). - const auto pages = ceilinged_divide(reserved, page_); - words_ = ceilinged_divide(pages, page_bound); - dirty_ = std::make_unique(words_); - intent_ = std::make_unique(words_); - released_ = std::make_unique(words_); - sweep_ = std::make_unique(words_); - writers_reset_(); - - // Released file prefix (full pages below logical). - const auto floor = page_floor(logical); - if (is_nonzero(floor) && (mmap_settle(memory_map_[zero], floor, - opened_[zero], zero) == fail)) - { - set_first_code(error::mmap_failure); - return false; - } - - // Commit the remainder to the commitment target and populate the tail. - const auto target = std::max(page_ceiling(to_width( - to_commitment())), page_ceiling(logical)); - if ((target > floor) && (mmap_commit(std::next(memory_map_[zero], floor), - target - floor, headroom_) == fail)) - { - set_first_code(error::mmap_failure); - return false; - } - - if ((logical > floor) && !pread_all(opened_[zero], - std::next(memory_map_[zero], floor), logical - floor, floor)) - { - set_first_code(error::fsync_failure); - return false; - } - - if (target > floor) - mmap_wire(std::next(memory_map_[zero], floor), target - floor); - - declare_released_(); - - // Attribute the anonymous span for diagnostics (smaps decomposition). - mmap_name(std::next(memory_map_[zero], floor), reserved - floor, - filenames_.front().filename().string().c_str()); - - return true; -} - -// Head-creation fill: the fill is file content, written sequentially to the -// file (page cache, no mapping involvement) and installed released, so -// creation contributes no memory residency (an in-memory backfill of the -// head set is the store's largest allocation transient). -TEMPLATE -size_t CLASS::allocate_filled_(size_t count, uint8_t backfill) NOEXCEPT -{ - BC_ASSERT(!staged_ && dirty_ && !shared_.load()); - std::unique_lock field_lock(field_mutex_); - std::unique_lock map_lock(remap_mutex_); - - using namespace system; - const auto start = logical_.load(); - if (!loaded_.load() || fault_.load() || is_add_overflow(start, count)) - return storage::eof; - - // Provision the file physically (disk full detected here). - const auto end = start + count; - if (!resize_(end)) - return storage::eof; - - // Write the fill. - const auto from = to_width(start); - const auto to = to_width(end); - std::vector chunk(std::min(to - from, release_chunk), backfill); - for (auto at = from; at < to; ) - { - const auto size = std::min(to - at, chunk.size()); - if (!pwrite_all(opened_[zero], chunk.data(), size, at)) - { - set_first_code(error::fsync_failure); - return storage::eof; - } - - at += size; - } - - // The fill bounds capacity (installation populates only the fill). - BC_ASSERT(capacity_.load() <= end); - logical_.store(end); - file_.store(std::max(file_.load(), end)); - capacity_.store(end); - if (!lazy_install_()) - return storage::eof; - - check_invariants_(); - return start; -} - // Commit failure results in unmapped when final (the default); a non-final // refusal (in-reservation commit, replacement reservation, or replacement // commit) returns false with the standing mapping untouched, so the caller @@ -637,7 +300,7 @@ bool CLASS::commit_(size_t size, bool final) NOEXCEPT return false; } - if (!staged_ && dirty_ && (target > from)) + if (managed_ && (target > from)) mmap_wire(std::next(memory_map_[Column], from), target - from); // Committed growth is a new (unnamed) vma; reattribute it. @@ -713,7 +376,7 @@ bool CLASS::commit_(size_t size, bool final) NOEXCEPT std::copy_n(std::next(memory_map_[Column], settled), logical - settled, std::next(base, settled)); - if (!staged_ && dirty_) + if (managed_) mmap_wire(std::next(base, settled), target - settled); // Convert the settled prefix on the replacement reservation. @@ -736,31 +399,27 @@ bool CLASS::commit_(size_t size, bool final) NOEXCEPT // Regrow dirty tracking with the reservation (marks are excluded by the // remap lock, as unstaged mutations hold protected accessors). - if constexpr (is_zero(Column)) + if (managed_) { - if (!staged_ && dirty_) - { - const auto pages = ceilinged_divide(reserved, page_); - const auto words = ceilinged_divide(pages, page_bound); - auto grown = std::make_unique(words); - - for (size_t word{}; word < std::min(words_, words); ++word) - grown[word].store(dirty_[word].load(relaxed), - relaxed); - - dirty_ = std::move(grown); - words_ = words; - - // The replacement reservation is fully anonymous (content was - // copied above), so released and intent page state resets. - intent_ = std::make_unique(words); - released_ = std::make_unique(words); - sweep_ = std::make_unique(words); - } + const auto pages = ceilinged_divide(reserved, page_); + const auto words = ceilinged_divide(pages, page_bound); + auto grown = std::make_unique(words); + + for (size_t word{}; word < std::min(words_, words); ++word) + grown[word].store(dirty_[word].load(relaxed), relaxed); + + dirty_ = std::move(grown); + words_ = words; + + // The replacement reservation is fully anonymous (content was + // copied above), so released and intent page state resets. + intent_ = std::make_unique(words); + released_ = std::make_unique(words); + sweep_ = std::make_unique(words); } // Release the exhausted reservation and adopt the replacement. - const auto released = ::munmap(memory_map_[Column], reserved_[Column]) + const auto released = ::munmap(memory_map_[Column], reserved_[Column]) != fail; memory_map_[Column] = base; @@ -775,107 +434,6 @@ bool CLASS::commit_(size_t size, bool final) NOEXCEPT return true; } -// Convert flushed rows [from, to) to a read-only shared file mapping, page -// floored so the settle boundary page remains anonymous with its settled -// bytes retained. Releases the covered anonymous pages. Failure results in -// unmapped. -TEMPLATE -template -bool CLASS::settle_(size_t from, size_t to) NOEXCEPT -{ - if (!staged_) - return true; - - const auto begin = page_floor(to_width(from)); - const auto end = page_floor(to_width(to)); - if (begin == end) - return true; - - const auto address = std::next(memory_map_[Column], begin); - if (mmap_settle(address, end - begin, opened_[Column], begin) == fail) - { - teardown_(error::mmap_failure); - return false; - } - -#if !defined(WITHOUT_MADVISE) - if (!advise_(address, end - begin)) - { - teardown_(error::madvise_failure); - return false; - } - - // Demote the settled extent to reclaim-first order: cached for imminent - // re-read (validation follows archival) but reclaimed under pressure - // before active pages and anonymous heads (head residency priority). - // Ample hosts skip demotion: with nothing contesting memory it only - // converts warm cache into re-read faults (see demote_memory). - static const auto ample = system_memory() > demote_memory; - if (!ample && (mmap_cold(address, end - begin) == fail)) - { - teardown_(error::madvise_failure); - return false; - } -#endif - - return true; -} - -// Revert settled pages at/above rows to committed anonymous memory and -// restore the retained bytes below rows from the file (truncation below the -// settle boundary). Failure results in unmapped. -TEMPLATE -template -bool CLASS::unsettle_(size_t rows) NOEXCEPT -{ - const auto bytes = to_width(rows); - const auto begin = page_floor(bytes); - const auto end = page_floor(to_width(settled_.load())); - if (begin == end) - return true; - - const auto address = std::next(memory_map_[Column], begin); - if (mmap_unsettle(address, end - begin) == fail) - { - teardown_(error::mmap_failure); - return false; - } - - if ((begin < bytes) && !pread_all(opened_[Column], address, bytes - begin, - begin)) - { - teardown_(error::fsync_failure); - return false; - } - - return true; -} - -// Release cached pages of settled rows [from, to), page floored within the -// converted read-only mapping (its pages cannot be dirtied, and any pending -// settle write-back is synced by the release). Cost follows residency, so -// re-eviction is idempotent and free. The mapping is unaffected (a later -// read faults the page back from the file). Failure faults the store -// (advisory-class failures have no benign modes) but leaves it mapped, as -// the shared-lock caller cannot tear down under live accessors. -TEMPLATE -template -bool CLASS::evict_(size_t from, size_t to) NOEXCEPT -{ - const auto begin = page_floor(to_width(from)); - const auto end = page_floor(to_width(to)); - if (begin == end) - return true; - - if (mmap_evict(std::next(memory_map_[Column], begin), end - begin) == fail) - { - set_first_code(error::fsync_failure); - return false; - } - - return true; -} - // Teardown results in unmapped (release failure is not further reported). TEMPLATE template @@ -928,7 +486,8 @@ TEMPLATE size_t CLASS::to_reservation(size_t rows) const NOEXCEPT { using namespace system; - return ceilinged_multiply(to_capacity(std::max(rows, minimum_)), headroom); + return ceilinged_multiply(to_capacity(std::max(rows, minimum_)), + reserve_factor); } TEMPLATE @@ -945,824 +504,9 @@ size_t CLASS::page_ceiling(size_t bytes) const NOEXCEPT return page_floor(ceilinged_add(bytes, sub1(page_))); } -// settle scheduler, instance-owned (drains completed writes to clean cache). -// ---------------------------------------------------------------------------- -// private - -// Dirty marking without the writer uncount (no paired prepare), for internal -// mark restoration (mark() pairs with prepare() and uncounts its writer). -TEMPLATE -void CLASS::remark_(size_t offset, size_t size) NOEXCEPT -{ - auto page = offset / page_; - const auto end = (offset + sub1(size)) / page_; - while ((page <= end) && ((page / page_bound) < words_)) - { - const auto bit = system::bit_right(page % page_bound); - dirty_[page++ / page_bound].fetch_or(bit, relaxed); - } - - marks_.fetch_add(one, relaxed); -} - -// Transfer dirty pages within [0, bytes), clearing marks before content -// reads so that concurrently remarked pages transfer on the next pass. -// Marks beyond bytes are retained (backfill above logical transfers when -// logical grows over it). Adjacent dirty pages coalesce into single writes. -TEMPLATE -template -bool CLASS::transfer_(size_t bytes) NOEXCEPT -{ - using namespace system; - if (is_zero(bytes)) - return true; - - // One pass at a time: concurrent passes split the claimed dirty set. - std::unique_lock transfer_lock(transfer_mutex_); - - // Untracked (multi-column unstaged) instances transfer in full. - if (!dirty_) - return pwrite_all(opened_[Column], memory_map_[Column], bytes, zero); - - size_t from{}; - size_t to{}; - const auto write = [&]() NOEXCEPT - { - return (from >= to) || pwrite_all(opened_[Column], - std::next(memory_map_[Column], from), to - from, from); - }; - - const auto pages = ceilinged_divide(bytes, page_); - const auto bound = std::min(words_, ceilinged_divide(pages, page_bound)); - for (size_t word{}; word < bound; ++word) - { - auto bits = dirty_[word].exchange(zero, relaxed); - - // Retain marks at and above the page bound (boundary word only). - const auto first = word * page_bound; - if (pages < (first + page_bound)) - { - const auto keep = mask_right(pages - first); - dirty_[word].fetch_or(bit_and(bits, keep), relaxed); - bits = bit_and(bits, bit_not(keep)); - } - - for (size_t bit{}; !is_zero(bits) && (bit < page_bound); ++bit) - { - if (!get_right(bits, bit)) - continue; - - bits = set_right(bits, bit, false); - const auto start = (first + bit) * page_; - const auto end = std::min(start + page_, bytes); - if (start == to) - { - to = end; - continue; - } - - if (!write()) - { - // Restore claimed marks (the failed range, the page that - // ended it, and this word's unwritten remainder) so failure - // is retryable, not lossy. - remark_(from, to - from); - remark_(start, end - start); - dirty_[word].fetch_or(bits, relaxed); - return false; - } - - from = start; - to = end; - } - } - - if (!write()) - { - remark_(from, to - from); - return false; - } - - return true; -} - -// Durability barrier for the column file. -TEMPLATE -template -bool CLASS::sync_() NOEXCEPT -{ -#if defined(F_FULLFSYNC) - // non-standard macOS behavior: news.ycombinator.com/item?id=30372218 - return ::fcntl(opened_[Column], F_FULLFSYNC, 0) != fail; -#else - return ::fsync(opened_[Column]) != fail; -#endif -} - -TEMPLATE -void CLASS::settler_start_() NOEXCEPT -{ - // A shared head has no lazy writer (the kernel writes its mapping back). - if (!staged_ && head_shared) - return; - - limit_ = system_memory() / throttle_factor; - evicted_ = zero; - settling_.store(true); - settler_ = std::thread([this]() NOEXCEPT - { - staged_ ? settler_run_() : head_run_(); - }); -} - -TEMPLATE -void CLASS::settler_stop_() NOEXCEPT -{ - if (!settling_.exchange(false)) - return; - - signal_(); - settler_cv_.notify_all(); - if (settler_.joinable()) - settler_.join(); -} - -// Pressure-paced draining (the windows model): drain intensity follows memory -// conditions, settled pages remain cached, writers delay only at the staging -// memory bound (write throttle), sustained stillness drains the residual (the -// lazy writer) so a quiescent map converges to fully settled. -TEMPLATE -void CLASS::settler_run_() NOEXCEPT -{ - // Tiers derive from physical memory, pressure and compression occupancy - // (a pressure precursor: the kernel compresses without raising level). - const auto memory = system_memory(); - const auto urgent = memory / urgent_factor; - const auto active = memory / active_factor; - const auto squeeze = memory / compress_factor; - const auto scarce = memory / sweep_factor; - const auto chunk = std::max(one, settle_chunk / stride); - const auto sweep = std::max(one, evict_chunk / stride); - - // Ticks without allocation before idle draining (settling is not writing, - // so draining does not hold its own clock). - auto mark = logical_.load(); - size_t still{}; - - const auto backlog = [this]() NOEXCEPT - { - using namespace system; - return ceilinged_multiply(floored_subtract(frontier_.load(), - settled_.load()), stride); - }; - - while (settling_.load()) - { - std::unique_lock settler_lock(settler_mutex_); - settler_cv_.wait_for(settler_lock, std::chrono::seconds(1)); - settler_lock.unlock(); - - if (!settling_.load()) - return; - - const auto top = logical_.load(); - still = (top == mark) ? std::min(add1(still), idle_seconds) : zero; - mark = top; - -#if defined(STAGING_TELEMETRY) - // The settler drains to the frontier while the throttle measures debt - // to logical, so a pinned frontier stops settling while debt grows - // (unbounded). lag exposes the pin, debt the throttle pressure. One - // string per line, as settler threads share the stream. - if (is_zero(++telemetry_ % telemetry_seconds)) - { - using namespace system; - const auto settled = settled_.load(); - const auto frontier = frontier_.load(); - const auto [head_, size_] = unpack_word( - window_.load(relaxed)); - - std::ostringstream line{}; - line << "staging " << filenames_.front().filename().string() - << " logical=" << top - << " frontier=" << frontier - << " settled=" << settled - << " lag=" << floored_subtract(top, frontier) - << " debt=" << floored_subtract(top, settled) - << " window=" << size_ - << " head=" << head_ - << std::endl; - - std::cerr << line.str() << std::flush; - } -#endif - - // Scarcity is read directly: clean cache is reclaimable, so the - // kernel pressure level does not raise while free memory exhausts. - if (system_free() < scarce) - evict_next_(sweep); - - auto bytes = backlog(); - if (is_zero(bytes)) - continue; - - // Urgency drains continuously, activity/stillness one chunk per tick. - const auto driven = - (system_pressure() > one) || - (system_compressed() > squeeze) || - (bytes > urgent); - - if (!driven && - (bytes <= active) && - (still < idle_seconds)) - continue; - - do - { - if (!settle_next_(chunk)) - break; - - bytes = backlog(); - } - while (settling_.load() && driven && (bytes > active)); - } -} - -// Idle transfer for unstaged (rewrite-in-place head) instances: sustained -// mark stillness drains dirty pages (the lazy writer) so a quiescent map -// converges to persisted and snapshot/close transfer approximately nothing. -// Requires no write exclusion: marks follow writes and transfer clears -// before reading, so racing writes remark and transfer on the next pass -// (torn disk pages are unreachable, as live heads are only trusted -// following a clean close). -// -// Memory scarcity additionally pages the head to its own file (the windows -// model): transfer, then release cold clean pages to read-only file mappings -// (reclaimable cache), restored to anonymous by prepare() before any write. -// Release waits one pass after engagement so all writers declare intent. -TEMPLATE -void CLASS::head_run_() NOEXCEPT -{ - const auto scarce = system_memory() / evict_factor; - - // Ticks without a mark before idle draining. - auto mark = marks_.load(); - auto transferred = mark; - size_t still{}; - size_t hot{}; - size_t touched{}; - - while (settling_.load()) - { - std::unique_lock settler_lock(settler_mutex_); - settler_cv_.wait_for(settler_lock, std::chrono::seconds(1)); - settler_lock.unlock(); - - if (!settling_.load()) - return; - - const auto top = marks_.load(); - const auto writes = top - mark; - still = (top == mark) ? std::min(add1(still), idle_seconds) : zero; - mark = top; - -#if defined(STAGING_TELEMETRY) - // Per-minute head write profile: marks in the minute, the busiest - // tick within it, and the ticks that carried any write. - marked_ += writes; - peaked_ = std::max(peaked_, writes); - active_ += is_nonzero(writes) ? one : zero; - if (is_zero(++telemetry_ % telemetry_seconds)) - { - std::ostringstream line{}; - line << "head " << filenames_.front().filename().string() - << " marks=" << marked_ - << " peak=" << peaked_ - << " active=" << active_ - << " hot=" << hot - << " settled=" << shared_.load() - << std::endl; - std::cout << line.str(); - marked_ = zero; - peaked_ = zero; - active_ = zero; - } -#endif - - // A settled head under sustained writes reinstalls lazily. - if (shared_.load()) - { - hot = (writes >= unsettle_writes) ? add1(hot) : zero; - if (hot >= unsettle_seconds) - { - unshare_(); - transferred = top; - hot = zero; - } - - continue; - } - - // A drained head settles to its file mapping while the store is - // current (settled pages are droppable, anonymous pages are not). - if (current_.load() && (transferred == top) && share_(transferred)) - continue; - - // Touch pass: assert working-set residency at a bounded rate. - // Hash-uniform probing is per-page sparse in every phase, so the - // kernel ages head pages cold and swaps them under cache pressure, - // though the set is the process working set (and head misses are - // unshieldable serial faults on the probe path). A read sets the - // page-table accessed bit without dirtying the page; volatile - // prevents elision. The unprivileged equivalent of mlock, and soft: - // under true extremity the kernel can still take the pages. The - // per-tick budget revisits every page each touch_seconds regardless - // of instance size (a whole-instance lap gave the largest head a - // proportionally longer revisit, which lost the aging race first). - // Residency-guarded so the touch never faults: a swapped page that - // nothing probes rests in swap, one the workload needs returns by - // its own fault and is defended thereafter. - // Residency is only contested under scarcity, so at plenty the tick - // costs a counter test (the aging race has no other runner). - if (system_available() < (system_memory() / sweep_factor)) - { - using namespace system; - std::shared_lock touch_lock(remap_mutex_); - if (loaded_.load() && !fault_.load()) - { - const auto pages = to_width(logical_.load()) / page_; - auto budget = ceilinged_divide(pages, touch_seconds); -#if !defined(HAVE_APPLE) - // Probe buffer and map alias, declared with the touch they - // serve (excluded on darwin below), as they are otherwise - // unused there. - unsigned char resident[touch_span]; - const volatile auto* map = memory_map_[zero]; -#endif - while (!is_zero(pages) && !is_zero(budget)) - { - if (touched >= pages) - touched = zero; - - const auto count = std::min( - { touch_span, pages - touched, budget }); - -#if !defined(HAVE_APPLE) - // Darwin excluded: mincore hides compressed pages from - // the guard, and unguarded touching measured as pure - // decompression churn; release is the darwin mechanism. - const auto at = touched * page_; - if (mmap_resident(std::next(memory_map_[zero], at), - count * page_, resident) == 0) - for (size_t page{}; page < count; ++page) - if (is_odd(resident[page])) - (void)map[at + page * page_]; -#endif - - touched += count; - budget = floored_subtract(budget, count); - } - } - } - - // Available includes reclaimable file cache, which a loaded store - // keeps large while the kernel swaps cold anonymous pages, so free - // exhaustion also signals scarcity (anon is being displaced). - // A sync writes every head hot (a converted run restores at the next - // burst), so release engages only while the store is current. - const auto scarcity = head_release && dirty_ && current_.load() && - ((system_available() < scarce) || (system_free() < scarce)); - - // Once engaged, a quiet instance converts independent of momentary - // scarcity: the signal clears as swap absorbs the hot set, but the - // swapped pages remain anonymous, so reads fault them back one at a - // time (a serial swap-in per probe). Conversion instead settles - // clean pages without read-back (the file holds their content), - // freeing swap and routing reads through the file mapping. - const auto engaged = engaged_.load(); - const auto quiet = writes < release_quiet; - const auto draining = engaged && quiet; - if (!scarcity && !draining && - ((still < idle_seconds) || (transferred == top))) - continue; - - // A write-hot instance neither transfers nor releases under - // scarcity: transferred pages re-dirty immediately (write - // amplification without release payoff, as hash-scattered writes - // into released pages each cost a segment restore, and the sweep - // otherwise re-releases restored segments every pass). Its - // anonymous set is left to swap (dirty-exempt) until quiescence, - // typically the phase change. -#if !defined(HAVE_APPLE) - // EXPERIMENT: darwin releases while write-hot (page-level intent and - // dirty filters remain) to price segment restores against the - // measured compressed-hot crawl. - if (scarcity && !quiet) - continue; -#endif - - { - std::shared_lock map_lock(remap_mutex_); - - if (!loaded_.load() || fault_.load()) - continue; - - if constexpr (head_release) - if (scarcity && !engaged) - engaged_.store(true); - - // Scarcity passes pace writeback for release: settle maps page - // cache content, and durability remains a clean close property, - // so sync applies only to idle draining. - if (!transfer_(to_width(logical_.load())) || - (!scarcity && !sync_())) - { - set_first_code(error::fsync_failure); - continue; - } - - // Discard the page cache copy of the transfer: the anonymous - // head is the live copy, so caching the file doubles it. - file_discard(opened_[zero]); - - // Quiet is assured here (hot scarcity skipped above, and idle - // draining implies sixty still seconds). - // Lazy head load engages restore on all platforms; the release - // sweep remains a darwin response (see head_release). - if constexpr (head_release) - if (engaged && !release_pages_()) - continue; - } - - transferred = top; - still = zero; - } -} - -TEMPLATE -std::atomic& CLASS::writer_slot_() NOEXCEPT -{ - static std::atomic threads{}; - static const thread_local size_t slot = threads.fetch_add(one) % - writer_shards; - return writers_.at(slot).count; -} - -TEMPLATE -size_t CLASS::writers_count_() const NOEXCEPT -{ - size_t count{}; - for (const auto& shard: writers_) - count += shard.count.load(); - - return count; -} - -TEMPLATE -void CLASS::writers_reset_() NOEXCEPT -{ - for (auto& shard: writers_) - shard.count.store(zero); -} - -TEMPLATE -void CLASS::quiesce_() NOEXCEPT -{ - transition_.store(true); - while (!is_zero(writers_count_())) - std::this_thread::yield(); -} - -// Settle a quiescent managed head to a writable mapping of its file over the -// committed span (as a shared head loads): pages drop under pressure and -// writes dirty page cache for kernel writeback, so page tracking idles (it -// remains allocated, as writers read it unlocked). Exclusive remap excludes -// accessors and the transition excludes counted head writers (raw pointer -// writes hold no lock), so a mark count still at the drained count proves -// the file current (a write landing after the drain snapshot would otherwise -// be lost to the remap). -// The remap lock precedes the drain: a writer holding an accessor waits -// uncounted at the transition, so a drain that preceded the lock would return -// on its own and the lock would then wait on that writer forever. Every -// counted writer either holds no lock or took its accessor before its count, -// so under the lock the count drains. -TEMPLATE -bool CLASS::share_(size_t transferred) NOEXCEPT -{ - std::unique_lock map_lock(remap_mutex_); - quiesce_(); - - auto shared = false; - if (loaded_.load() && !fault_.load() && (marks_.load() == transferred)) - { - // The drain transfers below logical; the committed fill above it is - // content (a raised logical exposes it unwritten). - const auto logical = to_width(logical_.load()); - const auto span = to_width(capacity_.load()); - const auto persisted = (span <= logical) || pwrite_all(opened_[zero], - std::next(memory_map_[zero], logical), span - logical, logical); - - shared = persisted && (mmap_share(memory_map_[zero], span, - opened_[zero], zero) != fail); - - if (!persisted) - set_first_code(error::fsync_failure); - else if (!shared) - set_first_code(error::mmap_failure); -#if !defined(WITHOUT_MADVISE) - else if (!advise_(memory_map_[zero], span)) - set_first_code(error::madvise_failure); -#endif - } - - shared_.store(shared); - transition_.store(false); - return shared; -} - -// Return a settled head to the anonymous model without a bulk remap: every -// full page declares released (the shared mapping is droppable file content), -// so tracked writers restore segments to anonymous before writing, and the -// tail above the full-page floor restores here (as the load commits it), so -// no dirty page is ever transferred from the file mapping (a self-copy, which -// darwin serves as an uninterruptible wait). Cold pages remain mapped. -TEMPLATE -void CLASS::unshare_() NOEXCEPT -{ - std::unique_lock map_lock(remap_mutex_); - quiesce_(); - - const auto floor = page_floor(to_width(logical_.load())); - const auto ceiling = page_ceiling(to_width(capacity_.load())); - const auto restored = (ceiling <= floor) || (mmap_restore( - std::next(memory_map_[zero], floor), ceiling - floor) != fail); - - if (restored) - { - declare_released_(); - shared_.store(false); - } - else - { - set_first_code(error::mmap_failure); - } - - transition_.store(false); -} - -// Declare the full-page prefix below logical released and engage the restore -// protocol (prepare restores a released segment before its write). -TEMPLATE -void CLASS::declare_released_() NOEXCEPT -{ - using namespace system; - const auto flags = page_floor(to_width(logical_.load())) / page_; - for (size_t word{}; word < ceilinged_divide(flags, page_bound); ++word) - { - const auto first = word * page_bound; - released_[word].store((flags >= (first + page_bound)) ? - bit_all : unmask_right(flags - first)); - } - - lazy_.store(true); -} - -// Release cold clean head page runs to read-only file mappings (reclaimable), -// full pages below logical only. Conversion is run-granular (release_chunk -// minimum) as each conversion splits a mapping: page granularity fragments -// the address space beyond what host memory management tolerates. Writer -// synchronization is a per-page bit protocol: prepare() declares intent then -// loads released; release stores released then loads intent (both -// sequentially consistent), so a run converts only when no write can land on -// it unrestored. A wrong release costs one restore. Conversion and restore -// serialize on restore_mutex_. -TEMPLATE -bool CLASS::release_pages_() NOEXCEPT -{ - using namespace system; - const auto bytes = to_width(logical_.load()); - const auto pages = bytes / page_; - const auto bound = std::min(words_, ceilinged_divide(pages, page_bound)); - const auto chunk = std::max(one, release_chunk / page_); - - std::unique_lock restore_lock(restore_mutex_); - - // Materialize candidacy (hot aging clears only the snapshot bits, so a - // concurrent declaration on a candidate page is retained for the live - // rechecks below). - for (size_t word{}; word < bound; ++word) - { - const auto hot = intent_[word].load(); - intent_[word].fetch_and(bit_not(hot)); - sweep_[word] = release_below(release_candidates( - dirty_[word].load(relaxed), hot, released_[word].load(relaxed)), - pages, word * page_bound); - } - - // Convert maximal candidate runs of at least chunk pages, clamped to - // chunk bounds. Restore is chunk-aligned and chunk-granular, so a - // partially released chunk places unreleased pages inside a restored - // span. The bit protocol guards released pages only: a writer to an - // unreleased page observes released clear, so it neither restores nor - // takes restore_mutex_, and its write is dropped by the copy-install - // window. Whole chunks make the restored span exactly the released set. - for (auto run = next_run(sweep_.get(), pages, zero); run.first < pages; - run = next_run(sweep_.get(), pages, run.second)) - { - const auto first = ceilinged_divide(run.first, chunk) * chunk; - const auto second = (run.second / chunk) * chunk; - - if ((second <= first) || ((second - first) < chunk)) - continue; - - const auto begin = first / page_bound; - const auto end = sub1(second) / page_bound; - const auto mask = [&](size_t word) NOEXCEPT - { - return page_mask(first, second, word * page_bound); - }; - - // Declare the release (prepare() restores from this point). - for (auto word = begin; word <= end; ++word) - released_[word].fetch_or(mask(word)); - - // An in-flight writer (counted, unaged) or raced intent or mark - // invalidates the conversion (whole run). The count loads first: a - // writer counted later observes released and restores, one drained - // earlier has published its marks (both sequentially consistent). - auto raced = is_nonzero(writers_count_()); - for (auto word = begin; (word <= end) && !raced; ++word) - raced = !is_zero(bit_and(mask(word), - bit_or(intent_[word].load(), dirty_[word].load()))); - - if (!raced) - mmap_unwire(std::next(memory_map_[zero], first * page_), - (second - first) * page_); - - if (raced || (mmap_settle( - std::next(memory_map_[zero], first * page_), - (second - first) * page_, opened_[zero], - first * page_) == fail)) - { - for (auto word = begin; word <= end; ++word) - released_[word].fetch_and(bit_not(mask(word))); - - if (!raced) - { - set_first_code(error::mmap_failure); - return false; - } - } - } - - return true; -} - -// Restore released pages overlapping [offset, offset+size) to writable -// anonymous memory (content preserved by atomic installation), before a -// declared write. Restoration is segmented at release_chunk alignment: the -// containing segment restores whole (unifying any fragmentation within it) -// but never more, as released runs consolidate without bound and restoring -// a maximal run copies gigabytes per scattered write (a restore convoy). -TEMPLATE -void CLASS::restore_(size_t offset, size_t size) NOEXCEPT -{ - using namespace system; - std::unique_lock restore_lock(restore_mutex_); - - // Segments clamp to full pages below logical (as does release candidacy): - // the reservation above commitment is inaccessible (installation reads). - const auto pages = to_width(logical_.load()) / page_; - const auto span = std::max(one, release_chunk / page_); - const auto last = (offset + sub1(size)) / page_; - auto page = offset / page_; - page -= (page % span); - - for (; (page <= last) && (page < pages); page += span) - { - const auto stop = std::min(page + span, pages); - const auto mask = [&](size_t word) NOEXCEPT - { - return page_mask(page, stop, word * page_bound); - }; - - const auto begin = page / page_bound; - const auto end = sub1(stop) / page_bound; - auto any = false; - for (auto word = begin; (word <= end) && !any; ++word) - any = !is_zero(bit_and(released_[word].load(), mask(word))); - - if (!any) - continue; - - if (mmap_restore(std::next(memory_map_[zero], page * page_), - (stop - page) * page_) == fail) - { - set_first_code(error::mmap_failure); - return; - } - - for (auto word = begin; word <= end; ++word) - released_[word].fetch_and(bit_not(mask(word))); - } -} - -// Settle up to chunk completed rows: write under the shared remap lock -// (completed extents are immutable, writers proceed), convert under a brief -// exclusive. Durability remains a snapshot property (no sync here). -TEMPLATE -bool CLASS::settle_next_(size_t chunk) NOEXCEPT -{ - size_t from{}; - size_t target{}; - - { - std::shared_lock map_lock(remap_mutex_); - - if (!loaded_.load() || fault_.load()) - return false; - - from = settled_.load(); - target = std::min(frontier_.load(), - system::ceilinged_add(from, chunk)); - - if (target <= from) - return false; - - if (!settle_write_(from, target, sequence{})) - { - set_first_code(error::fsync_failure); - return false; - } - } - - std::unique_lock map_lock(remap_mutex_); - - if (!loaded_.load()) - return false; - - // A raced settle boundary (flush) invalidates only this conversion. - if (settled_.load() != from) - return true; - - return settle_all_(target, sequence{}); -} - -// Evict up to chunk settled rows under memory scarcity: a wrapping sweep -// from the oldest offset (offset is write age in an append-only body, and -// re-read probability decays with it). Scarcity-driven release keeps the -// machine out of the exhausted-cache regime, where the kernel periodic -// sync walk (cost follows resident pages, not dirty) otherwise degrades -// fault service. A wrong eviction costs one cold read, so the sweep needs -// no precision. Runs under the shared remap lock (readers proceed). -TEMPLATE -bool CLASS::evict_next_(size_t chunk) NOEXCEPT -{ - std::shared_lock map_lock(remap_mutex_); - - if (!loaded_.load() || fault_.load()) - return false; - - // A truncated boundary self-heals here (the cursor clamps to a lap). - const auto end = settled_.load(); - - // Confine the lap to the recently settled tail: cache holds what was - // just written (and what validation just read), while the body below - // is long evicted, so a full-body lap sweeps unresident rows for most - // of its length and the tail refills faster than the cursor returns. - // A lap can hold no more than physical memory, as nothing else is - // resident to evict. - const auto span = system::greater(chunk, system_memory() / stride); - const auto floor = (end > span) ? (end - span) : zero; - const auto from = ((evicted_ > floor) && (evicted_ < end)) ? evicted_ : - floor; - const auto target = std::min(end, system::ceilinged_add(from, chunk)); - - if (target <= from) - return false; - - if (!evict_all_(from, target, sequence{})) - return false; - - // Wrap at the lap end so the next sweep resumes at the tail floor. - evicted_ = (target == end) ? zero : target; - return true; -} - -TEMPLATE -template -bool CLASS::settle_write_(size_t from, size_t to, - std::index_sequence) NOEXCEPT -{ - return (pwrite_all(opened_[Index], - std::next(memory_map_[Index], to_width(from)), - to_width(to) - to_width(from), - to_width(from)) && ...); -} - -#endif // MANAGE_STAGING - } // namespace database } // namespace libbitcoin +#endif // MANAGE_STAGING + #endif diff --git a/include/bitcoin/database/impl/memory/mmap_storage.ipp b/include/bitcoin/database/impl/memory/mmap_storage.ipp index 31d863eb6..369bbc7af 100644 --- a/include/bitcoin/database/impl/memory/mmap_storage.ipp +++ b/include/bitcoin/database/impl/memory/mmap_storage.ipp @@ -143,13 +143,7 @@ code CLASS::load() NOEXCEPT } remap_mutex_.unlock(); - -#if defined(MANAGE_STAGING) - settler_start_(); -#elif defined(HAVE_MSC) - scanner_start_(); -#endif - + worker_start_(); return error::success; } @@ -188,106 +182,6 @@ code CLASS::reload() NOEXCEPT return error::reload_locked; } -TEMPLATE -void CLASS::prepare(size_t STAGING_ONLY(offset), - size_t STAGING_ONLY(size)) NOEXCEPT -{ -#if defined(MANAGE_STAGING) - if (is_zero(size) || !dirty_) - return; - - // Count the writer before loading released below (sequentially - // consistent), pairing with the release protocol: release either observes - // the count (and aborts) or this writer observes released (and restores). - // Intent bits age (hot sampling), so they cannot protect a write held - // in flight across passes; the count persists until mark. - // A settle transition pairs the same way, but the writer waits UNCOUNTED - // (it retires its count and does not retake one until the transition - // clears), so the count drains monotonically and the transition is - // guaranteed to observe zero rather than merely likely to. - auto& writers = writer_slot_(); - for (;;) - { - while (transition_.load()) - std::this_thread::yield(); - - writers.fetch_add(one); - if (!transition_.load()) - break; - - writers.fetch_sub(one); - } - - // A settled head writes through its mapping. - if (shared_.load()) - return; - - if (!engaged_.load(relaxed) && !lazy_.load(relaxed)) - return; - - // Declare intent before the write (sequentially consistent, pairing with - // the release protocol), then restore any released page in the range. - auto restore = false; - auto page = offset / page_; - const auto end = (offset + sub1(size)) / page_; - while ((page <= end) && ((page / page_bound) < words_)) - { - const auto word = page / page_bound; - const auto flag = system::bit_right(page % page_bound); - intent_[word].fetch_or(flag); - restore |= !is_zero(system::bit_and(released_[word].load(), flag)); - ++page; - } - - if (restore) - restore_(offset, size); -#endif -} - -TEMPLATE -void CLASS::mark(size_t STAGING_ONLY(offset), - size_t STAGING_ONLY(size)) NOEXCEPT -{ -#if defined(MANAGE_STAGING) - if (is_zero(size) || !dirty_) - return; - - // Marks follow content writes; transfer clears before reading, so pages - // remarked during a transfer are simply rewritten by the next pass. A - // settled head writes through its mapping, so its marks count only (the - // settler reads the rate); no transition intervenes (the writer is - // counted), so the path matches prepare. - if (shared_.load()) - marks_.fetch_add(one, relaxed); - else - remark_(offset, size); - - // Uncount the writer after its marks (sequentially consistent), so a - // release pass loading a drained count observes the dirty bits. Only - // prepare() counts, so only mark() may uncount (transfer failure restores - // marks by remark_, as an unpaired uncount here corrupts the count). - writer_slot_().fetch_sub(one); -#endif -} - -TEMPLATE -void CLASS::current(bool STAGING_ONLY(state)) NOEXCEPT -{ -#if defined(MANAGE_STAGING) - current_.store(state); -#endif -} - -TEMPLATE -bool CLASS::settled() const NOEXCEPT -{ -#if defined(MANAGE_STAGING) - return shared_.load(); -#else - return false; -#endif -} - // Suspend writes before calling. TEMPLATE code CLASS::flush() NOEXCEPT @@ -336,12 +230,7 @@ code CLASS::flush() NOEXCEPT TEMPLATE code CLASS::unload() NOEXCEPT { -#if defined(MANAGE_STAGING) - settler_stop_(); -#elif defined(HAVE_MSC) - scanner_stop_(); -#endif - + worker_stop_(); std::unique_lock field_lock(field_mutex_); if (remap_mutex_.try_lock()) @@ -370,12 +259,7 @@ code CLASS::unload() NOEXCEPT TEMPLATE code CLASS::shrink() NOEXCEPT { -#if defined(MANAGE_STAGING) - settler_stop_(); -#elif defined(HAVE_MSC) - scanner_stop_(); -#endif - + worker_stop_(); std::unique_lock field_lock(field_mutex_); if (remap_mutex_.try_lock()) @@ -401,13 +285,7 @@ code CLASS::shrink() NOEXCEPT } remap_mutex_.unlock(); - -#if defined(MANAGE_STAGING) - settler_start_(); -#elif defined(HAVE_MSC) - scanner_start_(); -#endif - + worker_start_(); return error::success; } @@ -459,44 +337,7 @@ bool CLASS::truncate(size_t count) NOEXCEPT return false; } - // Discard extents above the truncation and clamp any overlap. - if (staged_) - { - std::unique_lock extent_lock(extent_mutex_); - - using namespace system; - auto [head, size] = unpack_word(window_.load(relaxed)); - while (!is_zero(size)) - { - auto& tail = ring_.at((head + sub1(size)) % extents); - const auto start = tail.start.load(relaxed); - if (start >= count) - { - --size; - continue; - } - - if (ceilinged_add(start, tail.count.load(relaxed)) > count) - { - const auto trimmed = count - start; - tail.count.store(trimmed, relaxed); - - // Clamp outstanding within the state, generation unchanged - // (the extent is trimmed, not recycled; writers quiescent). - const auto limit = trimmed * columns; - const auto state = tail.state.load(relaxed); - if (bit_and(state, outstanding_mask) > limit) - tail.state.store(pack_extent_(shift_right( - state, generation_shift), limit), relaxed); - } - - break; - } - - window_.store(pack_word(head, size), release); - frontier_.store(is_zero(size) ? count : - ring_.at(head).start.load(relaxed)); - } + trim_(count); #endif logical_.store(count); @@ -504,42 +345,6 @@ bool CLASS::truncate(size_t count) NOEXCEPT return true; } -// Iterated growth (callers hold the remap lock). Growth asks are amortized -// (rate surplus over the necessity), and each is admitted only while it -// leaves the configured headroom of the backing resource unclaimed (probed -// with the ask, released on grant), so exhaustion never consumes the -// system's final bytes. A large amortization step can be refused while the -// necessity fits, so iterate: halve the refused surplus toward the -// necessity. Refusal of the necessity is exhaustion, not store damage: -// published as disk full (space set, store intact, writes fail fast until -// cleared), it clears by settle drainage or operator relief, where teardown -// would convert a shortage into a restore. -TEMPLATE -bool CLASS::grow_(size_t end) NOEXCEPT -{ - if (!is_zero(space_.load())) - return false; - - using namespace system; - for (auto extended = to_growth(end); - !remap_all_(extended, sequence{}, false); - extended = ceilinged_add(end, - to_half(floored_subtract(extended, end)))) - { - if (fault_.load()) - return false; - - if (extended <= end) - { - set_disk_space(ceilinged_add(headroom_, ceilinged_multiply( - floored_subtract(end, capacity_.load()), stride))); - return false; - } - } - - return true; -} - TEMPLATE bool CLASS::expand(size_t count) NOEXCEPT { @@ -684,7 +489,7 @@ TEMPLATE size_t CLASS::allocate(size_t count, uint8_t backfill) NOEXCEPT { #if defined(MANAGE_STAGING) - if (!staged_ && dirty_ && !shared_.load()) + if (managed_ && !shared_.load()) return allocate_filled_(count, backfill); #endif diff --git a/include/bitcoin/database/memory/mmap.hpp b/include/bitcoin/database/memory/mmap.hpp index 58c230a6c..64a51877a 100644 --- a/include/bitcoin/database/memory/mmap.hpp +++ b/include/bitcoin/database/memory/mmap.hpp @@ -1,4 +1,4 @@ -/** +/** * Copyright (c) 2011-2026 libbitcoin developers * * This file is part of libbitcoin. @@ -90,6 +90,9 @@ class mmap /// True if the memory map(s) are loaded. bool is_loaded() const NOEXCEPT; + /// True while a managed head shares its file mapping (settled). + bool shared() const NOEXCEPT; + /// storage interface /// ----------------------------------------------------------------------- @@ -128,9 +131,6 @@ class mmap /// Report store currency (permits managed head settlement). void current(bool state) NOEXCEPT override; - /// True when a managed head has settled to its file mapping. - bool settled() const NOEXCEPT; - /// Flush memory map(s) to disk, suspend writes for call, must be loaded. code flush() NOEXCEPT override; @@ -202,12 +202,11 @@ class mmap { return bytes / widths.front(); } - + static constexpr size_t to_rows(size_t bytes) NOEXCEPT { // Convert constructor's byte minimum to row denomination. - constexpr auto row = (Widths + ...); - return system::ceilinged_divide(bytes, row); + return system::ceilinged_divide(bytes, stride); } static size_t to_chunk() NOEXCEPT; @@ -221,13 +220,26 @@ class mmap private: static constexpr size_t page_bound = to_bits(sizeof(uint64_t)); - static constexpr size_t settle_chunk = system::power2(28u); - static constexpr size_t advise_chunk = system::power2(30u); + static constexpr auto fail = -1; + static constexpr auto relaxed = std::memory_order_relaxed; + static constexpr auto release = std::memory_order_release; + using sequence = std::make_index_sequence; + + // Growth (memory commitment and disk provisioning). static constexpr size_t commit_chunk = system::power2(28u); static constexpr size_t chunk_scale = 256; + static constexpr size_t advise_chunk = system::power2(30u); + static constexpr size_t reserve_factor = 4; + + // Staged bodies (settle, throttle, evict). + static constexpr size_t settle_chunk = system::power2(28u); static constexpr size_t evict_chunk = system::power2(30u); static constexpr size_t compress_factor = 32; static constexpr size_t evict_factor = 32; + static constexpr size_t throttle_factor = 8; + static constexpr size_t active_factor = 32; + static constexpr size_t urgent_factor = 4; + static constexpr size_t idle_seconds = 60; // Settled-extent demotion ceiling (installed memory). Demotion trades // body cache for head residency, which pays only while the head set @@ -243,23 +255,21 @@ class mmap // know that head residency is the store's priority). A higher floor // keeps free memory above the watermark, so the sweep is the reclaim. static constexpr size_t sweep_factor = 8; - static constexpr size_t throttle_factor = 8; - static constexpr size_t active_factor = 32; - static constexpr size_t urgent_factor = 4; - static constexpr size_t idle_seconds = 60; + + // Managed heads (touch, release, share). static constexpr size_t touch_seconds = 4; static constexpr size_t touch_span = 16384; + static constexpr size_t release_quiet = 128; // Release conversion granularity: chunked runs bound address space // fragmentation (each conversion splits a mapping) to the measured flat // zone of host memory management (heads / chunk fragments worst case). static constexpr size_t release_chunk = system::power2(20u); - static constexpr size_t release_quiet = 128; - // Writes per tick sustained for unsettle_seconds reinstall a settled head + // Writes per tick sustained for unshare_seconds reinstall a shared head // (a block at the top is a burst of one tick, a catch-up is sustained). - static constexpr size_t unsettle_writes = 1000; - static constexpr size_t unsettle_seconds = 10; + static constexpr size_t unshare_writes = 1000; + static constexpr size_t unshare_seconds = 10; #if defined(HAVE_APPLE) // Anonymous overflow feeds the darwin compressor (10.8GB measured at // 16GB), which mincore hides from the touch guard; release converts @@ -277,16 +287,8 @@ class mmap // ratio tuning does not return with it. Excludes head_release (nothing // to release) and the dirty bitmap (nothing to transfer). static constexpr bool head_shared = false; - static constexpr size_t headroom = 4; -#if defined(STAGING_TELEMETRY) - static constexpr size_t telemetry_seconds = 60; -#endif - static constexpr auto fail = -1; - static constexpr auto relaxed = std::memory_order_relaxed; - static constexpr auto release = std::memory_order_release; - using sequence = std::make_index_sequence; - // mman dispatch, not thread safe. + // column dispatch, not thread safe. template bool flush_all_(size_t rows, std::index_sequence) NOEXCEPT; template @@ -299,113 +301,110 @@ class mmap bool grow_(size_t end) NOEXCEPT; bool probe_(size_t capacity) NOEXCEPT; - // mman wrappers, not thread safe. + // backend wrappers (native or staged by build), not thread safe. template bool flush_(size_t rows) NOEXCEPT; template - bool persist_(size_t to) NOEXCEPT; - template bool map_() NOEXCEPT; template - bool release_(size_t size) NOEXCEPT; - template bool unmap_(size_t size) NOEXCEPT; template bool remap_(size_t size, bool final=true) NOEXCEPT; template bool resize_(size_t size, bool final=true) NOEXCEPT; + + // worker (instance-owned thread, load/unload lifecycle). + void worker_start_() NOEXCEPT; + void worker_stop_() NOEXCEPT; + void worker_run_() NOEXCEPT; + bool tick_() NOEXCEPT; + +#if defined(HAVE_MSC) template - bool finalize_(size_t size) NOEXCEPT; + bool release_(size_t size) NOEXCEPT; + template + bool finalize_() NOEXCEPT; -#if defined(MANAGE_STAGING) - // staging dispatch, not thread safe. - template - bool settle_all_(size_t rows, std::index_sequence) NOEXCEPT; - template - bool unsettle_all_(size_t rows, std::index_sequence) NOEXCEPT; - template - bool evict_all_(size_t from, size_t to, - std::index_sequence) NOEXCEPT; + // Working-set steering for the native (file-backed) mapping. The cache + // manager trims without knowing that head residency is the store's + // priority, so it takes head pages alongside cold body cache and every + // head miss is a serial fault on the probe path. The scan asserts head + // residency (a read sets the access bit) and leads the trim on bodies + // (unlock moves the range to the standby list, reclaimed first). + void scan_run_() NOEXCEPT; +#endif - // staging wrappers, not thread safe. +#if defined(MANAGE_STAGING) + // anonymous backend (reservation, commitment, conversion), not thread safe. template bool stage_() NOEXCEPT; template bool commit_(size_t size, bool final=true) NOEXCEPT; template - bool settle_(size_t from, size_t to) NOEXCEPT; - template - bool unsettle_(size_t rows) NOEXCEPT; + bool persist_(size_t to) NOEXCEPT; template - bool evict_(size_t from, size_t to) NOEXCEPT; + bool sync_() NOEXCEPT; template void teardown_(const error::error_t& ec) NOEXCEPT; + bool advise_(uint8_t* map, size_t size) const NOEXCEPT; + size_t to_reservation(size_t rows) const NOEXCEPT; + size_t page_floor(size_t bytes) const NOEXCEPT; + size_t page_ceiling(size_t bytes) const NOEXCEPT; - // staging utilities, not thread safe (claim_ is lock-free thread safe). + // extent ring (write completion), locked except claim_ (lock-free). struct extent; - size_t allocate_filled_(size_t count, uint8_t backfill) NOEXCEPT; - bool lazy_install_() NOEXCEPT; size_t record_(size_t count) NOEXCEPT; bool claim_(extent& record, size_t offset, size_t count) NOEXCEPT; void maintain_() NOEXCEPT; void discard_() NOEXCEPT; + void trim_(size_t count) NOEXCEPT; + + // staged body (settle, throttle, evict), not thread safe unless noted. + template + bool settle_all_(size_t rows, std::index_sequence) NOEXCEPT; + template + bool unsettle_all_(size_t rows, std::index_sequence) NOEXCEPT; + template + bool evict_all_(size_t from, size_t to, + std::index_sequence) NOEXCEPT; + template + bool settle_write_(size_t from, size_t to, + std::index_sequence) NOEXCEPT; + template + bool settle_(size_t from, size_t to) NOEXCEPT; + template + bool unsettle_(size_t rows) NOEXCEPT; + template + bool evict_(size_t from, size_t to) NOEXCEPT; void throttle_() NOEXCEPT; void signal_() NOEXCEPT; + void body_run_() NOEXCEPT; + bool settle_next_(size_t chunk) NOEXCEPT; + bool evict_next_(size_t chunk, size_t& cursor) NOEXCEPT; - // dirty page transfer (unstaged instances), lock-free with writers. + // managed head (lazy install, dirty transfer, release, share). + size_t allocate_filled_(size_t count, uint8_t backfill) NOEXCEPT; + bool lazy_install_() NOEXCEPT; template bool transfer_(size_t bytes) NOEXCEPT; - template - bool sync_() NOEXCEPT; void remark_(size_t offset, size_t size) NOEXCEPT; + void head_run_() NOEXCEPT; - // head page release (unstaged instances), synchronized with writers by - // the prepare/release bit protocol (see release_pages_). + // head page release, synchronized with writers by the prepare/release + // bit protocol (see release_pages_). bool release_pages_() NOEXCEPT; - void quiesce_() NOEXCEPT; + void restore_(size_t offset, size_t size) NOEXCEPT; + void declare_released_() NOEXCEPT; + + // head share transitions, synchronized with writers by the count. std::atomic& writer_slot_() NOEXCEPT; size_t writers_count_() const NOEXCEPT; void writers_reset_() NOEXCEPT; + void quiesce_() NOEXCEPT; bool share_(size_t transferred) NOEXCEPT; void unshare_() NOEXCEPT; - void declare_released_() NOEXCEPT; - void restore_(size_t offset, size_t size) NOEXCEPT; - - // settle scheduler (instance-owned thread, load/unload lifecycle). - void settler_start_() NOEXCEPT; - void settler_stop_() NOEXCEPT; - void settler_run_() NOEXCEPT; - void head_run_() NOEXCEPT; - bool settle_next_(size_t chunk) NOEXCEPT; - bool evict_next_(size_t chunk) NOEXCEPT; - template - bool settle_write_(size_t from, size_t to, - std::index_sequence) NOEXCEPT; - bool advise_(uint8_t* map, size_t size) const NOEXCEPT; - size_t to_reservation(size_t rows) const NOEXCEPT; - size_t page_floor(size_t bytes) const NOEXCEPT; - size_t page_ceiling(size_t bytes) const NOEXCEPT; #endif // MANAGE_STAGING -#if defined(HAVE_MSC) - // Working-set steering for the native (file-backed) mapping. The cache - // manager trims without knowing that head residency is the store's - // priority, so it takes head pages alongside cold body cache and every - // head miss is a serial fault on the probe path. The scanner asserts head - // residency (a read sets the access bit) and leads the trim on bodies - // (unlock moves the range to the standby list, reclaimed first). - void scanner_start_() NOEXCEPT; - void scanner_stop_() NOEXCEPT; - void scanner_run_() NOEXCEPT; - - std::thread scanner_{}; - std::atomic_bool scanning_{}; - std::condition_variable scanner_cv_{}; - mutable std::mutex scanner_mutex_{}; - size_t touched_{}; - size_t unlocked_{}; -#endif // HAVE_MSC - // These are thread safe (const). const paths filenames_; const size_t minimum_; @@ -414,6 +413,7 @@ class mmap const advice access_; const bool random_; const bool staged_; + const bool managed_; // These are thread safe (atomic). std::atomic error_{ error::success }; @@ -432,6 +432,12 @@ class mmap std::array memory_map_{}; mutable std::shared_mutex remap_mutex_{}; + // These are protected by worker_mutex_ (working_ is atomic). + std::thread worker_{}; + std::atomic_bool working_{}; + std::condition_variable worker_cv_{}; + mutable std::mutex worker_mutex_{}; + #if defined(MANAGE_STAGING) // Page-dirty bitmap for unstaged (rewrite-in-place head) instances. // Marks follow content writes; transfer clears before reading, so a @@ -471,16 +477,9 @@ class mmap std::atomic state; }; - // This is unshared (settler thread only). - size_t evicted_{}; - -#if defined(STAGING_TELEMETRY) - // These are unshared (settler thread only). - size_t telemetry_{}; - size_t marked_{}; - size_t peaked_{}; - size_t active_{}; -#endif + // These are set at map and constant while loaded. + size_t page_{}; + size_t limit_{}; // These are thread safe (atomic). std::atomic marks_{}; @@ -489,6 +488,7 @@ class mmap std::atomic window_{}; // These are protected by remap_mutex_. + std::array reserved_{}; std::unique_ptr dirty_{}; std::unique_ptr intent_{}; std::unique_ptr released_{}; @@ -523,25 +523,16 @@ class mmap // Serializes page release against restore (prepare slow path). mutable std::mutex restore_mutex_{}; - // Serializes transfer passes (settler tick against flush), as concurrent + // Serializes transfer passes (worker tick against flush), as concurrent // passes split the claimed dirty set, allowing a flush to complete while // claimed pages remain unwritten (a stale snapshot copy). mutable std::mutex transfer_mutex_{}; - // These are protected by extent_mutex_. - size_t page_{}; + // This is protected by extent_mutex_. std::array ring_{}; - std::array reserved_{}; mutable std::mutex extent_mutex_{}; - // These are protected by settler_mutex_. - std::thread settler_{}; - std::atomic_bool settling_{}; - std::condition_variable settler_cv_{}; - mutable std::mutex settler_mutex_{}; - - // These are protected by throttle_mutex_. - size_t limit_{}; + // This is protected by throttle_mutex_. std::condition_variable throttle_cv_{}; mutable std::mutex throttle_mutex_{}; #endif // MANAGE_STAGING @@ -559,10 +550,13 @@ BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT) #include #include -#include #include -#include #include +#include +#include +#include +#include +#include BC_POP_WARNING() diff --git a/include/bitcoin/database/memory/mstage.hpp b/include/bitcoin/database/memory/mstage.hpp index 54c046aa6..d5ef5eb21 100644 --- a/include/bitcoin/database/memory/mstage.hpp +++ b/include/bitcoin/database/memory/mstage.hpp @@ -36,6 +36,9 @@ #if defined(MANAGE_STAGING) +namespace libbitcoin { +namespace database { + /// Reserve inaccessible anonymous address space (MAP_FAILED on failure). void* mmap_reserve(size_t size) NOEXCEPT; @@ -85,6 +88,9 @@ bool pread_all(int fd, uint8_t* to, size_t size, size_t offset) NOEXCEPT; bool pwrite_all(int fd, const uint8_t* from, size_t size, size_t offset) NOEXCEPT; +} // namespace database +} // namespace libbitcoin + #endif // MANAGE_STAGING #endif diff --git a/src/memory/mstage.cpp b/src/memory/mstage.cpp index 189a5a17a..0eddf0d70 100644 --- a/src/memory/mstage.cpp +++ b/src/memory/mstage.cpp @@ -37,8 +37,10 @@ #include #endif -using namespace libbitcoin; -using namespace libbitcoin::system; +namespace libbitcoin { +namespace database { + +using namespace system; static constexpr auto transfer_chunk = power2(30u); void* mmap_reserve(size_t size) NOEXCEPT @@ -131,30 +133,34 @@ int mmap_settle(void* address, size_t size, int fd, size_t offset) NOEXCEPT // Mapped pages (released runs, where the file is the live copy) hold a // reference and are not discarded. Dirty pages are not discarded either, so // a page discards on the pass following its writeback. +#if defined(POSIX_FADV_DONTNEED) int file_discard(int fd) NOEXCEPT { -#if defined(POSIX_FADV_DONTNEED) return ::posix_fadvise(fd, 0, 0, POSIX_FADV_DONTNEED); +} #else - return is_zero(fd) ? 0 : 0; -#endif +int file_discard(int) NOEXCEPT +{ + return 0; } +#endif +// Deactivation expresses the residency priority the kernel cannot infer: +// settled bodies stay cached for imminent re-read (validation follows +// archival) but reclaim first under pressure, before active pages and +// anonymous heads (which otherwise swap to preserve the body cache). A +// read reactivates, so genuinely hot pages promote themselves back. +#if defined(MADV_COLD) int mmap_cold(void* address, size_t size) NOEXCEPT { -#if defined(MADV_COLD) - // Deactivation expresses the residency priority the kernel cannot infer: - // settled bodies stay cached for imminent re-read (validation follows - // archival) but reclaim first under pressure, before active pages and - // anonymous heads (which otherwise swap to preserve the body cache). A - // read reactivates, so genuinely hot pages promote themselves back. return ::madvise(address, size, MADV_COLD); +} #else - // Both parameters are unused without MADV_COLD (darwin), consumed by the - // test as file_discard consumes its own without POSIX_FADV_DONTNEED. - return (address != nullptr) && !is_zero(size) ? 0 : 0; -#endif +int mmap_cold(void*, size_t) NOEXCEPT +{ + return 0; } +#endif // Diagnostic attribution: names each anonymous vma in the range so smaps and // per-process accounting decompose by table (heads vs staged bodies vs @@ -197,24 +203,28 @@ int mmap_share(void* address, size_t size, int fd, size_t offset) NOEXCEPT // the touch guard cannot defend a head there; wiring can (the user wire // limit leaves the kernel its share, and refusal leaves the pages unpinned). // Linux defends by the touch pass (unprivileged mlock is capped at 8MB). +#if defined(HAVE_APPLE) int mmap_wire(void* address, size_t size) NOEXCEPT { -#if defined(HAVE_APPLE) return ::mlock(address, size); -#else - return (address != nullptr) && !is_zero(size) ? 0 : 0; -#endif } int mmap_unwire(void* address, size_t size) NOEXCEPT { -#if defined(HAVE_APPLE) return ::munlock(address, size); +} #else - return (address != nullptr) && !is_zero(size) ? 0 : 0; -#endif +int mmap_wire(void*, size_t) NOEXCEPT +{ + return 0; } +int mmap_unwire(void*, size_t) NOEXCEPT +{ + return 0; +} +#endif + int mmap_unsettle(void* address, size_t size) NOEXCEPT { // No reserve: an unsettled span can span hundreds of gigabytes (truncate @@ -350,4 +360,7 @@ bool pwrite_all(int fd, const uint8_t* from, size_t size, return true; } +} // namespace database +} // namespace libbitcoin + #endif // MANAGE_STAGING diff --git a/test/memory/mmap.cpp b/test/memory/mmap.cpp index cdfe9807b..004858f2f 100644 --- a/test/memory/mmap.cpp +++ b/test/memory/mmap.cpp @@ -1388,22 +1388,23 @@ BOOST_AUTO_TEST_CASE(mmap__allocate__concurrent__unique_dense_claims) #if defined(MANAGE_STAGING) -// Settle transitions are asynchronous (settler tick), so waits are bounded. +// Share transitions are asynchronous (worker tick), so waits are bounded. constexpr size_t drain_wait = 70; -constexpr size_t settle_wait = 120; -constexpr size_t unsettle_wait = 60; +constexpr size_t share_wait = 120; +constexpr size_t unshare_wait = 60; constexpr size_t cell_width = sizeof(uint64_t); -static bool settled_within(const map& instance, bool state, size_t seconds) NOEXCEPT +static bool shared_within(const map& instance, bool state, size_t seconds) NOEXCEPT { const std::vector ticks(seconds); return std::any_of(ticks.begin(), ticks.end(), [&](size_t) NOEXCEPT { std::this_thread::sleep_for(std::chrono::seconds(one)); - return instance.settled() == state; + return instance.shared() == state; }); } +// Unguarded write (hashmap head cell path). static void stamp(map& instance, size_t cell, uint64_t generation) NOEXCEPT { const auto offset = cell * cell_width; @@ -1413,6 +1414,16 @@ static void stamp(map& instance, size_t cell, uint64_t generation) NOEXCEPT instance.mark(offset, cell_width); } +// Accessor-guarded write (array/head map paths), held across the mark. +static void stamp_held(map& instance, size_t cell, uint64_t generation) NOEXCEPT +{ + const auto offset = cell * cell_width; + const auto memory = instance.get(offset); + instance.prepare(offset, cell_width); + system::unsafe_to_little_endian(memory.begin(), generation); + instance.mark(offset, cell_width); +} + static std::vector read_cells(const map& instance, size_t cells) NOEXCEPT { std::vector positions(cells); @@ -1430,17 +1441,17 @@ static std::vector read_cells(const map& instance, size_t cells) NOEXC // Sustained volume is repetition by definition, so the writer drive iterates. // Stamps rising generations across the cells until stopped, returning the // last generation stamped into each. -static std::vector drive(map& instance, size_t cells, const std::atomic_bool& stop) NOEXCEPT +static std::vector drive(map& instance, size_t cells, const std::atomic_bool& stop, bool held=false) NOEXCEPT { std::vector last(cells); for (uint64_t generation = one; !stop.load(); ++generation) for (size_t cell = zero; cell < cells; ++cell) - stamp(instance, cell, last.at(cell) = generation); + (held ? stamp_held : stamp)(instance, cell, last.at(cell) = generation); return last; } -BOOST_AUTO_TEST_CASE(mmap__settle__not_current__unsettled) +BOOST_AUTO_TEST_CASE(mmap__share__not_current__unshared) { const std::string file = TEST_PATH; BOOST_REQUIRE(test::create(file)); @@ -1450,13 +1461,13 @@ BOOST_AUTO_TEST_CASE(mmap__settle__not_current__unsettled) BOOST_REQUIRE(!instance.load()); BOOST_REQUIRE_NE(instance.allocate(cell_width), storage::eof); - BOOST_REQUIRE(!settled_within(instance, true, drain_wait)); + BOOST_REQUIRE(!shared_within(instance, true, drain_wait)); BOOST_REQUIRE(!instance.unload()); BOOST_REQUIRE(!instance.close()); BOOST_REQUIRE(!instance.get_fault()); } -BOOST_AUTO_TEST_CASE(mmap__settle__current_quiescent__settled_until_unload) +BOOST_AUTO_TEST_CASE(mmap__share__current_quiescent__shared_until_unload) { const std::string file = TEST_PATH; BOOST_REQUIRE(test::create(file)); @@ -1467,14 +1478,14 @@ BOOST_AUTO_TEST_CASE(mmap__settle__current_quiescent__settled_until_unload) BOOST_REQUIRE_NE(instance.allocate(cell_width), storage::eof); instance.current(true); - BOOST_REQUIRE(settled_within(instance, true, settle_wait)); + BOOST_REQUIRE(shared_within(instance, true, share_wait)); BOOST_REQUIRE(!instance.unload()); - BOOST_REQUIRE(!instance.settled()); + BOOST_REQUIRE(!instance.shared()); BOOST_REQUIRE(!instance.close()); BOOST_REQUIRE(!instance.get_fault()); } -BOOST_AUTO_TEST_CASE(mmap__settle__write_while_settled__persists) +BOOST_AUTO_TEST_CASE(mmap__share__write_while_shared__persists) { const std::string file = TEST_PATH; BOOST_REQUIRE(test::create(file)); @@ -1485,7 +1496,7 @@ BOOST_AUTO_TEST_CASE(mmap__settle__write_while_settled__persists) BOOST_REQUIRE_NE(instance.allocate(cell_width), storage::eof); instance.current(true); - BOOST_REQUIRE(settled_within(instance, true, settle_wait)); + BOOST_REQUIRE(shared_within(instance, true, share_wait)); stamp(instance, zero, 42); BOOST_REQUIRE(!instance.unload()); @@ -1498,7 +1509,7 @@ BOOST_AUTO_TEST_CASE(mmap__settle__write_while_settled__persists) BOOST_REQUIRE(!instance.get_fault()); } -BOOST_AUTO_TEST_CASE(mmap__settle__sustained_writes__unsettled_without_loss) +BOOST_AUTO_TEST_CASE(mmap__share__sustained_writes__unshared_without_loss) { constexpr size_t cells = 512; @@ -1511,12 +1522,46 @@ BOOST_AUTO_TEST_CASE(mmap__settle__sustained_writes__unsettled_without_loss) BOOST_REQUIRE_NE(instance.allocate(cells * cell_width), storage::eof); instance.current(true); - BOOST_REQUIRE(settled_within(instance, true, settle_wait)); + BOOST_REQUIRE(shared_within(instance, true, share_wait)); std::atomic_bool stop{}; std::vector expected{}; std::thread writer([&]() NOEXCEPT { expected = drive(instance, cells, stop); }); - const auto unsettled = settled_within(instance, false, unsettle_wait); + const auto unshared = shared_within(instance, false, unshare_wait); + stop.store(true); + writer.join(); + + BOOST_REQUIRE(!instance.unload()); + BOOST_REQUIRE(!instance.close()); + BOOST_REQUIRE(!instance.open()); + BOOST_REQUIRE(!instance.load()); + BOOST_REQUIRE(unshared); + BOOST_REQUIRE(!instance.shared()); + BOOST_REQUIRE(read_cells(instance, cells) == expected); + BOOST_REQUIRE(!instance.unload()); + BOOST_REQUIRE(!instance.close()); + BOOST_REQUIRE(!instance.get_fault()); +} + +BOOST_AUTO_TEST_CASE(mmap__share__sustained_held_writes__unshared_without_loss) +{ + constexpr size_t cells = 512; + + const std::string file = TEST_PATH; + BOOST_REQUIRE(test::create(file)); + + map instance(file, { 1, 50 }); + BOOST_REQUIRE(!instance.open()); + BOOST_REQUIRE(!instance.load()); + BOOST_REQUIRE_NE(instance.allocate(cells * cell_width), storage::eof); + + instance.current(true); + BOOST_REQUIRE(shared_within(instance, true, share_wait)); + + std::atomic_bool stop{}; + std::vector expected{}; + std::thread writer([&]() NOEXCEPT { expected = drive(instance, cells, stop, true); }); + const auto unshared = shared_within(instance, false, unshare_wait); stop.store(true); writer.join(); @@ -1524,15 +1569,15 @@ BOOST_AUTO_TEST_CASE(mmap__settle__sustained_writes__unsettled_without_loss) BOOST_REQUIRE(!instance.close()); BOOST_REQUIRE(!instance.open()); BOOST_REQUIRE(!instance.load()); - BOOST_REQUIRE(unsettled); - BOOST_REQUIRE(!instance.settled()); + BOOST_REQUIRE(unshared); + BOOST_REQUIRE(!instance.shared()); BOOST_REQUIRE(read_cells(instance, cells) == expected); BOOST_REQUIRE(!instance.unload()); BOOST_REQUIRE(!instance.close()); BOOST_REQUIRE(!instance.get_fault()); } -BOOST_AUTO_TEST_CASE(mmap__settle__filled_tail_above_logical__survives_settle) +BOOST_AUTO_TEST_CASE(mmap__share__filled_tail_above_logical__survives_share) { constexpr size_t cells = 512; constexpr size_t tail = add1(cells); @@ -1549,7 +1594,7 @@ BOOST_AUTO_TEST_CASE(mmap__settle__filled_tail_above_logical__survives_settle) BOOST_REQUIRE_GT(instance.capacity(), add1(tail) * cell_width); instance.current(true); - BOOST_REQUIRE(settled_within(instance, true, settle_wait)); + BOOST_REQUIRE(shared_within(instance, true, share_wait)); BOOST_REQUIRE(instance.get_filled(tail * cell_width, cell_width, system::bit_all)); BOOST_REQUIRE_EQUAL(read_cells(instance, add1(tail)).back(), fill); From 5b75e898d44bc977b577ff4d4d1e8d29328d7621 Mon Sep 17 00:00:00 2001 From: Eric Voskuil Date: Tue, 22 Sep 2026 17:09:18 -0400 Subject: [PATCH 5/5] Remove dead code. --- builds/msvc/build-msvc.cmd | 2 +- builds/msvc/build/build_all.bat | 6 +----- builds/msvc/build/build_base.bat | 4 +--- builds/msvc/build/nuget_all.bat | 18 ++++-------------- 4 files changed, 7 insertions(+), 23 deletions(-) diff --git a/builds/msvc/build-msvc.cmd b/builds/msvc/build-msvc.cmd index 07c3cd2aa..e31ef0553 100644 --- a/builds/msvc/build-msvc.cmd +++ b/builds/msvc/build-msvc.cmd @@ -105,7 +105,7 @@ if "!libbitcoin_database_TAG!" == "" ( ) if "!BUILD_VERSION!" == "" ( - set "BUILD_VERSION=vs2022" + set "BUILD_VERSION=vs2026" call :msg_warn "Build msvc version not provided, defaulting to '!BUILD_VERSION!'." ) diff --git a/builds/msvc/build/build_all.bat b/builds/msvc/build/build_all.bat index 4edaf926c..7c45c55d8 100644 --- a/builds/msvc/build/build_all.bat +++ b/builds/msvc/build/build_all.bat @@ -1,9 +1,5 @@ @ECHO OFF CALL nuget_all.bat ECHO. -CALL build_base.bat vs2022 libbitcoin-database "Microsoft Visual Studio\2022\Community\VC\Auxiliary\Build" -CALL build_base.bat vs2019 libbitcoin-database "Microsoft Visual Studio\2019\Community\VC\Auxiliary\Build" -CALL build_base.bat vs2017 libbitcoin-database "Microsoft Visual Studio\2017\Community\VC\Auxiliary\Build" -REM CALL build_base.bat vs2015 libbitcoin-database "Microsoft Visual Studio 14.0\VC" -REM CALL build_base.bat vs2013 libbitcoin-database "Microsoft Visual Studio 12.0\VC" +CALL build_base.bat vs2026 libbitcoin-database "Microsoft Visual Studio\18\Community\VC\Auxiliary\Build" PAUSE diff --git a/builds/msvc/build/build_base.bat b/builds/msvc/build/build_base.bat index 2ec0e4565..395d23298 100644 --- a/builds/msvc/build/build_base.bat +++ b/builds/msvc/build/build_base.bat @@ -1,7 +1,5 @@ @ECHO OFF -REM Usage: [buildbase.bat vs2017 libbitcoin "Microsoft Visual Studio\2017\Community\VC\Auxiliary\Build"] -REM Usage: [buildbase.bat vs2015 libbitcoin "Microsoft Visual Studio 14.0\VC"] -REM Usage: [buildbase.bat vs2013 libbitcoin "Microsoft Visual Studio 12.0\VC"] +REM Usage: [buildbase.bat vs2026 libbitcoin "Microsoft Visual Studio\18\Community\VC\Auxiliary\Build"] SET studio=%1 SET project=%2 diff --git a/builds/msvc/build/nuget_all.bat b/builds/msvc/build/nuget_all.bat index d7426c30a..5767e32d0 100644 --- a/builds/msvc/build/nuget_all.bat +++ b/builds/msvc/build/nuget_all.bat @@ -1,15 +1,5 @@ @ECHO OFF -ECHO Downloading libbitcoin vs2022 dependencies from NuGet -CALL nuget.exe install ..\vs2022\libbitcoin-database\packages.config -CALL nuget.exe install ..\vs2022\libbitcoin-database-tools\packages.config -CALL nuget.exe install ..\vs2022\libbitcoin-database-test\packages.config -ECHO. -ECHO Downloading libbitcoin vs2019 dependencies from NuGet -CALL nuget.exe install ..\vs2019\libbitcoin-database\packages.config -CALL nuget.exe install ..\vs2019\libbitcoin-database-tools\packages.config -CALL nuget.exe install ..\vs2019\libbitcoin-database-test\packages.config -ECHO. -ECHO Downloading libbitcoin vs2017 dependencies from NuGet -CALL nuget.exe install ..\vs2017\libbitcoin-database\packages.config -CALL nuget.exe install ..\vs2017\libbitcoin-database-tools\packages.config -CALL nuget.exe install ..\vs2017\libbitcoin-database-test\packages.config +ECHO Downloading libbitcoin vs2026 dependencies from NuGet +CALL nuget.exe install ..\vs2026\libbitcoin-database\packages.config +CALL nuget.exe install ..\vs2026\libbitcoin-database-tools\packages.config +CALL nuget.exe install ..\vs2026\libbitcoin-database-test\packages.config