Directed block broadcasting for long elections (#2505)

* Split election block broadcasting into directed and random floods

* Formattting

* Assert no errror on system.poll_until_true (cc comment)

* As one line

* Simplify and explain the two copies (Serg comment)

nano/core_test/active_transactions.cpp

@@ -590,11 +590,7 @@ TEST (active_transactions, update_difficulty)
 	node1.process_active (send1);
 	node1.process_active (send2);
 	node1.block_processor.flush ();
-	system.deadline_set (10s);
-	while (node1.active.size () != 2 || node2.active.size () != 2)
-	{
-		ASSERT_NO_ERROR (system.poll ());
-	}
+	ASSERT_NO_ERROR (system.poll_until_true (10s, [&node1, &node2] { return node1.active.size () == 2 && node2.active.size () == 2; }));
 	// Update work with higher difficulty
 	auto work1 = node1.work_generate_blocking (send1->root (), difficulty1 + 1);
 	auto work2 = node1.work_generate_blocking (send2->root (), difficulty2 + 1);
@@ -609,6 +605,10 @@ TEST (active_transactions, update_difficulty)
 	node1.process_active (send1);
 	node1.process_active (send2);
 	node1.block_processor.flush ();
+	// Share the updated blocks
+	node1.network.flood_block (send1);
+	node1.network.flood_block (send2);
+
 	system.deadline_set (10s);
 	bool done (false);
 	while (!done)

nano/node/confirmation_solicitor.cpp

@@ -7,6 +7,7 @@ nano::confirmation_solicitor::confirmation_solicitor (nano::network & network_a,
 max_confirm_req_batches (params_a.is_test_network () ? 1 : 20),
 max_block_broadcasts (params_a.is_test_network () ? 4 : 30),
 max_election_requests (30),
+max_election_broadcasts (network_a.fanout () / 2),
 network (network_a)
 {
 }
@@ -16,20 +17,34 @@ void nano::confirmation_solicitor::prepare (std::vector<nano::representative> co
 	debug_assert (!prepared);
 	requests.clear ();
 	rebroadcasted = 0;
-	representatives = representatives_a;
+	/** Two copies are required as representatives can be erased from \p representatives_requests */
+	representatives_requests = representatives_a;
+	representatives_broadcasts = representatives_a;
 	prepared = true;
 }
 
 bool nano::confirmation_solicitor::broadcast (nano::election const & election_a)
 {
 	debug_assert (prepared);
-	bool result (true);
+	bool error (true);
 	if (rebroadcasted++ < max_block_broadcasts)
 	{
-		network.flood_block (election_a.status.winner);
-		result = false;
+		nano::publish winner (election_a.status.winner);
+		unsigned count = 0;
+		// Directed broadcasting to principal representatives
+		for (auto i (representatives_broadcasts.begin ()), n (representatives_broadcasts.end ()); i != n && count < max_election_broadcasts; ++i)
+		{
+			if (election_a.last_votes.find (i->account) == election_a.last_votes.end ())
+			{
+				i->channel->send (winner);
+				++count;
+			}
+		}
+		// Random flood for block propagation
+		network.flood_message (winner, nano::buffer_drop_policy::limiter, 0.5f);
+		error = false;
 	}
-	return result;
+	return error;
 }
 
 bool nano::confirmation_solicitor::add (nano::election const & election_a)
@@ -37,7 +52,7 @@ bool nano::confirmation_solicitor::add (nano::election const & election_a)
 	debug_assert (prepared);
 	auto const max_channel_requests (max_confirm_req_batches * nano::network::confirm_req_hashes_max);
 	unsigned count = 0;
-	for (auto i (representatives.begin ()); i != representatives.end () && count < max_election_requests;)
+	for (auto i (representatives_requests.begin ()); i != representatives_requests.end () && count < max_election_requests;)
 	{
 		bool full_queue (false);
 		auto rep (*i);
@@ -54,7 +69,7 @@ bool nano::confirmation_solicitor::add (nano::election const & election_a)
 				full_queue = true;
 			}
 		}
-		i = !full_queue ? i + 1 : representatives.erase (i);
+		i = !full_queue ? i + 1 : representatives_requests.erase (i);
 	}
 	return count == 0;
 }

nano/node/confirmation_solicitor.hpp

@@ -28,12 +28,15 @@ public:
 	size_t const max_block_broadcasts;
 	/** Maximum amount of requests to be sent per election */
 	size_t const max_election_requests;
+	/** Maximum amount of directed broadcasts to be sent per election */
+	size_t const max_election_broadcasts;
 
 private:
 	nano::network & network;
 
 	unsigned rebroadcasted{ 0 };
-	std::vector<nano::representative> representatives;
+	std::vector<nano::representative> representatives_requests;
+	std::vector<nano::representative> representatives_broadcasts;
 	using vector_root_hashes = std::vector<std::pair<nano::block_hash, nano::root>>;
 	std::unordered_map<std::shared_ptr<nano::transport::channel>, vector_root_hashes> requests;
 	bool prepared{ false };

nano/node/network.cpp

@@ -148,9 +148,9 @@ void nano::network::send_node_id_handshake (std::shared_ptr<nano::transport::cha
 	channel_a->send (message);
 }
 
-void nano::network::flood_message (nano::message const & message_a, nano::buffer_drop_policy drop_policy_a)
+void nano::network::flood_message (nano::message const & message_a, nano::buffer_drop_policy const drop_policy_a, float const scale_a)
 {
-	for (auto & i : list (fanout ()))
+	for (auto & i : list (fanout (scale_a)))
 	{
 		i->send (message_a, nullptr, drop_policy_a);
 	}

nano/node/network.hpp

@@ -100,7 +100,7 @@ public:
 	~network ();
 	void start ();
 	void stop ();
-	void flood_message (nano::message const &, nano::buffer_drop_policy = nano::buffer_drop_policy::limiter);
+	void flood_message (nano::message const &, nano::buffer_drop_policy const = nano::buffer_drop_policy::limiter, float const = 1.0f);
 	void flood_keepalive ()
 	{
 		nano::keepalive message;