Run packet processing from start

nano/node/network.cpp

@@ -23,14 +23,62 @@ nano::network::network (nano::node & node_a, uint16_t port_a) :
 	tcp_channels (node_a, [this] (nano::message const & message, std::shared_ptr<nano::transport::channel> const & channel) {
 		inbound (message, channel);
 	}),
-	port (port_a), disconnect_observer ([] () {})
+	port (port_a)
 {
-	for (std::size_t i = 0; i < node.config.network_threads && !node.flags.disable_tcp_realtime; ++i)
+}
+
+nano::network::~network ()
 {
-		packet_processing_threads.emplace_back (nano::thread_attributes::get_default (), [this, i] () {
+	// All threads must be stopped before this destructor
+	debug_assert (processing_threads.empty ());
+}
+
+void nano::network::start ()
+{
+	if (!node.flags.disable_connection_cleanup)
+	{
+		ongoing_cleanup ();
+	}
+
+	ongoing_syn_cookie_cleanup ();
+	ongoing_keepalive ();
+
+	if (!node.flags.disable_tcp_realtime)
+	{
+		tcp_channels.start ();
+
+		for (std::size_t i = 0; i < node.config.network_threads; ++i)
+		{
+			processing_threads.emplace_back (nano::thread_attributes::get_default (), [this] () {
 				nano::thread_role::set (nano::thread_role::name::packet_processing);
+				run_processing ();
+			});
+		}
+	}
+}
+
+void nano::network::stop ()
+{
+	stopped = true;
+
+	tcp_channels.stop ();
+	resolver.cancel ();
+	tcp_message_manager.stop ();
+
+	for (auto & thread : processing_threads)
+	{
+		thread.join ();
+	}
+	processing_threads.clear ();
+
+	port = 0;
+}
+
+void nano::network::run_processing ()
+{
 	try
 	{
+		// TODO: Move responsibility of packet queuing and processing to the message_processor class
 		tcp_channels.process_messages ();
 	}
 	catch (boost::system::error_code & ec)
@@ -53,42 +101,6 @@ nano::network::network (nano::node & node_a, uint16_t port_a) :
 		node.logger.critical (nano::log::type::network, "Unknown error");
 		release_assert (false);
 	}
-		});
-	}
-}
-
-nano::network::~network ()
-{
-	stop ();
-}
-
-void nano::network::start ()
-{
-	if (!node.flags.disable_connection_cleanup)
-	{
-		ongoing_cleanup ();
-	}
-	ongoing_syn_cookie_cleanup ();
-	if (!node.flags.disable_tcp_realtime)
-	{
-		tcp_channels.start ();
-	}
-	ongoing_keepalive ();
-}
-
-void nano::network::stop ()
-{
-	if (!stopped.exchange (true))
-	{
-		tcp_channels.stop ();
-		resolver.cancel ();
-		tcp_message_manager.stop ();
-		port = 0;
-		for (auto & thread : packet_processing_threads)
-		{
-			thread.join ();
-		}
-	}
 }
 
 void nano::network::send_keepalive (std::shared_ptr<nano::transport::channel> const & channel_a)

nano/node/network.hpp

@@ -130,6 +130,7 @@ public: // Handshake
 	nano::node_id_handshake::response_payload prepare_handshake_response (nano::node_id_handshake::query_payload const & query, bool v2) const;
 
 private:
+	void run_processing ();
 	void process_message (nano::message const &, std::shared_ptr<nano::transport::channel> const &);
 
 private: // Dependencies
@@ -139,18 +140,20 @@ public:
 	nano::networks const id;
 	nano::syn_cookies syn_cookies;
 	boost::asio::ip::udp::resolver resolver;
-	std::vector<boost::thread> packet_processing_threads;
 	nano::peer_exclusion excluded_peers;
 	nano::tcp_message_manager tcp_message_manager;
 	nano::network_filter publish_filter;
 	nano::transport::tcp_channels tcp_channels;
 	std::atomic<uint16_t> port{ 0 };
-	std::function<void ()> disconnect_observer;
+
+public: // Callbacks
+	std::function<void ()> disconnect_observer{ [] () {} };
 	// Called when a new channel is observed
-	std::function<void (std::shared_ptr<nano::transport::channel>)> channel_observer;
+	std::function<void (std::shared_ptr<nano::transport::channel>)> channel_observer{ [] (auto) {} };
 
 private:
 	std::atomic<bool> stopped{ false };
+	std::vector<boost::thread> processing_threads; // Using boost::thread to enable increased stack size
 
 public:
 	static unsigned const broadcast_interval_ms = 10;