Daemon latch

nano/lib/signal_manager.cpp

@@ -82,3 +82,21 @@ void nano::signal_manager::base_handler (nano::signal_manager::signal_descriptor
 		logger.debug (nano::log::type::signal_manager, "Signal error: {} ({})", ec.message (), to_signal_name (signum));
 	}
 }
+
+std::string nano::to_signal_name (int signum)
+{
+	switch (signum)
+	{
+		case SIGINT:
+			return "SIGINT";
+		case SIGTERM:
+			return "SIGTERM";
+		case SIGSEGV:
+			return "SIGSEGV";
+		case SIGABRT:
+			return "SIGABRT";
+		case SIGILL:
+			return "SIGILL";
+	}
+	return std::to_string (signum);
+}

nano/lib/signal_manager.hpp

@@ -73,4 +73,5 @@ private:
 	boost::thread smthread;
 };
 
+std::string to_signal_name (int signum);
 }

nano/nano_node/daemon.cpp

@@ -14,11 +14,11 @@
 #include <nano/node/openclwork.hpp>
 #include <nano/rpc/rpc.hpp>
 
-#include <boost/format.hpp>
 #include <boost/process.hpp>
 
 #include <csignal>
 #include <iostream>
+#include <latch>
 
 #include <fmt/chrono.h>
 
@@ -56,8 +56,6 @@ void install_abort_signal_handler ()
 #endif
 }
 
-volatile sig_atomic_t sig_int_or_term = 0;
-
 constexpr std::size_t OPEN_FILE_DESCRIPTORS_LIMIT = 16384;
 }
 
@@ -146,16 +144,28 @@ void nano::daemon::run (std::filesystem::path const & data_path, nano::node_flag
 				logger.info (nano::log::type::daemon, "Database backend: {}", node->store.vendor_get ());
 				logger.info (nano::log::type::daemon, "Start time: {:%c} UTC", fmt::gmtime (dateTime));
 
+				// IO context runner should be started first and stopped last to allow asio handlers to execute during node start/stop
+				runner = std::make_unique<nano::thread_runner> (io_ctx, node->config.io_threads);
+
 				node->start ();
 
-				nano::ipc::ipc_server ipc_server (*node, config.rpc);
+				std::atomic stopped{ false };
+
+				std::unique_ptr<nano::ipc::ipc_server> ipc_server = std::make_unique<nano::ipc::ipc_server> (*node, config.rpc);
 				std::unique_ptr<boost::process::child> rpc_process;
-				std::shared_ptr<nano::rpc> rpc;
 				std::unique_ptr<nano::rpc_handler_interface> rpc_handler;
+				std::shared_ptr<nano::rpc> rpc;
+
 				if (config.rpc_enable)
 				{
 					if (!config.rpc.child_process.enable)
 					{
+						auto stop_callback = [this, &stopped] () {
+							logger.warn (nano::log::type::daemon, "RPC stop request received, stopping...");
+							stopped = true;
+							stopped.notify_all ();
+						};
+
 						// Launch rpc in-process
 						nano::rpc_config rpc_config{ config.node.network_params.network };
 						auto error = nano::read_rpc_config_toml (data_path, rpc_config, flags.rpc_config_overrides);
@@ -166,16 +176,7 @@ void nano::daemon::run (std::filesystem::path const & data_path, nano::node_flag
 						}
 
 						rpc_config.tls_config = tls_config;
-						rpc_handler = std::make_unique<nano::inprocess_rpc_handler> (*node, ipc_server, config.rpc,
+						rpc_handler = std::make_unique<nano::inprocess_rpc_handler> (*node, *ipc_server, config.rpc, stop_callback);
-						[&ipc_server, &workers = node->workers, io_ctx_w = std::weak_ptr{ io_ctx }] () {
-							ipc_server.stop ();
-							workers.add_timed_task (std::chrono::steady_clock::now () + std::chrono::seconds (3), [io_ctx_w] () {
-								if (auto io_ctx_l = io_ctx_w.lock ())
-								{
-									io_ctx_l->stop ();
-								}
-							});
-						});
 						rpc = nano::get_rpc (io_ctx, rpc_config, *rpc_handler);
 						rpc->start ();
 					}
@@ -191,38 +192,35 @@ void nano::daemon::run (std::filesystem::path const & data_path, nano::node_flag
 
 						rpc_process = std::make_unique<boost::process::child> (config.rpc.child_process.rpc_path, "--daemon", "--data_path", data_path.string (), "--network", network);
 					}
+					debug_assert (rpc || rpc_process);
 				}
 
-				auto signal_handler = [this, io_ctx_w = std::weak_ptr{ io_ctx }] (int signum) {
+				auto signal_handler = [this, &stopped] (int signum) {
-					logger.warn (nano::log::type::daemon, "Interrupt signal received, stopping...");
+					logger.warn (nano::log::type::daemon, "Interrupt signal received ({}), stopping...", to_signal_name (signum));
-
+					stopped = true;
-					if (auto io_ctx_l = io_ctx_w.lock ())
+					stopped.notify_all ();
-					{
-						io_ctx_l->stop ();
-					}
-					sig_int_or_term = 1;
 				};
 
 				nano::signal_manager sigman;
-
 				// keep trapping Ctrl-C to avoid a second Ctrl-C interrupting tasks started by the first
 				sigman.register_signal_handler (SIGINT, signal_handler, true);
-
 				// sigterm is less likely to come in bunches so only trap it once
 				sigman.register_signal_handler (SIGTERM, signal_handler, false);
 
-				runner = std::make_unique<nano::thread_runner> (io_ctx, node->config.io_threads);
+				// Keep running until stopped flag is set
+				stopped.wait (false);
+
+				logger.info (nano::log::type::daemon, "Stopping...");
+
+				if (rpc)
+				{
+					rpc->stop ();
+				}
+				ipc_server->stop ();
+				node->stop ();
+				io_ctx->stop ();
 				runner->join ();
 
-				if (sig_int_or_term == 1)
-				{
-					ipc_server.stop ();
-					node->stop ();
-					if (rpc)
-					{
-						rpc->stop ();
-					}
-				}
 				if (rpc_process)
 				{
 					rpc_process->wait ();
@@ -243,5 +241,5 @@ void nano::daemon::run (std::filesystem::path const & data_path, nano::node_flag
 		logger.critical (nano::log::type::daemon, "Error deserializing config: {}", error.get_message ());
 	}
 
-	logger.info (nano::log::type::daemon, "Daemon exiting");
+	logger.info (nano::log::type::daemon, "Daemon stopped");
 }

nano/nano_rpc/entry.cpp

@@ -15,10 +15,10 @@
 #include <boost/filesystem.hpp>
 #include <boost/program_options.hpp>
 
+#include <latch>
+
 namespace
 {
-volatile sig_atomic_t sig_int_or_term = 0;
-
 nano::logger logger{ "rpc_daemon" };
 
 void run (std::filesystem::path const & data_path, std::vector<std::string> const & config_overrides)
@@ -41,7 +41,7 @@ void run (std::filesystem::path const & data_path, std::vector<std::string> cons
 		error = nano::read_tls_config_toml (data_path, *tls_config, logger);
 		if (error)
 		{
-			logger.critical (nano::log::type::daemon, "Error reading RPC TLS config: {}", error.get_message ());
+			logger.critical (nano::log::type::daemon_rpc, "Error reading RPC TLS config: {}", error.get_message ());
 			std::exit (1);
 		}
 		else
@@ -51,42 +51,43 @@ void run (std::filesystem::path const & data_path, std::vector<std::string> cons
 
 		std::shared_ptr<boost::asio::io_context> io_ctx = std::make_shared<boost::asio::io_context> ();
 
-		nano::signal_manager sigman;
+		runner = std::make_unique<nano::thread_runner> (io_ctx, rpc_config.rpc_process.io_threads);
+
 		try
 		{
 			nano::ipc_rpc_processor ipc_rpc_processor (*io_ctx, rpc_config);
 			auto rpc = nano::get_rpc (io_ctx, rpc_config, ipc_rpc_processor);
 			rpc->start ();
 
-			auto signal_handler = [io_ctx_w = std::weak_ptr{ io_ctx }] (int signum) {
+			std::atomic stopped{ false };
-				logger.warn (nano::log::type::daemon, "Interrupt signal received, stopping...");
 
-				if (auto io_ctx_l = io_ctx_w.lock ())
+			auto signal_handler = [&stopped] (int signum) {
-				{
+				logger.warn (nano::log::type::daemon_rpc, "Interrupt signal received ({}), stopping...", nano::to_signal_name (signum));
-					io_ctx_l->stop ();
+				stopped = true;
-				}
+				stopped.notify_all ();
-				sig_int_or_term = 1;
 			};
 
+			nano::signal_manager sigman;
 			sigman.register_signal_handler (SIGINT, signal_handler, true);
 			sigman.register_signal_handler (SIGTERM, signal_handler, false);
 
-			runner = std::make_unique<nano::thread_runner> (io_ctx, rpc_config.rpc_process.io_threads);
+			// Keep running until stopped flag is set
-			runner->join ();
+			stopped.wait (false);
 
-			if (sig_int_or_term == 1)
+			logger.info (nano::log::type::daemon_rpc, "Stopping...");
-			{
+
-				rpc->stop ();
+			rpc->stop ();
-			}
+			io_ctx->stop ();
+			runner->join ();
 		}
 		catch (std::runtime_error const & e)
 		{
-			logger.critical (nano::log::type::daemon, "Error while running RPC: {}", e.what ());
+			logger.critical (nano::log::type::daemon_rpc, "Error while running RPC: {}", e.what ());
 		}
 	}
 	else
 	{
-		logger.critical (nano::log::type::daemon, "Error deserializing config: {}", error.get_message ());
+		logger.critical (nano::log::type::daemon_rpc, "Error deserializing config: {}", error.get_message ());
 	}
 
 	logger.info (nano::log::type::daemon_rpc, "Daemon stopped (RPC)");