Adding limiter

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/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/scheduler/limiter.cpp

@@ -0,0 +1,63 @@
+#include <nano/lib/locks.hpp>
+#include <nano/lib/stats.hpp>
+#include <nano/node/active_transactions.hpp>
+#include <nano/node/scheduler/limiter.hpp>
+
+#include <boost/format.hpp>
+
+nano::scheduler::limiter::limiter (insert_t const & insert, size_t limit, nano::election_behavior behavior) :
+	insert{ insert },
+	limit_m{ limit },
+	behavior{ behavior }
+{
+	debug_assert (limit > 0);
+}
+
+size_t nano::scheduler::limiter::limit () const
+{
+	return limit_m;
+}
+
+std::unordered_set<nano::qualified_root> nano::scheduler::limiter::elections () const
+{
+	nano::lock_guard<nano::mutex> lock{ mutex };
+	return elections_m;
+}
+
+bool nano::scheduler::limiter::available () const
+{
+	nano::lock_guard<nano::mutex> lock{ mutex };
+	auto result = elections_m.size () < limit ();
+	return result;
+}
+
+nano::election_insertion_result nano::scheduler::limiter::activate (std::shared_ptr<nano::block> const & block)
+{
+	if (!available ())
+	{
+		return { nullptr, false };
+	}
+
+	// This code section is not synchronous with respect to available ()
+	// It is assumed 'sink' is thread safe and only one call to
+	auto result = insert (block, behavior);
+	if (result.inserted)
+	{
+		nano::lock_guard<nano::mutex> lock{ mutex };
+		elections_m.insert (result.election->qualified_root);
+		// Capture via weak_ptr so we don't have to consider destruction order of nano::scheduler::limiter compared to nano::election.
+		result.election->destructor_observers.add ([this_w = std::weak_ptr<nano::scheduler::limiter>{ shared_from_this () }] (nano::qualified_root const & root) {
+			if (auto this_l = this_w.lock ())
+			{
+				this_l->election_destruction_notification (root);
+			}
+		});
+	}
+	return result;
+}
+
+size_t nano::scheduler::limiter::election_destruction_notification (nano::qualified_root const & root)
+{
+	nano::lock_guard<nano::mutex> lock{ mutex };
+	return elections_m.erase (root);
+}

nano/node/scheduler/limiter.hpp

@@ -0,0 +1,49 @@
+#pragma once
+
+#include <nano/lib/locks.hpp>
+#include <nano/lib/numbers.hpp>
+#include <nano/lib/stats_enums.hpp>
+#include <nano/node/election_insertion_result.hpp>
+
+#include <memory>
+#include <unordered_set>
+
+namespace nano
+{
+class block;
+class election;
+enum class election_behavior;
+class stats;
+}
+namespace nano::scheduler
+{
+/**
+	This class is a facade around active_transactions that limits the number of elections that can be inserted.
+*/
+class limiter : public std::enable_shared_from_this<limiter>
+{
+public:
+	using insert_t = std::function<nano::election_insertion_result (std::shared_ptr<nano::block> const & block, nano::election_behavior behavior)>;
+
+	limiter (insert_t const & insert, size_t limit, nano::election_behavior behavior);
+	// Checks whether there is availability to insert an election for 'block' and if so, spawns a new election
+	nano::election_insertion_result activate (std::shared_ptr<nano::block> const & block);
+	// Returns whether there is availability to insert a new election
+	bool available () const;
+	// Returns the upper limit on the number of elections allowed to be started
+	size_t limit () const;
+	std::unordered_set<nano::qualified_root> elections () const;
+
+private:
+	size_t election_destruction_notification (nano::qualified_root const & root);
+
+	insert_t insert;
+	size_t const limit_m;
+	nano::election_behavior behavior;
+	// Tracks the elections that have been started through this facade
+	std::unordered_set<nano::qualified_root> elections_m;
+	std::function<nano::election_insertion_result (std::shared_ptr<nano::block> block)> start_election;
+
+	mutable nano::mutex mutex;
+};
+} // namespace nano::scheduler

nano/test_common/testutil.cpp

@@ -244,3 +244,5 @@ bool nano::test::start_elections (nano::test::system & system_a, nano::node & no
 {
 	return nano::test::start_elections (system_a, node_a, blocks_to_hashes (blocks_a), forced_a);
 }
+
+std::function<nano::election_insertion_result (std::shared_ptr<nano::block> const & block, nano::election_behavior behavior)> nano::test::active_transactions_insert_null = [] (std::shared_ptr<nano::block> const & block, nano::election_behavior behavior) { return nano::election_insertion_result{}; };

nano/test_common/testutil.hpp

@@ -3,6 +3,7 @@
 #include <nano/lib/errors.hpp>
 #include <nano/lib/locks.hpp>
 #include <nano/lib/timer.hpp>
+#include <nano/node/election_insertion_result.hpp>
 #include <nano/node/transport/channel.hpp>
 #include <nano/node/transport/transport.hpp>
 
@@ -131,6 +132,7 @@ class network_params;
 class vote;
 class block;
 class election;
+enum class election_behavior;
 
 extern nano::uint128_t const & genesis_amount;
 
@@ -418,5 +420,6 @@ namespace test
 	 * NOTE: Each election is given 5 seconds to complete, if it does not complete in 5 seconds, it will return an error.
 	 */
 	[[nodiscard]] bool start_elections (nano::test::system &, nano::node &, std::vector<std::shared_ptr<nano::block>> const &, bool const forced_a = false);
+	extern std::function<nano::election_insertion_result (std::shared_ptr<nano::block> const & block, nano::election_behavior behavior)> active_transactions_insert_null;
 }
 }