Refactoring thread pool

core/parachain/validator/impl/parachain_processor.cpp

@@ -98,7 +98,8 @@ namespace kagome::parachain {
         parachain_host_(std::move(parachain_host)),
         app_config_(app_config),
         babe_status_observable_(std::move(babe_status_observable)),
-        query_audi_{std::move(query_audi)} {
+        query_audi_{std::move(query_audi)},
+        thread_handler_{thread_pool_->handler()} {
     BOOST_ASSERT(pm_);
     BOOST_ASSERT(peer_view_);
     BOOST_ASSERT(crypto_provider_);
@@ -253,6 +254,11 @@ namespace kagome::parachain {
     return outcome::success();
   }
 
+  bool ParachainProcessorImpl::start() {
+    thread_handler_->start();
+    return true;
+  }
+
   outcome::result<kagome::parachain::ParachainProcessorImpl::RelayParentState>
   ParachainProcessorImpl::initNewBackingTask(
       const primitives::BlockHash &relay_parent) {
@@ -605,7 +611,7 @@ namespace kagome::parachain {
         peer_id);
 
     sequenceIgnore(
-        thread_pool_->io_context()->wrap(
+        thread_handler_->io_context()->wrap(
             asAsync([wself{weak_from_this()},
                      candidate{std::move(candidate)},
                      pov{std::move(pov)},

core/parachain/validator/parachain_processor.hpp

@@ -428,6 +428,7 @@ namespace kagome::parachain {
     std::shared_ptr<authority_discovery::Query> query_audi_;
 
     std::shared_ptr<primitives::events::ChainEventSubscriber> chain_sub_;
+    std::shared_ptr<ThreadHandler> thread_handler_;
   };
 
 }  // namespace kagome::parachain

core/utils/thread_pool.hpp

@@ -6,6 +6,7 @@
 #ifndef KAGOME_UTILS_THREAD_POOL_HPP
 #define KAGOME_UTILS_THREAD_POOL_HPP
 
+#include <atomic>
 #include <boost/asio/executor_work_guard.hpp>
 #include <boost/asio/io_context.hpp>
 #include <memory>
@@ -14,40 +15,151 @@
 #include "utils/non_copyable.hpp"
 
 namespace kagome {
-  // TODO(turuslan): use `ThreadPool` in `RpcThreadPool` and `RpcContext`
+
+  class ThreadHandler final {
+    enum struct State : uint32_t { kStopped = 0, kStarted };
+
+   public:
+    ThreadHandler(ThreadHandler &&) = delete;
+    ThreadHandler(ThreadHandler const &) = delete;
+
+    ThreadHandler &operator=(ThreadHandler &&) = delete;
+    ThreadHandler &operator=(ThreadHandler const &) = delete;
+
+    explicit ThreadHandler(std::shared_ptr<boost::asio::io_context> io_context)
+        : execution_state_{State::kStopped}, ioc_{std::move(io_context)} {}
+    ~ThreadHandler() = default;
+
+    void start() {
+      execution_state_.store(State::kStarted, std::memory_order_release);
+    }
+
+    void stop() {
+      execution_state_.store(State::kStopped, std::memory_order_release);
+    }
+
+    template <typename F>
+    void execute(F &&func) {
+      BOOST_ASSERT(ioc_);
+      if (State::kStarted == execution_state_.load(std::memory_order_acquire)) {
+        ioc_->post(std::forward<F>(func));
+      }
+    }
+
+    template <typename F, typename... Args>
+    auto reinvoke(F &&func, Args &&...args)
+        -> std::optional<std::tuple<Args...>> {
+      if (!isInCurrentThread()) {
+        execute([func(std::forward<F>(func)),
+                 tup{std::tuple<std::decay_t<Args>...>{
+                     std::forward<Args>(args)...}}]() mutable {
+          std::apply([&](auto &&...a) mutable { func(std::move(a)...); },
+                     std::move(tup));
+        });
+        return std::nullopt;
+      }
+      return {std::make_tuple(std::forward<Args>(args)...)};
+    }
+
+    bool isInCurrentThread() const {
+      BOOST_ASSERT(ioc_);
+      return ioc_->get_executor().running_in_this_thread();
+    }
+
+    std::shared_ptr<boost::asio::io_context> io_context() const {
+      return ioc_;
+    }
+
+   private:
+    std::atomic<State> execution_state_;
+    std::shared_ptr<boost::asio::io_context> ioc_;
+  };
 
   /**
    * Creates `io_context` and runs it on `thread_count` threads.
    */
-  class ThreadPool final : NonCopyable, NonMovable {
+  class ThreadPool final {
+    enum struct State : uint32_t { kStopped = 0, kStarted };
+
    public:
+    ThreadPool(ThreadPool &&) = delete;
+    ThreadPool(ThreadPool const &) = delete;
+
+    ThreadPool &operator=(ThreadPool &&) = delete;
+    ThreadPool &operator=(ThreadPool const &) = delete;
+
     explicit ThreadPool(size_t thread_count)
-        : io_context_{std::make_shared<boost::asio::io_context>()},
-          work_guard_{io_context_->get_executor()} {
+        : ioc_{std::make_shared<boost::asio::io_context>()},
+          work_guard_{ioc_->get_executor()} {
+      BOOST_ASSERT(ioc_);
       BOOST_ASSERT(thread_count > 0);
+
       threads_.reserve(thread_count);
       for (size_t i = 0; i < thread_count; ++i) {
-        threads_.emplace_back([io{io_context_}] { io->run(); });
+        threads_.emplace_back([io{ioc_}] { io->run(); });
       }
     }
 
     ~ThreadPool() {
-      io_context_->stop();
+      ioc_->stop();
       for (auto &thread : threads_) {
         thread.join();
       }
     }
 
-    const std::shared_ptr<boost::asio::io_context> &io_context() const {
-      return io_context_;
+    std::shared_ptr<ThreadHandler> handler() {
+      BOOST_ASSERT(ioc_);
+      return std::make_shared<ThreadHandler>(ioc_);
     }
 
    private:
-    std::shared_ptr<boost::asio::io_context> io_context_;
-    boost::asio::executor_work_guard<boost::asio::io_context::executor_type>
+    std::shared_ptr<boost::asio::io_context> ioc_;
+    std::optional<boost::asio::executor_work_guard<
+        boost::asio::io_context::executor_type>>
         work_guard_;
     std::vector<std::thread> threads_;
   };
 }  // namespace kagome
 
+#define REINVOKE_3(ctx, func, in_1, in_2, in_3, out_1, out_2, out_3) \
+  auto res##func = (ctx).reinvoke(                                   \
+      [wself(weak_from_this())](auto &&x1, auto &&x2, auto &&x3) {   \
+        if (auto self = wself.lock()) {                              \
+          self->func(std::move(x1), std::move(x2), std::move(x3));   \
+        }                                                            \
+      },                                                             \
+      std::move(in_1),                                               \
+      std::move(in_2),                                               \
+      std::move(in_3));                                              \
+  if (!(res##func)) return;                                          \
+  auto &&[func##x1, func##x2, func##x3] = std::move(*(res##func));   \
+  auto && (out_1) = func##x1;                                        \
+  auto && (out_2) = func##x2;                                        \
+  auto && (out_3) = func##x3;
+
+#define REINVOKE_2(ctx, func, in_1, in_2, out_1, out_2)  \
+  auto res##func = (ctx).reinvoke(                       \
+      [wself(weak_from_this())](auto &&x1, auto &&x2) {  \
+        if (auto self = wself.lock()) {                  \
+          self->func(std::move(x1), std::move(x2));      \
+        }                                                \
+      },                                                 \
+      std::move(in_1),                                   \
+      std::move(in_2));                                  \
+  if (!(res##func)) return;                              \
+  auto &&[func##x1, func##x2] = std::move(*(res##func)); \
+  auto && (out_1) = func##x1;                            \
+  auto && (out_2) = func##x2;
+
+#define REINVOKE_1(ctx, func, in, out)      \
+  auto res##func = (ctx).reinvoke(          \
+      [wself(weak_from_this())](auto &&x) { \
+        if (auto self = wself.lock()) {     \
+          self->func(std::move(x));         \
+        }                                   \
+      },                                    \
+      std::move(in));                       \
+  if (!(res##func)) return;                 \
+  auto && (out) = std::get<0>(std::move(*(res##func)));
+
 #endif  // KAGOME_UTILS_THREAD_POOL_HPP