Skip to content

Commit

Permalink
Daemon latch
Browse files Browse the repository at this point in the history
  • Loading branch information
@pwojcikdev
pwojcikdev committed Apr 14, 2024
1 parent f9ea8fe commit 7ff5c04
Show file tree
Hide file tree
Showing 4 changed files with 70 additions and 54 deletions.
19 changes: 19 additions & 0 deletions nano/lib/signal_manager.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -82,3 +82,22 @@ 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_view 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";
default:
return "UNKNOWN";
}
}
1 change: 1 addition & 0 deletions nano/lib/signal_manager.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -73,4 +73,5 @@ class signal_manager final
boost::thread smthread;
};

std::string_view to_signal_name (int signum);
}
66 changes: 31 additions & 35 deletions nano/nano_node/daemon.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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>

Expand Down Expand Up @@ -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;
}

Expand Down Expand Up @@ -146,16 +144,27 @@ 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::latch latch{ 1 };

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, &latch] () {
logger.warn (nano::log::type::daemon, "RPC stop request received, stopping...");
latch.count_down ();
};

// 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);
Expand All @@ -166,16 +175,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,
[&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_handler = std::make_unique<nano::inprocess_rpc_handler> (*node, *ipc_server, config.rpc, stop_callback);
rpc = nano::get_rpc (io_ctx, rpc_config, *rpc_handler);
rpc->start ();
}
Expand All @@ -191,38 +191,34 @@ 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) {
logger.warn (nano::log::type::daemon, "Interrupt signal received, stopping...");

if (auto io_ctx_l = io_ctx_w.lock ())
{
io_ctx_l->stop ();
}
sig_int_or_term = 1;
auto signal_handler = [this, &latch] (int signum) {
logger.warn (nano::log::type::daemon, "Interrupt signal received ({}), stopping...", to_signal_name (signum));
latch.count_down ();
};

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);
runner->join ();
// Keep running until latch is triggered
latch.wait ();

if (sig_int_or_term == 1)
logger.info (nano::log::type::daemon, "Stopping...");

if (rpc)
{
ipc_server.stop ();
node->stop ();
if (rpc)
{
rpc->stop ();
}
rpc->stop ();
}
ipc_server->stop ();
node->stop ();
io_ctx->stop ();
runner->join ();

if (rpc_process)
{
rpc_process->wait ();
Expand All @@ -243,5 +239,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");
}
38 changes: 19 additions & 19 deletions nano/nano_rpc/entry.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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
Expand All @@ -51,42 +51,42 @@ 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) {
logger.warn (nano::log::type::daemon, "Interrupt signal received, stopping...");
std::latch latch{ 1 };

if (auto io_ctx_l = io_ctx_w.lock ())
{
io_ctx_l->stop ();
}
sig_int_or_term = 1;
auto signal_handler = [&latch] (int signum) {
logger.warn (nano::log::type::daemon_rpc, "Interrupt signal received ({}), stopping...", nano::to_signal_name (signum));
latch.count_down ();
};

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);
runner->join ();
// Keep running until latch is triggered
latch.wait ();

logger.info (nano::log::type::daemon_rpc, "Stopping...");

if (sig_int_or_term == 1)
{
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)");
Expand Down

0 comments on commit 7ff5c04

Please sign in to comment.