Sighup reload bandwidth params (#3257)

Feature: reload bandwidth parameters from config file on receipt of SIGHUP signal

This commit does not get compiled in the Windows version of the software.

The config keys that are read and updated are:
node.bandwidth_limit
node.bandwidth_limit_burst_ratio

nano/core_test/network.cpp

@@ -937,6 +937,22 @@ TEST (network, bandwidth_limiter)
 	channel2->send (message, nullptr, nano::buffer_drop_policy::no_limiter_drop);
 	ASSERT_TIMELY (1s, 1 == node.stats.count (nano::stat::type::drop, nano::stat::detail::publish, nano::stat::dir::out));
 
+	// change the bandwidth settings, 2 packets will be dropped
+	node.network.set_bandwidth_params (1.1, message_size * 2);
+	channel1->send (message);
+	channel2->send (message);
+	channel1->send (message);
+	channel2->send (message);
+	ASSERT_TIMELY (1s, 3 == node.stats.count (nano::stat::type::drop, nano::stat::detail::publish, nano::stat::dir::out));
+
+	// change the bandwidth settings, no packet will be dropped
+	node.network.set_bandwidth_params (4, message_size);
+	channel1->send (message);
+	channel2->send (message);
+	channel1->send (message);
+	channel2->send (message);
+	ASSERT_TIMELY (1s, 3 == node.stats.count (nano::stat::type::drop, nano::stat::detail::publish, nano::stat::dir::out));
+
 	node.stop ();
 }
 

nano/core_test/utility.cpp

@@ -51,6 +51,40 @@ TEST (rate, network)
 	ASSERT_FALSE (bucket.try_consume (1));
 }
 
+TEST (rate, reset)
+{
+	nano::rate::token_bucket bucket (0, 0);
+
+	// consume lots of tokens, buckets should be unlimited
+	ASSERT_TRUE (bucket.try_consume (1000000));
+	ASSERT_TRUE (bucket.try_consume (1000000));
+
+	// set bucket to be limited
+	bucket.reset (1000, 1000);
+	ASSERT_FALSE (bucket.try_consume (1001));
+	ASSERT_TRUE (bucket.try_consume (1000));
+	ASSERT_FALSE (bucket.try_consume (1000));
+	std::this_thread::sleep_for (2ms);
+	ASSERT_TRUE (bucket.try_consume (2));
+
+	// reduce the limit
+	bucket.reset (100, 100 * 1000);
+	ASSERT_FALSE (bucket.try_consume (101));
+	ASSERT_TRUE (bucket.try_consume (100));
+	std::this_thread::sleep_for (1ms);
+	ASSERT_TRUE (bucket.try_consume (100));
+
+	// increase the limit
+	bucket.reset (2000, 1);
+	ASSERT_FALSE (bucket.try_consume (2001));
+	ASSERT_TRUE (bucket.try_consume (2000));
+
+	// back to unlimited
+	bucket.reset (0, 0);
+	ASSERT_TRUE (bucket.try_consume (1000000));
+	ASSERT_TRUE (bucket.try_consume (1000000));
+}
+
 TEST (rate, unlimited)
 {
 	nano::rate::token_bucket bucket (0, 0);

nano/lib/rate_limiting.cpp

@@ -6,15 +6,7 @@
 
 nano::rate::token_bucket::token_bucket (size_t max_token_count_a, size_t refill_rate_a)
 {
-	// A token count of 0 indicates unlimited capacity. We use 1e9 as
-	// a sentinel, allowing largest burst to still be computed.
-	if (max_token_count_a == 0 || refill_rate_a == 0)
-	{
-		refill_rate_a = max_token_count_a = static_cast<size_t> (1e9);
-	}
-	max_token_count = smallest_size = current_size = max_token_count_a;
-	refill_rate = refill_rate_a;
-	last_refill = std::chrono::steady_clock::now ();
+	reset (max_token_count_a, refill_rate_a);
 }
 
 bool nano::rate::token_bucket::try_consume (unsigned tokens_required_a)
@@ -51,3 +43,18 @@ size_t nano::rate::token_bucket::largest_burst () const
 	nano::lock_guard<nano::mutex> lk (bucket_mutex);
 	return max_token_count - smallest_size;
 }
+
+void nano::rate::token_bucket::reset (size_t max_token_count_a, size_t refill_rate_a)
+{
+	nano::lock_guard<nano::mutex> lk (bucket_mutex);
+
+	// A token count of 0 indicates unlimited capacity. We use 1e9 as
+	// a sentinel, allowing largest burst to still be computed.
+	if (max_token_count_a == 0 || refill_rate_a == 0)
+	{
+		refill_rate_a = max_token_count_a = static_cast<size_t> (1e9);
+	}
+	max_token_count = smallest_size = current_size = max_token_count_a;
+	refill_rate = refill_rate_a;
+	last_refill = std::chrono::steady_clock::now ();
+}

nano/lib/rate_limiting.hpp

@@ -41,6 +41,9 @@ namespace rate
 		/** Returns the largest burst observed */
 		size_t largest_burst () const;
 
+		/** Update the max_token_count and/or refill_rate_a parameters */
+		void reset (size_t max_token_count_a, size_t refill_rate_a);
+
 	private:
 		void refill ();
 		size_t max_token_count;

nano/nano_node/daemon.cpp

@@ -21,6 +21,21 @@ namespace
 volatile sig_atomic_t sig_int_or_term = 0;
 }
 
+static void load_and_set_bandwidth_params (std::shared_ptr<nano::node> const & node, boost::filesystem::path const & data_path, nano::node_flags const & flags)
+{
+	nano::daemon_config config (data_path);
+
+	auto error = nano::read_node_config_toml (data_path, config, flags.config_overrides);
+	if (!error)
+	{
+		error = nano::flags_config_conflicts (flags, config.node);
+		if (!error)
+		{
+			node->set_bandwidth_params (config.node.bandwidth_limit, config.node.bandwidth_limit_burst_ratio);
+		}
+	}
+}
+
 void nano_daemon::daemon::run (boost::filesystem::path const & data_path, nano::node_flags const & flags)
 {
 	// Override segmentation fault and aborting.
@@ -141,6 +156,15 @@ void nano_daemon::daemon::run (boost::filesystem::path const & data_path, nano::
 				// sigterm is less likely to come in bunches so only trap it once
 				sigman.register_signal_handler (SIGTERM, &nano::signal_handler, false);
 
+#ifndef _WIN32
+				// on sighup we should reload the bandwidth parameters
+				std::function<void (int)> sighup_signal_handler ([&node, &data_path, &flags] (int signum) {
+					debug_assert (signum == SIGHUP);
+					load_and_set_bandwidth_params (node, data_path, flags);
+				});
+				sigman.register_signal_handler (SIGHUP, sighup_signal_handler, true);
+#endif
+
 				runner = std::make_unique<nano::thread_runner> (io_ctx, node->config.io_threads);
 				runner->join ();
 

nano/node/network.cpp

@@ -795,6 +795,11 @@ void nano::network::erase (nano::transport::channel const & channel_a)
 	}
 }
 
+void nano::network::set_bandwidth_params (double limit_burst_ratio_a, size_t limit_a)
+{
+	limiter.reset (limit_burst_ratio_a, limit_a);
+}
+
 nano::message_buffer_manager::message_buffer_manager (nano::stat & stats_a, size_t size, size_t count) :
 	stats (stats_a),
 	free (count),

nano/node/network.hpp

@@ -179,6 +179,7 @@ public:
 	float size_sqrt () const;
 	bool empty () const;
 	void erase (nano::transport::channel const &);
+	void set_bandwidth_params (double, size_t);
 	nano::message_buffer_manager buffer_container;
 	boost::asio::ip::udp::resolver resolver;
 	std::vector<boost::thread> packet_processing_threads;

nano/node/node.cpp

@@ -1438,6 +1438,14 @@ bool nano::node::epoch_upgrader (nano::raw_key const & prv_a, nano::epoch epoch_
 	return error;
 }
 
+void nano::node::set_bandwidth_params (size_t limit, double ratio)
+{
+	config.bandwidth_limit_burst_ratio = ratio;
+	config.bandwidth_limit = limit;
+	network.set_bandwidth_params (limit, ratio);
+	logger.always_log (boost::str (boost::format ("set_bandwidth_params(%1%, %2%)") % limit % ratio));
+}
+
 void nano::node::epoch_upgrader_impl (nano::raw_key const & prv_a, nano::epoch epoch_a, uint64_t count_limit, uint64_t threads)
 {
 	nano::thread_role::set (nano::thread_role::name::epoch_upgrader);

nano/node/node.hpp

@@ -148,6 +148,7 @@ public:
 	bool online () const;
 	bool init_error () const;
 	bool epoch_upgrader (nano::raw_key const &, nano::epoch, uint64_t, uint64_t);
+	void set_bandwidth_params (size_t limit, double ratio);
 	std::pair<uint64_t, decltype (nano::ledger::bootstrap_weights)> get_bootstrap_weights () const;
 	void populate_backlog ();
 	nano::write_database_queue write_database_queue;

nano/node/transport/transport.cpp

@@ -262,3 +262,8 @@ bool nano::bandwidth_limiter::should_drop (const size_t & message_size_a)
 {
 	return !bucket.try_consume (nano::narrow_cast<unsigned int> (message_size_a));
 }
+
+void nano::bandwidth_limiter::reset (const double limit_burst_ratio_a, const size_t limit_a)
+{
+	bucket.reset (static_cast<size_t> (limit_a * limit_burst_ratio_a), limit_a);
+}

nano/node/transport/transport.hpp

@@ -14,6 +14,7 @@ public:
 	// initialize with limit 0 = unbounded
 	bandwidth_limiter (const double, const size_t);
 	bool should_drop (const size_t &);
+	void reset (const double, const size_t);
 
 private:
 	nano::rate::token_bucket bucket;

systest/set_bandwidth_params.sh

@@ -0,0 +1,68 @@
+#!/bin/sh
+
+# the caller should set the env var NANO_NODE_EXE to point to the nano_node executable
+# if NANO_NODE_EXE is unser ot empty then "../../build/nano_node" is used
+NANO_NODE_EXE=${NANO_NODE_EXE:-../../build/nano_node}
+
+mkdir -p data/log
+rm data/log/log_*.log
+
+# start nano_node and store its pid so we can later send it
+# the SIGHUP signal and so we can terminate it
+echo start nano_node
+$NANO_NODE_EXE --daemon --data_path data &
+pid=$!
+echo pid=$pid
+
+# wait for the node to start-up
+sleep 2
+
+# set bandwidth params 42 and 43 in the config file
+cat > data/config-node.toml <<EOF
+[node]
+bandwidth_limit = 42
+bandwidth_limit_burst_ratio = 43
+EOF
+
+# send nano_node the SIGHUP signal
+kill -HUP $pid
+
+# wait for the signal handler to kick in
+sleep 2
+
+# set another set of bandwidth params 44 and 45 in the config file
+cat > data/config-node.toml <<EOF
+[node]
+bandwidth_limit = 44
+bandwidth_limit_burst_ratio = 45
+EOF
+
+# send nano_node the SIGHUP signal
+kill -HUP $pid
+
+# wait for the signal handler to kick in
+sleep 2
+
+# terminate nano node and wait for it to die
+kill $pid
+wait $pid
+
+# the bandwidth params are logged in logger and we check for that logging below
+
+# check that the first signal handler got run and the data is correct
+grep -q "set_bandwidth_params(42, 43)" data/log/log_*.log
+rc1=$?
+echo rc1=$rc1
+
+# check that the second signal handler got run and the data is correct
+grep -q "set_bandwidth_params(44, 45)" data/log/log_*.log
+rc2=$?
+echo rc2=$rc2
+
+if [ $rc1 -eq 0 -a $rc2 -eq 0 ]; then
+    echo set_bandwith_params PASSED
+    exit 0
+else
+    echo set_bandwith_params FAILED
+    exit 1
+fi