This change creates an scheduler::limiter class which tracks and limits how many elections can be spawned.

nano/core_test/CMakeLists.txt

@@ -40,6 +40,7 @@ add_executable(
   peer_container.cpp
   scheduler_buckets.cpp
   request_aggregator.cpp
+  scheduler_limiter.cpp
   signal_manager.cpp
   signing.cpp
   socket.cpp

nano/core_test/election.cpp

@@ -17,6 +17,23 @@ TEST (election, construction)
 	node, nano::dev::genesis, [] (auto const &) {}, [] (auto const &) {}, nano::election_behavior::normal);
 }
 
+// This tests the election destruction event notification
+// Since the notification is signalled in the destructor, it needs to be freed
+TEST (election, destructor_observer)
+{
+	std::atomic<bool> destroyed{ false };
+	{
+		nano::test::system system (1);
+		auto & node = *system.nodes[0];
+		auto election = std::make_shared<nano::election> (
+		node, nano::dev::genesis, [] (auto const &) {}, [] (auto const &) {}, nano::election_behavior::normal);
+		election->destructor_observers.add ([&destroyed] (auto const & qualified_root) {
+			destroyed = true;
+		});
+	}
+	ASSERT_TRUE (destroyed);
+}
+
 TEST (election, behavior)
 {
 	nano::test::system system (1);

nano/core_test/scheduler_buckets.cpp

@@ -1,5 +1,6 @@
 #include <nano/node/scheduler/buckets.hpp>
 #include <nano/secure/common.hpp>
+#include <nano/test_common/testutil.hpp>
 
 #include <gtest/gtest.h>
 
@@ -108,35 +109,35 @@ std::shared_ptr<nano::state_block> & block3 ()
 
 TEST (buckets, construction)
 {
-	nano::scheduler::buckets buckets;
+	nano::scheduler::buckets buckets{ nano::test::active_transactions_insert_null };
 	ASSERT_EQ (0, buckets.size ());
-	ASSERT_TRUE (buckets.empty ());
+	ASSERT_TRUE (buckets.empty () && !buckets.available ()); // Initial state
 	ASSERT_EQ (62, buckets.bucket_count ());
 }
 
 TEST (buckets, index_min)
 {
-	nano::scheduler::buckets buckets;
+	nano::scheduler::buckets buckets{ nano::test::active_transactions_insert_null };
 	ASSERT_EQ (0, buckets.index (std::numeric_limits<nano::uint128_t>::min ()));
 }
 
 TEST (buckets, index_max)
 {
-	nano::scheduler::buckets buckets;
+	nano::scheduler::buckets buckets{ nano::test::active_transactions_insert_null };
 	ASSERT_EQ (buckets.bucket_count () - 1, buckets.index (std::numeric_limits<nano::uint128_t>::max ()));
 }
 
 TEST (buckets, insert_Gxrb)
 {
-	nano::scheduler::buckets buckets;
+	nano::scheduler::buckets buckets{ nano::test::active_transactions_insert_null };
 	buckets.push (1000, block0 (), nano::Gxrb_ratio);
 	ASSERT_EQ (1, buckets.size ());
 	ASSERT_EQ (1, buckets.bucket_size (48));
 }
 
 TEST (buckets, insert_Mxrb)
 {
-	nano::scheduler::buckets buckets;
+	nano::scheduler::buckets buckets{ nano::test::active_transactions_insert_null };
 	buckets.push (1000, block1 (), nano::Mxrb_ratio);
 	ASSERT_EQ (1, buckets.size ());
 	ASSERT_EQ (1, buckets.bucket_size (13));
@@ -145,7 +146,7 @@ TEST (buckets, insert_Mxrb)
 // Test two blocks with the same priority
 TEST (buckets, insert_same_priority)
 {
-	nano::scheduler::buckets buckets;
+	nano::scheduler::buckets buckets{ nano::test::active_transactions_insert_null };
 	buckets.push (1000, block0 (), nano::Gxrb_ratio);
 	buckets.push (1000, block2 (), nano::Gxrb_ratio);
 	ASSERT_EQ (2, buckets.size ());
@@ -155,7 +156,7 @@ TEST (buckets, insert_same_priority)
 // Test the same block inserted multiple times
 TEST (buckets, insert_duplicate)
 {
-	nano::scheduler::buckets buckets;
+	nano::scheduler::buckets buckets{ nano::test::active_transactions_insert_null };
 	buckets.push (1000, block0 (), nano::Gxrb_ratio);
 	buckets.push (1000, block0 (), nano::Gxrb_ratio);
 	ASSERT_EQ (1, buckets.size ());
@@ -164,18 +165,18 @@ TEST (buckets, insert_duplicate)
 
 TEST (buckets, insert_older)
 {
-	nano::scheduler::buckets buckets;
+	nano::scheduler::buckets buckets{ nano::test::active_transactions_insert_null };
 	buckets.push (1000, block0 (), nano::Gxrb_ratio);
 	buckets.push (1100, block2 (), nano::Gxrb_ratio);
-	ASSERT_EQ (block0 (), buckets.top ());
+	ASSERT_EQ (block0 (), buckets.top ().first);
 	buckets.pop ();
-	ASSERT_EQ (block2 (), buckets.top ());
+	ASSERT_EQ (block2 (), buckets.top ().first);
 	buckets.pop ();
 }
 
 TEST (buckets, pop)
 {
-	nano::scheduler::buckets buckets;
+	nano::scheduler::buckets buckets{ nano::test::active_transactions_insert_null };
 	ASSERT_TRUE (buckets.empty ());
 	buckets.push (1000, block0 (), nano::Gxrb_ratio);
 	ASSERT_FALSE (buckets.empty ());
@@ -185,69 +186,69 @@ TEST (buckets, pop)
 
 TEST (buckets, top_one)
 {
-	nano::scheduler::buckets buckets;
+	nano::scheduler::buckets buckets{ nano::test::active_transactions_insert_null };
 	buckets.push (1000, block0 (), nano::Gxrb_ratio);
-	ASSERT_EQ (block0 (), buckets.top ());
+	ASSERT_EQ (block0 (), buckets.top ().first);
 }
 
 TEST (buckets, top_two)
 {
-	nano::scheduler::buckets buckets;
+	nano::scheduler::buckets buckets{ nano::test::active_transactions_insert_null };
 	buckets.push (1000, block0 (), nano::Gxrb_ratio);
 	buckets.push (1, block1 (), nano::Mxrb_ratio);
-	ASSERT_EQ (block0 (), buckets.top ());
+	ASSERT_EQ (block0 (), buckets.top ().first);
 	buckets.pop ();
-	ASSERT_EQ (block1 (), buckets.top ());
+	ASSERT_EQ (block1 (), buckets.top ().first);
 	buckets.pop ();
 	ASSERT_TRUE (buckets.empty ());
 }
 
 TEST (buckets, top_round_robin)
 {
-	nano::scheduler::buckets buckets;
+	nano::scheduler::buckets buckets{ nano::test::active_transactions_insert_null };
 	buckets.push (1000, blockzero (), 0);
-	ASSERT_EQ (blockzero (), buckets.top ());
+	ASSERT_EQ (blockzero (), buckets.top ().first);
 	buckets.push (1000, block0 (), nano::Gxrb_ratio);
 	buckets.push (1000, block1 (), nano::Mxrb_ratio);
 	buckets.push (1100, block3 (), nano::Mxrb_ratio);
 	buckets.pop (); // blockzero
-	EXPECT_EQ (block1 (), buckets.top ());
+	EXPECT_EQ (block1 (), buckets.top ().first);
 	buckets.pop ();
-	EXPECT_EQ (block0 (), buckets.top ());
+	EXPECT_EQ (block0 (), buckets.top ().first);
 	buckets.pop ();
-	EXPECT_EQ (block3 (), buckets.top ());
+	EXPECT_EQ (block3 (), buckets.top ().first);
 	buckets.pop ();
 	EXPECT_TRUE (buckets.empty ());
 }
 
 TEST (buckets, trim_normal)
 {
-	nano::scheduler::buckets buckets{ 1 };
+	nano::scheduler::buckets buckets{ nano::test::active_transactions_insert_null, 1 };
 	buckets.push (1000, block0 (), nano::Gxrb_ratio);
 	buckets.push (1100, block2 (), nano::Gxrb_ratio);
 	ASSERT_EQ (1, buckets.size ());
-	ASSERT_EQ (block0 (), buckets.top ());
+	ASSERT_EQ (block0 (), buckets.top ().first);
 }
 
 TEST (buckets, trim_reverse)
 {
-	nano::scheduler::buckets buckets{ 1 };
+	nano::scheduler::buckets buckets{ nano::test::active_transactions_insert_null, 1 };
 	buckets.push (1100, block2 (), nano::Gxrb_ratio);
 	buckets.push (1000, block0 (), nano::Gxrb_ratio);
 	ASSERT_EQ (1, buckets.size ());
-	ASSERT_EQ (block0 (), buckets.top ());
+	ASSERT_EQ (block0 (), buckets.top ().first);
 }
 
 TEST (buckets, trim_even)
 {
-	nano::scheduler::buckets buckets{ 2 };
+	nano::scheduler::buckets buckets{ nano::test::active_transactions_insert_null, 2 };
 	buckets.push (1000, block0 (), nano::Gxrb_ratio);
 	buckets.push (1100, block2 (), nano::Gxrb_ratio);
 	ASSERT_EQ (1, buckets.size ());
-	ASSERT_EQ (block0 (), buckets.top ());
+	ASSERT_EQ (block0 (), buckets.top ().first);
 	buckets.push (1000, block1 (), nano::Mxrb_ratio);
 	ASSERT_EQ (2, buckets.size ());
-	ASSERT_EQ (block0 (), buckets.top ());
+	ASSERT_EQ (block0 (), buckets.top ().first);
 	buckets.pop ();
-	ASSERT_EQ (block1 (), buckets.top ());
+	ASSERT_EQ (block1 (), buckets.top ().first);
 }

nano/core_test/scheduler_limiter.cpp

@@ -0,0 +1,37 @@
+#include <nano/node/scheduler/limiter.hpp>
+#include <nano/secure/common.hpp>
+#include <nano/test_common/system.hpp>
+#include <nano/test_common/testutil.hpp>
+
+#include <gtest/gtest.h>
+
+TEST (scheduler_limiter, construction)
+{
+	auto occupancy = std::make_shared<nano::scheduler::limiter> (nano::test::active_transactions_insert_null, 1, nano::election_behavior::normal);
+	ASSERT_EQ (1, occupancy->limit ());
+	ASSERT_TRUE (occupancy->available ());
+}
+
+TEST (scheduler_limiter, limit)
+{
+	auto occupancy = std::make_shared<nano::scheduler::limiter> (nano::test::active_transactions_insert_null, 1, nano::election_behavior::normal);
+	ASSERT_EQ (1, occupancy->limit ());
+	ASSERT_TRUE (occupancy->available ());
+}
+
+TEST (scheduler_limiter, election_activate_observer)
+{
+	nano::test::system system{ 1 };
+	auto occupancy = std::make_shared<nano::scheduler::limiter> ([&] (auto const & block, auto const & behavior) {
+		return system.node (0).active.insert (block, behavior);
+	},
+	1, nano::election_behavior::normal);
+	auto result = occupancy->activate (nano::dev::genesis);
+	ASSERT_TRUE (result.inserted);
+	auto elections = occupancy->elections ();
+	ASSERT_EQ (1, elections.size ());
+	ASSERT_EQ (1, elections.count (nano::dev::genesis->qualified_root ()));
+	ASSERT_FALSE (occupancy->available ());
+	result.election = nullptr; // Implicitly run election destructor notification by clearing the last reference
+	ASSERT_TIMELY (5s, occupancy->available ());
+}

nano/node/CMakeLists.txt

@@ -202,6 +202,8 @@ add_library(
   scheduler/component.cpp
   scheduler/hinted.hpp
   scheduler/hinted.cpp
+  scheduler/limiter.hpp
+  scheduler/limiter.cpp
   scheduler/manual.hpp
   scheduler/manual.cpp
   scheduler/optimistic.hpp

nano/node/active_transactions.cpp

@@ -373,6 +373,14 @@ nano::election_insertion_result nano::active_transactions::insert (const std::sh
 	return result;
 }
 
+std::function<nano::election_insertion_result (std::shared_ptr<nano::block> const & block, nano::election_behavior behavior)> nano::active_transactions::insert_fn ()
+{
+	return [this] (std::shared_ptr<nano::block> const & block, nano::election_behavior behavior) {
+		auto result = insert (block, behavior);
+		return result;
+	};
+}
+
 void nano::active_transactions::trim ()
 {
 	/*

nano/node/active_transactions.hpp

@@ -142,6 +142,8 @@ class active_transactions final
 	 * Starts new election with a specified behavior type
 	 */
 	nano::election_insertion_result insert (std::shared_ptr<nano::block> const & block, nano::election_behavior behavior = nano::election_behavior::normal);
+	// Function wrapper around call to ::insert
+	std::function<nano::election_insertion_result (std::shared_ptr<nano::block> const & block, nano::election_behavior behavior)> insert_fn ();
 	// Distinguishes replay votes, cannot be determined if the block is not in any election
 	nano::vote_code vote (std::shared_ptr<nano::vote> const &);
 	// Is the root of this block in the roots container

nano/node/election.cpp

@@ -32,6 +32,11 @@ nano::election::election (nano::node & node_a, std::shared_ptr<nano::block> cons
 	last_blocks.emplace (block_a->hash (), block_a);
 }
 
+nano::election::~election ()
+{
+	destructor_observers.notify (qualified_root);
+}
+
 void nano::election::confirm_once (nano::unique_lock<nano::mutex> & lock_a, nano::election_status_type type_a)
 {
 	debug_assert (lock_a.owns_lock ());

nano/node/election.hpp

@@ -138,7 +138,10 @@ class election final : public std::enable_shared_from_this<nano::election>
 
 public: // Interface
 	election (nano::node &, std::shared_ptr<nano::block> const & block, std::function<void (std::shared_ptr<nano::block> const &)> const & confirmation_action, std::function<void (nano::account const &)> const & vote_action, nano::election_behavior behavior);
+	~election ();
+	nano::observer_set<nano::qualified_root const &> destructor_observers;
 
+public:
 	std::shared_ptr<nano::block> find (nano::block_hash const &) const;
 	/*
 	 * Process vote. Internally uses cooldown to throttle non-final votes

nano/node/scheduler/bucket.cpp

@@ -1,5 +1,6 @@
 #include <nano/lib/blocks.hpp>
 #include <nano/node/scheduler/bucket.hpp>
+#include <nano/node/scheduler/limiter.hpp>
 
 bool nano::scheduler::bucket::value_type::operator< (value_type const & other_a) const
 {
@@ -11,20 +12,23 @@ bool nano::scheduler::bucket::value_type::operator== (value_type const & other_a
 	return time == other_a.time && block->hash () == other_a.block->hash ();
 }
 
-nano::scheduler::bucket::bucket (size_t maximum) :
-	maximum{ maximum }
+nano::scheduler::bucket::bucket (std::shared_ptr<nano::scheduler::limiter> limiter, size_t maximum) :
+	maximum{ maximum },
+	limiter{ limiter }
 {
 	debug_assert (maximum > 0);
+	debug_assert (limiter != nullptr);
 }
 
 nano::scheduler::bucket::~bucket ()
 {
 }
 
-std::shared_ptr<nano::block> nano::scheduler::bucket::top () const
+std::pair<std::shared_ptr<nano::block>, std::shared_ptr<nano::scheduler::limiter>> nano::scheduler::bucket::top () const
 {
 	debug_assert (!queue.empty ());
-	return queue.begin ()->block;
+	auto & first = *queue.begin ();
+	return { first.block, limiter };
 }
 
 void nano::scheduler::bucket::pop ()
@@ -53,6 +57,11 @@ bool nano::scheduler::bucket::empty () const
 	return queue.empty ();
 }
 
+bool nano::scheduler::bucket::available () const
+{
+	return !queue.empty () && limiter->available ();
+}
+
 void nano::scheduler::bucket::dump () const
 {
 	for (auto const & item : queue)

nano/node/scheduler/bucket.hpp

@@ -11,6 +11,7 @@ class block;
 }
 namespace nano::scheduler
 {
+class limiter;
 /** A class which holds an ordered set of blocks to be scheduled, ordered by their block arrival time
  */
 class bucket final
@@ -25,15 +26,17 @@ class bucket final
 	};
 	std::set<value_type> queue;
 	size_t const maximum;
+	std::shared_ptr<nano::scheduler::limiter> limiter;
 
 public:
-	bucket (size_t maximum);
+	bucket (std::shared_ptr<nano::scheduler::limiter> limiter, size_t maximum);
 	~bucket ();
-	std::shared_ptr<nano::block> top () const;
+	std::pair<std::shared_ptr<nano::block>, std::shared_ptr<nano::scheduler::limiter>> top () const;
 	void pop ();
 	void push (uint64_t time, std::shared_ptr<nano::block> block);
 	size_t size () const;
 	bool empty () const;
+	bool available () const;
 	void dump () const;
 };
 } // namespace nano::scheduler