Commit 4ff15bf7 authored by neverlord's avatar neverlord

event_based_actor

parent 5bc73557
......@@ -3,6 +3,7 @@ ACLOCAL_AMFLAGS = -I m4
lib_LTLIBRARIES = libcppa.la
libcppa_la_SOURCES = \
src/abstract_event_based_actor.cpp \
src/abstract_tuple.cpp \
src/actor_behavior.cpp \
src/actor_count.cpp \
......@@ -21,6 +22,7 @@ libcppa_la_SOURCES = \
src/local_actor.cpp \
src/converted_thread_context.cpp \
src/cppa.cpp \
src/event_based_actor.cpp \
src/delegate.cpp \
src/demangle.cpp \
src/deserializer.cpp \
......@@ -81,7 +83,9 @@ nobase_library_include_HEADERS = \
cppa/cow_ptr.hpp \
cppa/cppa.hpp \
cppa/deserializer.hpp \
cppa/event_based_actor.hpp \
cppa/detail/abstract_actor.hpp \
cppa/detail/abstract_event_based_actor.hpp \
cppa/detail/abstract_tuple.hpp \
cppa/detail/actor_count.hpp \
cppa/detail/actor_proxy_cache.hpp \
......@@ -169,6 +173,7 @@ nobase_library_include_HEADERS = \
cppa/util/disable_if.hpp \
cppa/util/disjunction.hpp \
cppa/util/duration.hpp \
cppa/util/either.hpp \
cppa/util/element_at.hpp \
cppa/util/enable_if.hpp \
cppa/util/eval_first_n.hpp \
......
......@@ -231,3 +231,8 @@ examples/hello_world_example.cpp
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
src/abstract_event_based_actor.cpp
cppa/event_based_actor.hpp
src/event_based_actor.cpp
......@@ -39,15 +39,17 @@
#include "cppa/actor.hpp"
#include "cppa/invoke.hpp"
#include "cppa/channel.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/receive.hpp"
#include "cppa/announce.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/to_string.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/invoke_rules.hpp"
#include "cppa/actor_behavior.hpp"
#include "cppa/scheduling_hint.hpp"
#include "cppa/event_based_actor.hpp"
#include "cppa/util/rm_ref.hpp"
#include "cppa/util/enable_if.hpp"
......@@ -459,6 +461,22 @@ inline void trap_exit(bool new_value)
self()->trap_exit(new_value);
}
inline actor_ptr spawn(actor_behavior* what)
{
return get_scheduler()->spawn(what, scheduled);
}
template<scheduling_hint Hint>
inline actor_ptr spawn(actor_behavior* what)
{
return get_scheduler()->spawn(what, Hint);
}
inline actor_ptr spawn(event_based_actor* what)
{
return get_scheduler()->spawn(what);
}
/**
* @brief Spawns a new actor that executes @p what with given arguments.
* @param Hint Hint to the scheduler for the best scheduling strategy.
......@@ -467,7 +485,11 @@ inline void trap_exit(bool new_value)
* @returns A pointer to the newly created {@link actor Actor}.
*/
template<scheduling_hint Hint, typename F, typename... Args>
actor_ptr spawn(F&& what, const Args&... 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
|| std::is_convertible<typename util::rm_ref<F>::type, event_based_actor*>::value,
actor_ptr>::type
{
typedef typename util::rm_ref<F>::type ftype;
std::integral_constant<bool, std::is_function<ftype>::value> is_fun;
......@@ -482,7 +504,11 @@ actor_ptr spawn(F&& what, const Args&... args)
* @returns A pointer to the newly created {@link actor Actor}.
*/
template<typename F, typename... Args>
inline actor_ptr spawn(F&& what, const Args&... 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
|| std::is_convertible<typename util::rm_ref<F>::type, event_based_actor*>::value,
actor_ptr>::type
{
return spawn<scheduled>(std::forward<F>(what), args...);
}
......
#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*/;
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
......@@ -6,6 +6,7 @@
#include <map>
#include <list>
#include <mutex>
#include <stack>
#include <atomic>
#include <vector>
#include <memory>
......@@ -15,7 +16,10 @@
#include "cppa/pattern.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/detail/abstract_actor.hpp"
#include "cppa/util/either.hpp"
#include "cppa/util/singly_linked_list.hpp"
namespace cppa { namespace detail {
......
......@@ -22,6 +22,14 @@ class delegate
{
}
template<typename Arg1, typename Arg2, typename Function>
void reset(Function* fun, Arg1* a1, Arg2* a2)
{
m_fun = reinterpret_cast<void_fun>(fun);
m_arg1 = reinterpret_cast<void*>(a1);
m_arg2 = reinterpret_cast<void*>(a2);
}
void operator()();
};
......
#ifndef GET_BEHAVIOR_HPP
#define GET_BEHAVIOR_HPP
#include <type_traits>
#include "cppa/invoke.hpp"
#include "cppa/util/rm_ref.hpp"
#include "cppa/detail/tdata.hpp"
......@@ -83,12 +85,20 @@ class ftor_behavior<false, true, F, Args...> : public actor_behavior
template<typename R>
actor_behavior* get_behavior(std::integral_constant<bool,true>, R (*fptr)())
{
static_assert(std::is_convertible<R, actor_behavior*>::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?");
return new ftor_behavior<true, false, R (*)()>(fptr);
}
template<typename F>
actor_behavior* get_behavior(std::integral_constant<bool,false>, F&& ftor)
{
static_assert(std::is_convertible<decltype(ftor()), actor_behavior*>::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?");
typedef typename util::rm_ref<F>::type ftype;
return new ftor_behavior<false, false, ftype>(std::forward<F>(ftor));
}
......@@ -99,6 +109,10 @@ actor_behavior* get_behavior(std::integral_constant<bool,true>,
const Arg0& arg0,
const Args&... args)
{
static_assert(std::is_convertible<decltype(fptr(arg0, args...)), actor_behavior*>::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?");
typedef ftor_behavior<true, true, F, Arg0, Args...> impl;
return new impl(fptr, arg0, args...);
}
......@@ -109,6 +123,10 @@ actor_behavior* get_behavior(std::integral_constant<bool,false>,
const Arg0& arg0,
const Args&... args)
{
static_assert(std::is_convertible<decltype(ftor(arg0, args...)), actor_behavior*>::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?");
typedef typename util::rm_ref<F>::type ftype;
typedef ftor_behavior<false, true, ftype, Arg0, Args...> impl;
return new impl(std::forward<F>(ftor), arg0, args...);
......
......@@ -10,6 +10,8 @@ class mock_scheduler : public scheduler
public:
actor_ptr spawn(event_based_actor* what);
actor_ptr spawn(actor_behavior*, scheduling_hint);
static actor_ptr spawn(actor_behavior*);
......
......@@ -2,9 +2,12 @@
#define SCHEDULED_ACTOR_HPP
#include "cppa/local_actor.hpp"
#include "cppa/util/fiber.hpp"
#include "cppa/actor_behavior.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"
......@@ -32,6 +35,41 @@ class scheduled_actor : public abstract_actor<local_actor>
typedef abstract_actor<local_actor> super;
typedef super::queue_node queue_node;
typedef util::singly_linked_list<queue_node> queue_node_buffer;
enum dq_result
{
dq_done,
dq_indeterminate,
dq_timeout_occured
};
enum filter_result
{
normal_exit_signal,
expired_timeout_message,
timeout_message,
ordinary_message
};
filter_result filter_msg(const any_tuple& msg);
dq_result dq(std::unique_ptr<queue_node>& node,
invoke_rules_base& rules,
queue_node_buffer& buffer);
bool has_pending_timeout()
{
return m_has_pending_timeout_request;
}
void request_timeout(const util::duration& d);
private:
bool m_has_pending_timeout_request;
std::uint32_t m_active_timeout_id;
pattern<atom_value, std::uint32_t> m_pattern;
public:
......@@ -48,6 +86,8 @@ class scheduled_actor : public abstract_actor<local_actor>
: next(nullptr)
, m_state(ready)
, m_enqueue_to_scheduler(enqueue_fun, sched, this)
, m_has_pending_timeout_request(false)
, m_active_timeout_id(0)
{
}
......
......@@ -29,8 +29,14 @@ class task_scheduler : public scheduler
void schedule(scheduled_actor* what);
actor_ptr spawn(event_based_actor* what);
actor_ptr spawn(actor_behavior*, scheduling_hint);
private:
actor_ptr spawn_impl(scheduled_actor* what);
};
} } // namespace cppa::detail
......
......@@ -22,6 +22,8 @@ class thread_pool_scheduler : public scheduler
void schedule(scheduled_actor* what) /*override*/;
actor_ptr spawn(event_based_actor* what);
actor_ptr spawn(actor_behavior* behavior, scheduling_hint hint);
private:
......@@ -32,6 +34,8 @@ class thread_pool_scheduler : public scheduler
scheduled_actor_dummy m_dummy;
thread m_supervisor;
actor_ptr spawn_impl(scheduled_actor* what);
static void worker_loop(worker*);
static void supervisor_loop(job_queue*, scheduled_actor*);
......
#ifndef YIELDING_ACTOR_HPP
#define YIELDING_ACTOR_HPP
#include <stack>
#include "cppa/pattern.hpp"
#include "cppa/detail/delegate.hpp"
#include "cppa/detail/scheduled_actor.hpp"
#include "cppa/detail/yield_interface.hpp"
#include "cppa/util/either.hpp"
#include "cppa/util/singly_linked_list.hpp"
namespace cppa { namespace detail {
......@@ -18,34 +23,12 @@ class yielding_actor : public scheduled_actor
util::fiber m_fiber;
actor_behavior* m_behavior;
bool m_has_pending_timeout_request;
std::uint32_t m_active_timeout_id;
pattern<atom_value, std::uint32_t> m_pattern;
static void run(void* _this);
void yield_until_not_empty();
enum dq_result
{
dq_done,
dq_indeterminate,
dq_timeout_occured
};
enum filter_result
{
normal_exit_signal,
expired_timeout_message,
timeout_message,
ordinary_message
};
void exec_loop_stack();
filter_result filter_msg(const any_tuple& msg);
dq_result dq(std::unique_ptr<queue_node>& node,
invoke_rules_base& rules,
queue_node_buffer& buffer);
void yield_until_not_empty();
public:
......@@ -56,12 +39,10 @@ class yielding_actor : public scheduled_actor
: super(enqueue_fun, sched)
, m_fiber(&yielding_actor::run, this)
, m_behavior(behavior)
, m_has_pending_timeout_request(false)
, m_active_timeout_id(0)
{
}
~yielding_actor();
~yielding_actor() /*override*/;
void dequeue(invoke_rules& rules) /*override*/;
......
#ifndef EVENT_BASED_ACTOR_HPP
#define EVENT_BASED_ACTOR_HPP
#include "cppa/detail/abstract_event_based_actor.hpp"
namespace cppa {
class event_based_actor : public detail::abstract_event_based_actor
{
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)...);
}
protected:
void unbecome() /*override*/;
void become(invoke_rules&& behavior) /*override*/;
void become(timed_invoke_rules&& behavior) /*override*/;
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;
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:
// provoke compiler errors for usage of receive() and related functions
template<typename... Args>
void receive(Args&&... rules);
template<typename... Args>
void receive_loop(Args&&... rules);
template<typename Statement>
void receive_while(Statement&& stmt);
template<typename... Args>
void do_receive(Args&&... args);
};
} // namespace cppa
#endif // EVENT_BASED_ACTOR_HPP
......@@ -21,6 +21,12 @@ static constexpr std::uint32_t normal = 0x00001;
*/
static constexpr std::uint32_t unhandled_exception = 0x00002;
/**
* @brief Indicates that an event-based actor
* tried to use receive().
*/
static constexpr std::uint32_t unallowed_function_call = 0x00003;
/**
* @brief Indicates that an actor finishied execution
* because a connection to a remote link was
......
......@@ -38,9 +38,8 @@ class intrusive_ptr : util::comparable<intrusive_ptr<T>, const T*>,
set_ptr(other.m_ptr);
}
intrusive_ptr(intrusive_ptr&& other) : m_ptr(nullptr)
intrusive_ptr(intrusive_ptr&& other) : m_ptr(other.take())
{
swap(other);
}
template<typename Y>
......@@ -52,9 +51,10 @@ class intrusive_ptr : util::comparable<intrusive_ptr<T>, const T*>,
}
template<typename Y>
intrusive_ptr(intrusive_ptr<Y>&& other) : m_ptr(nullptr)
intrusive_ptr(intrusive_ptr<Y>&& other) : m_ptr(other.take())
{
swap(other);
static_assert(std::is_convertible<Y*, T*>::value,
"Y* is not assignable to T*");
}
~intrusive_ptr()
......
......@@ -32,11 +32,16 @@
#include <list>
#include <memory>
#include "cppa/detail/invokable.hpp"
#include "cppa/detail/intermediate.hpp"
namespace cppa { namespace detail {
/*
class invokable;
class intermediate;
class timed_invokable;
*/
typedef std::unique_ptr<detail::invokable> invokable_ptr;
typedef std::unique_ptr<detail::timed_invokable> timed_invokable_ptr;
......@@ -120,6 +125,8 @@ public:
timed_invoke_rules(timed_invoke_rules&& arg);
timed_invoke_rules(detail::timed_invokable_ptr&& arg);
timed_invoke_rules& operator=(timed_invoke_rules&&);
const util::duration& timeout() const;
void handle_timeout() const;
......
......@@ -8,9 +8,10 @@
#include "cppa/atom.hpp"
#include "cppa/tuple.hpp"
#include "cppa/actor.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/attachable.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/scheduling_hint.hpp"
#include "cppa/event_based_actor.hpp"
#include "cppa/util/duration.hpp"
......@@ -55,6 +56,11 @@ class scheduler
virtual actor_ptr spawn(actor_behavior* behavior,
scheduling_hint hint) = 0;
/**
* @brief Spawns a new event-based actor.
*/
virtual actor_ptr spawn(event_based_actor* what) = 0;
/**
* @brief Informs the scheduler about a converted context
* (a thread that acts as actor).
......
#ifndef EITHER_HPP
#define EITHER_HPP
#include <new>
#include <stdexcept>
#include <type_traits>
namespace cppa { namespace util {
template<class Left, class Right>
class either
{
static_assert(std::is_same<Left, Right>::value == false, "Left == Right");
union
{
Left m_left;
Right m_right;
};
bool m_is_left;
void check_flag(bool flag, const char* side)
{
if (m_is_left != flag)
{
std::string err = "not a ";
err += side;
throw std::runtime_error(err);
}
}
void destroy()
{
if (m_is_left)
{
m_left.~Left();
}
else
{
m_right.~Right();
}
}
template<typename L>
void cr_left(L&& value)
{
new (&m_left) Left (std::forward<L>(value));
}
template<typename R>
void cr_right(R&& value)
{
new (&m_right) Right (std::forward<R>(value));
}
public:
// default constructor creates a left
either() : m_is_left(true)
{
new (&m_left) Left ();
}
either(const Left& value) : m_is_left(true)
{
cr_left(value);
}
either(Left&& value) : m_is_left(true)
{
cr_left(std::move(value));
}
either(const Right& value) : m_is_left(false)
{
cr_right(value);
}
either(Right&& value) : m_is_left(false)
{
cr_right(std::move(value));
}
either(const either& other) : m_is_left(other.m_is_left)
{
if (other.m_is_left)
{
cr_left(other.left());
}
else
{
cr_right(other.right());
}
}
either(either&& other) : m_is_left(other.m_is_left)
{
if (other.m_is_left)
{
cr_left(std::move(other.left()));
}
else
{
cr_right(std::move(other.right()));
}
}
~either()
{
destroy();
}
either& operator=(const either& other)
{
if (m_is_left == other.m_is_left)
{
if (m_is_left)
{
left() = other.left();
}
else
{
right() = other.right();
}
}
else
{
destroy();
m_is_left = other.m_is_left;
if (other.m_is_left)
{
cr_left(other.left());
}
else
{
cr_right(other.right());
}
}
return *this;
}
inline bool is_left() const
{
return m_is_left;
}
inline bool is_right() const
{
return !m_is_left;
}
Left& left()
{
return m_left;
}
const Left& left() const
{
return m_left;
}
Right& right()
{
return m_right;
}
const Right& right() const
{
return m_right;
}
};
template<typename Left, typename Right>
bool operator==(const either<Left, Right>& lhs,
const either<Left, Right>& rhs)
{
if (lhs.is_left() == rhs.is_left())
{
if (lhs.is_left())
{
return lhs.left() == rhs.left();
}
else
{
return lhs.right() == rhs.right();
}
}
return false;
}
template<typename Left, typename Right>
bool operator==(const either<Left, Right>& lhs, const Left& rhs)
{
return lhs.is_left() && lhs.left() == rhs;
}
template<typename Left, typename Right>
bool operator==(const Left& lhs, const either<Left, Right>& rhs)
{
return rhs == lhs;
}
template<typename Left, typename Right>
bool operator==(const either<Left, Right>& lhs, const Right& rhs)
{
return lhs.is_right() && lhs.right() == rhs;
}
template<typename Left, typename Right>
bool operator==(const Right& lhs, const either<Left, Right>& rhs)
{
return rhs == lhs;
}
template<typename Left, typename Right>
bool operator!=(const either<Left, Right>& lhs,
const either<Left, Right>& rhs)
{
return !(lhs == rhs);
}
template<typename Left, typename Right>
bool operator!=(const either<Left, Right>& lhs, const Left& rhs)
{
return !(lhs == rhs);
}
template<typename Left, typename Right>
bool operator!=(const Left& lhs, const either<Left, Right>& rhs)
{
return !(rhs == lhs);
}
template<typename Left, typename Right>
bool operator!=(const either<Left, Right>& lhs, const Right& rhs)
{
return !(lhs == rhs);
}
template<typename Left, typename Right>
bool operator!=(const Right& lhs, const either<Left, Right>& rhs)
{
return !(rhs == lhs);
}
} } // namespace cppa::util
#endif // EITHER_HPP
#include "cppa/detail/invokable.hpp"
#include "cppa/detail/abstract_event_based_actor.hpp"
namespace cppa { namespace detail {
void abstract_event_based_actor::dequeue(invoke_rules&)
{
quit(exit_reason::unallowed_function_call);
}
void abstract_event_based_actor::dequeue(timed_invoke_rules&)
{
quit(exit_reason::unallowed_function_call);
}
void abstract_event_based_actor::handle_message(std::unique_ptr<queue_node>& node,
invoke_rules& behavior)
{
// no need to handle result
(void) dq(node, behavior, m_buffer);
}
void abstract_event_based_actor::handle_message(std::unique_ptr<queue_node>& node,
timed_invoke_rules& behavior)
{
// request timeout only if we're running short on messages
if (m_mailbox.empty() && has_pending_timeout() == false)
{
request_timeout(behavior.timeout());
}
switch (dq(node, behavior, m_buffer))
{
case dq_timeout_occured:
{
behavior.handle_timeout();
}
default: break;
}
}
void abstract_event_based_actor::handle_message(std::unique_ptr<queue_node>& node)
{
if (m_loop_stack.top().is_left())
{
handle_message(node, m_loop_stack.top().left());
}
else
{
handle_message(node, m_loop_stack.top().right());
}
}
void abstract_event_based_actor::resume(util::fiber*, resume_callback* callback)
{
auto done_cb = [&]()
{
m_state.store(scheduled_actor::done);
while (!m_loop_stack.empty()) m_loop_stack.pop();
callback->exec_done();
};
bool actor_done = false;
std::unique_ptr<queue_node> node;
do
{
if (m_loop_stack.empty())
{
cleanup(exit_reason::normal);
done_cb();
return;
}
else if (m_mailbox.empty())
{
m_state.store(scheduled_actor::about_to_block);
CPPA_MEMORY_BARRIER();
if (!m_mailbox.empty())
{
// someone preempt us
m_state.store(scheduled_actor::ready);
}
else
{
// nothing to do
return;
}
}
node.reset(m_mailbox.pop());
try
{
handle_message(node);
}
catch (actor_exited& what)
{
cleanup(what.reason());
actor_done = true;
}
catch (...)
{
cleanup(exit_reason::unhandled_exception);
actor_done = true;
}
if (actor_done)
{
done_cb();
return;
}
}
while (callback->still_ready());
}
} } // namespace cppa::detail
......@@ -2,6 +2,7 @@
#include <algorithm>
#include "cppa/exception.hpp"
#include "cppa/detail/invokable.hpp"
#include "cppa/detail/intermediate.hpp"
#include "cppa/detail/converted_thread_context.hpp"
......
#include "cppa/event_based_actor.hpp"
namespace cppa {
void event_based_actor::unbecome()
{
if (!m_loop_stack.empty()) m_loop_stack.pop();
}
void event_based_actor::become(invoke_rules&& behavior)
{
m_loop_stack.push(std::move(behavior));
}
void event_based_actor::become(timed_invoke_rules&& behavior)
{
m_loop_stack.push(std::move(behavior));
}
} // namespace cppa
......@@ -89,6 +89,14 @@ timed_invoke_rules::timed_invoke_rules(invokable_list&& lhs,
m_list.splice(m_list.begin(), rhs.m_list);
}
timed_invoke_rules& timed_invoke_rules::operator=(timed_invoke_rules&& other)
{
m_list = std::move(other.m_list);
other.m_list.clear();
std::swap(m_ti, other.m_ti);
return *this;
}
const util::duration& timed_invoke_rules::timeout() const
{
return m_ti->timeout();
......
......@@ -50,6 +50,13 @@ actor_ptr mock_scheduler::spawn(actor_behavior* behavior)
return ctx;
}
actor_ptr mock_scheduler::spawn(event_based_actor* what)
{
// TODO: don't delete what :)
delete what;
return nullptr;
}
actor_ptr mock_scheduler::spawn(actor_behavior* 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"
......@@ -11,6 +12,8 @@ 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)
{
}
......@@ -25,6 +28,12 @@ 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))
......@@ -81,6 +90,77 @@ int scheduled_actor::compare_exchange_state(int expected, int new_value)
return e;
}
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));
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));
// 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*)
......
......@@ -4,6 +4,7 @@
#include "cppa/local_actor.hpp"
#include "cppa/util/fiber.hpp"
#include "cppa/actor_behavior.hpp"
#include "cppa/detail/invokable.hpp"
#include "cppa/detail/actor_count.hpp"
#include "cppa/detail/task_scheduler.hpp"
#include "cppa/detail/yielding_actor.hpp"
......@@ -86,16 +87,26 @@ void task_scheduler::schedule(scheduled_actor* what)
}
}
actor_ptr task_scheduler::spawn(actor_behavior* behavior, scheduling_hint)
actor_ptr task_scheduler::spawn_impl(scheduled_actor* what)
{
inc_actor_count();
intrusive_ptr<scheduled_actor> ctx(new yielding_actor(behavior,
enqueue_fun,
this));
CPPA_MEMORY_BARRIER();
intrusive_ptr<scheduled_actor> ctx(what);
// add an implicit reference to ctx
ctx->ref();
m_queue.push_back(ctx.get());
return ctx;
return std::move(ctx);
}
actor_ptr task_scheduler::spawn(event_based_actor* what)
{
return spawn_impl(what->attach_to_scheduler(enqueue_fun, this));
}
actor_ptr task_scheduler::spawn(actor_behavior* behavior, scheduling_hint)
{
return spawn_impl(new yielding_actor(behavior, enqueue_fun, this));
}
} } // namespace cppa::detail
......@@ -2,6 +2,7 @@
#include <cstddef>
#include <iostream>
#include "cppa/detail/invokable.hpp"
#include "cppa/detail/actor_count.hpp"
#include "cppa/detail/mock_scheduler.hpp"
#include "cppa/detail/yielding_actor.hpp"
......@@ -205,6 +206,22 @@ void thread_pool_scheduler::schedule(scheduled_actor* what)
m_queue.push_back(what);
}
actor_ptr thread_pool_scheduler::spawn_impl(scheduled_actor* what)
{
inc_actor_count();
CPPA_MEMORY_BARRIER();
intrusive_ptr<scheduled_actor> ctx(what);
ctx->ref();
m_queue.push_back(ctx.get());
return std::move(ctx);
}
actor_ptr thread_pool_scheduler::spawn(event_based_actor* what)
{
return spawn_impl(what->attach_to_scheduler(enqueue_fun, this));
}
actor_ptr thread_pool_scheduler::spawn(actor_behavior* behavior,
scheduling_hint hint)
{
......@@ -214,14 +231,9 @@ actor_ptr thread_pool_scheduler::spawn(actor_behavior* behavior,
}
else
{
inc_actor_count();
CPPA_MEMORY_BARRIER();
intrusive_ptr<scheduled_actor> ctx(new yielding_actor(behavior,
enqueue_fun,
this));
ctx->ref();
m_queue.push_back(ctx.get());
return ctx;
return spawn_impl(new yielding_actor(behavior,
enqueue_fun,
this));
}
}
......
#include "cppa/cppa.hpp"
#include "cppa/detail/invokable.hpp"
#include "cppa/detail/intermediate.hpp"
#include "cppa/detail/yielding_actor.hpp"
......@@ -57,77 +58,6 @@ void yielding_actor::yield_until_not_empty()
}
}
yielding_actor::filter_result yielding_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;
}
yielding_actor::dq_result yielding_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));
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));
// 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;
}
}
void yielding_actor::dequeue(invoke_rules& rules)
{
queue_node_buffer buffer;
......@@ -146,13 +76,9 @@ void yielding_actor::dequeue(timed_invoke_rules& rules)
// try until a message was successfully dequeued
for (;;)
{
if (m_mailbox.empty() && m_has_pending_timeout_request == false)
if (m_mailbox.empty() && has_pending_timeout() == false)
{
future_send(this,
rules.timeout(),
atom(":Timeout"),
++m_active_timeout_id);
m_has_pending_timeout_request = true;
request_timeout(rules.timeout());
}
yield_until_not_empty();
std::unique_ptr<queue_node> node(m_mailbox.pop());
......
......@@ -14,6 +14,7 @@
#include "cppa/scheduler.hpp"
#include "cppa/to_string.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/event_based_actor.hpp"
using std::cerr;
using std::cout;
......@@ -21,130 +22,82 @@ using std::endl;
using namespace cppa;
namespace { const any_tuple* s_lm = nullptr; }
class event_actor
class event_testee : public event_based_actor
{
std::stack<invoke_rules> m_behavior;
invoke_rules m_next_behavior;
invoke_rules* m_behavior_ptr;
public:
event_actor(invoke_rules&& behavior)
void init()
{
m_behavior.push(std::move(behavior));
m_behavior_ptr = &(m_behavior.top());
become
(
on<int>() >> [&](int i)
{
// do NOT call receive() here!
// this would hijack the worker thread
become
(
on<int>() >> [&](int i2)
{
cout << "event testee: (" << i << ", " << i2 << ")" << endl;
unbecome();
},
on<float>() >> [&](float f)
{
cout << "event testee: (" << i << ", " << f << ")" << endl;
become
(
on<float>() >> [&]()
{
// switch back to the outer behavior
unbecome();
unbecome();
},
others() >> []()
{
cout << "event testee[line " << __LINE__ << "]: "
<< to_string(last_received())
<< endl;
}
);
}
);
},
others() >> []()
{
cout << "event testee[line " << __LINE__ << "]: "
<< to_string(last_received())
<< endl;
}
);
}
void become(invoke_rules&& behavior)
{
m_behavior.push(std::move(behavior));
m_behavior_ptr = &(m_behavior.top());
}
};
void set_next_behavior(invoke_rules&& behavior)
{
m_next_behavior = std::move(behavior);
m_behavior_ptr = &m_next_behavior;
}
class testee_behavior : public actor_behavior
{
public:
void unbecome()
void act()
{
if (!m_behavior.empty())
{
if (m_behavior_ptr == &(m_behavior.top()))
{
m_behavior.pop();
m_behavior_ptr = m_behavior.empty() ? nullptr
: &(m_behavior.top());
}
else
cout << "testee_behavior::act()" << endl;
receive_loop
(
after(std::chrono::milliseconds(10)) >> []()
{
m_behavior.pop();
quit(exit_reason::user_defined);
}
}
);
}
void operator()(const any_tuple& msg)
void on_exit()
{
if (m_behavior_ptr != nullptr)
{
s_lm = &msg;
auto ptr = m_behavior_ptr;
(*ptr)(msg);
// execute m_next_behavior at most once
if (ptr == m_behavior_ptr && ptr == &m_next_behavior)
{
m_behavior_ptr = m_behavior.empty() ? nullptr
: &m_behavior.top();
}
}
cout << "testee_behavior::on_exit()" << endl;
}
};
namespace { event_actor* s_event_actor_self = nullptr; }
void set_next_behavior(invoke_rules&& behavior)
{
s_event_actor_self->set_next_behavior(std::move(behavior));
}
void become(invoke_rules&& behavior)
{
s_event_actor_self->become(std::move(behavior));
}
void unbecome()
{
s_event_actor_self->unbecome();
}
event_actor* event_testee()
{
return new event_actor
{(
on<int>() >> [](int i)
{
// do NOT call receive() here!
// this would hijack the worker thread
set_next_behavior
((
on<int>() >> [=](int i2)
{
cout << "event testee: (" << i << ", " << i2 << ")" << endl;
},
on<float>() >> [=](float f)
{
cout << "event testee: (" << i << ", " << f << ")" << endl;
become
((
on<float>() >> []()
{
unbecome();
},
others() >> []()
{
cout << "event testee[line " << __LINE__ << "]: "
<< to_string(*s_lm)
<< endl;
}
));
}
));
},
others() >> []()
{
cout << "event testee[line " << __LINE__ << "]: "
<< to_string(*s_lm)
<< endl;
}
)};
}
void testee1()
{
receive_loop
......@@ -194,19 +147,22 @@ size_t test__spawn()
CPPA_TEST(test__spawn);
spawn(testee1);
spawn(new testee_behavior);
await_all_others_done();
auto et = event_testee();
s_event_actor_self = et;
(*et)(make_tuple(42));
(*et)(make_tuple(24));
(*et)(make_tuple(42));
(*et)(make_tuple(.24f));
(*et)(make_tuple("hello event actor"));
(*et)(make_tuple(42));
(*et)(make_tuple(.24f));
(*et)(make_tuple("hello event actor"));
delete et;
auto et = spawn(new event_testee);
send(et, 42);
send(et, 24);
send(et, 42);
send(et, .24f);
send(et, "hello event actor");
send(et, 42);
send(et, 24.f);
send(et, "hello event actor");
send(et, atom(":Exit"), exit_reason::user_defined);
await_all_others_done();
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