Commit 4126af45 authored by Dominik Charousset's avatar Dominik Charousset

properly spawn new actors in same worker as parent

parent 1ad4a179
...@@ -188,7 +188,7 @@ class proper_actor : public proper_actor_base<Base, ...@@ -188,7 +188,7 @@ class proper_actor : public proper_actor_base<Base,
template <typename... Ts> template <typename... Ts>
proper_actor(Ts&&... args) : super(std::forward<Ts>(args)...) { } proper_actor(Ts&&... args) : super(std::forward<Ts>(args)...) { }
inline void launch(bool is_hidden) { inline void launch(bool is_hidden, execution_unit* host) {
CPPA_LOG_TRACE(""); CPPA_LOG_TRACE("");
this->hidden(is_hidden); this->hidden(is_hidden);
auto bhvr = this->make_behavior(); auto bhvr = this->make_behavior();
...@@ -196,7 +196,7 @@ class proper_actor : public proper_actor_base<Base, ...@@ -196,7 +196,7 @@ class proper_actor : public proper_actor_base<Base,
CPPA_LOG_WARNING_IF(this->bhvr_stack().empty(), CPPA_LOG_WARNING_IF(this->bhvr_stack().empty(),
"actor did not set a behavior"); "actor did not set a behavior");
if (!this->bhvr_stack().empty()) { if (!this->bhvr_stack().empty()) {
this->scheduling_policy().launch(this); this->scheduling_policy().launch(this, host);
} }
} }
...@@ -256,9 +256,9 @@ class proper_actor<Base, Policies,true> : public proper_actor_base<Base, ...@@ -256,9 +256,9 @@ class proper_actor<Base, Policies,true> : public proper_actor_base<Base,
this->resume_policy().await_ready(this); this->resume_policy().await_ready(this);
} }
inline void launch(bool is_hidden) { inline void launch(bool is_hidden, execution_unit* host) {
this->hidden(is_hidden); this->hidden(is_hidden);
this->scheduling_policy().launch(this); this->scheduling_policy().launch(this, host);
} }
// implement blocking_actor::dequeue_response // implement blocking_actor::dequeue_response
......
...@@ -52,7 +52,6 @@ ...@@ -52,7 +52,6 @@
#include "cppa/message_header.hpp" #include "cppa/message_header.hpp"
#include "cppa/abstract_actor.hpp" #include "cppa/abstract_actor.hpp"
#include "cppa/abstract_group.hpp" #include "cppa/abstract_group.hpp"
#include "cppa/execution_unit.hpp"
#include "cppa/mailbox_element.hpp" #include "cppa/mailbox_element.hpp"
#include "cppa/response_promise.hpp" #include "cppa/response_promise.hpp"
#include "cppa/message_priority.hpp" #include "cppa/message_priority.hpp"
...@@ -68,7 +67,6 @@ ...@@ -68,7 +67,6 @@
namespace cppa { namespace cppa {
// forward declarations // forward declarations
class execution_unit;
class sync_handle_helper; class sync_handle_helper;
/** /**
...@@ -94,28 +92,32 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> { ...@@ -94,28 +92,32 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
template<class C, spawn_options Os = no_spawn_options, typename... Ts> template<class C, spawn_options Os = no_spawn_options, typename... Ts>
actor spawn(Ts&&... args) { actor spawn(Ts&&... args) {
constexpr auto os = make_unbound(Os); constexpr auto os = make_unbound(Os);
auto res = cppa::spawn<C, os>(std::forward<Ts>(args)...); auto res = spawn_class<C, os>(m_host, empty_before_launch_callback{},
std::forward<Ts>(args)...);
return eval_opts(Os, std::move(res)); return eval_opts(Os, std::move(res));
} }
template<spawn_options Os = no_spawn_options, typename... Ts> template<spawn_options Os = no_spawn_options, typename... Ts>
actor spawn(Ts&&... args) { actor spawn(Ts&&... args) {
constexpr auto os = make_unbound(Os); constexpr auto os = make_unbound(Os);
auto res = cppa::spawn<os>(std::forward<Ts>(args)...); auto res = spawn_functor<os>(m_host, empty_before_launch_callback{},
std::forward<Ts>(args)...);
return eval_opts(Os, std::move(res)); return eval_opts(Os, std::move(res));
} }
template<spawn_options Os = no_spawn_options, typename... Ts> template<class C, spawn_options Os, typename... Ts>
actor spawn_in_group(const group& grp, Ts&&... args) { actor spawn_in_group(const group& grp, Ts&&... args) {
constexpr auto os = make_unbound(Os); constexpr auto os = make_unbound(Os);
auto res = cppa::spawn_in_group<os>(grp, std::forward<Ts>(args)...); auto res = spawn_class<C, os>(m_host, group_subscriber{grp},
std::forward<Ts>(args)...);
return eval_opts(Os, std::move(res)); return eval_opts(Os, std::move(res));
} }
template<class C, spawn_options Os, typename... Ts> template<spawn_options Os = no_spawn_options, typename... Ts>
actor spawn_in_group(const group& grp, Ts&&... args) { actor spawn_in_group(const group& grp, Ts&&... args) {
constexpr auto os = make_unbound(Os); constexpr auto os = make_unbound(Os);
auto res = cppa::spawn_in_group<C, os>(grp, std::forward<Ts>(args)...); auto res = spawn_functor<os>(m_host, group_subscriber{grp},
std::forward<Ts>(args)...);
return eval_opts(Os, std::move(res)); return eval_opts(Os, std::move(res));
} }
...@@ -123,6 +125,17 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> { ...@@ -123,6 +125,17 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
* spawn typed actors * * spawn typed actors *
**************************************************************************/ **************************************************************************/
template<class C, spawn_options Os = no_spawn_options, typename... Ts>
typename detail::actor_handle_from_signature_list<
typename C::signatures
>::type
spawn_typed(Ts&&... args) {
constexpr auto os = make_unbound(Os);
auto res = spawn_class<C, os>(m_host, empty_before_launch_callback{},
std::forward<Ts>(args)...);
return eval_opts(Os, std::move(res));
}
template<spawn_options Os = no_spawn_options, typename F, typename... Ts> template<spawn_options Os = no_spawn_options, typename F, typename... Ts>
typename detail::infer_typed_actor_handle< typename detail::infer_typed_actor_handle<
typename util::get_callable_trait<F>::result_type, typename util::get_callable_trait<F>::result_type,
...@@ -132,18 +145,10 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> { ...@@ -132,18 +145,10 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
>::type >::type
spawn_typed(F fun, Ts&&... args) { spawn_typed(F fun, Ts&&... args) {
constexpr auto os = make_unbound(Os); constexpr auto os = make_unbound(Os);
auto res = cppa::spawn_typed<os>(std::move(fun), auto res = cppa::spawn_typed_functor<os>(m_host,
std::forward<Ts>(args)...); empty_before_launch_callback{},
return eval_opts(Os, std::move(res)); std::move(fun),
} std::forward<Ts>(args)...);
template<class C, spawn_options Os = no_spawn_options, typename... Ts>
typename detail::actor_handle_from_signature_list<
typename C::signatures
>::type
spawn_typed(Ts&&... args) {
constexpr auto os = make_unbound(Os);
auto res = cppa::spawn_typed<C, os>(std::forward<Ts>(args)...);
return eval_opts(Os, std::move(res)); return eval_opts(Os, std::move(res));
} }
...@@ -472,10 +477,10 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> { ...@@ -472,10 +477,10 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
template<class ActorHandle> template<class ActorHandle>
inline ActorHandle eval_opts(spawn_options opts, ActorHandle res) { inline ActorHandle eval_opts(spawn_options opts, ActorHandle res) {
if (has_monitor_flag(opts)) { if (has_monitor_flag(opts)) {
monitor(res.address()); monitor(res->address());
} }
if (has_link_flag(opts)) { if (has_link_flag(opts)) {
link_to(res.address()); link_to(res->address());
} }
return res; return res;
} }
......
...@@ -70,7 +70,7 @@ class logging { ...@@ -70,7 +70,7 @@ class logging {
// associates given actor id with this thread, // associates given actor id with this thread,
// returns the previously set actor id // returns the previously set actor id
virtual actor_id set_aid(actor_id aid) = 0; actor_id set_aid(actor_id aid);
virtual void log(const char* level, virtual void log(const char* level,
const char* class_name, const char* class_name,
...@@ -176,6 +176,7 @@ oss_wr operator<<(oss_wr&& lhs, T rhs) { ...@@ -176,6 +176,7 @@ oss_wr operator<<(oss_wr&& lhs, T rhs) {
# define CPPA_LOG_IMPL(lvlname, classname, funname, message) \ # define CPPA_LOG_IMPL(lvlname, classname, funname, message) \
CPPA_PRINT_ERROR_IMPL(lvlname, classname, funname, message) CPPA_PRINT_ERROR_IMPL(lvlname, classname, funname, message)
# define CPPA_PUSH_AID(unused) static_cast<void>(0) # define CPPA_PUSH_AID(unused) static_cast<void>(0)
# define CPPA_PUSH_AID_FROM_PTR(unused) static_cast<void>(0)
# define CPPA_SET_AID(unused) cppa_set_aid_dummy() # define CPPA_SET_AID(unused) cppa_set_aid_dummy()
# define CPPA_LOG_LEVEL 1 # define CPPA_LOG_LEVEL 1
#else #else
...@@ -190,6 +191,9 @@ oss_wr operator<<(oss_wr&& lhs, T rhs) { ...@@ -190,6 +191,9 @@ oss_wr operator<<(oss_wr&& lhs, T rhs) {
auto aid_pop_guard = ::cppa::util::make_scope_guard([=] { \ auto aid_pop_guard = ::cppa::util::make_scope_guard([=] { \
::cppa::get_logger()->set_aid(prev_aid_in_scope); \ ::cppa::get_logger()->set_aid(prev_aid_in_scope); \
}) })
# define CPPA_PUSH_AID_FROM_PTR(some_ptr) \
auto aid_ptr_argument = some_ptr; \
CPPA_PUSH_AID(aid_ptr_argument ? aid_ptr_argument->id() : 0)
# define CPPA_SET_AID(aid_arg) \ # define CPPA_SET_AID(aid_arg) \
::cppa::get_logger()->set_aid(aid_arg) ::cppa::get_logger()->set_aid(aid_arg)
#endif #endif
......
...@@ -80,8 +80,8 @@ class context_switching_resume { ...@@ -80,8 +80,8 @@ class context_switching_resume {
resumable::resume_result resume(detail::cs_thread* from, resumable::resume_result resume(detail::cs_thread* from,
execution_unit* host) override { execution_unit* host) override {
CPPA_REQUIRE(from != nullptr); CPPA_REQUIRE(from != nullptr && host != nullptr);
CPPA_PUSH_AID(this->id()); CPPA_LOG_TRACE("");
this->m_host = host; this->m_host = host;
using namespace detail; using namespace detail;
for (;;) { for (;;) {
......
...@@ -82,10 +82,11 @@ class cooperative_scheduling { ...@@ -82,10 +82,11 @@ class cooperative_scheduling {
} }
template<class Actor> template<class Actor>
inline void launch(Actor* self) { 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();
get_scheduling_coordinator()->enqueue(self); if (host) host->exec_later(self);
else get_scheduling_coordinator()->enqueue(self);
} }
template<class Actor> template<class Actor>
......
...@@ -39,6 +39,7 @@ ...@@ -39,6 +39,7 @@
#include "cppa/config.hpp" #include "cppa/config.hpp"
#include "cppa/extend.hpp" #include "cppa/extend.hpp"
#include "cppa/logging.hpp"
#include "cppa/behavior.hpp" #include "cppa/behavior.hpp"
#include "cppa/scheduler.hpp" #include "cppa/scheduler.hpp"
#include "cppa/actor_state.hpp" #include "cppa/actor_state.hpp"
...@@ -74,13 +75,13 @@ class event_based_resume { ...@@ -74,13 +75,13 @@ class event_based_resume {
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);
auto d = dptr(); auto d = dptr();
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); || d->state() == actor_state::pending);
CPPA_PUSH_AID(d->id());
auto done_cb = [&]() -> bool { auto done_cb = [&]() -> bool {
CPPA_LOG_TRACE(""); CPPA_LOG_TRACE("");
d->bhvr_stack().clear(); d->bhvr_stack().clear();
......
...@@ -105,7 +105,7 @@ class no_scheduling { ...@@ -105,7 +105,7 @@ class no_scheduling {
} }
template<class Actor> template<class Actor>
void launch(Actor* self) { void launch(Actor* self, execution_unit*) {
CPPA_PUSH_AID(self->id()); CPPA_PUSH_AID(self->id());
CPPA_LOG_TRACE(CPPA_ARG(self)); CPPA_LOG_TRACE(CPPA_ARG(self));
CPPA_REQUIRE(self != nullptr); CPPA_REQUIRE(self != nullptr);
......
This diff is collapsed.
...@@ -34,6 +34,7 @@ ...@@ -34,6 +34,7 @@
#ifndef CPPA_SPAWN_FWD_HPP #ifndef CPPA_SPAWN_FWD_HPP
#define CPPA_SPAWN_FWD_HPP #define CPPA_SPAWN_FWD_HPP
#include "cppa/group.hpp"
#include "cppa/typed_actor.hpp" #include "cppa/typed_actor.hpp"
#include "cppa/spawn_options.hpp" #include "cppa/spawn_options.hpp"
...@@ -41,21 +42,42 @@ ...@@ -41,21 +42,42 @@
namespace cppa { namespace cppa {
/****************************************************************************** template<class C, spawn_options Os, typename BeforeLaunch, typename... Ts>
* untyped actors * intrusive_ptr<C> spawn_class(execution_unit* host,
******************************************************************************/ BeforeLaunch before_launch_fun,
Ts&&... args);
template<spawn_options Os, typename BeforeLaunch, typename F, typename... Ts>
actor spawn_functor(execution_unit* host,
BeforeLaunch before_launch_fun,
F fun,
Ts&&... args);
class group_subscriber {
public:
inline group_subscriber(const group& grp) : m_grp(grp) { }
template<typename T>
inline void operator()(T* ptr) const {
ptr->join(m_grp);
}
template<class Impl, spawn_options Os = no_spawn_options, typename... Ts> private:
actor spawn(Ts&&... args);
template<spawn_options Os = no_spawn_options, typename... Ts> group m_grp;
actor spawn(Ts&&... args);
template<class Impl, spawn_options Os = no_spawn_options, typename... Ts> };
actor spawn_in_group(const group&, Ts&&... args);
class empty_before_launch_callback {
template<spawn_options Os = no_spawn_options, typename... Ts> public:
actor spawn_in_group(const group&, Ts&&... args);
template<typename T>
inline void operator()(T*) const { }
};
/****************************************************************************** /******************************************************************************
* typed actors * * typed actors *
...@@ -88,20 +110,14 @@ struct actor_handle_from_signature_list<util::type_list<Rs...>> { ...@@ -88,20 +110,14 @@ struct actor_handle_from_signature_list<util::type_list<Rs...>> {
} // namespace detail } // namespace detail
template<class Impl, spawn_options Os = no_spawn_options, typename... Ts> template<spawn_options Os, typename BeforeLaunch, typename F, typename... Ts>
typename detail::actor_handle_from_signature_list<
typename Impl::signatures
>::type
spawn_typed(Ts&&... args);
template<spawn_options Os = no_spawn_options, typename F, typename... Ts>
typename detail::infer_typed_actor_handle< typename detail::infer_typed_actor_handle<
typename util::get_callable_trait<F>::result_type, typename util::get_callable_trait<F>::result_type,
typename util::tl_head< typename util::tl_head<
typename util::get_callable_trait<F>::arg_types typename util::get_callable_trait<F>::arg_types
>::type >::type
>::type >::type
spawn_typed(F fun, Ts&&... args); spawn_typed_functor(execution_unit*, BeforeLaunch bl, F fun, Ts&&... args);
} // namespace cppa } // namespace cppa
......
...@@ -104,12 +104,6 @@ class logging_impl : public logging { ...@@ -104,12 +104,6 @@ class logging_impl : public logging {
} }
} }
actor_id set_aid(actor_id aid) override {
actor_id prev = t_self_id;
t_self_id = aid;
return prev;
}
void log(const char* level, void log(const char* level,
const char* c_class_name, const char* c_class_name,
const char* function_name, const char* function_name,
...@@ -170,4 +164,10 @@ logging::~logging() { } ...@@ -170,4 +164,10 @@ logging::~logging() { }
logging* logging::create_singleton() { return new logging_impl; } logging* logging::create_singleton() { return new logging_impl; }
actor_id logging::set_aid(actor_id aid) {
actor_id prev = t_self_id;
t_self_id = aid;
return prev;
}
} // namespace cppa } // namespace cppa
...@@ -242,6 +242,7 @@ class coordinator::shutdown_helper : public resumable { ...@@ -242,6 +242,7 @@ class coordinator::shutdown_helper : public resumable {
resumable::resume_result resume(detail::cs_thread*, execution_unit* ptr) { resumable::resume_result resume(detail::cs_thread*, execution_unit* ptr) {
auto w = dynamic_cast<worker*>(ptr); auto w = dynamic_cast<worker*>(ptr);
CPPA_REQUIRE(w != nullptr); CPPA_REQUIRE(w != nullptr);
CPPA_LOG_DEBUG("shutdown_helper::resume => shutdown worker");
w->m_running = false; w->m_running = false;
std::unique_lock<std::mutex> guard(mtx); std::unique_lock<std::mutex> guard(mtx);
last_worker = w; last_worker = w;
...@@ -328,7 +329,7 @@ void coordinator::enqueue(resumable* what) { ...@@ -328,7 +329,7 @@ void coordinator::enqueue(resumable* what) {
******************************************************************************/ ******************************************************************************/
worker::worker(size_t id, coordinator* parent) worker::worker(size_t id, coordinator* parent)
: m_running(false), m_id(id), m_last_victim(id), m_parent(parent) { } : m_running(true), m_id(id), m_last_victim(id), m_parent(parent) { }
void worker::start() { void worker::start() {
...@@ -389,15 +390,20 @@ void worker::run() { ...@@ -389,15 +390,20 @@ void worker::run() {
} }
}; };
// scheduling loop // scheduling loop
m_running = true;
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_LOG_DEBUG("dequeued new 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
job->detach_from_scheduler(); job->detach_from_scheduler();
} }
job = nullptr; job = nullptr;
// give others the opportunity to steal from us
if (m_job_list.size() > 1 && m_exposed_queue.empty()) {
m_exposed_queue.push_back(m_job_list.front());
m_job_list.erase(m_job_list.begin());
}
} }
} }
...@@ -412,7 +418,12 @@ worker::job_ptr worker::raid() { ...@@ -412,7 +418,12 @@ worker::job_ptr worker::raid() {
m_last_victim = (m_last_victim + 1) % n; m_last_victim = (m_last_victim + 1) % 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) return job; if (job) {
CPPA_LOG_DEBUG("worker " << m_id
<< " has successfully stolen a job from "
<< m_last_victim);
return job;
}
} }
} }
return nullptr; return nullptr;
......
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