Rate limit cementing

nano/lib/stats_enums.hpp

@@ -149,6 +149,7 @@ enum class detail
 	sync,
 	requeued,
 	evicted,
+	rate_limited,
 
 	// processing queue
 	queue,

nano/node/confirming_set.cpp

@@ -16,6 +16,7 @@ nano::confirming_set::confirming_set (confirming_set_config const & config_a, na
 	block_processor{ block_processor_a },
 	stats{ stats_a },
 	logger{ logger_a },
+	limiter{ config.rate_limit, /* unlimited token bucket capacity */ 0 },
 	workers{ 1, nano::thread_role::name::confirmation_height_notifications }
 {
 	batch_cemented.add ([this] (auto const & cemented) {
@@ -134,7 +135,9 @@ void nano::confirming_set::run ()
 		}
 		else
 		{
-			condition.wait (lock, [&] () { return !set.empty () || stopped; });
+			condition.wait (lock, [&] () {
+				return !set.empty () || stopped;
+			});
 		}
 	}
 }
@@ -243,11 +246,24 @@ void nano::confirming_set::run_batch (std::unique_lock<std::mutex> & lock)
 				{
 					// Confirming this block may implicitly confirm more
 					stats.add (nano::stat::type::confirming_set, nano::stat::detail::cemented, added.size ());
-					for (auto & block : added)
+					for (auto const & block : added)
 					{
 						cemented.push_back ({ block, hash, election });
 					}
 					cemented_count += added.size ();
+
+					// Rate limit cementing
+					while (!limiter.should_pass (added.size ()))
+					{
+						stats.inc (nano::stat::type::confirming_set, nano::stat::detail::rate_limited);
+						transaction.commit ();
+						std::this_thread::sleep_for (100ms);
+						transaction.renew ();
+						if (stopped)
+						{
+							return;
+						}
+					}
 				}
 				else
 				{
@@ -336,6 +352,7 @@ nano::container_info nano::confirming_set::container_info () const
 	nano::container_info info;
 	info.put ("set", set);
 	info.put ("deferred", deferred);
+	info.put ("limiter", limiter.size ());
 	info.add ("workers", workers.container_info ());
 	return info;
 }

nano/node/confirming_set.hpp

@@ -3,6 +3,7 @@
 #include <nano/lib/numbers.hpp>
 #include <nano/lib/numbers_templ.hpp>
 #include <nano/lib/observer_set.hpp>
+#include <nano/lib/rate_limiting.hpp>
 #include <nano/lib/thread_pool.hpp>
 #include <nano/node/fwd.hpp>
 #include <nano/secure/common.hpp>
@@ -41,6 +42,9 @@ class confirming_set_config final
 	size_t max_deferred{ 16 * 1024 };
 	/** Max age of deferred blocks before they are dropped */
 	std::chrono::seconds deferred_age_cutoff{ 15min };
+
+	/** For bounded backlog testing */
+	size_t rate_limit{ 0 };
 };
 
 /**
@@ -120,6 +124,8 @@ class confirming_set final
 	// Blocks that are being cemented in the current batch
 	std::unordered_set<nano::block_hash> current;
 
+	nano::rate_limiter limiter;
+
 	std::atomic<bool> stopped{ false };
 	mutable std::mutex mutex;
 	std::condition_variable condition;