Commit ce955a9c authored by neverlord's avatar neverlord

nestable receive via mixin rather than policy

parent 8e7a9375
......@@ -267,4 +267,5 @@ cppa/match_expr.hpp
cppa/detail/pseudo_tuple.hpp
cppa/detail/recursive_queue_node.hpp
cppa/detail/scheduled_actor_dummy.hpp
cppa/detail/nestable_invoke_policy.hpp
cppa/detail/nestable_receive_actor.hpp
cppa/detail/filter_result.hpp
......@@ -81,7 +81,7 @@ class abstract_event_based_actor : public detail::abstract_scheduled_actor
cache_msg
};
auto handle_message(mailbox_element& iter) -> handle_message_result;
auto handle_message(mailbox_element& node) -> handle_message_result;
abstract_event_based_actor();
......
......@@ -97,6 +97,21 @@ class behavior
return m_fun;
}
inline bool operator()(any_tuple& value)
{
return m_fun(value);
}
inline bool operator()(any_tuple const& value)
{
return m_fun(value);
}
inline bool operator()(any_tuple&& value)
{
return m_fun(std::move(value));
}
private:
// terminates recursion
......
......@@ -42,6 +42,7 @@
#include "cppa/scheduled_actor.hpp"
#include "cppa/util/fiber.hpp"
#include "cppa/detail/filter_result.hpp"
#include "cppa/detail/recursive_queue_node.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
......@@ -58,14 +59,6 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor>
std::atomic<int> m_state;
enum filter_result
{
normal_exit_signal,
expired_timeout_message,
timeout_message,
ordinary_message
};
filter_result filter_msg(any_tuple const& msg)
{
auto& arr = detail::static_types_array<atom_value, std::uint32_t>::arr;
......@@ -130,7 +123,7 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor>
static constexpr int about_to_block = 0x04;
abstract_scheduled_actor(int state = done)
: m_state(state)
: super(true), m_state(state)
, m_has_pending_timeout_request(false)
, m_active_timeout_id(0)
{
......@@ -171,6 +164,7 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor>
bool enqueue_node(typename super::mailbox_element* node,
int target_state = ready)
{
CPPA_REQUIRE(node->marked == false);
if (this->m_mailbox._push_back(node))
{
for (;;)
......
......@@ -49,47 +49,21 @@
#include "cppa/exit_reason.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/intrusive/singly_linked_list.hpp"
#include "cppa/detail/nestable_invoke_policy.hpp"
#include "cppa/detail/nestable_receive_actor.hpp"
namespace cppa { namespace detail {
/**
* @brief Represents a thread that was converted to an Actor.
*/
class converted_thread_context : public abstract_actor<local_actor>
class converted_thread_context
: public nestable_receive_actor<converted_thread_context,
abstract_actor<local_actor> >
{
typedef abstract_actor<local_actor> super;
struct filter_policy;
friend struct filter_policy;
struct filter_policy
{
converted_thread_context* m_parent;
inline filter_policy(converted_thread_context* ptr) : m_parent(ptr)
{
}
inline bool operator()(any_tuple const& msg)
{
if ( m_parent->m_trap_exit == false
&& matches(msg, m_parent->m_exit_msg_pattern))
{
auto reason = msg.get_as<std::uint32_t>(1);
if (reason != exit_reason::normal)
{
m_parent->quit(reason);
}
return true;
}
return false;
}
};
typedef nestable_receive_actor<converted_thread_context,
abstract_actor<local_actor> >
super;
public:
......@@ -106,18 +80,17 @@ class converted_thread_context : public abstract_actor<local_actor>
void dequeue(partial_function& rules); //override
inline decltype(m_mailbox)& mailbox()
{
return m_mailbox;
}
inline void push_timeout() { }
private:
inline void pop_timeout() { }
filter_result filter_msg(any_tuple const& msg);
// a list is safe to use in a nested receive
typedef std::unique_ptr<recursive_queue_node> queue_node_ptr;
inline decltype(m_mailbox)& mailbox() { return m_mailbox; }
private:
pattern<atom_value, std::uint32_t> m_exit_msg_pattern;
nestable_invoke_policy<filter_policy> m_invoke;
};
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef FILTER_RESULT_HPP
#define FILTER_RESULT_HPP
namespace cppa { namespace detail {
enum filter_result
{
normal_exit_signal,
expired_timeout_message,
timeout_message,
ordinary_message
};
} } // namespace cppa::detail
#endif // FILTER_RESULT_HPP
......@@ -28,82 +28,88 @@
\******************************************************************************/
#ifndef NESTABLE_INVOKE_POLICY_HPP
#define NESTABLE_INVOKE_POLICY_HPP
#ifndef NESTABLE_RECEIVE_ACTOR_HPP
#define NESTABLE_RECEIVE_ACTOR_HPP
#include <list>
#include <memory>
#include "cppa/config.hpp"
#include "cppa/behavior.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/partial_function.hpp"
#include "cppa/detail/filter_result.hpp"
#include "cppa/detail/recursive_queue_node.hpp"
namespace cppa { namespace detail {
template<class FilterPolicy>
class nestable_invoke_policy
template<class Derived, class Base>
class nestable_receive_actor : public Base
{
public:
protected:
typedef std::unique_ptr<recursive_queue_node> queue_node_ptr;
template<typename... Args>
nestable_invoke_policy(local_actor* parent, Args&&... args)
: m_last_dequeued(parent->last_dequeued())
, m_last_sender(parent->last_sender())
, m_filter_policy(std::forward<Args>(args)...)
enum handle_message_result
{
}
hm_timeout_msg,
hm_skip_msg,
hm_drop_msg,
hm_cache_msg,
hm_success
};
template<typename... Args>
bool invoke_from_cache(partial_function& fun, Args... args)
template<class FunOrBehavior>
bool invoke_from_cache(FunOrBehavior& fun)
{
auto i = m_cache.begin();
auto e = m_cache.end();
while (i != e)
{
switch (handle_message(*(*i), fun, args...))
switch (this->handle_message(*(*i), fun))
{
case hm_drop_msg:
{
i = m_cache.erase(i);
break;
}
case hm_success:
{
this->release_node(i->release());
m_cache.erase(i);
return true;
}
case hm_drop_msg:
{
this->release_node(i->release());
i = m_cache.erase(i);
break;
}
case hm_skip_msg:
case hm_cache_msg:
{
++i;
break;
}
default: CPPA_CRITICAL("illegal result of handle_message");
default:
{
CPPA_CRITICAL("illegal result of handle_message");
}
}
}
return false;
}
template<typename... Args>
bool invoke(queue_node_ptr& ptr, partial_function& fun, Args... args)
template<class FunOrBehavior>
bool invoke(recursive_queue_node* node, FunOrBehavior& fun)
{
switch (handle_message(*ptr, fun, args...))
switch (this->handle_message(*node, fun))
{
case hm_drop_msg:
case hm_success:
{
break;
this->release_node(node);
return true;
}
case hm_success:
case hm_drop_msg:
{
return true; // done
this->release_node(node);
break;
}
case hm_cache_msg:
{
m_cache.push_back(std::move(ptr));
m_cache.emplace_back(node);
break;
}
case hm_skip_msg:
......@@ -120,54 +126,77 @@ class nestable_invoke_policy
private:
enum handle_message_result
std::list<std::unique_ptr<recursive_queue_node> > m_cache;
inline Derived* dthis()
{
hm_timeout_msg,
hm_skip_msg,
hm_drop_msg,
hm_cache_msg,
hm_success
};
return static_cast<Derived*>(this);
}
inline Derived const* dthis() const
{
return static_cast<Derived const*>(this);
}
void handle_timeout(behavior& bhvr)
{
bhvr.handle_timeout();
}
any_tuple& m_last_dequeued;
actor_ptr& m_last_sender;
FilterPolicy m_filter_policy;
std::list<queue_node_ptr> m_cache;
void handle_timeout(partial_function&)
{
CPPA_CRITICAL("handle_timeout(partial_function&)");
}
template<typename... Args>
template<class FunOrBehavior>
handle_message_result handle_message(recursive_queue_node& node,
partial_function& fun,
Args... args)
FunOrBehavior& fun)
{
if (node.marked)
{
return hm_skip_msg;
}
if (m_filter_policy(node.msg, args...))
{
return hm_drop_msg;
}
std::swap(m_last_dequeued, node.msg);
std::swap(m_last_sender, node.sender);
switch (dthis()->filter_msg(node.msg))
{
typename recursive_queue_node::guard qguard{&node};
if (fun(m_last_dequeued))
case normal_exit_signal:
case expired_timeout_message:
{
// client calls erase(iter)
qguard.release();
m_last_dequeued.reset();
m_last_sender.reset();
return hm_drop_msg;
}
case timeout_message:
{
handle_timeout(fun);
return hm_success;
}
case ordinary_message:
{
break;
}
default:
{
CPPA_CRITICAL("illegal result of filter_msg");
}
}
std::swap(this->m_last_dequeued, node.msg);
std::swap(this->m_last_sender, node.sender);
dthis()->push_timeout();
node.marked = true;
if (fun(this->m_last_dequeued))
{
this->m_last_dequeued.reset();
this->m_last_sender.reset();
return hm_success;
}
// no match (restore members)
std::swap(m_last_dequeued, node.msg);
std::swap(m_last_sender, node.sender);
std::swap(this->m_last_dequeued, node.msg);
std::swap(this->m_last_sender, node.sender);
dthis()->pop_timeout();
node.marked = false;
return hm_cache_msg;
}
};
} }
} } // namespace cppa::detail
#endif // NESTABLE_INVOKE_POLICY_HPP
#endif // NESTABLE_RECEIVE_ACTOR_HPP
......@@ -43,13 +43,11 @@ struct recursive_queue_node
actor_ptr sender;
any_tuple msg;
inline recursive_queue_node()
: next(nullptr)
, marked(false)
inline recursive_queue_node() : next(nullptr), marked(false)
{
}
inline recursive_queue_node(actor* from, any_tuple content)
inline recursive_queue_node(actor* from, any_tuple&& content)
: next(nullptr)
, marked(false)
, sender(from)
......@@ -57,27 +55,11 @@ struct recursive_queue_node
{
}
recursive_queue_node(recursive_queue_node&&) = delete;
recursive_queue_node(recursive_queue_node const&) = delete;
recursive_queue_node& operator=(recursive_queue_node&&) = delete;
recursive_queue_node& operator=(recursive_queue_node const&) = delete;
struct guard
{
recursive_queue_node* m_node;
inline guard(recursive_queue_node* ptr) : m_node(ptr)
{
ptr->marked = true;
}
inline void release()
{
m_node = nullptr;
}
inline ~guard()
{
if (m_node) m_node->marked = false;
}
};
};
} } // namespace cppa::detail
......
......@@ -41,15 +41,19 @@
#include "cppa/pattern.hpp"
#include "cppa/detail/yield_interface.hpp"
#include "cppa/detail/nestable_invoke_policy.hpp"
#include "cppa/detail/nestable_receive_actor.hpp"
#include "cppa/detail/abstract_scheduled_actor.hpp"
namespace cppa { namespace detail {
class yielding_actor : public abstract_scheduled_actor
class yielding_actor
: public nestable_receive_actor<yielding_actor,
abstract_scheduled_actor>
{
typedef abstract_scheduled_actor super;
typedef nestable_receive_actor<yielding_actor,
abstract_scheduled_actor>
super;
public:
......@@ -61,6 +65,16 @@ class yielding_actor : public abstract_scheduled_actor
void resume(util::fiber* from, scheduler::callback* callback); //override
inline void push_timeout()
{
++m_active_timeout_id;
}
inline void pop_timeout()
{
--m_active_timeout_id;
}
private:
typedef std::unique_ptr<recursive_queue_node> queue_node_ptr;
......@@ -69,53 +83,8 @@ class yielding_actor : public abstract_scheduled_actor
void yield_until_not_empty();
struct filter_policy;
friend struct filter_policy;
struct filter_policy
{
yielding_actor* m_parent;
inline filter_policy(yielding_actor* parent) : m_parent(parent) { }
inline bool operator()(any_tuple const& msg)
{
return m_parent->filter_msg(msg) != ordinary_message;
}
inline bool operator()(any_tuple const& msg,
behavior* bhvr,
bool* timeout_occured)
{
switch (m_parent->filter_msg(msg))
{
case timeout_message:
{
bhvr->handle_timeout();
*timeout_occured = true;
return true;
}
case normal_exit_signal:
case expired_timeout_message:
{
return true;
}
case ordinary_message:
{
return false;
}
default: CPPA_CRITICAL("illegal result of filter_msg");
}
return false;
}
};
util::fiber m_fiber;
std::function<void()> m_behavior;
nestable_invoke_policy<filter_policy> m_invoke;
};
......
......@@ -131,7 +131,7 @@ class single_reader_queue
*/
inline bool empty() const
{
return !m_head && m_stack.load() == nullptr;
return m_head == nullptr && m_stack.load() == nullptr;
}
/**
......@@ -170,7 +170,7 @@ class single_reader_queue
if (empty())
{
lock_type guard(m_mtx);
while (!(m_stack.load()))
while (m_stack.load() == nullptr)
{
if (detail::wait_until(guard, m_cv, timeout) == false)
{
......
......@@ -54,7 +54,7 @@ class scheduled_actor : public local_actor
public:
scheduled_actor();
scheduled_actor(bool enable_pending_enqueue = false);
scheduled_actor* next; // intrusive next pointer
......
......@@ -32,9 +32,11 @@
#include "cppa/to_string.hpp"
#include "cppa/self.hpp"
#include "cppa/detail/invokable.hpp"
#include "cppa/abstract_event_based_actor.hpp"
#include "cppa/detail/invokable.hpp"
#include "cppa/detail/filter_result.hpp"
namespace cppa {
abstract_event_based_actor::abstract_event_based_actor()
......@@ -55,15 +57,16 @@ void abstract_event_based_actor::dequeue(partial_function&)
auto abstract_event_based_actor::handle_message(mailbox_element& node) -> handle_message_result
{
CPPA_REQUIRE(node.marked == false);
CPPA_REQUIRE(m_loop_stack.empty() == false);
auto& bhvr = *(m_loop_stack.back());
switch (filter_msg(node.msg))
{
case normal_exit_signal:
case expired_timeout_message:
case detail::normal_exit_signal:
case detail::expired_timeout_message:
return drop_msg;
case timeout_message:
case detail::timeout_message:
m_has_pending_timeout_request = false;
CPPA_REQUIRE(bhvr.timeout().valid());
bhvr.handle_timeout();
......@@ -79,8 +82,6 @@ auto abstract_event_based_actor::handle_message(mailbox_element& node) -> handle
}
std::swap(m_last_dequeued, node.msg);
std::swap(m_last_sender, node.sender);
//m_last_dequeued = node.msg;
//m_last_sender = node.sender;
if ((bhvr.get_partial_function())(m_last_dequeued))
{
m_last_dequeued.reset();
......
......@@ -33,6 +33,7 @@
#include <algorithm>
#include "cppa/self.hpp"
#include "cppa/to_string.hpp"
#include "cppa/exception.hpp"
#include "cppa/detail/matches.hpp"
#include "cppa/detail/invokable.hpp"
......@@ -41,7 +42,7 @@
namespace cppa { namespace detail {
converted_thread_context::converted_thread_context()
: m_exit_msg_pattern(atom(":Exit")), m_invoke(this, this)
: m_exit_msg_pattern(atom(":Exit"))
{
}
......@@ -61,17 +62,24 @@ void converted_thread_context::cleanup(std::uint32_t reason)
void converted_thread_context::enqueue(actor* sender, any_tuple msg)
{
# ifdef CPPA_DEBUG
auto node = fetch_node(sender, std::move(msg));
CPPA_REQUIRE(node->marked == false);
m_mailbox.push_back(node);
# else
m_mailbox.push_back(fetch_node(sender, std::move(msg)));
# endif
}
void converted_thread_context::dequeue(partial_function& fun) // override
{
if (m_invoke.invoke_from_cache(fun) == false)
if (invoke_from_cache(fun) == false)
{
queue_node_ptr e{m_mailbox.pop()};
while (m_invoke.invoke(e, fun) == false)
recursive_queue_node* e = m_mailbox.pop();
CPPA_REQUIRE(e->marked == false);
while (invoke(e, fun) == false)
{
e.reset(m_mailbox.pop());
e = m_mailbox.pop();
}
}
}
......@@ -81,21 +89,40 @@ void converted_thread_context::dequeue(behavior& bhvr) // override
auto& fun = bhvr.get_partial_function();
if (bhvr.timeout().valid() == false)
{
dequeue(fun);
return;
// suppress virtual function call
converted_thread_context::dequeue(fun);
}
if (m_invoke.invoke_from_cache(fun) == false)
else if (invoke_from_cache(fun) == false)
{
auto timeout = now();
timeout += bhvr.timeout();
queue_node_ptr e{m_mailbox.try_pop(timeout)};
while (e)
recursive_queue_node* e = m_mailbox.try_pop(timeout);
while (e != nullptr)
{
if (m_invoke.invoke(e, fun)) return;
else e.reset(m_mailbox.try_pop(timeout));
if (e->marked)
{
std::cout << "ooops: " << to_string(e->msg) << std::endl;
}
CPPA_REQUIRE(e->marked == false);
if (invoke(e, fun)) return;
e = m_mailbox.try_pop(timeout);
}
bhvr.handle_timeout();
}
}
filter_result converted_thread_context::filter_msg(any_tuple const& msg)
{
if (m_trap_exit == false && matches(msg, m_exit_msg_pattern))
{
auto reason = msg.get_as<std::uint32_t>(1);
if (reason != exit_reason::normal)
{
quit(reason);
}
return normal_exit_signal;
}
return ordinary_message;
}
} } // namespace cppa::detail
......@@ -32,8 +32,8 @@
namespace cppa {
scheduled_actor::scheduled_actor() : local_actor(true)
, next(nullptr), m_scheduler(nullptr)
scheduled_actor::scheduled_actor(bool enable_pending_enqueue)
: local_actor(enable_pending_enqueue), next(nullptr), m_scheduler(nullptr)
{
}
......
......@@ -110,7 +110,7 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self)
decltype(detail::now()) now;
bool done = false;
// message handling rules
auto handle_msg =
auto mfun =
(
on<util::duration,actor_ptr,anything>() >> [&](const util::duration& d,
const actor_ptr&)
......@@ -152,6 +152,7 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self)
while (it != messages.end() && (it->first) <= now)
{
queue_node_ptr ptr{std::move(it->second)};
CPPA_REQUIRE(ptr->marked == false);
auto whom = const_cast<actor_ptr*>(
reinterpret_cast<actor_ptr const*>(
ptr->msg.at(1)));
......@@ -171,7 +172,7 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self)
}
}
}
handle_msg(msg_ptr->msg);
mfun(msg_ptr->msg);
}
}
......
......@@ -42,7 +42,6 @@ namespace cppa { namespace detail {
yielding_actor::yielding_actor(std::function<void()> fun)
: m_fiber(&yielding_actor::run, this)
, m_behavior(fun)
, m_invoke(this, this)
{
}
......@@ -89,19 +88,19 @@ void yielding_actor::yield_until_not_empty()
void yielding_actor::dequeue(partial_function& fun)
{
if (m_invoke.invoke_from_cache(fun) == false)
if (invoke_from_cache(fun) == false)
{
queue_node_ptr e;
recursive_queue_node* e = nullptr;
do
{
e.reset(m_mailbox.try_pop());
e = m_mailbox.try_pop();
while (!e)
{
yield_until_not_empty();
e.reset(m_mailbox.try_pop());
e = m_mailbox.try_pop();
}
}
while (!m_invoke.invoke(e, fun));
while (invoke(e, fun) == false);
}
}
......@@ -113,22 +112,20 @@ void yielding_actor::dequeue(behavior& bhvr)
// suppress virtual function call
yielding_actor::dequeue(fun);
}
else if (m_invoke.invoke_from_cache(fun) == false)
else if (invoke_from_cache(bhvr) == false)
{
request_timeout(bhvr.timeout());
bool timeout_occured = false;
queue_node_ptr e;
recursive_queue_node* e = nullptr;
do
{
e.reset(m_mailbox.try_pop());
e = m_mailbox.try_pop();
while (!e)
{
yield_until_not_empty();
e.reset(m_mailbox.try_pop());
e = m_mailbox.try_pop();
}
}
while ( !m_invoke.invoke(e, fun, &bhvr, &timeout_occured)
&& !timeout_occured);
while (invoke(e, bhvr) == false);
}
}
......
......@@ -198,10 +198,10 @@ int main(int argc, char** argv)
RUN_TEST(test__fixed_vector);
RUN_TEST(test__tuple);
RUN_TEST(test__serialization);
RUN_TEST(test__local_group);
RUN_TEST(test__atom);
RUN_TEST(test__yield_interface);
RUN_TEST(test__spawn);
RUN_TEST(test__local_group);
RUN_TEST_A3(test__remote_actor, argv[0], false, args);
cout << endl
<< "error(s) in all tests: " << errors
......
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