Commit 99626f2c authored by neverlord's avatar neverlord

event actor, stacked event actor, mixin, etc.

parent e09e8634
......@@ -51,6 +51,7 @@ libcppa_la_SOURCES = \
src/serializer.cpp \
src/shared_spinlock.cpp \
src/singleton_manager.cpp \
src/stacked_event_based_actor.cpp \
src/string_serialization.cpp \
src/task_scheduler.cpp \
src/thread_pool_scheduler.cpp \
......@@ -84,8 +85,9 @@ nobase_library_include_HEADERS = \
cppa/cppa.hpp \
cppa/deserializer.hpp \
cppa/event_based_actor.hpp \
cppa/event_based_actor_mixin.hpp \
cppa/detail/abstract_actor.hpp \
cppa/detail/abstract_event_based_actor.hpp \
cppa/abstract_event_based_actor.hpp \
cppa/detail/abstract_tuple.hpp \
cppa/detail/actor_count.hpp \
cppa/detail/actor_proxy_cache.hpp \
......@@ -156,6 +158,7 @@ nobase_library_include_HEADERS = \
cppa/scheduler.hpp \
cppa/scheduling_hint.hpp \
cppa/serializer.hpp \
cppa/stacked_event_based_actor.hpp \
cppa/to_string.hpp \
cppa/tuple.hpp \
cppa/tuple_view.hpp \
......
......@@ -232,7 +232,10 @@ examples/math_actor_example.cpp
cppa/detail/yielding_actor.hpp
src/yielding_actor.cpp
cppa/util/either.hpp
cppa/detail/abstract_event_based_actor.hpp
cppa/abstract_event_based_actor.hpp
src/abstract_event_based_actor.cpp
cppa/event_based_actor.hpp
src/event_based_actor.cpp
cppa/stacked_event_based_actor.hpp
src/stacked_event_based_actor.cpp
cppa/event_based_actor_mixin.hpp
#ifndef EVENT_DRIVEN_ACTOR_HPP
#define EVENT_DRIVEN_ACTOR_HPP
#include <stack>
#include <memory>
#include <vector>
#include "cppa/pattern.hpp"
#include "cppa/invoke_rules.hpp"
#include "cppa/util/either.hpp"
#include "cppa/detail/scheduled_actor.hpp"
namespace cppa {
class abstract_event_based_actor : public detail::scheduled_actor
{
typedef detail::scheduled_actor super;
typedef super::queue_node queue_node;
typedef super::queue_node_buffer queue_node_buffer;
protected:
struct stack_element
{
util::either<invoke_rules*, timed_invoke_rules*> m_ptr;
bool m_ownership;
inline stack_element(invoke_rules* ptr, bool take_ownership)
: m_ptr(ptr), m_ownership(take_ownership)
{
}
inline stack_element(invoke_rules&& from)
: m_ptr(new invoke_rules(std::move(from))), m_ownership(true)
{
}
inline stack_element(timed_invoke_rules* ptr, bool take_ownership)
: m_ptr(ptr), m_ownership(take_ownership)
{
}
inline stack_element(timed_invoke_rules&& from)
: m_ptr(new timed_invoke_rules(std::move(from))), m_ownership(true)
{
}
inline ~stack_element()
{
if (m_ownership)
{
if (m_ptr.is_left())
{
delete m_ptr.left();
}
else
{
delete m_ptr.right();
}
}
}
inline bool is_left()
{
return m_ptr.is_left();
}
inline bool is_right()
{
return m_ptr.is_right();
}
inline invoke_rules& left()
{
return *(m_ptr.left());
}
inline timed_invoke_rules& right()
{
return *(m_ptr.right());
}
};
queue_node_buffer m_buffer;
std::stack<stack_element, std::vector<stack_element> > m_loop_stack;
public:
void dequeue(invoke_rules&) /*override*/;
void dequeue(timed_invoke_rules&) /*override*/;
void resume(util::fiber*, resume_callback* callback) /*override*/;
virtual void init() = 0;
virtual void on_exit();
template<typename Scheduler>
abstract_event_based_actor*
attach_to_scheduler(void (*enqueue_fun)(Scheduler*, scheduled_actor*),
Scheduler* sched)
{
m_enqueue_to_scheduler.reset(enqueue_fun, sched, this);
this->init();
return this;
}
private:
void handle_message(std::unique_ptr<queue_node>& node,
invoke_rules& behavior);
void handle_message(std::unique_ptr<queue_node>& node,
timed_invoke_rules& behavior);
void handle_message(std::unique_ptr<queue_node>& node);
protected:
// provoke compiler errors for usage of receive() and related functions
template<typename... Args>
void receive(Args&&...)
{
static_assert(sizeof...(Args) < 0,
"You shall not use receive in an event-based actor. "
"Use become()/unbecome() instead.");
}
template<typename... Args>
void receive_loop(Args&&... args)
{
receive(std::forward<Args>(args)...);
}
template<typename... Args>
void receive_while(Args&&... args)
{
receive(std::forward<Args>(args)...);
}
template<typename... Args>
void do_receive(Args&&... args)
{
receive(std::forward<Args>(args)...);
}
};
} // namespace cppa
#endif // EVENT_DRIVEN_ACTOR_HPP
......@@ -7,7 +7,9 @@
#if defined(__APPLE__)
# define CPPA_MACOS
# ifndef _GLIBCXX_HAS_GTHREADS
# define _GLIBCXX_HAS_GTHREADS
# endif
#elif defined(__GNUC__) && defined(__linux__)
# define CPPA_LINUX
#elif defined(WIN32)
......
......@@ -472,7 +472,7 @@ inline actor_ptr spawn(actor_behavior* what)
return get_scheduler()->spawn(what, Hint);
}
inline actor_ptr spawn(event_based_actor* what)
inline actor_ptr spawn(abstract_event_based_actor* what)
{
return get_scheduler()->spawn(what);
}
......
#ifndef EVENT_DRIVEN_ACTOR_HPP
#define EVENT_DRIVEN_ACTOR_HPP
#include <stack>
#include <memory>
#include "cppa/pattern.hpp"
#include "cppa/invoke_rules.hpp"
#include "cppa/util/either.hpp"
#include "cppa/detail/scheduled_actor.hpp"
namespace cppa { namespace detail {
class abstract_event_based_actor : public scheduled_actor
{
typedef scheduled_actor super;
typedef super::queue_node queue_node;
typedef super::queue_node_buffer queue_node_buffer;
protected:
queue_node_buffer m_buffer;
std::stack< util::either<invoke_rules, timed_invoke_rules> > m_loop_stack;
public:
void dequeue(invoke_rules&) /*override*/;
void dequeue(timed_invoke_rules&) /*override*/;
void resume(util::fiber*, resume_callback* callback) /*override*/;
virtual void on_exit() = 0;
private:
void handle_message(std::unique_ptr<queue_node>& node,
invoke_rules& behavior);
void handle_message(std::unique_ptr<queue_node>& node,
timed_invoke_rules& behavior);
void handle_message(std::unique_ptr<queue_node>& node);
};
} } // namespace cppa::detail
#endif // EVENT_DRIVEN_ACTOR_HPP
......@@ -100,7 +100,7 @@ class mailman_job
mailman_add_peer m_add_socket;
};
constexpr mailman_job(job_type jt) : next(nullptr), m_type(jt) { }
inline mailman_job(job_type jt) : next(nullptr), m_type(jt) { }
};
......
......@@ -10,7 +10,7 @@ class mock_scheduler : public scheduler
public:
actor_ptr spawn(event_based_actor* what);
actor_ptr spawn(abstract_event_based_actor* what);
actor_ptr spawn(actor_behavior*, scheduling_hint);
......
......@@ -29,7 +29,7 @@ class task_scheduler : public scheduler
void schedule(scheduled_actor* what);
actor_ptr spawn(event_based_actor* what);
actor_ptr spawn(abstract_event_based_actor* what);
actor_ptr spawn(actor_behavior*, scheduling_hint);
......
......@@ -22,7 +22,7 @@ class thread_pool_scheduler : public scheduler
void schedule(scheduled_actor* what) /*override*/;
actor_ptr spawn(event_based_actor* what);
actor_ptr spawn(abstract_event_based_actor* what);
actor_ptr spawn(actor_behavior* behavior, scheduling_hint hint);
......
#ifndef EVENT_BASED_ACTOR_HPP
#define EVENT_BASED_ACTOR_HPP
#include "cppa/detail/abstract_event_based_actor.hpp"
#include "cppa/event_based_actor_mixin.hpp"
namespace cppa {
class event_based_actor : public detail::abstract_event_based_actor
class event_based_actor : public event_based_actor_mixin<event_based_actor>
{
inline void become_impl(invoke_rules& rules)
{
become(std::move(rules));
}
friend class event_based_actor_mixin<event_based_actor>;
inline void become_impl(timed_invoke_rules&& rules)
{
become(std::move(rules));
}
typedef abstract_event_based_actor::stack_element stack_element;
template<typename Head, typename... Tail>
void become_impl(invoke_rules& rules, Head&& head, Tail&&... tail)
{
become_impl(rules.splice(std::forward<Head>(head)),
std::forward<Tail>(tail)...);
}
protected:
void unbecome();
void become(invoke_rules&& behavior);
void become(timed_invoke_rules&& behavior);
template<typename Head, typename... Tail>
void become(invoke_rules&& rules, Head&& head, Tail&&... tail)
{
invoke_rules tmp(std::move(rules));
become_impl(tmp.splice(std::forward<Head>(head)),
std::forward<Tail>(tail)...);
}
public:
virtual void init() = 0;
virtual void on_exit();
template<typename Scheduler>
abstract_event_based_actor*
attach_to_scheduler(void (*enqueue_fun)(Scheduler*, scheduled_actor*),
Scheduler* sched)
{
m_enqueue_to_scheduler.reset(enqueue_fun, sched, this);
this->init();
return this;
}
protected:
// provoke compiler errors for usage of receive() and related functions
template<typename... Args>
void receive(Args&&...)
{
static_assert(sizeof...(Args) < 0,
"You shall not use receive in an event-based actor. "
"Use become()/unbecome() instead.");
}
template<typename... Args>
void receive_loop(Args&&... args)
{
receive(std::forward<Args>(args)...);
}
template<typename... Args>
void receive_while(Args&&... args)
{
receive(std::forward<Args>(args)...);
}
template<typename... Args>
void do_receive(Args&&... args)
{
receive(std::forward<Args>(args)...);
}
void clear();
void do_become(invoke_rules* behavior, bool has_ownership);
void do_become(timed_invoke_rules* behavior, bool has_ownership);
};
......
#ifndef EVENT_BASED_ACTOR_MIXIN_HPP
#define EVENT_BASED_ACTOR_MIXIN_HPP
#include "cppa/abstract_event_based_actor.hpp"
namespace cppa {
template<typename Derived>
class event_based_actor_mixin : public abstract_event_based_actor
{
typedef abstract_event_based_actor super;
inline void become_impl(invoke_rules& rules)
{
become(std::move(rules));
}
inline void become_impl(timed_invoke_rules&& rules)
{
become(std::move(rules));
}
template<typename Head, typename... Tail>
void become_impl(invoke_rules& rules, Head&& head, Tail&&... tail)
{
become_impl(rules.splice(std::forward<Head>(head)),
std::forward<Tail>(tail)...);
}
inline Derived* d_this() { return static_cast<Derived*>(this); }
protected:
void become(invoke_rules* behavior)
{
d_this()->do_become(behavior, false);
}
void become(timed_invoke_rules* behavior)
{
d_this()->do_become(behavior, false);
}
void become(invoke_rules&& behavior)
{
d_this()->do_become(new invoke_rules(std::move(behavior)), true);
}
void become(timed_invoke_rules&& behavior)
{
d_this()->do_become(new timed_invoke_rules(std::move(behavior)), true);
}
template<typename Head, typename... Tail>
void become(invoke_rules&& rules, Head&& head, Tail&&... tail)
{
invoke_rules tmp(std::move(rules));
become_impl(tmp.splice(std::forward<Head>(head)),
std::forward<Tail>(tail)...);
}
};
} // namespace cppa
#endif // EVENT_BASED_ACTOR_MIXIN_HPP
......@@ -113,15 +113,17 @@ class timed_invoke_rules : public invoke_rules_base
friend class invoke_rules;
timed_invoke_rules() = delete;
timed_invoke_rules(const timed_invoke_rules&) = delete;
timed_invoke_rules& operator=(const timed_invoke_rules&) = delete;
timed_invoke_rules(invokable_list&& prepended_list,
timed_invoke_rules&& other);
static util::duration default_timeout;
public:
timed_invoke_rules();
timed_invoke_rules(timed_invoke_rules&& arg);
timed_invoke_rules(detail::timed_invokable_ptr&& arg);
......
......@@ -11,7 +11,7 @@
#include "cppa/attachable.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/scheduling_hint.hpp"
#include "cppa/event_based_actor.hpp"
#include "cppa/abstract_event_based_actor.hpp"
#include "cppa/util/duration.hpp"
......@@ -59,7 +59,7 @@ class scheduler
/**
* @brief Spawns a new event-based actor.
*/
virtual actor_ptr spawn(event_based_actor* what) = 0;
virtual actor_ptr spawn(abstract_event_based_actor* what) = 0;
/**
* @brief Informs the scheduler about a converted context
......
#ifndef STACKED_EVENT_BASED_ACTOR_HPP
#define STACKED_EVENT_BASED_ACTOR_HPP
#include "cppa/event_based_actor_mixin.hpp"
namespace cppa {
class stacked_event_based_actor : public event_based_actor_mixin<stacked_event_based_actor>
{
friend class event_based_actor_mixin<stacked_event_based_actor>;
typedef abstract_event_based_actor::stack_element stack_element;
void do_become(invoke_rules* behavior, bool has_ownership);
void do_become(timed_invoke_rules* behavior, bool has_ownership);
protected:
void unbecome();
};
} // namespace cppa
#endif // STACKED_EVENT_BASED_ACTOR_HPP
#include "cppa/detail/invokable.hpp"
#include "cppa/detail/abstract_event_based_actor.hpp"
#include "cppa/abstract_event_based_actor.hpp"
namespace cppa { namespace detail {
namespace cppa {
void abstract_event_based_actor::dequeue(invoke_rules&)
{
......@@ -91,9 +91,23 @@ void abstract_event_based_actor::resume(util::fiber*, resume_callback* callback)
}
else
{
// nothing to do
// nothing to do (wait for new messages)
switch (compare_exchange_state(scheduled_actor::about_to_block,
scheduled_actor::blocked))
{
case scheduled_actor::ready:
{
// got a new job
break;
}
case scheduled_actor::blocked:
{
// done
return;
}
default: exit(7); // illegal state
};
}
}
node.reset(m_mailbox.pop());
try
......@@ -119,4 +133,8 @@ void abstract_event_based_actor::resume(util::fiber*, resume_callback* callback)
while (callback->still_ready());
}
} } // namespace cppa::detail
void abstract_event_based_actor::on_exit()
{
}
} // namespace cppa
......@@ -2,25 +2,25 @@
namespace cppa {
void event_based_actor::on_exit()
{
}
void event_based_actor::unbecome()
void event_based_actor::clear()
{
if (!m_loop_stack.empty()) m_loop_stack.pop();
}
void event_based_actor::become(invoke_rules&& behavior)
void event_based_actor::do_become(invoke_rules* behavior,
bool has_ownership)
{
clear();
reset_timeout();
m_loop_stack.push(std::move(behavior));
m_loop_stack.push(stack_element(behavior, has_ownership));
}
void event_based_actor::become(timed_invoke_rules&& behavior)
void event_based_actor::do_become(timed_invoke_rules* behavior,
bool has_ownership)
{
request_timeout(behavior.timeout());
m_loop_stack.push(std::move(behavior));
clear();
request_timeout(behavior->timeout());
m_loop_stack.push(stack_element(behavior, has_ownership));
}
} // namespace cppa
......@@ -33,6 +33,7 @@
namespace cppa {
util::duration timed_invoke_rules::default_timeout;
// invoke_rules_base
invoke_rules_base::invoke_rules_base(invoke_rules_base&& other)
......@@ -72,6 +73,10 @@ invoke_rules_base::get_intermediate(const any_tuple& t) const
// timed_invoke_rules
timed_invoke_rules::timed_invoke_rules()
{
}
timed_invoke_rules::timed_invoke_rules(timed_invoke_rules&& other)
: super(std::move(other)), m_ti(std::move(other.m_ti))
{
......@@ -99,13 +104,13 @@ timed_invoke_rules& timed_invoke_rules::operator=(timed_invoke_rules&& other)
const util::duration& timed_invoke_rules::timeout() const
{
return m_ti->timeout();
return (m_ti != nullptr) ? m_ti->timeout() : default_timeout;
}
void timed_invoke_rules::handle_timeout() const
{
// safe, because timed_invokable ignores the given argument
m_ti->invoke(*static_cast<any_tuple*>(nullptr));
if (m_ti != nullptr) m_ti->invoke(*static_cast<any_tuple*>(nullptr));
}
// invoke_rules
......@@ -145,6 +150,11 @@ invoke_rules invoke_rules::operator,(invoke_rules&& other)
return std::move(m_list);
}
timed_invoke_rules invoke_rules::operator,(timed_invoke_rules&& other)
{
return timed_invoke_rules(std::move(m_list), std::move(other));
}
invoke_rules& invoke_rules::operator=(invoke_rules&& other)
{
m_list = std::move(other.m_list);
......
......@@ -50,7 +50,7 @@ actor_ptr mock_scheduler::spawn(actor_behavior* behavior)
return ctx;
}
actor_ptr mock_scheduler::spawn(event_based_actor* what)
actor_ptr mock_scheduler::spawn(abstract_event_based_actor* what)
{
// TODO: don't delete what :)
delete what;
......
......@@ -28,12 +28,6 @@ void scheduled_actor::quit(std::uint32_t reason)
//yield(yield_state::done);
}
void scheduled_actor::request_timeout(const util::duration& d)
{
future_send(this, d, atom(":Timeout"), ++m_active_timeout_id);
m_has_pending_timeout_request = true;
}
void scheduled_actor::enqueue_node(queue_node* node)
{
if (m_mailbox._push_back(node))
......@@ -90,6 +84,12 @@ int scheduled_actor::compare_exchange_state(int expected, int new_value)
return e;
}
void scheduled_actor::request_timeout(const util::duration& d)
{
future_send(this, d, atom(":Timeout"), ++m_active_timeout_id);
m_has_pending_timeout_request = true;
}
scheduled_actor::filter_result scheduled_actor::filter_msg(const any_tuple& msg)
{
if (m_pattern(msg))
......
#include "cppa/stacked_event_based_actor.hpp"
namespace cppa {
void stacked_event_based_actor::unbecome()
{
if (!m_loop_stack.empty()) m_loop_stack.pop();
}
void stacked_event_based_actor::do_become(invoke_rules* behavior,
bool has_ownership)
{
reset_timeout();
m_loop_stack.push(stack_element(behavior, has_ownership));
}
void stacked_event_based_actor::do_become(timed_invoke_rules* behavior,
bool has_ownership)
{
request_timeout(behavior->timeout());
m_loop_stack.push(stack_element(behavior, has_ownership));
}
} // namespace cppa
......@@ -99,7 +99,7 @@ actor_ptr task_scheduler::spawn_impl(scheduled_actor* what)
}
actor_ptr task_scheduler::spawn(event_based_actor* what)
actor_ptr task_scheduler::spawn(abstract_event_based_actor* what)
{
return spawn_impl(what->attach_to_scheduler(enqueue_fun, this));
}
......
......@@ -217,7 +217,7 @@ actor_ptr thread_pool_scheduler::spawn_impl(scheduled_actor* what)
}
actor_ptr thread_pool_scheduler::spawn(event_based_actor* what)
actor_ptr thread_pool_scheduler::spawn(abstract_event_based_actor* what)
{
return spawn_impl(what->attach_to_scheduler(enqueue_fun, this));
}
......
......@@ -74,12 +74,13 @@ void yielding_actor::dequeue(timed_invoke_rules& rules)
{
queue_node_buffer buffer;
// try until a message was successfully dequeued
request_timeout(rules.timeout());
for (;;)
{
if (m_mailbox.empty() && has_pending_timeout() == false)
{
request_timeout(rules.timeout());
}
//if (m_mailbox.empty() && has_pending_timeout() == false)
//{
// request_timeout(rules.timeout());
//}
yield_until_not_empty();
std::unique_ptr<queue_node> node(m_mailbox.pop());
switch (dq(node, rules, buffer))
......
......@@ -15,6 +15,7 @@
#include "cppa/to_string.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/event_based_actor.hpp"
#include "cppa/stacked_event_based_actor.hpp"
using std::cerr;
using std::cout;
......@@ -22,37 +23,64 @@ using std::endl;
using namespace cppa;
// GCC 4.7 supports non-static member initialization
#if (__GNUC__ >= 4) && (__GNUC_MINOR__ >= 7)
class event_testee : public event_based_actor
{
invoke_rules wait4string()
{
return on<std::string>() >> [=](const std::string& value)
invoke_rules wait4string =
(
on<std::string>() >> [=](const std::string& value)
{
cout << "event_testee[string]: " << value << endl;
// switch back to wait4int
unbecome();
unbecome();
};
}
);
invoke_rules wait4float()
{
return on<float>() >> [=](float value)
invoke_rules wait4float =
(
on<float>() >> [=](float value)
{
cout << "event_testee[float]: " << value << endl;
become(wait4string());
};
become(wait4string);
}
);
invoke_rules wait4int()
{
return on<int>() >> [=](int value)
invoke_rules wait4int =
(
on<int>() >> [=](int value)
{
cout << "event_testee[int]: " << value << endl;
become(wait4float());
};
become(wait4float);
}
);
public:
void on_exit()
{
cout << "event_testee[GCC47]::on_exit()" << endl;
}
void init()
{
cout << "event_testee[GCC47]::init()" << endl;
become(wait4int);
}
};
#else
class event_testee : public event_based_actor
{
invoke_rules wait4string;
invoke_rules wait4float;
invoke_rules wait4int;
public:
......@@ -63,21 +91,52 @@ class event_testee : public event_based_actor
void init()
{
wait4string =
(
on<std::string>() >> [=](const std::string& value)
{
cout << "event_testee[string]: " << value << endl;
become(&wait4int);
}
);
wait4float =
(
on<float>() >> [=](float value)
{
cout << "event_testee[float]: " << value << endl;
become(&wait4string);
}
);
wait4int =
(
on<int>() >> [=](int value)
{
cout << "event_testee[int]: " << value << endl;
become(&wait4float);
}
);
cout << "event_testee::init()" << endl;
become(wait4int());
become(&wait4int);
}
};
event_based_actor* event_testee2()
#endif
abstract_event_based_actor* event_testee2()
{
struct impl : event_based_actor
{
int num_timeouts;
impl() : num_timeouts(0) { }
timed_invoke_rules state;
void init()
{
become
cout << "event_testee2::impl::init()" << endl;
state =
//become
(
others() >> []()
{
......@@ -85,13 +144,17 @@ event_based_actor* event_testee2()
<< to_string(last_received())
<< endl;
},
after(std::chrono::milliseconds(50)) >> [this]()
after(std::chrono::milliseconds(50)) >> [=]()
{
cout << "testee2 received timeout nr. "
<< (num_timeouts + 1) << endl;
if (++num_timeouts >= 5) unbecome();
if (++num_timeouts >= 5)
{
quit(exit_reason::normal);
}
}
);
become(&state);
}
};
return new impl;
......@@ -142,6 +205,7 @@ void testee1()
(
after(std::chrono::milliseconds(10)) >> []()
{
cout << "testee1::quit" << endl;
quit(exit_reason::user_defined);
}
);
......@@ -205,13 +269,9 @@ size_t test__spawn()
CPPA_TEST(test__spawn);
//spawn(testee1);
//spawn(new testee_behavior);
spawn(event_testee2());
//await_all_others_done();
//spawn(event_testee2());
//auto et = spawn(new event_testee);
await_all_others_done();
behavior_test<testee_behavior>();
behavior_test<event_testee>();
......
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