Commit 76eceb2e authored by Dominik Charousset's avatar Dominik Charousset

scheduler fine-tuning + logging improvements

the scheduler no longer runs event-based actors after spawning them
without any message in their mailbox;
this patch also improves logging in the scheduler implementation
to give better insights into the work-stealing workflow
parent 4126af45
...@@ -33,11 +33,32 @@ ...@@ -33,11 +33,32 @@
namespace cppa { namespace cppa {
/**
* @brief Denotes the state of a cooperatively scheduled actor.
*/
enum class actor_state : int { enum class actor_state : int {
/**
* @brief Indicates that the actor is either waiting to be executed
* or currently running.
*/
ready, ready,
/**
* @brief Indicates that the actor finished exection.
*/
done, done,
/**
* @brief Indicates that the actor awaits a new message.
*/
blocked, blocked,
pending,
/**
* @brief Indicates that the actor is about to change its state to
* {@link blocked}, but still can be interrupted by an
* incoming message.
*/
about_to_block about_to_block
}; };
......
...@@ -48,7 +48,7 @@ class functor_based_actor : public event_based_actor { ...@@ -48,7 +48,7 @@ class functor_based_actor : public event_based_actor {
typedef std::function<void(event_based_actor*)> void_fun; typedef std::function<void(event_based_actor*)> void_fun;
template<typename F, typename... Ts> template<typename F, typename... Ts>
functor_based_actor(F f, Ts&&... vs) { functor_based_actor(F f, Ts&&... vs) : m_void_impl(false) {
typedef typename util::get_callable_trait<F>::type trait; typedef typename util::get_callable_trait<F>::type trait;
typedef typename trait::arg_types arg_types; typedef typename trait::arg_types arg_types;
typedef typename trait::result_type result_type; typedef typename trait::result_type result_type;
...@@ -80,6 +80,7 @@ class functor_based_actor : public event_based_actor { ...@@ -80,6 +80,7 @@ class functor_based_actor : public event_based_actor {
template<typename F> template<typename F>
void set(std::false_type, std::true_type, F fun) { void set(std::false_type, std::true_type, F fun) {
// void (pointer) // void (pointer)
m_void_impl = true;
m_make_behavior = [fun](pointer ptr) { m_make_behavior = [fun](pointer ptr) {
fun(ptr); fun(ptr);
return behavior{}; return behavior{};
...@@ -95,6 +96,7 @@ class functor_based_actor : public event_based_actor { ...@@ -95,6 +96,7 @@ class functor_based_actor : public event_based_actor {
template<typename F> template<typename F>
void set(std::false_type, std::false_type, F fun) { void set(std::false_type, std::false_type, F fun) {
// void (void) // void (void)
m_void_impl = true;
m_make_behavior = [fun](pointer) { m_make_behavior = [fun](pointer) {
fun(); fun();
return behavior{}; return behavior{};
...@@ -116,6 +118,7 @@ class functor_based_actor : public event_based_actor { ...@@ -116,6 +118,7 @@ class functor_based_actor : public event_based_actor {
std::forward<Ts>(args)...)); std::forward<Ts>(args)...));
} }
bool m_void_impl;
make_behavior_fun m_make_behavior; make_behavior_fun m_make_behavior;
}; };
......
...@@ -191,6 +191,7 @@ class proper_actor : public proper_actor_base<Base, ...@@ -191,6 +191,7 @@ class proper_actor : public proper_actor_base<Base,
inline void launch(bool is_hidden, execution_unit* host) { inline void launch(bool is_hidden, execution_unit* host) {
CPPA_LOG_TRACE(""); CPPA_LOG_TRACE("");
this->hidden(is_hidden); this->hidden(is_hidden);
this->m_host = host; // may be accessed during make_behavior call
auto bhvr = this->make_behavior(); auto bhvr = this->make_behavior();
if (bhvr) this->become(std::move(bhvr)); if (bhvr) this->become(std::move(bhvr));
CPPA_LOG_WARNING_IF(this->bhvr_stack().empty(), CPPA_LOG_WARNING_IF(this->bhvr_stack().empty(),
......
...@@ -58,6 +58,10 @@ class event_based_actor : public extend<local_actor, event_based_actor>:: ...@@ -58,6 +58,10 @@ class event_based_actor : public extend<local_actor, event_based_actor>::
behavior_stack_based<behavior>::impl, behavior_stack_based<behavior>::impl,
sync_sender<nonblocking_response_handle_tag>::impl> { sync_sender<nonblocking_response_handle_tag>::impl> {
public:
event_based_actor();
protected: protected:
/** /**
......
...@@ -213,7 +213,7 @@ oss_wr operator<<(oss_wr&& lhs, T rhs) { ...@@ -213,7 +213,7 @@ oss_wr operator<<(oss_wr&& lhs, T rhs) {
#define CPPA_PRINT_IF1(stmt, lvlname, classname, funname, msg) \ #define CPPA_PRINT_IF1(stmt, lvlname, classname, funname, msg) \
CPPA_PRINT_IF0(stmt, lvlname, classname, funname, msg) CPPA_PRINT_IF0(stmt, lvlname, classname, funname, msg)
#if CPPA_LOG_LEVEL < 4 #if CPPA_LOG_LEVEL < CPPA_TRACE
# define CPPA_PRINT4(arg0, arg1, arg2, arg3) # define CPPA_PRINT4(arg0, arg1, arg2, arg3)
# else # else
# define CPPA_PRINT4(lvlname, classname, funname, msg) \ # define CPPA_PRINT4(lvlname, classname, funname, msg) \
...@@ -223,7 +223,7 @@ oss_wr operator<<(oss_wr&& lhs, T rhs) { ...@@ -223,7 +223,7 @@ oss_wr operator<<(oss_wr&& lhs, T rhs) {
} }
#endif #endif
#if CPPA_LOG_LEVEL < 3 #if CPPA_LOG_LEVEL < CPPA_DEBUG
# define CPPA_PRINT3(arg0, arg1, arg2, arg3) # define CPPA_PRINT3(arg0, arg1, arg2, arg3)
# define CPPA_PRINT_IF3(arg0, arg1, arg2, arg3, arg4) # define CPPA_PRINT_IF3(arg0, arg1, arg2, arg3, arg4)
# else # else
...@@ -233,7 +233,7 @@ oss_wr operator<<(oss_wr&& lhs, T rhs) { ...@@ -233,7 +233,7 @@ oss_wr operator<<(oss_wr&& lhs, T rhs) {
CPPA_PRINT_IF0(stmt, lvlname, classname, funname, msg) CPPA_PRINT_IF0(stmt, lvlname, classname, funname, msg)
#endif #endif
#if CPPA_LOG_LEVEL < 2 #if CPPA_LOG_LEVEL < CPPA_INFO
# define CPPA_PRINT2(arg0, arg1, arg2, arg3) # define CPPA_PRINT2(arg0, arg1, arg2, arg3)
# define CPPA_PRINT_IF2(arg0, arg1, arg2, arg3, arg4) # define CPPA_PRINT_IF2(arg0, arg1, arg2, arg3, arg4)
# else # else
......
...@@ -70,6 +70,10 @@ class context_switching_resume { ...@@ -70,6 +70,10 @@ class context_switching_resume {
, m_cs_thread(context_switching_resume::trampoline, , m_cs_thread(context_switching_resume::trampoline,
static_cast<blocking_actor*>(this)) { } static_cast<blocking_actor*>(this)) { }
inline bool exec_on_spawn() const {
return true;
}
void attach_to_scheduler() override { void attach_to_scheduler() override {
this->ref(); this->ref();
} }
......
...@@ -85,8 +85,10 @@ class cooperative_scheduling { ...@@ -85,8 +85,10 @@ class cooperative_scheduling {
inline void launch(Actor* self, execution_unit* host) { inline void launch(Actor* self, execution_unit* host) {
// detached in scheduler::worker::run // detached in scheduler::worker::run
self->attach_to_scheduler(); self->attach_to_scheduler();
if (host) host->exec_later(self); if (self->exec_on_spawn()) {
else get_scheduling_coordinator()->enqueue(self); if (host) host->exec_later(self);
else get_scheduling_coordinator()->enqueue(self);
}
} }
template<class Actor> template<class Actor>
...@@ -106,6 +108,7 @@ class cooperative_scheduling { ...@@ -106,6 +108,7 @@ class cooperative_scheduling {
switch (state) { switch (state) {
case actor_state::blocked: { case actor_state::blocked: {
if (set_ready()) { if (set_ready()) {
// re-schedule actor
if (host) host->exec_later(self); if (host) host->exec_later(self);
else get_scheduling_coordinator()->enqueue(self); else get_scheduling_coordinator()->enqueue(self);
return; return;
...@@ -114,6 +117,7 @@ class cooperative_scheduling { ...@@ -114,6 +117,7 @@ class cooperative_scheduling {
} }
case actor_state::about_to_block: { case actor_state::about_to_block: {
if (set_ready()) { if (set_ready()) {
// actor is still running
return; return;
} }
break; break;
......
...@@ -47,6 +47,7 @@ ...@@ -47,6 +47,7 @@
#include "cppa/policy/resume_policy.hpp" #include "cppa/policy/resume_policy.hpp"
#include "cppa/detail/cs_thread.hpp" #include "cppa/detail/cs_thread.hpp"
#include "cppa/detail/functor_based_actor.hpp"
namespace cppa { namespace policy { namespace cppa { namespace policy {
...@@ -73,6 +74,10 @@ class event_based_resume { ...@@ -73,6 +74,10 @@ class event_based_resume {
this->deref(); this->deref();
} }
inline bool exec_on_spawn() const {
return false;
}
resumable::resume_result resume(detail::cs_thread*, resumable::resume_result resume(detail::cs_thread*,
execution_unit* host) override { execution_unit* host) override {
CPPA_REQUIRE(host != nullptr); CPPA_REQUIRE(host != nullptr);
...@@ -80,8 +85,7 @@ class event_based_resume { ...@@ -80,8 +85,7 @@ class event_based_resume {
d->m_host = host; d->m_host = host;
CPPA_LOG_TRACE("id = " << d->id() CPPA_LOG_TRACE("id = " << d->id()
<< ", state = " << static_cast<int>(d->state())); << ", state = " << static_cast<int>(d->state()));
CPPA_REQUIRE( d->state() == actor_state::ready CPPA_REQUIRE(d->state() == actor_state::ready);
|| d->state() == actor_state::pending);
auto done_cb = [&]() -> bool { auto done_cb = [&]() -> bool {
CPPA_LOG_TRACE(""); CPPA_LOG_TRACE("");
d->bhvr_stack().clear(); d->bhvr_stack().clear();
...@@ -91,7 +95,7 @@ class event_based_resume { ...@@ -91,7 +95,7 @@ class event_based_resume {
} }
d->on_exit(); d->on_exit();
if (!d->bhvr_stack().empty()) { if (!d->bhvr_stack().empty()) {
CPPA_LOG_DEBUG("on_exit did set a new behavior"); CPPA_LOG_DEBUG("on_exit did set a new behavior in done_cb");
d->planned_exit_reason(exit_reason::not_exited); d->planned_exit_reason(exit_reason::not_exited);
return false; // on_exit did set a new behavior return false; // on_exit did set a new behavior
} }
...@@ -131,7 +135,8 @@ class event_based_resume { ...@@ -131,7 +135,8 @@ class event_based_resume {
} }
} }
else { else {
CPPA_LOG_DEBUG("no more element in mailbox; going to block"); CPPA_LOG_DEBUG("no more element in mailbox; "
"going to block");
d->set_state(actor_state::about_to_block); d->set_state(actor_state::about_to_block);
std::atomic_thread_fence(std::memory_order_seq_cst); std::atomic_thread_fence(std::memory_order_seq_cst);
if (!d->has_next_message()) { if (!d->has_next_message()) {
...@@ -145,7 +150,8 @@ class event_based_resume { ...@@ -145,7 +150,8 @@ class event_based_resume {
"arriving message"); "arriving message");
break; break;
case actor_state::blocked: case actor_state::blocked:
CPPA_LOG_DEBUG("set state successfully to blocked"); CPPA_LOG_DEBUG("set state successfully "
"to blocked");
// done setting actor to blocked // done setting actor to blocked
return resumable::resume_later; return resumable::resume_later;
default: default:
......
...@@ -207,9 +207,12 @@ class invoke_policy { ...@@ -207,9 +207,12 @@ class invoke_policy {
message_id& mid, message_id& mid,
Fun& fun, Fun& fun,
MaybeResponseHandle hdl = MaybeResponseHandle{}) { MaybeResponseHandle hdl = MaybeResponseHandle{}) {
# if CPPA_LOG_LEVEL >= CPPA_DEBUG
auto msg_str = to_string(msg);
# endif
auto res = fun(msg); // might change mid auto res = fun(msg); // might change mid
CPPA_LOG_DEBUG_IF(res, "actor did consume message"); CPPA_LOG_DEBUG_IF(res, "actor did consume message: " << msg_str);
CPPA_LOG_DEBUG_IF(!res, "actor did ignore message"); CPPA_LOG_DEBUG_IF(!res, "actor did ignore message: " << msg_str);
if (res) { if (res) {
//message_header hdr{self, sender, mid.is_request() ? mid.response_id() //message_header hdr{self, sender, mid.is_request() ? mid.response_id()
// : message_id{}}; // : message_id{}};
...@@ -357,8 +360,6 @@ class invoke_policy { ...@@ -357,8 +360,6 @@ class invoke_policy {
return hm_cache_msg; return hm_cache_msg;
} }
case ordinary_message: { case ordinary_message: {
CPPA_LOG_DEBUG("handle as ordinary message: "
<< CPPA_TARG(node->msg, to_string));
if (!awaited_response.valid()) { if (!awaited_response.valid()) {
auto previous_node = dptr()->hm_begin(self, node); auto previous_node = dptr()->hm_begin(self, node);
auto res = invoke_fun(self, auto res = invoke_fun(self,
......
...@@ -35,6 +35,10 @@ ...@@ -35,6 +35,10 @@
namespace cppa { namespace cppa {
event_based_actor::event_based_actor() {
m_state = actor_state::blocked;
}
void event_based_actor::forward_to(const actor& whom) { void event_based_actor::forward_to(const actor& whom) {
forward_message(whom, message_priority::normal); forward_message(whom, message_priority::normal);
} }
......
...@@ -34,6 +34,14 @@ namespace cppa { ...@@ -34,6 +34,14 @@ namespace cppa {
namespace detail { namespace detail {
behavior functor_based_actor::make_behavior() { behavior functor_based_actor::make_behavior() {
if (m_void_impl) {
enqueue({address(), this}, make_any_tuple(atom("RUN")), m_host);
return {
on(atom("RUN")) >> [=] {
become(m_make_behavior(this));
}
};
}
return m_make_behavior(this); return m_make_behavior(this);
} }
......
...@@ -78,7 +78,16 @@ class logging_impl : public logging { ...@@ -78,7 +78,16 @@ class logging_impl : public logging {
void initialize() { void initialize() {
m_thread = thread([this] { (*this)(); }); m_thread = thread([this] { (*this)(); });
log("TRACE", "logging", "run", __FILE__, __LINE__, "ENTRY"); std::string msg = "ENTRY log level = ";
switch (CPPA_LOG_LEVEL) {
default: msg += "????"; break;
case 0: msg += "ERROR"; break;
case 1: msg += "WARN"; break;
case 2: msg += "INFO"; break;
case 3: msg += "DEBUG"; break;
case 4: msg += "TRACE"; break;
}
log("TRACE", "logging", "run", __FILE__, __LINE__, msg);
} }
void destroy() { void destroy() {
......
...@@ -328,6 +328,9 @@ void coordinator::enqueue(resumable* what) { ...@@ -328,6 +328,9 @@ void coordinator::enqueue(resumable* what) {
* implementation of worker * * implementation of worker *
******************************************************************************/ ******************************************************************************/
#define CPPA_LOG_DEBUG_WORKER(msg) \
CPPA_LOG_DEBUG("worker " << m_id << ": " << msg)
worker::worker(size_t id, coordinator* parent) worker::worker(size_t id, coordinator* parent)
: m_running(true), m_id(id), m_last_victim(id), m_parent(parent) { } : m_running(true), m_id(id), m_last_victim(id), m_parent(parent) { }
...@@ -349,6 +352,7 @@ void worker::run() { ...@@ -349,6 +352,7 @@ void worker::run() {
if (!m_job_list.empty()) { if (!m_job_list.empty()) {
job = m_job_list.back(); job = m_job_list.back();
m_job_list.pop_back(); m_job_list.pop_back();
CPPA_LOG_DEBUG_WORKER("got job from m_job_list");
return true; return true;
} }
return false; return false;
...@@ -356,11 +360,17 @@ void worker::run() { ...@@ -356,11 +360,17 @@ void worker::run() {
auto aggressive_poll = [&]() -> bool { auto aggressive_poll = [&]() -> bool {
for (int i = 1; i < 101; ++i) { for (int i = 1; i < 101; ++i) {
job = m_exposed_queue.try_pop(); job = m_exposed_queue.try_pop();
if (job) return true; if (job) {
CPPA_LOG_DEBUG_WORKER("got job with aggressive polling");
return true;
}
// try to steal every 10 poll attempts // try to steal every 10 poll attempts
if ((i % 10) == 0) { if ((i % 10) == 0) {
job = raid(); job = raid();
if (job) return true; if (job) {
CPPA_LOG_DEBUG_WORKER("got job with aggressive polling");
return true;
}
} }
std::this_thread::yield(); std::this_thread::yield();
} }
...@@ -369,11 +379,17 @@ void worker::run() { ...@@ -369,11 +379,17 @@ void worker::run() {
auto moderate_poll = [&]() -> bool { auto moderate_poll = [&]() -> bool {
for (int i = 1; i < 550; ++i) { for (int i = 1; i < 550; ++i) {
job = m_exposed_queue.try_pop(); job = m_exposed_queue.try_pop();
if (job) return true; if (job) {
CPPA_LOG_DEBUG_WORKER("got job with moderate polling");
return true;
}
// try to steal every 5 poll attempts // try to steal every 5 poll attempts
if ((i % 5) == 0) { if ((i % 5) == 0) {
job = raid(); job = raid();
if (job) return true; if (job) {
CPPA_LOG_DEBUG_WORKER("got job with moderate polling");
return true;
}
} }
std::this_thread::sleep_for(std::chrono::microseconds(50)); std::this_thread::sleep_for(std::chrono::microseconds(50));
} }
...@@ -382,17 +398,22 @@ void worker::run() { ...@@ -382,17 +398,22 @@ void worker::run() {
auto relaxed_poll = [&]() -> bool { auto relaxed_poll = [&]() -> bool {
for (;;) { for (;;) {
job = m_exposed_queue.try_pop(); job = m_exposed_queue.try_pop();
if (job) return true; if (job) {
CPPA_LOG_DEBUG_WORKER("got job with relaxed polling");
return true;
}
// always try to steal at this stage // always try to steal at this stage
job = raid(); job = raid();
if (job) return true; if (job) {
CPPA_LOG_DEBUG_WORKER("got job with relaxed polling");
return true;
}
std::this_thread::sleep_for(std::chrono::milliseconds(10)); std::this_thread::sleep_for(std::chrono::milliseconds(10));
} }
}; };
// scheduling loop // scheduling loop
while (m_running) { while (m_running) {
local_poll() || aggressive_poll() || moderate_poll() || relaxed_poll(); local_poll() || aggressive_poll() || moderate_poll() || relaxed_poll();
CPPA_LOG_DEBUG("dequeued new job");
CPPA_PUSH_AID_FROM_PTR(dynamic_cast<abstract_actor*>(job)); CPPA_PUSH_AID_FROM_PTR(dynamic_cast<abstract_actor*>(job));
if (job->resume(&fself, this) == resumable::done) { if (job->resume(&fself, this) == resumable::done) {
// was attached in policy::cooperative_scheduling::launch // was attached in policy::cooperative_scheduling::launch
...@@ -413,15 +434,20 @@ worker::job_ptr worker::try_steal() { ...@@ -413,15 +434,20 @@ worker::job_ptr worker::try_steal() {
worker::job_ptr worker::raid() { worker::job_ptr worker::raid() {
// try once to steal from anyone // try once to steal from anyone
auto inc = [](size_t arg) -> size_t { return arg + 1; };
auto dec = [](size_t arg) -> size_t { return arg - 1; };
// reduce probability of 'steal collisions' by letting
// half the workers pick victims by increasing IDs and
// the other half by decreasing IDs
size_t (*next)(size_t) = (m_id % 2) == 0 ? inc : dec;
auto n = m_parent->num_workers(); auto n = m_parent->num_workers();
for (size_t i = 0; i < n; ++i) { for (size_t i = 0; i < n; ++i) {
m_last_victim = (m_last_victim + 1) % n; m_last_victim = next(m_last_victim) % n;
if (m_last_victim != m_id) { if (m_last_victim != m_id) {
auto job = m_parent->worker_by_id(m_last_victim)->try_steal(); auto job = m_parent->worker_by_id(m_last_victim)->try_steal();
if (job) { if (job) {
CPPA_LOG_DEBUG("worker " << m_id CPPA_LOG_DEBUG_WORKER("successfully stolen a job from "
<< " has successfully stolen a job from " << m_last_victim);
<< m_last_victim);
return job; return job;
} }
} }
......
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