diff --git a/include/bitcoin/node/impl/sessions/session_peer.ipp b/include/bitcoin/node/impl/sessions/session_peer.ipp index 95e62383..16ef7727 100644 --- a/include/bitcoin/node/impl/sessions/session_peer.ipp +++ b/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; diff --git a/include/bitcoin/node/protocols/protocol_block_out_106.hpp b/include/bitcoin/node/protocols/protocol_block_out_106.hpp index 7e3c8f96..d58e31ea 100644 --- a/include/bitcoin/node/protocols/protocol_block_out_106.hpp +++ b/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 #include #include @@ -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(session->log) + network::tracker(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 backlog_{}; }; } // namespace node diff --git a/include/bitcoin/node/protocols/protocol_block_out_70001.hpp b/include/bitcoin/node/protocols/protocol_block_out_70001.hpp index 09104759..f9ed70d3 100644 --- a/include/bitcoin/node/protocols/protocol_block_out_70001.hpp +++ b/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. diff --git a/include/bitcoin/node/protocols/protocol_filter_out_70015.hpp b/include/bitcoin/node/protocols/protocol_filter_out_70015.hpp index f15d16e7..53b8b102 100644 --- a/include/bitcoin/node/protocols/protocol_filter_out_70015.hpp +++ b/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; - 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 diff --git a/include/bitcoin/node/protocols/protocol_peer.hpp b/include/bitcoin/node/protocols/protocol_peer.hpp index 743323e2..1f3504fc 100644 --- a/include/bitcoin/node/protocols/protocol_peer.hpp +++ b/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. diff --git a/include/bitcoin/node/protocols/protocol_transaction_in_106.hpp b/include/bitcoin/node/protocols/protocol_transaction_in_106.hpp index 606866c1..5d667fc7 100644 --- a/include/bitcoin/node/protocols/protocol_transaction_in_106.hpp +++ b/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. diff --git a/include/bitcoin/node/protocols/protocol_transaction_in_70013.hpp b/include/bitcoin/node/protocols/protocol_transaction_in_70013.hpp index cb37d00e..0db53f90 100644 --- a/include/bitcoin/node/protocols/protocol_transaction_in_70013.hpp +++ b/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 diff --git a/include/bitcoin/node/protocols/protocol_transaction_out_106.hpp b/include/bitcoin/node/protocols/protocol_transaction_out_106.hpp index b515b23b..fe4a01ef 100644 --- a/include/bitcoin/node/protocols/protocol_transaction_out_106.hpp +++ b/include/bitcoin/node/protocols/protocol_transaction_out_106.hpp @@ -59,7 +59,8 @@ 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( @@ -67,7 +68,8 @@ class BCN_API protocol_transaction_out_106 /// 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; diff --git a/include/bitcoin/node/protocols/protocol_transaction_out_70001.hpp b/include/bitcoin/node/protocols/protocol_transaction_out_70001.hpp index 021af7b9..b02510c8 100644 --- a/include/bitcoin/node/protocols/protocol_transaction_out_70001.hpp +++ b/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. diff --git a/include/bitcoin/node/settings.hpp b/include/bitcoin/node/settings.hpp index d398af19..d6d48501 100644 --- a/include/bitcoin/node/settings.hpp +++ b/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; diff --git a/src/protocols/protocol_block_out_106.cpp b/src/protocols/protocol_block_out_106.cpp index ccd1d63a..ed7699da 100644 --- a/src/protocols/protocol_block_out_106.cpp +++ b/src/protocols/protocol_block_out_106.cpp @@ -163,42 +163,15 @@ 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)) @@ -206,13 +179,16 @@ void protocol_block_out_106::send_block(const code& ec) NOEXCEPT 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(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 diff --git a/src/protocols/protocol_block_out_70001.cpp b/src/protocols/protocol_block_out_70001.cpp index 06bc39b0..2ce81d94 100644 --- a/src/protocols/protocol_block_out_70001.cpp +++ b/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; } diff --git a/src/protocols/protocol_filter_out_70015.cpp b/src/protocols/protocol_filter_out_70015.cpp index 52bd329d..46a16d9f 100644 --- a/src/protocols/protocol_filter_out_70015.cpp +++ b/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(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(events::filter_msecs, start); - SEND(out, send_filter, _1, ancestry); + SEND(out, send_filter, _1, ancestry, gate); } BC_POP_WARNING() diff --git a/src/protocols/protocol_transaction_in_106.cpp b/src/protocols/protocol_transaction_in_106.cpp index 71a79088..3b331f70 100644 --- a/src/protocols/protocol_transaction_in_106.cpp +++ b/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()); diff --git a/src/protocols/protocol_transaction_in_70013.cpp b/src/protocols/protocol_transaction_in_70013.cpp index 7457e8cb..0dddc1f2 100644 --- a/src/protocols/protocol_transaction_in_70013.cpp +++ b/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() diff --git a/src/protocols/protocol_transaction_out_106.cpp b/src/protocols/protocol_transaction_out_106.cpp index a2b71837..a4a66553 100644 --- a/src/protocols/protocol_transaction_out_106.cpp +++ b/src/protocols/protocol_transaction_out_106.cpp @@ -133,9 +133,8 @@ bool protocol_transaction_out_106::handle_receive_get_data(const code& ec, if (stopped(ec)) return false; - // Post so the completion resubscribe runs outside the current notify(). - POST(send_transaction, error::success, zero, message); - return false; + send_transaction(error::success, zero, message, gate()); + return true; } // Outbound (tx). @@ -146,7 +145,8 @@ bool protocol_transaction_out_106::handle_receive_get_data(const code& ec, // using MSG_TX getdata messages." (derived protocol) void protocol_transaction_out_106::send_transaction(const code& ec, - size_t index, const get_data::cptr& message) NOEXCEPT + size_t index, const get_data::cptr& message, + const gate_t::ptr& gate) NOEXCEPT { BC_ASSERT(stranded()); @@ -189,18 +189,13 @@ void protocol_transaction_out_106::send_transaction(const code& ec, } // The report resumes this loop on completion, so it precedes the tx. - if (report_unservable(index, message)) + if (report_unservable(index, message, gate)) return; - // BUGBUG: registration race. if (index >= message->items.size()) - { - // Complete, resubscribe to transaction requests. - SUBSCRIBE_CHANNEL(get_data, handle_receive_get_data, _1, _2); return; - } - SEND(transaction{ ptr }, send_transaction, _1, add1(index), message); + SEND(transaction{ ptr }, send_transaction, _1, add1(index), message, gate); } // not_found is undefined below bip37, so the channel is stopped instead. @@ -218,7 +213,7 @@ bool protocol_transaction_out_106::handle_unservable( // There is nothing to report below bip37, the channel is stopped above. bool protocol_transaction_out_106::report_unservable(size_t, - const get_data::cptr&) NOEXCEPT + const get_data::cptr&, const gate_t::ptr&) NOEXCEPT { return false; } diff --git a/src/protocols/protocol_transaction_out_70001.cpp b/src/protocols/protocol_transaction_out_70001.cpp index dbe638ee..5a9a415b 100644 --- a/src/protocols/protocol_transaction_out_70001.cpp +++ b/src/protocols/protocol_transaction_out_70001.cpp @@ -52,7 +52,7 @@ bool protocol_transaction_out_70001::handle_unservable( // The items are answered and the send loop resumed, as with a transaction, so // nothing is produced until the prior write completes. bool protocol_transaction_out_70001::report_unservable(size_t index, - const get_data::cptr& message) NOEXCEPT + const get_data::cptr& message, const gate_t::ptr& gate) NOEXCEPT { BC_ASSERT(stranded()); @@ -62,7 +62,8 @@ bool protocol_transaction_out_70001::report_unservable(size_t index, auto items = std::move(unservable_); unservable_.clear(); - SEND(not_found{ std::move(items) }, send_transaction, _1, index, message); + SEND(not_found{ std::move(items) }, send_transaction, _1, index, + message, gate); return true; } diff --git a/src/settings.cpp b/src/settings.cpp index 5c0fe32e..f699509b 100644 --- a/src/settings.cpp +++ b/src/settings.cpp @@ -34,7 +34,6 @@ settings::settings() NOEXCEPT headers_first{ true }, memory_priority{ true }, thread_priority{ true }, - allow_overlapped{ true }, provide_blocks{ true }, require_blocks{ true }, provide_witness{ true }, diff --git a/test/settings.cpp b/test/settings.cpp index ab380d32..534b63bb 100644 --- a/test/settings.cpp +++ b/test/settings.cpp @@ -35,7 +35,6 @@ BOOST_AUTO_TEST_CASE(settings__node__default_context__expected) BOOST_REQUIRE_EQUAL(node.headers_first, true); BOOST_REQUIRE_EQUAL(node.memory_priority, true); BOOST_REQUIRE_EQUAL(node.thread_priority, true); - BOOST_REQUIRE_EQUAL(node.allow_overlapped, true); BOOST_REQUIRE_EQUAL(node.provide_blocks, true); BOOST_REQUIRE_EQUAL(node.require_blocks, true); BOOST_REQUIRE_EQUAL(node.provide_witness, true);