Commit fd309798 authored by Dominik Charousset's avatar Dominik Charousset

maintenance

parent 22d4342c
...@@ -261,3 +261,4 @@ unit_testing/test__yield_interface.cpp ...@@ -261,3 +261,4 @@ unit_testing/test__yield_interface.cpp
cppa/detail/receive_policy.hpp cppa/detail/receive_policy.hpp
cppa/detail/behavior_stack.hpp cppa/detail/behavior_stack.hpp
src/behavior_stack.cpp src/behavior_stack.cpp
cppa/detail/stacked_actor_mixin.hpp
...@@ -43,6 +43,7 @@ ...@@ -43,6 +43,7 @@
#include "cppa/detail/receive_policy.hpp" #include "cppa/detail/receive_policy.hpp"
#include "cppa/detail/behavior_stack.hpp" #include "cppa/detail/behavior_stack.hpp"
#include "cppa/detail/yield_interface.hpp" #include "cppa/detail/yield_interface.hpp"
#include "cppa/detail/stacked_actor_mixin.hpp"
#include "cppa/detail/abstract_scheduled_actor.hpp" #include "cppa/detail/abstract_scheduled_actor.hpp"
namespace cppa { namespace cppa {
...@@ -68,39 +69,41 @@ class context_switching_actor : public scheduled_actor { ...@@ -68,39 +69,41 @@ class context_switching_actor : public scheduled_actor {
#else #else
class context_switching_actor : public detail::abstract_scheduled_actor { class context_switching_actor : public detail::stacked_actor_mixin<
context_switching_actor,
detail::abstract_scheduled_actor> {
friend class detail::receive_policy; friend class detail::receive_policy;
typedef detail::abstract_scheduled_actor super;
public: public:
context_switching_actor(std::function<void()> fun); context_switching_actor(std::function<void()> fun);
void dequeue(behavior& bhvr); //override
void dequeue(partial_function& fun); //override
resume_result resume(util::fiber* from); //override resume_result resume(util::fiber* from); //override
void unbecome();
protected: protected:
context_switching_actor(); context_switching_actor();
virtual void run(); virtual void run();
void do_become(behavior* bhvr, bool ownership, bool discard);
private: private:
// required by detail::nestable_receive_policy // required by detail::nestable_receive_policy
static const detail::receive_policy_flag receive_flag = detail::rp_nestable; static const detail::receive_policy_flag receive_flag = detail::rp_nestable;
detail::recursive_queue_node* receive_node(); detail::recursive_queue_node* receive_node();
inline void push_timeout() { ++m_active_timeout_id; } inline int init_timeout(const util::duration& timeout) {
inline void pop_timeout() { --m_active_timeout_id; } // request timeout message
request_timeout(timeout);
return 0;
}
inline detail::recursive_queue_node* try_receive_node() {
return m_mailbox.try_pop();
}
inline detail::recursive_queue_node* try_receive_node(int) {
// timeout is triggered from an explicit timeout message
return receive_node();
}
// required by util::fiber // required by util::fiber
static void trampoline(void* _this); static void trampoline(void* _this);
......
...@@ -471,7 +471,7 @@ struct spawn_fwd_ { ...@@ -471,7 +471,7 @@ struct spawn_fwd_ {
template<> template<>
struct spawn_fwd_<self_type> { struct spawn_fwd_<self_type> {
static inline actor_ptr _(const self_type&) { return self; } static inline actor_ptr _(const self_type& s) { return s.get(); }
}; };
/** /**
...@@ -494,7 +494,7 @@ inline actor_ptr spawn(F bhvr, Arg0&& arg0, Args&&... args) { ...@@ -494,7 +494,7 @@ inline actor_ptr spawn(F bhvr, Arg0&& arg0, Args&&... args) {
/** /**
* @ingroup ActorManagement * @ingroup ActorManagement
* @brief Spawns a new context-switching actor that executes * @brief Spawns a new context-switching actor that executes
* <tt>bhvr(arg0, args...)</tt>, equal to * <tt>bhvr(arg0, args...)</tt>. Equal to
* <tt>spawn<scheduled>(bhvr, arg0, args...)</tt>. * <tt>spawn<scheduled>(bhvr, arg0, args...)</tt>.
* @param bhvr A functor implementing the actor's behavior. * @param bhvr A functor implementing the actor's behavior.
* @param arg0 First argument to @p bhvr. * @param arg0 First argument to @p bhvr.
...@@ -517,7 +517,7 @@ inline actor_ptr spawn(F&& bhvr, Arg0&& arg0, Args&&... args) { ...@@ -517,7 +517,7 @@ inline actor_ptr spawn(F&& bhvr, Arg0&& arg0, Args&&... args) {
* @param what Message elements. * @param what Message elements.
*/ */
template<typename... Args> template<typename... Args>
void send(channel_ptr& whom, const Args&... what); void send(const channel_ptr& whom, Args&&... what);
/** /**
* @ingroup MessageHandling * @ingroup MessageHandling
...@@ -531,30 +531,18 @@ void send(channel_ptr& whom, const Args&... what); ...@@ -531,30 +531,18 @@ void send(channel_ptr& whom, const Args&... what);
* @param what Message as instance of {@link any_tuple}. * @param what Message as instance of {@link any_tuple}.
* @returns @p whom. * @returns @p whom.
*/ */
channel_ptr& operator<<(channel_ptr& whom, const any_tuple& what); const channel_ptr& operator<<(const channel_ptr& whom, any_tuple what);
#else #else
namespace detail { namespace detail {
inline void send_impl(channel* whom, any_tuple&& what) { inline void send_impl(channel* whom, any_tuple&& what) {
if (whom != nullptr) whom->enqueue(self.get(), std::move(what));
}
template<typename Arg0, typename... Args>
inline void send_tpl_impl(channel* whom, Arg0&& arg0, Args&&... args) {
if (whom != nullptr) {
whom->enqueue(self.get(), make_any_tuple(std::forward<Arg0>(arg0),
std::forward<Args>(args)...));
}
}
inline void send_impl(actor* whom, any_tuple&& what) {
if (whom) self->send_message(whom, std::move(what)); if (whom) self->send_message(whom, std::move(what));
} }
template<typename Arg0, typename... Args> template<typename Arg0, typename... Args>
inline void send_tpl_impl(actor* whom, Arg0&& arg0, Args&&... args) { inline void send_tpl_impl(channel* whom, Arg0&& arg0, Args&&... args) {
if (whom) { if (whom) {
self->send_message(whom, make_any_tuple(std::forward<Arg0>(arg0), self->send_message(whom, make_any_tuple(std::forward<Arg0>(arg0),
std::forward<Args>(args)...)); std::forward<Args>(args)...));
...@@ -564,8 +552,8 @@ inline void send_tpl_impl(actor* whom, Arg0&& arg0, Args&&... args) { ...@@ -564,8 +552,8 @@ inline void send_tpl_impl(actor* whom, Arg0&& arg0, Args&&... args) {
} // namespace detail } // namespace detail
template<class C> template<class C>
typename std::enable_if<std::is_base_of<channel, C>::value, inline typename std::enable_if<std::is_base_of<channel, C>::value,
const intrusive_ptr<C>& >::type const intrusive_ptr<C>& >::type
operator<<(const intrusive_ptr<C>& whom, any_tuple what) { operator<<(const intrusive_ptr<C>& whom, any_tuple what) {
detail::send_impl(whom.get(), std::move(what)); detail::send_impl(whom.get(), std::move(what));
return whom; return whom;
...@@ -576,7 +564,6 @@ inline const self_type& operator<<(const self_type& s, any_tuple what) { ...@@ -576,7 +564,6 @@ inline const self_type& operator<<(const self_type& s, any_tuple what) {
return s; return s;
} }
template<class C, typename Arg0, typename... Args> template<class C, typename Arg0, typename... Args>
inline typename std::enable_if<std::is_base_of<channel, C>::value>::type inline typename std::enable_if<std::is_base_of<channel, C>::value>::type
send(const intrusive_ptr<C>& whom, Arg0&& arg0, Args&&... args) { send(const intrusive_ptr<C>& whom, Arg0&& arg0, Args&&... args) {
...@@ -585,7 +572,7 @@ send(const intrusive_ptr<C>& whom, Arg0&& arg0, Args&&... args) { ...@@ -585,7 +572,7 @@ send(const intrusive_ptr<C>& whom, Arg0&& arg0, Args&&... args) {
std::forward<Args>(args)...); std::forward<Args>(args)...);
} }
// matches "send(this, ...)" in event-based actors // matches "send(this, ...)" and "send(self, ...)"
template<typename Arg0, typename... Args> template<typename Arg0, typename... Args>
inline void send(local_actor* whom, Arg0&& arg0, Args&&... args) { inline void send(local_actor* whom, Arg0&& arg0, Args&&... args) {
detail::send_tpl_impl(whom, detail::send_tpl_impl(whom,
...@@ -601,8 +588,8 @@ inline void send(local_actor* whom, Arg0&& arg0, Args&&... args) { ...@@ -601,8 +588,8 @@ inline void send(local_actor* whom, Arg0&& arg0, Args&&... args) {
* @param what Message elements. * @param what Message elements.
*/ */
template<typename... Args> template<typename... Args>
inline void reply(const Args&... what) { inline void reply(Args&&... what) {
send(self->last_sender(), what...); send(self->last_sender(), std::forward<Args>(what)...);
} }
/** /**
...@@ -614,10 +601,13 @@ inline void reply(const Args&... what) { ...@@ -614,10 +601,13 @@ inline void reply(const Args&... what) {
* @param what Message elements. * @param what Message elements.
*/ */
template<class Rep, class Period, typename... Args> template<class Rep, class Period, typename... Args>
inline void delayed_send(actor_ptr whom, inline void delayed_send(const actor_ptr& whom,
const std::chrono::duration<Rep, Period>& rel_time, const std::chrono::duration<Rep, Period>& rel_time,
const Args&... what) { Args&&... what) {
get_scheduler()->delayed_send(whom, rel_time, what...); if (whom) {
get_scheduler()->delayed_send(whom, rel_time,
std::forward<Args>(what)...);
}
} }
/** /**
...@@ -630,8 +620,8 @@ inline void delayed_send(actor_ptr whom, ...@@ -630,8 +620,8 @@ inline void delayed_send(actor_ptr whom,
*/ */
template<class Rep, class Period, typename... Args> template<class Rep, class Period, typename... Args>
inline void delayed_reply(const std::chrono::duration<Rep, Period>& rel_time, inline void delayed_reply(const std::chrono::duration<Rep, Period>& rel_time,
const Args&... what) { Args&&... what) {
delayed_send(self->last_sender(), rel_time, what...); delayed_send(self->last_sender(), rel_time, std::forward<Args>(what)...);
} }
/** /**
......
...@@ -110,6 +110,14 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor> { ...@@ -110,6 +110,14 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor> {
reset_timeout(); reset_timeout();
} }
inline void push_timeout() {
++m_active_timeout_id;
}
inline void pop_timeout() {
--m_active_timeout_id;
}
bool m_has_pending_timeout_request; bool m_has_pending_timeout_request;
std::uint32_t m_active_timeout_id; std::uint32_t m_active_timeout_id;
......
...@@ -124,6 +124,37 @@ class receive_policy { ...@@ -124,6 +124,37 @@ class receive_policy {
} }
} }
template<class Client>
void receive(Client* client, behavior& bhvr) {
auto& fun = bhvr.get_partial_function();
if (bhvr.timeout().valid() == false) {
receive(client, fun);
}
else if (invoke_from_cache(client, fun) == false) {
if (bhvr.timeout().is_zero()) {
recursive_queue_node* e = nullptr;
while ((e = client->try_receive_node()) != nullptr) {
CPPA_REQUIRE(e->marked == false);
if (invoke(client, e, bhvr)) {
return; // done
}
}
handle_timeout(client, bhvr);
}
else {
auto timeout = client->init_timeout(bhvr.timeout());
recursive_queue_node* e = nullptr;
while ((e = client->try_receive_node(timeout)) != nullptr) {
CPPA_REQUIRE(e->marked == false);
if (invoke(client, e, bhvr)) {
return; // done
}
}
handle_timeout(client, bhvr);
}
}
}
private: private:
typedef std::integral_constant<receive_policy_flag, rp_nestable> nestable; typedef std::integral_constant<receive_policy_flag, rp_nestable> nestable;
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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 CPPA_STACKED_ACTOR_MIXIN_HPP
#define CPPA_STACKED_ACTOR_MIXIN_HPP
#include <memory>
#include "cppa/behavior.hpp"
#include "cppa/detail/behavior_stack.hpp"
#include "cppa/detail/recursive_queue_node.hpp"
namespace cppa { namespace detail {
template<class Derived, class Base>
class stacked_actor_mixin : public Base {
inline Derived* dthis() {
return static_cast<Derived*>(this);
}
public:
virtual void unbecome() {
if (m_bhvr_stack_ptr) {
m_bhvr_stack_ptr->pop_back();
}
else if (this->initialized()) {
this->quit();
}
}
virtual void dequeue(partial_function& fun) {
m_recv_policy.receive(dthis(), fun);
}
virtual void dequeue(behavior& bhvr) {
m_recv_policy.receive(dthis(), bhvr);
}
protected:
receive_policy m_recv_policy;
std::unique_ptr<behavior_stack> m_bhvr_stack_ptr;
virtual void do_become(behavior* bhvr, bool owns_bhvr, bool discard_old) {
if (this->m_bhvr_stack_ptr) {
if (discard_old) this->m_bhvr_stack_ptr->pop_back();
this->m_bhvr_stack_ptr->push_back(bhvr, owns_bhvr);
}
else {
this->m_bhvr_stack_ptr.reset(new behavior_stack);
if (this->initialized()) {
this->m_bhvr_stack_ptr->exec();
this->quit();
}
}
}
};
} } // namespace cppa::detail
#endif // CPPA_STACKED_ACTOR_MIXIN_HPP
...@@ -50,9 +50,7 @@ class fsm_actor : public event_based_actor { ...@@ -50,9 +50,7 @@ class fsm_actor : public event_based_actor {
* @brief Overrides {@link event_based_actor::init()} and sets * @brief Overrides {@link event_based_actor::init()} and sets
* the initial actor behavior to <tt>Derived::init_state</tt>. * the initial actor behavior to <tt>Derived::init_state</tt>.
*/ */
void init() { void init() { become(&(static_cast<Derived*>(this)->init_state)); }
become(&(static_cast<Derived*>(this)->init_state));
}
}; };
......
...@@ -255,14 +255,16 @@ class local_actor : public actor { ...@@ -255,14 +255,16 @@ class local_actor : public actor {
local_actor(bool is_scheduled = false); local_actor(bool is_scheduled = false);
inline void send_message(channel* whom, any_tuple what) { virtual bool initialized() = 0;
inline void send_message(channel* whom, any_tuple&& what) {
whom->enqueue(this, std::move(what)); whom->enqueue(this, std::move(what));
} }
inline void send_message(actor* whom, any_tuple what) { inline void send_message(actor* whom, any_tuple&& what) {
if (m_chaining && !m_chained) { if (m_chaining && !m_chained_actor) {
if (whom->chained_enqueue(this, std::move(what))) { if (whom->chained_enqueue(this, std::move(what))) {
m_chained = whom; m_chained_actor = whom;
} }
} }
else { else {
...@@ -271,7 +273,7 @@ class local_actor : public actor { ...@@ -271,7 +273,7 @@ class local_actor : public actor {
} }
inline actor_ptr& chained_actor() { inline actor_ptr& chained_actor() {
return m_chained; return m_chained_actor;
} }
protected: protected:
...@@ -279,15 +281,15 @@ class local_actor : public actor { ...@@ -279,15 +281,15 @@ class local_actor : public actor {
bool m_chaining; bool m_chaining;
bool m_trap_exit; bool m_trap_exit;
bool m_is_scheduled; bool m_is_scheduled;
actor_ptr m_chained;
actor_ptr m_last_sender; actor_ptr m_last_sender;
actor_ptr m_chained_actor;
any_tuple m_last_dequeued; any_tuple m_last_dequeued;
# endif // CPPA_DOCUMENTATION # endif // CPPA_DOCUMENTATION
protected: protected:
virtual void do_become(behavior* bhvr, bool ownership, bool discard) = 0; virtual void do_become(behavior* bhvr, bool owns_ptr, bool discard_old) = 0;
}; };
......
...@@ -81,6 +81,8 @@ class scheduled_actor : public local_actor { ...@@ -81,6 +81,8 @@ class scheduled_actor : public local_actor {
scheduler* m_scheduler; scheduler* m_scheduler;
bool initialized();
}; };
} // namespace cppa } // namespace cppa
......
...@@ -109,10 +109,11 @@ class scheduler { ...@@ -109,10 +109,11 @@ class scheduler {
template<typename Duration, typename... Data> template<typename Duration, typename... Data>
void delayed_send(const actor_ptr& to, void delayed_send(const actor_ptr& to,
const Duration& rel_time, const Data&... data) { const Duration& rel_time,
Data&&... data) {
static_assert(sizeof...(Data) > 0, "no message to send"); static_assert(sizeof...(Data) > 0, "no message to send");
any_tuple data_tup = make_cow_tuple(data...); auto sub = make_any_tuple(std::forward<Data>(data)...);
any_tuple tup = make_cow_tuple(util::duration(rel_time), to, data_tup); auto tup = make_any_tuple(util::duration{rel_time}, to, std::move(sub));
delayed_send_helper()->enqueue(self, std::move(tup)); delayed_send_helper()->enqueue(self, std::move(tup));
} }
......
...@@ -53,6 +53,7 @@ ...@@ -53,6 +53,7 @@
#include "cppa/detail/receive_policy.hpp" #include "cppa/detail/receive_policy.hpp"
#include "cppa/detail/behavior_stack.hpp" #include "cppa/detail/behavior_stack.hpp"
#include "cppa/detail/stacked_actor_mixin.hpp"
#include "cppa/detail/recursive_queue_node.hpp" #include "cppa/detail/recursive_queue_node.hpp"
namespace cppa { namespace cppa {
...@@ -69,38 +70,28 @@ class thread_mapped_actor : public local_actor { }; ...@@ -69,38 +70,28 @@ class thread_mapped_actor : public local_actor { };
class self_type; class self_type;
class thread_mapped_actor : public detail::abstract_actor<local_actor> { class thread_mapped_actor : public detail::stacked_actor_mixin<
thread_mapped_actor,
detail::abstract_actor<local_actor> > {
friend class self_type;
friend class detail::receive_policy; friend class detail::receive_policy;
typedef detail::abstract_actor<local_actor> super;
public: public:
void quit(std::uint32_t reason = exit_reason::normal); //override void quit(std::uint32_t reason = exit_reason::normal); //override
void enqueue(actor* sender, any_tuple msg); //override void enqueue(actor* sender, any_tuple msg); //override
void dequeue(behavior& rules); //override
void dequeue(partial_function& rules); //override
detail::filter_result filter_msg(const any_tuple& msg); detail::filter_result filter_msg(const any_tuple& msg);
inline decltype(m_mailbox)& mailbox() { return m_mailbox; } inline decltype(m_mailbox)& mailbox() { return m_mailbox; }
void unbecome();
protected: protected:
void do_become(behavior* bhvr, bool ownership, bool discard); bool initialized();
private: private:
detail::receive_policy m_recv_policy;
std::unique_ptr<detail::behavior_stack> m_bhvr_stack_ptr;
// required by nestable_receive_policy // required by nestable_receive_policy
static const detail::receive_policy_flag receive_flag = detail::rp_nestable; static const detail::receive_policy_flag receive_flag = detail::rp_nestable;
inline void push_timeout() { } inline void push_timeout() { }
...@@ -108,6 +99,18 @@ class thread_mapped_actor : public detail::abstract_actor<local_actor> { ...@@ -108,6 +99,18 @@ class thread_mapped_actor : public detail::abstract_actor<local_actor> {
inline detail::recursive_queue_node* receive_node() { inline detail::recursive_queue_node* receive_node() {
return m_mailbox.pop(); return m_mailbox.pop();
} }
inline auto init_timeout(const util::duration& tout) -> decltype(std::chrono::high_resolution_clock::now()) {
auto result = std::chrono::high_resolution_clock::now();
result += tout;
return result;
}
inline detail::recursive_queue_node* try_receive_node() {
return m_mailbox.try_pop();
}
template<typename Timeout>
inline detail::recursive_queue_node* try_receive_node(const Timeout& tout) {
return m_mailbox.try_pop(tout);
}
inline void handle_timeout(behavior& bhvr) { inline void handle_timeout(behavior& bhvr) {
bhvr.handle_timeout(); bhvr.handle_timeout();
} }
......
...@@ -93,29 +93,6 @@ detail::recursive_queue_node* context_switching_actor::receive_node() { ...@@ -93,29 +93,6 @@ detail::recursive_queue_node* context_switching_actor::receive_node() {
return e; return e;
} }
void context_switching_actor::dequeue(partial_function& fun) {
m_recv_policy.receive(this, fun);
}
void context_switching_actor::dequeue(behavior& bhvr) {
if (bhvr.timeout().valid() == false) {
m_recv_policy.receive(this, bhvr.get_partial_function());
}
else if (m_recv_policy.invoke_from_cache(this, bhvr) == false) {
if (bhvr.timeout().is_zero()) {
for (auto e = m_mailbox.try_pop(); e != 0; e = m_mailbox.try_pop()){
CPPA_REQUIRE(e->marked == false);
if (m_recv_policy.invoke(this, e, bhvr)) return;
}
bhvr.handle_timeout();
}
else {
request_timeout(bhvr.timeout());
while (m_recv_policy.invoke(this, receive_node(), bhvr) == false) {}
}
}
}
resume_result context_switching_actor::resume(util::fiber* from) { resume_result context_switching_actor::resume(util::fiber* from) {
using namespace detail; using namespace detail;
scoped_self_setter sss{this}; scoped_self_setter sss{this};
...@@ -150,31 +127,6 @@ resume_result context_switching_actor::resume(util::fiber* from) { ...@@ -150,31 +127,6 @@ resume_result context_switching_actor::resume(util::fiber* from) {
} }
} }
void context_switching_actor::unbecome() {
if (m_bhvr_stack_ptr) {
m_bhvr_stack_ptr->pop_back();
}
else if (m_scheduler != nullptr) {
quit();
}
}
void context_switching_actor::do_become(behavior* bhvr, bool ownership, bool discard) {
if (m_bhvr_stack_ptr) {
if (discard) m_bhvr_stack_ptr->pop_back();
m_bhvr_stack_ptr->push_back(bhvr, ownership);
}
else {
m_bhvr_stack_ptr.reset(new detail::behavior_stack);
// scheduler == nullptr if and only if become() is called inside init()
if (m_scheduler != nullptr) {
m_bhvr_stack_ptr->exec();
quit();
}
}
}
} // namespace cppa } // namespace cppa
#else // ifdef CPPA_DISABLE_CONTEXT_SWITCHING #else // ifdef CPPA_DISABLE_CONTEXT_SWITCHING
......
...@@ -54,4 +54,8 @@ scheduled_actor* scheduled_actor::attach_to_scheduler(scheduler* sched) { ...@@ -54,4 +54,8 @@ scheduled_actor* scheduled_actor::attach_to_scheduler(scheduler* sched) {
return this; return this;
} }
bool scheduled_actor::initialized() {
return m_scheduler != nullptr;
}
} // namespace cppa } // namespace cppa
...@@ -43,7 +43,7 @@ ...@@ -43,7 +43,7 @@
namespace cppa { namespace cppa {
void thread_mapped_actor::quit(std::uint32_t reason) { void thread_mapped_actor::quit(std::uint32_t reason) {
super::cleanup(reason); cleanup(reason);
// actor_exited should not be catched, but if anyone does, // actor_exited should not be catched, but if anyone does,
// self must point to a newly created instance // self must point to a newly created instance
//self.set(nullptr); //self.set(nullptr);
...@@ -56,37 +56,8 @@ void thread_mapped_actor::enqueue(actor* sender, any_tuple msg) { ...@@ -56,37 +56,8 @@ void thread_mapped_actor::enqueue(actor* sender, any_tuple msg) {
m_mailbox.push_back(node); m_mailbox.push_back(node);
} }
void thread_mapped_actor::dequeue(partial_function& fun) { // override bool thread_mapped_actor::initialized() {
m_recv_policy.receive(this, fun); return true;
}
void thread_mapped_actor::dequeue(behavior& bhvr) { // override
auto& fun = bhvr.get_partial_function();
if (bhvr.timeout().valid() == false) {
// suppress virtual function call
thread_mapped_actor::dequeue(fun);
}
else if (m_recv_policy.invoke_from_cache(this, fun) == false) {
if (bhvr.timeout().is_zero()) {
for (auto e = m_mailbox.try_pop(); e != nullptr; e = m_mailbox.try_pop()) {
CPPA_REQUIRE(e->marked == false);
if (m_recv_policy.invoke(this, e, bhvr)) return;
e = m_mailbox.try_pop();
}
bhvr.handle_timeout();
}
else {
auto timeout = std::chrono::high_resolution_clock::now();
timeout += bhvr.timeout();
detail::recursive_queue_node* e = m_mailbox.try_pop(timeout);
while (e != nullptr) {
CPPA_REQUIRE(e->marked == false);
if (m_recv_policy.invoke(this, e, fun)) return;
e = m_mailbox.try_pop(timeout);
}
bhvr.handle_timeout();
}
}
} }
detail::filter_result thread_mapped_actor::filter_msg(const any_tuple& msg) { detail::filter_result thread_mapped_actor::filter_msg(const any_tuple& msg) {
...@@ -108,25 +79,6 @@ detail::filter_result thread_mapped_actor::filter_msg(const any_tuple& msg) { ...@@ -108,25 +79,6 @@ detail::filter_result thread_mapped_actor::filter_msg(const any_tuple& msg) {
return detail::ordinary_message; return detail::ordinary_message;
} }
void thread_mapped_actor::unbecome() {
if (m_bhvr_stack_ptr) {
m_bhvr_stack_ptr->pop_back();
}
else {
quit();
}
}
void thread_mapped_actor::do_become(behavior* ptr, bool owns_ptr, bool discard){
if (m_bhvr_stack_ptr) {
if (discard) m_bhvr_stack_ptr->pop_back();
m_bhvr_stack_ptr->push_back(ptr, owns_ptr);
}
else {
m_bhvr_stack_ptr.reset(new detail::behavior_stack);
m_bhvr_stack_ptr->exec();
quit();
}
}
} // namespace cppa::detail } // namespace cppa::detail
...@@ -336,22 +336,26 @@ class actor_factory { ...@@ -336,22 +336,26 @@ class actor_factory {
}; };
template<typename InitFun, typename... Members> template<typename InitFun, typename CleanupFun, typename... Members>
class simple_event_based_actor_impl : public event_based_actor { class simple_event_based_actor_impl : public event_based_actor {
public: public:
simple_event_based_actor_impl(InitFun fun) : m_fun(std::move(fun)) { } simple_event_based_actor_impl(InitFun fun, CleanupFun cfun)
: m_init(std::move(fun)), m_cleanup(std::move(cfun)) { }
void init() { apply(); } void init() { apply(); }
void on_exit() { m_cleanup(); }
private: private:
InitFun m_fun; InitFun m_init;
CleanupFun m_cleanup;
std::tuple<Members...> m_members; std::tuple<Members...> m_members;
void apply(typename std::add_pointer<Members>::type... args) { void apply(typename std::add_pointer<Members>::type... args) {
m_fun(args...); m_init(args...);
} }
template<typename... Args> template<typename... Args>
...@@ -371,7 +375,8 @@ class actor_tpl : public actor_factory { ...@@ -371,7 +375,8 @@ class actor_tpl : public actor_factory {
actor_tpl(InitFun fun) : m_init_fun(std::move(fun)) { } actor_tpl(InitFun fun) : m_init_fun(std::move(fun)) { }
actor_ptr spawn() { actor_ptr spawn() {
return cppa::spawn(new simple_event_based_actor_impl<InitFun, Members...>(m_init_fun)); auto void_fun = []() { };
return cppa::spawn(new simple_event_based_actor_impl<InitFun, decltype(void_fun), Members...>(m_init_fun, void_fun));
} }
}; };
...@@ -384,64 +389,6 @@ struct actor_tpl_from_type_list<InitFun, util::type_list<Ts...> > { ...@@ -384,64 +389,6 @@ struct actor_tpl_from_type_list<InitFun, util::type_list<Ts...> > {
typedef actor_tpl<InitFun, Ts...> type; typedef actor_tpl<InitFun, Ts...> type;
}; };
class str_wrapper {
str_wrapper() = delete;
str_wrapper(str_wrapper&&) = delete;
str_wrapper(const str_wrapper&) = delete;
public:
inline str_wrapper(std::string s) : m_str(s) { }
const std::string& str() const {
return m_str;
}
private:
std::string m_str;
};
struct some_integer {
void set(int value) { m_value = value; }
int get() const { return m_value; }
private:
int m_value;
};
template<class T, class MatchExpr = match_expr<>, class Parent = void>
class actor_facade_builder {
public:
private:
MatchExpr m_expr;
};
bool operator==(const str_wrapper& lhs, const std::string& rhs) {
return lhs.str() == rhs;
}
void foobar(const str_wrapper& x, const std::string& y) {
receive (
on(atom("same")).when(gref(x) == gref(y)) >> [&]() {
reply(atom("yes"));
},
on(atom("same")) >> [&]() {
reply(atom("no"));
}
);
}
template<typename Fun> template<typename Fun>
std::unique_ptr<actor_factory> foobaz(Fun fun) { std::unique_ptr<actor_factory> foobaz(Fun fun) {
typedef typename util::get_callable_trait<Fun>::type ctrait; typedef typename util::get_callable_trait<Fun>::type ctrait;
...@@ -457,21 +404,6 @@ size_t test__spawn() { ...@@ -457,21 +404,6 @@ size_t test__spawn() {
using std::string; using std::string;
CPPA_TEST(test__spawn); CPPA_TEST(test__spawn);
/*
actor_tpl<int, float, std::string> atpl;
atpl([](int* i, float* f, std::string* s) {
cout << "SUCCESS: " << *i << ", " << *f << ", \"" << *s << "\"" << endl;
});
*/
/*
actor_ptr tst = actor_facade<some_integer>()
.reply_to().when().with()
.reply_to().when().with()
.spawn();
send(tst, atom("EXIT"), exit_reason::user_defined);
*/
CPPA_IF_VERBOSE(cout << "test send() ... " << std::flush); CPPA_IF_VERBOSE(cout << "test send() ... " << std::flush);
send(self, 1, 2, 3); send(self, 1, 2, 3);
receive(on(1, 2, 3) >> []() { }); receive(on(1, 2, 3) >> []() { });
...@@ -566,7 +498,6 @@ size_t test__spawn() { ...@@ -566,7 +498,6 @@ size_t test__spawn() {
await_all_others_done(); await_all_others_done();
CPPA_IF_VERBOSE(cout << "ok" << endl); CPPA_IF_VERBOSE(cout << "ok" << endl);
/*
CPPA_IF_VERBOSE(cout << "test factory ... " << std::flush); CPPA_IF_VERBOSE(cout << "test factory ... " << std::flush);
auto factory = foobaz([&](int* i, float*, std::string*) { auto factory = foobaz([&](int* i, float*, std::string*) {
self->become ( self->become (
...@@ -581,7 +512,6 @@ size_t test__spawn() { ...@@ -581,7 +512,6 @@ size_t test__spawn() {
} }
); );
}); });
auto foobaz_actor = factory->spawn(); auto foobaz_actor = factory->spawn();
send(foobaz_actor, atom("set_int"), 42); send(foobaz_actor, atom("set_int"), 42);
send(foobaz_actor, atom("get_int")); send(foobaz_actor, atom("get_int"));
...@@ -593,25 +523,6 @@ size_t test__spawn() { ...@@ -593,25 +523,6 @@ size_t test__spawn() {
); );
await_all_others_done(); await_all_others_done();
CPPA_IF_VERBOSE(cout << "ok" << endl); CPPA_IF_VERBOSE(cout << "ok" << endl);
*/
{
bool invoked = false;
str_wrapper actor_facade_builder{"x"};
std::string actor_facade_builder_interim{"y"};
auto foo_actor = spawn(foobar, std::cref(actor_facade_builder), actor_facade_builder_interim);
send(foo_actor, atom("same"));
receive (
on(atom("yes")) >> [&]() {
CPPA_ERROR("x == y");
},
on(atom("no")) >> [&]() {
invoked = true;
}
);
CPPA_CHECK_EQUAL(true, invoked);
await_all_others_done();
}
CPPA_CHECK_EQUAL(behavior_test<testee_actor>(spawn(testee_actor{})), "wait4int"); CPPA_CHECK_EQUAL(behavior_test<testee_actor>(spawn(testee_actor{})), "wait4int");
CPPA_CHECK_EQUAL(behavior_test<event_testee>(spawn(new event_testee)), "wait4int"); CPPA_CHECK_EQUAL(behavior_test<event_testee>(spawn(new event_testee)), "wait4int");
......
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