Commit 920472e6 authored by Dominik Charousset's avatar Dominik Charousset

fsm_actor and scheduled_actor

parent 99626f2c
......@@ -5,7 +5,7 @@ lib_LTLIBRARIES = libcppa.la
libcppa_la_SOURCES = \
src/abstract_event_based_actor.cpp \
src/abstract_tuple.cpp \
src/actor_behavior.cpp \
src/scheduled_actor.cpp \
src/actor_count.cpp \
src/actor.cpp \
src/actor_proxy_cache.cpp \
......@@ -46,7 +46,7 @@ libcppa_la_SOURCES = \
src/primitive_variant.cpp \
src/process_information.cpp \
src/ripemd_160.cpp \
src/scheduled_actor.cpp \
src/abstract_scheduled_actor.cpp \
src/scheduler.cpp \
src/serializer.cpp \
src/shared_spinlock.cpp \
......@@ -68,7 +68,8 @@ library_includedir = $(includedir)/
endif
nobase_library_include_HEADERS = \
cppa/actor_behavior.hpp \
cppa/abstract_actor.hpp \
cppa/scheduled_actor.hpp \
cppa/actor.hpp \
cppa/actor_proxy.hpp \
cppa/announce.hpp \
......@@ -86,7 +87,6 @@ nobase_library_include_HEADERS = \
cppa/deserializer.hpp \
cppa/event_based_actor.hpp \
cppa/event_based_actor_mixin.hpp \
cppa/detail/abstract_actor.hpp \
cppa/abstract_event_based_actor.hpp \
cppa/detail/abstract_tuple.hpp \
cppa/detail/actor_count.hpp \
......@@ -124,7 +124,7 @@ nobase_library_include_HEADERS = \
cppa/detail/ptype_to_type.hpp \
cppa/detail/receive_loop_helper.hpp \
cppa/detail/ref_counted_impl.hpp \
cppa/detail/scheduled_actor.hpp \
cppa/detail/abstract_scheduled_actor.hpp \
cppa/detail/serialize_tuple.hpp \
cppa/detail/singleton_manager.hpp \
cppa/detail/swap_bytes.hpp \
......@@ -141,6 +141,7 @@ nobase_library_include_HEADERS = \
cppa/exception.hpp \
cppa/exit_reason.hpp \
cppa/from_string.hpp \
cppa/fsm_actor.hpp \
cppa/get.hpp \
cppa/get_view.hpp \
cppa/group.hpp \
......
......@@ -69,10 +69,10 @@ cppa/util/eval_first_n.hpp
cppa/local_actor.hpp
cppa/channel.hpp
src/channel.cpp
cppa/actor_behavior.hpp
cppa/scheduled_actor.hpp
cppa/cppa.hpp
cppa/scheduling_hint.hpp
src/actor_behavior.cpp
src/scheduled_actor.cpp
cppa/group.hpp
src/group.cpp
queue_performances/main.cpp
......@@ -163,7 +163,7 @@ cppa/detail/atom_val.hpp
src/atom.cpp
src/cppa.cpp
cppa/exit_reason.hpp
cppa/detail/abstract_actor.hpp
cppa/abstract_actor.hpp
cppa/detail/mailman.hpp
src/mailman.cpp
cppa/detail/native_socket.hpp
......@@ -179,8 +179,8 @@ cppa/util/fiber.hpp
src/fiber.cpp
cppa/detail/yield_interface.hpp
src/yield_interface.cpp
cppa/detail/scheduled_actor.hpp
src/scheduled_actor.cpp
cppa/detail/abstract_scheduled_actor.hpp
src/abstract_scheduled_actor.cpp
src/intermediate.cpp
src/invokable.cpp
cppa/detail/thread_pool_scheduler.hpp
......@@ -239,3 +239,4 @@ src/event_based_actor.cpp
cppa/stacked_event_based_actor.hpp
src/stacked_event_based_actor.cpp
cppa/event_based_actor_mixin.hpp
cppa/fsm_actor.hpp
......@@ -16,7 +16,7 @@
#include "cppa/attachable.hpp"
#include "cppa/exit_reason.hpp"
namespace cppa { namespace detail {
namespace cppa {
// implements linking and monitoring for actors
template<class Base>
......@@ -257,6 +257,6 @@ class abstract_actor : public Base
};
} } // namespace cppa::detail
} // namespace cppa
#endif // ABSTRACT_ACTOR_HPP
......@@ -8,17 +8,55 @@
#include "cppa/pattern.hpp"
#include "cppa/invoke_rules.hpp"
#include "cppa/util/either.hpp"
#include "cppa/detail/scheduled_actor.hpp"
#include "cppa/detail/abstract_scheduled_actor.hpp"
namespace cppa {
class abstract_event_based_actor : public detail::scheduled_actor
class abstract_event_based_actor : public detail::abstract_scheduled_actor
{
typedef detail::scheduled_actor super;
typedef detail::abstract_scheduled_actor super;
typedef super::queue_node queue_node;
typedef super::queue_node_buffer queue_node_buffer;
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*, abstract_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:
struct stack_element
......@@ -29,18 +67,10 @@ class abstract_event_based_actor : public detail::scheduled_actor
: 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)
......@@ -76,40 +106,6 @@ class abstract_event_based_actor : public detail::scheduled_actor
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>
......
......@@ -2,17 +2,17 @@
#define ACTOR_PROXY_HPP
#include "cppa/actor.hpp"
#include "cppa/detail/abstract_actor.hpp"
#include "cppa/abstract_actor.hpp"
namespace cppa {
/**
* @brief Represents a remote Actor.
*/
class actor_proxy : public detail::abstract_actor<actor>
class actor_proxy : public abstract_actor<actor>
{
typedef detail::abstract_actor<actor> super;
typedef abstract_actor<actor> super;
void forward_message(const process_information_ptr&,
actor*, const any_tuple&);
......
......@@ -47,7 +47,7 @@
#include "cppa/local_actor.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/invoke_rules.hpp"
#include "cppa/actor_behavior.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/scheduling_hint.hpp"
#include "cppa/event_based_actor.hpp"
......@@ -461,13 +461,13 @@ inline void trap_exit(bool new_value)
self()->trap_exit(new_value);
}
inline actor_ptr spawn(actor_behavior* what)
inline actor_ptr spawn(scheduled_actor* what)
{
return get_scheduler()->spawn(what, scheduled);
}
template<scheduling_hint Hint>
inline actor_ptr spawn(actor_behavior* what)
inline actor_ptr spawn(scheduled_actor* what)
{
return get_scheduler()->spawn(what, Hint);
}
......@@ -487,7 +487,7 @@ inline actor_ptr spawn(abstract_event_based_actor* what)
template<scheduling_hint Hint, typename F, typename... Args>
auto //actor_ptr
spawn(F&& what, const Args&... args)
-> typename util::disable_if_c< std::is_convertible<typename util::rm_ref<F>::type, actor_behavior*>::value
-> typename util::disable_if_c< std::is_convertible<typename util::rm_ref<F>::type, scheduled_actor*>::value
|| std::is_convertible<typename util::rm_ref<F>::type, event_based_actor*>::value,
actor_ptr>::type
{
......@@ -506,7 +506,7 @@ spawn(F&& what, const Args&... args)
template<typename F, typename... Args>
auto // actor_ptr
spawn(F&& what, const Args&... args)
-> typename util::disable_if_c< std::is_convertible<typename util::rm_ref<F>::type, actor_behavior*>::value
-> typename util::disable_if_c< std::is_convertible<typename util::rm_ref<F>::type, scheduled_actor*>::value
|| std::is_convertible<typename util::rm_ref<F>::type, event_based_actor*>::value,
actor_ptr>::type
{
......
......@@ -2,14 +2,14 @@
#define SCHEDULED_ACTOR_HPP
#include "cppa/local_actor.hpp"
#include "cppa/actor_behavior.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/util/fiber.hpp"
#include "cppa/util/singly_linked_list.hpp"
#include "cppa/util/single_reader_queue.hpp"
#include "cppa/detail/delegate.hpp"
#include "cppa/detail/abstract_actor.hpp"
namespace cppa { namespace detail {
......@@ -18,13 +18,13 @@ class task_scheduler;
/**
* @brief A spawned, scheduled Actor.
*/
class scheduled_actor : public abstract_actor<local_actor>
class abstract_scheduled_actor : public abstract_actor<local_actor>
{
friend class task_scheduler;
friend class util::single_reader_queue<scheduled_actor>;
friend class util::single_reader_queue<abstract_scheduled_actor>;
scheduled_actor* next; // intrusive next pointer (single_reader_queue)
abstract_scheduled_actor* next; // intrusive next pointer (single_reader_queue)
void enqueue_node(queue_node* node);
......@@ -33,7 +33,7 @@ class scheduled_actor : public abstract_actor<local_actor>
std::atomic<int> m_state;
delegate m_enqueue_to_scheduler;
typedef abstract_actor<local_actor> super;
typedef abstract_actor super;
typedef super::queue_node queue_node;
typedef util::singly_linked_list<queue_node> queue_node_buffer;
......@@ -87,10 +87,11 @@ class scheduled_actor : public abstract_actor<local_actor>
static constexpr int blocked = 0x02;
static constexpr int about_to_block = 0x04;
scheduled_actor();
abstract_scheduled_actor();
template<typename Scheduler>
scheduled_actor(void (*enqueue_fun)(Scheduler*, scheduled_actor*),
abstract_scheduled_actor(void (*enqueue_fun)(Scheduler*,
abstract_scheduled_actor*),
Scheduler* sched)
: next(nullptr)
, m_state(ready)
......@@ -106,8 +107,6 @@ class scheduled_actor : public abstract_actor<local_actor>
void enqueue(actor* sender, const any_tuple& msg);
//inline std::atomic<int>& state() { return m_mailbox.m_state; }
int compare_exchange_state(int expected, int new_value);
struct resume_callback
......@@ -126,7 +125,7 @@ class scheduled_actor : public abstract_actor<local_actor>
};
struct scheduled_actor_dummy : scheduled_actor
struct scheduled_actor_dummy : abstract_scheduled_actor
{
void resume(util::fiber*, resume_callback*);
void quit(std::uint32_t);
......
......@@ -16,8 +16,7 @@
#include "cppa/pattern.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/detail/abstract_actor.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/util/either.hpp"
#include "cppa/util/singly_linked_list.hpp"
......
......@@ -6,13 +6,13 @@
#include "cppa/invoke.hpp"
#include "cppa/util/rm_ref.hpp"
#include "cppa/detail/tdata.hpp"
#include "cppa/actor_behavior.hpp"
#include "cppa/scheduled_actor.hpp"
namespace cppa { namespace detail {
// default: <true, false, F>
template<bool IsFunctionPtr, bool HasArguments, typename F, typename... Args>
class ftor_behavior : public actor_behavior
class ftor_behavior : public scheduled_actor
{
F m_fun;
......@@ -26,7 +26,7 @@ class ftor_behavior : public actor_behavior
};
template<typename F, typename... Args>
class ftor_behavior<true, true, F, Args...> : public actor_behavior
class ftor_behavior<true, true, F, Args...> : public scheduled_actor
{
static_assert(sizeof...(Args) > 0, "sizeof...(Args) == 0");
......@@ -43,7 +43,7 @@ class ftor_behavior<true, true, F, Args...> : public actor_behavior
};
template<typename F>
class ftor_behavior<false, false, F> : public actor_behavior
class ftor_behavior<false, false, F> : public scheduled_actor
{
F m_fun;
......@@ -59,7 +59,7 @@ class ftor_behavior<false, false, F> : public actor_behavior
};
template<typename F, typename... Args>
class ftor_behavior<false, true, F, Args...> : public actor_behavior
class ftor_behavior<false, true, F, Args...> : public scheduled_actor
{
static_assert(sizeof...(Args) > 0, "sizeof...(Args) == 0");
......@@ -83,9 +83,9 @@ class ftor_behavior<false, true, F, Args...> : public actor_behavior
};
template<typename R>
actor_behavior* get_behavior(std::integral_constant<bool,true>, R (*fptr)())
scheduled_actor* get_behavior(std::integral_constant<bool,true>, R (*fptr)())
{
static_assert(std::is_convertible<R, actor_behavior*>::value == false,
static_assert(std::is_convertible<R, scheduled_actor*>::value == false,
"Spawning a function returning an actor_behavior? "
"Are you sure that you do not want to spawn the behavior "
"returned by that function?");
......@@ -93,9 +93,9 @@ actor_behavior* get_behavior(std::integral_constant<bool,true>, R (*fptr)())
}
template<typename F>
actor_behavior* get_behavior(std::integral_constant<bool,false>, F&& ftor)
scheduled_actor* get_behavior(std::integral_constant<bool,false>, F&& ftor)
{
static_assert(std::is_convertible<decltype(ftor()), actor_behavior*>::value == false,
static_assert(std::is_convertible<decltype(ftor()), scheduled_actor*>::value == false,
"Spawning a functor returning an actor_behavior? "
"Are you sure that you do not want to spawn the behavior "
"returned by that functor?");
......@@ -104,12 +104,12 @@ actor_behavior* get_behavior(std::integral_constant<bool,false>, F&& ftor)
}
template<typename F, typename Arg0, typename... Args>
actor_behavior* get_behavior(std::integral_constant<bool,true>,
scheduled_actor* get_behavior(std::integral_constant<bool,true>,
F fptr,
const Arg0& arg0,
const Args&... args)
{
static_assert(std::is_convertible<decltype(fptr(arg0, args...)), actor_behavior*>::value == false,
static_assert(std::is_convertible<decltype(fptr(arg0, args...)), scheduled_actor*>::value == false,
"Spawning a function returning an actor_behavior? "
"Are you sure that you do not want to spawn the behavior "
"returned by that function?");
......@@ -118,12 +118,12 @@ actor_behavior* get_behavior(std::integral_constant<bool,true>,
}
template<typename F, typename Arg0, typename... Args>
actor_behavior* get_behavior(std::integral_constant<bool,false>,
scheduled_actor* get_behavior(std::integral_constant<bool,false>,
F ftor,
const Arg0& arg0,
const Args&... args)
{
static_assert(std::is_convertible<decltype(ftor(arg0, args...)), actor_behavior*>::value == false,
static_assert(std::is_convertible<decltype(ftor(arg0, args...)), scheduled_actor*>::value == false,
"Spawning a functor returning an actor_behavior? "
"Are you sure that you do not want to spawn the behavior "
"returned by that functor?");
......
......@@ -12,9 +12,9 @@ class mock_scheduler : public scheduler
actor_ptr spawn(abstract_event_based_actor* what);
actor_ptr spawn(actor_behavior*, scheduling_hint);
actor_ptr spawn(scheduled_actor*, scheduling_hint);
static actor_ptr spawn(actor_behavior*);
static actor_ptr spawn(scheduled_actor*);
};
......
......@@ -3,7 +3,7 @@
#include "cppa/scheduler.hpp"
#include "cppa/detail/thread.hpp"
#include "cppa/detail/scheduled_actor.hpp"
#include "cppa/detail/abstract_scheduled_actor.hpp"
#include "cppa/util/single_reader_queue.hpp"
namespace cppa { namespace detail {
......@@ -13,13 +13,13 @@ class task_scheduler : public scheduler
typedef scheduler super;
typedef util::single_reader_queue<scheduled_actor> job_queue;
typedef util::single_reader_queue<abstract_scheduled_actor> job_queue;
job_queue m_queue;
scheduled_actor_dummy m_dummy;
thread m_worker;
static void worker_loop(job_queue*, scheduled_actor* dummy);
static void worker_loop(job_queue*, abstract_scheduled_actor* dummy);
public:
......@@ -27,15 +27,15 @@ class task_scheduler : public scheduler
virtual void stop();
void schedule(scheduled_actor* what);
void schedule(abstract_scheduled_actor* what);
actor_ptr spawn(abstract_event_based_actor* what);
actor_ptr spawn(actor_behavior*, scheduling_hint);
actor_ptr spawn(scheduled_actor*, scheduling_hint);
private:
actor_ptr spawn_impl(scheduled_actor* what);
actor_ptr spawn_impl(abstract_scheduled_actor* what);
};
......
......@@ -3,7 +3,7 @@
#include "cppa/scheduler.hpp"
#include "cppa/detail/thread.hpp"
#include "cppa/detail/scheduled_actor.hpp"
#include "cppa/detail/abstract_scheduled_actor.hpp"
namespace cppa { namespace detail {
......@@ -20,24 +20,24 @@ class thread_pool_scheduler : public scheduler
void stop() /*override*/;
void schedule(scheduled_actor* what) /*override*/;
void schedule(abstract_scheduled_actor* what) /*override*/;
actor_ptr spawn(abstract_event_based_actor* what);
actor_ptr spawn(actor_behavior* behavior, scheduling_hint hint);
actor_ptr spawn(scheduled_actor* behavior, scheduling_hint hint);
private:
typedef util::single_reader_queue<scheduled_actor> job_queue;
typedef util::single_reader_queue<abstract_scheduled_actor> job_queue;
job_queue m_queue;
scheduled_actor_dummy m_dummy;
thread m_supervisor;
actor_ptr spawn_impl(scheduled_actor* what);
actor_ptr spawn_impl(abstract_scheduled_actor* what);
static void worker_loop(worker*);
static void supervisor_loop(job_queue*, scheduled_actor*);
static void supervisor_loop(job_queue*, abstract_scheduled_actor*);
};
......
......@@ -6,7 +6,7 @@
#include "cppa/pattern.hpp"
#include "cppa/detail/delegate.hpp"
#include "cppa/detail/scheduled_actor.hpp"
#include "cppa/detail/abstract_scheduled_actor.hpp"
#include "cppa/detail/yield_interface.hpp"
#include "cppa/util/either.hpp"
......@@ -14,15 +14,15 @@
namespace cppa { namespace detail {
class yielding_actor : public scheduled_actor
class yielding_actor : public abstract_scheduled_actor
{
typedef scheduled_actor super;
typedef abstract_scheduled_actor super;
typedef super::queue_node queue_node;
typedef util::singly_linked_list<queue_node> queue_node_buffer;
util::fiber m_fiber;
actor_behavior* m_behavior;
scheduled_actor* m_behavior;
static void run(void* _this);
......@@ -33,8 +33,8 @@ class yielding_actor : public scheduled_actor
public:
template<typename Scheduler>
yielding_actor(actor_behavior* behavior,
void (*enqueue_fun)(Scheduler*, scheduled_actor*),
yielding_actor(scheduled_actor* behavior,
void (*enqueue_fun)(Scheduler*, abstract_scheduled_actor*),
Scheduler* sched)
: super(enqueue_fun, sched)
, m_fiber(&yielding_actor::run, this)
......
#ifndef FSM_ACTOR_HPP
#define FSM_ACTOR_HPP
#include "cppa/event_based_actor.hpp"
namespace cppa {
template<class Derived>
class fsm_actor : public event_based_actor
{
public:
void init()
{
become(&(static_cast<Derived*>(this)->init_state));
}
};
} // namespace cppa
#endif // FSM_ACTOR_HPP
......@@ -3,12 +3,12 @@
namespace cppa {
class actor_behavior
class scheduled_actor
{
public:
virtual ~actor_behavior();
virtual ~scheduled_actor();
virtual void on_exit();
virtual void act() = 0;
......
......@@ -18,7 +18,7 @@
namespace cppa {
//class local_actor;
class actor_behavior;
class scheduled_actor;
class scheduler_helper;
/**
......@@ -53,7 +53,7 @@ class scheduler
* @brief Spawns a new actor that executes <code>behavior->act()</code>
* with the scheduling policy @p hint if possible.
*/
virtual actor_ptr spawn(actor_behavior* behavior,
virtual actor_ptr spawn(scheduled_actor* behavior,
scheduling_hint hint) = 0;
/**
......
......@@ -64,7 +64,7 @@ void abstract_event_based_actor::resume(util::fiber*, resume_callback* callback)
set_self(this);
auto done_cb = [&]()
{
m_state.store(scheduled_actor::done);
m_state.store(abstract_scheduled_actor::done);
while (!m_loop_stack.empty()) m_loop_stack.pop();
on_exit();
callback->exec_done();
......@@ -82,25 +82,25 @@ void abstract_event_based_actor::resume(util::fiber*, resume_callback* callback)
}
else if (m_mailbox.empty())
{
m_state.store(scheduled_actor::about_to_block);
m_state.store(abstract_scheduled_actor::about_to_block);
CPPA_MEMORY_BARRIER();
if (!m_mailbox.empty())
{
// someone preempt us
m_state.store(scheduled_actor::ready);
m_state.store(abstract_scheduled_actor::ready);
}
else
{
// nothing to do (wait for new messages)
switch (compare_exchange_state(scheduled_actor::about_to_block,
scheduled_actor::blocked))
switch (compare_exchange_state(abstract_scheduled_actor::about_to_block,
abstract_scheduled_actor::blocked))
{
case scheduled_actor::ready:
case abstract_scheduled_actor::ready:
{
// got a new job
break;
}
case scheduled_actor::blocked:
case abstract_scheduled_actor::blocked:
{
// done
return;
......
#include "cppa/cppa.hpp"
#include "cppa/exception.hpp"
#include "cppa/detail/task_scheduler.hpp"
#include "cppa/detail/abstract_scheduled_actor.hpp"
#include "cppa/detail/yield_interface.hpp"
namespace { void dummy_enqueue(void*, cppa::detail::abstract_scheduled_actor*) { } }
namespace cppa { namespace detail {
abstract_scheduled_actor::abstract_scheduled_actor()
: next(nullptr)
, m_state(abstract_scheduled_actor::done)
, m_enqueue_to_scheduler(dummy_enqueue, static_cast<void*>(nullptr), this)
, m_has_pending_timeout_request(false)
, m_active_timeout_id(0)
{
}
abstract_scheduled_actor::resume_callback::~resume_callback()
{
}
void abstract_scheduled_actor::quit(std::uint32_t reason)
{
cleanup(reason);
throw actor_exited(reason);
//yield(yield_state::done);
}
void abstract_scheduled_actor::enqueue_node(queue_node* node)
{
if (m_mailbox._push_back(node))
{
for (;;)
{
int state = m_state.load();
switch (state)
{
case blocked:
{
if (m_state.compare_exchange_weak(state, ready))
{
m_enqueue_to_scheduler();
return;
}
break;
}
case about_to_block:
{
if (m_state.compare_exchange_weak(state, ready))
{
return;
}
break;
}
default: return;
}
}
}
}
void abstract_scheduled_actor::enqueue(actor* sender, any_tuple&& msg)
{
enqueue_node(new queue_node(sender, std::move(msg)));
}
void abstract_scheduled_actor::enqueue(actor* sender, const any_tuple& msg)
{
enqueue_node(new queue_node(sender, msg));
}
int abstract_scheduled_actor::compare_exchange_state(int expected, int new_value)
{
int e = expected;
do
{
if (m_state.compare_exchange_weak(e, new_value))
{
return new_value;
}
}
while (e == expected);
return e;
}
void abstract_scheduled_actor::request_timeout(const util::duration& d)
{
future_send(this, d, atom(":Timeout"), ++m_active_timeout_id);
m_has_pending_timeout_request = true;
}
abstract_scheduled_actor::filter_result abstract_scheduled_actor::filter_msg(const any_tuple& msg)
{
if (m_pattern(msg))
{
auto v0 = *reinterpret_cast<const atom_value*>(msg.at(0));
auto v1 = *reinterpret_cast<const std::uint32_t*>(msg.at(1));
if (v0 == atom(":Exit"))
{
if (m_trap_exit == false)
{
if (v1 != exit_reason::normal)
{
quit(v1);
}
return normal_exit_signal;
}
}
else if (v0 == atom(":Timeout"))
{
return (v1 == m_active_timeout_id) ? timeout_message
: expired_timeout_message;
}
}
return ordinary_message;
}
abstract_scheduled_actor::dq_result abstract_scheduled_actor::dq(std::unique_ptr<queue_node>& node,
invoke_rules_base& rules,
queue_node_buffer& buffer)
{
switch (filter_msg(node->msg))
{
case normal_exit_signal:
case expired_timeout_message:
{
// skip message
return dq_indeterminate;
}
case timeout_message:
{
// m_active_timeout_id is already invalid
m_has_pending_timeout_request = false;
// restore mailbox before calling client
if (!buffer.empty())
{
m_mailbox.push_front(std::move(buffer));
buffer.clear();
}
return dq_timeout_occured;
}
default: break;
}
auto imd = rules.get_intermediate(node->msg);
if (imd)
{
m_last_dequeued = std::move(node->msg);
m_last_sender = std::move(node->sender);
// restore mailbox before invoking imd
if (!buffer.empty())
{
m_mailbox.push_front(std::move(buffer));
buffer.clear();
}
// expire pending request
if (m_has_pending_timeout_request)
{
++m_active_timeout_id;
m_has_pending_timeout_request = false;
}
imd->invoke();
return dq_done;
}
else
{
buffer.push_back(node.release());
return dq_indeterminate;
}
}
// dummy
void scheduled_actor_dummy::resume(util::fiber*, resume_callback*)
{
}
void scheduled_actor_dummy::quit(std::uint32_t)
{
}
void scheduled_actor_dummy::dequeue(invoke_rules&)
{
}
void scheduled_actor_dummy::dequeue(timed_invoke_rules&)
{
}
void scheduled_actor_dummy::link_to(intrusive_ptr<actor>&)
{
}
void scheduled_actor_dummy::unlink_from(intrusive_ptr<actor>&)
{
}
bool scheduled_actor_dummy::establish_backlink(intrusive_ptr<actor>&)
{
return false;
}
bool scheduled_actor_dummy::remove_backlink(intrusive_ptr<actor>&)
{
return false;
}
void scheduled_actor_dummy::detach(const attachable::token&)
{
}
bool scheduled_actor_dummy::attach(attachable*)
{
return false;
}
} } // namespace cppa::detail
#include "cppa/actor_behavior.hpp"
namespace cppa {
actor_behavior::~actor_behavior()
{
}
void actor_behavior::on_exit()
{
}
} // namespace cppa
......@@ -9,7 +9,7 @@
#include "cppa/scheduler.hpp"
#include "cppa/attachable.hpp"
#include "cppa/invoke_rules.hpp"
#include "cppa/actor_behavior.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/detail/thread.hpp"
#include "cppa/detail/actor_count.hpp"
......@@ -23,7 +23,7 @@ using std::endl;
namespace {
void run_actor(cppa::intrusive_ptr<cppa::local_actor> m_self,
cppa::actor_behavior* behavior)
cppa::scheduled_actor* behavior)
{
cppa::set_self(m_self.get());
if (behavior)
......@@ -41,7 +41,7 @@ void run_actor(cppa::intrusive_ptr<cppa::local_actor> m_self,
namespace cppa { namespace detail {
actor_ptr mock_scheduler::spawn(actor_behavior* behavior)
actor_ptr mock_scheduler::spawn(scheduled_actor* behavior)
{
inc_actor_count();
CPPA_MEMORY_BARRIER();
......@@ -57,7 +57,7 @@ actor_ptr mock_scheduler::spawn(abstract_event_based_actor* what)
return nullptr;
}
actor_ptr mock_scheduler::spawn(actor_behavior* behavior, scheduling_hint)
actor_ptr mock_scheduler::spawn(scheduled_actor* behavior, scheduling_hint)
{
return spawn(behavior);
}
......
#include "cppa/cppa.hpp"
#include "cppa/exception.hpp"
#include "cppa/detail/task_scheduler.hpp"
#include "cppa/detail/scheduled_actor.hpp"
#include "cppa/detail/yield_interface.hpp"
#include "cppa/scheduled_actor.hpp"
namespace { void dummy_enqueue(void*, cppa::detail::scheduled_actor*) { } }
namespace cppa {
namespace cppa { namespace detail {
scheduled_actor::scheduled_actor()
: next(nullptr)
, m_state(scheduled_actor::done)
, m_enqueue_to_scheduler(dummy_enqueue, static_cast<void*>(nullptr), this)
, m_has_pending_timeout_request(false)
, m_active_timeout_id(0)
{
}
scheduled_actor::resume_callback::~resume_callback()
{
}
void scheduled_actor::quit(std::uint32_t reason)
{
cleanup(reason);
throw actor_exited(reason);
//yield(yield_state::done);
}
void scheduled_actor::enqueue_node(queue_node* node)
{
if (m_mailbox._push_back(node))
{
for (;;)
{
int state = m_state.load();
switch (state)
{
case blocked:
{
if (m_state.compare_exchange_weak(state, ready))
{
m_enqueue_to_scheduler();
return;
}
break;
}
case about_to_block:
{
if (m_state.compare_exchange_weak(state, ready))
{
return;
}
break;
}
default: return;
}
}
}
}
void scheduled_actor::enqueue(actor* sender, any_tuple&& msg)
{
enqueue_node(new queue_node(sender, std::move(msg)));
}
void scheduled_actor::enqueue(actor* sender, const any_tuple& msg)
{
enqueue_node(new queue_node(sender, msg));
}
int scheduled_actor::compare_exchange_state(int expected, int new_value)
{
int e = expected;
do
{
if (m_state.compare_exchange_weak(e, new_value))
{
return new_value;
}
}
while (e == expected);
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))
{
auto v0 = *reinterpret_cast<const atom_value*>(msg.at(0));
auto v1 = *reinterpret_cast<const std::uint32_t*>(msg.at(1));
if (v0 == atom(":Exit"))
{
if (m_trap_exit == false)
{
if (v1 != exit_reason::normal)
{
quit(v1);
}
return normal_exit_signal;
}
}
else if (v0 == atom(":Timeout"))
{
return (v1 == m_active_timeout_id) ? timeout_message
: expired_timeout_message;
}
}
return ordinary_message;
}
scheduled_actor::dq_result scheduled_actor::dq(std::unique_ptr<queue_node>& node,
invoke_rules_base& rules,
queue_node_buffer& buffer)
{
switch (filter_msg(node->msg))
{
case normal_exit_signal:
case expired_timeout_message:
{
// skip message
return dq_indeterminate;
}
case timeout_message:
{
// m_active_timeout_id is already invalid
m_has_pending_timeout_request = false;
// restore mailbox before calling client
if (!buffer.empty())
{
m_mailbox.push_front(std::move(buffer));
buffer.clear();
}
return dq_timeout_occured;
}
default: break;
}
auto imd = rules.get_intermediate(node->msg);
if (imd)
{
m_last_dequeued = std::move(node->msg);
m_last_sender = std::move(node->sender);
// restore mailbox before invoking imd
if (!buffer.empty())
{
m_mailbox.push_front(std::move(buffer));
buffer.clear();
}
// expire pending request
if (m_has_pending_timeout_request)
{
++m_active_timeout_id;
m_has_pending_timeout_request = false;
}
imd->invoke();
return dq_done;
}
else
{
buffer.push_back(node.release());
return dq_indeterminate;
}
}
// dummy
void scheduled_actor_dummy::resume(util::fiber*, resume_callback*)
{
}
void scheduled_actor_dummy::quit(std::uint32_t)
{
}
void scheduled_actor_dummy::dequeue(invoke_rules&)
{
}
void scheduled_actor_dummy::dequeue(timed_invoke_rules&)
{
}
void scheduled_actor_dummy::link_to(intrusive_ptr<actor>&)
{
}
void scheduled_actor_dummy::unlink_from(intrusive_ptr<actor>&)
{
}
bool scheduled_actor_dummy::establish_backlink(intrusive_ptr<actor>&)
{
return false;
}
bool scheduled_actor_dummy::remove_backlink(intrusive_ptr<actor>&)
{
return false;
}
void scheduled_actor_dummy::detach(const attachable::token&)
scheduled_actor::~scheduled_actor()
{
}
bool scheduled_actor_dummy::attach(attachable*)
void scheduled_actor::on_exit()
{
return false;
}
} } // namespace cppa::detail
} // namespace cppa
......@@ -3,7 +3,7 @@
#include "cppa/config.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/util/fiber.hpp"
#include "cppa/actor_behavior.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/detail/invokable.hpp"
#include "cppa/detail/actor_count.hpp"
#include "cppa/detail/task_scheduler.hpp"
......@@ -18,7 +18,7 @@ using std::endl;
namespace {
void enqueue_fun(cppa::detail::task_scheduler* where,
cppa::detail::scheduled_actor* what)
cppa::detail::abstract_scheduled_actor* what)
{
where->schedule(what);
}
......@@ -27,14 +27,14 @@ void enqueue_fun(cppa::detail::task_scheduler* where,
namespace cppa { namespace detail {
void task_scheduler::worker_loop(job_queue* jq, scheduled_actor* dummy)
void task_scheduler::worker_loop(job_queue* jq, abstract_scheduled_actor* dummy)
{
cppa::util::fiber fself;
scheduled_actor* job = nullptr;
struct handler : scheduled_actor::resume_callback
abstract_scheduled_actor* job = nullptr;
struct handler : abstract_scheduled_actor::resume_callback
{
scheduled_actor** job_ptr;
handler(scheduled_actor** jptr) : job_ptr(jptr) { }
abstract_scheduled_actor** job_ptr;
handler(abstract_scheduled_actor** jptr) : job_ptr(jptr) { }
bool still_ready()
{
return true;
......@@ -72,7 +72,7 @@ void task_scheduler::stop()
super::stop();
}
void task_scheduler::schedule(scheduled_actor* what)
void task_scheduler::schedule(abstract_scheduled_actor* what)
{
if (what)
{
......@@ -87,11 +87,11 @@ void task_scheduler::schedule(scheduled_actor* what)
}
}
actor_ptr task_scheduler::spawn_impl(scheduled_actor* what)
actor_ptr task_scheduler::spawn_impl(abstract_scheduled_actor* what)
{
inc_actor_count();
CPPA_MEMORY_BARRIER();
intrusive_ptr<scheduled_actor> ctx(what);
intrusive_ptr<abstract_scheduled_actor> ctx(what);
// add an implicit reference to ctx
ctx->ref();
m_queue.push_back(ctx.get());
......@@ -104,7 +104,7 @@ actor_ptr task_scheduler::spawn(abstract_event_based_actor* what)
return spawn_impl(what->attach_to_scheduler(enqueue_fun, this));
}
actor_ptr task_scheduler::spawn(actor_behavior* behavior, scheduling_hint)
actor_ptr task_scheduler::spawn(scheduled_actor* behavior, scheduling_hint)
{
return spawn_impl(new yielding_actor(behavior, enqueue_fun, this));
}
......
......@@ -16,7 +16,7 @@ namespace cppa { namespace detail {
namespace {
void enqueue_fun(cppa::detail::thread_pool_scheduler* where,
cppa::detail::scheduled_actor* what)
cppa::detail::abstract_scheduled_actor* what)
{
where->schedule(what);
}
......@@ -33,7 +33,7 @@ struct thread_pool_scheduler::worker
worker* next;
bool m_done;
job_queue* m_job_queue;
volatile scheduled_actor* m_job;
volatile abstract_scheduled_actor* m_job;
worker_queue* m_supervisor_queue;
mutex m_mtx;
condition_variable m_cv;
......@@ -60,11 +60,11 @@ struct thread_pool_scheduler::worker
// enqueue as idle worker
m_supervisor_queue->push_back(this);
util::fiber fself;
struct handler : scheduled_actor::resume_callback
struct handler : abstract_scheduled_actor::resume_callback
{
time_type timeout;
bool reschedule;
scheduled_actor* job;
abstract_scheduled_actor* job;
handler() : timeout(now()), reschedule(false), job(nullptr)
{
}
......@@ -97,7 +97,7 @@ struct thread_pool_scheduler::worker
}
if (m_done) return;
}
h.job = const_cast<scheduled_actor*>(m_job);
h.job = const_cast<abstract_scheduled_actor*>(m_job);
// run actor up to 300ms
h.reschedule = false;
h.timeout = now();
......@@ -121,7 +121,7 @@ void thread_pool_scheduler::worker_loop(thread_pool_scheduler::worker* w)
}
void thread_pool_scheduler::supervisor_loop(job_queue* jqueue,
scheduled_actor* dummy)
abstract_scheduled_actor* dummy)
{
worker_queue wqueue;
std::vector<worker_ptr> workers;
......@@ -142,7 +142,7 @@ void thread_pool_scheduler::supervisor_loop(job_queue* jqueue,
do
{
// fetch next job
scheduled_actor* job = jqueue->pop();
abstract_scheduled_actor* job = jqueue->pop();
if (job == dummy)
{
done = true;
......@@ -201,16 +201,16 @@ void thread_pool_scheduler::stop()
super::stop();
}
void thread_pool_scheduler::schedule(scheduled_actor* what)
void thread_pool_scheduler::schedule(abstract_scheduled_actor* what)
{
m_queue.push_back(what);
}
actor_ptr thread_pool_scheduler::spawn_impl(scheduled_actor* what)
actor_ptr thread_pool_scheduler::spawn_impl(abstract_scheduled_actor* what)
{
inc_actor_count();
CPPA_MEMORY_BARRIER();
intrusive_ptr<scheduled_actor> ctx(what);
intrusive_ptr<abstract_scheduled_actor> ctx(what);
ctx->ref();
m_queue.push_back(ctx.get());
return std::move(ctx);
......@@ -222,7 +222,7 @@ actor_ptr thread_pool_scheduler::spawn(abstract_event_based_actor* what)
return spawn_impl(what->attach_to_scheduler(enqueue_fun, this));
}
actor_ptr thread_pool_scheduler::spawn(actor_behavior* behavior,
actor_ptr thread_pool_scheduler::spawn(scheduled_actor* behavior,
scheduling_hint hint)
{
if (hint == detached)
......
......@@ -40,7 +40,7 @@ void yielding_actor::yield_until_not_empty()
{
while (m_mailbox.empty())
{
m_state.store(scheduled_actor::about_to_block);
m_state.store(abstract_scheduled_actor::about_to_block);
CPPA_MEMORY_BARRIER();
// make sure mailbox is empty
if (!m_mailbox.empty())
......@@ -48,7 +48,7 @@ void yielding_actor::yield_until_not_empty()
// someone preempt us
//compare_exchange_state(scheduled_actor::about_to_block,
// scheduled_actor::ready);
m_state.store(scheduled_actor::ready);
m_state.store(abstract_scheduled_actor::ready);
return;
}
else
......@@ -120,15 +120,15 @@ void yielding_actor::resume(util::fiber* from, resume_callback* callback)
}
case yield_state::blocked:
{
switch (compare_exchange_state(scheduled_actor::about_to_block,
scheduled_actor::blocked))
switch (compare_exchange_state(abstract_scheduled_actor::about_to_block,
abstract_scheduled_actor::blocked))
{
case scheduled_actor::ready:
case abstract_scheduled_actor::ready:
{
if (callback->still_ready()) break;
else return;
}
case scheduled_actor::blocked:
case abstract_scheduled_actor::blocked:
{
// wait until someone re-schedules that actor
return;
......
#define CPPA_VERBOSE_CHECK
#include <stack>
#include <chrono>
#include <iostream>
......@@ -12,6 +10,7 @@
#include "cppa/cppa.hpp"
#include "cppa/actor.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/fsm_actor.hpp"
#include "cppa/to_string.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/event_based_actor.hpp"
......@@ -26,117 +25,126 @@ using namespace cppa;
// GCC 4.7 supports non-static member initialization
#if (__GNUC__ >= 4) && (__GNUC_MINOR__ >= 7)
class event_testee : public event_based_actor
class event_testee : public fsm_actor<event_testee>
{
friend class fsm_actor<event_testee>;
invoke_rules wait4string =
(
on<std::string>() >> [=](const std::string& value)
on<std::string>() >> [=]()
{
cout << "event_testee[string]: " << value << endl;
// switch back to wait4int
unbecome();
unbecome();
become(&init_state);
},
on<atom("GetState")>() >> [=]()
{
reply("wait4string");
}
);
invoke_rules wait4float =
(
on<float>() >> [=](float value)
on<float>() >> [=]()
{
cout << "event_testee[float]: " << value << endl;
become(wait4string);
}
);
invoke_rules wait4int =
(
on<int>() >> [=](int value)
become(&wait4string);
},
on<atom("GetState")>() >> [=]()
{
cout << "event_testee[int]: " << value << endl;
become(wait4float);
reply("wait4float");
}
);
public:
void on_exit()
invoke_rules init_state =
(
on<int>() >> [=]()
{
cout << "event_testee[GCC47]::on_exit()" << endl;
}
void init()
become(&wait4float);
},
on<atom("GetState")>() >> [=]()
{
cout << "event_testee[GCC47]::init()" << endl;
become(wait4int);
reply("init_state");
}
);
};
#else
class event_testee : public event_based_actor
class event_testee : public fsm_actor<event_testee>
{
friend class fsm_actor<event_testee>;
invoke_rules wait4string;
invoke_rules wait4float;
invoke_rules wait4int;
invoke_rules init_state;
public:
void on_exit()
{
cout << "event_testee::on_exit()" << endl;
}
void init()
event_testee()
{
wait4string =
(
on<std::string>() >> [=](const std::string& value)
on<std::string>() >> [=]()
{
become(&init_state);
},
on<atom("GetState")>() >> [=]()
{
cout << "event_testee[string]: " << value << endl;
become(&wait4int);
reply("wait4string");
}
);
wait4float =
(
on<float>() >> [=](float value)
on<float>() >> [=]()
{
cout << "event_testee[float]: " << value << endl;
become(&wait4string);
//become(&wait4string);
become
(
on<std::string>() >> [=]()
{
become(&init_state);
},
on<atom("GetState")>() >> []()
{
reply("wait4string");
}
);
},
on<atom("GetState")>() >> [=]()
{
reply("wait4float");
}
);
wait4int =
init_state =
(
on<int>() >> [=](int value)
on<int>() >> [=]()
{
cout << "event_testee[int]: " << value << endl;
become(&wait4float);
},
on<atom("GetState")>() >> [=]()
{
reply("init_state");
}
);
cout << "event_testee::init()" << endl;
become(&wait4int);
}
};
#endif
// quits after 5 timeouts
abstract_event_based_actor* event_testee2()
{
struct impl : event_based_actor
struct impl : fsm_actor<impl>
{
int num_timeouts;
impl() : num_timeouts(0) { }
timed_invoke_rules state;
void init()
timed_invoke_rules init_state;
impl() : num_timeouts(0)
{
cout << "event_testee2::impl::init()" << endl;
state =
//become
init_state =
(
others() >> []()
{
......@@ -146,66 +154,79 @@ abstract_event_based_actor* event_testee2()
},
after(std::chrono::milliseconds(50)) >> [=]()
{
cout << "testee2 received timeout nr. "
<< (num_timeouts + 1) << endl;
if (++num_timeouts >= 5)
{
quit(exit_reason::normal);
}
}
);
become(&state);
}
};
return new impl;
}
class testee_behavior : public actor_behavior
class testee_actor : public scheduled_actor
{
public:
void act()
{
receive_loop
(
on<int>() >> [&](int i)
void wait4string()
{
cout << "testee_behavior[int]: " << i << endl;
receive
bool string_received = false;
receive_while([&]() { return !string_received; })
(
on<float>() >> [&](float f)
on<std::string>() >> [&]()
{
cout << "testee_behavior[float]: " << f << endl;
receive
(
on<std::string>() >> [&](const std::string& str)
string_received = true;
},
on<atom("GetState")>() >> [&]()
{
cout << "testee_behavior[string]: "
<< str
<< endl;
reply("wait4string");
}
);
}
);
void wait4float()
{
bool float_received = false;
receive_while([&]() { return !float_received; })
(
on<float>() >> [&]()
{
float_received = true;
wait4string();
},
on<atom("GetState")>() >> [&]()
{
reply("wait4float");
}
);
}
void on_exit()
public:
void act()
{
receive_loop
(
on<int>() >> [&]()
{
cout << "testee_behavior::on_exit()" << endl;
wait4float();
},
on<atom("GetState")>() >> [&]()
{
reply("init_state");
}
);
}
};
// receives one timeout and quits
void testee1()
{
receive_loop
(
after(std::chrono::milliseconds(10)) >> []()
{
cout << "testee1::quit" << endl;
quit(exit_reason::user_defined);
}
);
......@@ -245,10 +266,10 @@ void testee3(actor_ptr parent)
}
template<class Testee>
void behavior_test()
std::string behavior_test()
{
std::string result;
std::string testee_name = detail::to_uniform_name(typeid(Testee));
cout << "behavior_test<" << testee_name << ">()" << endl;
auto et = spawn(new Testee);
send(et, 1);
send(et, 2);
......@@ -259,22 +280,34 @@ void behavior_test()
send(et, .3f);
send(et, "hello again " + testee_name);
send(et, "goodbye " + testee_name);
send(et, atom("GetState"));
receive
(
on<std::string>() >> [&](const std::string& str)
{
result = str;
},
after(std::chrono::seconds(2)) >> [&]()
{
throw std::runtime_error(testee_name + " does not reply");
}
);
send(et, atom(":Exit"), exit_reason::user_defined);
await_all_others_done();
cout << endl;
return result;
}
size_t test__spawn()
{
CPPA_TEST(test__spawn);
//spawn(testee1);
spawn(testee1);
spawn(event_testee2());
await_all_others_done();
behavior_test<testee_behavior>();
behavior_test<event_testee>();
CPPA_CHECK_EQUAL(behavior_test<testee_actor>(), "init_state");
CPPA_CHECK_EQUAL(behavior_test<event_testee>(), "init_state");
return CPPA_TEST_RESULT;
......
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