don't share pool handler

core/application/app_state_manager.hpp

@@ -6,7 +6,9 @@
 
 #pragma once
 
-#include "log/logger.hpp"
+#include <functional>
+#include <stdexcept>
+#include <string>
 
 namespace kagome::application {
 

core/blockchain/impl/block_tree_impl.cpp

@@ -126,7 +126,7 @@ namespace kagome::blockchain {
       std::shared_ptr<const class JustificationStoragePolicy>
           justification_storage_policy,
       std::shared_ptr<storage::trie_pruner::TriePruner> state_pruner,
-      std::shared_ptr<common::MainPoolHandler> main_pool_handler) {
+      common::MainThreadPool &main_thread_pool) {
     BOOST_ASSERT(storage != nullptr);
     BOOST_ASSERT(header_repo != nullptr);
 
@@ -285,7 +285,7 @@ namespace kagome::blockchain {
                           std::move(extrinsic_event_key_repo),
                           std::move(justification_storage_policy),
                           state_pruner,
-                          std::move(main_pool_handler)));
+                          main_thread_pool));
 
     // Add non-finalized block to the block tree
     for (auto &e : collected) {
@@ -421,7 +421,7 @@ namespace kagome::blockchain {
       std::shared_ptr<const JustificationStoragePolicy>
           justification_storage_policy,
       std::shared_ptr<storage::trie_pruner::TriePruner> state_pruner,
-      std::shared_ptr<common::MainPoolHandler> main_pool_handler)
+      common::MainThreadPool &main_thread_pool)
       : block_tree_data_{BlockTreeData{
           .header_repo_ = std::move(header_repo),
           .storage_ = std::move(storage),
@@ -435,7 +435,7 @@ namespace kagome::blockchain {
           .genesis_block_hash_ = {},
           .blocks_pruning_ = {app_config.blocksPruning(), finalized.number},
       }},
-        main_pool_handler_(std::move(main_pool_handler)) {
+        main_pool_handler_{main_thread_pool.handlerStarted()} {
     block_tree_data_.sharedAccess([&](const BlockTreeData &p) {
       BOOST_ASSERT(p.header_repo_ != nullptr);
       BOOST_ASSERT(p.storage_ != nullptr);

core/blockchain/impl/block_tree_impl.hpp

@@ -33,14 +33,18 @@
 #include "telemetry/service.hpp"
 #include "utils/safe_object.hpp"
 
+namespace kagome {
+  class PoolHandler;
+}  // namespace kagome
+
 namespace kagome::blockchain {
   struct ReorgAndPrune;
   class TreeNode;
   class CachedTree;
 }  // namespace kagome::blockchain
 
 namespace kagome::common {
-  class MainPoolHandler;
+  class MainThreadPool;
 }
 
 namespace kagome::storage::trie_pruner {
@@ -67,7 +71,7 @@ namespace kagome::blockchain {
         std::shared_ptr<const class JustificationStoragePolicy>
             justification_storage_policy,
         std::shared_ptr<storage::trie_pruner::TriePruner> state_pruner,
-        std::shared_ptr<common::MainPoolHandler> main_pool_handler);
+        common::MainThreadPool &main_thread_pool);
 
     /// Recover block tree state at provided block
     static outcome::result<void> recover(
@@ -199,7 +203,7 @@ namespace kagome::blockchain {
         std::shared_ptr<const class JustificationStoragePolicy>
             justification_storage_policy,
         std::shared_ptr<storage::trie_pruner::TriePruner> state_pruner,
-        std::shared_ptr<common::MainPoolHandler> main_pool_handler);
+        common::MainThreadPool &main_thread_pool);
 
     outcome::result<void> reorgAndPrune(BlockTreeData &p,
                                         const ReorgAndPrune &changes);
@@ -280,7 +284,7 @@ namespace kagome::blockchain {
     metrics::Gauge *metric_best_block_height_;
     metrics::Gauge *metric_finalized_block_height_;
     metrics::Gauge *metric_known_chain_leaves_;
-    std::shared_ptr<common::MainPoolHandler> main_pool_handler_;
+    std::shared_ptr<PoolHandler> main_pool_handler_;
     telemetry::Telemetry telemetry_ = telemetry::createTelemetryService();
   };
 }  // namespace kagome::blockchain

core/common/main_thread_pool.hpp

@@ -22,16 +22,4 @@ namespace kagome::common {
     // Ctor for test purposes
     MainThreadPool(TestThreadPool test) : ThreadPool{test} {}
   };
-
-  class MainPoolHandler final : public PoolHandler {
-   public:
-    MainPoolHandler(
-        std::shared_ptr<application::AppStateManager> app_state_manager,
-        std::shared_ptr<MainThreadPool> thread_pool)
-        : PoolHandler(thread_pool->io_context()) {
-      BOOST_ASSERT(app_state_manager);
-      start();
-      app_state_manager->takeControl(*this);
-    }
-  };
 }  // namespace kagome::common

core/common/worker_thread_pool.hpp

@@ -26,15 +26,4 @@ namespace kagome::common {
     // Ctor for test purposes
     WorkerThreadPool(TestThreadPool test) : ThreadPool{test} {}
   };
-
-  class WorkerPoolHandler final : public PoolHandler {
-   public:
-    WorkerPoolHandler(
-        std::shared_ptr<application::AppStateManager> app_state_manager,
-        std::shared_ptr<WorkerThreadPool> thread_pool)
-        : PoolHandler(thread_pool->io_context()) {
-      BOOST_ASSERT(app_state_manager);
-      app_state_manager->takeControl(*this);
-    }
-  };
 }  // namespace kagome::common

core/consensus/babe/impl/babe.cpp

@@ -69,6 +69,7 @@ namespace {
 namespace kagome::consensus::babe {
 
   Babe::Babe(
+      application::AppStateManager &app_state_manager,
       const application::AppConfiguration &app_config,
       const clock::SystemClock &clock,
       std::shared_ptr<blockchain::BlockTree> block_tree,
@@ -91,8 +92,8 @@ namespace kagome::consensus::babe {
       std::shared_ptr<runtime::OffchainWorkerApi> offchain_worker_api,
       std::shared_ptr<offchain::OffchainWorkerFactory> offchain_worker_factory,
       std::shared_ptr<offchain::OffchainWorkerPool> offchain_worker_pool,
-      std::shared_ptr<common::MainPoolHandler> main_pool_handler,
-      std::shared_ptr<common::WorkerPoolHandler> worker_pool_handler)
+      common::MainThreadPool &main_thread_pool,
+      common::WorkerThreadPool &worker_thread_pool)
       : log_(log::createLogger("Babe", "babe")),
         clock_(clock),
         block_tree_(std::move(block_tree)),
@@ -115,8 +116,8 @@ namespace kagome::consensus::babe {
         offchain_worker_api_(std::move(offchain_worker_api)),
         offchain_worker_factory_(std::move(offchain_worker_factory)),
         offchain_worker_pool_(std::move(offchain_worker_pool)),
-        main_pool_handler_(std::move(main_pool_handler)),
-        worker_pool_handler_(std::move(worker_pool_handler)),
+        main_pool_handler_{main_thread_pool.handler(app_state_manager)},
+        worker_pool_handler_{worker_thread_pool.handler(app_state_manager)},
         is_validator_by_config_(app_config.roles().flags.authority != 0),
         telemetry_{telemetry::createTelemetryService()} {
     BOOST_ASSERT(block_tree_);

core/consensus/babe/impl/babe.hpp

@@ -18,8 +18,13 @@
 #include "primitives/event_types.hpp"
 #include "telemetry/service.hpp"
 
+namespace kagome {
+  class PoolHandler;
+}  // namespace kagome
+
 namespace kagome::application {
   class AppConfiguration;
+  class AppStateManager;
 }  // namespace kagome::application
 
 namespace kagome::authorship {
@@ -31,8 +36,8 @@ namespace kagome::blockchain {
 }
 
 namespace kagome::common {
-  class WorkerPoolHandler;
-  class MainPoolHandler;
+  class WorkerThreadPool;
+  class MainThreadPool;
 }  // namespace kagome::common
 
 namespace kagome::consensus {
@@ -94,6 +99,7 @@ namespace kagome::consensus::babe {
     };
 
     Babe(
+        application::AppStateManager &app_state_manager,
         const application::AppConfiguration &app_config,
         const clock::SystemClock &clock,
         std::shared_ptr<blockchain::BlockTree> block_tree,
@@ -117,8 +123,8 @@ namespace kagome::consensus::babe {
         std::shared_ptr<offchain::OffchainWorkerFactory>
             offchain_worker_factory,
         std::shared_ptr<offchain::OffchainWorkerPool> offchain_worker_pool,
-        std::shared_ptr<common::MainPoolHandler> main_pool_handler,
-        std::shared_ptr<common::WorkerPoolHandler> worker_pool_handler);
+        common::MainThreadPool &main_thread_pool,
+        common::WorkerThreadPool &worker_thread_pool);
 
     bool isGenesisConsensus() const override;
 
@@ -187,8 +193,8 @@ namespace kagome::consensus::babe {
     std::shared_ptr<runtime::OffchainWorkerApi> offchain_worker_api_;
     std::shared_ptr<offchain::OffchainWorkerFactory> offchain_worker_factory_;
     std::shared_ptr<offchain::OffchainWorkerPool> offchain_worker_pool_;
-    std::shared_ptr<common::MainPoolHandler> main_pool_handler_;
-    std::shared_ptr<common::WorkerPoolHandler> worker_pool_handler_;
+    std::shared_ptr<PoolHandler> main_pool_handler_;
+    std::shared_ptr<PoolHandler> worker_pool_handler_;
 
     const bool is_validator_by_config_;
     bool is_active_validator_;

core/consensus/beefy/impl/beefy_impl.cpp

@@ -48,8 +48,8 @@ namespace kagome::network {
       std::shared_ptr<runtime::BeefyApi> beefy_api,
       std::shared_ptr<crypto::EcdsaProvider> ecdsa,
       std::shared_ptr<storage::SpacedStorage> db,
-      std::shared_ptr<common::MainPoolHandler> main_thread_handler,
-      std::shared_ptr<BeefyThreadPool> beefy_thread_pool,
+      common::MainThreadPool &main_thread_pool,
+      BeefyThreadPool &beefy_thread_pool,
       std::shared_ptr<libp2p::basic::Scheduler> scheduler,
       LazySPtr<consensus::Timeline> timeline,
       std::shared_ptr<crypto::SessionKeys> session_keys,
@@ -59,11 +59,8 @@ namespace kagome::network {
         beefy_api_{std::move(beefy_api)},
         ecdsa_{std::move(ecdsa)},
         db_{db->getSpace(storage::Space::kBeefyJustification)},
-        main_pool_handler_(std::move(main_thread_handler)),
-        beefy_pool_handler_{[&] {
-          BOOST_ASSERT(beefy_thread_pool != nullptr);
-          return beefy_thread_pool->handler();
-        }()},
+        main_pool_handler_{main_thread_pool.handler(app_state_manager)},
+        beefy_pool_handler_{beefy_thread_pool.handler(app_state_manager)},
         scheduler_{std::move(scheduler)},
         timeline_{std::move(timeline)},
         session_keys_{std::move(session_keys)},
@@ -249,7 +246,6 @@ namespace kagome::network {
   }
 
   void BeefyImpl::start() {
-    beefy_pool_handler_->start();
     beefy_pool_handler_->execute([weak{weak_from_this()}] {
       if (auto self = weak.lock()) {
         std::ignore = self->update();
@@ -258,10 +254,6 @@ namespace kagome::network {
     setTimer();
   }
 
-  void BeefyImpl::stop() {
-    beefy_pool_handler_->stop();
-  }
-
   bool BeefyImpl::hasJustification(primitives::BlockNumber block) const {
     auto r = db_->contains(BlockNumberKey::encode(block));
     return r and r.value();

core/consensus/beefy/impl/beefy_impl.hpp

@@ -32,8 +32,7 @@ namespace kagome::blockchain {
 }
 
 namespace kagome::common {
-  class MainPoolHandler;
-  class WorkerPoolHandler;
+  class MainThreadPool;
 }  // namespace kagome::common
 
 namespace kagome::consensus {
@@ -66,8 +65,8 @@ namespace kagome::network {
               std::shared_ptr<runtime::BeefyApi> beefy_api,
               std::shared_ptr<crypto::EcdsaProvider> ecdsa,
               std::shared_ptr<storage::SpacedStorage> db,
-              std::shared_ptr<common::MainPoolHandler> main_thread_handler,
-              std::shared_ptr<BeefyThreadPool> beefy_thread_pool,
+              common::MainThreadPool &main_thread_pool,
+              BeefyThreadPool &beefy_thread_pool,
               std::shared_ptr<libp2p::basic::Scheduler> scheduler,
               LazySPtr<consensus::Timeline> timeline,
               std::shared_ptr<crypto::SessionKeys> session_keys,
@@ -76,7 +75,6 @@ namespace kagome::network {
 
     void prepare();
     void start();
-    void stop();
 
     primitives::BlockNumber finalized() const override;
 
@@ -122,7 +120,7 @@ namespace kagome::network {
     std::shared_ptr<runtime::BeefyApi> beefy_api_;
     std::shared_ptr<crypto::EcdsaProvider> ecdsa_;
     std::shared_ptr<storage::BufferStorage> db_;
-    std::shared_ptr<common::MainPoolHandler> main_pool_handler_;
+    std::shared_ptr<PoolHandler> main_pool_handler_;
     std::shared_ptr<PoolHandler> beefy_pool_handler_;
     std::shared_ptr<libp2p::basic::Scheduler> scheduler_;
     LazySPtr<consensus::Timeline> timeline_;

core/consensus/grandpa/impl/environment_impl.cpp

@@ -44,6 +44,7 @@ namespace kagome::consensus::grandpa {
   using primitives::Justification;
 
   EnvironmentImpl::EnvironmentImpl(
+      application::AppStateManager &app_state_manager,
       std::shared_ptr<blockchain::BlockTree> block_tree,
       std::shared_ptr<blockchain::BlockHeaderRepository> header_repository,
       std::shared_ptr<AuthorityManager> authority_manager,
@@ -58,7 +59,7 @@ namespace kagome::consensus::grandpa {
       std::shared_ptr<crypto::Hasher> hasher,
       std::shared_ptr<offchain::OffchainWorkerFactory> offchain_worker_factory,
       std::shared_ptr<offchain::OffchainWorkerPool> offchain_worker_pool,
-      std::shared_ptr<common::MainPoolHandler> main_pool_handler)
+      common::MainThreadPool &main_thread_pool)
       : block_tree_{std::move(block_tree)},
         header_repository_{std::move(header_repository)},
         authority_manager_{std::move(authority_manager)},
@@ -73,7 +74,7 @@ namespace kagome::consensus::grandpa {
         hasher_(std::move(hasher)),
         offchain_worker_factory_(std::move(offchain_worker_factory)),
         offchain_worker_pool_(std::move(offchain_worker_pool)),
-        main_pool_handler_(std::move(main_pool_handler)),
+        main_pool_handler_{main_thread_pool.handler(app_state_manager)},
         logger_{log::createLogger("GrandpaEnvironment", "grandpa")} {
     BOOST_ASSERT(block_tree_ != nullptr);
     BOOST_ASSERT(header_repository_ != nullptr);

core/consensus/grandpa/impl/environment_impl.hpp

@@ -13,13 +13,21 @@
 #include "metrics/metrics.hpp"
 #include "parachain/approval/approved_ancestor.hpp"
 
+namespace kagome {
+  class PoolHandler;
+}  // namespace kagome
+
+namespace kagome::application {
+  class AppStateManager;
+}  // namespace kagome::application
+
 namespace kagome::blockchain {
   class BlockHeaderRepository;
   class BlockTree;
 }  // namespace kagome::blockchain
 
 namespace kagome::common {
-  class MainPoolHandler;
+  class MainThreadPool;
 }
 
 namespace kagome::consensus::grandpa {
@@ -55,6 +63,7 @@ namespace kagome::consensus::grandpa {
                           public std::enable_shared_from_this<EnvironmentImpl> {
    public:
     EnvironmentImpl(
+        application::AppStateManager &app_state_manager,
         std::shared_ptr<blockchain::BlockTree> block_tree,
         std::shared_ptr<blockchain::BlockHeaderRepository> header_repository,
         std::shared_ptr<AuthorityManager> authority_manager,
@@ -71,7 +80,7 @@ namespace kagome::consensus::grandpa {
         std::shared_ptr<offchain::OffchainWorkerFactory>
             offchain_worker_factory,
         std::shared_ptr<offchain::OffchainWorkerPool> offchain_worker_pool,
-        std::shared_ptr<common::MainPoolHandler> main_pool_handler);
+        common::MainThreadPool &main_thread_pool);
 
     ~EnvironmentImpl() override = default;
 
@@ -154,7 +163,7 @@ namespace kagome::consensus::grandpa {
     std::shared_ptr<crypto::Hasher> hasher_;
     std::shared_ptr<offchain::OffchainWorkerFactory> offchain_worker_factory_;
     std::shared_ptr<offchain::OffchainWorkerPool> offchain_worker_pool_;
-    std::shared_ptr<common::MainPoolHandler> main_pool_handler_;
+    std::shared_ptr<PoolHandler> main_pool_handler_;
 
     metrics::RegistryPtr metrics_registry_ = metrics::createRegistry();
     metrics::Gauge *metric_approval_lag_;