Modifying buckets

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 () && !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/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/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 ()
@@ -55,7 +59,7 @@ bool nano::scheduler::bucket::empty () const
 
 bool nano::scheduler::bucket::available () const
 {
-	return !queue.empty ();
+	return !queue.empty () && limiter->available ();
 }
 
 void nano::scheduler::bucket::dump () const

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,11 +26,12 @@ 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;

nano/node/scheduler/buckets.cpp

@@ -1,7 +1,9 @@
 #include <nano/lib/blocks.hpp>
 #include <nano/lib/utility.hpp>
+#include <nano/node/election.hpp>
 #include <nano/node/scheduler/bucket.hpp>
 #include <nano/node/scheduler/buckets.hpp>
+#include <nano/node/scheduler/limiter.hpp>
 
 #include <string>
 
@@ -29,7 +31,7 @@ void nano::scheduler::buckets::seek ()
  * Prioritization constructor, construct a container containing approximately 'maximum' number of blocks.
  * @param maximum number of blocks that this container can hold, this is a soft and approximate limit.
  */
-nano::scheduler::buckets::buckets (uint64_t maximum) :
+nano::scheduler::buckets::buckets (insert_t const & insert, uint64_t maximum) :
 	maximum{ maximum }
 {
 	auto build_region = [this] (uint128_t const & begin, uint128_t const & end, size_t count) {
@@ -52,7 +54,8 @@ nano::scheduler::buckets::buckets (uint64_t maximum) :
 	auto bucket_max = std::max<size_t> (1u, maximum / minimums.size ());
 	for (size_t i = 0u, n = minimums.size (); i < n; ++i)
 	{
-		buckets_m.push_back (std::make_unique<scheduler::bucket> (bucket_max));
+		auto limiter = std::make_shared<scheduler::limiter> (insert, bucket_max, nano::election_behavior::normal);
+		buckets_m.push_back (std::make_unique<scheduler::bucket> (limiter, bucket_max));
 	}
 	current = buckets_m.begin ();
 }
@@ -83,7 +86,7 @@ void nano::scheduler::buckets::push (uint64_t time, std::shared_ptr<nano::block>
 }
 
 /** Return the highest priority block of the current bucket */
-std::shared_ptr<nano::block> nano::scheduler::buckets::top () const
+std::pair<std::shared_ptr<nano::block>, std::shared_ptr<nano::scheduler::limiter>> nano::scheduler::buckets::top () const
 {
 	debug_assert (available ());
 	auto result = (*current)->top ();

nano/node/scheduler/buckets.hpp

@@ -1,6 +1,8 @@
 #pragma once
 #include <nano/lib/numbers.hpp>
 #include <nano/lib/utility.hpp>
+#include <nano/node/election.hpp>
+#include <nano/node/election_insertion_result.hpp>
 
 #include <cstddef>
 #include <cstdint>
@@ -9,11 +11,13 @@
 
 namespace nano
 {
+class active_transactions;
 class block;
 }
 namespace nano::scheduler
 {
 class bucket;
+class limiter;
 /** A container for holding blocks and their arrival/creation time.
  *
  *  The container consists of a number of buckets. Each bucket holds an ordered set of 'value_type' items.
@@ -40,14 +44,16 @@ class buckets final
 	uint64_t const maximum;
 
 	void next ();
-	void seek ();
 
 public:
-	buckets (uint64_t maximum = 250000u);
+	using insert_t = std::function<nano::election_insertion_result (std::shared_ptr<nano::block> const & block, nano::election_behavior behavior)>;
+
+	buckets (insert_t const & insert, uint64_t maximum = 250000u);
 	~buckets ();
 	void push (uint64_t time, std::shared_ptr<nano::block> block, nano::amount const & priority);
-	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 seek ();
 	std::size_t size () const;
 	std::size_t bucket_count () const;
 	std::size_t bucket_size (std::size_t index) const;

nano/node/scheduler/priority.cpp

@@ -5,7 +5,7 @@
 nano::scheduler::priority::priority (nano::node & node_a, nano::stats & stats_a) :
 	node{ node_a },
 	stats{ stats_a },
-	buckets{ std::make_unique<scheduler::buckets> () }
+	buckets{ std::make_unique<scheduler::buckets> (node_a.active.insert_fn ()) }
 {
 }
 
@@ -110,7 +110,7 @@ void nano::scheduler::priority::run ()
 				buckets->pop ();
 				lock.unlock ();
 				stats.inc (nano::stat::type::election_scheduler, nano::stat::detail::insert_priority);
-				auto result = node.active.insert (block);
+				auto result = node.active.insert (block.first);
 				if (result.inserted)
 				{
 					stats.inc (nano::stat::type::election_scheduler, nano::stat::detail::insert_priority_success);