Commit e957731f authored by Dominik Charousset's avatar Dominik Charousset

optimized event-based actors using functions

this patch saves one message per function-based actor to increase
scalability and performance
parent 54e383dd
...@@ -83,7 +83,7 @@ class event_based_actor : public scheduled_actor { ...@@ -83,7 +83,7 @@ class event_based_actor : public scheduled_actor {
protected: protected:
event_based_actor(); event_based_actor(actor_state st = actor_state::blocked);
// provoke compiler errors for usage of receive() and related functions // provoke compiler errors for usage of receive() and related functions
......
...@@ -126,7 +126,7 @@ class actor_facade<Ret(Args...)> : public actor { ...@@ -126,7 +126,7 @@ class actor_facade<Ret(Args...)> : public actor {
template<long... Is> template<long... Is>
void enqueue_impl(const actor_ptr& sender, any_tuple msg, message_id id, util::int_list<Is...>) { void enqueue_impl(const actor_ptr& sender, any_tuple msg, message_id id, util::int_list<Is...>) {
auto opt = tuple_cast<Args...>(msg); auto opt = tuple_cast<typename util::rm_ref<Args>::type...>(msg);
if (opt) { if (opt) {
response_handle handle{this, sender, id}; response_handle handle{this, sender, id};
size_t number_of_values = 1; size_t number_of_values = 1;
......
...@@ -36,6 +36,8 @@ ...@@ -36,6 +36,8 @@
#include "cppa/logging.hpp" #include "cppa/logging.hpp"
#include "cppa/event_based_actor.hpp" #include "cppa/event_based_actor.hpp"
using namespace std;
namespace cppa { namespace cppa {
class default_scheduled_actor : public event_based_actor { class default_scheduled_actor : public event_based_actor {
...@@ -46,16 +48,28 @@ class default_scheduled_actor : public event_based_actor { ...@@ -46,16 +48,28 @@ class default_scheduled_actor : public event_based_actor {
typedef std::function<void()> fun_type; typedef std::function<void()> fun_type;
default_scheduled_actor(fun_type&& fun) : m_fun(std::move(fun)) { } default_scheduled_actor(fun_type&& fun)
: super(actor_state::ready), m_fun(std::move(fun)), m_initialized(false) { }
void init() { }
void init() { resume_result resume(util::fiber* f, actor_ptr& next) {
become ( if (!m_initialized) {
on(atom("RUN")) >> [=] { scoped_self_setter sss{this};
CPPA_LOGS_TRACE("init$lambda", ""); m_initialized = true;
unbecome();
m_fun(); m_fun();
if (m_bhvr_stack.empty()) {
if (exit_reason() == exit_reason::not_exited) quit(exit_reason::normal);
set_state(actor_state::done);
m_bhvr_stack.clear();
m_bhvr_stack.cleanup();
on_exit();
next.swap(m_chained_actor);
set_state(actor_state::done);
return resume_result::actor_done;
}
} }
); return event_based_actor::resume(f, next);
} }
scheduled_actor_type impl_type() { scheduled_actor_type impl_type() {
...@@ -65,6 +79,7 @@ class default_scheduled_actor : public event_based_actor { ...@@ -65,6 +79,7 @@ class default_scheduled_actor : public event_based_actor {
private: private:
fun_type m_fun; fun_type m_fun;
bool m_initialized;
}; };
...@@ -72,7 +87,7 @@ intrusive_ptr<event_based_actor> event_based_actor::from(std::function<void()> f ...@@ -72,7 +87,7 @@ intrusive_ptr<event_based_actor> event_based_actor::from(std::function<void()> f
return detail::memory::create<default_scheduled_actor>(std::move(fun)); return detail::memory::create<default_scheduled_actor>(std::move(fun));
} }
event_based_actor::event_based_actor() : super(actor_state::blocked, true) { } event_based_actor::event_based_actor(actor_state st) : super(st, true) { }
void event_based_actor::dequeue(behavior&) { void event_based_actor::dequeue(behavior&) {
quit(exit_reason::unallowed_function_call); quit(exit_reason::unallowed_function_call);
......
...@@ -209,20 +209,10 @@ actor_ptr thread_pool_scheduler::exec(spawn_options os, scheduled_actor_ptr p) { ...@@ -209,20 +209,10 @@ actor_ptr thread_pool_scheduler::exec(spawn_options os, scheduled_actor_ptr p) {
return std::move(p); return std::move(p);
} }
p->attach_to_scheduler(this, is_hidden); p->attach_to_scheduler(this, is_hidden);
if (p->has_behavior()) { if (p->has_behavior() || p->impl_type() == default_event_based_impl) {
if (!is_hidden) get_actor_registry()->inc_running(); if (!is_hidden) get_actor_registry()->inc_running();
p->ref(); // implicit reference that's released if actor dies p->ref(); // implicit reference that's released if actor dies
switch (p->impl_type()) { if (p->impl_type() != event_based_impl) m_queue.push_back(p.get());
case default_event_based_impl: {
p->enqueue(nullptr, make_any_tuple(atom("RUN")));
break;
}
case context_switching_impl: {
m_queue.push_back(p.get());
break;
}
default: break; // nothing to do
}
} }
else p->on_exit(); else p->on_exit();
return std::move(p); return std::move(p);
......
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