Unverified Commit 2134a261 authored by Noir's avatar Noir Committed by GitHub

Merge pull request #1242

Tie lifetime of meta objects table to CAF threads
parents 46a3c6a4 01e8f098
......@@ -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"
......@@ -543,6 +545,23 @@ 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), meta_objects_guard_};
}
auto meta_objects_guard() const noexcept {
return meta_objects_guard_;
}
const auto& metrics_actors_includes() const noexcept {
return metrics_actors_includes_;
}
......@@ -722,6 +741,9 @@ private:
/// Manages threads for detached actors.
detail::private_thread_pool private_threads_;
/// Ties the lifetime of the meta objects table to the actor system.
detail::global_meta_objects_guard_type meta_objects_guard_;
};
} // namespace caf
......@@ -240,9 +240,18 @@ struct IUnknown;
// Convenience macros.
#define CAF_IGNORE_UNUSED(x) static_cast<void>(x)
/// Prints `error` to `stderr` and aborts program execution.
#define CAF_CRITICAL(error) \
do { \
fprintf(stderr, "[FATAL] %s:%u: critical error: '%s'\n", __FILE__, \
fprintf(stderr, "[FATAL] critical error (%s:%d): %s\n", __FILE__, \
__LINE__, error); \
::abort(); \
} while (false)
/// Prints `error` to `stderr` and aborts program execution.
#define CAF_CRITICAL_FMT(fmt_str, ...) \
do { \
fprintf(stderr, "[FATAL] critical error (%s:%d): " fmt_str "\n", __FILE__, \
__LINE__, __VA_ARGS__); \
::abort(); \
} while (false)
......@@ -51,6 +51,15 @@ struct meta_object {
void (*stringify)(std::string&, const void*);
};
/// An opaque type for shared object lifetime management of the global meta
/// objects table.
using global_meta_objects_guard_type = intrusive_ptr<ref_counted>;
/// 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.
CAF_CORE_EXPORT global_meta_objects_guard_type global_meta_objects_guard();
/// Returns the global storage for all meta objects. The ::type_id of an object
/// is the index for accessing the corresonding meta object.
CAF_CORE_EXPORT span<const meta_object> global_meta_objects();
......
......@@ -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_;
......
......@@ -361,6 +361,8 @@ CAF_CORE_EXPORT void intrusive_ptr_release(const dynamic_message_data*);
CAF_CORE_EXPORT dynamic_message_data*
intrusive_cow_ptr_unshare(dynamic_message_data*&);
using global_meta_objects_guard_type = intrusive_ptr<ref_counted>;
} // namespace detail
// -- weak pointer aliases -----------------------------------------------------
......
......@@ -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;
......
......@@ -277,6 +277,9 @@ actor_system::actor_system(actor_system_config& cfg)
tracing_context_(cfg.tracing_context),
private_threads_(this) {
CAF_SET_LOGGER_SYS(this);
meta_objects_guard_ = detail::global_meta_objects_guard();
if (!meta_objects_guard_)
CAF_CRITICAL("unable to obtain the global meta objects guard");
for (auto& hook : cfg.thread_hooks_)
hook->init(*this);
// Cache some configuration parameters for faster lookups at runtime.
......
......@@ -9,40 +9,44 @@
#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 {
#define fatal(str) \
do { \
fprintf(stderr, "FATAL: " str "\n"); \
abort(); \
} while (false)
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;
}
span<const meta_object> global_meta_objects() {
return {meta_objects, meta_objects_size};
}
......@@ -65,8 +69,8 @@ void clear_global_meta_objects() {
span<meta_object> resize_global_meta_objects(size_t size) {
if (size <= meta_objects_size)
fatal("resize_global_meta_objects called with a new size that does not "
"grow the array");
CAF_CRITICAL("resize_global_meta_objects called with a new size that does "
"not grow the array");
auto new_storage = new meta_object[size];
std::copy(meta_objects, meta_objects + meta_objects_size, new_storage);
delete[] meta_objects;
......@@ -79,9 +83,9 @@ void set_global_meta_objects(type_id_t first_id, span<const meta_object> xs) {
auto new_size = first_id + xs.size();
if (first_id < meta_objects_size) {
if (new_size > meta_objects_size)
fatal("set_global_meta_objects called with "
"'first_id < meta_objects_size' and "
"'new_size > meta_objects_size'");
CAF_CRITICAL("set_global_meta_objects called with "
"'first_id < meta_objects_size' and "
"'new_size > meta_objects_size'");
auto out = meta_objects + first_id;
for (const auto& x : xs) {
if (out->type_name.empty()) {
......@@ -94,12 +98,10 @@ void set_global_meta_objects(type_id_t first_id, span<const meta_object> xs) {
// Get null-terminated strings.
auto name1 = to_string(out->type_name);
auto name2 = to_string(x.type_name);
fprintf(stderr,
"FATAL: type ID %d already assigned to %s (tried to override "
"with %s)\n",
static_cast<int>(std::distance(meta_objects, out)),
name1.c_str(), name2.c_str());
abort();
CAF_CRITICAL_FMT("type ID %d already assigned to %s "
"(tried to override with %s)",
static_cast<int>(std::distance(meta_objects, out)),
name1.c_str(), name2.c_str());
}
++out;
}
......
......@@ -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() {
......
......@@ -20,6 +20,7 @@
#include "caf/config.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/get_process_id.hpp"
#include "caf/detail/meta_object.hpp"
#include "caf/detail/pretty_type_name.hpp"
#include "caf/detail/set_thread_name.hpp"
#include "caf/intrusive/task_result.hpp"
......@@ -642,12 +643,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();
}};
system_.thread_started();
run();
system_.thread_terminates();
};
thread_ = std::thread{f, detail::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