Commit 7ce78f07 authored by Dominik Charousset's avatar Dominik Charousset

added scheduled_and_hidden for system-internal event-based actors

parent f22a121f
...@@ -55,15 +55,19 @@ class thread_pool_scheduler : public scheduler { ...@@ -55,15 +55,19 @@ class thread_pool_scheduler : public scheduler {
void enqueue(scheduled_actor* what) /*override*/; void enqueue(scheduled_actor* what) /*override*/;
actor_ptr spawn(scheduled_actor* what); actor_ptr spawn(scheduled_actor* what,
scheduling_hint hint);
actor_ptr spawn(scheduled_actor* what, init_callback init_cb); actor_ptr spawn(scheduled_actor* what,
init_callback init_cb,
scheduling_hint hint);
actor_ptr spawn(void_function fun, scheduling_hint hint); actor_ptr spawn(void_function fun,
scheduling_hint hint);
actor_ptr spawn(void_function what, actor_ptr spawn(void_function what,
scheduling_hint hint, init_callback init_cb,
init_callback init_cb); scheduling_hint hint);
private: private:
......
...@@ -66,15 +66,18 @@ class scheduled_actor : public local_actor { ...@@ -66,15 +66,18 @@ class scheduled_actor : public local_actor {
// called from worker thread // called from worker thread
virtual resume_result resume(util::fiber* from) = 0; virtual resume_result resume(util::fiber* from) = 0;
void attach_to_scheduler(scheduler* sched); void attach_to_scheduler(scheduler* sched, bool hidden);
virtual bool has_behavior() = 0; virtual bool has_behavior() = 0;
virtual scheduled_actor_type impl_type() = 0; virtual scheduled_actor_type impl_type() = 0;
inline bool is_hidden() const { return m_hidden; }
protected: protected:
scheduler* m_scheduler; scheduler* m_scheduler;
bool m_hidden;
bool initialized(); bool initialized();
......
...@@ -156,20 +156,23 @@ class scheduler { ...@@ -156,20 +156,23 @@ class scheduler {
* it starts execution. * it starts execution.
*/ */
virtual actor_ptr spawn(void_function fun, virtual actor_ptr spawn(void_function fun,
scheduling_hint hint, init_callback init_cb,
init_callback init_cb) = 0; scheduling_hint hint) = 0;
/** /**
* @brief Spawns a new event-based actor. * @brief Spawns a new event-based actor.
*/ */
virtual actor_ptr spawn(scheduled_actor* what) = 0; virtual actor_ptr spawn(scheduled_actor* what,
scheduling_hint hint = scheduled) = 0;
/** /**
* @brief Spawns a new event-based actor and calls * @brief Spawns a new event-based actor and calls
* <code>init_cb</code> after the actor is initialized but before * <code>init_cb</code> after the actor is initialized but before
* it starts execution. * it starts execution.
*/ */
virtual actor_ptr spawn(scheduled_actor* what, init_callback init_cb) = 0; virtual actor_ptr spawn(scheduled_actor* what,
init_callback init_cb,
scheduling_hint hint = scheduled) = 0;
// hide implementation details for documentation // hide implementation details for documentation
# ifndef CPPA_DOCUMENTATION # ifndef CPPA_DOCUMENTATION
...@@ -198,15 +201,15 @@ class scheduler { ...@@ -198,15 +201,15 @@ class scheduler {
std::forward<Fun>(fun), std::forward<Fun>(fun),
detail::spawn_fwd_<typename util::rm_ref<Arg0>::type>::_(arg0), detail::spawn_fwd_<typename util::rm_ref<Arg0>::type>::_(arg0),
detail::spawn_fwd_<typename util::rm_ref<Args>::type>::_(args)...), detail::spawn_fwd_<typename util::rm_ref<Args>::type>::_(args)...),
hint, std::forward<InitCallback>(init_cb),
std::forward<InitCallback>(init_cb)); hint);
} }
template<typename InitCallback, typename Fun> template<typename InitCallback, typename Fun>
actor_ptr spawn_cb_impl(scheduling_hint hint, InitCallback&& init_cb, Fun&& fun) { actor_ptr spawn_cb_impl(scheduling_hint hint, InitCallback&& init_cb, Fun&& fun) {
return this->spawn(std::forward<Fun>(fun), return this->spawn(std::forward<Fun>(fun),
hint, std::forward<InitCallback>(init_cb),
std::forward<InitCallback>(init_cb)); hint);
} }
# endif // CPPA_DOCUMENTATION # endif // CPPA_DOCUMENTATION
......
...@@ -50,10 +50,16 @@ enum scheduling_hint { ...@@ -50,10 +50,16 @@ enum scheduling_hint {
detached, detached,
/** /**
* @brief Indicates that an actor should run in its own thread but should * @brief Indicates that an actor should run in its own thread,
* be ignored by {@link await_others_done()}. * but it is ignored by {@link await_others_done()}.
*/ */
detached_and_hidden detached_and_hidden,
/**
* @brief Indicates that an actor takes part in cooperative scheduling,
* but it is ignored by {@link await_others_done()}.
*/
scheduled_and_hidden
}; };
......
...@@ -34,10 +34,11 @@ ...@@ -34,10 +34,11 @@
namespace cppa { namespace cppa {
scheduled_actor::scheduled_actor(bool enable_chained_send) scheduled_actor::scheduled_actor(bool enable_chained_send)
: local_actor(enable_chained_send), next(nullptr), m_scheduler(nullptr) { } : local_actor(enable_chained_send), next(0), m_scheduler(0), m_hidden(false) { }
void scheduled_actor::attach_to_scheduler(scheduler* sched) { void scheduled_actor::attach_to_scheduler(scheduler* sched, bool hidden) {
CPPA_REQUIRE(sched != nullptr); CPPA_REQUIRE(sched != nullptr);
m_hidden = hidden;
// init is called by the spawning actor, manipulate self to // init is called by the spawning actor, manipulate self to
// point to this actor // point to this actor
scoped_self_setter sss{this}; scoped_self_setter sss{this};
......
...@@ -51,24 +51,6 @@ namespace { ...@@ -51,24 +51,6 @@ namespace {
typedef std::unique_lock<std::mutex> guard_type; typedef std::unique_lock<std::mutex> guard_type;
typedef intrusive::single_reader_queue<thread_pool_scheduler::worker> worker_queue; typedef intrusive::single_reader_queue<thread_pool_scheduler::worker> worker_queue;
/*
template<typename F>
actor_ptr spawn_void_fun_impl(thread_pool_scheduler* _this,
void_function fun,
scheduling_hint sh,
F cb ) {
if (sh == scheduled) {
scheduled_actor_ptr ptr{new context_switching_actor(std::move(fun))};
ptr->attach_to_scheduler(_this);
cb(ptr.get());
return _this->spawn_impl(std::move(ptr));
}
else {
return _this->spawn_as_thread(std::move(fun), std::move(cb));
}
}
*/
} // namespace <anonmyous> } // namespace <anonmyous>
struct thread_pool_scheduler::worker { struct thread_pool_scheduler::worker {
...@@ -154,9 +136,10 @@ struct thread_pool_scheduler::worker { ...@@ -154,9 +136,10 @@ struct thread_pool_scheduler::worker {
switch (job->resume(&fself)) { switch (job->resume(&fself)) {
case resume_result::actor_done: { case resume_result::actor_done: {
auto pending = fetch_pending(); auto pending = fetch_pending();
bool hidden = job->is_hidden();
if (!job->deref()) delete job; if (!job->deref()) delete job;
std::atomic_thread_fence(std::memory_order_seq_cst); std::atomic_thread_fence(std::memory_order_seq_cst);
dec_actor_count(); if (!hidden) dec_actor_count();
job = pending; job = pending;
break; break;
} }
...@@ -204,9 +187,10 @@ void thread_pool_scheduler::stop() { ...@@ -204,9 +187,10 @@ void thread_pool_scheduler::stop() {
auto ptr = m_queue.try_pop(); auto ptr = m_queue.try_pop();
while (ptr != nullptr) { while (ptr != nullptr) {
if (ptr != &m_dummy) { if (ptr != &m_dummy) {
bool hidden = ptr->is_hidden();
if (!ptr->deref()) delete ptr; if (!ptr->deref()) delete ptr;
std::atomic_thread_fence(std::memory_order_seq_cst); std::atomic_thread_fence(std::memory_order_seq_cst);
dec_actor_count(); if (!hidden) dec_actor_count();
ptr = m_queue.try_pop(); ptr = m_queue.try_pop();
} }
} }
...@@ -240,7 +224,7 @@ actor_ptr thread_pool_scheduler::spawn_as_thread(void_function fun, ...@@ -240,7 +224,7 @@ actor_ptr thread_pool_scheduler::spawn_as_thread(void_function fun,
actor_ptr thread_pool_scheduler::spawn_impl(scheduled_actor_ptr what) { actor_ptr thread_pool_scheduler::spawn_impl(scheduled_actor_ptr what) {
if (what->has_behavior()) { if (what->has_behavior()) {
inc_actor_count(); if (!what->is_hidden()) { inc_actor_count(); }
what->ref(); what->ref();
// event-based actors are not pushed to the job queue on startup // event-based actors are not pushed to the job queue on startup
if (what->impl_type() == context_switching_impl) { if (what->impl_type() == context_switching_impl) {
...@@ -253,46 +237,49 @@ actor_ptr thread_pool_scheduler::spawn_impl(scheduled_actor_ptr what) { ...@@ -253,46 +237,49 @@ actor_ptr thread_pool_scheduler::spawn_impl(scheduled_actor_ptr what) {
return std::move(what); return std::move(what);
} }
actor_ptr thread_pool_scheduler::spawn(scheduled_actor* raw) { actor_ptr thread_pool_scheduler::spawn(scheduled_actor* raw,
scheduling_hint hint) {
scheduled_actor_ptr ptr{raw}; scheduled_actor_ptr ptr{raw};
ptr->attach_to_scheduler(this); ptr->attach_to_scheduler(this, hint == scheduled_and_hidden);
return spawn_impl(std::move(ptr)); return spawn_impl(std::move(ptr));
} }
actor_ptr thread_pool_scheduler::spawn(scheduled_actor* raw, init_callback cb) { actor_ptr thread_pool_scheduler::spawn(scheduled_actor* raw,
init_callback cb,
scheduling_hint hint) {
scheduled_actor_ptr ptr{raw}; scheduled_actor_ptr ptr{raw};
ptr->attach_to_scheduler(this); ptr->attach_to_scheduler(this, hint == scheduled_and_hidden);
cb(ptr.get()); cb(ptr.get());
return spawn_impl(std::move(ptr)); return spawn_impl(std::move(ptr));
} }
#ifndef CPPA_DISABLE_CONTEXT_SWITCHING #ifndef CPPA_DISABLE_CONTEXT_SWITCHING
actor_ptr thread_pool_scheduler::spawn(void_function fun, scheduling_hint sh) { actor_ptr thread_pool_scheduler::spawn(void_function fun, scheduling_hint hint) {
if (sh == scheduled) { if (hint == scheduled || hint == scheduled_and_hidden) {
scheduled_actor_ptr ptr{new context_switching_actor(std::move(fun))}; scheduled_actor_ptr ptr{new context_switching_actor(std::move(fun))};
ptr->attach_to_scheduler(this); ptr->attach_to_scheduler(this, hint == scheduled_and_hidden);
return spawn_impl(std::move(ptr)); return spawn_impl(std::move(ptr));
} }
else { else {
return spawn_as_thread(std::move(fun), return spawn_as_thread(std::move(fun),
[](local_actor*) { }, [](local_actor*) { },
sh == detached_and_hidden); hint == detached_and_hidden);
} }
} }
actor_ptr thread_pool_scheduler::spawn(void_function fun, actor_ptr thread_pool_scheduler::spawn(void_function fun,
scheduling_hint sh, init_callback init_cb,
init_callback init_cb) { scheduling_hint hint) {
if (sh == scheduled) { if (hint == scheduled || hint == scheduled_and_hidden) {
scheduled_actor_ptr ptr{new context_switching_actor(std::move(fun))}; scheduled_actor_ptr ptr{new context_switching_actor(std::move(fun))};
ptr->attach_to_scheduler(this); ptr->attach_to_scheduler(this, hint == scheduled_and_hidden);
init_cb(ptr.get()); init_cb(ptr.get());
return spawn_impl(std::move(ptr)); return spawn_impl(std::move(ptr));
} }
else { else {
return spawn_as_thread(std::move(fun), return spawn_as_thread(std::move(fun),
std::move(init_cb), std::move(init_cb),
sh == detached_and_hidden); hint == detached_and_hidden);
} }
} }
......
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