posix: wire pages against compressor, shard write count, release on current.

include/bitcoin/database/impl/memory/mmap_staging.ipp

@@ -373,7 +373,7 @@ bool CLASS::stage_() NOEXCEPT
             intent_ = std::make_unique<dirty_bitmaps>(words_);
             released_ = std::make_unique<dirty_bitmaps>(words_);
             sweep_ = std::make_unique<uint64_t[]>(words_);
-            writers_.store(zero);
+            writers_reset_();
         }
     }
 
@@ -500,7 +500,7 @@ bool CLASS::lazy_install_() NOEXCEPT
     intent_ = std::make_unique<dirty_bitmaps>(words_);
     released_ = std::make_unique<dirty_bitmaps>(words_);
     sweep_ = std::make_unique<uint64_t[]>(words_);
-    writers_.store(zero);
+    writers_reset_();
 
     // Released file prefix (full pages below logical).
     const auto floor = page_floor(logical);
@@ -528,6 +528,9 @@ bool CLASS::lazy_install_() NOEXCEPT
         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).
@@ -634,6 +637,9 @@ bool CLASS::commit_(size_t size, bool final) NOEXCEPT
             return false;
         }
 
+        if (!staged_ && dirty_ && (target > from))
+            mmap_wire(std::next(memory_map_[Column], from), target - from);
+
         // Committed growth is a new (unnamed) vma; reattribute it.
         if (target > from)
             mmap_name(std::next(memory_map_[Column], from), target - from,
@@ -707,6 +713,9 @@ 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_)
+        mmap_wire(std::next(base, settled), target - settled);
+
     // Convert the settled prefix on the replacement reservation.
     if (!is_zero(settled) &&
         (mmap_settle(replace, settled, opened_[Column], zero) == fail))
@@ -1319,7 +1328,9 @@ void CLASS::head_run_() NOEXCEPT
         // 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).
-        const auto scarcity = head_release && dirty_ &&
+        // 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
@@ -1400,11 +1411,37 @@ void CLASS::head_run_() NOEXCEPT
 // 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<size_t>& CLASS::writer_slot_() NOEXCEPT
+{
+    static std::atomic<size_t> 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_.load()))
+    while (!is_zero(writers_count_()))
         std::this_thread::yield();
 }
 
@@ -1552,11 +1589,15 @@ bool CLASS::release_pages_() NOEXCEPT
         // 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_.load());
+        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],

include/bitcoin/database/impl/memory/mmap_storage.ipp

@@ -205,16 +205,17 @@ void CLASS::prepare(size_t STAGING_ONLY(offset),
     // (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);
+        writers.fetch_add(one);
         if (!transition_.load())
             break;
 
-        writers_.fetch_sub(one);
+        writers.fetch_sub(one);
     }
 
     // A settled head writes through its mapping.
@@ -265,7 +266,7 @@ void CLASS::mark(size_t STAGING_ONLY(offset),
     // 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).
-    writers_.fetch_sub(one);
+    writer_slot_().fetch_sub(one);
 #endif
 }
 

include/bitcoin/database/memory/mmap.hpp

@@ -363,6 +363,9 @@ class mmap
     // the prepare/release bit protocol (see release_pages_).
     bool release_pages_() NOEXCEPT;
     void quiesce_() NOEXCEPT;
+    std::atomic<size_t>& writer_slot_() NOEXCEPT;
+    size_t writers_count_() const NOEXCEPT;
+    void writers_reset_() NOEXCEPT;
     bool share_(size_t transferred) NOEXCEPT;
     void unshare_() NOEXCEPT;
     void declare_released_() NOEXCEPT;
@@ -511,8 +514,11 @@ class mmap
     std::atomic_bool transition_{};
 
     // Writers between prepare and mark (unaged, unlike intent bits), so a
-    // release pass cannot settle under a preempted in-flight write.
-    std::atomic<size_t> writers_{};
+    // release pass cannot settle under a preempted in-flight write. Sharded
+    // by thread on separate lines: the write path touches one, drains sum.
+    struct alignas(64) writer_shard { std::atomic<size_t> count{}; };
+    static constexpr size_t writer_shards = 16;
+    std::array<writer_shard, writer_shards> writers_{};
 
     // Serializes page release against restore (prepare slow path).
     mutable std::mutex restore_mutex_{};

include/bitcoin/database/memory/mstage.hpp

@@ -74,6 +74,12 @@ int file_discard(int fd) NOEXCEPT;
 /// anonymous memory carrying its content (readers never observe zeros).
 int mmap_restore(void* address, size_t size) NOEXCEPT;
 
+/// Pin anonymous pages resident (best effort, no-op where not unprivileged).
+int mmap_wire(void* address, size_t size) NOEXCEPT;
+
+/// Release pinned pages to normal reclaim.
+int mmap_unwire(void* address, size_t size) NOEXCEPT;
+
 /// Full-transfer positional file read/write (false on failure or early eof).
 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,

src/memory/mstage.cpp

@@ -187,11 +187,34 @@ int mmap_resident(const void* address, size_t size,
 
 int mmap_share(void* address, size_t size, int fd, size_t offset) NOEXCEPT
 {
+    mmap_unwire(address, size);
     return ::mmap(address, size, PROT_READ | PROT_WRITE, MAP_SHARED |
         MAP_FIXED, fd, possible_narrow_sign_cast<off_t>(offset)) == MAP_FAILED ?
         -1 : 0;
 }
 
+// Darwin compresses cold anonymous pages, which mincore reports resident, so
+// 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).
+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_unsettle(void* address, size_t size) NOEXCEPT
 {
     // No reserve: an unsettled span can span hundreds of gigabytes (truncate
@@ -268,6 +291,7 @@ int mmap_restore(void* address, size_t size) NOEXCEPT
         return -1;
     }
 
+    mmap_wire(address, size);
     return 0;
 }