Parachains. Pipelined thread pool implementation.

core/injector/application_injector.cpp

@@ -113,6 +113,7 @@
 #include "outcome/outcome.hpp"
 #include "parachain/availability/bitfield/store_impl.hpp"
 #include "parachain/availability/store/store_impl.hpp"
+#include "parachain/thread_pool.hpp"
 #include "parachain/validator/parachain_observer.hpp"
 #include "parachain/validator/parachain_processor.hpp"
 #include "runtime/binaryen/binaryen_memory_provider.hpp"
@@ -1586,6 +1587,10 @@ namespace kagome::injector {
     return pimpl_->injector_.create<sptr<parachain::ParachainProcessorImpl>>();
   }
 
+  std::shared_ptr<thread::ThreadPool> KagomeNodeInjector::injectThreadPool() {
+    return pimpl_->injector_.create<sptr<thread::ThreadPool>>();
+  }
+
   std::shared_ptr<consensus::babe::Babe> KagomeNodeInjector::injectBabe() {
     return pimpl_->injector_.create<sptr<consensus::babe::Babe>>();
   }

core/injector/application_injector.hpp

@@ -18,6 +18,10 @@ namespace soralog {
 }
 
 namespace kagome {
+  namespace thread {
+    struct ThreadPool;
+  }
+
   namespace application {
     class AppConfiguration;
     class ChainSpec;
@@ -102,6 +106,7 @@ namespace kagome::injector {
     std::shared_ptr<parachain::ParachainObserverImpl> injectParachainObserver();
     std::shared_ptr<parachain::ParachainProcessorImpl>
     injectParachainProcessor();
+    std::shared_ptr<thread::ThreadPool> injectThreadPool();
     std::shared_ptr<consensus::grandpa::Grandpa> injectGrandpa();
     std::shared_ptr<soralog::LoggingSystem> injectLoggingSystem();
     std::shared_ptr<storage::trie::TrieStorage> injectTrieStorage();

core/parachain/CMakeLists.txt

@@ -1,3 +1,9 @@
+add_library(thread_pool
+        thread_pool.cpp
+        )
+target_link_libraries(thread_pool
+        Boost::boost
+        )
 
 add_library(validator_parachain
     availability/bitfield/signer.cpp
@@ -14,4 +20,5 @@ target_link_libraries(validator_parachain
     collation_protocol
     protocol_error
     peer_view
+    thread_pool
     )

core/parachain/tasks_sequence.hpp

@@ -0,0 +1,144 @@
+//
+// Created by iceseer on 11/20/22.
+//
+
+#ifndef KAGOME_TASKS_SEQUENCE_HPP
+#define KAGOME_TASKS_SEQUENCE_HPP
+
+#include <memory>
+#include <type_traits>
+
+#include <boost/asio/executor_work_guard.hpp>
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/io_service.hpp>
+#include <boost/asio/post.hpp>
+#include <boost/asio/signal_set.hpp>
+
+namespace kagome::thread {
+
+  // clang-format off
+  /*
+   * Code of `sequence` allows to execute multiple tasks in different threads
+   * like so the output of the task is the incoming argument for the next one.
+   * Additionally it makes check of the returned outcome::result<T>.
+   *
+   * ThreadQueueContext allows you to make wrapper around any sync/async subsystem.
+   *
+   * Example:
+   *    tp_ - std::shared_ptr<kagome::thread::ThreadPool>
+   *    c_ - std::shared_ptr<boost::asio::io_context>
+   *
+   *    thread::sequence(
+   *        thread::createTask(tp_, []() -> outcome::result<int> { return 100; }),
+   *        thread::createTask(
+   *            c_, [](auto a) -> outcome::result<float> { return 10.f + a; }),
+   *        thread::createTask(tp_,
+   *                           [](auto b) -> outcome::result<std::string> {
+   *                             static_assert(std::is_floating_point_v<decltype(b)>);
+   *                             return std::to_string(static_cast<int>(b))
+   *                                    + " is the result";
+   *                           }),
+   *        thread::createTask(tp_,
+   *                           [](auto c) { assert(c == "110 is the result"); }));
+   * */
+  // clang-format on
+
+  template <typename T>
+  struct ThreadQueueContext {
+    template <typename D>
+    [[maybe_unused]] ThreadQueueContext(D &&);
+    template <typename F>
+    [[maybe_unused]] void operator()(F &&func);
+  };
+
+  template <>
+  struct ThreadQueueContext<std::weak_ptr<boost::asio::io_context>> {
+    using Type = std::weak_ptr<boost::asio::io_context>;
+    Type t;
+
+    template <typename D>
+    ThreadQueueContext(D &&arg) : t{std::forward<D>(arg)} {}
+
+    template <typename F>
+    void operator()(F &&func) {
+      if (auto call_context = t.lock()) {
+        boost::asio::post(*call_context, std::forward<F>(func));
+      }
+    }
+  };
+
+  template <>
+  struct ThreadQueueContext<std::shared_ptr<boost::asio::io_context>>
+      : ThreadQueueContext<std::weak_ptr<boost::asio::io_context>> {
+    template <typename D>
+    ThreadQueueContext(D &&arg)
+        : ThreadQueueContext<std::weak_ptr<boost::asio::io_context>>(
+            std::forward<D>(arg)) {}
+  };
+
+  template <typename T>
+  auto createThreadQueueContext(T &&t) {
+    return ThreadQueueContext<std::decay_t<T>>{std::forward<T>(t)};
+  }
+
+  template <typename C, typename F>
+  auto createTask(C &&c, F &&f) {
+    return std::make_pair(createThreadQueueContext(std::forward<C>(c)),
+                          std::forward<F>(f));
+  }
+
+  template <typename T, typename K, typename... Args>
+  void sequence(std::pair<T, K> &&t, Args &&...args) {
+    __internal_contextCall(std::move(t.first),
+                           [func{std::move(t.second)},
+                            forwarding_func{__internal_bindArgs(
+                                std::forward<Args>(args)...)}]() mutable {
+                             forwarding_func(func());
+                           });
+  }
+
+  template <typename T, typename K>
+  void sequence(std::pair<T, K> &&t) {
+    __internal_contextCall(std::move(t.first), std::move(t.second));
+  }
+
+  template <typename T, typename F>
+  void __internal_contextCall(ThreadQueueContext<T> &&t, F &&f) {
+    t(std::forward<F>(f));
+  }
+
+  template <typename R, typename T, typename K>
+  void __internal_executeI(R &&r, std::pair<T, K> &&t) {
+    if (std::forward<R>(r).has_value()) {
+      __internal_contextCall(
+          std::move(t.first),
+          [r{std::forward<R>(r).value()}, func{std::move(t.second)}]() mutable {
+            func(std::move(r));
+          });
+    }
+  }
+
+  template <typename R, typename T, typename K, typename... Args>
+  void __internal_executeI(R &&r, std::pair<T, K> &&t, Args &&...args) {
+    if (std::forward<R>(r).has_value()) {
+      __internal_contextCall(std::move(t.first),
+                             [r{std::forward<R>(r).value()},
+                              func{std::move(t.second)},
+                              forwarding_func{__internal_bindArgs(
+                                  std::forward<Args>(args)...)}]() mutable {
+                               forwarding_func(func(std::move(r)));
+                             });
+    }
+  }
+
+  template <typename... Args>
+  auto __internal_bindArgs(Args &&...args) {
+    return std::bind(
+        [](auto &&...args) mutable { __internal_executeI(std::move(args)...); },
+        std::placeholders::_1,
+        std::forward<Args>(args)...);
+  }
+
+}  // namespace kagome::thread
+
+#endif  // KAGOME_TASKS_SEQUENCE_HPP

core/parachain/thread_pool.cpp

@@ -0,0 +1,63 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+#include "parachain/thread_pool.hpp"
+
+namespace kagome::thread {
+
+  ThreadPool::ThreadPool(
+      std::shared_ptr<application::AppStateManager> app_state_manager,
+      size_t thread_count)
+      : thread_count_{thread_count} {
+    BOOST_ASSERT(thread_count_ > 0);
+
+    if (app_state_manager) app_state_manager->takeControl(*this);
+  }
+
+  ThreadPool::ThreadPool(size_t thread_count) : thread_count_{thread_count} {
+    BOOST_ASSERT(thread_count_ > 0);
+  }
+
+  ThreadPool::~ThreadPool() {
+    /// check that all workers are stopped.
+    BOOST_ASSERT(workers_.empty());
+  }
+
+  bool ThreadPool::prepare() {
+    context_ = std::make_shared<WorkersContext>();
+    work_guard_ = std::make_unique<WorkGuard>(context_->get_executor());
+    workers_.reserve(thread_count_);
+    return true;
+  }
+
+  bool ThreadPool::start() {
+    BOOST_ASSERT(context_);
+    BOOST_ASSERT(work_guard_);
+    for (size_t ix = 0; ix < thread_count_; ++ix) {
+      workers_.emplace_back(
+          [wptr{this->weak_from_this()}, context{context_}]() {
+            if (auto self = wptr.lock()) {
+              self->logger_->debug("Started thread worker with id: {}",
+                                   std::this_thread::get_id());
+            }
+            context->run();
+          });
+    }
+    return true;
+  }
+
+  void ThreadPool::stop() {
+    work_guard_.reset();
+    if (context_) {
+      context_->stop();
+    }
+    for (auto &worker : workers_) {
+      if (worker.joinable()) {
+        worker.join();
+      }
+    }
+    workers_.clear();
+  }
+
+}  // namespace kagome::thread

core/parachain/thread_pool.hpp

@@ -0,0 +1,84 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#ifndef KAGOME_THREAD_POOL_HPP
+#define KAGOME_THREAD_POOL_HPP
+
+#include <memory>
+#include <queue>
+#include <thread>
+#include <type_traits>
+
+#include <boost/asio/executor_work_guard.hpp>
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/io_service.hpp>
+#include <boost/asio/post.hpp>
+#include <boost/asio/signal_set.hpp>
+
+#include "application/app_state_manager.hpp"
+#include "parachain/tasks_sequence.hpp"
+
+namespace kagome::thread {
+
+  /**
+   * Thread pool with sequenced task execution on different threads.
+   */
+  struct ThreadPool final : std::enable_shared_from_this<ThreadPool> {
+    using WorkersContext = boost::asio::io_context;
+    using WorkGuard = boost::asio::executor_work_guard<
+        boost::asio::io_context::executor_type>;
+
+    ThreadPool() = delete;
+    ThreadPool(const ThreadPool &) = delete;
+    ThreadPool(ThreadPool &&) = delete;
+    ThreadPool &operator=(const ThreadPool &) = delete;
+    ThreadPool &operator=(ThreadPool &&) = delete;
+
+    ThreadPool(size_t thread_count = 5ull);
+    ThreadPool(std::shared_ptr<application::AppStateManager> app_state_manager,
+               size_t thread_count = 5ull);
+    ~ThreadPool();
+
+    /// AppStateManager Impl
+    bool prepare();
+    bool start();
+    void stop();
+
+    template <typename F>
+    void pushTask(F &&f) {
+      boost::asio::post(*context_, std::forward<F>(f));
+    }
+
+   private:
+    friend struct ThreadQueueContext<ThreadPool>;
+
+    const size_t thread_count_;
+    std::shared_ptr<WorkersContext> context_;
+    std::unique_ptr<WorkGuard> work_guard_;
+    std::vector<std::thread> workers_;
+    log::Logger logger_ = log::createLogger("ThreadPool", "thread");
+  };
+
+  /*
+   * Wrapper for sequenced execution.
+   * */
+  template <>
+  struct ThreadQueueContext<std::shared_ptr<ThreadPool>> {
+    using Type = std::weak_ptr<ThreadPool>;
+    Type t;
+
+    ThreadQueueContext(const std::shared_ptr<ThreadPool> &arg) : t{arg} {}
+
+    template <typename F>
+    void operator()(F &&func) {
+      if (auto tp = t.lock()) {
+        tp->pushTask(std::forward<F>(func));
+      }
+    }
+  };
+
+}  // namespace kagome::thread
+
+#endif  // KAGOME_THREAD_POOL_HPP

core/subscription/subscriber.hpp

@@ -64,9 +64,9 @@ namespace kagome::subscription {
 
    public:
     template <typename... SubscriberConstructorArgs>
-    explicit Subscriber(SubscriptionEnginePtr &ptr,
+    explicit Subscriber(SubscriptionEnginePtr ptr,
                         SubscriberConstructorArgs &&...args)
-        : engine_(ptr),
+        : engine_(std::move(ptr)),
           object_(std::forward<SubscriberConstructorArgs>(args)...) {}
 
     ~Subscriber() {