SOON

include/libp2p/injector/host_injector.hpp

@@ -24,6 +24,7 @@
 #include <libp2p/peer/key_repository/inmem_key_repository.hpp>
 #include <libp2p/peer/peer_repository.hpp>
 #include <libp2p/peer/protocol_repository/inmem_protocol_repository.hpp>
+#include <libp2p/peer/rtt_repository/inmem_rtt_repository.hpp>
 #include <libp2p/protocol/gossip/config.hpp>
 #include <libp2p/protocol_muxer/multiselect.hpp>
 
@@ -98,6 +99,7 @@ namespace libp2p::injector {
         di::bind<peer::KeyRepository>.to<peer::InmemKeyRepository>(),
         di::bind<peer::AddressRepository>.to<peer::InmemAddressRepository>(),
         di::bind<peer::ProtocolRepository>.to<peer::InmemProtocolRepository>(),
+        di::bind<peer::RttRepository>.to<peer::InmemRttRepository>(),
         di::bind<protocol_muxer::ProtocolMuxer>.to<protocol_muxer::multiselect::Multiselect>(),
         di::bind<crypto::validator::KeyValidator>.to<crypto::validator::KeyValidatorImpl>(),
         di::bind<crypto::CryptoProvider>.to<crypto::CryptoProviderImpl>(),

include/libp2p/peer/peer_repository.hpp

@@ -13,6 +13,7 @@
 #include <libp2p/peer/peer_id.hpp>
 #include <libp2p/peer/peer_info.hpp>
 #include <libp2p/peer/protocol_repository.hpp>
+#include <libp2p/peer/rtt_repository.hpp>
 
 namespace libp2p::peer {
   /**
@@ -23,7 +24,8 @@ namespace libp2p::peer {
    public:
     PeerRepository(std::shared_ptr<AddressRepository> addrRepo,
                    std::shared_ptr<KeyRepository> keyRepo,
-                   std::shared_ptr<ProtocolRepository> protocolRepo);
+                   std::shared_ptr<ProtocolRepository> protocolRepo,
+                   std::shared_ptr<RttRepository> rttRepo);
     /**
      * @brief Getter for an address repository.
      * @return associated instance of an address repository.
@@ -42,6 +44,12 @@ namespace libp2p::peer {
      */
     ProtocolRepository &getProtocolRepository();
 
+    /**
+     * @brief Getter for an RTT repository.
+     * @return associated instance of an RTT repository.
+     */
+    RttRepository &getRttRepository();
+
     /**
      * @brief Returns set of peer ids known by this peer repository.
      * @return unordered set of peers
@@ -61,5 +69,6 @@ namespace libp2p::peer {
     std::shared_ptr<AddressRepository> addr_;
     std::shared_ptr<KeyRepository> key_;
     std::shared_ptr<ProtocolRepository> proto_;
+    std::shared_ptr<RttRepository> rtt_;
   };
 }  // namespace libp2p::peer

include/libp2p/peer/rtt_repository.hpp

@@ -0,0 +1,47 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <chrono>
+#include <optional>
+
+#include <libp2p/basic/garbage_collectable.hpp>
+#include <libp2p/peer/peer_id.hpp>
+
+namespace libp2p::peer {
+
+  /**
+   * @brief Repository to store Round Trip Time (RTT) for peers.
+   * It calculates smoothed RTT using EWMA.
+   */
+  class RttRepository : public basic::GarbageCollectable {
+   public:
+    virtual ~RttRepository() = default;
+
+    /**
+     * @brief Update RTT for a peer.
+     * @param p PeerId
+     * @param rtt Measured RTT
+     */
+    virtual void updateRtt(const PeerId &p, std::chrono::microseconds rtt) = 0;
+
+    /**
+     * @brief Get smoothed RTT for a peer.
+     * @param p PeerId
+     * @return Smoothed RTT if available, std::nullopt otherwise.
+     */
+    virtual std::optional<std::chrono::microseconds> getRtt(
+        const PeerId &p) const = 0;
+
+    /**
+     * @brief Remove RTT information for a peer.
+     * @param p PeerId
+     */
+    virtual void clear(const PeerId &p) = 0;
+  };
+
+}  // namespace libp2p::peer

include/libp2p/peer/rtt_repository/inmem_rtt_repository.hpp

@@ -0,0 +1,35 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <libp2p/peer/rtt_repository.hpp>
+
+#include <unordered_map>
+
+namespace libp2p::peer {
+
+  class InmemRttRepository : public RttRepository {
+   public:
+    void updateRtt(const PeerId &p, std::chrono::microseconds rtt) override;
+
+    std::optional<std::chrono::microseconds> getRtt(
+        const PeerId &p) const override;
+
+    void clear(const PeerId &p) override;
+
+    void collectGarbage() override;
+
+   private:
+    struct PeerRtt {
+      std::chrono::microseconds value;
+      std::chrono::steady_clock::time_point last_updated;
+    };
+
+    std::unordered_map<PeerId, PeerRtt> rtt_map_;
+  };
+
+}  // namespace libp2p::peer

include/libp2p/protocol/gossip/config.hpp

@@ -192,6 +192,11 @@ namespace libp2p::protocol::gossip {
     size_t opportunistic_graft_ticks = 60;
     size_t opportunistic_graft_peers = 2;
 
+    /// Number of peers from the mesh to which we propagate messages.
+    /// Peers are selected by the smallest RTT.
+    /// If -1, propagate to all peers in the mesh.
+    int soon_delta = -1;
+
     ScoreConfig score;
   };
 }  // namespace libp2p::protocol::gossip

src/CMakeLists.txt

@@ -41,6 +41,7 @@ target_link_libraries(libp2p
         p2p_inmem_address_repository
         p2p_inmem_key_repository
         p2p_inmem_protocol_repository
+        p2p_inmem_rtt_repository
         p2p_key_validator
         p2p_listener_manager
         p2p_literals

src/peer/CMakeLists.txt

@@ -7,6 +7,7 @@
 add_subdirectory(address_repository)
 add_subdirectory(key_repository)
 add_subdirectory(protocol_repository)
+add_subdirectory(rtt_repository)
 
 libp2p_add_library(p2p_peer_errors
     errors.cpp

src/peer/peer_repository.cpp

@@ -23,13 +23,16 @@ namespace libp2p::peer {
   PeerRepository::PeerRepository(
       std::shared_ptr<AddressRepository> addr_repo,
       std::shared_ptr<KeyRepository> key_repo,
-      std::shared_ptr<ProtocolRepository> protocol_repo)
+      std::shared_ptr<ProtocolRepository> protocol_repo,
+      std::shared_ptr<RttRepository> rtt_repo)
       : addr_(std::move(addr_repo)),
         key_(std::move(key_repo)),
-        proto_(std::move(protocol_repo)) {
+        proto_(std::move(protocol_repo)),
+        rtt_(std::move(rtt_repo)) {
     BOOST_ASSERT(addr_ != nullptr);
     BOOST_ASSERT(key_ != nullptr);
     BOOST_ASSERT(proto_ != nullptr);
+    BOOST_ASSERT(rtt_ != nullptr);
   }
 
   AddressRepository &PeerRepository::getAddressRepository() {
@@ -44,11 +47,17 @@ namespace libp2p::peer {
     return *proto_;
   }
 
+  RttRepository &PeerRepository::getRttRepository() {
+    return *rtt_;
+  }
+
   std::unordered_set<PeerId> PeerRepository::getPeers() const {
     std::unordered_set<PeerId> peers;
     merge_sets<PeerId>(peers, addr_->getPeers());
     merge_sets<PeerId>(peers, key_->getPeers());
     merge_sets<PeerId>(peers, proto_->getPeers());
+    // RttRepository doesn't have getPeers() yet, but it's fine.
+    // Usually peers in RttRepository should be in others too.
     return peers;
   }
 

src/peer/rtt_repository/CMakeLists.txt

@@ -0,0 +1,13 @@
+#
+# Copyright Quadrivium LLC
+# All Rights Reserved
+# SPDX-License-Identifier: Apache-2.0
+#
+
+libp2p_add_library(p2p_inmem_rtt_repository
+    inmem_rtt_repository.cpp
+    )
+target_link_libraries(p2p_inmem_rtt_repository
+    p2p_peer_id
+    )
+

src/peer/rtt_repository/inmem_rtt_repository.cpp

@@ -0,0 +1,53 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include <libp2p/peer/rtt_repository/inmem_rtt_repository.hpp>
+
+namespace libp2p::peer {
+
+  void InmemRttRepository::updateRtt(const PeerId &p,
+                                     std::chrono::microseconds rtt) {
+    auto now = std::chrono::steady_clock::now();
+    auto it = rtt_map_.find(p);
+    if (it == rtt_map_.end()) {
+      rtt_map_.emplace(p, PeerRtt{rtt, now});
+    } else {
+      // EWMA: SRTT = (1 - alpha) * SRTT + alpha * RTT
+      // alpha = 0.125 (1/8) is standard for TCP
+      // SRTT = (7 * SRTT + RTT) / 8
+      it->second.value = (it->second.value * 7 + rtt) / 8;
+      it->second.last_updated = now;
+    }
+  }
+
+  std::optional<std::chrono::microseconds> InmemRttRepository::getRtt(
+      const PeerId &p) const {
+    auto it = rtt_map_.find(p);
+    if (it != rtt_map_.end()) {
+      return it->second.value;
+    }
+    return std::nullopt;
+  }
+
+  void InmemRttRepository::clear(const PeerId &p) {
+    rtt_map_.erase(p);
+  }
+
+  void InmemRttRepository::collectGarbage() {
+    auto now = std::chrono::steady_clock::now();
+    // Expire entries older than 1 hour
+    auto ttl = std::chrono::hours(1);
+
+    for (auto it = rtt_map_.begin(); it != rtt_map_.end();) {
+      if (now - it->second.last_updated > ttl) {
+        it = rtt_map_.erase(it);
+      } else {
+        ++it;
+      }
+    }
+  }
+
+}  // namespace libp2p::peer

src/protocol/gossip/gossip.cpp

@@ -5,6 +5,7 @@
  */
 
 #include <generated/protocol/gossip/gossip.pb.h>
+#include <algorithm>
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/steady_timer.hpp>
 #include <boost/endian/conversion.hpp>
@@ -729,8 +730,34 @@ namespace libp2p::protocol::gossip {
           add_peer(peer);
         }
       }
-      for (auto &peer : topic.mesh_peers_) {
-        add_peer(peer);
+      if (config_.soon_delta >= 0) {
+        if (topic.mesh_peers_.size()
+            <= static_cast<size_t>(config_.soon_delta)) {
+          for (auto &peer : topic.mesh_peers_) {
+            add_peer(peer);
+          }
+        } else {
+          std::vector<PeerPtr> sorted_mesh_peers(topic.mesh_peers_.begin(),
+                                                 topic.mesh_peers_.end());
+          auto &rtt_repo = host_->getPeerRepository().getRttRepository();
+          std::sort(sorted_mesh_peers.begin(), sorted_mesh_peers.end(),
+                    [&](const PeerPtr &a, const PeerPtr &b) {
+                      auto rtt_a = rtt_repo.getRtt(a->peer_id_);
+                      auto rtt_b = rtt_repo.getRtt(b->peer_id_);
+                      auto val_a =
+                          rtt_a.value_or(std::chrono::microseconds::max());
+                      auto val_b =
+                          rtt_b.value_or(std::chrono::microseconds::max());
+                      return val_a < val_b;
+                    });
+          for (int i = 0; i < config_.soon_delta; ++i) {
+            add_peer(sorted_mesh_peers[i]);
+          }
+        }
+      } else {
+        for (auto &peer : topic.mesh_peers_) {
+          add_peer(peer);
+        }
       }
       if (publish) {
         if (auto more =

src/protocol/ping.cpp

@@ -81,12 +81,16 @@ namespace libp2p::protocol {
         }
         stream = stream_result.value();
       }
-      auto r = co_await ping(
-          stream,
-          std::chrono::duration_cast<std::chrono::milliseconds>(config_.timeout));
+      auto r =
+          co_await ping(stream,
+                        std::chrono::duration_cast<std::chrono::milliseconds>(
+                            config_.timeout));
       if (not r.has_value()) {
         stream->reset();
         stream.reset();
+      } else {
+        host_->getPeerRepository().getRttRepository().updateRtt(
+            connection->remotePeer(), r.value());
       }
       timer.expires_after(config_.interval);
       co_await timer.async_wait(boost::asio::use_awaitable);
@@ -103,6 +107,10 @@ namespace libp2p::protocol {
     auto stream = stream_result.value();
     auto res = co_await ping(stream, timeout);
     stream->close();
+    if (res.has_value()) {
+      host_->getPeerRepository().getRttRepository().updateRtt(
+          conn->remotePeer(), res.value());
+    }
     co_return res;
   }
 
@@ -132,8 +140,8 @@ namespace libp2p::protocol {
     auto end = std::chrono::steady_clock::now();
     timer.cancel();
     if (r.has_value()) {
-      co_return std::chrono::duration_cast<std::chrono::microseconds>(end -
-                                                                      start);
+      co_return std::chrono::duration_cast<std::chrono::microseconds>(end
+                                                                      - start);
     }
     co_return r.error();
   }