Support throttling block syncing to peers.

plugins/net_plugin/include/eosio/net_plugin/net_plugin.hpp

@@ -54,6 +54,7 @@ namespace eosio {
                std::chrono::nanoseconds last_bytes_received{0};
                size_t bytes_sent{0};
                std::chrono::nanoseconds last_bytes_sent{0};
+               size_t block_sync_bytes_sent{0};
                std::chrono::nanoseconds connection_start_time{0};
                std::string log_p2p_address;
             };

plugins/net_plugin/include/eosio/net_plugin/protocol.hpp

@@ -16,9 +16,10 @@ namespace eosio {
 
    // Longest domain name is 253 characters according to wikipedia.
    // Addresses include ":port" where max port is 65535, which adds 6 chars.
+   // Addresses may also include ":bitrate" with suffix and separators, which adds 30 chars.
    // We also add our own extentions of "[:trx|:blk] - xxxxxxx", which adds 14 chars, total= 273.
    // Allow for future extentions as well, hence 384.
-   constexpr size_t max_p2p_address_length = 253 + 6;
+   constexpr size_t max_p2p_address_length = 253 + 6 + 30;
    constexpr size_t max_handshake_str_length = 384;
 
    struct handshake_message {

plugins/net_plugin/net_plugin.cpp

@@ -29,6 +29,7 @@
 #include <cmath>
 #include <memory>
 #include <new>
+#include <regex>
 
 // should be defined for c++17, but clang++16 still has not implemented it
 #ifdef __cpp_lib_hardware_interference_size
@@ -475,6 +476,7 @@ namespace eosio {
       std::function<void()> increment_dropped_trxs;
 
    private:
+      static const std::map<std::string, int> prefix_multipliers;
       alignas(hardware_destructive_interference_size)
       mutable fc::mutex             chain_info_mtx; // protects chain_info_t
       chain_info_t                  chain_info GUARDED_BY(chain_info_mtx);
@@ -529,14 +531,15 @@ namespace eosio {
 
       constexpr static uint16_t to_protocol_version(uint16_t v);
 
+      size_t parse_connection_limit(const string& limit_str);
       void plugin_initialize(const variables_map& options);
       void plugin_startup();
       void plugin_shutdown();
       bool in_sync() const;
       fc::logger& get_logger() { return logger; }
 
-      void create_session(tcp::socket&& socket, const string listen_address);
-   };
+      void create_session(tcp::socket&& socket, const string listen_address, size_t limit);
+   }; //net_plugin_impl
 
    // peer_[x]log must be called from thread in connection strand
 #define peer_dlog( PEER, FORMAT, ... ) \
@@ -771,7 +774,7 @@ namespace eosio {
       /// @brief ctor
       /// @param socket created by boost::asio in fc::listener
       /// @param address identifier of listen socket which accepted this new connection
-      explicit connection( tcp::socket&& socket, const string& listen_address );
+      explicit connection( tcp::socket&& socket, const string& listen_address, size_t block_sync_rate_limit );
       ~connection() = default;
 
       connection( const connection& ) = delete;
@@ -800,6 +803,7 @@ namespace eosio {
       std::chrono::nanoseconds get_last_bytes_received() const { return last_bytes_received.load(); }
       size_t get_bytes_sent() const { return bytes_sent.load(); }
       std::chrono::nanoseconds get_last_bytes_sent() const { return last_bytes_sent.load(); }
+      size_t get_block_sync_bytes_sent() const { return block_sync_bytes_sent.load(); }
       boost::asio::ip::port_type get_remote_endpoint_port() const { return remote_endpoint_port.load(); }
       void set_heartbeat_timeout(std::chrono::milliseconds msec) {
          hb_timeout = msec;
@@ -827,6 +831,8 @@ namespace eosio {
          blocks_only
       };
 
+      size_t                          block_sync_rate_limit{0};  // bytes/second, default unlimited
+
       std::atomic<connection_types>   connection_type{both};
       std::atomic<uint32_t>           peer_start_block_num{0};
       std::atomic<uint32_t>           peer_head_block_num{0};
@@ -835,6 +841,7 @@ namespace eosio {
       std::atomic<size_t>             bytes_received{0};
       std::atomic<std::chrono::nanoseconds>   last_bytes_received{0ns};
       std::atomic<size_t>             bytes_sent{0};
+      std::atomic<size_t>             block_sync_bytes_sent{0};
       std::atomic<std::chrono::nanoseconds>   last_bytes_sent{0ns};
       std::atomic<boost::asio::ip::port_type> remote_endpoint_port{0};
 
@@ -962,10 +969,11 @@ namespace eosio {
       void send_time(const time_message& msg);
       /** \brief Read system time and convert to a 64 bit integer.
        *
-       * There are five calls to this routine in the program.  One
+       * There are six calls to this routine in the program.  One
        * when a packet arrives from the network, one when a packet
-       * is placed on the send queue, one during start session, and
-       * one each when data is counted as received or sent.
+       * is placed on the send queue, one during start session, one
+       * when a sync block is queued and one each when data is
+       * counted as received or sent.
        * Calls the kernel time of day routine and converts to 
        * a (at least) 64 bit integer.
        */
@@ -980,7 +988,7 @@ namespace eosio {
       void stop_send();
 
       void enqueue( const net_message &msg );
-      void enqueue_block( const signed_block_ptr& sb, bool to_sync_queue = false);
+      size_t enqueue_block( const signed_block_ptr& sb, bool to_sync_queue = false);
       void enqueue_buffer( const std::shared_ptr<std::vector<char>>& send_buffer,
                            go_away_reason close_after_send,
                            bool to_sync_queue = false);
@@ -1055,6 +1063,10 @@ namespace eosio {
    }; // class connection
 
    const string connection::unknown = "<unknown>";
+   const std::map<std::string, int> net_plugin_impl::prefix_multipliers{
+      {"",1},{"K",pow(10,3)},{"M",pow(10,6)},{"G",pow(10, 9)},{"T",pow(10, 12)},
+             {"Ki",pow(2,10)},{"Mi",pow(2,20)},{"Gi",pow(2,30)},{"Ti",pow(2,40)}
+   };
 
    // called from connection strand
    struct msg_handler : public fc::visitor<void> {
@@ -1185,9 +1197,10 @@ namespace eosio {
       fc_ilog( logger, "created connection ${c} to ${n}", ("c", connection_id)("n", endpoint) );
    }
 
-   connection::connection(tcp::socket&& s, const string& listen_address)
+   connection::connection(tcp::socket&& s, const string& listen_address, size_t block_sync_rate_limit)
       : listen_address( listen_address ),
         peer_addr(),
+        block_sync_rate_limit(block_sync_rate_limit),
         strand( my_impl->thread_pool.get_executor() ),
         socket( new tcp::socket( std::move(s) ) ),
         connection_id( ++my_impl->current_connection_id ),
@@ -1199,7 +1212,6 @@ namespace eosio {
       fc_dlog( logger, "new connection object created for peer ${address}:${port} from listener ${addr}", ("address", log_remote_endpoint_ip)("port", log_remote_endpoint_port)("addr", listen_address) );
    }
 
-   // called from connection strand
    void connection::update_endpoints() {
       boost::system::error_code ec;
       boost::system::error_code ec2;
@@ -1649,7 +1661,7 @@ namespace eosio {
       } else {
          peer_dlog( this, "enqueue sync block ${num}", ("num", peer_requested->last + 1) );
       }
-      uint32_t num = ++peer_requested->last;
+      uint32_t num = peer_requested->last + 1;
       if(num == peer_requested->end_block) {
          peer_requested.reset();
          peer_dlog( this, "completing enqueue_sync_block ${num}", ("num", num) );
@@ -1661,14 +1673,25 @@ namespace eosio {
          sb = cc.fetch_block_by_number( num ); // thread-safe
       } FC_LOG_AND_DROP();
       if( sb ) {
-         enqueue_block( sb, true );
+         if( block_sync_rate_limit > 0 ) {
+            auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(get_time() - connection_start_time);
+            auto current_rate = double(block_sync_bytes_sent) / elapsed.count();
+            if( current_rate < block_sync_rate_limit ) {
+               block_sync_bytes_sent += enqueue_block( sb, true );
+               ++peer_requested->last;
+            } else {
+               return false;
+            }
+         } else {
+            block_sync_bytes_sent += enqueue_block( sb, true );
+            ++peer_requested->last;
+         }
       } else {
          peer_ilog( this, "enqueue sync, unable to fetch block ${num}, sending benign_other go away", ("num", num) );
          peer_requested.reset(); // unable to provide requested blocks
          no_retry = benign_other;
          enqueue( go_away_message( benign_other ) );
       }
-
       return true;
    }
 
@@ -1781,14 +1804,15 @@ namespace eosio {
    }
 
    // called from connection strand
-   void connection::enqueue_block( const signed_block_ptr& b, bool to_sync_queue) {
+   size_t connection::enqueue_block( const signed_block_ptr& b, bool to_sync_queue) {
       peer_dlog( this, "enqueue block ${num}", ("num", b->block_num()) );
       verify_strand_in_this_thread( strand, __func__, __LINE__ );
 
       block_buffer_factory buff_factory;
       auto sb = buff_factory.get_send_buffer( b );
       latest_blk_time = std::chrono::system_clock::now();
       enqueue_buffer( sb, no_reason, to_sync_queue);
+      return sb->size();
    }
 
    // called from connection strand
@@ -2719,7 +2743,7 @@ namespace eosio {
    }
 
 
-   void net_plugin_impl::create_session(tcp::socket&& socket, const string listen_address) {
+   void net_plugin_impl::create_session(tcp::socket&& socket, const string listen_address, size_t limit) {
       uint32_t                  visitors  = 0;
       uint32_t                  from_addr = 0;
       boost::system::error_code rec;
@@ -2745,7 +2769,7 @@ namespace eosio {
                visitors < connections.get_max_client_count())) {
             fc_ilog(logger, "Accepted new connection: " + paddr_str);
 
-            connection_ptr new_connection = std::make_shared<connection>(std::move(socket), listen_address);
+            connection_ptr new_connection = std::make_shared<connection>(std::move(socket), listen_address, limit);
             new_connection->strand.post([new_connection, this]() {
                if (new_connection->start_session()) {
                   connections.add(new_connection);
@@ -3927,7 +3951,15 @@ namespace eosio {
    void net_plugin::set_program_options( options_description& /*cli*/, options_description& cfg )
    {
       cfg.add_options()
-         ( "p2p-listen-endpoint", bpo::value< vector<string> >()->default_value( vector<string>(1, string("0.0.0.0:9876")) ), "The actual host:port used to listen for incoming p2p connections. May be used multiple times.")
+         ( "p2p-listen-endpoint", bpo::value< vector<string> >()->default_value( vector<string>(1, string("0.0.0.0:9876:0")) ), "The actual host:port[:<rate-cap>] used to listen for incoming p2p connections. May be used multiple times. "
+           "  The optional rate cap will limit block sync bandwidth to the specified rate.  A number alone will be "
+           "  interpreted as bytes per second.  The number may be suffixed with units.  Supported units are: "
+           "  'B/s', 'KB/s', 'MB/s, 'GB/s', 'TB/s', 'KiB/s', 'MiB/s', 'GiB/s', 'TiB/s'."
+           "  Transactions and blocks outside of sync mode are not throttled."
+           "  Examples:\n"
+           "    192.168.0.100:9876:1MiB/s\n"
+           "    node.eos.io:9876:250KB/s\n"
+           "    node.eos.io:9876:0.5GB/s")
          ( "p2p-server-address", bpo::value< vector<string> >(), "An externally accessible host:port for identifying this node. Defaults to p2p-listen-endpoint. May be used as many times as p2p-listen-endpoint. If provided, the first address will be used in handshakes with other nodes. Otherwise the default is used.")
          ( "p2p-peer-address", bpo::value< vector<string> >()->composing(),
            "The public endpoint of a peer node to connect to. Use multiple p2p-peer-address options as needed to compose a network.\n"
@@ -3983,6 +4015,29 @@ namespace eosio {
       return fc::json::from_string(s).as<T>();
    }
 
+   size_t net_plugin_impl::parse_connection_limit( const std::string& limit_str) {
+      std::istringstream in(limit_str);
+      fc_dlog( logger, "parsing connection endpoint with locale ${l}", ("l", std::locale("").name()));
+      in.imbue(std::locale(""));
+      double limit{0};
+      in >> limit;
+      size_t block_sync_rate_limit = 0;
+      if( limit > 0.0f ) {
+         std::string units;
+         in >> units;
+         std::regex units_regex{"([KMGT]?[i]?)B/s"};
+         std::smatch units_match;
+         std::regex_match(units, units_match, units_regex);
+         try {
+            block_sync_rate_limit = boost::numeric_cast<size_t>(limit * prefix_multipliers.at(units_match[1].str()));
+            fc_dlog( logger, "setting block_sync_rate_limit to ${limit}", ("limit", block_sync_rate_limit));
+         } catch (boost::numeric::bad_numeric_cast&) {
+            EOS_ASSERT(false, plugin_config_exception, "block sync limit specification overflowed: ${limit}", ("limit", limit_str));
+         }
+      }
+      return block_sync_rate_limit;
+   }
+
    void net_plugin_impl::plugin_initialize( const variables_map& options ) {
       try {
          fc_ilog( logger, "Initialize net plugin" );
@@ -4192,10 +4247,20 @@ namespace eosio {
 
                std::string extra_listening_log_info =
                      ", max clients is " + std::to_string(my->connections.get_max_client_count());
-
+
+               auto listen_addr = address;
+               auto limit = string("0");
+               if( std::count(address.begin(), address.end(), ':') > 1 ) {
+                  auto last_colon_location = address.rfind(':');
+                  listen_addr = std::string(address, 0, last_colon_location);
+                  limit = std::string(address, last_colon_location+1);
+               }
+
+               auto block_sync_rate_limit = my->parse_connection_limit(limit);
+
                fc::create_listener<tcp>(
-                     my->thread_pool.get_executor(), logger, accept_timeout, address, extra_listening_log_info,
-                     [my = my, addr = p2p_addr](tcp::socket&& socket) { my->create_session(std::move(socket), addr); });
+                     my->thread_pool.get_executor(), logger, accept_timeout, listen_addr, extra_listening_log_info,
+                     [my = my, addr = p2p_addr, block_sync_rate_limit = block_sync_rate_limit](tcp::socket&& socket) { fc_dlog( logger, "start listening on ${addr} with peer sync throttle ${limit}", ("addr", addr)("limit", block_sync_rate_limit)); my->create_session(std::move(socket), addr, block_sync_rate_limit); });
             } catch (const std::exception& e) {
                fc_elog( logger, "net_plugin::plugin_startup failed to listen on ${addr}, ${what}",
                      ("addr", address)("what", e.what()) );
@@ -4479,6 +4544,7 @@ namespace eosio {
                , .last_bytes_received = (*it)->get_last_bytes_received()
                , .bytes_sent = (*it)->get_bytes_sent()
                , .last_bytes_sent = (*it)->get_last_bytes_sent()
+               , .block_sync_bytes_sent = (*it)->get_block_sync_bytes_sent()
                , .connection_start_time = (*it)->connection_start_time
                , .log_p2p_address = (*it)->log_p2p_address
             };

plugins/prometheus_plugin/metrics.hpp

@@ -190,6 +190,7 @@ struct catalog_type {
          add_and_set_gauge("last_bytes_received", peer.last_bytes_received.count());
          add_and_set_gauge("bytes_sent", peer.bytes_sent);
          add_and_set_gauge("last_bytes_sent", peer.last_bytes_sent.count());
+         add_and_set_gauge("block_sync_bytes_sent", peer.block_sync_bytes_sent);
          add_and_set_gauge("connection_start_time", peer.connection_start_time.count());
          add_and_set_gauge(peer.log_p2p_address, 0); // Empty gauge; we only want the label
       }

tests/CMakeLists.txt

@@ -51,6 +51,8 @@ configure_file(${CMAKE_CURRENT_SOURCE_DIR}/http_plugin_test.py ${CMAKE_CURRENT_B
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/p2p_high_latency_test.py ${CMAKE_CURRENT_BINARY_DIR}/p2p_high_latency_test.py COPYONLY)
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/p2p_multiple_listen_test.py ${CMAKE_CURRENT_BINARY_DIR}/p2p_multiple_listen_test.py COPYONLY)
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/p2p_no_listen_test.py ${CMAKE_CURRENT_BINARY_DIR}/p2p_no_listen_test.py COPYONLY)
+configure_file(${CMAKE_CURRENT_SOURCE_DIR}/p2p_sync_throttle_test.py ${CMAKE_CURRENT_BINARY_DIR}/p2p_sync_throttle_test.py COPYONLY)
+configure_file(${CMAKE_CURRENT_SOURCE_DIR}/p2p_sync_throttle_test_shape.json ${CMAKE_CURRENT_BINARY_DIR}/p2p_sync_throttle_test_shape.json COPYONLY)
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/compute_transaction_test.py ${CMAKE_CURRENT_BINARY_DIR}/compute_transaction_test.py COPYONLY)
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/subjective_billing_test.py ${CMAKE_CURRENT_BINARY_DIR}/subjective_billing_test.py COPYONLY)
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/get_account_test.py ${CMAKE_CURRENT_BINARY_DIR}/get_account_test.py COPYONLY)
@@ -186,6 +188,8 @@ add_test(NAME p2p_multiple_listen_test COMMAND tests/p2p_multiple_listen_test.py
 set_property(TEST p2p_multiple_listen_test PROPERTY LABELS nonparallelizable_tests)
 add_test(NAME p2p_no_listen_test COMMAND tests/p2p_no_listen_test.py -v ${UNSHARE} WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
 set_property(TEST p2p_no_listen_test PROPERTY LABELS nonparallelizable_tests)
+add_test(NAME p2p_sync_throttle_test COMMAND tests/p2p_sync_throttle_test.py -v -d 2 ${UNSHARE} WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
+set_property(TEST p2p_sync_throttle_test PROPERTY LABELS nonparallelizable_tests)
 
 # needs iproute-tc or iproute2 depending on platform
 #add_test(NAME p2p_high_latency_test COMMAND tests/p2p_high_latency_test.py -v WORKING_DIRECTORY ${CMAKE_BINARY_DIR})

tests/TestHarness/testUtils.py

@@ -256,6 +256,9 @@ def waitForObj(lam, timeout=None, sleepTime=1, reporter=None):
                 if reporter is not None:
                     reporter()
                 time.sleep(sleepTime)
+            else:
+                if timeout == 60:
+                    raise RuntimeError('waitForObj reached 60 second timeout')
         finally:
             if needsNewLine:
                 Utils.Print()

tests/block_log_util_test.py

@@ -70,7 +70,7 @@ def verifyBlockLog(expected_block_num, trimmedBlockLog):
     node0.kill(signal.SIGTERM)
 
     Print("Wait for node0's head block to become irreversible")
-    node1.waitForBlock(headBlockNum, blockType=BlockType.lib)
+    node1.waitForBlock(headBlockNum, blockType=BlockType.lib, timeout=90)
     infoAfter=node1.getInfo(exitOnError=True)
     headBlockNumAfter=infoAfter["head_block_num"]