Merge pull request #4609 from pwojcikdev/fair-queuing/message-process…

…or-3

Replace `tcp_message_manager`

nano/core_test/network.cpp

@@ -878,68 +878,6 @@ TEST (network, tcp_no_accept_excluded_peers)
 	ASSERT_TIMELY_EQ (5s, node0->network.size (), 1);
 }
 
-/*
-namespace nano
-{
-TEST (network, tcp_message_manager)
-{
-	nano::transport::tcp_message_manager manager (1);
-	item.node_id = nano::account (100);
-	ASSERT_EQ (0, manager.entries.size ());
-	manager.put_message (item);
-	ASSERT_EQ (1, manager.entries.size ());
-	ASSERT_EQ (manager.get_message ().node_id, item.node_id);
-	ASSERT_EQ (0, manager.entries.size ());
-
-	// Fill the queue
-	manager.entries = decltype (manager.entries) (manager.max_entries, item);
-	ASSERT_EQ (manager.entries.size (), manager.max_entries);
-
-	// This task will wait until a message is consumed
-	auto future = std::async (std::launch::async, [&] {
-		manager.put_message (item);
-	});
-
-	// This should give sufficient time to execute put_message
-	// and prove that it waits on condition variable
-	std::this_thread::sleep_for (200ms);
-
-	ASSERT_EQ (manager.entries.size (), manager.max_entries);
-	ASSERT_EQ (manager.get_message ().node_id, item.node_id);
-	ASSERT_NE (std::future_status::timeout, future.wait_for (1s));
-	ASSERT_EQ (manager.entries.size (), manager.max_entries);
-
-	nano::tcp_message_manager manager2 (2);
-	size_t message_count = 10'000;
-	std::vector<std::thread> consumers;
-	for (auto i = 0; i < 4; ++i)
-	{
-		consumers.emplace_back ([&] {
-			for (auto i = 0; i < message_count; ++i)
-			{
-				ASSERT_EQ (manager.get_message ().node_id, item.node_id);
-			}
-		});
-	}
-	std::vector<std::thread> producers;
-	for (auto i = 0; i < 4; ++i)
-	{
-		producers.emplace_back ([&] {
-			for (auto i = 0; i < message_count; ++i)
-			{
-				manager.put_message (item);
-			}
-		});
-	}
-
-	for (auto & t : boost::range::join (producers, consumers))
-	{
-		t.join ();
-	}
-}
-}
-*/
-
 TEST (network, cleanup_purge)
 {
 	auto test_start = std::chrono::steady_clock::now ();

nano/core_test/toml.cpp

@@ -273,6 +273,9 @@ TEST (toml, daemon_config_deserialize_defaults)
 	ASSERT_EQ (conf.node.request_aggregator.max_queue, defaults.node.request_aggregator.max_queue);
 	ASSERT_EQ (conf.node.request_aggregator.threads, defaults.node.request_aggregator.threads);
 	ASSERT_EQ (conf.node.request_aggregator.batch_size, defaults.node.request_aggregator.batch_size);
+
+	ASSERT_EQ (conf.node.message_processor.threads, defaults.node.message_processor.threads);
+	ASSERT_EQ (conf.node.message_processor.max_queue, defaults.node.message_processor.max_queue);
 }
 
 TEST (toml, optional_child)
@@ -584,6 +587,10 @@ TEST (toml, daemon_config_deserialize_no_defaults)
 	threads = 999
 	batch_size = 999
 
+	[node.message_processor]
+	threads = 999
+	max_queue = 999
+
 	[opencl]
 	device = 999
 	enable = true
@@ -741,6 +748,9 @@ TEST (toml, daemon_config_deserialize_no_defaults)
 	ASSERT_NE (conf.node.request_aggregator.max_queue, defaults.node.request_aggregator.max_queue);
 	ASSERT_NE (conf.node.request_aggregator.threads, defaults.node.request_aggregator.threads);
 	ASSERT_NE (conf.node.request_aggregator.batch_size, defaults.node.request_aggregator.batch_size);
+
+	ASSERT_NE (conf.node.message_processor.threads, defaults.node.message_processor.threads);
+	ASSERT_NE (conf.node.message_processor.max_queue, defaults.node.message_processor.max_queue);
 }
 
 /** There should be no required values **/

nano/lib/logging_enums.hpp

@@ -49,7 +49,7 @@ enum class type
 	election,
 	blockprocessor,
 	network,
-	network_processed,
+	message,
 	channel,
 	channel_sent,
 	socket,
@@ -78,6 +78,7 @@ enum class type
 	thread_runner,
 	signal_manager,
 	peer_history,
+	message_processor,
 
 	// bootstrap
 	bulk_pull_client,

nano/lib/stats_enums.hpp

@@ -72,6 +72,9 @@ enum class type
 	syn_cookies,
 	peer_history,
 	port_mapping,
+	message_processor,
+	message_processor_overfill,
+	message_processor_type,
 
 	bootstrap_ascending,
 	bootstrap_ascending_accounts,

nano/lib/thread_roles.cpp

@@ -22,8 +22,8 @@ std::string nano::thread_role::get_string (nano::thread_role::name role)
 		case nano::thread_role::name::work:
 			thread_role_name_string = "Work pool";
 			break;
-		case nano::thread_role::name::packet_processing:
-			thread_role_name_string = "Pkt processing";
+		case nano::thread_role::name::message_processing:
+			thread_role_name_string = "Msg processing";
 			break;
 		case nano::thread_role::name::vote_processing:
 			thread_role_name_string = "Vote processing";

nano/lib/thread_roles.hpp

@@ -12,7 +12,7 @@ enum class name
 	unknown,
 	io,
 	work,
-	packet_processing,
+	message_processing,
 	vote_processing,
 	block_processing,
 	request_loop,

nano/node/CMakeLists.txt

@@ -99,6 +99,8 @@ add_library(
   local_vote_history.hpp
   make_store.hpp
   make_store.cpp
+  message_processor.hpp
+  message_processor.cpp
   network.hpp
   network.cpp
   nodeconfig.hpp