Cleanup TCP throttling code (performance) + move connection checks

contrib/epee/include/net/abstract_tcp_server2.h

@@ -77,6 +77,13 @@ namespace net_utils
     virtual ~i_connection_filter(){}
   };
 
+  struct i_connection_limit
+  {
+    virtual bool is_host_limit(const epee::net_utils::network_address &address)=0;
+  protected:
+    virtual ~i_connection_limit(){}
+  };
+  
 
   /************************************************************************/
   /*                                                                      */
@@ -260,10 +267,11 @@ namespace net_utils
     struct shared_state : connection_basic_shared_state, t_protocol_handler::config_type
     {
       shared_state()
-        : connection_basic_shared_state(), t_protocol_handler::config_type(), pfilter(nullptr), stop_signal_sent(false)
+        : connection_basic_shared_state(), t_protocol_handler::config_type(), pfilter(nullptr), plimit(nullptr), stop_signal_sent(false)
       {}
 
       i_connection_filter* pfilter;
+      i_connection_limit* plimit;
       bool stop_signal_sent;
     };
 
@@ -369,6 +377,7 @@ namespace net_utils
     size_t get_threads_count(){return m_threads_count;}
 
     void set_connection_filter(i_connection_filter* pfilter);
+    void set_connection_limit(i_connection_limit* plimit);
 
     void set_default_remote(epee::net_utils::network_address remote)
     {

contrib/epee/include/net/abstract_tcp_server2.inl

@@ -328,7 +328,7 @@ namespace net_utils
       return;
     }
     auto self = connection<T>::shared_from_this();
-    if (m_connection_type != e_connection_type_RPC) {
+    if (speed_limit_is_enabled()) {
       auto calc_duration = []{
         CRITICAL_REGION_LOCAL(
           network_throttle_manager_t::m_lock_get_global_throttle_in
@@ -382,7 +382,7 @@ namespace net_utils
             m_conn_context.m_max_speed_down,
             speed
           );
-          {
+          if (speed_limit_is_enabled()) {
             CRITICAL_REGION_LOCAL(
               network_throttle_manager_t::m_lock_get_global_throttle_in
             );
@@ -454,7 +454,7 @@ namespace net_utils
       return;
     }
     auto self = connection<T>::shared_from_this();
-    if (m_connection_type != e_connection_type_RPC) {
+    if (speed_limit_is_enabled()) {
       auto calc_duration = [this]{
         CRITICAL_REGION_LOCAL(
           network_throttle_manager_t::m_lock_get_global_throttle_out
@@ -513,7 +513,7 @@ namespace net_utils
             m_conn_context.m_max_speed_down,
             speed
           );
-          {
+          if (speed_limit_is_enabled()) {
             CRITICAL_REGION_LOCAL(
               network_throttle_manager_t::m_lock_get_global_throttle_out
             );
@@ -873,6 +873,13 @@ namespace net_utils
     ).pfilter;
     if (filter && !filter->is_remote_host_allowed(*real_remote))
       return false;
+
+    auto *limit = static_cast<shared_state&>(
+      connection_basic::get_state()
+    ).plimit;
+    if (limit && limit->is_host_limit(*real_remote))
+      return false;
+
     ec_t ec;
     #if !defined(_WIN32) || !defined(__i686)
     connection_basic::socket_.next_layer().set_option(
@@ -1022,7 +1029,7 @@ namespace net_utils
   template<typename T>
   bool connection<T>::speed_limit_is_enabled() const
   {
-    return m_connection_type != e_connection_type_RPC;
+    return m_connection_type == e_connection_type_P2P;
   }
 
   template<typename T>
@@ -1349,6 +1356,13 @@ namespace net_utils
   }
   //---------------------------------------------------------------------------------
   template<class t_protocol_handler>
+  void boosted_tcp_server<t_protocol_handler>::set_connection_limit(i_connection_limit* plimit)
+  {
+    assert(m_state != nullptr); // always set in constructor
+    m_state->plimit = plimit;
+  }
+  //---------------------------------------------------------------------------------
+  template<class t_protocol_handler>
   bool boosted_tcp_server<t_protocol_handler>::run_server(size_t threads_count, bool wait, const boost::thread::attributes& attrs)
   {
     TRY_ENTRY();

contrib/epee/include/net/network_throttle-detail.hpp

@@ -46,13 +46,13 @@ namespace net_utils
 
 
 class network_throttle : public i_network_throttle {
-	private:
+	public:
 		struct packet_info {
 			size_t m_size; // octets sent. Summary for given small-window (e.g. for all packaged in 1 second)
 			packet_info();
 		};
 
-
+	private:
 		network_speed_bps m_target_speed;
 		size_t m_network_add_cost; // estimated add cost of headers 
 		size_t m_network_minimal_segment; // estimated minimal cost of sending 1 byte to round up to

contrib/epee/src/network_throttle-detail.cpp

@@ -46,7 +46,7 @@
 #include "misc_log_ex.h" 
 #include <boost/chrono.hpp>
 #include "misc_language.h"
-#include <sstream>
+#include <fstream>
 #include <iomanip>
 #include <algorithm>
 
@@ -186,6 +186,23 @@ void network_throttle::handle_trafic_exact(size_t packet_size)
 	_handle_trafic_exact(packet_size, packet_size);
 }
 
+namespace
+{
+	struct output_history
+	{
+		const boost::circular_buffer< network_throttle::packet_info >& history;
+	};
+
+	std::ostream& operator<<(std::ostream& out, const output_history& source)
+	{
+		out << '[';
+		for (auto sample: source.history)
+			out << sample.m_size << ' ';
+		out << ']';
+		return out;
+	}
+}
+
 void network_throttle::_handle_trafic_exact(size_t packet_size, size_t orginal_size)
 {
 	tick();
@@ -196,14 +213,11 @@ void network_throttle::_handle_trafic_exact(size_t packet_size, size_t orginal_s
 	m_total_packets++;
 	m_total_bytes += packet_size;
 
-	std::ostringstream oss; oss << "["; 	for (auto sample: m_history) oss << sample.m_size << " ";	 oss << "]" << std::ends;
-	std::string history_str = oss.str();
-
 	MTRACE("Throttle " << m_name << ": packet of ~"<<packet_size<<"b " << " (from "<<orginal_size<<" b)"
         << " Speed AVG=" << std::setw(4) <<  ((long int)(cts .average/1024)) <<"[w="<<cts .window<<"]"
         <<           " " << std::setw(4) <<  ((long int)(cts2.average/1024)) <<"[w="<<cts2.window<<"]"
 				<<" / " << " Limit="<< ((long int)(m_target_speed/1024)) <<" KiB/sec "
-				<< " " << history_str
+				<< " " << output_history{m_history}
 		);
 }
 
@@ -289,8 +303,6 @@ void network_throttle::calculate_times(size_t packet_size, calculate_times_struc
     }
 
 	if (dbg) {
-		std::ostringstream oss; oss << "["; 	for (auto sample: m_history) oss << sample.m_size << " ";	 oss << "]" << std::ends;
-		std::string history_str = oss.str();
 		MTRACE((cts.delay > 0 ? "SLEEP" : "")
 			<< "dbg " << m_name << ": " 
 			<< "speed is A=" << std::setw(8) <<cts.average<<" vs "
@@ -300,7 +312,7 @@ void network_throttle::calculate_times(size_t packet_size, calculate_times_struc
 			<< "E="<< std::setw(8) << E << " (Enow="<<std::setw(8)<<Enow<<") "
             << "M=" << std::setw(8) << M <<" W="<< std::setw(8) << cts.window << " "
             << "R=" << std::setw(8) << cts.recomendetDataSize << " Wgood" << std::setw(8) << Wgood << " "
-			<< "History: " << std::setw(8) << history_str << " "
+			<< "History: " << std::setw(8) << output_history{m_history} << " "
 			<< "m_last_sample_time=" << std::setw(8) << m_last_sample_time
 		);
 

src/p2p/net_node.h

@@ -124,7 +124,8 @@ namespace nodetool
   template<class t_payload_net_handler>
   class node_server: public epee::levin::levin_commands_handler<p2p_connection_context_t<typename t_payload_net_handler::connection_context> >,
                      public i_p2p_endpoint<typename t_payload_net_handler::connection_context>,
-                     public epee::net_utils::i_connection_filter
+                     public epee::net_utils::i_connection_filter,
+                     public epee::net_utils::i_connection_limit
   {
     struct by_conn_id{};
     struct by_peer_id{};
@@ -350,7 +351,10 @@ namespace nodetool
     virtual bool add_host_fail(const epee::net_utils::network_address &address, unsigned int score = 1);
     //----------------- i_connection_filter  --------------------------------------------------------
     virtual bool is_remote_host_allowed(const epee::net_utils::network_address &address, time_t *t = NULL);
+    //----------------- i_connection_limit  ---------------------------------------------------------
+    virtual bool is_host_limit(const epee::net_utils::network_address &address);
     //-----------------------------------------------------------------------------------------------
+
     bool parse_peer_from_string(epee::net_utils::network_address& pe, const std::string& node_addr, uint16_t default_port = 0);
     bool handle_command_line(
         const boost::program_options::variables_map& vm

src/p2p/net_node.inl

@@ -222,6 +222,26 @@ namespace nodetool
     // not found in hosts or subnets, allowed
     return true;
   }
+  //-----------------------------------------------------------------------------------
+  template<class t_payload_net_handler>
+  bool node_server<t_payload_net_handler>::is_host_limit(const epee::net_utils::network_address &address)
+  {
+    const network_zone& zone = m_network_zones.at(address.get_zone());
+    if (zone.m_current_number_of_in_peers >= zone.m_config.m_net_config.max_in_connection_count) // in peers limit
+    {
+      MWARNING("Exceeded max incoming connections, so dropping this one.");
+      return true;
+    }
+
+    if(has_too_many_connections(address))
+    {
+      MWARNING("CONNECTION FROM " << address.host_str() << " REFUSED, too many connections from the same address");
+      return true;
+    }
+
+    return false;
+  }
+
   //-----------------------------------------------------------------------------------
   template<class t_payload_net_handler>
   bool node_server<t_payload_net_handler>::block_host(epee::net_utils::network_address addr, time_t seconds, bool add_only)
@@ -967,6 +987,7 @@ namespace nodetool
         std::string ipv6_addr = "";
         std::string ipv6_port = "";
         zone.second.m_net_server.set_connection_filter(this);
+        zone.second.m_net_server.set_connection_limit(this);
         MINFO("Binding (IPv4) on " << zone.second.m_bind_ip << ":" << zone.second.m_port);
         if (!zone.second.m_bind_ipv6_address.empty() && m_use_ipv6)
         {
@@ -2543,13 +2564,6 @@ namespace nodetool
       return 1;
     }
 
-    if (zone.m_current_number_of_in_peers >= zone.m_config.m_net_config.max_in_connection_count) // in peers limit
-    {
-      LOG_WARNING_CC(context, "COMMAND_HANDSHAKE came, but already have max incoming connections, so dropping this one.");
-      drop_connection(context);
-      return 1;
-    }
-
     if(!m_payload_handler.process_payload_sync_data(arg.payload_data, context, true))
     {
       LOG_WARNING_CC(context, "COMMAND_HANDSHAKE came, but process_payload_sync_data returned false, dropping connection.");
@@ -2559,13 +2573,6 @@ namespace nodetool
 
     zone.m_notifier.on_handshake_complete(context.m_connection_id, context.m_is_income);
 
-    if(has_too_many_connections(context.m_remote_address))
-    {
-      LOG_PRINT_CCONTEXT_L1("CONNECTION FROM " << context.m_remote_address.host_str() << " REFUSED, too many connections from the same address");
-      drop_connection(context);
-      return 1;
-    }
-
     //associate peer_id with this connection
     context.peer_id = arg.node_data.peer_id;
     context.m_in_timedsync = false;
@@ -2885,15 +2892,16 @@ namespace nodetool
       if (cntxt.m_is_income && cntxt.m_remote_address.is_same_host(address)) {
         count++;
 
-        if (count > max_connections) {
+        // the only call location happens BEFORE foreach_connection list is updated
+        if (count >= max_connections) {
           return false;
         }
       }
 
       return true;
     });
-
-    return count > max_connections;
+    // the only call location happens BEFORE foreach_connection list is updated
+    return count >= max_connections;
   }
 
   template<class t_payload_net_handler>

tests/unit_tests/node_server.cpp

@@ -224,6 +224,18 @@ TEST(ban, subnet)
   test_core pr_core;
   cryptonote::t_cryptonote_protocol_handler<test_core> cprotocol(pr_core, NULL);
   Server server(cprotocol);
+  {
+    boost::program_options::options_description opts{};
+    Server::init_options(opts);
+    cryptonote::core::init_options(opts);
+
+    char** args = nullptr;
+    boost::program_options::variables_map vm;
+    boost::program_options::store(
+      boost::program_options::parse_command_line(0, args, opts), vm
+    );
+    server.init(vm);
+  }
   cprotocol.set_p2p_endpoint(&server);
 
   ASSERT_TRUE(server.block_subnet(MAKE_IPV4_SUBNET(1,2,3,4,24), 10));