update gossip (part)

Signed-off-by: turuslan <[email protected]>

Author: turuslan

Diff for: include/libp2p/basic/scheduler/manual_scheduler_backend.hpp

@@ -66,9 +66,9 @@ namespace libp2p::basic {
       }
     }
 
-   private:
     void callDeferred();
 
+   private:
     /// Current time, set manually
     std::chrono::milliseconds current_clock_;
 

Diff for: include/libp2p/connection/stream_pair.hpp

@@ -0,0 +1,20 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <libp2p/peer/peer_id.hpp>
+
+namespace libp2p::basic {
+  class Scheduler;
+}  // namespace libp2p::basic
+
+namespace libp2p::connection {
+  struct Stream;
+
+  std::pair<std::shared_ptr<Stream>, std::shared_ptr<Stream>> streamPair(
+      std::shared_ptr<basic::Scheduler> post, PeerId peer1, PeerId peer2);
+}  // namespace libp2p::connection

Diff for: include/libp2p/peer/protocol.hpp

@@ -17,3 +17,7 @@ namespace libp2p::peer {
       std::string;
 
 }  // namespace libp2p::peer
+
+namespace libp2p {
+  using peer::ProtocolName;
+}  // namespace libp2p

Diff for: include/libp2p/protocol/gossip/explicit_peers.hpp

@@ -0,0 +1,18 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <libp2p/peer/peer_id.hpp>
+
+namespace libp2p::protocol::gossip {
+  class ExplicitPeers {
+   public:
+    bool contains(const PeerId &) const {
+      return false;
+    }
+  };
+}  // namespace libp2p::protocol::gossip

Diff for: include/libp2p/protocol/gossip/gossip.hpp

@@ -10,14 +10,18 @@
 #include <functional>
 #include <set>
 #include <string>
+#include <unordered_map>
 #include <vector>
 
 #include <boost/optional.hpp>
 
 #include <libp2p/common/byteutil.hpp>
 #include <libp2p/multi/multiaddress.hpp>
 #include <libp2p/peer/peer_id.hpp>
+#include <libp2p/peer/protocol.hpp>
 #include <libp2p/protocol/common/subscription.hpp>
+#include <libp2p/protocol/gossip/peer_kind.hpp>
+#include <libp2p/protocol/gossip/score_config.hpp>
 
 namespace libp2p {
   struct Host;
@@ -41,6 +45,7 @@ namespace libp2p::protocol::gossip {
   struct Config {
     /// Network density factors for gossip meshes
     size_t D_min = 5;
+    size_t D = 6;
     size_t D_max = 10;
 
     /// Ideal number of connected peers to support the network
@@ -50,27 +55,12 @@ namespace libp2p::protocol::gossip {
     /// incoming peers will be rejected
     size_t max_connections_num = 1000;
 
-    /// Forward messages to all subscribers not in mesh
-    /// (floodsub mode compatibility)
-    bool floodsub_forward_mode = false;
-
     /// Forward local message to local subscribers
     bool echo_forward_mode = false;
 
     /// Read or write timeout per whole network operation
     std::chrono::milliseconds rw_timeout_msec{std::chrono::seconds(10)};
 
-    /// Lifetime of a message in message cache
-    std::chrono::milliseconds message_cache_lifetime_msec{
-        std::chrono::minutes(2)};
-
-    /// Topic's message seen cache lifetime
-    std::chrono::milliseconds seen_cache_lifetime_msec{
-        message_cache_lifetime_msec * 3 / 4};
-
-    /// Topic's seen cache limit
-    unsigned seen_cache_limit = 100;
-
     /// Heartbeat interval
     std::chrono::milliseconds heartbeat_interval_msec{1000};
 
@@ -86,11 +76,36 @@ namespace libp2p::protocol::gossip {
     /// Max RPC message size
     size_t max_message_size = 1 << 24;
 
-    /// Protocol version
-    std::string protocol_version = "/meshsub/1.0.0";
+    /// Protocol versions
+    std::unordered_map<ProtocolName, PeerKind> protocol_versions{
+        {"/floodsub/1.0.0", PeerKind::Floodsub},
+        {"/meshsub/1.0.0", PeerKind::Gossipsub},
+        {"/meshsub/1.1.0", PeerKind::Gossipsubv1_1},
+        {"/meshsub/1.2.0", PeerKind::Gossipsubv1_2},
+    };
 
     /// Sign published messages
     bool sign_messages = false;
+
+    size_t history_length{5};
+
+    size_t history_gossip{3};
+
+    std::chrono::seconds fanout_ttl{60};
+
+    std::chrono::seconds duplicate_cache_time{60};
+
+    std::chrono::seconds prune_backoff{60};
+
+    std::chrono::seconds unsubscribe_backoff{10};
+
+    size_t backoff_slack = 1;
+
+    bool flood_publish = true;
+
+    size_t max_ihave_length = 5000;
+
+    ScoreConfig score;
   };
 
   using TopicId = std::string;

Diff for: include/libp2p/protocol/gossip/peer_kind.hpp

@@ -0,0 +1,19 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <cstdint>
+
+namespace libp2p::protocol::gossip {
+  enum class PeerKind : uint8_t {
+    NotSupported,
+    Floodsub,
+    Gossipsub,
+    Gossipsubv1_1,
+    Gossipsubv1_2,
+  };
+}  // namespace libp2p::protocol::gossip

Diff for: include/libp2p/protocol/gossip/score.hpp

@@ -0,0 +1,18 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <libp2p/peer/peer_id.hpp>
+
+namespace libp2p::protocol::gossip {
+  class Score {
+   public:
+    bool below(const PeerId &peer_id, double threshold) {
+      return false;
+    }
+  };
+}  // namespace libp2p::protocol::gossip

Diff for: include/libp2p/protocol/gossip/score_config.hpp

@@ -0,0 +1,15 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+namespace libp2p::protocol::gossip {
+  struct ScoreConfig {
+    double zero = 0;
+    double gossip_threshold = -10;
+    double publish_threshold = -50;
+  };
+}  // namespace libp2p::protocol::gossip

Diff for: include/libp2p/protocol/gossip/time_cache.hpp

@@ -0,0 +1,81 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <chrono>
+#include <deque>
+#include <qtils/empty.hpp>
+#include <unordered_map>
+#include <unordered_set>
+
+namespace libp2p::protocol::gossip::time_cache {
+  using Ttl = std::chrono::milliseconds;
+  using Clock = std::chrono::steady_clock;
+  using Time = Clock::time_point;
+
+  template <typename K, typename V>
+  class TimeCache {
+   public:
+    TimeCache(Ttl ttl) : ttl_{ttl} {}
+
+    bool contains(const K &key) const {
+      return map_.contains(key);
+    }
+
+    void clearExpired(Time now = Clock::now()) {
+      while (not expirations_.empty() and expirations_.front().first <= now) {
+        map_.erase(expirations_.front().second);
+        expirations_.pop_front();
+      }
+    }
+
+    V &getOrDefault(const K &key, Time now = Clock::now()) {
+      clearExpired(now);
+      auto it = map_.find(key);
+      if (it == map_.end()) {
+        it = map_.emplace(key, V()).first;
+        expirations_.emplace_back(now + ttl_, it);
+      }
+      return it->second;
+    }
+
+   private:
+    using Map = std::unordered_map<K, V>;
+
+    Ttl ttl_;
+    Map map_;
+    std::deque<std::pair<Clock::time_point, typename Map::iterator>>
+        expirations_;
+  };
+
+  template <typename K>
+  class DuplicateCache {
+   public:
+    DuplicateCache(Ttl ttl) : cache_{ttl} {}
+
+    bool contains(const K &key) const {
+      return cache_.contains(key);
+    }
+
+    bool insert(const K &key, Time now = Clock::now()) {
+      cache_.clearExpired(now);
+      if (cache_.contains(key)) {
+        return false;
+      }
+      cache_.getOrDefault(key);
+      return true;
+    }
+
+   private:
+    TimeCache<K, qtils::Empty> cache_;
+  };
+}  // namespace libp2p::protocol::gossip::time_cache
+
+namespace libp2p::protocol::gossip {
+  using time_cache::DuplicateCache;
+  using time_cache::TimeCache;
+}  // namespace libp2p::protocol::gossip

Diff for: src/basic/message_read_writer_uvarint.cpp

@@ -32,8 +32,8 @@ namespace libp2p::basic {
           }
 
           auto msg_len = varint_res.value().toUInt64();
+          auto buffer = std::make_shared<std::vector<uint8_t>>(msg_len, 0);
           if (0 != msg_len) {
-            auto buffer = std::make_shared<std::vector<uint8_t>>(msg_len, 0);
             self->conn_->read(
                 *buffer,
                 msg_len,
@@ -44,7 +44,7 @@ namespace libp2p::basic {
                   cb(std::move(buffer));
                 });
           } else {
-            cb(ResultType{});
+            cb(buffer);
           }
         });
   }

Diff for: src/connection/CMakeLists.txt

@@ -19,5 +19,12 @@ target_link_libraries(p2p_loopback_stream
     p2p_peer_id
     )
 
+libp2p_add_library(p2p_stream_pair
+    stream_pair.cpp
+    )
+target_link_libraries(p2p_stream_pair
+    p2p_peer_id
+    )
+
 libp2p_install(p2p_loopback_stream)