Make read/write sequencing consistent across channels.

include/bitcoin/node/impl/sessions/session_peer.ipp

@@ -50,7 +50,7 @@ inline void CLASS::attach_handshake(const channel_ptr& channel,
     network::result_handler&& handler) NOEXCEPT
 {
     BC_ASSERT(channel->stranded());
-    BC_ASSERT(channel->paused());
+    BC_ASSERT(!channel->held());
 
     // Set the current top for version protocol, before handshake.
     const auto top = this->archive().get_top_confirmed();
@@ -65,7 +65,7 @@ TEMPLATE
 inline void CLASS::attach_protocols(const channel_ptr& channel) NOEXCEPT
 {
     BC_ASSERT(channel->stranded());
-    BC_ASSERT(channel->paused());
+    BC_ASSERT(channel->held());
 
     using namespace system;
     using namespace network;

include/bitcoin/node/protocols/protocol_block_out_106.hpp

@@ -19,7 +19,6 @@
 #ifndef LIBBITCOIN_NODE_PROTOCOLS_PROTOCOL_BLOCK_OUT_106_HPP
 #define LIBBITCOIN_NODE_PROTOCOLS_PROTOCOL_BLOCK_OUT_106_HPP
 
-#include <deque>
 #include <bitcoin/node/define.hpp>
 #include <bitcoin/node/protocols/protocol_peer.hpp>
 
@@ -40,8 +39,7 @@ class BCN_API protocol_block_out_106
             session->system_settings().top_checkpoint().height()),
         node_pruned_(session->node_settings().limited_blocks),
         node_witness_(session->node_settings().provide_witness),
-        allow_overlapped_(session->node_settings().allow_overlapped),
-        network::tracker<protocol_block_out_106>(session->log)
+       network::tracker<protocol_block_out_106>(session->log)
     {
     }
 
@@ -71,13 +69,15 @@ class BCN_API protocol_block_out_106
         const get_blocks::cptr& message) NOEXCEPT;
     virtual bool handle_receive_get_data(const code& ec,
         const get_data::cptr& message) NOEXCEPT;
-    virtual void send_block(const code& ec) NOEXCEPT;
+    virtual void send_block(const code& ec, size_t index,
+        const get_data::cptr& message, const gate_t::ptr& gate) NOEXCEPT;
 
     /// The item cannot be served, stops the channel and returns false.
     virtual bool handle_unservable(const inventory_item& item) NOEXCEPT;
 
     /// Send any unservable items accumulated above, false if none.
-    virtual bool report_unservable() NOEXCEPT;
+    virtual bool report_unservable(size_t index,
+        const get_data::cptr& message, const gate_t::ptr& gate) NOEXCEPT;
 
 private:
     using inventory = network::messages::peer::inventory;
@@ -86,16 +86,11 @@ class BCN_API protocol_block_out_106
         const database::header_link& link) NOEXCEPT;
     bool is_under_checkpoint(const database::header_link& link) NOEXCEPT;
     inventory create_inventory(const get_blocks& locator) const NOEXCEPT;
-    void merge_inventory(const inventory_items& items) NOEXCEPT;
 
     // These are thread safe.
     const size_t top_checkpoint_height_;
     const bool node_pruned_;
     const bool node_witness_;
-    const bool allow_overlapped_;
-
-    // This is protected by strand.
-    std::deque<inventory_item> backlog_{};
 };
 
 } // namespace node

include/bitcoin/node/protocols/protocol_block_out_70001.hpp

@@ -45,7 +45,8 @@ class BCN_API protocol_block_out_70001
     bool handle_unservable(const inventory_item& item) NOEXCEPT override;
 
     /// Replies not_found with the accumulated items, false if none.
-    bool report_unservable() NOEXCEPT override;
+    bool report_unservable(size_t index, const get_data::cptr& message,
+        const gate_t::ptr& gate) NOEXCEPT override;
 
 private:
     // This is thread safe.

include/bitcoin/node/protocols/protocol_filter_out_70015.hpp

@@ -52,7 +52,8 @@ class BCN_API protocol_filter_out_70015
 
 private:
     using ancestry_ptr = std::shared_ptr<database::header_links>;
-    void send_filter(const code& ec, const ancestry_ptr& ancestry) NOEXCEPT;
+    void send_filter(const code& ec, const ancestry_ptr& ancestry,
+        const gate_t::ptr& gate) NOEXCEPT;
 };
 
 } // namespace node

include/bitcoin/node/protocols/protocol_peer.hpp

@@ -37,6 +37,7 @@ class BCN_API protocol_peer
 public:
     // Replace base class channel_t (network::channel_peer). 
     using channel_t = node::channel_peer;
+    using gate_t = network::channel_peer::gate_t;
 
 protected:
     /// Constructors.

include/bitcoin/node/protocols/protocol_transaction_in_106.hpp

@@ -56,8 +56,10 @@ class BCN_API protocol_transaction_in_106
     /// Accept incoming transaction message.
     virtual bool handle_receive_transaction(const code& ec,
         const network::messages::peer::transaction::cptr& message) NOEXCEPT;
-    virtual void handle_submit(const code& ec, size_t index) NOEXCEPT;
-    virtual void do_handle_submit(const code& ec) NOEXCEPT;
+    virtual void handle_submit(const code& ec, size_t index,
+        const gate_t::ptr& gate) NOEXCEPT;
+    virtual void do_handle_submit(const code& ec,
+        const gate_t::ptr& gate) NOEXCEPT;
 
 private:
     /// Squash duplicates and provide constant time retrieval.

include/bitcoin/node/protocols/protocol_transaction_in_70013.hpp

@@ -41,7 +41,8 @@ class BCN_API protocol_transaction_in_70013
 
 protected:
     /// The peer is advised of the fee rate and of suspension (see fee_filter).
-    void do_handle_submit(const code& ec) NOEXCEPT override;
+    void do_handle_submit(const code& ec,
+        const gate_t::ptr& gate) NOEXCEPT override;
 };
 
 } // namespace node

include/bitcoin/node/protocols/protocol_transaction_out_106.hpp

@@ -59,15 +59,17 @@ class BCN_API protocol_transaction_out_106
     virtual bool handle_receive_get_data(const code& ec,
         const network::messages::peer::get_data::cptr& message) NOEXCEPT;
     virtual void send_transaction(const code& ec, size_t index,
-        const network::messages::peer::get_data::cptr& message) NOEXCEPT;
+        const network::messages::peer::get_data::cptr& message,
+        const gate_t::ptr& gate) NOEXCEPT;
 
     /// The item cannot be served, false terminates the send loop.
     virtual bool handle_unservable(
         const network::messages::peer::inventory_item& item) NOEXCEPT;
 
     /// Replies not_found with the accumulated items, false if none.
     virtual bool report_unservable(size_t index,
-        const network::messages::peer::get_data::cptr& message) NOEXCEPT;
+        const network::messages::peer::get_data::cptr& message,
+        const gate_t::ptr& gate) NOEXCEPT;
 
     virtual bool announce(const system::hash_digest& hash) NOEXCEPT;
 

include/bitcoin/node/protocols/protocol_transaction_out_70001.hpp

@@ -47,8 +47,8 @@ class BCN_API protocol_transaction_out_70001
 
     /// Replies not_found with the accumulated items, false if none.
     bool report_unservable(size_t index,
-        const network::messages::peer::get_data::cptr& message) NOEXCEPT
-        override;
+        const network::messages::peer::get_data::cptr& message,
+        const gate_t::ptr& gate) NOEXCEPT override;
 
 private:
     // This is thread safe.

include/bitcoin/node/settings.hpp

@@ -39,7 +39,6 @@ class BCN_API settings
     bool headers_first;
     bool thread_priority;
     bool memory_priority;
-    bool allow_overlapped;
     bool provide_blocks;
     bool require_blocks;
     bool provide_witness;

src/protocols/protocol_block_out_106.cpp

@@ -163,56 +163,32 @@ bool protocol_block_out_106::handle_receive_get_data(const code& ec,
     if (is_zero(size))
         return true;
 
-    const auto total = ceilinged_add(backlog_.size(), size);
-    if (total > network::messages::peer::max_inventory)
-    {
-        LOGR("Blocks requested (" << total << ") exceeds inv limit ["
-            << opposite() << "].");
-        stop(network::error::protocol_violation);
-        return false;
-    }
-
-    // Satoshi sends overlapping get_data requests, but assumes that the
-    // recipient is blocking *all traffic* until the previous is completed.
-    // So to prevent frequent drops of satoshi peers, and not let one protocol
-    // block all others, we must accumulate the requests into a backlog. If the
-    // backlog exceeds the *individual* message limit we drop the peer.
-    const auto idle = backlog_.empty();
-    if (!allow_overlapped_ && !idle)
-    {
-        LOGR("Overlapping block requests [" << opposite() << "].");
-        stop(network::error::protocol_violation);
-        return false;
-    }
-
-    // Append the new inventory the request queue.
-    merge_inventory(message->items);
-
-    // Bump the idle async send loop if no pending send.
-    if (idle)
-        send_block(error::success);
-
+    send_block(error::success, zero, message, gate());
     return true;
 }
 
 // Outbound (block).
 // ----------------------------------------------------------------------------
 
-void protocol_block_out_106::send_block(const code& ec) NOEXCEPT
+void protocol_block_out_106::send_block(const code& ec, size_t index,
+    const get_data::cptr& message, const gate_t::ptr& gate) NOEXCEPT
 {
     BC_ASSERT(stranded());
     if (stopped(ec))
         return;
 
     const auto& query = archive();
 
-    // Drain unservable items from the front of the backlog. The derived
+    // Drain unservable items, skipping non-block inventory. The derived
     // protocol accumulates them if it reports them, and otherwise stops the
-    // channel on the first. Copied because each is handled after the pop.
+    // channel on the first.
     database::header_link link{};
-    while (!backlog_.empty())
+    for (; index < message->items.size(); ++index)
     {
-        const auto item = backlog_.front();
+        const auto& item = message->items.at(index);
+        if (!item.is_block())
+            continue;
+
         if (item.is_witness_type() && !node_witness_)
         {
             LOGR("Unsupported witness get_data from [" << opposite() << "].");
@@ -224,18 +200,18 @@ void protocol_block_out_106::send_block(const code& ec) NOEXCEPT
         if (is_servable(item, link))
             break;
 
-        backlog_.pop_front();
         if (!handle_unservable(item))
             return;
     }
 
     // The report resumes this loop on completion, so it precedes the block.
-    if (report_unservable())
+    if (report_unservable(index, message, gate))
         return;
 
-    if (backlog_.empty()) return;
+    if (index >= message->items.size())
+        return;
 
-    const auto item = backlog_.front();
+    const auto& item = message->items.at(index);
     const auto witness = item.is_witness_type();
     const auto start = logger::now();
     messages::peer::block out
@@ -249,16 +225,14 @@ void protocol_block_out_106::send_block(const code& ec) NOEXCEPT
         LOGV("Requested block " << encode_hash(item.hash) << " from ["
             << opposite() << "] not obtained.");
 
-        backlog_.pop_front();
         if (handle_unservable(item))
-            report_unservable();
+            report_unservable(add1(index), message, gate);
 
         return;
     }
 
-    backlog_.pop_front();
     span<microseconds>(events::block_usecs, start);
-    SEND(std::move(out), send_block, _1);
+    SEND(std::move(out), send_block, _1, add1(index), message, gate);
 }
 
 // The checkpoint, milestone and association queries assume an archived header.
@@ -308,7 +282,8 @@ bool protocol_block_out_106::handle_unservable(
 }
 
 // There is nothing to report below bip37, the channel is stopped above.
-bool protocol_block_out_106::report_unservable() NOEXCEPT
+bool protocol_block_out_106::report_unservable(size_t,
+    const get_data::cptr&, const gate_t::ptr&) NOEXCEPT
 {
     BC_ASSERT(stranded());
     return false;
@@ -317,13 +292,6 @@ bool protocol_block_out_106::report_unservable() NOEXCEPT
 // utilities
 // ----------------------------------------------------------------------------
 
-void protocol_block_out_106::merge_inventory(
-    const inventory_items& items) NOEXCEPT
-{
-    for (const auto& item: items)
-        if (item.is_block())
-            backlog_.push_back(item);
-}
 
 protocol_block_out_106::inventory protocol_block_out_106::create_inventory(
     const get_blocks& locator) const NOEXCEPT

src/protocols/protocol_block_out_70001.cpp

@@ -52,7 +52,8 @@ bool protocol_block_out_70001::handle_unservable(
 
 // The items are answered and the send loop resumed, as with a block, so
 // nothing is produced until the prior write completes.
-bool protocol_block_out_70001::report_unservable() NOEXCEPT
+bool protocol_block_out_70001::report_unservable(size_t index,
+    const get_data::cptr& message, const gate_t::ptr& gate) NOEXCEPT
 {
     BC_ASSERT(stranded());
 
@@ -62,7 +63,7 @@ bool protocol_block_out_70001::report_unservable() NOEXCEPT
     auto items = std::move(unservable_);
     unservable_.clear();
 
-    SEND(not_found{ std::move(items) }, send_block, _1);
+    SEND(not_found{ std::move(items) }, send_block, _1, index, message, gate);
     return true;
 }
 

src/protocols/protocol_filter_out_70015.cpp

@@ -220,25 +220,20 @@ bool protocol_filter_out_70015::handle_receive_get_filters(const code& ec,
         return false;
     }
 
-    // Post so the completion resubscribe runs outside the current notify().
     span<milliseconds>(events::ancestry_msecs, start);
-    POST(send_filter, error::success, ancestry);
-    return false;
+    send_filter(error::success, ancestry, gate());
+    return true;
 }
 
 void protocol_filter_out_70015::send_filter(const code& ec,
-    const ancestry_ptr& ancestry) NOEXCEPT
+    const ancestry_ptr& ancestry, const gate_t::ptr& gate) NOEXCEPT
 {
     BC_ASSERT(stranded());
     if (stopped(ec))
         return;
 
     if (ancestry->empty())
-    {
-        // Complete, resubscribe to get_client_filters.
-        SUBSCRIBE_CHANNEL(get_client_filters, handle_receive_get_filters, _1, _2);
         return;
-    }
 
     const auto& query = archive();
     const auto start = logger::now();
@@ -255,7 +250,7 @@ void protocol_filter_out_70015::send_filter(const code& ec,
     out.block_hash = query.get_header_key(link);
     out.filter_type = client_filter::type_id::neutrino;
     span<milliseconds>(events::filter_msecs, start);
-    SEND(out, send_filter, _1, ancestry);
+    SEND(out, send_filter, _1, ancestry, gate);
 }
 
 BC_POP_WARNING()

src/protocols/protocol_transaction_in_106.cpp

@@ -148,19 +148,21 @@ bool protocol_transaction_in_106::handle_receive_transaction(const code& ec,
     }
 
     submit(to_shared(chain::transaction_cptrs{ tx }), false,
-        BIND(handle_submit, _1, _2));
+        BIND(handle_submit, _1, _2, gate()));
 
     return true;
 }
 
 // protected
-void protocol_transaction_in_106::handle_submit(const code& ec, size_t) NOEXCEPT
+void protocol_transaction_in_106::handle_submit(const code& ec, size_t,
+    const gate_t::ptr& gate) NOEXCEPT
 {
-    POST(do_handle_submit, ec);
+    POST(do_handle_submit, ec, gate);
 }
 
 // protected
-void protocol_transaction_in_106::do_handle_submit(const code& ec) NOEXCEPT
+void protocol_transaction_in_106::do_handle_submit(const code& ec,
+    const gate_t::ptr&) NOEXCEPT
 {
     BC_ASSERT(stranded());
 

src/protocols/protocol_transaction_in_70013.cpp

@@ -36,7 +36,8 @@ BC_PUSH_WARNING(NO_VALUE_OR_CONST_REF_SHARED_PTR)
 
 // bip133: the peer is sent our rate when current and the maximum otherwise, so
 // a tx below the rate, or any tx while suspended, is sent against instruction.
-void protocol_transaction_in_70013::do_handle_submit(const code& ec) NOEXCEPT
+void protocol_transaction_in_70013::do_handle_submit(const code& ec,
+    const gate_t::ptr& gate) NOEXCEPT
 {
     BC_ASSERT(stranded());
 
@@ -50,7 +51,7 @@ void protocol_transaction_in_70013::do_handle_submit(const code& ec) NOEXCEPT
         return;
     }
 
-    protocol_transaction_in_70001::do_handle_submit(ec);
+    protocol_transaction_in_70001::do_handle_submit(ec, gate);
 }
 
 BC_POP_WARNING()