Commit 15d04e37 authored by Dominik Charousset's avatar Dominik Charousset

Merge pull request #610

parents f214b791 d8dc4018
...@@ -491,6 +491,14 @@ public: ...@@ -491,6 +491,14 @@ public:
/// Blocks the caller until all detached threads are done. /// Blocks the caller until all detached threads are done.
void await_detached_threads(); void await_detached_threads();
/// Calls all thread started hooks
/// @warning must be called by thread which is about to start
void thread_started();
/// Calls all thread terminates hooks
/// @warning must be called by thread which is about to terminate
void thread_terminates();
/// @endcond /// @endcond
private: private:
......
...@@ -30,6 +30,7 @@ ...@@ -30,6 +30,7 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/stream.hpp" #include "caf/stream.hpp"
#include "caf/thread_hook.hpp"
#include "caf/config_value.hpp" #include "caf/config_value.hpp"
#include "caf/config_option.hpp" #include "caf/config_option.hpp"
#include "caf/actor_factory.hpp" #include "caf/actor_factory.hpp"
...@@ -51,6 +52,8 @@ public: ...@@ -51,6 +52,8 @@ public:
using hook_factory_vector = std::vector<hook_factory>; using hook_factory_vector = std::vector<hook_factory>;
using thread_hooks = std::vector<std::unique_ptr<thread_hook>>;
template <class K, class V> template <class K, class V>
using hash_map = std::unordered_map<K, V>; using hash_map = std::unordered_map<K, V>;
...@@ -224,6 +227,13 @@ public: ...@@ -224,6 +227,13 @@ public:
}); });
} }
/// Adds a hook type to the scheduler.
template <class Hook, class... Ts>
actor_system_config& add_thread_hook(Ts&&... ts) {
thread_hooks_.emplace_back(new Hook(std::forward<Ts>(ts)...));
return *this;
}
/// Stores whether the help text for this config object was /// Stores whether the help text for this config object was
/// printed. If set to `true`, the application should not use /// printed. If set to `true`, the application should not use
/// this config object to initialize an `actor_system` and /// this config object to initialize an `actor_system` and
...@@ -310,6 +320,10 @@ public: ...@@ -310,6 +320,10 @@ public:
hook_factory_vector hook_factories; hook_factory_vector hook_factories;
group_module_factory_vector group_module_factories; group_module_factory_vector group_module_factories;
// -- hooks ------------------------------------------------------------------
thread_hooks thread_hooks_;
// -- run-time type information ---------------------------------------------- // -- run-time type information ----------------------------------------------
portable_name_map type_names_by_rtti; portable_name_map type_names_by_rtti;
......
...@@ -56,6 +56,7 @@ ...@@ -56,6 +56,7 @@
#include "caf/exit_reason.hpp" #include "caf/exit_reason.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/thread_hook.hpp"
#include "caf/typed_actor.hpp" #include "caf/typed_actor.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/deserializer.hpp" #include "caf/deserializer.hpp"
......
...@@ -55,7 +55,9 @@ public: ...@@ -55,7 +55,9 @@ public:
CAF_ASSERT(this_thread_.get_id() == std::thread::id{}); CAF_ASSERT(this_thread_.get_id() == std::thread::id{});
auto this_worker = this; auto this_worker = this;
this_thread_ = std::thread{[this_worker] { this_thread_ = std::thread{[this_worker] {
this_worker->system().thread_started();
this_worker->run(); this_worker->run();
this_worker->system().thread_terminates();
}}; }};
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_THREAD_HOOK_HPP
#define CAF_THREAD_HOOK_HPP
namespace caf {
/// Interface to define thread hooks.
class thread_hook {
public:
virtual ~thread_hook() {
// nop
}
/// Called by the actor system once before starting any threads.
virtual void init(actor_system&) = 0;
/// Called whenever the actor system has started a new thread.
/// To access a reference to the started thread use `std::this_thread`.
/// @warning must the thread-safe
virtual void thread_started() = 0;
/// Called whenever a thread is about to quit.
/// To access a reference to the terminating thread use `std::this_thread`.
/// @warning must the thread-safe
virtual void thread_terminates() = 0;
};
} // namespace caf
#endif // CAF_THREAD_HOOK_HPP
...@@ -215,6 +215,8 @@ actor_system::actor_system(actor_system_config& cfg) ...@@ -215,6 +215,8 @@ actor_system::actor_system(actor_system_config& cfg)
cfg_(cfg), cfg_(cfg),
logger_dtor_done_(false) { logger_dtor_done_(false) {
CAF_SET_LOGGER_SYS(this); CAF_SET_LOGGER_SYS(this);
for (auto& hook : cfg.thread_hooks_)
hook->init(*this);
for (auto& f : cfg.module_factories) { for (auto& f : cfg.module_factories) {
auto mod_ptr = f(*this); auto mod_ptr = f(*this);
modules_[mod_ptr->id()].reset(mod_ptr); modules_[mod_ptr->id()].reset(mod_ptr);
...@@ -417,6 +419,16 @@ void actor_system::await_detached_threads() { ...@@ -417,6 +419,16 @@ void actor_system::await_detached_threads() {
detached_cv.wait(guard); detached_cv.wait(guard);
} }
void actor_system::thread_started() {
for (auto& hook : cfg_.thread_hooks_)
hook->thread_started();
}
void actor_system::thread_terminates() {
for (auto& hook : cfg_.thread_hooks_)
hook->thread_terminates();
}
expected<strong_actor_ptr> expected<strong_actor_ptr>
actor_system::dyn_spawn_impl(const std::string& name, message& args, actor_system::dyn_spawn_impl(const std::string& name, message& args,
execution_unit* ctx, bool check_interface, execution_unit* ctx, bool check_interface,
......
...@@ -91,6 +91,7 @@ void blocking_actor::launch(execution_unit*, bool, bool hide) { ...@@ -91,6 +91,7 @@ void blocking_actor::launch(execution_unit*, bool, bool hide) {
home_system().inc_detached_threads(); home_system().inc_detached_threads();
std::thread([](strong_actor_ptr ptr) { std::thread([](strong_actor_ptr ptr) {
// actor lives in its own thread // actor lives in its own thread
ptr->home_system->thread_started();
auto this_ptr = ptr->get(); auto this_ptr = ptr->get();
CAF_ASSERT(dynamic_cast<blocking_actor*>(this_ptr) != 0); CAF_ASSERT(dynamic_cast<blocking_actor*>(this_ptr) != 0);
auto self = static_cast<blocking_actor*>(this_ptr); auto self = static_cast<blocking_actor*>(this_ptr);
...@@ -119,6 +120,7 @@ void blocking_actor::launch(execution_unit*, bool, bool hide) { ...@@ -119,6 +120,7 @@ void blocking_actor::launch(execution_unit*, bool, bool hide) {
# endif # endif
self->cleanup(std::move(rsn), self->context()); self->cleanup(std::move(rsn), self->context());
ptr->home_system->dec_detached_threads(); ptr->home_system->dec_detached_threads();
ptr->home_system->thread_terminates();
}, strong_actor_ptr{ctrl()}).detach(); }, strong_actor_ptr{ctrl()}).detach();
} }
......
...@@ -505,7 +505,11 @@ void logger::start() { ...@@ -505,7 +505,11 @@ void logger::start() {
if (inline_output_) if (inline_output_)
log_first_line(); log_first_line();
else else
thread_ = std::thread{[this] { this->run(); }}; thread_ = std::thread{[this] {
this->system_.thread_started();
this->run();
this->system_.thread_terminates();
}};
#endif #endif
} }
......
...@@ -85,12 +85,14 @@ void private_thread::shutdown() { ...@@ -85,12 +85,14 @@ void private_thread::shutdown() {
} }
void private_thread::exec(private_thread* this_ptr) { void private_thread::exec(private_thread* this_ptr) {
this_ptr->system_.thread_started();
this_ptr->run(); this_ptr->run();
// make sure to not destroy the private thread object before the // make sure to not destroy the private thread object before the
// detached actor is destroyed and this object is unreachable // detached actor is destroyed and this object is unreachable
this_ptr->await_self_destroyed(); this_ptr->await_self_destroyed();
// signalize destruction of detached thread to registry // signalize destruction of detached thread to registry
this_ptr->system_.dec_detached_threads(); this_ptr->system_.dec_detached_threads();
this_ptr->system_.thread_terminates();
// done // done
delete this_ptr; delete this_ptr;
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/config.hpp"
// this suite tests whether actors terminate as expect in several use cases
#define CAF_SUITE thread_hook
#include "caf/all.hpp"
#include "caf/test/unit_test.hpp"
using namespace caf;
namespace {
class test_thread_hooks : public thread_hook {
using atomic_cnt = std::atomic<size_t>;
public:
test_thread_hooks(size_t assumed_init, size_t assumed_thread_started,
size_t assumed_thread_termintes_cb)
: count_init_{0},
count_thread_started_{0},
count_thread_terminates_{0},
assumed_init_{assumed_init},
assumed_thread_started_{assumed_thread_started},
assumed_thread_termintes_cb_{assumed_thread_termintes_cb} {
// nop
}
virtual ~test_thread_hooks() {
CAF_CHECK_EQUAL(count_init_, assumed_init_);
CAF_CHECK_EQUAL(count_thread_started_, assumed_thread_started_);
CAF_CHECK_EQUAL(count_thread_terminates_, assumed_thread_termintes_cb_);
}
virtual void init(actor_system&) {
count_init_.fetch_add(1, std::memory_order_relaxed);
}
virtual void thread_started() {
count_thread_started_.fetch_add(1, std::memory_order_relaxed);
}
virtual void thread_terminates() {
count_thread_terminates_.fetch_add(1, std::memory_order_relaxed);
}
private:
atomic_cnt count_init_;
atomic_cnt count_thread_started_;
atomic_cnt count_thread_terminates_;
atomic_cnt assumed_init_;
atomic_cnt assumed_thread_started_;
atomic_cnt assumed_thread_termintes_cb_;
};
} // namespace <anonymous>
CAF_TEST(test_no_args) {
actor_system_config cfg{};
struct a : public thread_hook {
virtual void init(actor_system&) {}
virtual void thread_started() {}
virtual void thread_terminates() {}
};
cfg.add_thread_hook<a>();
}
CAF_TEST(test_no_system) {
actor_system_config cfg{};
cfg.add_thread_hook<test_thread_hooks>(0,0,0);
}
CAF_TEST(test_system_no_actor) {
actor_system_config cfg{};
size_t assumed_threads = 2 + cfg.scheduler_max_threads;
#ifdef CAF_LOG_LEVEL
assumed_threads += 1; // If logging is enabled, we have a additional thread.
#endif
cfg.add_thread_hook<test_thread_hooks>(1, assumed_threads, assumed_threads);
actor_system system{cfg};
}
...@@ -269,6 +269,7 @@ void middleman::start() { ...@@ -269,6 +269,7 @@ void middleman::start() {
std::condition_variable cv; std::condition_variable cv;
thread_ = std::thread{[&,this] { thread_ = std::thread{[&,this] {
CAF_SET_LOGGER_SYS(&system()); CAF_SET_LOGGER_SYS(&system());
system().thread_started();
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
{ {
std::unique_lock<std::mutex> guard{mtx}; std::unique_lock<std::mutex> guard{mtx};
...@@ -277,6 +278,7 @@ void middleman::start() { ...@@ -277,6 +278,7 @@ void middleman::start() {
cv.notify_one(); cv.notify_one();
} }
backend().run(); backend().run();
system().thread_terminates();
}}; }};
std::unique_lock<std::mutex> guard{mtx}; std::unique_lock<std::mutex> guard{mtx};
while (init_done == false) while (init_done == false)
......
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