tx-relay-v2: add maximum in_flight requests for each connection

Author: 0xFFFC0000

src/cryptonote_basic/connection_context.cpp

@@ -129,6 +129,7 @@ namespace cryptonote
   void cryptonote_connection_context::add_requested_from_peer()
   {
     epee::write_lock w_lock(m_connection_stats->mutex);
+    add_in_flight_requests();
     ++m_connection_stats->requested_from_peer;
   }
 
@@ -138,6 +139,25 @@ namespace cryptonote
     ++m_connection_stats->sent;
   }
 
+  void cryptonote_connection_context::add_in_flight_requests()
+  {
+    epee::write_lock w_lock(m_connection_stats->mutex);
+    ++m_connection_stats->in_flight_requests;
+  }
+
+  void cryptonote_connection_context::remove_in_flight_request()
+  {
+    epee::write_lock w_lock(m_connection_stats->mutex);
+    if (m_connection_stats->in_flight_requests > 0)
+      --m_connection_stats->in_flight_requests;
+  }
+
+  bool cryptonote_connection_context::can_process_additional_request(size_t requests)
+  {
+    epee::read_lock r_lock(m_connection_stats->mutex);
+    return (m_connection_stats->in_flight_requests + requests) < P2P_MAX_IN_FLIGHT_REQUESTS;
+  }
+
   size_t cryptonote_connection_context::get_announcement_size() const
   {
     epee::read_lock r_lock(m_connection_stats->mutex);
@@ -184,6 +204,12 @@ namespace cryptonote
     return m_connection_stats->missed;
   }
 
+  size_t cryptonote_connection_context::get_in_flight_requests() const
+  {
+    epee::read_lock r_lock(m_connection_stats->mutex);
+    return m_connection_stats->in_flight_requests;
+  }
+
   std::string cryptonote_connection_context::get_info() const
   {
     epee::read_lock r_lock(m_connection_stats->mutex);

src/cryptonote_basic/connection_context.h

@@ -131,6 +131,7 @@ namespace cryptonote
         size_t requested_from_peer = 0;
         size_t sent = 0;
         size_t missed = 0;
+        size_t in_flight_requests = 0;
       };
       std::shared_ptr<connection_statistics> m_connection_stats;
 
@@ -141,6 +142,9 @@ namespace cryptonote
       void add_requested_from_me();
       void add_requested_from_peer();
       void add_sent();
+      void add_in_flight_requests();
+      void remove_in_flight_request();
+      bool can_process_additional_request(size_t requests = 1);
 
       size_t get_announcement_size() const;
       size_t get_received() const;
@@ -149,6 +153,7 @@ namespace cryptonote
       size_t get_sent() const;
       size_t get_total() const;
       size_t get_missed() const;
+      size_t get_in_flight_requests() const;
       std::string get_info() const;
       bool missed_announced_tx(const crypto::hash &tx_hash);
   };

src/cryptonote_config.h

@@ -153,6 +153,7 @@
 #define P2P_DEFAULT_LIMIT_RATE_UP                       8192       // kB/s
 #define P2P_DEFAULT_LIMIT_RATE_DOWN                     32768       // kB/s
 #define P2P_REQUEST_FAILURE_THRESHOLD_PERCENTAGE        70          // if more than 70% of requests fail, the peer is dropped
+#define P2P_MAX_IN_FLIGHT_REQUESTS                      100
 
 #define P2P_FAILED_ADDR_FORGET_SECONDS                  (60*60)     //1 hour
 #define P2P_IP_BLOCKTIME                                (60*60*24)  //24 hour

src/cryptonote_protocol/cryptonote_protocol_handler.inl

@@ -904,13 +904,17 @@ namespace cryptonote
     // Iterate over each advertised transaction hash and check our pool and our requested tracker.
     for (const auto &tx_hash : arg.t)
     {
+      if (!context.can_process_additional_request())
+      {
+        LOG_ERROR_CCONTEXT("Too many in-flight requests, cannot accept");
+        break;
+      }
       // If we have the tx in our pool, also remove it from request manager and skip.
       if (m_core.pool_has_tx(tx_hash))
       {
         m_request_manager.remove_transaction(tx_hash);
         continue;
       }
-
       bool need_request = m_request_manager.add_transaction(tx_hash, context.m_connection_id, now);
       context.add_announcement(tx_hash);
       if (need_request) {
@@ -1043,6 +1047,7 @@ namespace cryptonote
             && m_request_manager.already_requested_tx(tx_hash))
         {
           m_request_manager.remove_transaction(tx_hash);
+          context.remove_in_flight_request();
         }
       }
 

src/cryptonote_protocol/txrequestqueue.h

@@ -47,7 +47,7 @@
 /**
  * \brief A queue of requests. The front of the queue shows the current request.
  *
- * This queue holds peers that have the requested transaction and keeps
+ * This queue holds peers that have announced the transaction and keeps
  * track of the time the last request was made. It provides capabilities
  * to add a new peer and to retrieve the next peer that hasn't been requested.
  */
@@ -72,8 +72,14 @@ class tx_request_queue {
   public:
     struct order {
       bool operator()(const tx_request &lhs, const tx_request &rhs) const {
-        return lhs.from_connection_id != rhs.from_connection_id &&
-               lhs.first_seen < rhs.first_seen;
+        if (lhs.get_connection_id() == rhs.get_connection_id())
+          return false; // same peer => equivalent (disallow duplicates)
+        if (lhs.get_first_seen() < rhs.get_first_seen())
+          return true;
+        if (rhs.get_first_seen() < lhs.get_first_seen())
+          return false;
+        // tie-break to maintain strict weak ordering when first_seen is equal
+        return lhs.get_connection_id() < rhs.get_connection_id();
       }
     };
 
@@ -84,13 +90,6 @@ class tx_request_queue {
         : from_connection_id(other.from_connection_id),
           first_seen(other.first_seen), request_time(other.request_time) {}
 
-    bool operator==(const tx_request &other) const {
-      return from_connection_id == other.get_connection_id();
-    }
-    bool operator!=(const tx_request &other) const {
-      return from_connection_id != other.get_connection_id();
-    }
-
     const boost::uuids::uuid &get_connection_id() const {
       return from_connection_id;
     }