fetch msg (#552)

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

Author: turuslan

core/node/blocksync_request.cpp

@@ -13,6 +13,7 @@
 #include "common/logger.hpp"
 #include "primitives/tipset/load.hpp"
 #include "storage/ipfs/datastore.hpp"
+#include "vm/actor/builtin/types/miner/policy.hpp"
 
 #define TRACE_ENABLED 0
 
@@ -22,7 +23,8 @@ namespace fc::sync::blocksync {
   using libp2p::basic::Scheduler;
 
   namespace {
-    constexpr size_t kMaxDepth = 100;
+    constexpr size_t kMaxDepth =
+        vm::actor::builtin::types::miner::kChainFinality;
 
     using primitives::block::MsgMeta;
 
@@ -32,7 +34,7 @@ namespace fc::sync::blocksync {
     }
 
     template <typename... Args>
-    inline void trace(spdlog::string_view_t fmt, const Args &... args) {
+    inline void trace(spdlog::string_view_t fmt, const Args &...args) {
 #if TRACE_ENABLED
       log()->trace(fmt, args...);
 #endif
@@ -214,8 +216,8 @@ namespace fc::sync::blocksync {
           if (meta_cid != header.messages) {
             return BlocksyncRequest::Error::kStoreCidsMismatch;
           }
-          block_stored(std::move(*asBlake(block_cid)), std::move(header));
         }
+        block_stored(std::move(*asBlake(block_cid)), std::move(header));
       }
 
       return outcome::success();
@@ -246,8 +248,10 @@ namespace fc::sync::blocksync {
                        std::function<void(Result)> callback) {
         callback_ = std::move(callback);
         result_.emplace();
+        result_->from = peer;
         result_->blocks_requested = std::move(blocks);
         result_->messages_stored = (options & kMessagesOnly);
+        result_->messages_only = options == kMessagesOnly;
 
         std::vector<CbCid> blocks_reduced =
             tryReduceRequest(result_->blocks_requested,
@@ -260,11 +264,16 @@ namespace fc::sync::blocksync {
           return;
         }
 
-        if (options == kMessagesOnly) {
-          // not supported yet
-          result_->error = BlocksyncRequest::Error::kNotImplemented;
-          scheduleResult();
-          return;
+        if (result_->messages_only) {
+          auto cb{[&](std::error_code ec) {
+            result_->error = ec;
+            scheduleResult();
+          }};
+          result_->blocks_available.clear();
+          for (const auto &cid : blocks_reduced) {
+            OUTCOME_CB(result_->blocks_available.emplace_back(),
+                       getCbor<BlockHeader>(ipld_, CID{cid}));
+          }
         }
 
         waitlist_.insert(blocks_reduced.begin(), blocks_reduced.end());
@@ -285,7 +294,6 @@ namespace fc::sync::blocksync {
         }
 
         options_ = options;
-        result_->from = peer;
 
         host_.newStream(
             // peer must be already connected
@@ -462,6 +470,9 @@ namespace fc::sync::blocksync {
 
         boost::optional<TipsetHash> expected_parent;
 
+        if (result_->messages_only) {
+          chain[0].blocks = std::move(result_->blocks_available);
+        }
         auto res = storeTipsetBundle(
             ipld_,
             put_block_header_,

core/node/blocksync_request.hpp

@@ -48,6 +48,8 @@ namespace fc::sync::blocksync {
 
       /// All their meta/messages is also available
       bool messages_stored = false;
+
+      bool messages_only{false};
     };
 
     /// Cancels existing request if still active and makes the new one

core/node/fetch_msg.hpp

@@ -0,0 +1,100 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <unordered_set>
+
+#include "node/blocksync_request.hpp"
+#include "node/peer_height.hpp"
+
+namespace fc::sync::fetch_msg {
+  using blocksync::BlocksyncRequest;
+  using libp2p::Host;
+  using libp2p::basic::Scheduler;
+  using peer_height::PeerHeight;
+  using primitives::ChainEpoch;
+
+  constexpr size_t kQueueMax{1000};
+  constexpr size_t kFetchingMax{20};
+
+  struct FetchMsg {
+    std::shared_ptr<Host> host;
+    std::shared_ptr<Scheduler> scheduler;
+    std::shared_ptr<PeerHeight> peers;
+    IpldPtr ipld;
+    std::set<std::pair<ChainEpoch, TipsetKey>> queue;
+    std::unordered_map<TipsetKey, std::shared_ptr<BlocksyncRequest>>
+        fetching_tsk;
+    std::unordered_set<PeerId> fetching_peer;
+    std::function<void(TipsetKey)> on_fetch;
+
+    FetchMsg(std::shared_ptr<Host> host,
+             std::shared_ptr<Scheduler> scheduler,
+             std::shared_ptr<PeerHeight> peers,
+             IpldPtr ipld)
+        : host{host}, scheduler{scheduler}, peers{peers}, ipld{ipld} {}
+
+    bool has(const TipsetCPtr &ts, bool priority) {
+      for (const auto &block : ts->blks) {
+        if (!ipld->contains(block.messages).value()) {
+          if (queue.size() >= kQueueMax) {
+            queue.erase(std::prev(queue.end()));
+          }
+          queue.emplace(priority ? -1 : ts->height(), ts->key);
+          dequeue();
+          return false;
+        }
+      }
+      return true;
+    }
+
+    void dequeue() {
+      if (queue.empty()) {
+        return;
+      }
+      if (fetching_tsk.size() >= kFetchingMax) {
+        return;
+      }
+      auto it{queue.begin()};
+      // note: BlocksyncRequests caches duplicate requests
+      peers->visit(it->first, [&](const PeerId &peer) {
+        if (fetching_peer.count(peer) != 0) {
+          return true;
+        }
+        auto tsk{std::move(it->second)};
+        queue.erase(it);
+        fetching_peer.emplace(peer);
+        fetching_tsk.emplace(
+            tsk,
+            BlocksyncRequest::newRequest(
+                *host,
+                *scheduler,
+                ipld,
+                nullptr,
+                peer,
+                tsk.cids(),
+                1,
+                blocksync::kMessagesOnly,
+                10000,
+                [this](BlocksyncRequest::Result r) { onFetch(std::move(r)); }));
+        return false;
+      });
+    }
+
+    void onFetch(BlocksyncRequest::Result &&r) {
+      TipsetKey tsk{std::move(r.blocks_requested)};
+      fetching_tsk.erase(tsk);
+      fetching_peer.erase(*r.from);
+      if (r.error) {
+        peers->onError(*r.from);
+      }
+      dequeue();
+      if (on_fetch) {
+        on_fetch(std::move(tsk));
+      }
+    }
+  };
+}  // namespace fc::sync::fetch_msg

core/node/identify.cpp

@@ -57,6 +57,13 @@ namespace fc::sync {
           }
         });
 
+    on_disconnect_ =
+        host_->getBus()
+            .getChannel<libp2p::event::network::OnPeerDisconnectedChannel>()
+            .subscribe([this](const PeerId &peer) {
+              events_->signalPeerDisconnected({peer});
+            });
+
     handleProtocol(*host_, *identify_protocol_);
     handleProtocol(*host_, *identify_push_protocol_);
     handleProtocol(*host_, *identify_delta_protocol_);

core/node/identify.hpp

@@ -7,6 +7,7 @@
 
 #include "node/events_fwd.hpp"
 
+#include <libp2p/event/bus.hpp>
 #include <libp2p/peer/peer_info.hpp>
 
 namespace fc::sync {
@@ -34,6 +35,7 @@ namespace fc::sync {
 
     std::shared_ptr<events::Events> events_;
     events::Connection on_identify_;
+    libp2p::event::Handle on_disconnect_;
   };
 
 }  // namespace fc::sync

core/node/main/main.cpp

@@ -14,9 +14,9 @@
 #include "api/rpc/make.hpp"
 #include "api/rpc/ws.hpp"
 #include "common/api_secret.hpp"
-#include "common/local_ip.hpp"
 #include "common/libp2p/peer/peer_info_helper.hpp"
 #include "common/libp2p/soralog.hpp"
+#include "common/local_ip.hpp"
 #include "common/logger.hpp"
 #include "common/prometheus/rpc.hpp"
 #include "drand/impl/http.hpp"

core/node/peer_height.hpp

@@ -0,0 +1,62 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <boost/bimap.hpp>
+#include <boost/bimap/multiset_of.hpp>
+#include <boost/bimap/unordered_set_of.hpp>
+
+#include "node/events.hpp"
+
+namespace fc::sync::peer_height {
+  using primitives::ChainEpoch;
+
+  struct PeerHeight {
+    boost::bimap<boost::bimaps::unordered_set_of<PeerId, std::hash<PeerId>>,
+                 boost::bimaps::multiset_of<ChainEpoch>>
+        peers;
+    events::Connection on_peer_head;
+    events::Connection on_disconnect;
+
+    PeerHeight(const std::shared_ptr<events::Events> &events) {
+      on_peer_head =
+          events->subscribePossibleHead([this](const events::PossibleHead &e) {
+            if (e.source) {
+              // note: may require successful fetch
+              onHeight(*e.source, e.height);
+            }
+          });
+      on_disconnect = events->subscribePeerDisconnected(
+          [this](const events::PeerDisconnected &e) { onError(e.peer_id); });
+    }
+
+    void onHeight(const PeerId &peer, ChainEpoch height) {
+      if (height <= 0) {
+        return;
+      }
+      // note: may ignore peers behind our height
+      const auto it{peers.left.find(peer)};
+      if (it == peers.left.end()) {
+        peers.insert({peer, height});
+      } else if (height > it->second) {
+        peers.left.replace_data(it, height);
+      }
+    }
+
+    void onError(const PeerId &peer) {
+      peers.left.erase(peer);
+    }
+
+    template <typename F>
+    void visit(ChainEpoch min, const F &f) {
+      for (auto it{peers.right.rbegin()}; it != peers.right.rend(); ++it) {
+        if (it->first < min || !f(it->second)) {
+          break;
+        }
+      }
+    }
+  };
+}  // namespace fc::sync::peer_height