GH-1155 Add disconnect of connections over peer-limit

Author: heifner

plugins/net_plugin/net_plugin.cpp

@@ -248,6 +248,7 @@ namespace eosio {
    constexpr auto     def_max_clients = 25; // 0 for unlimited clients
    constexpr auto     def_max_nodes_per_host = 1;
    constexpr auto     def_conn_retry_wait = 30;
+   constexpr auto     def_conn_retry_wait_peer_limit_multiplier = 10;
    constexpr auto     def_txn_expire_wait = std::chrono::seconds(3);
    constexpr auto     def_resp_expected_wait = std::chrono::seconds(5);
    constexpr auto     def_sync_fetch_span = 1000;
@@ -310,6 +311,7 @@ namespace eosio {
       mutable std::shared_mutex        connections_mtx;
       connection_details_index         connections;
       chain::flat_set<string>          supplied_peers;
+      std::atomic<uint64_t>            max_peer_ping_time_ns{std::numeric_limits<uint64_t>::max()};;
 
       alignas(hardware_destructive_interference_sz)
       fc::mutex                             connector_check_timer_mtx;
@@ -322,6 +324,7 @@ namespace eosio {
       fc::microseconds                                         max_cleanup_time;
       boost::asio::steady_timer::duration                      connector_period{0};
       uint32_t                                                 max_client_count{def_max_clients};
+      uint16_t                                                 peer_limit{6};
       std::function<void(net_plugin::p2p_connections_metrics)> update_p2p_connection_metrics;
 
    private: // must call with held mutex
@@ -332,13 +335,16 @@ namespace eosio {
 
    public:
       size_t number_connections() const;
+      size_t number_connected_peers() const;
       void add_supplied_peers(const vector<string>& peers );
+      void maybe_disconnect(connection_ptr c) const;
 
       // not thread safe, only call on startup
       void init(std::chrono::milliseconds heartbeat_timeout_ms,
                 fc::microseconds conn_max_cleanup_time,
                 boost::asio::steady_timer::duration conn_period,
-                uint32_t maximum_client_count);
+                uint32_t maximum_client_count,
+                uint16_t peer_limit);
 
       std::chrono::milliseconds get_heartbeat_timeout() const { return heartbeat_timeout; }
 
@@ -907,6 +913,8 @@ namespace eosio {
       block_status_monitor    block_status_monitor_;
       std::atomic<time_point> last_vote_received;
       std::atomic<uint32_t>   unique_votes_rcvd_count{0};
+      std::atomic<time_point> last_unique_block_received;
+      std::atomic<bool>       closed_by_peer_limit = false;
 
       alignas(hardware_destructive_interference_sz)
       fc::mutex                        sync_response_expected_timer_mtx;
@@ -1399,6 +1407,7 @@ namespace eosio {
          peer_dlog( this, "connected" );
          socket_open = true;
          connection_start_time = get_time();
+         closed_by_peer_limit = false;
          start_read_message();
          return true;
       }
@@ -3665,6 +3674,8 @@ namespace eosio {
       // make sure we also get the latency we need
       if (peer_ping_time_ns == std::numeric_limits<uint64_t>::max()) {
          send_time();
+      } else {
+         my_impl->connections.maybe_disconnect(shared_from_this());
       }
    }
 
@@ -4019,6 +4030,7 @@ namespace eosio {
 
          if (fork_db_add_result == fork_db_add_t::appended_to_head || fork_db_add_result == fork_db_add_t::fork_switch) {
             ++c->unique_blocks_rcvd_count;
+            c->last_unique_block_received = fc::time_point::now();
             fc_dlog(logger, "post process_incoming_block to app thread, block ${n}", ("n", ptr->block_num()));
             my_impl->producer_plug->process_blocks();
 
@@ -4352,7 +4364,7 @@ namespace eosio {
            "   p2p.trx.eos.io:9876:trx\n"
            "   p2p.blk.eos.io:9876:blk\n")
          ( "p2p-peer-limit", bpo::value<uint16_t>()->default_value(6),
-           "Limit the number of p2p-peer-address to remain connected to. Selects the best peers of p2p-peer-address-list.")
+           "Soft limit on the number of p2p-peer-address to remain connected to. Selects the best peers of p2p-peer-address-list.")
          ( "p2p-max-nodes-per-host", bpo::value<int>()->default_value(def_max_nodes_per_host), "Maximum number of client nodes from any single IP address")
          ( "p2p-accept-transactions", bpo::value<bool>()->default_value(true), "Allow transactions received over p2p network to be evaluated and relayed if valid.")
          ( "p2p-disable-block-nack", bpo::value<bool>()->default_value(false),
@@ -4371,7 +4383,7 @@ namespace eosio {
            "Tuple of [PublicKey, WIF private key] (may specify multiple times)")
          ( "max-clients", bpo::value<uint32_t>()->default_value(def_max_clients), "Maximum number of clients from which connections are accepted, use 0 for no limit")
          ( "connection-cleanup-period", bpo::value<int>()->default_value(def_conn_retry_wait), "number of seconds to wait before cleaning up dead connections")
-         ( "max-cleanup-time-msec", bpo::value<uint32_t>()->default_value(10), "max connection cleanup time per cleanup call in milliseconds")
+         ( "max-cleanup-time-msec", bpo::value<uint32_t>()->default_value(50), "max connection cleanup time per cleanup call in milliseconds")
          ( "p2p-dedup-cache-expire-time-sec", bpo::value<uint32_t>()->default_value(10), "Maximum time to track transaction for duplicate optimization")
          ( "net-threads", bpo::value<uint16_t>()->default_value(my->thread_pool_size),
            "Number of worker threads in net_plugin thread pool" )
@@ -4418,6 +4430,16 @@ namespace eosio {
          p2p_accept_transactions = options.at( "p2p-accept-transactions" ).as<bool>();
          p2p_disable_block_nack = options.at( "p2p-disable-block-nack" ).as<bool>();
 
+         if (p2p_accept_transactions || chain_plug->accept_votes()) {
+            if (p2p_accept_transactions && chain_plug->accept_votes()) {
+               EOS_ASSERT( p2p_peer_limit >= 6, chain::plugin_config_exception,
+                           "p2p-peer-limit with vote processing and trx processing should be >= 6, two connections each reserved for votes and trxs.");
+            } else {
+               EOS_ASSERT( p2p_peer_limit >= 4, chain::plugin_config_exception,
+                           "p2p-peer-limit with vote processing or trx processing should be >= 4, two connections reserved for vote or trx processing");
+            }
+         }
+
          use_socket_read_watermark = options.at( "use-socket-read-watermark" ).as<bool>();
          keepalive_interval = std::chrono::milliseconds( options.at( "p2p-keepalive-interval-ms" ).as<int>() );
          EOS_ASSERT( keepalive_interval.count() > 0, chain::plugin_config_exception,
@@ -4436,7 +4458,8 @@ namespace eosio {
          connections.init( std::chrono::milliseconds( options.at("p2p-keepalive-interval-ms").as<int>() * 2 ),
                                fc::milliseconds( options.at("max-cleanup-time-msec").as<uint32_t>() ),
                                std::chrono::seconds( options.at("connection-cleanup-period").as<int>() ),
-                               options.at("max-clients").as<uint32_t>() );
+                               options.at("max-clients").as<uint32_t>(),
+                               p2p_peer_limit );
 
          if( options.count( "p2p-listen-endpoint" )) {
             auto p2ps =  options.at("p2p-listen-endpoint").as<vector<string>>();
@@ -4727,6 +4750,16 @@ namespace eosio {
       return connections.size();
    }
 
+   size_t connections_manager::number_connected_peers() const {
+      size_t num = 0;
+      for_each_block_connection([&num](const auto& cc) {
+         if (cc->socket_is_open() && !cc->incoming()) {
+            ++num;
+         }
+      });
+      return num;
+   }
+
    connection_ptr connections_manager::find_connection(uint32_t connection_id) const {
       std::shared_lock g(connections_mtx);
       const auto& index = connections.get<by_connection_id>();
@@ -4740,15 +4773,48 @@ namespace eosio {
       supplied_peers.insert( peers.begin(), peers.end() );
    }
 
+   void connections_manager::maybe_disconnect(connection_ptr c) const {
+      assert(c);
+      // if accepting trx then don't disconnect from any peers
+      if (my_impl->p2p_accept_transactions) return;
+      // if syncing then other connections are basically idle
+      if (my_impl->sync_master->syncing_from_peer()) return;
+      // verify current, not syncing from us
+      if (!c->current()) return;
+      // remain connected for a heartbeat to gather info
+      if (c->connection_start_time.load() > connection::get_time() - get_heartbeat_timeout()) return;
+      // we have received votes first from this connection
+      if (c->unique_votes_rcvd_count > 0) return;
+      // we received block first from this connection recently
+      if (c->last_unique_block_received.load() > fc::time_point::now() - fc::minutes(10)) return;
+      // quick check if more connections than limit
+      if (number_connections() <= peer_limit) return;
+      // do we have more than the limit currently
+      if (number_connected_peers() <= peer_limit) return;
+
+      auto peer_ping_time = c->get_peer_ping_time_ns();
+      if (peer_ping_time < max_peer_ping_time_ns) {
+         return;
+      }
+
+      peer_ilog(c, "Disconnecting from peer that is not one of the best of p2p-peer-address, ping ${p}ms, max ${m}ms",
+                ("p", peer_ping_time/1000000)("m", max_peer_ping_time_ns.load()/1000000));
+
+      c->closed_by_peer_limit = true;
+      c->close(false);
+   }
+
    // not thread safe, only call on startup
    void connections_manager::init( std::chrono::milliseconds heartbeat_timeout_ms,
              fc::microseconds conn_max_cleanup_time,
              boost::asio::steady_timer::duration conn_period,
-             uint32_t maximum_client_count ) {
+             uint32_t maximum_client_count,
+             uint16_t p2p_peer_limit) {
       heartbeat_timeout = heartbeat_timeout_ms;
       max_cleanup_time = conn_max_cleanup_time;
       connector_period = conn_period;
       max_client_count = maximum_client_count;
+      peer_limit = p2p_peer_limit;
    }
 
    fc::microseconds connections_manager::get_connector_period() const {
@@ -4958,11 +5024,16 @@ namespace eosio {
    // called from any thread
    void connections_manager::connection_monitor(const std::weak_ptr<connection>& from_connection) {
       size_t num_rm = 0, num_clients = 0, num_peers = 0, num_bp_peers = 0;
-      auto cleanup = [&num_rm, this](vector<connection_ptr>&& reconnecting, vector<connection_ptr>&& removing) {
-         for( auto& c : reconnecting ) {
-            if (!c->resolve_and_connect()) {
-               ++num_rm;
-               removing.push_back(c);
+      std::multiset<uint64_t> ping_times;
+      auto cleanup = [&num_rm, num_peers, this](vector<connection_ptr>&& reconnecting, vector<connection_ptr>&& removing) {
+         // if num_peers less than or equal the peer_limit then attempt to connect to all configured peers
+         // the best of the configured peers will be chosen after they connect.
+         if (num_peers <= peer_limit) {
+            for( auto& c : reconnecting ) {
+               if (!c->resolve_and_connect()) {
+                  ++num_rm;
+                  removing.push_back(c);
+               }
             }
          }
          std::scoped_lock g( connections_mtx );
@@ -4971,6 +5042,14 @@ namespace eosio {
             index.erase(c->connection_id);
          }
       };
+      auto record_max_peer_ping_time = [&]() {
+         if (ping_times.size() > peer_limit) {
+            auto itr = ping_times.begin();
+            std::advance(itr, peer_limit-1);
+            max_peer_ping_time_ns = *itr;
+         }
+      };
+      auto peer_limit_reconnect_time = connection::get_time() - (connector_period * def_conn_retry_wait_peer_limit_multiplier);
       auto max_time = fc::time_point::now().safe_add(max_cleanup_time);
       std::vector<connection_ptr> reconnecting, removing;
       auto from = from_connection.lock();
@@ -5001,16 +5080,23 @@ namespace eosio {
          if (!c->socket_is_open() && c->state() != connection::connection_state::connecting) {
             if (!c->incoming()) {
                --num_peers;
-               reconnecting.push_back(c);
+               if (!c->closed_by_peer_limit || (c->connection_start_time.load() < peer_limit_reconnect_time)) {
+                  reconnecting.push_back(c);
+               }
             } else {
                --num_clients;
                ++num_rm;
                removing.push_back(c);
             }
          }
+         auto ping_time = c->get_peer_ping_time_ns();
+         if (ping_time < std::numeric_limits<uint64_t>::max()) {
+            ping_times.insert(ping_time);
+         }
          ++it;
       }
       g.unlock();
+      record_max_peer_ping_time();
       cleanup(std::move(reconnecting), std::move(removing));
 
       if( num_clients > 0 || num_peers > 0 ) {