Commit 4e70994d authored by Dominik Charousset's avatar Dominik Charousset

Tie lifetime of meta objects table to CAF threads

Fixes undefined behavior if an application leaves `main` while the
threads started by CAF are still running. Usually, an application
destroys the `actor_system` before leaving `main`. However, some
applications may use the `main` function only for spinning up threads.
In this case, the meta objects table must exceed the lifetime of the
static helper object.
parent 46a3c6a4
......@@ -31,6 +31,9 @@ is based on [Keep a Changelog](https://keepachangelog.com).
an error. The `includes` and `excludes` filters are now consistently handled
and accepted in config files as well as on the command line (#1238).
- Silence a deprecated-enum-conversion warning for `std::byte` (#1230).
- Fix heap-use-after-free when accessing the meta objects table in applications
that leave the `main` function while the actor system and its worker threads
are still running (#1241).
## [0.18.1] - 2021-03-19
......
......@@ -12,6 +12,7 @@
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <typeinfo>
#include "caf/abstract_actor.hpp"
......@@ -24,6 +25,7 @@
#include "caf/detail/core_export.hpp"
#include "caf/detail/init_fun_factory.hpp"
#include "caf/detail/private_thread_pool.hpp"
#include "caf/detail/set_thread_name.hpp"
#include "caf/detail/spawn_fwd.hpp"
#include "caf/detail/spawnable.hpp"
#include "caf/fwd.hpp"
......@@ -83,6 +85,20 @@ std::string get_rtti_from_mpi() {
namespace caf {
/// An opaque type for shared object lifetime management of the global meta
/// objects table.
using global_meta_objects_guard_type = intrusive_ptr<ref_counted>;
// Note: for technical reasons (dependencies), `global_meta_objects_guard` is
// implemented in src/detail/meta_object.cpp.
/// Returns a shared ownership wrapper for global state to manage meta objects.
/// Any thread that accesses the actor system should participate in the lifetime
/// management of the global state by using a meta objects guard.
/// @warning The guard does *not* extend the lifetime of the actor system.
/// @relates actor_system
CAF_CORE_EXPORT global_meta_objects_guard_type global_meta_objects_guard();
/// Actor environment including scheduler, registry, and optional components
/// such as a middleman.
class CAF_CORE_EXPORT actor_system {
......@@ -543,6 +559,19 @@ public:
/// @warning must be called by thread which is about to terminate
void thread_terminates();
template <class F>
std::thread launch_thread(const char* thread_name, F fun) {
auto body = [this, thread_name, f{std::move(fun)}](auto guard) {
CAF_IGNORE_UNUSED(guard);
CAF_SET_LOGGER_SYS(this);
detail::set_thread_name(thread_name);
thread_started();
f();
thread_terminates();
};
return std::thread{std::move(body), global_meta_objects_guard()};
}
const auto& metrics_actors_includes() const noexcept {
return metrics_actors_includes_;
}
......
......@@ -25,8 +25,6 @@ public:
private:
void run(actor_system* sys);
static void exec(actor_system* sys, private_thread* this_ptr);
std::pair<resumable*, bool> await();
std::thread thread_;
......
......@@ -12,6 +12,7 @@
#include "caf/detail/core_export.hpp"
#include "caf/detail/make_meta_object.hpp"
#include "caf/detail/meta_object.hpp"
#include "caf/fwd.hpp"
#include "caf/span.hpp"
#include "caf/type_id.hpp"
......
......@@ -60,13 +60,8 @@ protected:
w->start();
// Launch an additional background thread for dispatching timeouts and
// delayed messages.
timer_ = std::thread{[&] {
CAF_SET_LOGGER_SYS(&system());
detail::set_thread_name("caf.clock");
system().thread_started();
clock_.run_dispatch_loop();
system().thread_terminates();
}};
timer_ = system().launch_thread("caf.clock",
[this] { clock_.run_dispatch_loop(); });
// Run remaining startup code.
super::start();
}
......
......@@ -37,14 +37,7 @@ public:
void start() {
CAF_ASSERT(this_thread_.get_id() == std::thread::id{});
auto this_worker = this;
this_thread_ = std::thread{[this_worker] {
CAF_SET_LOGGER_SYS(&this_worker->system());
detail::set_thread_name("caf.worker");
this_worker->system().thread_started();
this_worker->run();
this_worker->system().thread_terminates();
}};
this_thread_ = system().launch_thread("caf.worker", [this] { run(); });
}
worker(const worker&) = delete;
......
......@@ -9,16 +9,19 @@
#include <cstdlib>
#include <cstring>
#include "caf/actor_system.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/config.hpp"
#include "caf/deserializer.hpp"
#include "caf/error.hpp"
#include "caf/error_code.hpp"
#include "caf/make_counted.hpp"
#include "caf/ref_counted.hpp"
#include "caf/serializer.hpp"
#include "caf/span.hpp"
namespace caf::detail {
namespace caf {
#define fatal(str) \
do { \
......@@ -29,20 +32,31 @@ namespace caf::detail {
namespace {
// Stores global type information.
meta_object* meta_objects;
detail::meta_object* meta_objects;
// Stores the size of `meta_objects`.
size_t meta_objects_size;
// Make sure to clean up all meta objects on program exit.
struct meta_objects_cleanup {
~meta_objects_cleanup() {
struct meta_objects_cleanup : ref_counted {
~meta_objects_cleanup() override {
delete[] meta_objects;
}
} cleanup_helper;
};
global_meta_objects_guard_type cleanup_helper
= make_counted<meta_objects_cleanup>();
} // namespace
global_meta_objects_guard_type global_meta_objects_guard() {
return cleanup_helper;
}
} // namespace caf
namespace caf::detail {
span<const meta_object> global_meta_objects() {
return {meta_objects, meta_objects_size};
}
......
......@@ -63,16 +63,10 @@ std::pair<resumable*, bool> private_thread::await() {
private_thread* private_thread::launch(actor_system* sys) {
auto ptr = std::make_unique<private_thread>();
ptr->thread_ = std::thread{exec, sys, ptr.get()};
auto raw_ptr = ptr.get();
ptr->thread_ = sys->launch_thread("caf.thread",
[raw_ptr, sys] { raw_ptr->run(sys); });
return ptr.release();
}
void private_thread::exec(actor_system* sys, private_thread* this_ptr) {
CAF_SET_LOGGER_SYS(sys);
detail::set_thread_name("caf.thread");
sys->thread_started();
this_ptr->run(sys);
sys->thread_terminates();
}
} // namespace caf::detail
......@@ -18,6 +18,7 @@
#include "caf/detail/private_thread_pool.hpp"
#include "caf/actor_system.hpp"
#include "caf/config.hpp"
#include "caf/detail/private_thread.hpp"
......@@ -28,7 +29,7 @@ private_thread_pool::node::~node() {
}
void private_thread_pool::start() {
loop_ = std::thread{[](private_thread_pool* ptr) { ptr->run_loop(); }, this};
loop_ = sys_->launch_thread("caf.pool", [this] { run_loop(); });
}
void private_thread_pool::stop() {
......
......@@ -642,12 +642,16 @@ void logger::start() {
open_file();
log_first_line();
} else {
thread_ = std::thread{[this] {
// Note: we don't call system_->launch_thread here since we don't want to
// set a logger context in the logger thread.
auto f = [this](auto guard) {
CAF_IGNORE_UNUSED(guard);
detail::set_thread_name("caf.logger");
this->system_.thread_started();
this->run();
this->system_.thread_terminates();
}};
};
thread_ = std::thread{f, global_meta_objects_guard()};
}
}
......
......@@ -68,7 +68,7 @@ template <class Hook>
struct config : actor_system_config {
config() {
add_thread_hook<Hook>();
set("logger.verbosity", "quiet");
set("caf.logger.verbosity", "quiet");
}
};
......@@ -103,7 +103,7 @@ CAF_TEST(counting_system_without_actor) {
assumed_init_calls = 1;
auto fallback = scheduler::abstract_coordinator::default_thread_count();
assumed_thread_count = get_or(cfg, "caf.scheduler.max-threads", fallback)
+ 1; // caf.clock
+ 2; // clock and private thread pool
auto& sched = sys.scheduler();
if (sched.detaches_utility_actors())
assumed_thread_count += sched.num_utility_actors();
......@@ -113,7 +113,7 @@ CAF_TEST(counting_system_with_actor) {
assumed_init_calls = 1;
auto fallback = scheduler::abstract_coordinator::default_thread_count();
assumed_thread_count = get_or(cfg, "caf.scheduler.max-threads", fallback)
+ 2; // caf.clock and detached actor
+ 3; // clock, private thread pool, and detached actor
auto& sched = sys.scheduler();
if (sched.detaches_utility_actors())
assumed_thread_count += sched.num_utility_actors();
......@@ -122,4 +122,3 @@ CAF_TEST(counting_system_with_actor) {
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -146,7 +146,8 @@ public:
using impl = detail::prometheus_broker;
actor_config cfg{&mpx_};
broker_ = mpx_.system().spawn_impl<impl, hidden>(cfg, std::move(dptr));
thread_ = std::thread{[this] { mpx_.run(); }};
thread_ = mpx_.system().launch_thread("caf.io.prom",
[this] { mpx_.run(); });
return actual_port;
}
......@@ -424,10 +425,7 @@ void middleman::start() {
std::atomic<bool> init_done{false};
std::mutex mtx;
std::condition_variable cv;
thread_ = std::thread{[&, this] {
CAF_SET_LOGGER_SYS(&system());
detail::set_thread_name("caf.multiplexer");
system().thread_started();
auto run_backend = [this, &mtx, &cv, &init_done] {
CAF_LOG_TRACE("");
{
std::unique_lock<std::mutex> guard{mtx};
......@@ -436,8 +434,8 @@ void middleman::start() {
cv.notify_one();
}
backend().run();
system().thread_terminates();
}};
};
thread_ = system().launch_thread("caf.io.mpx", run_backend);
std::unique_lock<std::mutex> guard{mtx};
while (init_done == false)
cv.wait(guard);
......
......@@ -22,6 +22,7 @@
#include "caf/config_option_adder.hpp"
#include "caf/config_option_set.hpp"
#include "caf/config_value.hpp"
#include "caf/detail/set_thread_name.hpp"
#include "caf/settings.hpp"
#include "caf/string_algorithms.hpp"
#include "caf/test/unit_test.hpp"
......@@ -36,6 +37,7 @@ public:
private:
watchdog(int secs) {
thread_ = std::thread{[=] {
caf::detail::set_thread_name("test.watchdog");
auto tp = std::chrono::high_resolution_clock::now()
+ std::chrono::seconds(secs);
std::unique_lock<std::mutex> guard{mtx_};
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment