Update sig batch to eliminate write contention.

include/bitcoin/node/chasers/chaser_validate.hpp

@@ -87,7 +87,7 @@ class BCN_API chaser_validate
     virtual bool mark_invalids(header_links& prevalids,
         const header_links& invalids, bool startup) NOEXCEPT;
 
-    /// Turnstile (drain/capture exclusion without blocking writers).
+    /// Turnstile (drain/commit exclusion without blocking writers).
     virtual bool enter_capture() NOEXCEPT;
     virtual void exit_capture() NOEXCEPT;
 
@@ -97,7 +97,6 @@ class BCN_API chaser_validate
 
 private:
     using atomic_counter = std::atomic<size_t>;
-    using atomic_counter_ptr = std::shared_ptr<atomic_counter>;
     struct counters
     {
         atomic_counter ecdsa_{};
@@ -110,31 +109,17 @@ class BCN_API chaser_validate
         atomic_counter missed_threshold_{};
     };
 
-    using schnorr_link = database::schnorr_link;
-    using schnorr_link_ptr = std::shared_ptr<schnorr_link>;
-    using cursor = system::chain::threshold::cursor;
     using missed = signatures::miss;
 
     // Capture handlers.
     void do_log(const system::chain::script& missed) NOEXCEPT;
     void do_fire(missed miss, size_t count) NOEXCEPT;
-    bool do_ecdsa(const system::hash_digest& digest,
-        const system::ec_compressed& point, const system::ec_signature& sign,
-        const header_link& link, const atomic_counter_ptr& sequence) NOEXCEPT;
-    bool do_schnorr(const system::hash_digest& digest,
-        const system::ec_xonly& point, const system::ec_signature& sign,
-        const header_link& link) NOEXCEPT;
-    bool do_multisig(const system::hash_digest& digest,
-        const std::span<const system::ec_compressed>& points,
-        const std::span<const system::ec_signature>& signs,
-        const header_link& link, const atomic_counter_ptr& sequence) NOEXCEPT;
-    bool do_threshold(const system::hash_digest& digest,
-        const system::ec_xonly& point, const system::ec_signature& sign,
-        const schnorr_link_ptr& fk_ptr, const header_link& link) NOEXCEPT;
-    cursor open_threshold(size_t rows, const header_link& link) NOEXCEPT;
 
     // Capture helpers.
     signatures get_capture(const header_link& link) NOEXCEPT;
+    code commit_capture(bool& batched, const header_link& link) NOEXCEPT;
+    void clear_capture() NOEXCEPT;
+    void do_purge_capture() NOEXCEPT;
     std::string log_ratio(const std::string& name, size_t numerator,
         size_t denominator) const NOEXCEPT;
     void log_captures() const NOEXCEPT;

include/bitcoin/node/error.hpp

@@ -114,10 +114,7 @@ enum error_t : uint8_t
     batch2,
     batch3,
     batch4,
-    batch5,
-    batch6,
-    batch7,
-    batch8
+    batch5
 };
 
 // No current need for error_code equivalence mapping.

src/chasers/chaser_validate.cpp

@@ -305,6 +305,12 @@ void chaser_validate::complete_block(const code& ec, const header_link& link,
     {
         process_batch(true);
     }
+
+    // Batching has subsided, release slab capacity (no-op once released).
+    if (current)
+    {
+        POST(do_purge_capture);
+    }
 }
 
 void chaser_validate::notify_block(const code& ec, size_t height, 

src/chasers/chaser_validate_batch.cpp

@@ -59,12 +59,12 @@ void chaser_validate::process_batch(bool residual) NOEXCEPT
         return;
 
     // Claim the drain (at most one, losers rely on the winner).
-    // Arrivals now pause at enter_capture (briefly, until verifying_).
     if (draining_.exchange(true))
         return;
 
-    // Wait for in-flight captures to complete or abandon on close.
-    // Bounded: arrivals pause rather than enter, so writers_ only drains.
+    // Wait for in-flight commits to complete or abandon on close.
+    // Bounded: the writer epoch spans only the per-block slab commit, so
+    // writers_ only drains (arriving commits divert to in-place verify).
     while (is_nonzero(writers_.load()))
     {
         if (closed())
@@ -76,22 +76,17 @@ void chaser_validate::process_batch(bool residual) NOEXCEPT
         std::this_thread::yield();
     }
 
-    // Divert paused/new arrivals to inline for the verify duration.
-    ////verifying_.store(true);
-
     // Batch tables are now quiescent (no writers admitted, none in flight).
     // ========================================================================
 
     // Retest under the claim, another drain may have just emptied the tables.
     if (!is_mature(residual))
     {
-        ////verifying_.store(false);
         draining_.store(false);
         return;
     }
 
     const auto ec = do_process_batch(false);
-    ////verifying_.store(false);
     draining_.store(false);
     if (ec == network::error::operation_canceled)
         return;
@@ -269,22 +264,8 @@ std::string chaser_validate::log_rate(const std::string& name,
 
 bool chaser_validate::enter_capture() NOEXCEPT
 {
-#if defined(DISABLED)
-    // Blocked by batch (also requires other verifying_ enabled).
-    while (true)
-    {
-        ++writers_;
-        if (!draining_.load())
-            return true;
 
-        --writers_;
-        if (verifying_.load() || closed())
-            return false;
-
-        std::this_thread::yield();
-    }
-#else
-    // Bypassed by batch (faster overall, but high bypass).
+    // Capture bypassed by batch.
     ++writers_;
     if (draining_.load())
     {
@@ -293,7 +274,6 @@ bool chaser_validate::enter_capture() NOEXCEPT
     }
 
     return true;
-#endif
 }
 
 void chaser_validate::exit_capture() NOEXCEPT

src/chasers/chaser_validate_capture.cpp

@@ -30,9 +30,6 @@ using namespace system::chain;
 using namespace database;
 using namespace std::placeholders;
 
-// Shared pointers required for lifetime in handler parameters.
-BC_PUSH_WARNING(NO_VALUE_OR_CONST_REF_SHARED_PTR)
-BC_PUSH_WARNING(SMART_PTR_NOT_NEEDED)
 BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT)
 
 // Capture handlers.
@@ -63,68 +60,6 @@ void chaser_validate::do_fire(missed miss, size_t count) NOEXCEPT
     }
 }
 
-bool chaser_validate::do_ecdsa(const hash_digest& digest,
-    const ec_compressed& point, const ec_signature& sign,
-    const header_link& link, const atomic_counter_ptr& sequence) NOEXCEPT
-{
-    ++counters_.ecdsa_;
-    const auto id = (*sequence)++;
-    if (is_limited<uint16_t>(id)) return false;
-    const auto group = narrow_cast<uint16_t>(id);
-    const auto set = archive().set_signature(digest, point, sign, group, link);
-    if (!set) fault(error::batch5);
-    return set;
-}
-
-bool chaser_validate::do_schnorr(const hash_digest& digest,
-    const ec_xonly& point, const ec_signature& sign,
-    const header_link& link) NOEXCEPT
-{
-    ++counters_.schnorr_;
-    const auto set = archive().set_signature(digest, point, sign, link);
-    if (!set) fault(error::batch6);
-    return set;
-}
-
-bool chaser_validate::do_multisig(const hash_digest& digest,
-    const std::span<const system::ec_compressed>& points,
-    const std::span<const system::ec_signature>& signs,
-    const header_link& link, const atomic_counter_ptr& sequence) NOEXCEPT
-{
-    counters_.multisig_ += points.size();
-    const auto id = (*sequence)++;
-    if (is_limited<uint16_t>(id)) return false;
-    const auto group = narrow_cast<uint16_t>(id);
-    const auto set = archive().set_signatures(digest, points, signs, group, link);
-    if (!set) fault(error::batch7);
-    return set;
-}
-
-bool chaser_validate::do_threshold(const hash_digest& digest,
-    const ec_xonly& point, const ec_signature& sign,
-    const schnorr_link_ptr& fk_ptr, const header_link& link) NOEXCEPT
-{
-    auto set = archive().set_signature((*fk_ptr)++, digest, point, sign, link);
-    if (!set) fault(error::batch8);
-    return set;
-}
-
-chaser_validate::cursor chaser_validate::open_threshold(size_t rows,
-    const header_link& link) NOEXCEPT
-{
-    auto first = archive().allocate_signatures(rows);
-    const auto fk = emplace_shared<schnorr_link>(first);
-    if (fk->is_terminal())
-        return {};
-
-    counters_.threshold_ += rows;
-    return
-    {
-        .put = BIND_THIS(do_threshold, _1, _2, _3, fk, link),
-        .rows = rows
-    };
-}
-
 // Capture helpers.
 // ----------------------------------------------------------------------------
 // private
@@ -134,19 +69,87 @@ signatures chaser_validate::get_capture(const header_link& link) NOEXCEPT
     if (!batch_enabled_ || link.is_terminal() || is_current_header(link))
         return { false };
 
-    const auto sequence = to_shared<atomic_counter>();
+    // The capture populates this thread's accumulators (commit_capture consumes).
     return signatures
     {
         .enabled = true,
         .log = BIND_THIS(do_log, _1),
-        .fire = BIND_THIS(do_fire, _1, _2),
-        .ecdsa = BIND_THIS(do_ecdsa, _1, _2, _3, link, sequence),
-        .schnorr = BIND_THIS(do_schnorr, _1, _2, _3, link),
-        .multisig = BIND_THIS(do_multisig, _1, _2, _3, link, sequence),
-        .threshold = BIND_THIS(open_threshold, _1, link)
+        .fire = BIND_THIS(do_fire, _1, _2)
     };
 }
 
+// Commit this thread's captured signatures as the block's batch rows. All
+// batch table state is written inside the commit epoch (turnstile). When
+// diverted by a drain, or upon store decline, the rows are verified in
+// place, equivalent to the inline evaluation their capture fabricated
+// (batched is cleared so the block completes by the non-batched path).
+code chaser_validate::commit_capture(bool& batched,
+    const header_link& link) NOEXCEPT
+{
+    code ec{};
+    auto committed = false;
+    auto& ecdsa = signatures::ecdsa_rows();
+    auto& schnorr = signatures::schnorr_rows();
+
+    if (enter_capture())
+    {
+        auto& query = archive();
+        committed =
+            query.set_signatures(ecdsa, link) &&
+            query.set_signatures(schnorr, link) &&
+            query.set_prevalid(link);
+        exit_capture();
+
+        // Store decline (e.g. disk full), recoverable once faulted.
+        if (!committed)
+            fault(error::batch5);
+    }
+
+    const auto thresholds = schnorr.thresholds();
+    const auto singles = schnorr.rows().size() - thresholds;
+
+    if (committed)
+    {
+        counters_.ecdsa_ += ecdsa.singles();
+        counters_.multisig_ += ecdsa.multisig_keys();
+        counters_.schnorr_ += singles;
+        counters_.threshold_ += thresholds;
+    }
+    else
+    {
+        counters_.missed_ecdsa_ += ecdsa.singles();
+        counters_.missed_multisig_ += ecdsa.multisig_keys();
+        counters_.missed_schnorr_ += singles;
+        counters_.missed_threshold_ += thresholds;
+
+        // Block validity remains fully determined.
+        batched = false;
+        if (!ecdsa.verify() || !schnorr.verify())
+            ec = system::error::invalid_signature;
+    }
+
+    ecdsa.clear();
+    schnorr.clear();
+    return ec;
+}
+
+// Discard this thread's captured signatures (script failure preempted commit).
+void chaser_validate::clear_capture() NOEXCEPT
+{
+    signatures::ecdsa_rows().clear();
+    signatures::schnorr_rows().clear();
+}
+
+// Release all threads' accumulator capacity (batching subsided). Safe on the
+// strand with an empty backlog: accumulators are populated only within
+// validate_block, and the strand is the sole poster of validations.
+void chaser_validate::do_purge_capture() NOEXCEPT
+{
+    BC_ASSERT(stranded());
+    if (is_zero(validate_backlog_.load()))
+        signatures::purge();
+}
+
 std::string chaser_validate::log_ratio(const std::string& name,
     size_t numerator, size_t denominator) const NOEXCEPT
 {
@@ -167,11 +170,9 @@ void chaser_validate::log_captures() const NOEXCEPT
     LOGN(log_ratio("Capture schnorr..", counters_.schnorr_,
         counters_.schnorr_   + counters_.missed_schnorr_));
     LOGN(log_ratio("Capture threshold", counters_.threshold_,
-        counters_.threshold_ + zero));
+        counters_.threshold_ + counters_.missed_threshold_));
 }
 
-BC_POP_WARNING()
-BC_POP_WARNING()
 BC_POP_WARNING()
 
 } // namespace node

src/chasers/chaser_validate_parallel.cpp

@@ -114,30 +114,25 @@ code chaser_validate::validate(bool& batched, bool& capturing, bool bypass,
         if ((ec = block.accept(ctx, subsidy_interval_, initial_subsidy_)))
             return ec;
 
-        // This critical section is mutually-exclusive with batch verification
-        // (turnstile). All batch table state written inside the capture epoch.
-        // ====================================================================
-
-        // Enter the capture turnstile. If drain is in progress, full validate.
-        const auto entered = enter_capture();
-
-        // Initialize signature capture.
-        const auto capture = get_capture(entered ? link : header_link{});
-        capturing = entered ? capture.enabled : true;
+        // Initialize signature capture (appends to this thread's accumulators).
+        const auto capture = get_capture(link);
+        capturing = capture.enabled;
 
         ec = block.connect(ctx, capture);
 
         // At least one signature batch was attempted (batch completion).
         batched = capture.batched;
 
-        // Mark block as valid contingent on batch verification.
-        if (!ec && batched && !query.set_prevalid(link))
-            ec = error::validate6;
-
-        if (entered)
-            exit_capture();
-
-        // ====================================================================
+        // Commit (or discard) the captured signatures, marking the block prevalid
+        // contingent on batch verification. The commit epoch is mutually
+        // exclusive with batch verification (turnstile).
+        if (capturing)
+        {
+            if (!ec && batched)
+                ec = commit_capture(batched, link);
+            else
+                clear_capture();
+        }
 
         if (ec)
             return ec;

src/error.cpp

@@ -104,10 +104,7 @@ DEFINE_ERROR_T_MESSAGE_MAP(error)
     { batch2, "batch2" },
     { batch3, "batch3" },
     { batch4, "batch4" },
-    { batch5, "batch5" },
-    { batch6, "batch6" },
-    { batch7, "batch7" },
-    { batch8, "batch8" }
+    { batch5, "batch5" }
 };
 
 DEFINE_ERROR_T_CATEGORY(error, "node", "node code")