Skip to content

Use sigwait for POSIX signal handling #5

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
huangminghuang merged 2 commits into from
Jun 3, 2026
Merged

Use sigwait for POSIX signal handling #5

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
aee7562
Use sigwait for appbase signal handling
huangminghuang Jun 1, 2026
7057a01
Keep signal waiter alive after sigwait errors
huangminghuang Jun 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
120 changes: 73 additions & 47 deletions application_base.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,16 +1,17 @@
#include <appbase/application_base.hpp>
#include <appbase/version.hpp>

#include <boost/algorithm/string.hpp>
#include <boost/asio/signal_set.hpp>
#include <atomic>
#include <boost/algorithm/string.hpp>

#include <cstring>
#include <iostream>
#include <fstream>
#include <unordered_map>
#include <future>
#include <optional>

#include <pthread.h>
#include <unistd.h>
#include <signal.h>

Expand All @@ -28,17 +29,12 @@ class application_impl {

#ifdef _WIN32
application_impl():_app_options("Application Options"){}
// appbase does not provide Windows console-control signal handling.
void start_signal_thread(std::function<void(int)>) {}
void stop_signal_thread() {}
#else

application_impl():_app_options("Application Options"){
// Create a separate thread to handle signals, so that they don't interrupt I/O.
// stdio does not recover from EINTR.
_signal_catching_thread = std::thread([&ioctx = _signal_catching_io_ctx]() {
auto workwork = boost::asio::make_work_guard(ioctx);
ioctx.run();
});

// after creating the thread for handling signals, we can block signals in the current thread
sigset_t blocked_signals;
get_target_sigset(&blocked_signals);
pthread_sigmask(SIG_BLOCK, &blocked_signals, nullptr);
Expand All @@ -52,11 +48,53 @@ class application_impl {
sigaddset(blocked_signals, SIGHUP);
}

~application_impl() {
/**
* @brief Start the POSIX signal waiter after application callbacks are initialized.
*/
void start_signal_thread(std::function<void(int)> signal_callback) {
if(_signal_catching_thread.joinable())
return;
_signal_callback = std::move(signal_callback);
_signal_thread_running.store(true, std::memory_order_release);
_signal_catching_thread = std::thread([this]() { signal_loop(); });
}

/**
* @brief Stop the POSIX signal waiter without dispatching another shutdown signal.
*/
void stop_signal_thread() {
if(_signal_catching_thread.joinable()) {
_signal_catching_io_ctx.stop();
_signal_thread_running.store(false, std::memory_order_release);
(void)pthread_kill(_signal_catching_thread.native_handle(), SIGTERM);
_signal_catching_thread.join();
}
}

/**
* @brief Wait for blocked process signals and dispatch them to appbase.
*/
void signal_loop() {
sigset_t blocked_signals;
get_target_sigset(&blocked_signals);
pthread_sigmask(SIG_BLOCK, &blocked_signals, nullptr);

while(_signal_thread_running.load(std::memory_order_acquire)) {
int signal_number = 0;
const int result = sigwait(&blocked_signals, &signal_number);
if(result != 0) {
std::cerr << "appbase failed waiting for signal: " << std::strerror(result) << std::endl;
continue;
}

if(!_signal_thread_running.load(std::memory_order_acquire))
break;

_signal_callback(signal_number);
}
}

~application_impl() {
stop_signal_thread();

// need to unblock signals, otherwise next thread created will inherit blocked signals
sigset_t blocked_signals;
Expand Down Expand Up @@ -84,7 +122,8 @@ class application_impl {
any_type_compare_map _any_compare_map;

std::thread _signal_catching_thread;
boost::asio::io_context _signal_catching_io_ctx;
std::atomic_bool _signal_thread_running{false};
std::function<void(int)> _signal_callback;
};

application_base::application_base(std::shared_ptr<void>&& e) :
Expand All @@ -104,7 +143,10 @@ application_base::application_base(std::shared_ptr<void>&& e) :
register_config_type<std::filesystem::path>();
}

application_base::~application_base() { }
application_base::~application_base() {
// Stop before callbacks/plugin lists are destroyed; application_impl's destructor is only a final safety net.
my->stop_signal_thread();
}

void application_base::set_version(uint64_t version) {
my->_version = version;
Expand Down Expand Up @@ -142,25 +184,7 @@ std::filesystem::path application_base::get_logging_conf() const {
return my->_logging_conf;
}

void application_base::wait_for_signal(std::shared_ptr<boost::asio::signal_set> ss) {
ss->async_wait([this, ss](const boost::system::error_code& ec, int) {
if(ec)
return;
quit();
wait_for_signal(ss);
});
}

std::shared_ptr<boost::asio::signal_set> application_base::setup_signal_handling_on_ioc(boost::asio::io_context& io_ctx) {
std::shared_ptr<boost::asio::signal_set> ss = std::make_shared<boost::asio::signal_set>(io_ctx, SIGINT, SIGTERM);
#ifdef SIGPIPE
ss->add(SIGPIPE);
#endif
wait_for_signal(ss);
return ss;
}

void application_base::startup(boost::asio::io_context& io_ctx) {
void application_base::startup() {
try {
for( auto plugin : initialized_plugins ) {
if( is_quiting() ) break;
Expand All @@ -171,27 +195,29 @@ void application_base::startup(boost::asio::io_context& io_ctx) {
shutdown_plugins();
throw;
}

#ifdef SIGHUP
std::shared_ptr<boost::asio::signal_set> sighup_set(new boost::asio::signal_set(io_ctx, SIGHUP));
start_sighup_handler( sighup_set );
#endif
}

void application_base::start_sighup_handler( std::shared_ptr<boost::asio::signal_set> sighup_set ) {
void application_base::handle_signal(int signal_number) {
#ifdef SIGHUP
sighup_set->async_wait([sighup_set, this](const boost::system::error_code& err, int /*num*/) {
if( err ) return;
post_cb(priority::medium, [sighup_set, this]() {
sighup_callback();
if(signal_number == SIGHUP) {
post_cb(priority::medium, [this]() {
if(sighup_callback)
sighup_callback();
for( auto plugin : initialized_plugins ) {
if( is_quiting() ) return;
plugin->handle_sighup();
}
});
start_sighup_handler( sighup_set );
});
return;
}
#endif

#ifdef SIGPIPE
if(signal_number == SIGPIPE)
return;
#endif

quit();
}

void application_base::register_config_type_comparison(std::type_index i, config_comparison_f comp) {
Expand Down Expand Up @@ -379,7 +405,7 @@ bool application_base::initialize_impl(int argc, char** argv, vector<abstract_pl
auto error_header = [&]() { return std::string("appbase: exception thrown during plugin \"") + plugin_name + "\" initialization.\n"; };

// setup handling of SIGINT/SIGTERM/SIGPIPE during initialize
auto ss = setup_signal_handling_on_ioc(my->_signal_catching_io_ctx);
my->start_signal_thread([this](int signal_number) { handle_signal(signal_number); });

try {
if(options.count("plugin") > 0)
Expand Down Expand Up @@ -587,7 +613,7 @@ std::filesystem::path application_base::full_config_file_path() const {
}

void application_base::set_sighup_callback(std::function<void()> callback) {
sighup_callback = callback;
sighup_callback = callback ? std::move(callback) : []() {};
}

const bpo::variables_map& application_base::get_options() const{
Expand Down
19 changes: 10 additions & 9 deletions include/appbase/application_base.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,7 +120,7 @@ class application_base {
return initialize_impl(argc, argv, {find_plugin<Plugin>()...}, initialize_logging, extra_program_options_providers);
}

void startup(boost::asio::io_context& io_ctx);
void startup();

/**
* Wait until quit(), SIGINT or SIGTERM and then shutdown.
Expand Down Expand Up @@ -342,9 +342,10 @@ class application_base {

private:
// members are ordered taking into account that the last one is destructed first
std::function<void()> sighup_callback;
std::function<void()> stop_executor_cb;
std::function<void(int, std::function<void()>)> post_cb;
std::function<void()> sighup_callback = []() {};
std::function<void()> stop_executor_cb = []() {};
/// Default dispatcher is only used before application_t wires in the real executor-backed post callback.
std::function<void(int, std::function<void()>)> post_cb = [](int, std::function<void()> cb) { cb(); };

map<std::type_index, erased_method_ptr> methods;
map<std::type_index, erased_channel_ptr> channels;
Expand All @@ -358,14 +359,14 @@ class application_base {
using plugin_registration_entry = std::pair<std::string, std::function<void(application_base&)>>;
inline static std::vector<plugin_registration_entry> plugin_registrations;

void start_sighup_handler(std::shared_ptr<boost::asio::signal_set> sighup_set);

void set_program_options(const std::vector<extra_program_options_provider>& option_providers = {});
void write_default_config(const std::filesystem::path& cfg_file);
void print_default_config(std::ostream& os);

void wait_for_signal(std::shared_ptr<boost::asio::signal_set> ss);
std::shared_ptr<boost::asio::signal_set> setup_signal_handling_on_ioc(boost::asio::io_context& io_ctx);
/**
* @brief Dispatch a POSIX signal received by the application signal waiter.
*/
void handle_signal(int signal_number);

void handle_exception(std::exception_ptr eptr, std::string_view origin);
};
Expand Down Expand Up @@ -437,7 +438,7 @@ class application_t : public application_base {
}

void startup() {
application_base::startup(get_io_context());
application_base::startup();
}

application_t() : application_base(std::make_shared<executor_t>()) {
Expand Down
78 changes: 78 additions & 0 deletions tests/basic_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,17 @@
#include <appbase/application.hpp>
#include <atomic>
#include <chrono>
#include <csignal>
#include <iostream>
#include <string_view>
#include <thread>
#include <future>
#include <boost/exception/diagnostic_information.hpp>

#ifndef _WIN32
#include <unistd.h>
#endif


#define BOOST_TEST_MODULE Basic Tests
#include <boost/test/included/unit_test.hpp>
Expand Down Expand Up @@ -55,12 +62,18 @@ class pluginA : public appbase::plugin<pluginA>
if (shutdown_counter)
++(*shutdown_counter);
}

void handle_sighup() override {
if (sighup_counter)
++(*sighup_counter);
}

uint64_t dbsize() const { return dbsize_; }
bool readonly() const { return readonly_; }

void do_throw(const std::string& msg) { throw std::runtime_error(msg); }
void set_shutdown_counter(uint32_t &c) { shutdown_counter = &c; }
void set_sighup_counter(std::atomic<uint32_t>& c) { sighup_counter = &c; }

void log(std::string_view s) const {
if (log_)
Expand All @@ -75,8 +88,23 @@ class pluginA : public appbase::plugin<pluginA>
bool log_ {false};
uint64_t dbsize_ {0};
uint32_t* shutdown_counter { nullptr };
std::atomic<uint32_t>* sighup_counter { nullptr };
};

/**
* @brief Wait until the condition becomes true or the timeout expires.
*/
template <typename Predicate>
bool wait_for_condition(Predicate&& predicate, std::chrono::milliseconds timeout) {
const auto deadline = std::chrono::steady_clock::now() + timeout;
while(std::chrono::steady_clock::now() < deadline) {
if(predicate())
return true;
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
return predicate();
}

class pluginB : public appbase::plugin<pluginB>
{
public:
Expand Down Expand Up @@ -183,6 +211,56 @@ BOOST_AUTO_TEST_CASE(app_execution)
app_thread.join();
}

#if !defined(_WIN32) && defined(SIGHUP)
// -----------------------------------------------------------------------------
// Check POSIX signal handling: SIGHUP reloads, SIGTERM quits.
// -----------------------------------------------------------------------------
BOOST_AUTO_TEST_CASE(posix_signal_handling)
{
appbase::application::register_plugin<pluginB>();

appbase::scoped_app app;

const char* argv[] = { bu::framework::current_test_case().p_name->c_str() };

BOOST_REQUIRE(app->initialize<pluginB>(sizeof(argv) / sizeof(char*), const_cast<char**>(argv)));

std::atomic<uint32_t> sighup_callback_counter{0};
std::atomic<uint32_t> plugin_sighup_counter{0};
app->set_sighup_callback([&]() { ++sighup_callback_counter; });
app->get_plugin<pluginA>().set_sighup_counter(plugin_sighup_counter);

std::promise<void> startup_promise;
auto startup_fut = startup_promise.get_future();
std::thread app_thread([&]() {
app->startup();
startup_promise.set_value();
app->exec();
});

BOOST_REQUIRE(startup_fut.wait_for(std::chrono::seconds(5)) == std::future_status::ready);

BOOST_REQUIRE(::kill(::getpid(), SIGHUP) == 0);
BOOST_REQUIRE(wait_for_condition(
[&]() {
return sighup_callback_counter.load() == 1 && plugin_sighup_counter.load() == 1;
},
std::chrono::seconds(5)));
BOOST_CHECK(!app->is_quiting());

#ifdef SIGPIPE
BOOST_REQUIRE(::kill(::getpid(), SIGPIPE) == 0);
std::this_thread::sleep_for(std::chrono::milliseconds(100));
BOOST_CHECK(!app->is_quiting());
#endif

BOOST_REQUIRE(::kill(::getpid(), SIGTERM) == 0);
BOOST_REQUIRE(wait_for_condition([&]() { return app->is_quiting(); }, std::chrono::seconds(5)));

app_thread.join();
}
#endif

// -----------------------------------------------------------------------------
// Check application lifetime managed by appbase::scoped_app
// -----------------------------------------------------------------------------
Expand Down
Loading