Merge pull request #4757 from pwojcikdev/activate-cemented-blocks

Activate cemented blocks

nano/core_test/active_elections.cpp

@@ -1232,8 +1232,8 @@ TEST (active_elections, activate_inactive)
 	ASSERT_TIMELY_EQ (5s, 1, node.stats.count (nano::stat::type::confirmation_observer, nano::stat::detail::active_quorum, nano::stat::dir::out));
 	ASSERT_ALWAYS_EQ (50ms, 0, node.stats.count (nano::stat::type::confirmation_observer, nano::stat::detail::active_conf_height, nano::stat::dir::out));
 
-	// The first block was not active so no activation takes place
+	// Cementing of send should activate open
-	ASSERT_FALSE (node.active.active (open->qualified_root ()) || node.block_confirmed_or_being_confirmed (open->hash ()));
+	ASSERT_TIMELY (5s, node.active.active (open->qualified_root ()));
 }
 
 TEST (active_elections, list_active)
@@ -1308,7 +1308,9 @@ TEST (active_elections, vacancy)
 				.sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
 				.work (*system.work.generate (nano::dev::genesis->hash ()))
 				.build ();
-	node.active.vacancy_update = [&updated] () { updated = true; };
+	node.active.vacancy_updated.add ([&updated] () {
+		updated = true;
+	});
 	ASSERT_EQ (nano::block_status::progress, node.process (send));
 	ASSERT_EQ (1, node.active.vacancy (nano::election_behavior::priority));
 	ASSERT_EQ (0, node.active.size ());

nano/node/active_elections.cpp

@@ -29,21 +29,38 @@ nano::active_elections::active_elections (nano::node & node_a, nano::confirming_
 {
 	count_by_behavior.fill (0); // Zero initialize array
 
+	// Cementing blocks might implicitly confirm dependent elections
 	confirming_set.batch_cemented.add ([this] (auto const & cemented) {
-		auto transaction = node.ledger.tx_begin_read ();
+		std::deque<block_cemented_result> results;
+		{
+			// Process all cemented blocks while holding the lock to avoid races where an election for a block that is already cemented is inserted
+			nano::lock_guard<nano::mutex> guard{ mutex };
 			for (auto const & [block, confirmation_root, source_election] : cemented)
+			{
+				auto result = block_cemented (block, confirmation_root, source_election);
+				results.push_back (result);
+			}
+		}
+		{
+			// TODO: This could be offloaded to a separate notification worker, profiling is needed
+			auto transaction = node.ledger.tx_begin_read ();
+			for (auto const & [status, votes] : results)
 			{
 				transaction.refresh_if_needed ();
-			block_cemented (transaction, block, confirmation_root, source_election);
+				notify_observers (transaction, status, votes);
+			}
 		}
 	});
 
 	// Notify elections about alternative (forked) blocks
-	block_processor.block_processed.add ([this] (auto const & result, auto const & context) {
+	block_processor.batch_processed.add ([this] (auto const & batch) {
+		for (auto const & [result, context] : batch)
+		{
 			if (result == nano::block_status::fork)
 			{
 				publish (context.block);
 			}
+		}
 	});
 }
 
@@ -79,16 +96,17 @@ void nano::active_elections::stop ()
 	clear ();
 }
 
-void nano::active_elections::block_cemented (nano::secure::transaction const & transaction, std::shared_ptr<nano::block> const & block, nano::block_hash const & confirmation_root, std::shared_ptr<nano::election> const & source_election)
+auto nano::active_elections::block_cemented (std::shared_ptr<nano::block> const & block, nano::block_hash const & confirmation_root, std::shared_ptr<nano::election> const & source_election) -> block_cemented_result
 {
+	debug_assert (!mutex.try_lock ());
 	debug_assert (node.block_confirmed (block->hash ()));
 
 	// Dependent elections are implicitly confirmed when their block is cemented
-	auto dependend_election = election (block->qualified_root ());
+	auto dependend_election = election_impl (block->qualified_root ());
 	if (dependend_election)
 	{
 		node.stats.inc (nano::stat::type::active_elections, nano::stat::detail::confirm_dependent);
-		dependend_election->try_confirm (block->hash ());
+		dependend_election->try_confirm (block->hash ()); // TODO: This should either confirm or cancel the election
 	}
 
 	nano::election_status status;
@@ -122,16 +140,7 @@ void nano::active_elections::block_cemented (nano::secure::transaction const & t
 	nano::log::arg{ "confirmation_root", confirmation_root },
 	nano::log::arg{ "source_election", source_election });
 
-	notify_observers (transaction, status, votes);
+	return { status, votes };
-
-	bool cemented_bootstrap_count_reached = node.ledger.cemented_count () >= node.ledger.bootstrap_weight_max_blocks;
-	bool was_active = status.type == nano::election_status_type::active_confirmed_quorum || status.type == nano::election_status_type::active_confirmation_height;
-
-	// Next-block activations are only done for blocks with previously active elections
-	if (cemented_bootstrap_count_reached && was_active && !node.flags.disable_activate_successors)
-	{
-		activate_successors (transaction, block);
-	}
 }
 
 void nano::active_elections::notify_observers (nano::secure::transaction const & transaction, nano::election_status const & status, std::vector<nano::vote_with_weight_info> const & votes) const
@@ -169,17 +178,6 @@ void nano::active_elections::notify_observers (nano::secure::transaction const &
 	}
 }
 
-void nano::active_elections::activate_successors (nano::secure::transaction const & transaction, std::shared_ptr<nano::block> const & block)
-{
-	node.scheduler.priority.activate (transaction, block->account ());
-
-	// Start or vote for the next unconfirmed block in the destination account
-	if (block->is_send () && !block->destination ().is_zero () && block->destination () != block->account ())
-	{
-		node.scheduler.priority.activate (transaction, block->destination ());
-	}
-}
-
 int64_t nano::active_elections::limit (nano::election_behavior behavior) const
 {
 	switch (behavior)
@@ -312,7 +310,7 @@ void nano::active_elections::cleanup_election (nano::unique_lock<nano::mutex> &
 		entry.erased_callback (election);
 	}
 
-	vacancy_update ();
+	vacancy_updated.notify ();
 
 	for (auto const & [hash, block] : blocks_l)
 	{
@@ -435,7 +433,7 @@ nano::election_insertion_result nano::active_elections::insert (std::shared_ptr<
 
 		node.vote_cache_processor.trigger (hash);
 		node.observers.active_started.notify (hash);
-		vacancy_update ();
+		vacancy_updated.notify ();
 	}
 
 	// Votes are generated for inserted or ongoing elections
@@ -459,11 +457,17 @@ bool nano::active_elections::active (nano::block const & block_a) const
 	return roots.get<tag_root> ().find (block_a.qualified_root ()) != roots.get<tag_root> ().end ();
 }
 
-std::shared_ptr<nano::election> nano::active_elections::election (nano::qualified_root const & root_a) const
+std::shared_ptr<nano::election> nano::active_elections::election (nano::qualified_root const & root) const
 {
-	std::shared_ptr<nano::election> result;
 	nano::lock_guard<nano::mutex> lock{ mutex };
-	auto existing = roots.get<tag_root> ().find (root_a);
+	return election_impl (root);
+}
+
+std::shared_ptr<nano::election> nano::active_elections::election_impl (nano::qualified_root const & root) const
+{
+	debug_assert (!mutex.try_lock ());
+	std::shared_ptr<nano::election> result;
+	auto existing = roots.get<tag_root> ().find (root);
 	if (existing != roots.get<tag_root> ().end ())
 	{
 		result = existing->election;
@@ -541,8 +545,7 @@ void nano::active_elections::clear ()
 		nano::lock_guard<nano::mutex> guard{ mutex };
 		roots.clear ();
 	}
-
+	vacancy_updated.notify ();
-	vacancy_update ();
 }
 
 nano::container_info nano::active_elections::container_info () const

nano/node/active_elections.hpp

@@ -2,6 +2,7 @@
 
 #include <nano/lib/enum_util.hpp>
 #include <nano/lib/numbers.hpp>
+#include <nano/lib/observer_set.hpp>
 #include <nano/node/election_behavior.hpp>
 #include <nano/node/election_insertion_result.hpp>
 #include <nano/node/election_status.hpp>
@@ -104,7 +105,7 @@ public:
 	bool active (nano::qualified_root const &) const;
 	std::shared_ptr<nano::election> election (nano::qualified_root const &) const;
 	// Returns a list of elections sorted by difficulty
-	std::vector<std::shared_ptr<nano::election>> list_active (std::size_t = std::numeric_limits<std::size_t>::max ());
+	std::vector<std::shared_ptr<nano::election>> list_active (std::size_t max_count = std::numeric_limits<std::size_t>::max ());
 	bool erase (nano::block const &);
 	bool erase (nano::qualified_root const &);
 	bool empty () const;
@@ -121,21 +122,24 @@ public:
 	 * How many election slots are available for specified election type
 	 */
 	int64_t vacancy (nano::election_behavior behavior) const;
-	std::function<void ()> vacancy_update{ [] () {} };
 
 	nano::container_info container_info () const;
 
+public: // Events
+	nano::observer_set<> vacancy_updated;
+
 private:
 	void request_loop ();
 	void request_confirm (nano::unique_lock<nano::mutex> &);
 	// Erase all blocks from active and, if not confirmed, clear digests from network filters
 	void cleanup_election (nano::unique_lock<nano::mutex> & lock_a, std::shared_ptr<nano::election>);
-	nano::stat::type completion_type (nano::election const & election) const;
+
-	// Returns a list of elections sorted by difficulty, mutex must be locked
+	using block_cemented_result = std::pair<nano::election_status, std::vector<nano::vote_with_weight_info>>;
-	std::vector<std::shared_ptr<nano::election>> list_active_impl (std::size_t) const;
+	block_cemented_result block_cemented (std::shared_ptr<nano::block> const & block, nano::block_hash const & confirmation_root, std::shared_ptr<nano::election> const & source_election);
-	void activate_successors (nano::secure::transaction const &, std::shared_ptr<nano::block> const & block);
 	void notify_observers (nano::secure::transaction const &, nano::election_status const & status, std::vector<nano::vote_with_weight_info> const & votes) const;
-	void block_cemented (nano::secure::transaction const &, std::shared_ptr<nano::block> const & block, nano::block_hash const & confirmation_root, std::shared_ptr<nano::election> const & source_election);
+
+	std::shared_ptr<nano::election> election_impl (nano::qualified_root const &) const;
+	std::vector<std::shared_ptr<nano::election>> list_active_impl (std::size_t max_count) const;
 
 private: // Dependencies
 	active_elections_config const & config;

nano/node/node.cpp

@@ -137,7 +137,7 @@ nano::node::node (std::shared_ptr<boost::asio::io_context> io_ctx_a, std::filesy
 	generator{ *generator_impl },
 	final_generator_impl{ std::make_unique<nano::vote_generator> (config, *this, ledger, wallets, vote_processor, history, network, stats, logger, /* final */ true) },
 	final_generator{ *final_generator_impl },
-	scheduler_impl{ std::make_unique<nano::scheduler::component> (*this) },
+	scheduler_impl{ std::make_unique<nano::scheduler::component> (config, *this, ledger, block_processor, active, online_reps, vote_cache, confirming_set, stats, logger) },
 	scheduler{ *scheduler_impl },
 	aggregator_impl{ std::make_unique<nano::request_aggregator> (config.request_aggregator, *this, stats, generator, final_generator, history, ledger, wallets, vote_router) },
 	aggregator{ *aggregator_impl },
@@ -187,13 +187,6 @@ nano::node::node (std::shared_ptr<boost::asio::io_context> io_ctx_a, std::filesy
 
 	if (!init_error ())
 	{
-		// Notify election schedulers when AEC frees election slot
-		active.vacancy_update = [this] () {
-			scheduler.priority.notify ();
-			scheduler.hinted.notify ();
-			scheduler.optimistic.notify ();
-		};
-
 		wallets.observer = [this] (bool active) {
 			observers.wallet.notify (active);
 		};

nano/node/process_live_dispatcher.cpp

@@ -43,12 +43,6 @@ void nano::process_live_dispatcher::inspect (nano::block_status const & result,
 
 void nano::process_live_dispatcher::process_live (nano::block const & block, secure::transaction const & transaction)
 {
-	// Start collecting quorum on block
-	if (ledger.dependents_confirmed (transaction, block))
-	{
-		scheduler.activate (transaction, block.account ());
-	}
-
 	if (websocket.server && websocket.server->any_subscriber (nano::websocket::topic::new_unconfirmed_block))
 	{
 		websocket.server->broadcast (nano::websocket::message_builder ().new_block_arrived (block));

nano/node/scheduler/component.cpp

@@ -5,16 +5,22 @@
 #include <nano/node/scheduler/optimistic.hpp>
 #include <nano/node/scheduler/priority.hpp>
 
-nano::scheduler::component::component (nano::node & node) :
+nano::scheduler::component::component (nano::node_config & node_config, nano::node & node, nano::ledger & ledger, nano::block_processor & block_processor, nano::active_elections & active, nano::online_reps & online_reps, nano::vote_cache & vote_cache, nano::confirming_set & confirming_set, nano::stats & stats, nano::logger & logger) :
-	hinted_impl{ std::make_unique<nano::scheduler::hinted> (node.config.hinted_scheduler, node, node.vote_cache, node.active, node.online_reps, node.stats) },
+	hinted_impl{ std::make_unique<nano::scheduler::hinted> (node_config.hinted_scheduler, node, vote_cache, active, online_reps, stats) },
 	manual_impl{ std::make_unique<nano::scheduler::manual> (node) },
-	optimistic_impl{ std::make_unique<nano::scheduler::optimistic> (node.config.optimistic_scheduler, node, node.ledger, node.active, node.network_params.network, node.stats) },
+	optimistic_impl{ std::make_unique<nano::scheduler::optimistic> (node_config.optimistic_scheduler, node, ledger, active, node_config.network_params.network, stats) },
-	priority_impl{ std::make_unique<nano::scheduler::priority> (node, node.stats) },
+	priority_impl{ std::make_unique<nano::scheduler::priority> (node_config, node, ledger, block_processor, active, confirming_set, stats, logger) },
 	hinted{ *hinted_impl },
 	manual{ *manual_impl },
 	optimistic{ *optimistic_impl },
 	priority{ *priority_impl }
 {
+	// Notify election schedulers when AEC frees election slot
+	active.vacancy_updated.add ([this] () {
+		priority.notify ();
+		hinted.notify ();
+		optimistic.notify ();
+	});
 }
 
 nano::scheduler::component::~component ()

nano/node/scheduler/component.hpp

@@ -10,12 +10,10 @@ namespace nano::scheduler
 class component final
 {
 public:
-	explicit component (nano::node & node);
+	component (nano::node_config &, nano::node &, nano::ledger &, nano::block_processor &, nano::active_elections &, nano::online_reps &, nano::vote_cache &, nano::confirming_set &, nano::stats &, nano::logger &);
 	~component ();
 
-	// Starts all schedulers
 	void start ();
-	// Stops all schedulers
 	void stop ();
 
 	nano::container_info container_info () const;

nano/node/scheduler/priority.cpp

@@ -7,10 +7,15 @@
 #include <nano/secure/ledger_set_any.hpp>
 #include <nano/secure/ledger_set_confirmed.hpp>
 
-nano::scheduler::priority::priority (nano::node & node_a, nano::stats & stats_a) :
+nano::scheduler::priority::priority (nano::node_config & node_config, nano::node & node_a, nano::ledger & ledger_a, nano::block_processor & block_processor_a, nano::active_elections & active_a, nano::confirming_set & confirming_set_a, nano::stats & stats_a, nano::logger & logger_a) :
-	config{ node_a.config.priority_scheduler },
+	config{ node_config.priority_scheduler },
 	node{ node_a },
-	stats{ stats_a }
+	ledger{ ledger_a },
+	block_processor{ block_processor_a },
+	active{ active_a },
+	confirming_set{ confirming_set_a },
+	stats{ stats_a },
+	logger{ logger_a }
 {
 	std::vector<nano::uint128_t> minimums;
 
@@ -34,13 +39,41 @@ nano::scheduler::priority::priority (nano::node & node_a, nano::stats & stats_a)
 	build_region (uint128_t{ 1 } << 116, uint128_t{ 1 } << 120, 2);
 	minimums.push_back (uint128_t{ 1 } << 120);
 
-	node.logger.debug (nano::log::type::election_scheduler, "Number of buckets: {}", minimums.size ());
+	logger.debug (nano::log::type::election_scheduler, "Number of buckets: {}", minimums.size ());
 
 	for (size_t i = 0u, n = minimums.size (); i < n; ++i)
 	{
-		auto bucket = std::make_unique<scheduler::bucket> (minimums[i], node.config.priority_bucket, node.active, stats);
+		auto bucket = std::make_unique<scheduler::bucket> (minimums[i], node_config.priority_bucket, active, stats);
 		buckets.emplace_back (std::move (bucket));
 	}
+
+	// Activate accounts with fresh blocks
+	block_processor.batch_processed.add ([this] (auto const & batch) {
+		auto transaction = ledger.tx_begin_read ();
+		for (auto const & [result, context] : batch)
+		{
+			if (result == nano::block_status::progress)
+			{
+				release_assert (context.block != nullptr);
+				activate (transaction, context.block->account ());
+			}
+		}
+	});
+
+	// Activate successors of cemented blocks
+	confirming_set.batch_cemented.add ([this] (auto const & batch) {
+		if (node.flags.disable_activate_successors)
+		{
+			return;
+		}
+
+		auto transaction = ledger.tx_begin_read ();
+		for (auto const & context : batch)
+		{
+			release_assert (context.block != nullptr);
+			activate_successors (transaction, *context.block);
+		}
+	});
 }
 
 nano::scheduler::priority::~priority ()
@@ -85,11 +118,10 @@ void nano::scheduler::priority::stop ()
 bool nano::scheduler::priority::activate (secure::transaction const & transaction, nano::account const & account)
 {
 	debug_assert (!account.is_zero ());
-	auto info = node.ledger.any.account_get (transaction, account);
+	if (auto info = ledger.any.account_get (transaction, account))
-	if (info)
 	{
 		nano::confirmation_height_info conf_info;
-		node.store.confirmation_height.get (transaction, account, conf_info);
+		ledger.store.confirmation_height.get (transaction, account, conf_info);
 		if (conf_info.height < info->block_count)
 		{
 			return activate (transaction, account, *info, conf_info);
@@ -103,14 +135,14 @@ bool nano::scheduler::priority::activate (secure::transaction const & transactio
 {
 	debug_assert (conf_info.frontier != account_info.head);
 
-	auto hash = conf_info.height == 0 ? account_info.open_block : node.ledger.any.block_successor (transaction, conf_info.frontier).value ();
+	auto hash = conf_info.height == 0 ? account_info.open_block : ledger.any.block_successor (transaction, conf_info.frontier).value ();
-	auto block = node.ledger.any.block_get (transaction, hash);
+	auto block = ledger.any.block_get (transaction, hash);
 	release_assert (block != nullptr);
 
-	if (node.ledger.dependents_confirmed (transaction, *block))
+	if (ledger.dependents_confirmed (transaction, *block))
 	{
 		auto const balance = block->balance ();
-		auto const previous_balance = node.ledger.any.block_balance (transaction, conf_info.frontier).value_or (0);
+		auto const previous_balance = ledger.any.block_balance (transaction, conf_info.frontier).value_or (0);
 		auto const balance_priority = std::max (balance, previous_balance);
 
 		bool added = false;
@@ -120,8 +152,8 @@ bool nano::scheduler::priority::activate (secure::transaction const & transactio
 		}
 		if (added)
 		{
-			node.stats.inc (nano::stat::type::election_scheduler, nano::stat::detail::activated);
+			stats.inc (nano::stat::type::election_scheduler, nano::stat::detail::activated);
-			node.logger.trace (nano::log::type::election_scheduler, nano::log::detail::block_activated,
+			logger.trace (nano::log::type::election_scheduler, nano::log::detail::block_activated,
 			nano::log::arg{ "account", account.to_account () }, // TODO: Convert to lazy eval
 			nano::log::arg{ "block", block },
 			nano::log::arg{ "time", account_info.modified },
@@ -131,7 +163,7 @@ bool nano::scheduler::priority::activate (secure::transaction const & transactio
 		}
 		else
 		{
-			node.stats.inc (nano::stat::type::election_scheduler, nano::stat::detail::activate_full);
+			stats.inc (nano::stat::type::election_scheduler, nano::stat::detail::activate_full);
 		}
 
 		return true; // Activated
@@ -141,6 +173,17 @@ bool nano::scheduler::priority::activate (secure::transaction const & transactio
 	return false; // Not activated
 }
 
+bool nano::scheduler::priority::activate_successors (secure::transaction const & transaction, nano::block const & block)
+{
+	bool result = activate (transaction, block.account ());
+	// Start or vote for the next unconfirmed block in the destination account
+	if (block.is_send () && !block.destination ().is_zero () && block.destination () != block.account ())
+	{
+		result |= activate (transaction, block.destination ());
+	}
+	return result;
+}
+
 void nano::scheduler::priority::notify ()
 {
 	condition.notify_all ();

nano/node/scheduler/priority.hpp

@@ -26,7 +26,7 @@ public:
 class priority final
 {
 public:
-	priority (nano::node &, nano::stats &);
+	priority (nano::node_config &, nano::node &, nano::ledger &, nano::block_processor &, nano::active_elections &, nano::confirming_set &, nano::stats &, nano::logger &);
 	~priority ();
 
 	void start ();
@@ -36,8 +36,9 @@ public:
 	 * Activates the first unconfirmed block of \p account_a
 	 * @return true if account was activated
 	 */
-	bool activate (secure::transaction const &, nano::account const &);
+	bool activate (nano::secure::transaction const &, nano::account const &);
-	bool activate (secure::transaction const &, nano::account const &, nano::account_info const &, nano::confirmation_height_info const &);
+	bool activate (nano::secure::transaction const &, nano::account const &, nano::account_info const &, nano::confirmation_height_info const &);
+	bool activate_successors (nano::secure::transaction const &, nano::block const &);
 
 	void notify ();
 	std::size_t size () const;
@@ -48,7 +49,12 @@ public:
 private: // Dependencies
 	priority_config const & config;
 	nano::node & node;
+	nano::ledger & ledger;
+	nano::block_processor & block_processor;
+	nano::active_elections & active;
+	nano::confirming_set & confirming_set;
 	nano::stats & stats;
+	nano::logger & logger;
 
 private:
 	void run ();