Commit 35405ce6 authored by Dominik Charousset's avatar Dominik Charousset

Streamline thread_hook unit test

parent 15d04e37
...@@ -30,6 +30,7 @@ ...@@ -30,6 +30,7 @@
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/duration.hpp" #include "caf/duration.hpp"
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/actor_cast.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
namespace caf { namespace caf {
...@@ -41,10 +42,28 @@ namespace scheduler { ...@@ -41,10 +42,28 @@ namespace scheduler {
/// chosen workers. /// chosen workers.
class abstract_coordinator : public actor_system::module { class abstract_coordinator : public actor_system::module {
public: public:
enum utility_actor_id : size_t {
printer_id,
timer_id,
max_id
};
explicit abstract_coordinator(actor_system& sys); explicit abstract_coordinator(actor_system& sys);
/// Returns a handle to the central printing actor. /// Returns a handle to the central printing actor.
actor printer() const; inline actor printer() const {
return actor_cast<actor>(utility_actors_[printer_id]);
}
/// Returns a handle to the central timer actor.
inline actor timer() const {
return actor_cast<actor>(utility_actors_[timer_id]);
}
/// Returns the number of utility actors.
inline size_t num_utility_actors() const {
return utility_actors_.size();
}
/// Puts `what` into the queue of a randomly chosen worker. /// Puts `what` into the queue of a randomly chosen worker.
virtual void enqueue(resumable* what) = 0; virtual void enqueue(resumable* what) = 0;
...@@ -52,10 +71,11 @@ public: ...@@ -52,10 +71,11 @@ public:
template <class Duration, class... Data> template <class Duration, class... Data>
void delayed_send(Duration rel_time, strong_actor_ptr from, void delayed_send(Duration rel_time, strong_actor_ptr from,
strong_actor_ptr to, message_id mid, message data) { strong_actor_ptr to, message_id mid, message data) {
timer_->enqueue(nullptr, invalid_message_id, auto& dest = utility_actors_[timer_id];
make_message(duration{rel_time}, std::move(from), dest->enqueue(nullptr, invalid_message_id,
std::move(to), mid, std::move(data)), make_message(duration{rel_time}, std::move(from),
nullptr); std::move(to), mid, std::move(data)),
nullptr);
} }
inline actor_system& system() { inline actor_system& system() {
...@@ -70,6 +90,9 @@ public: ...@@ -70,6 +90,9 @@ public:
return num_workers_; return num_workers_;
} }
/// Returns `true` if this scheduler detaches its utility actors.
virtual bool detaches_utility_actors() const;
void start() override; void start() override;
void init(actor_system_config& cfg) override; void init(actor_system_config& cfg) override;
...@@ -92,8 +115,7 @@ protected: ...@@ -92,8 +115,7 @@ protected:
// configured number of workers // configured number of workers
size_t num_workers_; size_t num_workers_;
strong_actor_ptr timer_; std::array<actor, max_id> utility_actors_;
strong_actor_ptr printer_;
actor_system& system_; actor_system& system_;
}; };
......
...@@ -141,6 +141,8 @@ public: ...@@ -141,6 +141,8 @@ public:
/// hook. /// hook.
void inline_all_enqueues(); void inline_all_enqueues();
bool detaches_utility_actors() const override;
protected: protected:
void start() override; void start() override;
......
...@@ -315,16 +315,16 @@ public: ...@@ -315,16 +315,16 @@ public:
* implementation of coordinator * * implementation of coordinator *
******************************************************************************/ ******************************************************************************/
actor abstract_coordinator::printer() const { bool abstract_coordinator::detaches_utility_actors() const {
CAF_ASSERT(printer_ != nullptr); return true;
return actor_cast<actor>(printer_);
} }
void abstract_coordinator::start() { void abstract_coordinator::start() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// launch utility actors // launch utility actors
timer_ = actor_cast<strong_actor_ptr>(system_.spawn<timer_actor, hidden + detached>()); static constexpr auto fs = hidden + detached;
printer_ = actor_cast<strong_actor_ptr>(system_.spawn<printer_actor, hidden + detached>()); utility_actors_[timer_id] = system_.spawn<timer_actor, fs>();
utility_actors_[printer_id] = system_.spawn<printer_actor, fs>();
} }
void abstract_coordinator::init(actor_system_config& cfg) { void abstract_coordinator::init(actor_system_config& cfg) {
...@@ -343,9 +343,9 @@ void* abstract_coordinator::subtype_ptr() { ...@@ -343,9 +343,9 @@ void* abstract_coordinator::subtype_ptr() {
void abstract_coordinator::stop_actors() { void abstract_coordinator::stop_actors() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
scoped_actor self{system_, true}; scoped_actor self{system_, true};
anon_send_exit(timer_, exit_reason::user_shutdown); for (auto& x : utility_actors_)
anon_send_exit(printer_, exit_reason::user_shutdown); anon_send_exit(x, exit_reason::user_shutdown);
self->wait_for(timer_, printer_); self->wait_for(utility_actors_);
} }
abstract_coordinator::abstract_coordinator(actor_system& sys) abstract_coordinator::abstract_coordinator(actor_system& sys)
......
...@@ -95,13 +95,18 @@ private: ...@@ -95,13 +95,18 @@ private:
test_coordinator::test_coordinator(actor_system& sys) : super(sys) { test_coordinator::test_coordinator(actor_system& sys) : super(sys) {
// nop // nop
} }
bool test_coordinator::detaches_utility_actors() const {
return false;
}
void test_coordinator::start() { void test_coordinator::start() {
dummy_worker worker{this}; dummy_worker worker{this};
actor_config cfg{&worker}; actor_config cfg{&worker};
auto& sys = system(); auto& sys = system();
timer_ = make_actor<dummy_timer, strong_actor_ptr>( utility_actors_[timer_id] = make_actor<dummy_timer, actor>(
sys.next_actor_id(), sys.node(), &sys, cfg, this); sys.next_actor_id(), sys.node(), &sys, cfg, this);
printer_ = make_actor<dummy_printer, strong_actor_ptr>( utility_actors_[printer_id] = make_actor<dummy_printer, actor>(
sys.next_actor_id(), sys.node(), &sys, cfg); sys.next_actor_id(), sys.node(), &sys, cfg);
} }
......
...@@ -29,65 +29,110 @@ using namespace caf; ...@@ -29,65 +29,110 @@ using namespace caf;
namespace { namespace {
class test_thread_hooks : public thread_hook { using atomic_count = std::atomic<size_t>;
using atomic_cnt = std::atomic<size_t>;
size_t assumed_thread_count;
size_t assumed_init_calls;
struct dummy_thread_hook : thread_hook {
void init(actor_system&) override {
// nop
}
void thread_started() override {
// nop
}
void thread_terminates() override {
// nop
}
};
class counting_thread_hook : public thread_hook {
public: public:
test_thread_hooks(size_t assumed_init, size_t assumed_thread_started, counting_thread_hook()
size_t assumed_thread_termintes_cb) : count_init_{0},
: count_init_{0}, count_thread_started_{0},
count_thread_started_{0}, count_thread_terminates_{0} {
count_thread_terminates_{0},
assumed_init_{assumed_init},
assumed_thread_started_{assumed_thread_started},
assumed_thread_termintes_cb_{assumed_thread_termintes_cb} {
// nop // nop
} }
virtual ~test_thread_hooks() {
CAF_CHECK_EQUAL(count_init_, assumed_init_); ~counting_thread_hook() override {
CAF_CHECK_EQUAL(count_thread_started_, assumed_thread_started_); CAF_CHECK_EQUAL(count_init_, assumed_init_calls);
CAF_CHECK_EQUAL(count_thread_terminates_, assumed_thread_termintes_cb_); CAF_CHECK_EQUAL(count_thread_started_, assumed_thread_count);
CAF_CHECK_EQUAL(count_thread_terminates_, assumed_thread_count);
} }
virtual void init(actor_system&) {
count_init_.fetch_add(1, std::memory_order_relaxed); void init(actor_system&) override {
++count_init_;
} }
virtual void thread_started() {
count_thread_started_.fetch_add(1, std::memory_order_relaxed); void thread_started() override {
++count_thread_started_;
} }
virtual void thread_terminates() {
count_thread_terminates_.fetch_add(1, std::memory_order_relaxed); void thread_terminates() override {
++count_thread_terminates_;
} }
private: private:
atomic_cnt count_init_; atomic_count count_init_;
atomic_cnt count_thread_started_; atomic_count count_thread_started_;
atomic_cnt count_thread_terminates_; atomic_count count_thread_terminates_;
atomic_cnt assumed_init_; };
atomic_cnt assumed_thread_started_;
atomic_cnt assumed_thread_termintes_cb_; template <class Hook>
struct config : actor_system_config {
config() {
add_thread_hook<Hook>();
logger_verbosity = atom("quiet");
}
};
template <class Hook>
struct fixture {
config<Hook> cfg;
actor_system sys;
fixture() : sys(cfg) {
// nop
}
}; };
} // namespace <anonymous> } // namespace <anonymous>
CAF_TEST(test_no_args) { CAF_TEST(counting_no_system) {
actor_system_config cfg{}; assumed_init_calls = 0;
struct a : public thread_hook { actor_system_config cfg;
virtual void init(actor_system&) {} cfg.add_thread_hook<counting_thread_hook>();
virtual void thread_started() {}
virtual void thread_terminates() {}
};
cfg.add_thread_hook<a>();
} }
CAF_TEST(test_no_system) { CAF_TEST_FIXTURE_SCOPE(dummy_hook, fixture<dummy_thread_hook>)
actor_system_config cfg{};
cfg.add_thread_hook<test_thread_hooks>(0,0,0); CAF_TEST(counting_no_args) {
// nop
} }
CAF_TEST(test_system_no_actor) { CAF_TEST_FIXTURE_SCOPE_END()
actor_system_config cfg{};
size_t assumed_threads = 2 + cfg.scheduler_max_threads; CAF_TEST_FIXTURE_SCOPE(counting_hook, fixture<counting_thread_hook>)
#ifdef CAF_LOG_LEVEL
assumed_threads += 1; // If logging is enabled, we have a additional thread. CAF_TEST(counting_system_without_actor) {
#endif assumed_init_calls = 1;
cfg.add_thread_hook<test_thread_hooks>(1, assumed_threads, assumed_threads); assumed_thread_count = cfg.scheduler_max_threads;
actor_system system{cfg}; auto& sched = sys.scheduler();
if (sched.detaches_utility_actors())
assumed_thread_count += sched.num_utility_actors();
} }
CAF_TEST(counting_system_with_actor) {
assumed_init_calls = 1;
assumed_thread_count = cfg.scheduler_max_threads + 1;
auto& sched = sys.scheduler();
if (sched.detaches_utility_actors())
assumed_thread_count += sched.num_utility_actors();
sys.spawn<detached>([] {});
sys.spawn([] {});
}
CAF_TEST_FIXTURE_SCOPE_END()
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