Batch control.

include/bitcoin/node/chase.hpp

@@ -112,13 +112,13 @@ enum class chase
     /// Issued by 'block_in_31800' and handled by 'organize'.
     unchecked,
 
+    /// A downloaded window is completed by check (height_t).
+    /// Issued by 'check' and handled by 'validate'.
+    windowed,
+
     /// Accept/Connect.
     /// -----------------------------------------------------------------------
 
-    /// A branch has become provisionally valid, download unblocked (height_t).
-    /// Issued by 'validate' and handled by 'check'.
-    prevalid,
-
     /// A branch has become valid (height_t).
     /// Issued by 'validate' and handled by 'check', 'confirm', 'snapshot'.
     valid,

include/bitcoin/node/chasers/chaser_validate.hpp

@@ -20,7 +20,6 @@
 #define LIBBITCOIN_NODE_CHASERS_CHASER_VALIDATE_HPP
 
 #include <atomic>
-#include <mutex>
 #include <bitcoin/node/chasers/chaser.hpp>
 #include <bitcoin/node/define.hpp>
 
@@ -78,26 +77,41 @@ class BCN_API chaser_validate
     virtual void notify_block(const code& ec, size_t height,
         const header_link& link, bool bypass, bool startup=false) NOEXCEPT;
 
-    /// Batching.
+    /// Batching (lock-free, self-serviced by completing pool threads).
+    /// Batch state is the store: sig tables carry rows, prevalid table
+    /// carries the per-block link set (write-through, crash-durable).
     virtual code start_batch() NOEXCEPT;
-    virtual void close_batch() NOEXCEPT;
-    virtual void push_batch(const header_link& link, size_t height) NOEXCEPT;
     virtual void process_batch(bool residual) NOEXCEPT;
     virtual code do_process_batch(bool startup) NOEXCEPT;
-    virtual bool mark_valids(bool startup) NOEXCEPT;
-    virtual bool mark_invalids(const header_links& invalids,
-        bool startup) NOEXCEPT;
+    virtual bool mark_valids(header_links& prevalids, bool startup) NOEXCEPT;
+    virtual bool mark_invalids(header_links& prevalids,
+        const header_links& invalids, bool startup) NOEXCEPT;
+
+    /// Turnstile (drain/capture exclusion without blocking writers).
+    virtual bool enter_capture() NOEXCEPT;
+    virtual void exit_capture() NOEXCEPT;
 
     // Override base class strand because it sits on the network thread pool.
     network::asio::strand& strand() NOEXCEPT override;
     bool stranded() const NOEXCEPT override;
 
 private:
-    static constexpr auto relaxed = std::memory_order_relaxed;
-    using schnorr_link = database::schnorr_link;
-    using schnorr_link_ptr = std::shared_ptr<schnorr_link>;
     using atomic_counter = std::atomic<size_t>;
     using atomic_counter_ptr = std::shared_ptr<atomic_counter>;
+    struct counters
+    {
+        atomic_counter ecdsa_{};
+        atomic_counter schnorr_{};
+        atomic_counter multisig_{};
+        atomic_counter threshold_{};
+        atomic_counter missed_ecdsa_{};
+        atomic_counter missed_schnorr_{};
+        atomic_counter missed_multisig_{};
+        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;
 
@@ -128,31 +142,24 @@ class BCN_API chaser_validate
     // Batching helpers.
     bool is_residual() NOEXCEPT;
     bool is_mature(bool residual) NOEXCEPT;
-    std::string log_rate(const std::string& name, size_t numerator,
-        size_t denominator) const NOEXCEPT;
+    std::string log_rate(const std::string& name, size_t signatures,
+        size_t milliseconds) const NOEXCEPT;
 
-    // These are protected by strand.
-    header_links batched_{};
-    header_links invalids_{};
+    // This is not thread safe.
     network::threadpool validation_threadpool_;
 
     // These are thread safe.
-    stopper stopping_{};
-    std::shared_mutex mutex_{};
-    std::atomic<size_t> ecdsa_{};
-    std::atomic<size_t> schnorr_{};
-    std::atomic<size_t> multisig_{};
-    std::atomic<size_t> threshold_{};
-    std::atomic<size_t> missed_ecdsa_{};
-    std::atomic<size_t> missed_schnorr_{};
-    std::atomic<size_t> missed_multisig_{};
-    std::atomic<size_t> missed_threshold_{};
-    std::atomic<size_t> validate_backlog_{};
-    std::atomic<size_t> batch_backlog_{};
+    network::asio::strand validation_strand_;
+    atomic_counter validate_backlog_{};
+    std::atomic_bool disk_recovering_{};
+    std::atomic_bool window_archived_{};
     std::atomic_bool maximum_posted_{};
-    std::atomic_bool recovering_{};
+    std::atomic_bool draining_{};
+    atomic_counter writers_{};
+    counters counters_{};
+    stopper stopping_{};
 
-    network::asio::strand validation_strand_;
+    // These are thread safe.
     const uint32_t subsidy_interval_;
     const uint64_t initial_subsidy_;
     const size_t silent_start_height_;

include/bitcoin/node/error.hpp

@@ -92,6 +92,7 @@ enum error_t : uint8_t
     validate7,
     validate8,
     validate9,
+    validate10,
     confirm1,
     confirm2,
     confirm3,

src/chasers/chaser_check.cpp

@@ -88,6 +88,9 @@ map_ptr chaser_check::split(const map_ptr& map) NOEXCEPT
 // start/stop
 // ----------------------------------------------------------------------------
 
+// `position` tracks the contiguous chain of associated candidate blocks.
+// `requested` tracks the last height of the current download window.
+// `advanced` tracks `requested` less window blocks not yet validated.
 code chaser_check::start() NOEXCEPT
 {
     start_tracking();
@@ -155,7 +158,7 @@ bool chaser_check::handle_chase(const code&, chase event_,
             break;
         }
         case chase::valid:
-        case chase::prevalid:
+        ////case chase::prevalid:
         {
             BC_ASSERT(std::holds_alternative<height_t>(value));
             POST(do_advanced, std::get<height_t>(value));
@@ -360,7 +363,6 @@ void chaser_check::do_advanced(height_t) NOEXCEPT
     BC_ASSERT(stranded());
 
     // Validations are not ordered, so accumulate vs. compare height.
-    // Advancement through window is a quantity, not dedicated to a branch.
     ++advanced_;
 
     // The full count of requested hashes has been validated.
@@ -389,8 +391,14 @@ void chaser_check::do_bump(height_t) NOEXCEPT
     // TODO: query.is_associated() is expensive (hashmap search).
     // Skip checked blocks starting immediately after last checked.
     while (!closed() && query.is_associated(query.to_candidate(++height)))
+    {
         set_position(height);
 
+        // Notify validator that no more blocks are coming.
+        if (height == requested_)
+            notify(error::success, chase::windowed, height);
+    }
+
     do_headers({});
 }
 
@@ -484,21 +492,25 @@ size_t chaser_check::set_unassociated() NOEXCEPT
     if (closed() || purging())
         return {};
 
-    // Defer new work issuance until gaps filled and validation caught up.
-    if (position() < requested_ || advanced_ < requested_)
+    // Defer new work issuance until gaps filled.
+    if (position() < requested_)
+        return {};
+
+    // Defer new work until validation caught up to request.
+    if (advanced_ < requested_)
         return {};
 
     // Inventory size gets set only once.
     if (is_zero(inventory_))
         if (is_zero((inventory_ = get_inventory_size())))
-            return zero;
+            return {};
 
     // Due to previous downloads, validation can race ahead of last request.
     // The last request (requested_) stops at the last gap in the window, but
     // validation continues until the next gap. Start next scan above validated
     // not last requested, since all between are already downloaded.
     const auto& query = archive();
-    const auto requested = requested_;
+    const auto previous = requested_;
     const auto step = ceilinged_add(position(), maximum_concurrency_);
     const auto stop = std::min(step, maximum_height_);
     size_t count{};
@@ -518,7 +530,7 @@ size_t chaser_check::set_unassociated() NOEXCEPT
 
     LOGN("Advance by ("
         << maximum_concurrency_ << ") above ("
-        << requested << ") from ("
+        << previous << ") from ("
         << position() << ") stop ("
         << stop << ") found ("
         << count << ") last ("
@@ -530,7 +542,7 @@ size_t chaser_check::set_unassociated() NOEXCEPT
 size_t chaser_check::get_inventory_size() const NOEXCEPT
 {
     if (is_zero(connections_) || !is_current_chain(false))
-        return zero;
+        return {};
 
     const auto& query = archive();
     const auto fork = query.get_fork();

src/chasers/chaser_validate.cpp

@@ -18,7 +18,6 @@
  */
 #include <bitcoin/node/chasers/chaser_validate.hpp>
 
-#include <shared_mutex>
 #include <bitcoin/node/chasers/chaser.hpp>
 #include <bitcoin/node/define.hpp>
 #include <bitcoin/node/full_node.hpp>
@@ -75,31 +74,54 @@ bool chaser_validate::handle_chase(const code&, chase event_,
     // Because in-flight blocks are lost, reset position when backlog clears.
     if (event_ == chase::unfull)
     {
-        recovering_.store(true);
+        disk_recovering_.store(true);
         return true;
     }
 
+    // Record durable state before the suspension gate. A consequence is that
+    // recovery with window_archived_ drains small batches until next checked.
+    if (event_ == chase::windowed)
+        window_archived_.store(true);
+    else if (event_ == chase::checked)
+        window_archived_.store(false);
+
     // Stop generating query during suspension.
     // Incoming events may already be flushed to the strand at this point.
     if (suspended())
         return true;
 
     switch (event_)
     {
-        case chase::resume:
         case chase::start:
         case chase::bump:
         {
             POST(do_bump, height_t{});
             break;
         }
+        case chase::resume:
+        {
+            // A windowed drain may have been elided during suspension.
+            process_batch(is_residual());
+            POST(do_bump, height_t{});
+            break;
+        }
         case chase::checked:
         {
             // value is checked block height.
             BC_ASSERT(std::holds_alternative<height_t>(value));
             POST(do_checked, std::get<height_t>(value));
             break;
         }
+        case chase::windowed:
+        {
+            // value is last height in window. Called directly (not posted):
+            // the drain must never depend on scheduling, which saturated
+            // validations can exhaust for the entire window (strand for
+            // admission-side control, never for release-side).
+            BC_ASSERT(std::holds_alternative<height_t>(value));
+            process_batch(is_residual());
+            break;
+        }
         case chase::regressed:
         case chase::disorganized:
         {
@@ -146,9 +168,9 @@ void chaser_validate::do_bump(height_t) NOEXCEPT
 {
     BC_ASSERT(stranded());
 
-    if (recovering_.load() && is_zero(validate_backlog_.load()))
+    if (disk_recovering_.load() && is_zero(validate_backlog_.load()))
     {
-        recovering_.store(false);
+        disk_recovering_.store(false);
         set_position(archive().get_fork());
     }
 
@@ -263,28 +285,25 @@ void chaser_validate::complete_block(const code& ec, const header_link& link,
         notify_block({}, height, link, bypass);
     }
 
-    // Batch jobs (all posting from unstranded).
+    // Batch jobs (self-serviced from any thread).
     // ------------------------------------------------------------------------
 
     if (closed() || !batch_enabled_)
         return;
 
-    // Queue block and process batch if ready.
     if (batched)
     {
-        ++batch_backlog_;
-        POST(push_batch, link, height);
+        process_batch(is_residual());
         return;
     }
 
     // Capturing disabled when confirmed chain current (and not under bypass).
-    // When not in effect must drain last batch by last block validation.
     const auto current = !capturing && !bypass;
 
-    // Drain batch when recent (current, or maximum reached without backlog).
+    // Drain batch when recent (current, or window/maximum without backlog).
     if (current || is_residual())
     {
-        POST(process_batch, true);
+        process_batch(true);
     }
 }
 
@@ -313,9 +332,6 @@ void chaser_validate::notify_block(const code& ec, size_t height,
 
 void chaser_validate::stopping(const code& ec) NOEXCEPT
 {
-    // closed() is now true so time to bump batched_ drain.
-    POST(close_batch);
-
     // Stop long-running batch validations.
     stopping_.store(true);