Commit 899766cd authored by Dominik Charousset's avatar Dominik Charousset

Add test coordinator and per-actor configs

parent fdd24706
......@@ -88,6 +88,7 @@ set (LIBCAF_CORE_SRCS
src/splitter.cpp
src/sync_request_bouncer.cpp
src/stringification_inspector.cpp
src/test_coordinator.cpp
src/timestamp.cpp
src/try_match.cpp
src/type_erased_value.cpp
......
......@@ -39,6 +39,7 @@
#include "caf/abstract_actor.hpp"
#include "caf/actor_registry.hpp"
#include "caf/string_algorithms.hpp"
#include "caf/named_actor_config.hpp"
#include "caf/scoped_execution_unit.hpp"
#include "caf/uniform_type_info_map.hpp"
#include "caf/composable_behavior_based_actor.hpp"
......@@ -163,6 +164,9 @@ public:
using module_array = std::array<module_ptr, module::num_ids>;
using named_actor_config_map = std::unordered_map<std::string,
named_actor_config>;
/// @warning The system stores a reference to `cfg`, which means the
/// config object must outlive the actor system.
explicit actor_system(actor_system_config& cfg);
......@@ -454,6 +458,11 @@ public:
return cfg_;
}
/// Returns configuration parameters for individual named actors types.
inline const named_actor_config_map& named_actor_configs() const {
return named_actor_configs_;
}
/// @cond PRIVATE
/// Increases running-detached-threads-count by one.
......@@ -521,6 +530,7 @@ private:
std::mutex logger_dtor_mtx_;
std::condition_variable logger_dtor_cv_;
volatile bool logger_dtor_done_;
named_actor_config_map named_actor_configs_;
};
} // namespace caf
......
......@@ -92,7 +92,6 @@
#include "caf/composable_behavior.hpp"
#include "caf/typed_actor_pointer.hpp"
#include "caf/scoped_execution_unit.hpp"
#include "caf/typed_continue_helper.hpp"
#include "caf/typed_response_promise.hpp"
#include "caf/typed_event_based_actor.hpp"
#include "caf/abstract_composable_behavior.hpp"
......@@ -105,6 +104,7 @@
#include "caf/meta/load_callback.hpp"
#include "caf/meta/omittable_if_empty.hpp"
#include "caf/scheduler/test_coordinator.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
/// @author Dominik Charousset <dominik.charousset (at) haw-hamburg.de>
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* 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_NAMED_ACTOR_CONFIG_HPP
#define CAF_NAMED_ACTOR_CONFIG_HPP
#include <cstddef>
#include "caf/atom.hpp"
#include "caf/deep_to_string.hpp"
namespace caf {
/// Stores a flow-control configuration.
struct named_actor_config {
atom_value strategy;
size_t low_watermark;
size_t max_pending;
};
template <class Processor>
void serialize(Processor& proc, named_actor_config& x, const unsigned int) {
proc & x.strategy;
proc & x.low_watermark;
proc & x.max_pending;
}
inline std::string to_string(const named_actor_config& x) {
return "named_actor_config"
+ deep_to_string_as_tuple(x.strategy, x.low_watermark, x.max_pending);
}
} // namespace caf
#endif //CAF_NAMED_ACTOR_CONFIG_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* 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_TEST_COORDINATOR_HPP
#define CAF_TEST_COORDINATOR_HPP
#include "caf/config.hpp"
#include <deque>
#include <chrono>
#include <cstddef>
#include "caf/scheduler/abstract_coordinator.hpp"
namespace caf {
namespace scheduler {
/// Coordinator for testing purposes.
class test_coordinator : public abstract_coordinator {
public:
using super = abstract_coordinator;
test_coordinator(actor_system& sys);
std::deque<resumable*> jobs;
struct delayed_msg {
strong_actor_ptr from;
strong_actor_ptr to;
message_id mid;
message msg;
};
using hrc = std::chrono::high_resolution_clock;
std::multimap<hrc::time_point, delayed_msg> delayed_messages;
template <class T = resumable>
T& next_job() {
if (jobs.empty())
CAF_RAISE_ERROR("jobs.empty())");
return dynamic_cast<T&>(*jobs.front());
}
/// Tries to execute a single event.
bool run_once();
/// Executes events until the job queue is empty and no pending timeouts are
/// left. Returns the number of processed events.
size_t run();
/// Tries to dispatch a single delayed message.
bool dispatch_once();
/// Dispatches all pending delayed messages. Returns the number of dispatched
/// messages.
size_t dispatch();
/// Loops until no job or delayed message remains. Returns the total number of
/// events (first) and dispatched delayed messages (second).
std::pair<size_t, size_t> run_dispatch_loop();
protected:
void start() override;
void stop() override;
void enqueue(resumable* ptr) override;
};
} // namespace scheduler
} // namespace caf
#endif // CAF_TEST_COORDINATOR_HPP
......@@ -30,6 +30,7 @@
#include "caf/policy/work_stealing.hpp"
#include "caf/scheduler/coordinator.hpp"
#include "caf/scheduler/test_coordinator.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/scheduler/profiled_coordinator.hpp"
......@@ -205,6 +206,7 @@ actor_system::actor_system(actor_system_config& cfg)
if (clptr)
opencl_manager_ = reinterpret_cast<opencl::manager*>(clptr->subtype_ptr());
auto& sched = modules_[module::scheduler];
using test = scheduler::test_coordinator;
using share = scheduler::coordinator<policy::work_sharing>;
using steal = scheduler::coordinator<policy::work_stealing>;
using profiled_share = scheduler::profiled_coordinator<policy::work_sharing>;
......@@ -214,6 +216,7 @@ actor_system::actor_system(actor_system_config& cfg)
enum sched_conf {
stealing = 0x0001,
sharing = 0x0002,
testing = 0x0003,
profiled = 0x0100,
profiled_stealing = 0x0101,
profiled_sharing = 0x0102
......@@ -221,6 +224,8 @@ actor_system::actor_system(actor_system_config& cfg)
sched_conf sc = stealing;
if (cfg.scheduler_policy == atom("sharing"))
sc = sharing;
else if (cfg.scheduler_policy == atom("testing"))
sc = testing;
else if (cfg.scheduler_policy != atom("stealing"))
std::cerr << "[WARNING] " << deep_to_string(cfg.scheduler_policy)
<< " is an unrecognized scheduler pollicy, "
......@@ -241,6 +246,8 @@ actor_system::actor_system(actor_system_config& cfg)
case profiled_sharing:
sched.reset(new profiled_share(*this));
break;
case testing:
sched.reset(new test(*this));
}
}
// initialize state for each module and give each module the opportunity
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* 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/scheduler/test_coordinator.hpp"
#include "caf/resumable.hpp"
#include "caf/monitorable_actor.hpp"
namespace caf {
namespace scheduler {
namespace {
class dummy_worker : public execution_unit {
public:
dummy_worker(test_coordinator* parent)
: execution_unit(&parent->system()),
parent_(parent) {
// nop
}
void exec_later(resumable* ptr) override {
parent_->jobs.push_back(ptr);
}
private:
test_coordinator* parent_;
};
class dummy_printer : public monitorable_actor {
public:
dummy_printer(actor_config& cfg) : monitorable_actor(cfg) {
mh_.assign(
[&](add_atom, actor_id, const std::string& str) {
std::cout << str;
}
);
}
void enqueue(mailbox_element_ptr what, execution_unit*) override {
mh_(what->content());
}
private:
message_handler mh_;
};
class dummy_timer : public monitorable_actor {
public:
dummy_timer(actor_config& cfg, test_coordinator* parent)
: monitorable_actor(cfg),
parent_(parent) {
mh_.assign(
[&](const duration& d, strong_actor_ptr& from,
strong_actor_ptr& to, message_id mid, message& msg) {
auto tout = test_coordinator::hrc::now();
tout += d;
using delayed_msg = test_coordinator::delayed_msg;
parent_->delayed_messages.emplace(tout, delayed_msg{std::move(from),
std::move(to), mid,
std::move(msg)});
}
);
}
void enqueue(mailbox_element_ptr what, execution_unit*) override {
mh_(what->content());
}
private:
test_coordinator* parent_;
message_handler mh_;
};
} // namespace <anonymous>
test_coordinator::test_coordinator(actor_system& sys) : super(sys) {
// nop
}
void test_coordinator::start() {
dummy_worker worker{this};
actor_config cfg{&worker};
auto& sys = system();
timer_ = make_actor<dummy_timer, strong_actor_ptr>(
sys.next_actor_id(), sys.node(), &sys, cfg, this);
printer_ = make_actor<dummy_printer, strong_actor_ptr>(
sys.next_actor_id(), sys.node(), &sys, cfg);
}
void test_coordinator::stop() {
run_dispatch_loop();
}
void test_coordinator::enqueue(resumable* ptr) {
jobs.push_back(ptr);
}
bool test_coordinator::run_once() {
if (jobs.empty())
return false;
auto job = jobs.front();
jobs.pop_front();
dummy_worker worker{this};
switch (job->resume(&worker, 1)) {
case resumable::resume_later:
jobs.push_front(job);
break;
case resumable::done:
case resumable::awaiting_message:
intrusive_ptr_release(job);
break;
case resumable::shutdown_execution_unit:
break;
}
return true;
}
size_t test_coordinator::run() {
size_t res = 0;
while (run_once())
++res;
return res;
}
bool test_coordinator::dispatch_once() {
auto i = delayed_messages.begin();
if (i == delayed_messages.end())
return false;
auto& dm = i->second;
dm.to->enqueue(dm.from, dm.mid, std::move(dm.msg), nullptr);
delayed_messages.erase(i);
return true;
}
size_t test_coordinator::dispatch() {
size_t res = 0;
while (dispatch_once())
++res;
return res;
}
std::pair<size_t, size_t> test_coordinator::run_dispatch_loop() {
std::pair<size_t, size_t> res;
size_t i = 0;
do {
auto x = run();
auto y = dispatch();
res.first += x;
res.second += y;
i = x + y;
} while (i > 0);
return res;
}
} // namespace caf
} // namespace scheduler
This diff is collapsed.
......@@ -36,17 +36,20 @@ using ms = std::chrono::milliseconds;
using reset_atom = atom_constant<atom("reset")>;
using timer = typed_actor<reacts_to<reset_atom>>;
timer::behavior_type timer_impl(timer::pointer self) {
auto had_reset = std::make_shared<bool>(false);
struct timer_state {
bool had_reset = false;
};
timer::behavior_type timer_impl(timer::stateful_pointer<timer_state> self) {
self->delayed_send(self, ms(100), reset_atom::value);
return {
[=](reset_atom) {
CAF_MESSAGE("timer reset");
*had_reset = true;
self->state.had_reset = true;
},
after(ms(600)) >> [=] {
CAF_MESSAGE("timer expired");
CAF_REQUIRE(*had_reset);
CAF_REQUIRE(self->state.had_reset);
self->quit();
}
};
......@@ -68,18 +71,34 @@ timer::behavior_type timer_impl2(timer::pointer self) {
};
}
struct fixture {
fixture() : system(cfg) {
// nop
struct config : actor_system_config {
config() {
scheduler_policy = atom("testing");
}
};
actor_system_config cfg;
struct fixture {
config cfg;
actor_system system;
scoped_actor self;
scheduler::test_coordinator& sched;
fixture()
: system(cfg),
self(system),
sched(dynamic_cast<scheduler::test_coordinator&>(system.scheduler())) {
CAF_REQUIRE(sched.jobs.empty());
CAF_REQUIRE(sched.delayed_messages.empty());
}
~fixture() {
sched.run_dispatch_loop();
}
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(timeout_tests, fixture)
CAF_TEST_FIXTURE_SCOPE(simple_timeout_tests, fixture)
CAF_TEST(duration_conversion) {
duration d1{time_unit::milliseconds, 100};
......
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