stream open queue (#488)

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

Author: turuslan

core/api/full_node/make.cpp

@@ -667,6 +667,7 @@ namespace fc::api {
                   vm::message::MessageSignerImpl{key_store}.sign(message.from,
                                                                  message));
       OUTCOME_TRY(mpool->add(signed_message));
+      spdlog::info("MpoolPushMessage {}", signed_message.getCid());
       return std::move(signed_message);
     };
     api->MpoolSelect = [=](auto &tsk, auto ticket_quality)

core/common/libp2p/CMakeLists.txt

@@ -10,3 +10,10 @@ add_library(cbor_stream
 target_link_libraries(cbor_stream
     cbor
     )
+
+add_library(stream_open_queue
+    stream_open_queue.cpp
+    )
+target_link_libraries(stream_open_queue
+    p2p::p2p
+    )

core/common/libp2p/stream_open_queue.cpp

@@ -0,0 +1,59 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "common/libp2p/stream_open_queue.hpp"
+
+namespace libp2p::connection {
+  StreamOpenQueue::Active::Active(std::shared_ptr<Stream> stream,
+                                  std::weak_ptr<StreamOpenQueue> weak,
+                                  List::iterator it)
+      : StreamProxy{std::move(stream)},
+        weak{std::move(weak)},
+        it{std::move(it)} {}
+
+  StreamOpenQueue::Active::~Active() {
+    stream->close([stream{stream}](auto) { stream->reset(); });
+    if (auto queue{weak.lock()}) {
+      queue->active.erase(it);
+      queue->check();
+    }
+  }
+
+  StreamOpenQueue::StreamOpenQueue(std::shared_ptr<Host> host,
+                                   size_t max_active)
+      : host{std::move(host)}, max_active{max_active} {
+    assert(max_active >= 1);
+  }
+
+  void StreamOpenQueue::open(Pending item) {
+    queue.emplace(std::move(item));
+    check();
+  }
+
+  void StreamOpenQueue::check() {
+    while (!queue.empty() && active.size() < max_active) {
+      auto item{std::move(queue.front())};
+      queue.pop();
+      auto it{active.insert(active.begin(), Active::List::value_type{})};
+      host->newStream(item.peer,
+                      item.protocol,
+                      [weak{weak_from_this()},
+                       cb{std::move(item.cb)},
+                       it{std::move(it)}](Host::StreamResult _stream) {
+                        if (auto self{weak.lock()}) {
+                          if (!_stream) {
+                            self->active.erase(it);
+                            self->check();
+                            return cb(_stream);
+                          }
+                          auto stream{std::make_shared<Active>(
+                              std::move(_stream.value()), weak, it)};
+                          *it = stream;
+                          cb(std::move(stream));
+                        }
+                      });
+    }
+  }
+}  // namespace libp2p::connection

core/common/libp2p/stream_open_queue.hpp

@@ -0,0 +1,45 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <libp2p/host/host.hpp>
+#include <list>
+#include <queue>
+
+#include "common/libp2p/stream_proxy.hpp"
+
+namespace libp2p::connection {
+  struct StreamOpenQueue : std::enable_shared_from_this<StreamOpenQueue> {
+    struct Active : StreamProxy {
+      using List = std::list<std::weak_ptr<Active>>;
+
+      std::weak_ptr<StreamOpenQueue> weak;
+      List::iterator it;
+
+      Active(std::shared_ptr<Stream> stream,
+             std::weak_ptr<StreamOpenQueue> weak,
+             List::iterator it);
+      ~Active() override;
+    };
+
+    struct Pending {
+      peer::PeerInfo peer;
+      peer::Protocol protocol;
+      Host::StreamResultHandler cb;
+    };
+
+    std::shared_ptr<Host> host;
+    size_t max_active;
+    std::queue<Pending> queue;
+    Active::List active;
+
+    StreamOpenQueue(std::shared_ptr<Host> host, size_t max_active);
+
+    void open(Pending item);
+
+    void check();
+  };
+}  // namespace libp2p::connection

core/data_transfer/CMakeLists.txt

@@ -9,4 +9,5 @@ add_library(data_transfer
 target_link_libraries(data_transfer
     cbor_stream
     graphsync
+    stream_open_queue
     )

core/data_transfer/dt.cpp

@@ -24,6 +24,8 @@ size_t std::hash<fc::data_transfer::PeerDtId>::operator()(
 }
 
 namespace fc::data_transfer {
+  constexpr size_t kStreamOpenMax{20};
+
   using common::libp2p::CborStream;
   using libp2p::protocol::Subscription;
   using storage::ipld::kAllSelector;
@@ -52,6 +54,7 @@ namespace fc::data_transfer {
       std::shared_ptr<Host> host, std::shared_ptr<Graphsync> gs) {
     auto dt{std::make_shared<DataTransfer>()};
     dt->host = host;
+    dt->streams = std::make_shared<StreamOpenQueue>(host, kStreamOpenMax);
     dt->gs = gs;
     gs->setRequestHandler(
         [_dt{weaken(dt)}](auto pgsid, auto req) {
@@ -356,11 +359,15 @@ namespace fc::data_transfer {
 
   void DataTransfer::dtSend(const PeerInfo &peer,
                             const DataTransferMessage &msg) {
-    host->newStream(peer, kProtocol, [msg](auto _s) {
-      if (_s) {
-        auto s{std::make_shared<CborStream>(_s.value())};
-        s->write(msg, [s](auto) { s->close(); });
-      }
+    streams->open({
+        peer,
+        kProtocol,
+        [msg](Host::StreamResult _stream) {
+          if (_stream) {
+            auto stream{std::make_shared<CborStream>(_stream.value())};
+            stream->write(msg, [stream](outcome::result<size_t>) {});
+          }
+        },
     });
   }
 }  // namespace fc::data_transfer

core/data_transfer/dt.hpp

@@ -5,6 +5,7 @@
 
 #pragma once
 
+#include "common/libp2p/stream_open_queue.hpp"
 #include "data_transfer/message.hpp"
 #include "fwd.hpp"
 #include "storage/ipfs/graphsync/graphsync.hpp"
@@ -26,6 +27,7 @@ namespace fc::data_transfer {
 
   using gsns::Graphsync;
   using libp2p::Host;
+  using libp2p::connection::StreamOpenQueue;
   using libp2p::peer::PeerId;
   using storage::ipld::traverser::Traverser;
   using PeerGsId = gsns::FullRequestId;
@@ -94,6 +96,7 @@ namespace fc::data_transfer {
     void dtSend(const PeerInfo &peer, const DataTransferMessage &msg);
 
     std::shared_ptr<Host> host;
+    std::shared_ptr<StreamOpenQueue> streams;
     std::shared_ptr<Graphsync> gs;
     std::map<std::string, OnPush> on_push;
     std::map<std::string, OnPull> on_pull;

core/markets/retrieval/client/impl/retrieval_client_impl.cpp

@@ -112,9 +112,10 @@ namespace fc::markets::retrieval::client {
             auto unseal{res.status == DealStatus::kDealStatusFundsNeededUnseal};
             const auto last{res.status
                             == DealStatus::kDealStatusFundsNeededLastPayment};
+            const auto pay{res.status == DealStatus::kDealStatusFundsNeeded};
             const auto done{res.status == DealStatus::kDealStatusCompleted};
             const auto accepted{res.status == DealStatus::kDealStatusAccepted};
-            deal->accepted = unseal || accepted || last || done;
+            deal->accepted = unseal || accepted || pay || last || done;
             if (!deal->accepted) {
               deal->handler(
                   res.status == DealStatus::kDealStatusRejected

core/markets/storage/client/CMakeLists.txt

@@ -21,4 +21,5 @@ target_link_libraries(storage_market_client
     tipset
     memory_indexed_car
     storage_market_import_manager
+    stream_open_queue
     )