From b2a672db9d51e8de1b0571f0d56d812a778af5cb Mon Sep 17 00:00:00 2001 From: RickiNano <81099017+RickiNano@users.noreply.github.com> Date: Mon, 11 Mar 2024 13:52:29 +0100 Subject: [PATCH] Multithreaded request aggregator (#4469) * Run aggregator threaded * Configurable number of aggregator threads * Default to max 4 threads * Added unit test for request_aggregator_threads toml * Improve the description of request_aggregator_threads --------- Co-authored-by: Dimitrios Siganos --- nano/core_test/toml.cpp | 7 +++++++ nano/node/nodeconfig.cpp | 5 +++++ nano/node/nodeconfig.hpp | 4 ++++ nano/node/request_aggregator.cpp | 16 ++++++++++++---- nano/node/request_aggregator.hpp | 4 +++- 5 files changed, 31 insertions(+), 5 deletions(-) diff --git a/nano/core_test/toml.cpp b/nano/core_test/toml.cpp index 744e70acd8..c4f973adcf 100644 --- a/nano/core_test/toml.cpp +++ b/nano/core_test/toml.cpp @@ -193,6 +193,8 @@ TEST (toml, daemon_config_deserialize_defaults) ASSERT_EQ (conf.node.work_peers, defaults.node.work_peers); ASSERT_EQ (conf.node.work_threads, defaults.node.work_threads); ASSERT_EQ (conf.node.max_queued_requests, defaults.node.max_queued_requests); + ASSERT_EQ (conf.node.request_aggregator_threads, defaults.node.request_aggregator_threads); + ASSERT_EQ (conf.node.max_unchecked_blocks, defaults.node.max_unchecked_blocks); ASSERT_EQ (conf.node.backlog_scan_batch_size, defaults.node.backlog_scan_batch_size); ASSERT_EQ (conf.node.backlog_scan_frequency, defaults.node.backlog_scan_frequency); @@ -450,6 +452,10 @@ TEST (toml, daemon_config_deserialize_no_defaults) work_threads = 999 max_work_generate_multiplier = 1.0 max_queued_requests = 999 + + request_aggregator_threads = 999 + max_unchecked_blocks = 999 + frontiers_confirmation = "always" backlog_scan_batch_size = 999 backlog_scan_frequency = 999 @@ -639,6 +645,7 @@ TEST (toml, daemon_config_deserialize_no_defaults) ASSERT_NE (conf.node.work_peers, defaults.node.work_peers); ASSERT_NE (conf.node.work_threads, defaults.node.work_threads); ASSERT_NE (conf.node.max_queued_requests, defaults.node.max_queued_requests); + ASSERT_NE (conf.node.request_aggregator_threads, defaults.node.request_aggregator_threads); ASSERT_NE (conf.node.backlog_scan_batch_size, defaults.node.backlog_scan_batch_size); ASSERT_NE (conf.node.backlog_scan_frequency, defaults.node.backlog_scan_frequency); diff --git a/nano/node/nodeconfig.cpp b/nano/node/nodeconfig.cpp index 2453c2c24d..898f487b6e 100644 --- a/nano/node/nodeconfig.cpp +++ b/nano/node/nodeconfig.cpp @@ -129,6 +129,10 @@ nano::error nano::node_config::serialize_toml (nano::tomlconfig & toml) const toml.put ("max_work_generate_multiplier", max_work_generate_multiplier, "Maximum allowed difficulty multiplier for work generation.\ntype:double,[1..]"); toml.put ("frontiers_confirmation", serialize_frontiers_confirmation (frontiers_confirmation), "Mode controlling frontier confirmation rate.\ntype:string,{auto,always,disabled}"); toml.put ("max_queued_requests", max_queued_requests, "Limit for number of queued confirmation requests for one channel, after which new requests are dropped until the queue drops below this value.\ntype:uint32"); + + toml.put ("request_aggregator_threads", request_aggregator_threads, "Number of threads to dedicate to request aggregator. The default value is the minimum of 4 or the number returned by nano::hardware_concurency(), which is the number of hardware threads or the value of the environment variable NANO_HARDWARE_CONCURRENCY."); + toml.put ("max_unchecked_blocks", max_unchecked_blocks, "Maximum number of unchecked blocks to store in memory. Defaults to 65536. \ntype:uint64,[0..]"); + toml.put ("rep_crawler_weight_minimum", rep_crawler_weight_minimum.to_string_dec (), "Rep crawler minimum weight, if this is less than minimum principal weight then this is taken as the minimum weight a rep must have to be tracked. If you want to track all reps set this to 0. If you do not want this to influence anything then set it to max value. This is only useful for debugging or for people who really know what they are doing.\ntype:string,amount,raw"); toml.put ("backlog_scan_batch_size", backlog_scan_batch_size, "Number of accounts per second to process when doing backlog population scan. Increasing this value will help unconfirmed frontiers get into election prioritization queue faster, however it will also increase resource usage. \ntype:uint"); toml.put ("backlog_scan_frequency", backlog_scan_frequency, "Backlog scan divides the scan into smaller batches, number of which is controlled by this value. Higher frequency helps to utilize resources more uniformly, however it also introduces more overhead. The resulting number of accounts per single batch is `backlog_scan_batch_size / backlog_scan_frequency` \ntype:uint"); @@ -425,6 +429,7 @@ nano::error nano::node_config::deserialize_toml (nano::tomlconfig & toml) toml.get ("max_work_generate_multiplier", max_work_generate_multiplier); toml.get ("max_queued_requests", max_queued_requests); + toml.get ("request_aggregator_threads", request_aggregator_threads); auto rep_crawler_weight_minimum_l (rep_crawler_weight_minimum.to_string_dec ()); if (toml.has_key ("rep_crawler_weight_minimum")) diff --git a/nano/node/nodeconfig.hpp b/nano/node/nodeconfig.hpp index 7324193118..0adade97b7 100644 --- a/nano/node/nodeconfig.hpp +++ b/nano/node/nodeconfig.hpp @@ -116,6 +116,10 @@ class node_config bool backup_before_upgrade{ false }; double max_work_generate_multiplier{ 64. }; uint32_t max_queued_requests{ 512 }; + + unsigned request_aggregator_threads{ std::min (nano::hardware_concurrency (), 4u) }; // Max 4 threads if available + unsigned max_unchecked_blocks{ 65536 }; + std::chrono::seconds max_pruning_age{ !network_params.network.is_beta_network () ? std::chrono::seconds (24 * 60 * 60) : std::chrono::seconds (5 * 60) }; // 1 day; 5 minutes for beta network uint64_t max_pruning_depth{ 0 }; nano::rocksdb_config rocksdb_config; diff --git a/nano/node/request_aggregator.cpp b/nano/node/request_aggregator.cpp index 7f6aa2d2e1..0f7a773def 100644 --- a/nano/node/request_aggregator.cpp +++ b/nano/node/request_aggregator.cpp @@ -15,15 +15,20 @@ nano::request_aggregator::request_aggregator (nano::node_config const & config_a max_delay (config_a.network_params.network.is_dev_network () ? 50 : 300), small_delay (config_a.network_params.network.is_dev_network () ? 10 : 50), max_channel_requests (config_a.max_queued_requests), + request_aggregator_threads (config_a.request_aggregator_threads), stats (stats_a), local_votes (history_a), ledger (ledger_a), wallets (wallets_a), active (active_a), generator (generator_a), - final_generator (final_generator_a), - thread ([this] () { run (); }) + final_generator (final_generator_a) { + for (auto i = 0; i < request_aggregator_threads; ++i) + { + threads.emplace_back ([this] () { run (); }); + } + generator.set_reply_action ([this] (std::shared_ptr const & vote_a, std::shared_ptr const & channel_a) { this->reply_action (vote_a, channel_a); }); @@ -132,9 +137,12 @@ void nano::request_aggregator::stop () stopped = true; } condition.notify_all (); - if (thread.joinable ()) + for (auto & thread : threads) { - thread.join (); + if (thread.joinable ()) + { + thread.join (); + } } } diff --git a/nano/node/request_aggregator.hpp b/nano/node/request_aggregator.hpp index 657b7b56f3..f88a6b4faa 100644 --- a/nano/node/request_aggregator.hpp +++ b/nano/node/request_aggregator.hpp @@ -13,6 +13,7 @@ #include #include #include +#include namespace mi = boost::multi_index; @@ -74,6 +75,7 @@ class request_aggregator final std::chrono::milliseconds const max_delay; std::chrono::milliseconds const small_delay; std::size_t const max_channel_requests; + std::size_t const request_aggregator_threads; private: void run (); @@ -105,7 +107,7 @@ class request_aggregator final bool started{ false }; nano::condition_variable condition; nano::mutex mutex{ mutex_identifier (mutexes::request_aggregator) }; - std::thread thread; + std::vector threads; friend std::unique_ptr collect_container_info (request_aggregator &, std::string const &); };