Commit 0498f5c3 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'issue/1341'

Close #1341.
parents 2aeb1e8b c8441bf8
...@@ -35,6 +35,9 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -35,6 +35,9 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- Remote spawning of actors is no longer considered experimental. - Remote spawning of actors is no longer considered experimental.
- The output of `--dump-config` now prints valid config file syntax. - The output of `--dump-config` now prints valid config file syntax.
- When starting a new thread via CAF, the thread hooks API now receives an
additional tag that identifies the component responsible for launching the new
thread.
### Deprecated ### Deprecated
......
...@@ -64,6 +64,7 @@ caf_add_component( ...@@ -64,6 +64,7 @@ caf_add_component(
message_priority message_priority
pec pec
sec sec
thread_owner
HEADERS HEADERS
${CAF_CORE_HEADERS} ${CAF_CORE_HEADERS}
SOURCES SOURCES
......
...@@ -536,19 +536,19 @@ public: ...@@ -536,19 +536,19 @@ public:
/// Calls all thread started hooks /// Calls all thread started hooks
/// @warning must be called by thread which is about to start /// @warning must be called by thread which is about to start
void thread_started(); void thread_started(thread_owner);
/// Calls all thread terminates hooks /// Calls all thread terminates hooks
/// @warning must be called by thread which is about to terminate /// @warning must be called by thread which is about to terminate
void thread_terminates(); void thread_terminates();
template <class F> template <class F>
std::thread launch_thread(const char* thread_name, F fun) { std::thread launch_thread(const char* thread_name, thread_owner tag, F fun) {
auto body = [this, thread_name, f{std::move(fun)}](auto guard) { auto body = [this, thread_name, tag, f{std::move(fun)}](auto guard) {
CAF_IGNORE_UNUSED(guard); CAF_IGNORE_UNUSED(guard);
CAF_SET_LOGGER_SYS(this); CAF_SET_LOGGER_SYS(this);
detail::set_thread_name(thread_name); detail::set_thread_name(thread_name);
thread_started(); thread_started(tag);
f(); f();
thread_terminates(); thread_terminates();
}; };
......
...@@ -178,6 +178,7 @@ enum class exit_reason : uint8_t; ...@@ -178,6 +178,7 @@ enum class exit_reason : uint8_t;
enum class invoke_message_result; enum class invoke_message_result;
enum class pec : uint8_t; enum class pec : uint8_t;
enum class sec : uint8_t; enum class sec : uint8_t;
enum class thread_owner;
// -- aliases ------------------------------------------------------------------ // -- aliases ------------------------------------------------------------------
......
...@@ -11,6 +11,7 @@ ...@@ -11,6 +11,7 @@
#include "caf/execution_unit.hpp" #include "caf/execution_unit.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/resumable.hpp" #include "caf/resumable.hpp"
#include "caf/thread_owner.hpp"
namespace caf::scheduler { namespace caf::scheduler {
...@@ -37,7 +38,8 @@ public: ...@@ -37,7 +38,8 @@ public:
void start() { void start() {
CAF_ASSERT(this_thread_.get_id() == std::thread::id{}); CAF_ASSERT(this_thread_.get_id() == std::thread::id{});
this_thread_ = system().launch_thread("caf.worker", [this] { run(); }); this_thread_ = system().launch_thread("caf.worker", thread_owner::scheduler,
[this] { run(); });
} }
worker(const worker&) = delete; worker(const worker&) = delete;
......
...@@ -18,10 +18,11 @@ public: ...@@ -18,10 +18,11 @@ public:
/// Called by the actor system once before starting any threads. /// Called by the actor system once before starting any threads.
virtual void init(actor_system&) = 0; virtual void init(actor_system&) = 0;
/// Called whenever the actor system has started a new thread. /// Called whenever the actor system has started a new thread. To access a
/// To access a reference to the started thread use `std::this_thread`. /// reference to the started thread use `std::this_thread`.
/// @param owner Identifies the CAF component that created this thread.
/// @warning must the thread-safe /// @warning must the thread-safe
virtual void thread_started() = 0; virtual void thread_started(thread_owner owner) = 0;
/// Called whenever a thread is about to quit. /// Called whenever a thread is about to quit.
/// To access a reference to the terminating thread use `std::this_thread`. /// To access a reference to the terminating thread use `std::this_thread`.
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/core_export.hpp"
#include <type_traits>
namespace caf {
/// Denotes the component that launched a CAF thread.
enum class thread_owner {
/// Indicates that the thread belongs to the cooperative scheduler.
scheduler,
/// Indicates that the thread belongs to the internal thread pool for detached
/// and blocking actors.
pool,
/// Indicates that the thread runs background activity such as logging for the
/// actor system.
system,
/// Indicates that the thread has been launched by request of a user without
/// using any of the default mechanisms above.
other,
};
/// @relates sec
CAF_CORE_EXPORT std::string to_string(thread_owner);
/// @relates thread_owner
CAF_CORE_EXPORT bool from_string(std::string_view, thread_owner&);
/// @relates thread_owner
CAF_CORE_EXPORT bool from_integer(std::underlying_type_t<thread_owner>,
thread_owner&);
/// @relates sec
template <class Inspector>
bool inspect(Inspector& f, thread_owner& x) {
return default_enum_inspect(f, x);
}
} // namespace caf
...@@ -499,9 +499,9 @@ size_t actor_system::detached_actors() const noexcept { ...@@ -499,9 +499,9 @@ size_t actor_system::detached_actors() const noexcept {
return private_threads_.running(); return private_threads_.running();
} }
void actor_system::thread_started() { void actor_system::thread_started(thread_owner owner) {
for (auto& hook : cfg_.thread_hooks_) for (auto& hook : cfg_.thread_hooks_)
hook->thread_started(); hook->thread_started(owner);
} }
void actor_system::thread_terminates() { void actor_system::thread_terminates() {
......
...@@ -10,6 +10,7 @@ ...@@ -10,6 +10,7 @@
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/resumable.hpp" #include "caf/resumable.hpp"
#include "caf/scoped_execution_unit.hpp" #include "caf/scoped_execution_unit.hpp"
#include "caf/thread_owner.hpp"
namespace caf::detail { namespace caf::detail {
...@@ -64,7 +65,7 @@ std::pair<resumable*, bool> private_thread::await() { ...@@ -64,7 +65,7 @@ std::pair<resumable*, bool> private_thread::await() {
private_thread* private_thread::launch(actor_system* sys) { private_thread* private_thread::launch(actor_system* sys) {
auto ptr = std::make_unique<private_thread>(); auto ptr = std::make_unique<private_thread>();
auto raw_ptr = ptr.get(); auto raw_ptr = ptr.get();
ptr->thread_ = sys->launch_thread("caf.thread", ptr->thread_ = sys->launch_thread("caf.thread", thread_owner::pool,
[raw_ptr, sys] { raw_ptr->run(sys); }); [raw_ptr, sys] { raw_ptr->run(sys); });
return ptr.release(); return ptr.release();
} }
......
...@@ -21,6 +21,7 @@ ...@@ -21,6 +21,7 @@
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/detail/private_thread.hpp" #include "caf/detail/private_thread.hpp"
#include "caf/thread_owner.hpp"
namespace caf::detail { namespace caf::detail {
...@@ -29,7 +30,8 @@ private_thread_pool::node::~node() { ...@@ -29,7 +30,8 @@ private_thread_pool::node::~node() {
} }
void private_thread_pool::start() { void private_thread_pool::start() {
loop_ = sys_->launch_thread("caf.pool", [this] { run_loop(); }); loop_ = sys_->launch_thread("caf.pool", thread_owner::pool,
[this] { run_loop(); });
} }
void private_thread_pool::stop() { void private_thread_pool::stop() {
......
...@@ -9,6 +9,7 @@ ...@@ -9,6 +9,7 @@
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/system_messages.hpp" #include "caf/system_messages.hpp"
#include "caf/thread_owner.hpp"
namespace caf::detail { namespace caf::detail {
...@@ -52,7 +53,8 @@ void thread_safe_actor_clock::run() { ...@@ -52,7 +53,8 @@ void thread_safe_actor_clock::run() {
} }
void thread_safe_actor_clock::start_dispatch_loop(caf::actor_system& sys) { void thread_safe_actor_clock::start_dispatch_loop(caf::actor_system& sys) {
dispatcher_ = sys.launch_thread("caf.clock", [this] { run(); }); dispatcher_ = sys.launch_thread("caf.clock", thread_owner::system,
[this] { run(); });
} }
void thread_safe_actor_clock::stop_dispatch_loop() { void thread_safe_actor_clock::stop_dispatch_loop() {
......
...@@ -28,6 +28,7 @@ ...@@ -28,6 +28,7 @@
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/string_algorithms.hpp" #include "caf/string_algorithms.hpp"
#include "caf/term.hpp" #include "caf/term.hpp"
#include "caf/thread_owner.hpp"
#include "caf/timestamp.hpp" #include "caf/timestamp.hpp"
namespace caf { namespace caf {
...@@ -633,7 +634,7 @@ void logger::start() { ...@@ -633,7 +634,7 @@ void logger::start() {
auto f = [this](auto guard) { auto f = [this](auto guard) {
CAF_IGNORE_UNUSED(guard); CAF_IGNORE_UNUSED(guard);
detail::set_thread_name("caf.logger"); detail::set_thread_name("caf.logger");
system_.thread_started(); system_.thread_started(thread_owner::system);
run(); run();
system_.thread_terminates(); system_.thread_terminates();
}; };
......
...@@ -22,7 +22,7 @@ struct dummy_thread_hook : thread_hook { ...@@ -22,7 +22,7 @@ struct dummy_thread_hook : thread_hook {
// nop // nop
} }
void thread_started() override { void thread_started(thread_owner) override {
// nop // nop
} }
...@@ -48,7 +48,7 @@ public: ...@@ -48,7 +48,7 @@ public:
++count_init_; ++count_init_;
} }
void thread_started() override { void thread_started(thread_owner) override {
++count_thread_started_; ++count_thread_started_;
} }
......
...@@ -42,6 +42,7 @@ ...@@ -42,6 +42,7 @@
#include "caf/scoped_actor.hpp" #include "caf/scoped_actor.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/thread_owner.hpp"
#include "caf/typed_event_based_actor.hpp" #include "caf/typed_event_based_actor.hpp"
#ifdef CAF_WINDOWS #ifdef CAF_WINDOWS
...@@ -149,7 +150,8 @@ public: ...@@ -149,7 +150,8 @@ public:
sync_ptr->count_down(); sync_ptr->count_down();
mpx_.run(); mpx_.run();
}; };
thread_ = mpx_.system().launch_thread("caf.io.prom", run_mpx); thread_ = mpx_.system().launch_thread("caf.io.prom", thread_owner::system,
run_mpx);
sync.wait(); sync.wait();
CAF_LOG_INFO("expose Prometheus metrics at port" << actual_port); CAF_LOG_INFO("expose Prometheus metrics at port" << actual_port);
return actual_port; return actual_port;
...@@ -448,7 +450,8 @@ void middleman::start() { ...@@ -448,7 +450,8 @@ void middleman::start() {
sync_ptr->count_down(); sync_ptr->count_down();
backend().run(); backend().run();
}; };
thread_ = system().launch_thread("caf.io.mpx", run_backend); thread_ = system().launch_thread("caf.io.mpx", thread_owner::system,
run_backend);
sync.wait(); sync.wait();
} }
// Spawn utility actors. // Spawn utility actors.
......
...@@ -13,6 +13,7 @@ ...@@ -13,6 +13,7 @@
#include "caf/raise_error.hpp" #include "caf/raise_error.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/thread_owner.hpp"
#include "caf/uri.hpp" #include "caf/uri.hpp"
namespace caf::net { namespace caf::net {
...@@ -66,11 +67,12 @@ middleman::~middleman() { ...@@ -66,11 +67,12 @@ middleman::~middleman() {
void middleman::start() { void middleman::start() {
if (!get_or(config(), "caf.middleman.manual-multiplexing", false)) { if (!get_or(config(), "caf.middleman.manual-multiplexing", false)) {
mpx_thread_ = sys_.launch_thread("caf.net.mpx", [this] { auto fn = [this] {
mpx_->set_thread_id(); mpx_->set_thread_id();
launch_background_tasks(sys_); launch_background_tasks(sys_);
mpx_->run(); mpx_->run();
}); };
mpx_thread_ = sys_.launch_thread("caf.net.mpx", thread_owner::system, fn);
} else { } else {
mpx_->set_thread_id(); mpx_->set_thread_id();
} }
......
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