Commit c91bf3fc authored by Dominik Charousset's avatar Dominik Charousset

API refactoring

parent 3eee072f
...@@ -138,6 +138,7 @@ set(LIBCPPA_SRC ...@@ -138,6 +138,7 @@ set(LIBCPPA_SRC
src/binary_serializer.cpp src/binary_serializer.cpp
src/broker.cpp src/broker.cpp
src/buffer.cpp src/buffer.cpp
src/blocking_untyped_actor.cpp
src/channel.cpp src/channel.cpp
src/common_actor_ops.cpp src/common_actor_ops.cpp
src/context_switching_resume.cpp src/context_switching_resume.cpp
...@@ -150,11 +151,12 @@ set(LIBCPPA_SRC ...@@ -150,11 +151,12 @@ set(LIBCPPA_SRC
src/deserializer.cpp src/deserializer.cpp
src/duration.cpp src/duration.cpp
src/empty_tuple.cpp src/empty_tuple.cpp
src/event_based_resume.cpp
src/exception.cpp src/exception.cpp
src/exit_reason.cpp src/exit_reason.cpp
src/fd_util.cpp src/fd_util.cpp
src/fiber.cpp src/fiber.cpp
src/functor_based_actor.cpp
src/functor_based_blocking_actor.cpp
src/get_root_uuid.cpp src/get_root_uuid.cpp
src/get_mac_addresses.cpp src/get_mac_addresses.cpp
src/group.cpp src/group.cpp
...@@ -178,11 +180,10 @@ set(LIBCPPA_SRC ...@@ -178,11 +180,10 @@ set(LIBCPPA_SRC
src/partial_function.cpp src/partial_function.cpp
src/primitive_variant.cpp src/primitive_variant.cpp
src/ref_counted.cpp src/ref_counted.cpp
src/resumable.cpp
src/remote_actor_proxy.cpp src/remote_actor_proxy.cpp
src/response_handle.cpp src/response_handle.cpp
src/ripemd_160.cpp src/ripemd_160.cpp
src/scheduled_actor.cpp
src/scheduled_actor_dummy.cpp
src/scheduler.cpp src/scheduler.cpp
src/scoped_actor.cpp src/scoped_actor.cpp
src/serializer.cpp src/serializer.cpp
...@@ -190,13 +191,13 @@ set(LIBCPPA_SRC ...@@ -190,13 +191,13 @@ set(LIBCPPA_SRC
src/singleton_manager.cpp src/singleton_manager.cpp
src/string_serialization.cpp src/string_serialization.cpp
src/sync_request_bouncer.cpp src/sync_request_bouncer.cpp
src/thread_mapped_actor.cpp
src/thread_pool_scheduler.cpp src/thread_pool_scheduler.cpp
src/to_uniform_name.cpp src/to_uniform_name.cpp
src/type_lookup_table.cpp src/type_lookup_table.cpp
src/unicast_network.cpp src/unicast_network.cpp
src/uniform_type_info.cpp src/uniform_type_info.cpp
src/uniform_type_info_map.cpp src/uniform_type_info_map.cpp
src/untyped_actor.cpp
src/weak_ptr_anchor.cpp src/weak_ptr_anchor.cpp
src/yield_interface.cpp) src/yield_interface.cpp)
......
...@@ -13,7 +13,6 @@ cppa/attachable.hpp ...@@ -13,7 +13,6 @@ cppa/attachable.hpp
cppa/behavior.hpp cppa/behavior.hpp
cppa/binary_deserializer.hpp cppa/binary_deserializer.hpp
cppa/binary_serializer.hpp cppa/binary_serializer.hpp
cppa/blocking_actor.hpp
cppa/blocking_untyped_actor.hpp cppa/blocking_untyped_actor.hpp
cppa/channel.hpp cppa/channel.hpp
cppa/common_actor_ops.hpp cppa/common_actor_ops.hpp
...@@ -37,7 +36,6 @@ cppa/detail/demangle.hpp ...@@ -37,7 +36,6 @@ cppa/detail/demangle.hpp
cppa/detail/disablable_delete.hpp cppa/detail/disablable_delete.hpp
cppa/detail/empty_tuple.hpp cppa/detail/empty_tuple.hpp
cppa/detail/fd_util.hpp cppa/detail/fd_util.hpp
cppa/detail/get_behavior.hpp
cppa/detail/group_manager.hpp cppa/detail/group_manager.hpp
cppa/detail/handle.hpp cppa/detail/handle.hpp
cppa/detail/ieee_754.hpp cppa/detail/ieee_754.hpp
...@@ -53,7 +51,6 @@ cppa/detail/projection.hpp ...@@ -53,7 +51,6 @@ cppa/detail/projection.hpp
cppa/detail/pseudo_tuple.hpp cppa/detail/pseudo_tuple.hpp
cppa/detail/ptype_to_type.hpp cppa/detail/ptype_to_type.hpp
cppa/detail/raw_access.hpp cppa/detail/raw_access.hpp
cppa/detail/scheduled_actor_dummy.hpp
cppa/detail/serialize_tuple.hpp cppa/detail/serialize_tuple.hpp
cppa/detail/singleton_manager.hpp cppa/detail/singleton_manager.hpp
cppa/detail/singleton_mixin.hpp cppa/detail/singleton_mixin.hpp
...@@ -105,7 +102,6 @@ cppa/io/remote_actor_proxy.hpp ...@@ -105,7 +102,6 @@ cppa/io/remote_actor_proxy.hpp
cppa/io/stream.hpp cppa/io/stream.hpp
cppa/local_actor.hpp cppa/local_actor.hpp
cppa/logging.hpp cppa/logging.hpp
cppa/mailbox_based.hpp
cppa/mailbox_element.hpp cppa/mailbox_element.hpp
cppa/match.hpp cppa/match.hpp
cppa/match_expr.hpp cppa/match_expr.hpp
...@@ -117,7 +113,6 @@ cppa/message_header.hpp ...@@ -117,7 +113,6 @@ cppa/message_header.hpp
cppa/message_id.hpp cppa/message_id.hpp
cppa/message_priority.hpp cppa/message_priority.hpp
cppa/node_id.hpp cppa/node_id.hpp
cppa/nonblocking_actor.hpp
cppa/none.hpp cppa/none.hpp
cppa/object.hpp cppa/object.hpp
cppa/on.hpp cppa/on.hpp
...@@ -133,7 +128,7 @@ cppa/opt.hpp ...@@ -133,7 +128,7 @@ cppa/opt.hpp
cppa/optional.hpp cppa/optional.hpp
cppa/optional_variant.hpp cppa/optional_variant.hpp
cppa/partial_function.hpp cppa/partial_function.hpp
cppa/policy/invoke_policy_base.hpp cppa/policy/invoke_policy.hpp
cppa/policy/nestable_invoke.hpp cppa/policy/nestable_invoke.hpp
cppa/policy/prioritizing.hpp cppa/policy/prioritizing.hpp
cppa/policy/sequential_invoke.hpp cppa/policy/sequential_invoke.hpp
...@@ -144,7 +139,6 @@ cppa/ref_counted.hpp ...@@ -144,7 +139,6 @@ cppa/ref_counted.hpp
cppa/replies_to.hpp cppa/replies_to.hpp
cppa/response_handle.hpp cppa/response_handle.hpp
cppa/sb_actor.hpp cppa/sb_actor.hpp
cppa/scheduled_actor.hpp
cppa/scheduler.hpp cppa/scheduler.hpp
cppa/scoped_actor.hpp cppa/scoped_actor.hpp
cppa/send.hpp cppa/send.hpp
...@@ -152,7 +146,6 @@ cppa/serializer.hpp ...@@ -152,7 +146,6 @@ cppa/serializer.hpp
cppa/singletons.hpp cppa/singletons.hpp
cppa/spawn.hpp cppa/spawn.hpp
cppa/spawn_options.hpp cppa/spawn_options.hpp
cppa/thread_mapped_actor.hpp
cppa/timeout_definition.hpp cppa/timeout_definition.hpp
cppa/to_string.hpp cppa/to_string.hpp
cppa/tpartial_function.hpp cppa/tpartial_function.hpp
...@@ -245,7 +238,6 @@ src/demangle.cpp ...@@ -245,7 +238,6 @@ src/demangle.cpp
src/deserializer.cpp src/deserializer.cpp
src/duration.cpp src/duration.cpp
src/empty_tuple.cpp src/empty_tuple.cpp
src/event_based_resume.cpp
src/exception.cpp src/exception.cpp
src/exit_reason.cpp src/exit_reason.cpp
src/factory.cpp src/factory.cpp
...@@ -286,8 +278,6 @@ src/ref_counted.cpp ...@@ -286,8 +278,6 @@ src/ref_counted.cpp
src/remote_actor_proxy.cpp src/remote_actor_proxy.cpp
src/response_handle.cpp src/response_handle.cpp
src/ripemd_160.cpp src/ripemd_160.cpp
src/scheduled_actor.cpp
src/scheduled_actor_dummy.cpp
src/scheduler.cpp src/scheduler.cpp
src/scoped_actor.cpp src/scoped_actor.cpp
src/send.cpp src/send.cpp
...@@ -296,7 +286,6 @@ src/shared_spinlock.cpp ...@@ -296,7 +286,6 @@ src/shared_spinlock.cpp
src/singleton_manager.cpp src/singleton_manager.cpp
src/string_serialization.cpp src/string_serialization.cpp
src/sync_request_bouncer.cpp src/sync_request_bouncer.cpp
src/thread_mapped_actor.cpp
src/thread_pool_scheduler.cpp src/thread_pool_scheduler.cpp
src/to_uniform_name.cpp src/to_uniform_name.cpp
src/type_lookup_table.cpp src/type_lookup_table.cpp
...@@ -338,3 +327,14 @@ cppa/policy/resume_policy.hpp ...@@ -338,3 +327,14 @@ cppa/policy/resume_policy.hpp
cppa/policy/cooperative_scheduling.hpp cppa/policy/cooperative_scheduling.hpp
cppa/policy/scheduling_policy.hpp cppa/policy/scheduling_policy.hpp
cppa/policy/priority_policy.hpp cppa/policy/priority_policy.hpp
cppa/policy/not_prioritizing.hpp
cppa/detail/functor_based_blocking_actor.hpp
src/functor_based_actor.cpp
cppa/mailbox_based.hpp
cppa/behavior_stack_based.hpp
cppa/resumable.hpp
src/resumable.cpp
cppa/policy/middleman_scheduling.hpp
src/untyped_actor.cpp
src/functor_based_blocking_actor.cpp
src/blocking_untyped_actor.cpp
...@@ -143,6 +143,12 @@ class abstract_actor : public abstract_channel { ...@@ -143,6 +143,12 @@ class abstract_actor : public abstract_channel {
*/ */
inline const node_id& node() const; inline const node_id& node() const;
/**
* @brief Returns the actor's exit reason of
* <tt>exit_reason::not_exited</tt> if it's still alive.
*/
inline std::uint32_t exit_reason() const;
protected: protected:
abstract_actor(); abstract_actor();
...@@ -166,12 +172,6 @@ class abstract_actor : public abstract_channel { ...@@ -166,12 +172,6 @@ class abstract_actor : public abstract_channel {
*/ */
bool unlink_from_impl(const actor_addr& other); bool unlink_from_impl(const actor_addr& other);
/**
* @brief Returns the actor's exit reason of
* <tt>exit_reason::not_exited</tt> if it's still alive.
*/
inline std::uint32_t exit_reason() const;
/** /**
* @brief Returns <tt>exit_reason() != exit_reason::not_exited</tt>. * @brief Returns <tt>exit_reason() != exit_reason::not_exited</tt>.
*/ */
......
...@@ -45,6 +45,10 @@ namespace cppa { ...@@ -45,6 +45,10 @@ namespace cppa {
namespace detail { class raw_access; } namespace detail { class raw_access; }
struct invalid_actor_t { constexpr invalid_actor_t() { } };
constexpr invalid_actor_t invalid_actor = invalid_actor_t{};
/** /**
* @brief Identifies an untyped actor. * @brief Identifies an untyped actor.
*/ */
...@@ -65,6 +69,8 @@ class actor : util::comparable<actor> { ...@@ -65,6 +69,8 @@ class actor : util::comparable<actor> {
inline bool operator!() const; inline bool operator!() const;
actor(const invalid_actor_t&);
void enqueue(const message_header& hdr, any_tuple msg) const; void enqueue(const message_header& hdr, any_tuple msg) const;
inline common_actor_ops* operator->() const { inline common_actor_ops* operator->() const {
...@@ -89,7 +95,7 @@ inline actor::operator bool() const { ...@@ -89,7 +95,7 @@ inline actor::operator bool() const {
inline bool actor::operator!() const { inline bool actor::operator!() const {
return !static_cast<bool>(*this); return !static_cast<bool>(*this);
} }
} // namespace cppa } // namespace cppa
#endif // CPPA_ACTOR_HPP #endif // CPPA_ACTOR_HPP
...@@ -25,7 +25,7 @@ ...@@ -25,7 +25,7 @@
* * * *
* You should have received a copy of the GNU Lesser General Public License * * You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. * * along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/ \******************************************************************************/
#ifndef CPPA_ACTOR_ADDR_HPP #ifndef CPPA_ACTOR_ADDR_HPP
......
...@@ -25,7 +25,7 @@ ...@@ -25,7 +25,7 @@
* * * *
* You should have received a copy of the GNU Lesser General Public License * * You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. * * along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/ \******************************************************************************/
#ifndef CPPA_ACTOR_NAMESPACE_HPP #ifndef CPPA_ACTOR_NAMESPACE_HPP
......
...@@ -28,12 +28,9 @@ ...@@ -28,12 +28,9 @@
\******************************************************************************/ \******************************************************************************/
#ifndef CPPA_STACKLESS_HPP #ifndef CPPA_BEHAVIOR_STACK_BASED_HPP
#define CPPA_STACKLESS_HPP #define CPPA_BEHAVIOR_STACK_BASED_HPP
#include "cppa/util/dptr.hpp"
#include "cppa/detail/receive_policy.hpp"
#include "cppa/detail/behavior_stack.hpp" #include "cppa/detail/behavior_stack.hpp"
namespace cppa { namespace cppa {
...@@ -41,14 +38,14 @@ namespace cppa { ...@@ -41,14 +38,14 @@ namespace cppa {
#ifdef CPPA_DOCUMENTATION #ifdef CPPA_DOCUMENTATION
/** /**
* @brief Policy tag that causes {@link event_based_actor::become} to * @brief Policy tag that causes {@link untyped_actor::become} to
* discard the current behavior. * discard the current behavior.
* @relates local_actor * @relates local_actor
*/ */
constexpr auto discard_behavior; constexpr auto discard_behavior;
/** /**
* @brief Policy tag that causes {@link event_based_actor::become} to * @brief Policy tag that causes {@link untyped_actor::become} to
* keep the current behavior available. * keep the current behavior available.
* @relates local_actor * @relates local_actor
*/ */
...@@ -75,25 +72,14 @@ constexpr keep_behavior_t keep_behavior = keep_behavior_t{}; ...@@ -75,25 +72,14 @@ constexpr keep_behavior_t keep_behavior = keep_behavior_t{};
#endif #endif
template<class Base, class Subtype> template<class Base, class Subtype>
class stackless : public Base { class behavior_stack_based : public Base {
protected:
typedef stackless combined_type;
public: public:
template<typename... Ts> typedef behavior_stack_based combined_type;
stackless(Ts&&... args) : Base(std::forward<Ts>(args)...) { }
static constexpr auto receive_flag = detail::rp_sequential; inline void unbecome() {
m_bhvr_stack.pop_async_back();
bool has_behavior() {
return this->m_bhvr_stack.empty() == false;
}
void unbecome() {
this->m_bhvr_stack.pop_async_back();
} }
/** /**
...@@ -110,49 +96,35 @@ class stackless : public Base { ...@@ -110,49 +96,35 @@ class stackless : public Base {
std::forward<Ts>(args)...), std::forward<Ts>(args)...),
true); true);
} }
template<bool Discard, typename... Ts> template<bool Discard, typename... Ts>
inline void become(behavior_policy<Discard>, Ts&&... args) { inline void become(behavior_policy<Discard>, Ts&&... args) {
do_become(match_expr_convert(std::forward<Ts>(args)...), Discard); do_become(match_expr_convert(std::forward<Ts>(args)...), Discard);
} }
void become_waiting_for(behavior bhvr, message_id mf) { void become_waiting_for(behavior bhvr, message_id mf) {
if (bhvr.timeout().valid()) { if (bhvr.timeout().valid()) {
this->reset_timeout(); //FIXME
this->request_timeout(bhvr.timeout());
} }
this->m_bhvr_stack.push_back(std::move(bhvr), mf); m_bhvr_stack.push_back(std::move(bhvr), mf);
} }
void do_become(behavior&& bhvr, bool discard_old) { void do_become(behavior&& bhvr, bool discard_old) {
this->reset_timeout(); if (discard_old) m_bhvr_stack.pop_async_back();
this->request_timeout(bhvr.timeout()); m_bhvr_stack.push_back(std::move(bhvr));
if (discard_old) this->m_bhvr_stack.pop_async_back();
this->m_bhvr_stack.push_back(std::move(bhvr));
} }
inline bool has_behavior() const { inline bool has_behavior() const {
return this->m_bhvr_stack.empty() == false; return m_bhvr_stack.empty() == false;
} }
inline behavior& get_behavior() { inline behavior& get_behavior() {
CPPA_REQUIRE(this->m_bhvr_stack.empty() == false); CPPA_REQUIRE(m_bhvr_stack.empty() == false);
return this->m_bhvr_stack.back(); return m_bhvr_stack.back();
} }
inline void handle_timeout(behavior& bhvr) { inline void handle_timeout(behavior& bhvr) {
CPPA_REQUIRE(bhvr.timeout().valid());
this->reset_timeout();
bhvr.handle_timeout(); bhvr.handle_timeout();
if (this->m_bhvr_stack.empty() == false) {
this->request_timeout(get_behavior().timeout());
}
}
void exec_bhvr_stack() {
while (!m_bhvr_stack.empty()) {
m_bhvr_stack.exec(m_recv_policy, util::dptr<Subtype>(this));
}
} }
inline detail::behavior_stack& bhvr_stack() { inline detail::behavior_stack& bhvr_stack() {
...@@ -168,11 +140,8 @@ class stackless : public Base { ...@@ -168,11 +140,8 @@ class stackless : public Base {
// allows actors to keep previous behaviors and enables unbecome() // allows actors to keep previous behaviors and enables unbecome()
detail::behavior_stack m_bhvr_stack; detail::behavior_stack m_bhvr_stack;
// used for message handling in subclasses
detail::receive_policy m_recv_policy;
}; };
} // namespace cppa } // namespace cppa
#endif // CPPA_STACKLESS_HPP #endif // CPPA_BEHAVIOR_STACK_BASED_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* 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 2.1 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 <chrono>
#include <memory>
#include <functional>
#include "cppa/behavior.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/mailbox_element.hpp"
namespace cppa {
class blocking_actor : public local_actor {
friend class detail::receive_policy;
friend class detail::behavior_stack;
protected:
typedef stacked combined_type;
public:
static constexpr auto receive_flag = detail::rp_nestable;
struct receive_while_helper {
std::function<void(behavior&)> m_dq;
std::function<bool()> m_stmt;
template<typename... Ts>
void operator()(Ts&&... args) {
static_assert(sizeof...(Ts) > 0,
"operator() requires at least one argument");
behavior bhvr = match_expr_convert(std::forward<Ts>(args)...);
while (m_stmt()) m_dq(bhvr);
}
};
template<typename T>
struct receive_for_helper {
std::function<void(behavior&)> m_dq;
T& begin;
T end;
template<typename... Ts>
void operator()(Ts&&... args) {
behavior bhvr = match_expr_convert(std::forward<Ts>(args)...);
for ( ; begin != end; ++begin) m_dq(bhvr);
}
};
struct do_receive_helper {
std::function<void(behavior&)> m_dq;
behavior m_bhvr;
template<typename Statement>
void until(Statement stmt) {
do { m_dq(m_bhvr); }
while (stmt() == false);
}
};
/**
* @brief Dequeues the next message from the mailbox that
* is matched by given behavior.
*/
template<typename... Ts>
void receive(Ts&&... args) {
static_assert(sizeof...(Ts), "receive() requires at least one argument");
dequeue(match_expr_convert(std::forward<Ts>(args)...));
}
/**
* @brief Receives messages in an endless loop.
* Semantically equal to: <tt>for (;;) { receive(...); }</tt>
*/
template<typename... Ts>
void receive_loop(Ts&&... args) {
behavior bhvr = match_expr_convert(std::forward<Ts>(args)...);
for (;;) dequeue(bhvr);
}
/**
* @brief Receives messages as in a range-based loop.
*
* Semantically equal to:
* <tt>for ( ; begin != end; ++begin) { receive(...); }</tt>.
*
* <b>Usage example:</b>
* @code
* int i = 0;
* receive_for(i, 10) (
* on(atom("get")) >> [&]() -> any_tuple { return {"result", i}; }
* );
* @endcode
* @param begin First value in range.
* @param end Last value in range (excluded).
* @returns A functor implementing the loop.
*/
template<typename T>
receive_for_helper<T> receive_for(T& begin, const T& end) {
return {make_dequeue_callback(), begin, end};
}
/**
* @brief Receives messages as long as @p stmt returns true.
*
* Semantically equal to: <tt>while (stmt()) { receive(...); }</tt>.
*
* <b>Usage example:</b>
* @code
* int i = 0;
* receive_while([&]() { return (++i <= 10); })
* (
* on<int>() >> int_fun,
* on<float>() >> float_fun
* );
* @endcode
* @param stmt Lambda expression, functor or function returning a @c bool.
* @returns A functor implementing the loop.
*/
template<typename Statement>
receive_while_helper receive_while(Statement stmt) {
static_assert(std::is_same<bool, decltype(stmt())>::value,
"functor or function does not return a boolean");
return {make_dequeue_callback(), stmt};
}
/**
* @brief Receives messages until @p stmt returns true.
*
* Semantically equal to: <tt>do { receive(...); } while (stmt() == false);</tt>
*
* <b>Usage example:</b>
* @code
* int i = 0;
* do_receive
* (
* on<int>() >> int_fun,
* on<float>() >> float_fun
* )
* .until([&]() { return (++i >= 10); };
* @endcode
* @param bhvr Denotes the actor's response the next incoming message.
* @returns A functor providing the @c until member function.
*/
template<typename... Ts>
do_receive_helper do_receive(Ts&&... args) {
return { make_dequeue_callback()
, match_expr_convert(std::forward<Ts>(args)...)};
}
virtual void run() {
if (m_behavior) m_behavior();
auto dthis = util::dptr<Subtype>(this);
auto rsn = dthis->planned_exit_reason();
dthis->cleanup(rsn == exit_reason::not_exited ? exit_reason::normal : rsn);
}
inline void set_behavior(std::function<void()> fun) {
m_behavior = std::move(fun);
}
inline optional<behavior&> sync_handler(message_id msg_id) {
auto i = m_sync_handler.find(msg_id);
if (i != m_sync_handler.end()) return i->second;
return none;
}
protected:
template<typename... Ts>
stacked(Ts&&... args) : Base(std::forward<Ts>(args)...) { }
virtual bool has_behavior() {
return static_cast<bool>(m_behavior);
}
std::function<void()> m_behavior;
detail::receive_policy m_recv_policy;
private:
std::map<message_id, behavior> m_sync_handler;
std::function<void(behavior&)> make_dequeue_callback() {
return [=](behavior& bhvr) { dequeue(bhvr); };
}
void dequeue(behavior& bhvr) {
m_recv_policy.receive(util::dptr<Subtype>(this), bhvr);
}
void dequeue_response(behavior& bhvr, message_id request_id) {
m_recv_policy.receive(util::dptr<Subtype>(this), bhvr, request_id);
}
};
} // namespace cppa
#endif // CPPA_STACKED_ACTOR_MIXIN_HPP
...@@ -25,23 +25,210 @@ ...@@ -25,23 +25,210 @@
* * * *
* You should have received a copy of the GNU Lesser General Public License * * You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. * * along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/ \******************************************************************************/
#ifndef CPPA_BLOCKING_UNTYPED_ACTOR_HPP #ifndef CPPA_BLOCKING_UNTYPED_ACTOR_HPP
#define CPPA_BLOCKING_UNTYPED_ACTOR_HPP #define CPPA_BLOCKING_UNTYPED_ACTOR_HPP
#include "cppa/extend.hpp" #include "cppa/extend.hpp"
#include "cppa/stacked.hpp"
#include "cppa/local_actor.hpp" #include "cppa/local_actor.hpp"
#include "cppa/mailbox_based.hpp"
#include "cppa/mailbox_element.hpp"
namespace cppa { namespace cppa {
/** /**
* @extends local_actor * @extends local_actor
*/ */
class blocking_untyped_actor : public extend<local_actor>::with<stacked> { class blocking_untyped_actor : public extend<local_actor>::with<mailbox_based> {
public:
typedef std::chrono::high_resolution_clock::time_point timeout_type;
struct receive_while_helper {
std::function<void(behavior&)> m_dq;
std::function<bool()> m_stmt;
template<typename... Ts>
void operator()(Ts&&... args) {
static_assert(sizeof...(Ts) > 0,
"operator() requires at least one argument");
behavior bhvr = match_expr_convert(std::forward<Ts>(args)...);
while (m_stmt()) m_dq(bhvr);
}
};
template<typename T>
struct receive_for_helper {
std::function<void(behavior&)> m_dq;
T& begin;
T end;
template<typename... Ts>
void operator()(Ts&&... args) {
behavior bhvr = match_expr_convert(std::forward<Ts>(args)...);
for ( ; begin != end; ++begin) m_dq(bhvr);
}
};
struct do_receive_helper {
std::function<void(behavior&)> m_dq;
behavior m_bhvr;
template<typename Statement>
void until(Statement stmt) {
do { m_dq(m_bhvr); }
while (stmt() == false);
}
};
/**
* @brief Dequeues the next message from the mailbox that
* is matched by given behavior.
*/
template<typename... Ts>
void receive(Ts&&... args) {
static_assert(sizeof...(Ts), "receive() requires at least one argument");
dequeue(match_expr_convert(std::forward<Ts>(args)...));
}
/**
* @brief Receives messages in an endless loop.
* Semantically equal to: <tt>for (;;) { receive(...); }</tt>
*/
template<typename... Ts>
void receive_loop(Ts&&... args) {
behavior bhvr = match_expr_convert(std::forward<Ts>(args)...);
for (;;) dequeue(bhvr);
}
/**
* @brief Receives messages as in a range-based loop.
*
* Semantically equal to:
* <tt>for ( ; begin != end; ++begin) { receive(...); }</tt>.
*
* <b>Usage example:</b>
* @code
* int i = 0;
* receive_for(i, 10) (
* on(atom("get")) >> [&]() -> any_tuple { return {"result", i}; }
* );
* @endcode
* @param begin First value in range.
* @param end Last value in range (excluded).
* @returns A functor implementing the loop.
*/
template<typename T>
receive_for_helper<T> receive_for(T& begin, const T& end) {
return {make_dequeue_callback(), begin, end};
}
/**
* @brief Receives messages as long as @p stmt returns true.
*
* Semantically equal to: <tt>while (stmt()) { receive(...); }</tt>.
*
* <b>Usage example:</b>
* @code
* int i = 0;
* receive_while([&]() { return (++i <= 10); })
* (
* on<int>() >> int_fun,
* on<float>() >> float_fun
* );
* @endcode
* @param stmt Lambda expression, functor or function returning a @c bool.
* @returns A functor implementing the loop.
*/
template<typename Statement>
receive_while_helper receive_while(Statement stmt) {
static_assert(std::is_same<bool, decltype(stmt())>::value,
"functor or function does not return a boolean");
return {make_dequeue_callback(), stmt};
}
/**
* @brief Handles a synchronous response message in an event-based way.
* @param handle A future for a synchronous response.
* @throws std::logic_error if @p handle is not valid or if the actor
* already received the response for @p handle
* @relates message_future
*/
inline sync_receive_helper receive_response(const message_future& f) {
return {f};
}
/**
* @brief Receives messages until @p stmt returns true.
*
* Semantically equal to: <tt>do { receive(...); } while (stmt() == false);</tt>
*
* <b>Usage example:</b>
* @code
* int i = 0;
* do_receive
* (
* on<int>() >> int_fun,
* on<float>() >> float_fun
* )
* .until([&]() { return (++i >= 10); };
* @endcode
* @param bhvr Denotes the actor's response the next incoming message.
* @returns A functor providing the @c until member function.
*/
template<typename... Ts>
do_receive_helper do_receive(Ts&&... args) {
return { make_dequeue_callback()
, match_expr_convert(std::forward<Ts>(args)...)};
}
inline optional<behavior&> sync_handler(message_id msg_id) {
auto i = m_sync_handler.find(msg_id);
if (i != m_sync_handler.end()) return i->second;
return none;
}
inline void dequeue(behavior&& bhvr) {
behavior tmp{std::move(bhvr)};
dequeue(tmp);
}
virtual void dequeue(behavior& bhvr) = 0;
virtual mailbox_element* dequeue() = 0;
virtual mailbox_element* try_dequeue() = 0;
virtual mailbox_element* try_dequeue(const timeout_type&) = 0;
void await_all_other_actors_done();
virtual void act() = 0;
private:
std::map<message_id, behavior> m_sync_handler;
std::function<void(behavior&)> make_dequeue_callback() {
return [=](behavior& bhvr) { dequeue(bhvr); };
}
/*
void dequeue_response(behavior& bhvr, message_id request_id) {
m_recv_policy.receive(util::dptr<Subtype>(this), bhvr, request_id);
}
*/
}; };
} // namespace cppa } // namespace cppa
......
...@@ -25,7 +25,7 @@ ...@@ -25,7 +25,7 @@
* * * *
* You should have received a copy of the GNU Lesser General Public License * * You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. * * along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/ \******************************************************************************/
#ifndef CPPA_CHANNEL_HPP #ifndef CPPA_CHANNEL_HPP
......
...@@ -46,7 +46,6 @@ ...@@ -46,7 +46,6 @@
#include "cppa/behavior.hpp" #include "cppa/behavior.hpp"
#include "cppa/announce.hpp" #include "cppa/announce.hpp"
#include "cppa/sb_actor.hpp" #include "cppa/sb_actor.hpp"
#include "cppa/threaded.hpp"
#include "cppa/scheduler.hpp" #include "cppa/scheduler.hpp"
#include "cppa/to_string.hpp" #include "cppa/to_string.hpp"
#include "cppa/any_tuple.hpp" #include "cppa/any_tuple.hpp"
...@@ -56,14 +55,12 @@ ...@@ -56,14 +55,12 @@
#include "cppa/typed_actor.hpp" #include "cppa/typed_actor.hpp"
#include "cppa/exit_reason.hpp" #include "cppa/exit_reason.hpp"
#include "cppa/local_actor.hpp" #include "cppa/local_actor.hpp"
#include "cppa/prioritizing.hpp" #include "cppa/scoped_actor.hpp"
#include "cppa/spawn_options.hpp" #include "cppa/spawn_options.hpp"
#include "cppa/untyped_actor.hpp" #include "cppa/untyped_actor.hpp"
#include "cppa/abstract_actor.hpp" #include "cppa/abstract_actor.hpp"
#include "cppa/message_future.hpp" #include "cppa/message_future.hpp"
#include "cppa/response_handle.hpp" #include "cppa/response_handle.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/event_based_actor.hpp"
#include "cppa/blocking_untyped_actor.hpp" #include "cppa/blocking_untyped_actor.hpp"
#include "cppa/util/type_traits.hpp" #include "cppa/util/type_traits.hpp"
...@@ -79,7 +76,6 @@ ...@@ -79,7 +76,6 @@
#include "cppa/io/connection_handle.hpp" #include "cppa/io/connection_handle.hpp"
#include "cppa/detail/memory.hpp" #include "cppa/detail/memory.hpp"
#include "cppa/detail/get_behavior.hpp"
#include "cppa/detail/actor_registry.hpp" #include "cppa/detail/actor_registry.hpp"
/** /**
...@@ -436,8 +432,7 @@ ...@@ -436,8 +432,7 @@
namespace cppa { namespace cppa {
template<typename... Ts> inline void send_tuple_as(const actor& from, const channel& to, any_tuple msg) {
void send_tuple_as(const actor& from, const channel& to, any_tuple msg) {
to.enqueue({from->address(), to}, std::move(msg)); to.enqueue({from->address(), to}, std::move(msg));
} }
...@@ -450,6 +445,23 @@ template<typename... Ts> ...@@ -450,6 +445,23 @@ template<typename... Ts>
void send_as(const actor& from, const channel& to, Ts&&... args) { void send_as(const actor& from, const channel& to, Ts&&... args) {
send_tuple_as(from, to, make_any_tuple(std::forward<Ts>(args)...)); send_tuple_as(from, to, make_any_tuple(std::forward<Ts>(args)...));
} }
inline void anon_send_tuple(const channel& to, any_tuple msg) {
to.enqueue({invalid_actor_addr, to}, std::move(msg));
}
/**
* @brief Anonymously sends a message to @p receiver;
*/
template<typename... Ts>
inline void anon_send(const channel& receiver, Ts&&... args) {
anon_send_tuple(receiver, make_any_tuple(std::forward<Ts>(args)...));
}
inline void anon_send_exit(const actor_addr&, std::uint32_t) {
}
/** /**
* @brief Blocks execution of this actor until all * @brief Blocks execution of this actor until all
* other actors finished execution. * other actors finished execution.
...@@ -517,10 +529,16 @@ template<class Impl, spawn_options Options = no_spawn_options, typename... Ts> ...@@ -517,10 +529,16 @@ template<class Impl, spawn_options Options = no_spawn_options, typename... Ts>
actor spawn_io(io::input_stream_ptr in, actor spawn_io(io::input_stream_ptr in,
io::output_stream_ptr out, io::output_stream_ptr out,
Ts&&... args) { Ts&&... args) {
using namespace io; using namespace policy;
using namespace std; using proper_impl = detail::proper_actor<Impl,
auto ptr = make_counted<Impl>(move(in), move(out), forward<Ts>(args)...); middleman_scheduling,
return eval_sopts(Options, io::init_and_launch(move(ptr))); not_prioritizing,
no_resume,
cooperative_scheduling>;
auto ptr = make_counted<proper_impl>(std::move(in), std::move(out),
std::forward<Ts>(args)...);
ptr->launch();
return ptr;
} }
/** /**
...@@ -530,16 +548,12 @@ actor spawn_io(io::input_stream_ptr in, ...@@ -530,16 +548,12 @@ actor spawn_io(io::input_stream_ptr in,
* @returns An {@link actor_ptr} to the spawned {@link actor}. * @returns An {@link actor_ptr} to the spawned {@link actor}.
*/ */
template<spawn_options Options = no_spawn_options, template<spawn_options Options = no_spawn_options,
typename F = std::function<void (io::broker*)>, typename F = std::function<void (io::broker*)>>
typename... Ts>
actor spawn_io(F fun, actor spawn_io(F fun,
io::input_stream_ptr in, io::input_stream_ptr in,
io::output_stream_ptr out, io::output_stream_ptr out) {
Ts&&... args) { return spawn_io<io::default_broker>(std::move(fun), std::move(in),
using namespace std; std::move(out));
auto ptr = io::broker::from(move(fun), move(in), move(out),
forward<Ts>(args)...);
return eval_sopts(Options, io::init_and_launch(move(ptr)));
} }
/* /*
...@@ -551,21 +565,17 @@ actor_ptr spawn_io(const char* host, uint16_t port, Ts&&... args) { ...@@ -551,21 +565,17 @@ actor_ptr spawn_io(const char* host, uint16_t port, Ts&&... args) {
*/ */
template<spawn_options Options = no_spawn_options, template<spawn_options Options = no_spawn_options,
typename F = std::function<void (io::broker*)>, typename F = std::function<void (io::broker*)>>
typename... Ts> actor spawn_io(F fun, const std::string& host, uint16_t port) {
actor spawn_io(F fun, const std::string& host, uint16_t port, Ts&&... args) {
auto ptr = io::ipv4_io_stream::connect_to(host.c_str(), port); auto ptr = io::ipv4_io_stream::connect_to(host.c_str(), port);
return spawn_io(std::move(fun), ptr, ptr, std::forward<Ts>(args)...); return spawn_io(std::move(fun), ptr, ptr);
} }
template<spawn_options Options = no_spawn_options, template<spawn_options Options = no_spawn_options,
typename F = std::function<void (io::broker*)>, typename F = std::function<void (io::broker*)>>
typename... Ts> actor spawn_io_server(F fun, uint16_t port) {
actor spawn_io_server(F fun, uint16_t port, Ts&&... args) {
using namespace std; using namespace std;
auto ptr = io::broker::from(move(fun), io::ipv4_acceptor::create(port), return spawn_io(move(fun), io::ipv4_acceptor::create(port));
forward<Ts>(args)...);
return eval_sopts(Options, io::init_and_launch(move(ptr)));
} }
/** /**
......
...@@ -31,13 +31,100 @@ ...@@ -31,13 +31,100 @@
#ifndef CPPA_FUNCTOR_BASED_ACTOR_HPP #ifndef CPPA_FUNCTOR_BASED_ACTOR_HPP
#define CPPA_FUNCTOR_BASED_ACTOR_HPP #define CPPA_FUNCTOR_BASED_ACTOR_HPP
#include <type_traits>
#include "cppa/untyped_actor.hpp" #include "cppa/untyped_actor.hpp"
namespace cppa { namespace detail { namespace cppa { namespace detail {
class functor_based_actor : public untyped_actor { class functor_based_actor : public untyped_actor {
public:
typedef std::function<behavior(untyped_actor*)> make_behavior_fun;
typedef std::function<void(untyped_actor*)> void_fun;
template<typename F, typename... Ts>
functor_based_actor(F f, Ts&&... vs) {
untyped_actor* dummy = nullptr;
create(dummy, f, std::forward<Ts>(vs)...);
}
behavior make_behavior() override;
private:
void create(untyped_actor*, void_fun);
template<class Actor, typename F, typename... Ts>
auto create(Actor*, F f, Ts&&... vs) ->
typename std::enable_if<
std::is_convertible<
decltype(f(std::forward<Ts>(vs)...)),
behavior
>::value
>::type {
auto fun = std::bind(f, std::forward<Ts>(vs)...);
m_make_behavior = [fun](Actor*) -> behavior { return fun(); };
}
template<class Actor, typename F, typename... Ts>
auto create(Actor* dummy, F f, Ts&&... vs) ->
typename std::enable_if<
std::is_convertible<
decltype(f(dummy, std::forward<Ts>(vs)...)),
behavior
>::value
>::type {
auto fun = std::bind(f, std::placeholders::_1, std::forward<Ts>(vs)...);
m_make_behavior = [fun](Actor* self) -> behavior { return fun(self); };
}
template<class Actor, typename F, typename... Ts>
auto create(Actor*, F f, Ts&&... vs) ->
typename std::enable_if<
std::is_same<
decltype(f(std::forward<Ts>(vs)...)),
void
>::value
>::type {
std::function<void()> fun = std::bind(f, std::forward<Ts>(vs)...);
m_make_behavior = [fun](Actor*) -> behavior {
fun();
return behavior{};
};
}
template<class Actor, typename F, typename... Ts>
auto create(Actor* dummy, F f, Ts&&... vs) ->
typename std::enable_if<
std::is_same<
decltype(f(dummy, std::forward<Ts>(vs)...)),
void
>::value
>::type {
std::function<void(Actor*)> fun = std::bind(f, std::placeholders::_1,
std::forward<Ts>(vs)...);
m_make_behavior = [fun](Actor* self) -> behavior {
fun(self);
return behavior{};
};
}
/*
template<class Actor, typename F, typename T0, typename... Ts>
void create(void*, F, T0&&, Ts&&...) {
// this overload acts as 'catch-all' rule to
// give the user a clear hint at what's wrong
// with the provided functor
static_assert(sizeof...(Ts) != sizeof...(Ts),
"spawn: provided functor must either return 'void'' "
"or 'cppa::behavior'");
}
*/
make_behavior_fun m_make_behavior;
}; };
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* 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 2.1 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 FUNCTOR_BASED_BLOCKING_ACTOR_HPP
#define FUNCTOR_BASED_BLOCKING_ACTOR_HPP
#include "cppa/blocking_untyped_actor.hpp"
namespace cppa {
namespace detail {
class functor_based_blocking_actor : public blocking_untyped_actor {
public:
typedef std::function<void (blocking_untyped_actor*)> act_fun;
template<typename F, typename... Ts>
functor_based_blocking_actor(F f, Ts&&... vs) {
blocking_untyped_actor* dummy = nullptr;
create(dummy, f, std::forward<Ts>(vs)...);
}
protected:
void act() override;
private:
void create(blocking_untyped_actor*, act_fun);
template<class Actor, typename F, typename T0, typename... Ts>
auto create(Actor* dummy, F f, T0&& v0, Ts&&... vs) ->
typename std::enable_if<
std::is_same<
decltype(f(dummy, std::forward<T0>(v0), std::forward<Ts>(vs)...)),
void
>::value
>::type {
create(dummy, std::bind(f, std::placeholders::_1,
std::forward<T0>(v0), std::forward<Ts>(vs)...));
}
template<class Actor, typename F, typename T0, typename... Ts>
auto create(Actor* dummy, F f, T0&& v0, Ts&&... vs) ->
typename std::enable_if<
std::is_same<
decltype(f(std::forward<T0>(v0), std::forward<Ts>(vs)...)),
void
>::value
>::type {
std::function<void()> fun = std::bind(f, std::forward<T0>(v0),
std::forward<Ts>(vs)...);
create(dummy, [fun](Actor*) { fun(); });
}
act_fun m_act;
};
} // namespace detail
} // namespace cppa
#endif // FUNCTOR_BASED_BLOCKING_ACTOR_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* 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 2.1 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_GET_BEHAVIOR_HPP
#define CPPA_GET_BEHAVIOR_HPP
#include <type_traits>
#include "cppa/util/call.hpp"
#include "cppa/util/int_list.hpp"
#include "cppa/util/type_traits.hpp"
#include "cppa/detail/tdata.hpp"
#include "cppa/scheduled_actor.hpp"
namespace cppa { namespace detail {
// default: <true, false, F>
template<bool IsFunctionPtr, bool HasArguments, typename F, typename... Ts>
class ftor_behavior : public scheduled_actor {
F m_fun;
public:
ftor_behavior(F ptr) : m_fun(ptr) { }
virtual void act() { m_fun(); }
};
template<typename F, typename... Ts>
class ftor_behavior<true, true, F, Ts...> : public scheduled_actor {
static_assert(sizeof...(Ts) > 0, "sizeof...(Ts) == 0");
F m_fun;
typedef typename tdata_from_type_list<
typename util::tl_map<util::type_list<Ts...>,
implicit_conversions>::type>::type
tdata_type;
tdata_type m_args;
public:
ftor_behavior(F ptr, const Ts&... args) : m_fun(ptr), m_args(args...) { }
virtual void act() {
util::apply_args(m_fun, m_args, util::get_indices(m_args));
}
};
template<typename F>
class ftor_behavior<false, false, F> : public scheduled_actor {
F m_fun;
public:
ftor_behavior(const F& arg) : m_fun(arg) { }
ftor_behavior(F&& arg) : m_fun(std::move(arg)) { }
virtual void act() { m_fun(); }
};
template<typename F, typename... Ts>
class ftor_behavior<false, true, F, Ts...> : public scheduled_actor {
static_assert(sizeof...(Ts) > 0, "sizeof...(Ts) == 0");
F m_fun;
typedef typename tdata_from_type_list<
typename util::tl_map<util::type_list<Ts...>,
implicit_conversions>::type>::type
tdata_type;
tdata_type m_args;
public:
ftor_behavior(const F& f, const Ts&... args) : m_fun(f), m_args(args...) {
}
ftor_behavior(F&& f, const Ts&... args)
: m_fun(std::move(f)), m_args(args...) { }
virtual void act() {
util::apply_args(m_fun, m_args, util::get_indices(m_args));
}
};
template<typename R>
scheduled_actor* get_behavior(std::integral_constant<bool, true>, R (*fptr)()) {
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?");
return new ftor_behavior<true, false, R (*)()>(fptr);
}
template<typename F>
scheduled_actor* get_behavior(std::integral_constant<bool, false>, F&& ftor) {
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?");
typedef typename util::rm_const_and_ref<F>::type ftype;
return new ftor_behavior<false, false, ftype>(std::forward<F>(ftor));
}
template<typename F, typename T, typename... Ts>
scheduled_actor* get_behavior(std::true_type, F fptr, const T& arg, const Ts&... args) {
static_assert(std::is_convertible<decltype(fptr(arg, 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?");
typedef ftor_behavior<true, true, F, T, Ts...> impl;
return new impl(fptr, arg, args...);
}
template<typename F, typename T, typename... Ts>
scheduled_actor* get_behavior(std::false_type, F ftor, const T& arg, const Ts&... args) {
static_assert(std::is_convertible<decltype(ftor(arg, 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?");
typedef typename util::rm_const_and_ref<F>::type ftype;
typedef ftor_behavior<false, true, ftype, T, Ts...> impl;
return new impl(std::forward<F>(ftor), arg, args...);
}
} } // namespace cppa::detail
#endif // CPPA_GET_BEHAVIOR_HPP
...@@ -3,37 +3,46 @@ ...@@ -3,37 +3,46 @@
#include <type_traits> #include <type_traits>
#include "cppa/blocking_actor.hpp" #include "cppa/resumable.hpp"
#include "cppa/mailbox_element.hpp" #include "cppa/mailbox_element.hpp"
#include "cppa/blocking_untyped_actor.hpp"
namespace cppa { namespace util { class fiber; } } #include "cppa/util/duration.hpp"
namespace cppa { namespace detail { namespace cppa {
namespace util {
struct fiber;
} // namespace util
} // namespace cppa
namespace cppa {
namespace detail {
template<class Base, template<class Base,
class SchedulingPolicy, class SchedulingPolicy,
class PriorityPolicy, class PriorityPolicy,
class ResumePolicy, class ResumePolicy,
class InvokePolicy, class InvokePolicy,
bool OverrideDequeue = std::is_base_of<blocking_actor, Base>::value> bool OverrideDequeue = std::is_base_of<blocking_untyped_actor, Base>::value>
class proper_actor : public Base { class proper_actor : public ResumePolicy::template mixin<Base> {
friend SchedulingPolicy;
friend PriorityPolicy;
friend ResumePolicy;
friend InvokePolicy;
typedef typename ResumePolicy::template mixin<Base> super;
public: public:
template<typename... Ts> template <typename... Ts>
proper_actor(Ts&&... args) : Base(std::forward<Ts>(args)) { } proper_actor(Ts&&... args) : super(std::forward<Ts>(args)...) { }
void enqueue(const message_header& hdr, any_tuple msg) override { void enqueue(const message_header& hdr, any_tuple msg) override {
m_scheduling_policy.enqueue(this, hdr, msg); m_scheduling_policy.enqueue(this, hdr, msg);
} }
void resume(util::fiber* fb) override { inline void launch() { m_scheduling_policy.launch(this); }
m_resume_policy.resume(this, fb);
}
inline void launch() {
m_scheduling_policy.launch(this);
}
inline mailbox_element* next_message() { inline mailbox_element* next_message() {
return m_priority_policy.next_message(this); return m_priority_policy.next_message(this);
...@@ -44,74 +53,113 @@ class proper_actor : public Base { ...@@ -44,74 +53,113 @@ class proper_actor : public Base {
} }
// grant access to the actor's mailbox // grant access to the actor's mailbox
Base::mailbox_type& mailbox() { typename Base::mailbox_type& mailbox() { return this->m_mailbox; }
return this->m_mailbox;
SchedulingPolicy& scheduling_policy() { return m_scheduling_policy; }
PriorityPolicy& priority_policy() { return m_priority_policy; }
ResumePolicy& resume_policy() { return m_resume_policy; }
InvokePolicy& invoke_policy() { return m_invoke_policy; }
inline void push_timeout() {
m_scheduling_policy.push_timeout();
}
inline void current_node(mailbox_element* ptr) {
this->m_current_node = ptr;
} }
inline void request_timeout(const util::duration& rel_timeout) {
m_invoke_policy.request_timeout(this, rel_timeout);
}
inline mailbox_element* current_node() { return this->m_current_node; }
detail::behavior_stack& bhvr_stack() { return this->m_bhvr_stack; }
protected: protected:
SchedulingPolicy m_scheduling_policy; SchedulingPolicy m_scheduling_policy;
PriorityPolicy m_priority_policy; PriorityPolicy m_priority_policy;
ResumePolicy m_resume_policy; ResumePolicy m_resume_policy;
InvokePolicy m_invoke_policy; InvokePolicy m_invoke_policy;
}; };
// for blocking actors, there's one more member function to implement // for blocking actors, there's one more member function to implement
template<class Base, template <class Base, class SchedulingPolicy, class PriorityPolicy,
class SchedulingPolicy, class ResumePolicy, class InvokePolicy>
class PriorityPolicy, class proper_actor<
class ResumePolicy, Base, SchedulingPolicy, PriorityPolicy, ResumePolicy, InvokePolicy,
class InvokePolicy> true> final : public proper_actor<Base, SchedulingPolicy, PriorityPolicy,
class proper_actor<Base, ResumePolicy, InvokePolicy, false> {
SchedulingPolicy,
PriorityPolicy, typedef proper_actor<Base, SchedulingPolicy, PriorityPolicy,
ResumePolicy, ResumePolicy, InvokePolicy, false>
InvokePolicy, super;
true> : public proper_actor<Base,
SchedulingPolicy,
PriorityPolicy,
ResumePolicy,
InvokePolicy,
false> {
public: public:
template <typename... Ts>
proper_actor(Ts&&... args) : super(std::forward<Ts>(args)...) { }
void dequeue(behavior& bhvr) override { void dequeue(behavior& bhvr) override {
if (bhvr.timeout().valid()) { if (bhvr.timeout().valid()) {
auto abs_time = m_scheduling_policy.init_timeout(this, bhvr.timeout()); auto tout =
this->m_scheduling_policy.init_timeout(this, bhvr.timeout());
auto done = false; auto done = false;
while (!done) { while (!done) {
if (!m_resume_policy.await_data(this, abs_time)) { if (!this->m_resume_policy.await_data(this, tout)) {
bhvr.handle_timeout(); bhvr.handle_timeout();
done = true; done = true;
} } else {
else { auto msg = this->m_priority_policy.next_message(this);
auto msg = m_priority_policy.next_message(this);
// must not return nullptr, because await_data guarantees // must not return nullptr, because await_data guarantees
// at least one message in our mailbox // at least one message in our mailbox
CPPA_REQUIRE(msg != nullptr); CPPA_REQUIRE(msg != nullptr);
done = m_invoke_policy.invoke(this, bhvr, msg); done = this->m_invoke_policy.invoke(this, msg, bhvr);
} }
} }
} } else {
else {
for (;;) { for (;;) {
auto msg = m_priority_policy.next_message(this); auto msg = this->m_priority_policy.next_message(this);
while (msg) { while (msg) {
if (m_invoke_policy.invoke(this, bhvr, msg)) { if (this->m_invoke_policy.invoke(this, msg, bhvr)) {
// we're done // we're done
return; return;
} }
msg = m_priority_policy.next_message(this); msg = this->m_priority_policy.next_message(this);
} }
m_resume_policy.await_data(this); this->m_resume_policy.await_data(this);
} }
} }
} }
mailbox_element* dequeue() override {
auto e = try_dequeue();
if (!e) {
this->m_resume_policy.await_data(this);
return try_dequeue(); // guaranteed to succeed after await_data
}
return e;
}
mailbox_element* try_dequeue() override {
return this->m_priority_policy.next_message(this);
}
mailbox_element* try_dequeue(const typename Base::timeout_type& tout) override {
if (this->m_resume_policy.await_data(this, tout)) {
return try_dequeue();
}
return nullptr;
}
}; };
} } // namespace cppa::detail } // namespace detail
} // namespace cppa
#endif // PROPER_ACTOR_HPP #endif // PROPER_ACTOR_HPP
...@@ -25,7 +25,7 @@ ...@@ -25,7 +25,7 @@
* * * *
* You should have received a copy of the GNU Lesser General Public License * * You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. * * along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/ \******************************************************************************/
#ifndef CPPA_RAW_ACCESS_HPP #ifndef CPPA_RAW_ACCESS_HPP
......
...@@ -33,10 +33,10 @@ ...@@ -33,10 +33,10 @@
#include <thread> #include <thread>
#include "cppa/resumable.hpp"
#include "cppa/scheduler.hpp" #include "cppa/scheduler.hpp"
#include "cppa/context_switching_actor.hpp"
#include "cppa/util/producer_consumer_list.hpp" #include "cppa/util/producer_consumer_list.hpp"
#include "cppa/detail/scheduled_actor_dummy.hpp"
namespace cppa { namespace detail { namespace cppa { namespace detail {
...@@ -46,8 +46,9 @@ class thread_pool_scheduler : public scheduler { ...@@ -46,8 +46,9 @@ class thread_pool_scheduler : public scheduler {
public: public:
using scheduler::init_callback; struct dummy : resumable {
using scheduler::actor_fun; resume_result resume(util::fiber*) override;
};
struct worker; struct worker;
...@@ -59,24 +60,20 @@ class thread_pool_scheduler : public scheduler { ...@@ -59,24 +60,20 @@ class thread_pool_scheduler : public scheduler {
void destroy(); void destroy();
void enqueue(scheduled_actor* what); void enqueue(resumable* what) override;
local_actor_ptr exec(spawn_options opts, scheduled_actor_ptr ptr) override;
local_actor_ptr exec(spawn_options opts, init_callback init_cb, actor_fun f) override;
private: private:
//typedef util::single_reader_queue<abstract_scheduled_actor> job_queue; //typedef util::single_reader_queue<abstract_scheduled_actor> job_queue;
typedef util::producer_consumer_list<scheduled_actor> job_queue; typedef util::producer_consumer_list<resumable> job_queue;
size_t m_num_threads; size_t m_num_threads;
job_queue m_queue; job_queue m_queue;
scheduled_actor_dummy m_dummy; dummy m_dummy;
std::thread m_supervisor; std::thread m_supervisor;
static void worker_loop(worker*); static void worker_loop(worker*);
static void supervisor_loop(job_queue*, scheduled_actor*, size_t); static void supervisor_loop(job_queue*, resumable*, size_t);
}; };
......
...@@ -51,8 +51,8 @@ struct extend_helper<D, B, M, Ms...> : extend_helper<D, M<B, D>, Ms...> { }; ...@@ -51,8 +51,8 @@ struct extend_helper<D, B, M, Ms...> : extend_helper<D, M<B, D>, Ms...> { };
/** /**
* @brief Allows convenient definition of types using mixins. * @brief Allows convenient definition of types using mixins.
* For example, @p extend<ar, T>::with<ob, fo> is an alias for * For example, "extend<ar, T>::with<ob, fo>" is an alias for
* @p fo<ob<ar, T>, T>. * "fo<ob<ar, T>, T>".
* *
* Mixins in libcppa always have two template parameters: base type and * Mixins in libcppa always have two template parameters: base type and
* derived type. This allows mixins to make use of the curiously recurring * derived type. This allows mixins to make use of the curiously recurring
...@@ -64,7 +64,7 @@ struct extend { ...@@ -64,7 +64,7 @@ struct extend {
/** /**
* @brief Identifies the combined type. * @brief Identifies the combined type.
*/ */
template<template<class, class> class... Mixins> template<CPPA_MIXIN... Mixins>
using with = typename detail::extend_helper<Derived, Base, Mixins...>::type; using with = typename detail::extend_helper<Derived, Base, Mixins...>::type;
}; };
......
...@@ -33,9 +33,9 @@ ...@@ -33,9 +33,9 @@
#include <map> #include <map>
#include "cppa/stackless.hpp" #include "cppa/extend.hpp"
#include "cppa/threadless.hpp"
#include "cppa/local_actor.hpp" #include "cppa/local_actor.hpp"
#include "cppa/behavior_stack_based.hpp"
#include "cppa/util/buffer.hpp" #include "cppa/util/buffer.hpp"
...@@ -45,7 +45,10 @@ ...@@ -45,7 +45,10 @@
#include "cppa/io/accept_handle.hpp" #include "cppa/io/accept_handle.hpp"
#include "cppa/io/connection_handle.hpp" #include "cppa/io/connection_handle.hpp"
namespace cppa { namespace io { #include "cppa/policy/sequential_invoke.hpp"
namespace cppa {
namespace io {
class broker; class broker;
...@@ -58,7 +61,9 @@ local_actor_ptr init_and_launch(broker_ptr); ...@@ -58,7 +61,9 @@ local_actor_ptr init_and_launch(broker_ptr);
* and other components in the network. * and other components in the network.
* @extends local_actor * @extends local_actor
*/ */
class broker : public extend<local_actor>::with<threadless, stackless> { class broker : public extend<local_actor>::with<behavior_stack_based> {
friend class policy::sequential_invoke;
typedef combined_type super; typedef combined_type super;
...@@ -97,6 +102,7 @@ class broker : public extend<local_actor>::with<threadless, stackless> { ...@@ -97,6 +102,7 @@ class broker : public extend<local_actor>::with<threadless, stackless> {
void write(const connection_handle& hdl, util::buffer&& buf); void write(const connection_handle& hdl, util::buffer&& buf);
/*
template<typename F, typename... Ts> template<typename F, typename... Ts>
static broker_ptr from(F fun, static broker_ptr from(F fun,
input_stream_ptr in, input_stream_ptr in,
...@@ -121,6 +127,7 @@ class broker : public extend<local_actor>::with<threadless, stackless> { ...@@ -121,6 +127,7 @@ class broker : public extend<local_actor>::with<threadless, stackless> {
std::forward<Ts>(args)...), std::forward<Ts>(args)...),
std::move(in)); std::move(in));
} }
*/
template<typename F, typename... Ts> template<typename F, typename... Ts>
actor fork(F fun, connection_handle hdl, Ts&&... args) { actor fork(F fun, connection_handle hdl, Ts&&... args) {
...@@ -154,14 +161,16 @@ class broker : public extend<local_actor>::with<threadless, stackless> { ...@@ -154,14 +161,16 @@ class broker : public extend<local_actor>::with<threadless, stackless> {
explicit broker(scribe_pointer); explicit broker(scribe_pointer);
virtual behavior make_behavior() = 0;
private: private:
actor_addr fork_impl(std::function<void (broker*)> fun, actor fork_impl(std::function<void (broker*)> fun,
connection_handle hdl); connection_handle hdl);
static broker_ptr from_impl(std::function<void (broker*)> fun, //static broker_ptr from_impl(std::function<void (broker*)> fun,
input_stream_ptr in, // input_stream_ptr in,
output_stream_ptr out); // output_stream_ptr out);
void invoke_message(const message_header& hdr, any_tuple msg); void invoke_message(const message_header& hdr, any_tuple msg);
...@@ -178,10 +187,27 @@ class broker : public extend<local_actor>::with<threadless, stackless> { ...@@ -178,10 +187,27 @@ class broker : public extend<local_actor>::with<threadless, stackless> {
std::map<accept_handle, doorman_pointer> m_accept; std::map<accept_handle, doorman_pointer> m_accept;
std::map<connection_handle, scribe_pointer> m_io; std::map<connection_handle, scribe_pointer> m_io;
policy::sequential_invoke m_invoke;
}; };
//typedef intrusive_ptr<broker> broker_ptr; class default_broker : public broker {
public:
typedef std::function<void (broker*)> function_type;
default_broker(input_stream_ptr in, output_stream_ptr out, function_type f);
behavior make_behavior() override;
private:
function_type m_fun;
};
} } // namespace cppa::network } // namespace io
} // namespace cppa
#endif // CPPA_BROKER_HPP #endif // CPPA_BROKER_HPP
...@@ -31,6 +31,7 @@ ...@@ -31,6 +31,7 @@
#ifndef CPPA_CONTEXT_HPP #ifndef CPPA_CONTEXT_HPP
#define CPPA_CONTEXT_HPP #define CPPA_CONTEXT_HPP
#include <atomic>
#include <cstdint> #include <cstdint>
#include <functional> #include <functional>
...@@ -42,9 +43,10 @@ ...@@ -42,9 +43,10 @@
#include "cppa/any_tuple.hpp" #include "cppa/any_tuple.hpp"
#include "cppa/message_id.hpp" #include "cppa/message_id.hpp"
#include "cppa/match_expr.hpp" #include "cppa/match_expr.hpp"
#include "cppa/actor_state.hpp"
#include "cppa/exit_reason.hpp" #include "cppa/exit_reason.hpp"
#include "cppa/typed_actor.hpp" #include "cppa/typed_actor.hpp"
#include "cppa/mailbox_based.hpp" #include "cppa/spawn_options.hpp"
#include "cppa/memory_cached.hpp" #include "cppa/memory_cached.hpp"
#include "cppa/message_future.hpp" #include "cppa/message_future.hpp"
#include "cppa/message_header.hpp" #include "cppa/message_header.hpp"
...@@ -58,6 +60,8 @@ ...@@ -58,6 +60,8 @@
#include "cppa/detail/behavior_stack.hpp" #include "cppa/detail/behavior_stack.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
namespace cppa { namespace cppa {
// forward declarations // forward declarations
...@@ -66,22 +70,69 @@ class message_future; ...@@ -66,22 +70,69 @@ class message_future;
class local_scheduler; class local_scheduler;
class sync_handle_helper; class sync_handle_helper;
namespace detail { class receive_policy; } namespace util {
struct fiber;
} // namespace util
template<class Impl, spawn_options Options = no_spawn_options, typename... Ts>
actor spawn(Ts&&... args);
template<spawn_options Options = no_spawn_options, typename... Ts>
actor spawn(Ts&&... args);
/** /**
* @brief Base class for local running Actors. * @brief Base class for local running Actors.
* @extends actor * @extends actor
*/ */
class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_based> { class local_actor : public extend<abstract_actor>::with<memory_cached> {
friend class detail::receive_policy;
typedef combined_type super; typedef combined_type super;
public: public:
typedef detail::disposer del;
typedef intrusive::single_reader_queue<mailbox_element, del> mailbox_type;
~local_actor(); ~local_actor();
inline actor eval_opts(spawn_options opts, actor res) {
if (has_monitor_flag(opts)) {
monitor(res);
}
if (has_link_flag(opts)) {
link_to(res);
}
return res;
}
template<class Impl, spawn_options Options = no_spawn_options, typename... Ts>
actor spawn(Ts&&... args) {
auto res = cppa::spawn<Impl, make_unbound(Options)>(std::forward<Ts>(args)...);
return eval_opts(Options, std::move(res));
}
template<spawn_options Options = no_spawn_options, typename... Ts>
actor spawn(Ts&&... args) {
auto res = cppa::spawn<make_unbound(Options)>(std::forward<Ts>(args)...);
return eval_opts(Options, std::move(res));
}
/**
* @brief Sends @p what to the receiver specified in @p dest.
*/
void send_tuple(message_priority prio, const channel& dest, any_tuple what);
/**
* @brief Sends <tt>{what...}</tt> to the receiver specified in @p hdr.
* @pre <tt>sizeof...(Ts) > 0</tt>
*/
template<typename... Ts>
void send(message_priority prio, const channel& whom, Ts&&... what) {
send_tuple(prio, whom, make_any_tuple(std::forward<Ts>(what)...));
}
/** /**
* @brief Sends @p what to the receiver specified in @p dest. * @brief Sends @p what to the receiver specified in @p dest.
*/ */
...@@ -96,6 +147,8 @@ class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_b ...@@ -96,6 +147,8 @@ class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_b
send_tuple(whom, make_any_tuple(std::forward<Ts>(what)...)); send_tuple(whom, make_any_tuple(std::forward<Ts>(what)...));
} }
void send_exit(const actor_addr& whom, std::uint32_t reason);
/** /**
* @brief Sends @p what as a synchronous message to @p whom. * @brief Sends @p what as a synchronous message to @p whom.
* @param whom Receiver of the message. * @param whom Receiver of the message.
...@@ -128,8 +181,8 @@ class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_b ...@@ -128,8 +181,8 @@ class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_b
} }
template<typename... Ts> template<typename... Ts>
message_future timed_sync_send(const util::duration& rtime, message_future timed_sync_send(const actor& dest,
const actor& dest, const util::duration& rtime,
Ts&&... what) { Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
return timed_sync_send_tuple(rtime, dest, make_any_tuple(std::forward<Ts>(what)...)); return timed_sync_send_tuple(rtime, dest, make_any_tuple(std::forward<Ts>(what)...));
...@@ -146,6 +199,12 @@ class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_b ...@@ -146,6 +199,12 @@ class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_b
const util::duration& rtime, const util::duration& rtime,
any_tuple data); any_tuple data);
template<typename... Ts>
void delayed_send(const channel& dest, const util::duration& rtime,
Ts&&... args) {
delayed_send_tuple(dest, rtime, make_any_tuple(std::forward<Ts>(args)...));
}
/** /**
* @brief Causes this actor to subscribe to the group @p what. * @brief Causes this actor to subscribe to the group @p what.
* *
...@@ -234,15 +293,6 @@ class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_b ...@@ -234,15 +293,6 @@ class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_b
demonitor(whom->address()); demonitor(whom->address());
} }
/**
* @brief Can be overridden to initialize an actor before any
* message is handled.
* @warning Must not call blocking functions such as
* {@link cppa::receive receive}.
* @note Calling {@link become} to set an initial behavior is supported.
*/
virtual void init();
/** /**
* @brief Can be overridden to perform cleanup code after an actor * @brief Can be overridden to perform cleanup code after an actor
* finished execution. * finished execution.
...@@ -309,9 +359,7 @@ class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_b ...@@ -309,9 +359,7 @@ class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_b
else quit(exit_reason::unhandled_sync_failure); else quit(exit_reason::unhandled_sync_failure);
} }
local_actor(bool is_scheduled = false); local_actor();
virtual bool initialized() const = 0;
inline bool chaining_enabled(); inline bool chaining_enabled();
...@@ -332,20 +380,32 @@ class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_b ...@@ -332,20 +380,32 @@ class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_b
inline void mark_arrived(message_id response_id); inline void mark_arrived(message_id response_id);
//virtual void become_waiting_for(behavior, message_id) = 0; inline std::uint32_t planned_exit_reason() const;
//inline detail::behavior_stack& bhvr_stack();
//virtual void do_become(behavior&& bhvr, bool discard_old) = 0; inline void planned_exit_reason(std::uint32_t value);
//inline void do_become(const behavior& bhvr, bool discard_old); actor_state cas_state(actor_state expected, actor_state desired) {
auto e = expected;
do { if (m_state.compare_exchange_weak(e, desired)) return desired; }
while (e == expected);
return e;
}
inline std::uint32_t planned_exit_reason() const; inline void set_state(actor_state new_value) {
m_state.store(new_value);
}
inline void planned_exit_reason(std::uint32_t value); inline actor_state state() const {
return m_state;
}
protected: protected:
template<typename... Ts>
inline mailbox_element* new_mailbox_element(Ts&&... args) {
return mailbox_element::create(std::forward<Ts>(args)...);
}
void remove_handler(message_id id); void remove_handler(message_id id);
void cleanup(std::uint32_t reason); void cleanup(std::uint32_t reason);
...@@ -353,9 +413,6 @@ class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_b ...@@ -353,9 +413,6 @@ class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_b
// true if this actor receives EXIT messages as ordinary messages // true if this actor receives EXIT messages as ordinary messages
bool m_trap_exit; bool m_trap_exit;
// true if this actor takes part in cooperative scheduling
bool m_is_scheduled;
// identifies the ID of the last sent synchronous request // identifies the ID of the last sent synchronous request
message_id m_last_request_id; message_id m_last_request_id;
...@@ -372,12 +429,12 @@ class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_b ...@@ -372,12 +429,12 @@ class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_b
// {group => subscription} map of all joined groups // {group => subscription} map of all joined groups
std::map<group_ptr, group::subscription> m_subscriptions; std::map<group_ptr, group::subscription> m_subscriptions;
// allows actors to keep previous behaviors and enables unbecome()
//detail::behavior_stack m_bhvr_stack;
// set by quit // set by quit
std::uint32_t m_planned_exit_reason; std::uint32_t m_planned_exit_reason;
// the state of the (possibly cooperatively scheduled) actor
std::atomic<actor_state> m_state;
/** @endcond */ /** @endcond */
private: private:
......
...@@ -53,6 +53,11 @@ class mailbox_based : public Base { ...@@ -53,6 +53,11 @@ class mailbox_based : public Base {
} }
} }
template<typename... Ts>
inline mailbox_element* new_mailbox_element(Ts&&... args) {
return mailbox_element::create(std::forward<Ts>(args)...);
}
protected: protected:
typedef mailbox_based combined_type; typedef mailbox_based combined_type;
...@@ -68,11 +73,6 @@ class mailbox_based : public Base { ...@@ -68,11 +73,6 @@ class mailbox_based : public Base {
Base::cleanup(reason); Base::cleanup(reason);
} }
template<typename... Ts>
inline mailbox_element* new_mailbox_element(Ts&&... args) {
return mailbox_element::create(std::forward<Ts>(args)...);
}
mailbox_type m_mailbox; mailbox_type m_mailbox;
}; };
......
...@@ -64,7 +64,7 @@ template<typename T1, typename T2> ...@@ -64,7 +64,7 @@ template<typename T1, typename T2>
inline const T2& deduce_const(const T1&, T2& rhs) { return rhs; } inline const T2& deduce_const(const T1&, T2& rhs) { return rhs; }
template<class FilteredPattern> template<class FilteredPattern>
struct invoke_policy_base { struct invoke_util_base {
typedef FilteredPattern filtered_pattern; typedef FilteredPattern filtered_pattern;
typedef typename pseudo_tuple_from_type_list<filtered_pattern>::type typedef typename pseudo_tuple_from_type_list<filtered_pattern>::type
tuple_type; tuple_type;
...@@ -72,9 +72,9 @@ struct invoke_policy_base { ...@@ -72,9 +72,9 @@ struct invoke_policy_base {
// covers wildcard_position::multiple and wildcard_position::in_between // covers wildcard_position::multiple and wildcard_position::in_between
template<wildcard_position, class Pattern, class FilteredPattern> template<wildcard_position, class Pattern, class FilteredPattern>
struct invoke_policy_impl : invoke_policy_base<FilteredPattern> { struct invoke_util_impl : invoke_util_base<FilteredPattern> {
typedef invoke_policy_base<FilteredPattern> super; typedef invoke_util_base<FilteredPattern> super;
template<class Tuple> template<class Tuple>
static bool can_invoke(const std::type_info& type_token, static bool can_invoke(const std::type_info& type_token,
...@@ -113,12 +113,12 @@ struct invoke_policy_impl : invoke_policy_base<FilteredPattern> { ...@@ -113,12 +113,12 @@ struct invoke_policy_impl : invoke_policy_base<FilteredPattern> {
}; };
template<> template<>
struct invoke_policy_impl<wildcard_position::nil, struct invoke_util_impl<wildcard_position::nil,
util::empty_type_list, util::empty_type_list,
util::empty_type_list > util::empty_type_list >
: invoke_policy_base<util::empty_type_list> { : invoke_util_base<util::empty_type_list> {
typedef invoke_policy_base<util::empty_type_list> super; typedef invoke_util_base<util::empty_type_list> super;
template<typename PtrType, class Tuple> template<typename PtrType, class Tuple>
static bool prepare_invoke(typename super::tuple_type&, static bool prepare_invoke(typename super::tuple_type&,
...@@ -137,12 +137,12 @@ struct invoke_policy_impl<wildcard_position::nil, ...@@ -137,12 +137,12 @@ struct invoke_policy_impl<wildcard_position::nil,
}; };
template<class Pattern, typename... Ts> template<class Pattern, typename... Ts>
struct invoke_policy_impl<wildcard_position::nil, struct invoke_util_impl<wildcard_position::nil,
Pattern, Pattern,
util::type_list<Ts...>> util::type_list<Ts...>>
: invoke_policy_base<util::type_list<Ts...>> { : invoke_util_base<util::type_list<Ts...>> {
typedef invoke_policy_base<util::type_list<Ts...>> super; typedef invoke_util_base<util::type_list<Ts...>> super;
typedef typename super::tuple_type tuple_type; typedef typename super::tuple_type tuple_type;
...@@ -242,12 +242,12 @@ struct invoke_policy_impl<wildcard_position::nil, ...@@ -242,12 +242,12 @@ struct invoke_policy_impl<wildcard_position::nil,
}; };
template<> template<>
struct invoke_policy_impl<wildcard_position::leading, struct invoke_util_impl<wildcard_position::leading,
util::type_list<anything>, util::type_list<anything>,
util::empty_type_list> util::empty_type_list>
: invoke_policy_base<util::empty_type_list> { : invoke_util_base<util::empty_type_list> {
typedef invoke_policy_base<util::empty_type_list> super; typedef invoke_util_base<util::empty_type_list> super;
template<class Tuple> template<class Tuple>
static inline bool can_invoke(const std::type_info&, const Tuple&) { static inline bool can_invoke(const std::type_info&, const Tuple&) {
...@@ -266,12 +266,12 @@ struct invoke_policy_impl<wildcard_position::leading, ...@@ -266,12 +266,12 @@ struct invoke_policy_impl<wildcard_position::leading,
}; };
template<class Pattern, typename... Ts> template<class Pattern, typename... Ts>
struct invoke_policy_impl<wildcard_position::trailing, struct invoke_util_impl<wildcard_position::trailing,
Pattern, Pattern,
util::type_list<Ts...>> util::type_list<Ts...>>
: invoke_policy_base<util::type_list<Ts...>> { : invoke_util_base<util::type_list<Ts...>> {
typedef invoke_policy_base<util::type_list<Ts...>> super; typedef invoke_util_base<util::type_list<Ts...>> super;
template<class Tuple> template<class Tuple>
static bool can_invoke(const std::type_info& arg_types, static bool can_invoke(const std::type_info& arg_types,
...@@ -308,12 +308,12 @@ struct invoke_policy_impl<wildcard_position::trailing, ...@@ -308,12 +308,12 @@ struct invoke_policy_impl<wildcard_position::trailing,
}; };
template<class Pattern, typename... Ts> template<class Pattern, typename... Ts>
struct invoke_policy_impl<wildcard_position::leading, struct invoke_util_impl<wildcard_position::leading,
Pattern, Pattern,
util::type_list<Ts...>> util::type_list<Ts...>>
: invoke_policy_base<util::type_list<Ts...>> { : invoke_util_base<util::type_list<Ts...>> {
typedef invoke_policy_base<util::type_list<Ts...>> super; typedef invoke_util_base<util::type_list<Ts...>> super;
template<class Tuple> template<class Tuple>
static bool can_invoke(const std::type_info& arg_types, static bool can_invoke(const std::type_info& arg_types,
...@@ -352,8 +352,8 @@ struct invoke_policy_impl<wildcard_position::leading, ...@@ -352,8 +352,8 @@ struct invoke_policy_impl<wildcard_position::leading,
}; };
template<class Pattern> template<class Pattern>
struct invoke_policy struct invoke_util
: invoke_policy_impl< : invoke_util_impl<
get_wildcard_position<Pattern>(), get_wildcard_position<Pattern>(),
Pattern, Pattern,
typename util::tl_filter_not_type<Pattern, anything>::type> { typename util::tl_filter_not_type<Pattern, anything>::type> {
...@@ -507,7 +507,7 @@ Result unroll_expr(PPFPs& fs, ...@@ -507,7 +507,7 @@ Result unroll_expr(PPFPs& fs,
auto& f = get<N>(fs); auto& f = get<N>(fs);
typedef typename util::rm_const_and_ref<decltype(f)>::type Fun; typedef typename util::rm_const_and_ref<decltype(f)>::type Fun;
typedef typename Fun::pattern_type pattern_type; typedef typename Fun::pattern_type pattern_type;
typedef detail::invoke_policy<pattern_type> policy; typedef detail::invoke_util<pattern_type> policy;
typename policy::tuple_type targs; typename policy::tuple_type targs;
if (policy::prepare_invoke(targs, type_token, is_dynamic, ptr, tup)) { if (policy::prepare_invoke(targs, type_token, is_dynamic, ptr, tup)) {
auto is = util::get_indices(targs); auto is = util::get_indices(targs);
...@@ -536,7 +536,7 @@ inline bool can_unroll_expr(PPFPs& fs, ...@@ -536,7 +536,7 @@ inline bool can_unroll_expr(PPFPs& fs,
auto& f = get<N>(fs); auto& f = get<N>(fs);
typedef typename util::rm_const_and_ref<decltype(f)>::type Fun; typedef typename util::rm_const_and_ref<decltype(f)>::type Fun;
typedef typename Fun::pattern_type pattern_type; typedef typename Fun::pattern_type pattern_type;
typedef detail::invoke_policy<pattern_type> policy; typedef detail::invoke_util<pattern_type> policy;
return policy::can_invoke(arg_types, tup); return policy::can_invoke(arg_types, tup);
} }
...@@ -556,7 +556,7 @@ inline std::uint64_t calc_bitmask(PPFPs& fs, ...@@ -556,7 +556,7 @@ inline std::uint64_t calc_bitmask(PPFPs& fs,
auto& f = get<N>(fs); auto& f = get<N>(fs);
typedef typename util::rm_const_and_ref<decltype(f)>::type Fun; typedef typename util::rm_const_and_ref<decltype(f)>::type Fun;
typedef typename Fun::pattern_type pattern_type; typedef typename Fun::pattern_type pattern_type;
typedef detail::invoke_policy<pattern_type> policy; typedef detail::invoke_util<pattern_type> policy;
std::uint64_t result = policy::can_invoke(tinf, tup) ? (0x01 << N) : 0x00; std::uint64_t result = policy::can_invoke(tinf, tup) ? (0x01 << N) : 0x00;
return result | calc_bitmask(fs, long_constant<N-1l>(), tinf, tup); return result | calc_bitmask(fs, long_constant<N-1l>(), tinf, tup);
} }
......
...@@ -34,7 +34,9 @@ ...@@ -34,7 +34,9 @@
#include <cstdint> #include <cstdint>
#include <type_traits> #include <type_traits>
#include "cppa/on.hpp"
#include "cppa/match_expr.hpp" #include "cppa/match_expr.hpp"
#include "cppa/message_id.hpp"
#include "cppa/partial_function.hpp" #include "cppa/partial_function.hpp"
namespace cppa { namespace cppa {
...@@ -56,8 +58,10 @@ class continue_helper { ...@@ -56,8 +58,10 @@ class continue_helper {
inline continue_helper(message_id mid) : m_mid(mid) { } inline continue_helper(message_id mid) : m_mid(mid) { }
template<typename F> template<typename F>
continue_helper& continue_with(F fun); continue_helper& continue_with(F) {
}
inline message_id get_message_id() const { inline message_id get_message_id() const {
return m_mid; return m_mid;
} }
...@@ -78,12 +82,14 @@ class message_future { ...@@ -78,12 +82,14 @@ class message_future {
message_future() = delete; message_future() = delete;
inline continue_helper then(const partial_function& pfun) { inline continue_helper then(const behavior&) {
//FIXME
return message_id{};
} }
inline continue_helper await(const partial_function& pfun) { inline continue_helper await(const behavior&) {
//FIXME
return message_id{};
} }
/** /**
...@@ -113,7 +119,7 @@ class message_future { ...@@ -113,7 +119,7 @@ class message_future {
continue_helper continue_helper
>::type >::type
then(Fs... fs) { then(Fs... fs) {
return then(fs2bhvr(std::move(fs)...)); return then(partial_function{(on_arg_match >> std::move(fs))...});
} }
/** /**
...@@ -124,7 +130,7 @@ class message_future { ...@@ -124,7 +130,7 @@ class message_future {
template<typename... Fs> template<typename... Fs>
typename std::enable_if<util::all_callable<Fs...>::value>::type typename std::enable_if<util::all_callable<Fs...>::value>::type
await(Fs... fs) { await(Fs... fs) {
await(fs2bhvr({std::move(fs)...})); await(partial_function{(on_arg_match >> std::move(fs))...});
} }
/** /**
...@@ -142,8 +148,6 @@ class message_future { ...@@ -142,8 +148,6 @@ class message_future {
message_id m_mid; message_id m_mid;
local_actor* self; local_actor* self;
partial_function fs2bhvr(partial_function pf);
inline void check_consistency() { } inline void check_consistency() { }
}; };
...@@ -193,17 +197,6 @@ inline sync_handle_helper handle_response(const message_future& f) { ...@@ -193,17 +197,6 @@ inline sync_handle_helper handle_response(const message_future& f) {
return {f}; return {f};
} }
/**
* @brief Handles a synchronous response message in an event-based way.
* @param handle A future for a synchronous response.
* @throws std::logic_error if @p handle is not valid or if the actor
* already received the response for @p handle
* @relates message_future
*/
inline sync_receive_helper receive_response(const message_future& f) {
return {f};
}
} // namespace cppa } // namespace cppa
#endif // CPPA_MESSAGE_FUTURE_HPP #endif // CPPA_MESSAGE_FUTURE_HPP
This diff is collapsed.
...@@ -33,10 +33,14 @@ ...@@ -33,10 +33,14 @@
#include "cppa/policy/no_resume.hpp" #include "cppa/policy/no_resume.hpp"
#include "cppa/policy/prioritizing.hpp" #include "cppa/policy/prioritizing.hpp"
#include "cppa/policy/no_scheduling.hpp"
#include "cppa/policy/invoke_policy.hpp"
#include "cppa/policy/nestable_invoke.hpp" #include "cppa/policy/nestable_invoke.hpp"
#include "cppa/policy/not_prioritizing.hpp"
#include "cppa/policy/sequential_invoke.hpp" #include "cppa/policy/sequential_invoke.hpp"
#include "cppa/policy/event_based_resume.hpp" #include "cppa/policy/event_based_resume.hpp"
#include "cppa/policy/invoke_policy_base.hpp" #include "cppa/policy/middleman_scheduling.hpp"
#include "cppa/policy/cooperative_scheduling.hpp"
#include "cppa/policy/context_switching_resume.hpp" #include "cppa/policy/context_switching_resume.hpp"
#endif // CPPA_POLICY_HPP #endif // CPPA_POLICY_HPP
...@@ -27,19 +27,26 @@ ...@@ -27,19 +27,26 @@
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. * * along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/ \******************************************************************************/
#ifndef CPPA_CONTEXT_SWITCHING_ACTOR_HPP #ifndef CPPA_CONTEXT_SWITCHING_ACTOR_HPP
#define CPPA_CONTEXT_SWITCHING_ACTOR_HPP #define CPPA_CONTEXT_SWITCHING_ACTOR_HPP
#include "cppa/config.hpp" #include "cppa/config.hpp"
#include "cppa/resumable.hpp"
#include "cppa/actor_state.hpp"
#include "cppa/mailbox_element.hpp" #include "cppa/mailbox_element.hpp"
#include "cppa/util/fiber.hpp" #include "cppa/util/fiber.hpp"
#include "cppa/policy/resume_policy.hpp"
#include "cppa/detail/yield_interface.hpp" #include "cppa/detail/yield_interface.hpp"
namespace cppa { class local_actor; } namespace cppa {
class local_actor;
}
namespace cppa { namespace policy { namespace cppa {
namespace policy {
/** /**
* @brief Context-switching actor implementation. * @brief Context-switching actor implementation.
...@@ -49,9 +56,57 @@ class context_switching_resume { ...@@ -49,9 +56,57 @@ class context_switching_resume {
public: public:
// required by util::fiber
static void trampoline(void* _this);
// Base must be a mailbox-based actor
template<class Base>
struct mixin : Base, resumable {
template<typename... Ts>
mixin(Ts&&... args)
: Base(std::forward<Ts>(args)...)
, m_fiber(context_switching_resume::trampoline,
static_cast<blocking_untyped_actor*>(this)) { }
resumable::resume_result resume(util::fiber* from) override {
CPPA_REQUIRE(from != nullptr);
using namespace detail;
for (;;) {
switch (call(&m_fiber, from)) {
case yield_state::done: {
CPPA_REQUIRE(next_job == nullptr);
return resumable::done;
}
case yield_state::ready: { break; }
case yield_state::blocked: {
CPPA_REQUIRE(next_job == nullptr);
CPPA_REQUIRE(m_chained_actor == nullptr);
switch (this->cas_state(actor_state::about_to_block,
actor_state::blocked)) {
case actor_state::ready: {
// restore variables
CPPA_REQUIRE(next_job == nullptr);
break;
}
case actor_state::blocked: {
// wait until someone re-schedules that actor
return resumable::resume_later;
}
default: { CPPA_CRITICAL("illegal yield result"); }
}
break;
}
default: { CPPA_CRITICAL("illegal state"); }
}
}
}
util::fiber m_fiber;
};
template<class Actor, typename F> template <class Actor, typename F>
void fetch_messages(Actor* self, F cb) { void fetch_messages(Actor* self, F cb) {
auto e = self->m_mailbox.try_pop(); auto e = self->m_mailbox.try_pop();
while (e == nullptr) { while (e == nullptr) {
...@@ -63,7 +118,8 @@ class context_switching_resume { ...@@ -63,7 +118,8 @@ class context_switching_resume {
self->set_state(actor_state::ready); self->set_state(actor_state::ready);
} }
// wait until actor becomes rescheduled // wait until actor becomes rescheduled
else detail::yield(detail::yield_state::blocked); else
detail::yield(detail::yield_state::blocked);
} }
} }
// ok, we have at least one message // ok, we have at least one message
...@@ -73,7 +129,7 @@ class context_switching_resume { ...@@ -73,7 +129,7 @@ class context_switching_resume {
} }
} }
template<class Actor, typename F> template <class Actor, typename F>
void try_fetch_messages(Actor* self, F cb) { void try_fetch_messages(Actor* self, F cb) {
auto e = self->m_mailbox.try_pop(); auto e = self->m_mailbox.try_pop();
while (e) { while (e) {
...@@ -83,57 +139,33 @@ class context_switching_resume { ...@@ -83,57 +139,33 @@ class context_switching_resume {
} }
template<class Actor> template<class Actor>
resume_result resume(Actor* self, util::fiber* from) { void await_data(Actor* self) {
CPPA_LOG_TRACE("state = " << static_cast<int>(self->state())); while (self->m_mailbox.can_fetch_more() == false) {
CPPA_REQUIRE(from != nullptr); self->set_state(actor_state::about_to_block);
CPPA_REQUIRE(next_job == nullptr); // make sure mailbox is empty
using namespace detail; if (self->m_mailbox.can_fetch_more()) {
for (;;) { // someone preempt us => continue
switch (call(&m_fiber, from)) { self->set_state(actor_state::ready);
case yield_state::done: {
CPPA_REQUIRE(next_job == nullptr);
return resume_result::actor_done;
}
case yield_state::ready: {
break;
}
case yield_state::blocked: {
CPPA_REQUIRE(next_job == nullptr);
CPPA_REQUIRE(m_chained_actor == nullptr);
switch (compare_exchange_state(actor_state::about_to_block,
actor_state::blocked)) {
case actor_state::ready: {
// restore variables
CPPA_REQUIRE(next_job == nullptr);
break;
}
case actor_state::blocked: {
// wait until someone re-schedules that actor
return resume_result::actor_blocked;
}
default: {
CPPA_CRITICAL("illegal yield result");
}
}
break;
}
default: {
CPPA_CRITICAL("illegal state");
}
} }
// wait until actor becomes rescheduled
else detail::yield(detail::yield_state::blocked);
} }
} }
private: template<class Actor, typename AbsTimeout>
bool await_data(Actor* self, const AbsTimeout&) {
await_data(self);
return true;
}
// required by util::fiber private:
static void trampoline(void* _this);
// members // members
util::fiber m_fiber; util::fiber m_fiber;
}; };
} } // namespace cppa::policy } // namespace policy
} // namespace cppa
#endif // CPPA_CONTEXT_SWITCHING_ACTOR_HPP #endif // CPPA_CONTEXT_SWITCHING_ACTOR_HPP
...@@ -31,6 +31,17 @@ ...@@ -31,6 +31,17 @@
#ifndef CPPA_COOPERATIVE_SCHEDULING_HPP #ifndef CPPA_COOPERATIVE_SCHEDULING_HPP
#define CPPA_COOPERATIVE_SCHEDULING_HPP #define CPPA_COOPERATIVE_SCHEDULING_HPP
#include <atomic>
#include "cppa/any_tuple.hpp"
#include "cppa/resumable.hpp"
#include "cppa/actor_state.hpp"
#include "cppa/message_header.hpp"
#include "cppa/detail/yield_interface.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
namespace cppa { namespace policy { namespace cppa { namespace policy {
class cooperative_scheduling { class cooperative_scheduling {
...@@ -49,12 +60,12 @@ class cooperative_scheduling { ...@@ -49,12 +60,12 @@ class cooperative_scheduling {
// this does return nullptr // this does return nullptr
template<class Actor, typename F> template<class Actor, typename F>
void fetch_messages(Actor* self, F cb) { void fetch_messages(Actor* self, F cb) {
auto e = self->m_mailbox.try_pop(); auto e = self->mailbox().try_pop();
while (e == nullptr) { while (e == nullptr) {
if (self->m_mailbox.can_fetch_more() == false) { if (self->mailbox().can_fetch_more() == false) {
self->set_state(actor_state::about_to_block); self->set_state(actor_state::about_to_block);
// make sure mailbox is empty // make sure mailbox is empty
if (self->m_mailbox.can_fetch_more()) { if (self->mailbox().can_fetch_more()) {
// someone preempt us => continue // someone preempt us => continue
self->set_state(actor_state::ready); self->set_state(actor_state::ready);
} }
...@@ -65,7 +76,7 @@ class cooperative_scheduling { ...@@ -65,7 +76,7 @@ class cooperative_scheduling {
// ok, we have at least one message // ok, we have at least one message
while (e) { while (e) {
cb(e); cb(e);
e = self->m_mailbox.try_pop(); e = self->mailbox().try_pop();
} }
} }
...@@ -76,6 +87,59 @@ class cooperative_scheduling { ...@@ -76,6 +87,59 @@ class cooperative_scheduling {
fetch_messages(self, cb); fetch_messages(self, cb);
} }
template<class Actor>
inline void launch(Actor*) {
static_cast<void>(m_hidden);
}
template<class Actor>
void enqueue(Actor* self, const message_header& hdr, any_tuple& msg) {
auto e = self->new_mailbox_element(hdr, std::move(msg));
switch (self->mailbox().enqueue(e)) {
case intrusive::first_enqueued: {
auto state = self->state();
auto set_ready = [&]() -> bool {
auto s = self->cas_state(state, actor_state::ready);
return s == actor_state::ready;
};
for (;;) {
switch (state) {
case actor_state::blocked: {
if (set_ready()) {
CPPA_REQUIRE(m_scheduler != nullptr);
//m_scheduler->enqueue(this);
return;
}
break;
}
case actor_state::about_to_block: {
if (set_ready()) {
return;
}
break;
}
default: return;
}
}
break;
}
case intrusive::queue_closed: {
if (hdr.id.is_request()) {
//FIXME
//detail::sync_request_bouncer f{exit_reason()};
//f(hdr.sender, hdr.id);
}
break;
}
default: break;
}
}
private:
// denotes whether this actor is ignored by await_all_actors_done()
bool m_hidden;
}; };
} } // namespace cppa::policy } } // namespace cppa::policy
......
This diff is collapsed.
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* 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 2.1 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_POLICY_MIDDLEMAN_SCHEDULING_HPP
#define CPPA_POLICY_MIDDLEMAN_SCHEDULING_HPP
#include "cppa/singletons.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/policy/cooperative_scheduling.hpp"
namespace cppa {
namespace policy {
// Actor must implement invoke_message
class middleman_scheduling {
public:
template<class Actor>
class continuation {
public:
typedef intrusive_ptr<Actor> pointer;
continuation(pointer ptr, const message_header& hdr, any_tuple&& msg)
: m_self(std::move(ptr)), m_hdr(hdr), m_data(std::move(msg)) { }
inline void operator()() const {
m_self->invoke_message(m_hdr, move(m_data));
}
private:
pointer m_self;
message_header m_hdr;
any_tuple m_data;
};
using timeout_type = int;
template<class Actor>
timeout_type init_timeout(Actor* self, const util::duration& rel_time) {
// request explicit timeout message
self->request_timeout(rel_time);
return 0; // return some dummy value
}
// this does return nullptr
template<class Actor, typename F>
inline void fetch_messages(Actor*, F) {
// clients cannot fetch messages
}
template<class Actor, typename F>
inline void fetch_messages(Actor* self, F cb, timeout_type) {
// a call to this call is always preceded by init_timeout,
// which will trigger a timeout message
fetch_messages(self, cb);
}
template<class Actor>
inline void launch(Actor*) {
// nothing to do
}
template<class Actor>
void enqueue(Actor* self, const message_header& hdr, any_tuple& msg) {
get_middleman()->run_later(continuation<Actor>{self, hdr, std::move(msg)});
}
};
} // namespace policy
} // namespace cppa
#endif // CPPA_POLICY_MIDDLEMAN_SCHEDULING_HPP
...@@ -44,13 +44,13 @@ ...@@ -44,13 +44,13 @@
#include "cppa/intrusive/single_reader_queue.hpp" #include "cppa/intrusive/single_reader_queue.hpp"
#include "cppa/policy/invoke_policy_base.hpp" #include "cppa/policy/invoke_policy.hpp"
namespace cppa { namespace detail { class receive_policy; } } namespace cppa { namespace detail { class receive_policy; } }
namespace cppa { namespace policy { namespace cppa { namespace policy {
class nestable_invoke : public invoke_policy_base<nestable_invoke> { class nestable_invoke : public invoke_policy<nestable_invoke> {
typedef std::unique_lock<std::mutex> lock_type; typedef std::unique_lock<std::mutex> lock_type;
...@@ -62,34 +62,32 @@ class nestable_invoke : public invoke_policy_base<nestable_invoke> { ...@@ -62,34 +62,32 @@ class nestable_invoke : public invoke_policy_base<nestable_invoke> {
template<class Client> template<class Client>
static inline pointer hm_begin(Client* client, pointer node) { static inline pointer hm_begin(Client* client, pointer node) {
auto previous = client->m_current_node; auto previous = client->current_node();
client->m_current_node = node; client->current_node(node);
return previous; return previous;
} }
template<class Client> template<class Client>
static inline void hm_cleanup(Client* client, pointer /*previous*/) { static inline void hm_cleanup(Client* client, pointer /*previous*/) {
client->m_current_node = &(client->m_dummy_node); client->current_node(&(client->m_dummy_node));
if (client->has_behavior()) {
client->request_timeout(client->get_behavior().timeout());
}
else client->reset_timeout();
} }
template<class Client> template<class Client>
static inline void hm_revert(Client* client, pointer previous) { static inline void hm_revert(Client* client, pointer previous) {
client->m_current_node = previous; client->current_node(previous);
} }
typedef std::chrono::high_resolution_clock::time_point timeout_type; typedef std::chrono::high_resolution_clock::time_point timeout_type;
nestable_invoke() : m_initialized(false) { }
inline void reset_timeout() { } inline void reset_timeout() { }
inline void request_timeout(const util::duration&) { } template<class Actor>
inline void request_timeout(Actor*, const util::duration&) { }
inline void handle_timeout(behavior& bhvr) { bhvr.handle_timeout(); } template<class Actor>
inline void handle_timeout(Actor*, behavior& bhvr) {
bhvr.handle_timeout();
}
inline void pop_timeout() { } inline void pop_timeout() { }
...@@ -97,19 +95,7 @@ class nestable_invoke : public invoke_policy_base<nestable_invoke> { ...@@ -97,19 +95,7 @@ class nestable_invoke : public invoke_policy_base<nestable_invoke> {
inline bool waits_for_timeout(std::uint32_t) { return false; } inline bool waits_for_timeout(std::uint32_t) { return false; }
virtual mailbox_element* try_pop() { /*
return this->m_mailbox.try_pop();
}
mailbox_element* pop() {
wait_for_data();
return try_pop();
}
inline mailbox_element* try_pop(const timeout_type& abs_time) {
return (timed_wait_for_data(abs_time)) ? try_pop() : nullptr;
}
void run_detached() { void run_detached() {
auto dthis = util::dptr<Subtype>(this); auto dthis = util::dptr<Subtype>(this);
dthis->init(); dthis->init();
...@@ -120,87 +106,7 @@ class nestable_invoke : public invoke_policy_base<nestable_invoke> { ...@@ -120,87 +106,7 @@ class nestable_invoke : public invoke_policy_base<nestable_invoke> {
auto rsn = dthis->planned_exit_reason(); auto rsn = dthis->planned_exit_reason();
dthis->cleanup(rsn == exit_reason::not_exited ? exit_reason::normal : rsn); dthis->cleanup(rsn == exit_reason::not_exited ? exit_reason::normal : rsn);
} }
*/
inline void initialized(bool value) {
m_initialized = value;
}
bool initialized() const override {
return m_initialized;
}
protected:
typedef threaded combined_type;
void enqueue_impl(typename Base::mailbox_type& mbox,
const message_header& hdr,
any_tuple&& msg) {
auto ptr = this->new_mailbox_element(hdr, std::move(msg));
switch (mbox.enqueue(ptr)) {
case intrusive::first_enqueued: {
lock_type guard(m_mtx);
m_cv.notify_one();
break;
}
default: break;
case intrusive::queue_closed:
if (hdr.id.valid()) {
detail::sync_request_bouncer f{this->exit_reason()};
f(hdr.sender, hdr.id);
}
break;
}
}
void enqueue(const message_header& hdr, any_tuple msg) override {
enqueue_impl(this->m_mailbox, hdr, std::move(msg));
}
timeout_type init_timeout(const util::duration& rel_time) {
auto result = std::chrono::high_resolution_clock::now();
result += rel_time;
return result;
}
mailbox_element* await_message() {
return pop();
}
mailbox_element* await_message(const timeout_type& abs_time) {
return try_pop(abs_time);
}
virtual bool mailbox_empty() {
return this->m_mailbox.empty();
}
bool timed_wait_for_data(const timeout_type& abs_time) {
CPPA_REQUIRE(not this->m_mailbox.closed());
if (mailbox_empty()) {
lock_type guard(m_mtx);
while (mailbox_empty()) {
if (m_cv.wait_until(guard, abs_time) == std::cv_status::timeout) {
return false;
}
}
}
return true;
}
void wait_for_data() {
if (mailbox_empty()) {
lock_type guard(m_mtx);
while (mailbox_empty()) m_cv.wait(guard);
}
}
std::mutex m_mtx;
std::condition_variable m_cv;
private:
bool m_initialized;
}; };
......
...@@ -5,144 +5,41 @@ ...@@ -5,144 +5,41 @@
#include "cppa/policy/resume_policy.hpp" #include "cppa/policy/resume_policy.hpp"
namespace cppa {
namespace util {
struct fiber;
} // namespace util
} // namespace cppa
namespace cppa { namespace policy { namespace cppa { namespace policy {
// this policy simply forwards calls to @p await_data to the scheduling
// policy and throws an exception whenever @p resume is called;
// it intentionally works only with the no_scheduling policy
class no_resume { class no_resume {
public: public:
using timeout_type = std::chrono::high_resolution_clock::time_point; template<class Base>
struct mixin : Base {
template<class Actor> template<typename... Ts>
timeout_type init_timeout(Actor* self, const util::duration& rel_time) { mixin(Ts&&... args) : Base(std::forward<Ts>(args)...) { }
// request explicit timeout message
self->request_timeout(rel_time);
return 0; // return some dummy value
}
template<class Actor, typename F> inline resumable::resume_result resume(util::fiber*) {
void fetch_messages(Actor* self, F cb) { this->act();
auto e = self->m_mailbox.try_pop(); return resumable::done;
while (e) {
cb(e);
e = self->m_mailbox.try_pop();
} }
} };
template<class Actor, typename F>
inline void fetch_messages(Actor* self, F cb, timeout_type) {
// a call to this call is always preceded by init_timeout,
// which will trigger a timeout message
fetch_messages(self, cb);
}
template<class Actor, typename F> template<class Actor>
void try_fetch_messages(Actor* self, F cb) { void await_data(Actor* self) {
// try_fetch and fetch have the same semantics for event-based actors self->scheduling_policy().await_data(self);
fetch_messages(self, cb);
} }
template<class Actor> template<class Actor, typename AbsTimeout>
resume_result resume(Actor* self, util::fiber*) { bool await_data(Actor* self, const AbsTimeout& abs_time) {
CPPA_LOG_TRACE("id = " << self->id() return self->scheduling_policy().await_data(self, abs_time);
<< ", state = " << static_cast<int>(self->state()));
CPPA_REQUIRE( self->state() == actor_state::ready
|| self->state() == actor_state::pending);
auto done_cb = [&]() -> bool {
CPPA_LOG_TRACE("");
if (self->exit_reason() == self->exit_reason::not_exited) {
if (self->planned_exit_reason() == self->exit_reason::not_exited) {
self->planned_exit_reason(exit_reason::normal);
}
self->on_exit();
if (!m_bhvr_stack.empty()) {
planned_exit_reason(exit_reason::not_exited);
return false; // on_exit did set a new behavior
}
cleanup(planned_exit_reason());
}
set_state(actor_state::done);
m_bhvr_stack.clear();
m_bhvr_stack.cleanup();
on_exit();
CPPA_REQUIRE(next_job == nullptr);
return true;
};
CPPA_REQUIRE(next_job == nullptr);
try {
//auto e = m_mailbox.try_pop();
for (auto e = m_mailbox.try_pop(); ; e = m_mailbox.try_pop()) {
//e = m_mailbox.try_pop();
if (e == nullptr) {
CPPA_REQUIRE(next_job == nullptr);
CPPA_LOGMF(CPPA_DEBUG, self, "no more element in mailbox; going to block");
set_state(actor_state::about_to_block);
std::atomic_thread_fence(std::memory_order_seq_cst);
if (this->m_mailbox.can_fetch_more() == false) {
switch (compare_exchange_state(actor_state::about_to_block,
actor_state::blocked)) {
case actor_state::ready:
// interrupted by arriving message
// restore members
CPPA_REQUIRE(m_chained_actor == nullptr);
CPPA_LOGMF(CPPA_DEBUG, self, "switched back to ready: "
"interrupted by arriving message");
break;
case actor_state::blocked:
CPPA_LOGMF(CPPA_DEBUG, self, "set state successfully to blocked");
// done setting actor to blocked
return resume_result::actor_blocked;
default:
CPPA_LOGMF(CPPA_ERROR, self, "invalid state");
CPPA_CRITICAL("invalid state");
};
}
else {
CPPA_LOGMF(CPPA_DEBUG, self, "switched back to ready: "
"mailbox can fetch more");
set_state(actor_state::ready);
CPPA_REQUIRE(m_chained_actor == nullptr);
}
}
else {
CPPA_LOGMF(CPPA_DEBUG, self, "try to invoke message: " << to_string(e->msg));
if (m_bhvr_stack.invoke(m_recv_policy, this, e)) {
CPPA_LOG_DEBUG_IF(m_chained_actor,
"set actor with ID "
<< m_chained_actor->id()
<< " as successor");
if (m_bhvr_stack.empty() && done_cb()) {
CPPA_LOGMF(CPPA_DEBUG, self, "behavior stack empty");
return resume_result::actor_done;
}
m_bhvr_stack.cleanup();
}
}
}
}
catch (actor_exited& what) {
CPPA_LOG_INFO("actor died because of exception: actor_exited, "
"reason = " << what.reason());
if (exit_reason() == exit_reason::not_exited) {
quit(what.reason());
}
}
catch (std::exception& e) {
CPPA_LOG_WARNING("actor died because of exception: "
<< detail::demangle(typeid(e))
<< ", what() = " << e.what());
if (exit_reason() == exit_reason::not_exited) {
quit(exit_reason::unhandled_exception);
}
}
catch (...) {
CPPA_LOG_WARNING("actor died because of an unknown exception");
if (exit_reason() == exit_reason::not_exited) {
quit(exit_reason::unhandled_exception);
}
}
done_cb();
return resume_result::actor_done;
} }
}; };
......
...@@ -33,8 +33,14 @@ ...@@ -33,8 +33,14 @@
#include <mutex> #include <mutex>
#include <thread> #include <thread>
#include <chrono>
#include <condition_variable> #include <condition_variable>
#include "cppa/util/fiber.hpp"
#include "cppa/util/duration.hpp"
#include "cppa/policy/scheduling_policy.hpp"
#include "cppa/detail/sync_request_bouncer.hpp" #include "cppa/detail/sync_request_bouncer.hpp"
#include "cppa/intrusive/single_reader_queue.hpp" #include "cppa/intrusive/single_reader_queue.hpp"
...@@ -47,6 +53,43 @@ class no_scheduling { ...@@ -47,6 +53,43 @@ class no_scheduling {
public: public:
typedef std::chrono::high_resolution_clock::time_point timeout_type;
template<class Actor>
inline timeout_type init_timeout(Actor*, const util::duration& rel_time) {
auto result = std::chrono::high_resolution_clock::now();
result += rel_time;
return result;
}
template<class Actor, typename F>
bool fetch_messages(Actor* self, F cb) {
await_data(self);
fetch_messages_impl(self, cb);
}
template<class Actor, typename F>
bool try_fetch_messages(Actor* self, F cb) {
auto next = [&] { return self->mailbox().try_pop(); };
auto e = next();
if (!e) return false;
do {
cb(e);
e = next();
}
while (e);
return true;
}
template<class Actor, typename F>
timed_fetch_result fetch_messages(Actor* self, F cb, timeout_type abs_time) {
if (!await_data(self, abs_time)) {
return timed_fetch_result::no_message;
}
fetch_messages_impl(self, cb);
return timed_fetch_result::success;
}
template<class Actor> template<class Actor>
void enqueue(Actor* self, const message_header& hdr, any_tuple& msg) { void enqueue(Actor* self, const message_header& hdr, any_tuple& msg) {
auto ptr = self->new_mailbox_element(hdr, std::move(msg)); auto ptr = self->new_mailbox_element(hdr, std::move(msg));
...@@ -69,22 +112,29 @@ class no_scheduling { ...@@ -69,22 +112,29 @@ class no_scheduling {
template<class Actor> template<class Actor>
void launch(Actor* self) { void launch(Actor* self) {
std::thread([=] { std::thread([=] {
auto rr = resume_result::actor_blocked; util::fiber fself;
while (rr != resume_result::actor_done) { auto rr = resumable::resume_later;
wait_for_data(); while (rr != resumable::done) {
self->resume(); await_data(self);
self->resume(&fself);
} }
}).detach(); }).detach();
} }
private: template<class Actor>
void await_data(Actor* self) {
if (self->mailbox().empty()) {
lock_type guard(m_mtx);
while (self->mailbox().empty()) m_cv.wait(guard);
}
}
bool timed_wait_for_data(const timeout_type& abs_time) { template<class Actor>
CPPA_REQUIRE(not this->m_mailbox.closed()); bool await_data(Actor* self, const timeout_type& abs_time) {
if (mailbox_empty()) { CPPA_REQUIRE(!self->mailbox().closed());
if (self->mailbox().empty()) {
lock_type guard(m_mtx); lock_type guard(m_mtx);
while (mailbox_empty()) { while (self->mailbox().empty()) {
if (m_cv.wait_until(guard, abs_time) == std::cv_status::timeout) { if (m_cv.wait_until(guard, abs_time) == std::cv_status::timeout) {
return false; return false;
} }
...@@ -93,11 +143,15 @@ class no_scheduling { ...@@ -93,11 +143,15 @@ class no_scheduling {
return true; return true;
} }
void wait_for_data() { private:
if (mailbox_empty()) {
lock_type guard(m_mtx); template<class Actor, typename F>
while (mailbox_empty()) m_cv.wait(guard); void fetch_messages_impl(Actor* self, F cb) {
auto next = [&] { return self->mailbox().try_pop(); };
for (auto e = next(); e != nullptr; e = next()) {
cb(e);
} }
} }
std::mutex m_mtx; std::mutex m_mtx;
......
...@@ -28,24 +28,31 @@ ...@@ -28,24 +28,31 @@
\******************************************************************************/ \******************************************************************************/
#ifndef CPPA_SCHEDULED_ACTOR_DUMMY_HPP #ifndef NOT_PRIORITIZING_HPP
#define CPPA_SCHEDULED_ACTOR_DUMMY_HPP #define NOT_PRIORITIZING_HPP
#include "cppa/scheduled_actor.hpp" #include "cppa/mailbox_element.hpp"
namespace cppa { namespace detail { #include "cppa/policy/priority_policy.hpp"
struct scheduled_actor_dummy : scheduled_actor { namespace cppa {
scheduled_actor_dummy(); namespace policy {
void enqueue(const message_header&, any_tuple) override;
resume_result resume(util::fiber*) override; struct not_prioritizing {
void quit(std::uint32_t) override;
void dequeue(behavior&) override; template<class Actor>
void dequeue_response(behavior&, message_id) override; mailbox_element* next_message(Actor* self) {
bool has_behavior() override; return self->mailbox().try_pop();
scheduled_actor_type impl_type() override; }
template<class Actor, typename F>
bool fetch_messages(Actor*, F) {
//FIXME
}
}; };
} } // namespace cppa::detail } // namespace policy
} // namespace cppa
#endif // CPPA_SCHEDULED_ACTOR_DUMMY_HPP #endif // NOT_PRIORITIZING_HPP
...@@ -62,7 +62,7 @@ class priority_policy { ...@@ -62,7 +62,7 @@ class priority_policy {
* for re-scheduling. * for re-scheduling.
*/ */
template<class Actor, typename F> template<class Actor, typename F>
bool fetch_messages(Actor* self); bool fetch_messages(Actor* self, F cb);
}; };
......
...@@ -31,18 +31,15 @@ ...@@ -31,18 +31,15 @@
#ifndef CPPA_RESUME_POLICY_HPP #ifndef CPPA_RESUME_POLICY_HPP
#define CPPA_RESUME_POLICY_HPP #define CPPA_RESUME_POLICY_HPP
#include "cppa/resumable.hpp"
// this header consists all type definitions needed to // this header consists all type definitions needed to
// implement the resume_policy trait // implement the resume_policy trait
namespace cppa { namespace util { class duration; } } namespace cppa { namespace util { class duration; struct fiber; } }
namespace cppa { namespace policy { namespace cppa { namespace policy {
enum class resume_result {
actor_blocked,
actor_done
};
/** /**
* @brief The resume_policy <b>concept</b> class. Please note that this * @brief The resume_policy <b>concept</b> class. Please note that this
* class is <b>not</b> implemented. It only explains the all * class is <b>not</b> implemented. It only explains the all
...@@ -59,7 +56,7 @@ class resume_policy { ...@@ -59,7 +56,7 @@ class resume_policy {
* actor finishes execution. * actor finishes execution.
*/ */
template<class Actor> template<class Actor>
resume_result resume(Actor* self, util::fiber* from); resumable::resume_result resume(Actor* self, util::fiber* from);
/** /**
* @brief Waits unconditionally until a new message arrives. * @brief Waits unconditionally until a new message arrives.
......
...@@ -36,14 +36,14 @@ ...@@ -36,14 +36,14 @@
#include "cppa/util/duration.hpp" #include "cppa/util/duration.hpp"
#include "cppa/policy/invoke_policy_base.hpp" #include "cppa/policy/invoke_policy.hpp"
namespace cppa { namespace policy { namespace cppa { namespace policy {
/** /**
* @brief An actor that is scheduled or otherwise managed. * @brief An actor that is scheduled or otherwise managed.
*/ */
class sequential_invoke : public invoke_policy_base<sequential_invoke> { class sequential_invoke : public invoke_policy<sequential_invoke> {
public: public:
...@@ -53,26 +53,26 @@ class sequential_invoke : public invoke_policy_base<sequential_invoke> { ...@@ -53,26 +53,26 @@ class sequential_invoke : public invoke_policy_base<sequential_invoke> {
return node->marked; return node->marked;
} }
template<class Client> template<class Actor>
static inline pointer hm_begin(Client* client, pointer node) { inline pointer hm_begin(Actor* self, pointer node) {
auto previous = client->m_current_node; auto previous = self->m_current_node;
client->m_current_node = node; self->m_current_node = node;
client->push_timeout(); push_timeout();
node->marked = true; node->marked = true;
return previous; return previous;
} }
template<class Client> template<class Actor>
static inline void hm_cleanup(Client* client, pointer previous) { inline void hm_cleanup(Actor* self, pointer previous) {
client->m_current_node->marked = false; self->m_current_node->marked = false;
client->m_current_node = previous; self->m_current_node = previous;
} }
template<class Client> template<class Actor>
static inline void hm_revert(Client* client, pointer previous) { inline void hm_revert(Actor* self, pointer previous) {
client->m_current_node->marked = false; self->m_current_node->marked = false;
client->m_current_node = previous; self->m_current_node = previous;
client->pop_timeout(); pop_timeout();
} }
inline void reset_timeout() { inline void reset_timeout() {
...@@ -82,22 +82,24 @@ class sequential_invoke : public invoke_policy_base<sequential_invoke> { ...@@ -82,22 +82,24 @@ class sequential_invoke : public invoke_policy_base<sequential_invoke> {
} }
} }
void request_timeout(const util::duration& d) { template<class Actor>
void request_timeout(Actor* self, const util::duration& d) {
if (!d.valid()) m_has_pending_tout = false; if (!d.valid()) m_has_pending_tout = false;
else { else {
auto msg = make_any_tuple(atom("SYNC_TOUT"), ++m_pending_tout); auto msg = make_any_tuple(atom("SYNC_TOUT"), ++m_pending_tout);
if (d.is_zero()) { if (d.is_zero()) {
// immediately enqueue timeout message if duration == 0s // immediately enqueue timeout message if duration == 0s
this->enqueue({this->address(), this}, std::move(msg)); self->enqueue({self->address(), self}, std::move(msg));
//auto e = this->new_mailbox_element(this, std::move(msg)); //auto e = this->new_mailbox_element(this, std::move(msg));
//this->m_mailbox.enqueue(e); //this->m_mailbox.enqueue(e);
} }
else this->delayed_send_tuple(this, d, std::move(msg)); else self->delayed_send_tuple(self, d, std::move(msg));
m_has_pending_tout = true; m_has_pending_tout = true;
} }
} }
inline void handle_timeout(behavior& bhvr) { template<class Actor>
inline void handle_timeout(Actor*, behavior& bhvr) {
bhvr.handle_timeout(); bhvr.handle_timeout();
reset_timeout(); reset_timeout();
} }
......
...@@ -28,64 +28,32 @@ ...@@ -28,64 +28,32 @@
\******************************************************************************/ \******************************************************************************/
#ifndef CPPA_THREAD_BASED_ACTOR_HPP #ifndef CPPA_RESUMABLE_HPP
#define CPPA_THREAD_BASED_ACTOR_HPP #define CPPA_RESUMABLE_HPP
#include "cppa/config.hpp"
#include <map>
#include <list>
#include <mutex>
#include <stack>
#include <atomic>
#include <vector>
#include <memory>
#include <cstdint>
#include "cppa/atom.hpp"
#include "cppa/extend.hpp"
#include "cppa/stacked.hpp"
#include "cppa/threaded.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/mailbox_based.hpp"
#include "cppa/mailbox_element.hpp"
#include "cppa/detail/receive_policy.hpp"
namespace cppa { namespace cppa {
class self_type; namespace util {
class scheduler_helper; struct fiber;
} // namespace util
/**
* @brief An actor using the blocking API running in its own thread.
* @extends local_actor
*/
class thread_mapped_actor : public extend<local_actor, thread_mapped_actor>::
with<stacked, threaded> {
typedef combined_type super; struct resumable {
public: enum resume_result {
resume_later,
done
};
thread_mapped_actor(); // intrusive next pointer needed to use
// 'resumable' with 'single_reader_queue'
resumable* next;
thread_mapped_actor(std::function<void()> fun); virtual ~resumable();
inline void initialized(bool value) { m_initialized = value; } virtual resume_result resume(util::fiber*) = 0;
bool initialized() const override;
private:
bool m_initialized;
}; };
typedef intrusive_ptr<thread_mapped_actor> thread_mapped_actor_ptr;
} // namespace cppa } // namespace cppa
#endif // CPPA_THREAD_BASED_ACTOR_HPP #endif // CPPA_RESUMABLE_HPP
...@@ -35,7 +35,7 @@ ...@@ -35,7 +35,7 @@
#include <type_traits> #include <type_traits>
#include "cppa/util/dptr.hpp" #include "cppa/util/dptr.hpp"
#include "cppa/event_based_actor.hpp" #include "cppa/untyped_actor.hpp"
namespace cppa { namespace cppa {
...@@ -45,11 +45,11 @@ namespace cppa { ...@@ -45,11 +45,11 @@ namespace cppa {
* to initialize the derived actor with its @p init_state member. * to initialize the derived actor with its @p init_state member.
* @tparam Derived Direct subclass of @p sb_actor. * @tparam Derived Direct subclass of @p sb_actor.
*/ */
template<class Derived, class Base = event_based_actor> template<class Derived, class Base = untyped_actor>
class sb_actor : public Base { class sb_actor : public Base {
static_assert(std::is_base_of<event_based_actor, Base>::value, static_assert(std::is_base_of<untyped_actor, Base>::value,
"Base must be either event_based_actor or a derived type"); "Base must be either untyped_actor or a derived type");
protected: protected:
...@@ -58,11 +58,11 @@ class sb_actor : public Base { ...@@ -58,11 +58,11 @@ class sb_actor : public Base {
public: public:
/** /**
* @brief Overrides {@link event_based_actor::init()} and sets * @brief Overrides {@link untyped_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() override { behavior make_behavior() override {
become(util::dptr<Derived>(this)->init_state); return util::dptr<Derived>(this)->init_state;
} }
protected: protected:
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* 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 2.1 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_ACTOR_BEHAVIOR_HPP
#define CPPA_ACTOR_BEHAVIOR_HPP
#include <functional>
#include "cppa/config.hpp"
#include "cppa/extend.hpp"
#include "cppa/threadless.hpp"
#include "cppa/actor_state.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/mailbox_based.hpp"
#include "cppa/mailbox_element.hpp"
#include "cppa/message_priority.hpp"
#include "cppa/detail/memory.hpp"
#include "cppa/detail/receive_policy.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
namespace cppa {
class scheduler;
namespace util { struct fiber; }
enum scheduled_actor_type {
context_switching_impl, // enqueued to the job queue on startup
event_based_impl, // not enqueued to the job queue on startup
default_event_based_impl // scheduler enqueues a 'RUN' message on startup
};
class scheduled_actor;
/**
* @brief A base class for cooperatively scheduled actors.
* @extends local_actor
*/
class scheduled_actor : public extend<local_actor>::with<threadless> {
typedef combined_type super;
public:
~scheduled_actor();
/**
* @brief Intrusive next pointer needed by the scheduler's job queue.
*/
scheduled_actor* next;
/**
* @brief Continues execution of this actor.
* @param next_job Denotes the next actor to be resumed, should be
* set in case of chaining.
* @note This member function is called from the scheduler's worker threads.
*/
virtual resume_result resume(util::fiber* from) = 0;
/**
* @brief Called once by the scheduler after actor is initialized,
* but before @p resume is called.
*/
void attach_to_scheduler(scheduler* sched, bool hidden);
/**
* @brief Returns @p false if this actor can be disposed immediately
* after construction, @p true otherwise.
*/
virtual bool has_behavior() = 0;
/**
* @brief Denotes wheter this actor is enqueued to the scheduler's
* job queue on startup.
*/
virtual scheduled_actor_type impl_type() = 0;
/**
* @brief Returns @p true if this actor is ignored by
* {@link await_all_actors_done()}, false otherwise.
*/
inline bool is_hidden() const;
virtual void run_detached();
void enqueue(const message_header&, any_tuple) override;
protected:
scheduled_actor(actor_state init_state, bool enable_chained_send);
void cleanup(std::uint32_t reason) override;
typedef intrusive::single_reader_queue<mailbox_element, detail::disposer>
mailbox_type;
actor_state compare_exchange_state(actor_state expected, actor_state desired);
inline void set_state(actor_state new_value) {
m_state.store(new_value);
}
inline actor_state state() const {
return m_state;
}
bool initialized() const;
private:
bool enqueue_impl(actor_state, const message_header&, any_tuple&&);
std::atomic<actor_state> m_state;
protected:
scheduler* m_scheduler;
bool m_hidden;
};
typedef intrusive_ptr<scheduled_actor> scheduled_actor_ptr;
inline bool scheduled_actor::is_hidden() const {
return m_hidden;
}
} // namespace cppa
#endif // CPPA_ACTOR_BEHAVIOR_HPP
...@@ -41,8 +41,8 @@ ...@@ -41,8 +41,8 @@
#include "cppa/actor.hpp" #include "cppa/actor.hpp"
#include "cppa/channel.hpp" #include "cppa/channel.hpp"
#include "cppa/cow_tuple.hpp" #include "cppa/cow_tuple.hpp"
#include "cppa/resumable.hpp"
#include "cppa/attachable.hpp" #include "cppa/attachable.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/spawn_options.hpp" #include "cppa/spawn_options.hpp"
#include "cppa/util/duration.hpp" #include "cppa/util/duration.hpp"
...@@ -52,7 +52,7 @@ namespace cppa { ...@@ -52,7 +52,7 @@ namespace cppa {
class untyped_actor; class untyped_actor;
class scheduled_actor; class scheduled_actor;
class scheduler_helper; class scheduler_helper;
class event_based_actor; class untyped_actor;
typedef intrusive_ptr<scheduled_actor> scheduled_actor_ptr; typedef intrusive_ptr<scheduled_actor> scheduled_actor_ptr;
namespace detail { class singleton_manager; } // namespace detail namespace detail { class singleton_manager; } // namespace detail
...@@ -83,12 +83,9 @@ class scheduler { ...@@ -83,12 +83,9 @@ class scheduler {
public: public:
typedef std::function<void(local_actor*)> init_callback; actor printer() const;
typedef std::function<behavior(untyped_actor*)> actor_fun;
const actor& printer() const;
virtual void enqueue(scheduled_actor*) = 0; virtual void enqueue(resumable*) = 0;
/** /**
* @brief Informs the scheduler about a converted context * @brief Informs the scheduler about a converted context
...@@ -128,33 +125,13 @@ class scheduler { ...@@ -128,33 +125,13 @@ class scheduler {
//TODO: delayed_send_helper()->enqueue(nullptr, std::move(tup)); //TODO: delayed_send_helper()->enqueue(nullptr, std::move(tup));
} }
/**
* @brief Executes @p ptr in this scheduler.
*/
virtual local_actor_ptr exec(spawn_options opts,
scheduled_actor_ptr ptr) = 0;
/**
* @brief Creates a new actor from @p actor_behavior and executes it
* in this scheduler.
*/
virtual local_actor_ptr exec(spawn_options opts,
init_callback init_cb,
actor_fun actor_behavior) = 0;
template<typename F, typename T, typename... Ts>
local_actor_ptr exec(spawn_options opts, init_callback cb,
F f, T&& a0, Ts&&... as) {
return this->exec(opts, cb, std::bind(f, std::forward<T>(a0), std::forward<Ts>(as)...));
}
private: private:
static scheduler* create_singleton(); static scheduler* create_singleton();
inline void dispose() { delete this; } inline void dispose() { delete this; }
actor& delayed_send_helper(); actor delayed_send_helper();
scheduler_helper* m_helper; scheduler_helper* m_helper;
......
...@@ -25,131 +25,55 @@ ...@@ -25,131 +25,55 @@
* * * *
* You should have received a copy of the GNU Lesser General Public License * * You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. * * along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/ \******************************************************************************/
#ifndef CPPA_SCOPED_ACTOR_HPP #ifndef CPPA_SCOPED_ACTOR_HPP
#define CPPA_SCOPED_ACTOR_HPP #define CPPA_SCOPED_ACTOR_HPP
#include "cppa/behavior.hpp" #include "cppa/behavior.hpp"
#include "cppa/thread_mapped_actor.hpp" #include "cppa/blocking_untyped_actor.hpp"
namespace cppa { namespace cppa {
class scoped_actor { class scoped_actor {
public: public:
/** scoped_actor();
* @brief Dequeues the next message from the mailbox that
* is matched by given behavior. ~scoped_actor();
*/
template<typename... Ts> inline blocking_untyped_actor* operator->() const {
void receive(Ts&&... args); return m_self.get();
}
/**
* @brief Receives messages in an endless loop. inline blocking_untyped_actor& operator*() const {
* Semantically equal to: <tt>for (;;) { receive(...); }</tt> return *m_self;
*/ }
template<typename... Ts>
void receive_loop(Ts&&... args); inline blocking_untyped_actor* get() const {
return m_self.get();
/** }
* @brief Receives messages as in a range-based loop.
* operator channel() const {
* Semantically equal to: return get();
* <tt>for ( ; begin != end; ++begin) { receive(...); }</tt>. }
*
* <b>Usage example:</b> operator actor() const {
* @code return get();
* int i = 0; }
* receive_for(i, 10) (
* on(atom("get")) >> [&]() -> any_tuple { return {"result", i}; } operator actor_addr() const {
* ); return get()->address();
* @endcode }
* @param begin First value in range.
* @param end Last value in range (excluded). private:
* @returns A functor implementing the loop.
*/ intrusive_ptr<blocking_untyped_actor> m_self;
template<typename T>
detail::receive_for_helper<T> receive_for(T& begin, const T& end);
/**
* @brief Receives messages as long as @p stmt returns true.
*
* Semantically equal to: <tt>while (stmt()) { receive(...); }</tt>.
*
* <b>Usage example:</b>
* @code
* int i = 0;
* receive_while([&]() { return (++i <= 10); })
* (
* on<int>() >> int_fun,
* on<float>() >> float_fun
* );
* @endcode
* @param stmt Lambda expression, functor or function returning a @c bool.
* @returns A functor implementing the loop.
*/
template<typename Statement>
detail::receive_while_helper<Statement> receive_while(Statement&& stmt);
/**
* @brief Receives messages until @p stmt returns true.
*
* Semantically equal to: <tt>do { receive(...); } while (stmt() == false);</tt>
*
* <b>Usage example:</b>
* @code
* int i = 0;
* do_receive
* (
* on<int>() >> int_fun,
* on<float>() >> float_fun
* )
* .until([&]() { return (++i >= 10); };
* @endcode
* @param bhvr Denotes the actor's response the next incoming message.
* @returns A functor providing the @c until member function.
*/
template<typename... Ts>
detail::do_receive_helper do_receive(Ts&&... args);
private:
intrusive_ptr<thread_mapped_actor> m_self;
}; };
/******************************************************************************
* inline and template function implementations *
******************************************************************************/
template<typename... Ts>
void scoped_actor::receive(Ts&&... args) {
m_self->receive(std::forward<Ts(args)...);
}
template<typename... Ts>
void scoped_actor::receive_loop(Ts&&... args) {
m_self->receive_loop(std::forward<Ts(args)...);
}
template<typename T>
detail::receive_for_helper<T> scoped_actor::receive_for(T& begin, const T& end) {
m_self->receive_for(begin, end);
}
template<typename Statement>
detail::receive_while_helper<Statement> scoped_actor::receive_while(Statement&& stmt) {
m_self->receive_while(std::forward<Statement(stmt));
}
template<typename... Ts>
detail::do_receive_helper scoped_actor::do_receive(Ts&&... args) {
m_self->do_receive(std::forward<Ts(args)...);
}
} // namespace cppa } // namespace cppa
#endif // CPPA_SCOPED_ACTOR_HPP #endif // CPPA_SCOPED_ACTOR_HPP
...@@ -39,17 +39,11 @@ ...@@ -39,17 +39,11 @@
#include "cppa/spawn_options.hpp" #include "cppa/spawn_options.hpp"
#include "cppa/detail/proper_actor.hpp" #include "cppa/detail/proper_actor.hpp"
#include "cppa/detail/functor_based_actor.hpp"
#include "cppa/detail/functor_based_blocking_actor.hpp"
namespace cppa { namespace cppa {
/** @cond PRIVATE */
constexpr bool unbound_spawn_options(spawn_options opts) {
return !has_monitor_flag(opts) && !has_link_flag(opts);
}
/** @endcond */
/** /**
* @ingroup ActorCreation * @ingroup ActorCreation
* @{ * @{
...@@ -58,17 +52,21 @@ constexpr bool unbound_spawn_options(spawn_options opts) { ...@@ -58,17 +52,21 @@ constexpr bool unbound_spawn_options(spawn_options opts) {
/** /**
* @brief Spawns an actor of type @p Impl. * @brief Spawns an actor of type @p Impl.
* @param args Constructor arguments. * @param args Constructor arguments.
* @tparam Impl Subtype of {@link event_based_actor} or {@link sb_actor}. * @tparam Impl Subtype of {@link untyped_actor} or {@link sb_actor}.
* @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior. * @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior.
* @returns An {@link actor} to the spawned {@link actor}. * @returns An {@link actor} to the spawned {@link actor}.
*/ */
template<class Impl, spawn_options Options = no_spawn_options, typename... Ts> template<class Impl, spawn_options Options, typename... Ts>
actor spawn(Ts&&... args) { actor spawn(Ts&&... args) {
static_assert(std::is_base_of<event_based_actor, Impl>::value, static_assert(std::is_base_of<untyped_actor, Impl>::value ||
"Impl is not a derived type of event_based_actor"); (std::is_base_of<blocking_untyped_actor, Impl>::value &&
static_assert(unbound_spawn_options(Options), has_blocking_api_flag(Options)),
"Impl is not a derived type of untyped_actor or "
"is a derived type of blocking_untyped_actor but "
"blocking_api_flag is missing");
static_assert(is_unbound(Options),
"top-level spawns cannot have monitor or link flag"); "top-level spawns cannot have monitor or link flag");
static_assert(unbound_spawn_options(Options), static_assert(is_unbound(Options),
"top-level spawns cannot have monitor or link flag"); "top-level spawns cannot have monitor or link flag");
using scheduling_policy = typename std::conditional< using scheduling_policy = typename std::conditional<
has_detach_flag(Options), has_detach_flag(Options),
...@@ -100,6 +98,8 @@ actor spawn(Ts&&... args) { ...@@ -100,6 +98,8 @@ actor spawn(Ts&&... args) {
resume_policy, resume_policy,
invoke_policy>; invoke_policy>;
auto ptr = make_counted<proper_impl>(std::forward<Ts>(args)...); auto ptr = make_counted<proper_impl>(std::forward<Ts>(args)...);
ptr->launch();
return ptr;
/* /*
scheduled_actor_ptr ptr; scheduled_actor_ptr ptr;
if (has_priority_aware_flag(Options)) { if (has_priority_aware_flag(Options)) {
...@@ -121,7 +121,8 @@ actor spawn(Ts&&... args) { ...@@ -121,7 +121,8 @@ actor spawn(Ts&&... args) {
* @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior. * @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior.
* @returns An {@link actor} to the spawned {@link actor}. * @returns An {@link actor} to the spawned {@link actor}.
*/ */
template<spawn_options Options = no_spawn_options, typename... Ts> //template<spawn_options Options = no_spawn_options, typename... Ts>
template<spawn_options Options, typename... Ts>
actor spawn(Ts&&... args) { actor spawn(Ts&&... args) {
static_assert(sizeof...(Ts) > 0, "too few arguments provided"); static_assert(sizeof...(Ts) > 0, "too few arguments provided");
using base_class = typename std::conditional< using base_class = typename std::conditional<
...@@ -129,13 +130,7 @@ actor spawn(Ts&&... args) { ...@@ -129,13 +130,7 @@ actor spawn(Ts&&... args) {
detail::functor_based_blocking_actor, detail::functor_based_blocking_actor,
detail::functor_based_actor detail::functor_based_actor
>::type; >::type;
return spawn<base_class>(std::forward<Ts>(args)...); return spawn<base_class, Options>(std::forward<Ts>(args)...);
using impl = detail::proper_actor<untyped_actor,
scheduling_policy,
priority_policy,
resume_policy,
invoke_policy>;
return make_counted<impl>();
} }
/** /**
...@@ -163,7 +158,7 @@ actor spawn_in_group(const group_ptr& grp, Ts&&... args) { ...@@ -163,7 +158,7 @@ actor spawn_in_group(const group_ptr& grp, Ts&&... args) {
/** /**
* @brief Spawns an actor of type @p Impl that immediately joins @p grp. * @brief Spawns an actor of type @p Impl that immediately joins @p grp.
* @param args Constructor arguments. * @param args Constructor arguments.
* @tparam Impl Subtype of {@link event_based_actor} or {@link sb_actor}. * @tparam Impl Subtype of {@link untyped_actor} or {@link sb_actor}.
* @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior. * @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior.
* @returns An {@link actor} to the spawned {@link actor}. * @returns An {@link actor} to the spawned {@link actor}.
* @note The spawned has joined the group before this function returns. * @note The spawned has joined the group before this function returns.
......
...@@ -173,6 +173,21 @@ constexpr bool has_blocking_api_flag(spawn_options opts) { ...@@ -173,6 +173,21 @@ constexpr bool has_blocking_api_flag(spawn_options opts) {
/** @} */ /** @} */
/** @cond PRIVATE */
constexpr bool is_unbound(spawn_options opts) {
return !has_monitor_flag(opts) && !has_link_flag(opts);
}
constexpr spawn_options make_unbound(spawn_options opts) {
return static_cast<spawn_options>(
(static_cast<int>(opts)
& ~static_cast<int>(spawn_options::link_flag))
& ~static_cast<int>(spawn_options::monitor_flag));
}
/** @endcond */
} // namespace cppa } // namespace cppa
#endif // CPPA_SPAWN_OPTIONS_HPP #endif // CPPA_SPAWN_OPTIONS_HPP
...@@ -72,6 +72,10 @@ inline std::string to_string(const actor& what) { ...@@ -72,6 +72,10 @@ inline std::string to_string(const actor& what) {
return detail::to_string_impl(what); return detail::to_string_impl(what);
} }
inline std::string to_string(const actor_addr& what) {
return detail::to_string_impl(what);
}
inline std::string to_string(const group_ptr& what) { inline std::string to_string(const group_ptr& what) {
return detail::to_string_impl(what); return detail::to_string_impl(what);
} }
......
...@@ -25,7 +25,7 @@ ...@@ -25,7 +25,7 @@
* * * *
* You should have received a copy of the GNU Lesser General Public License * * You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. * * along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/ \******************************************************************************/
#ifndef CPPA_TYPED_ACTOR_HPP #ifndef CPPA_TYPED_ACTOR_HPP
......
...@@ -34,7 +34,7 @@ ...@@ -34,7 +34,7 @@
#include "cppa/replies_to.hpp" #include "cppa/replies_to.hpp"
#include "cppa/typed_behavior.hpp" #include "cppa/typed_behavior.hpp"
#include "cppa/message_future.hpp" #include "cppa/message_future.hpp"
#include "cppa/event_based_actor.hpp" #include "cppa/untyped_actor.hpp"
#include "cppa/detail/typed_actor_util.hpp" #include "cppa/detail/typed_actor_util.hpp"
...@@ -44,7 +44,7 @@ template<typename... Signatures> ...@@ -44,7 +44,7 @@ template<typename... Signatures>
class typed_actor_ptr; class typed_actor_ptr;
template<typename... Signatures> template<typename... Signatures>
class typed_actor : public event_based_actor { class typed_actor : public untyped_actor {
public: public:
......
...@@ -25,103 +25,30 @@ ...@@ -25,103 +25,30 @@
* * * *
* You should have received a copy of the GNU Lesser General Public License * * You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. * * along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/ \******************************************************************************/
#ifndef CPPA_UNTYPED_ACTOR_HPP #ifndef CPPA_UNTYPED_ACTOR_HPP
#define CPPA_UNTYPED_ACTOR_HPP #define CPPA_UNTYPED_ACTOR_HPP
#include "cppa/extend.hpp" #include "cppa/extend.hpp"
#include "cppa/stackless.hpp"
#include "cppa/local_actor.hpp" #include "cppa/local_actor.hpp"
#include "cppa/mailbox_based.hpp" #include "cppa/mailbox_based.hpp"
#include "cppa/behavior_stack_based.hpp"
namespace cppa { namespace cppa {
/** /**
* @extends local_actor * @extends local_actor
*/ */
class untyped_actor : public local_actor { class untyped_actor : public extend<local_actor>::with<mailbox_based,
behavior_stack_based> {
bool has_behavior() {
return this->m_bhvr_stack.empty() == false;
}
void unbecome() {
this->m_bhvr_stack.pop_async_back();
}
/**
* @brief Sets the actor's behavior and discards the previous behavior
* unless {@link keep_behavior} is given as first argument.
*/
template<typename T, typename... Ts>
inline typename std::enable_if<
!is_behavior_policy<typename util::rm_const_and_ref<T>::type>::value,
void
>::type
become(T&& arg, Ts&&... args) {
do_become(match_expr_convert(std::forward<T>(arg),
std::forward<Ts>(args)...),
true);
}
template<bool Discard, typename... Ts>
inline void become(behavior_policy<Discard>, Ts&&... args) {
do_become(match_expr_convert(std::forward<Ts>(args)...), Discard);
}
void become_waiting_for(behavior bhvr, message_id mf) {
if (bhvr.timeout().valid()) {
this->reset_timeout();
this->request_timeout(bhvr.timeout());
}
this->m_bhvr_stack.push_back(std::move(bhvr), mf);
}
void do_become(behavior&& bhvr, bool discard_old) {
this->reset_timeout();
this->request_timeout(bhvr.timeout());
if (discard_old) this->m_bhvr_stack.pop_async_back();
this->m_bhvr_stack.push_back(std::move(bhvr));
}
inline bool has_behavior() const {
return this->m_bhvr_stack.empty() == false;
}
inline behavior& get_behavior() {
CPPA_REQUIRE(this->m_bhvr_stack.empty() == false);
return this->m_bhvr_stack.back();
}
inline void handle_timeout(behavior& bhvr) {
CPPA_REQUIRE(bhvr.timeout().valid());
this->reset_timeout();
bhvr.handle_timeout();
if (this->m_bhvr_stack.empty() == false) {
this->request_timeout(get_behavior().timeout());
}
}
void exec_bhvr_stack() {
while (!m_bhvr_stack.empty()) {
m_bhvr_stack.exec(m_recv_policy, util::dptr<Subtype>(this));
}
}
inline detail::behavior_stack& bhvr_stack() {
return m_bhvr_stack;
}
inline optional<behavior&> sync_handler(message_id msg_id) {
return m_bhvr_stack.sync_handler(msg_id);
}
protected: protected:
// allows actors to keep previous behaviors and enables unbecome() virtual behavior make_behavior() = 0;
detail::behavior_stack m_bhvr_stack;
void forward_to(const actor& other);
}; };
......
This diff is collapsed.
...@@ -5,11 +5,11 @@ ...@@ -5,11 +5,11 @@
using namespace std; using namespace std;
using namespace cppa; using namespace cppa;
void mirror() { behavior mirror(untyped_actor* self) {
// wait for messages // wait for messages
become ( return (
// invoke this lambda expression if we receive a string // invoke this lambda expression if we receive a string
on_arg_match >> [](const string& what) -> string { on_arg_match >> [=](const string& what) -> string {
// prints "Hello World!" via aout (thread-safe cout wrapper) // prints "Hello World!" via aout (thread-safe cout wrapper)
aout << what << endl; aout << what << endl;
// terminates this actor ('become' otherwise loops forever) // terminates this actor ('become' otherwise loops forever)
...@@ -20,11 +20,11 @@ void mirror() { ...@@ -20,11 +20,11 @@ void mirror() {
); );
} }
void hello_world(const actor_ptr& buddy) { void hello_world(untyped_actor* self, const actor& buddy) {
// send "Hello World!" to our buddy ... // send "Hello World!" to our buddy ...
sync_send(buddy, "Hello World!").then( self->sync_send(buddy, "Hello World!").then(
// ... and wait for a response // ... and wait for a response
on_arg_match >> [](const string& what) { on_arg_match >> [=](const string& what) {
// prints "!dlroW olleH" // prints "!dlroW olleH"
aout << what << endl; aout << what << endl;
} }
......
...@@ -13,9 +13,9 @@ using std::endl; ...@@ -13,9 +13,9 @@ using std::endl;
using namespace cppa; using namespace cppa;
// implementation using the blocking API // implementation using the blocking API
void blocking_math_fun() { void blocking_math_fun(blocking_untyped_actor* self) {
bool done = false; bool done = false;
do_receive ( self->do_receive (
// "arg_match" matches the parameter types of given lambda expression // "arg_match" matches the parameter types of given lambda expression
// thus, it's equal to // thus, it's equal to
// - on<atom("plus"), int, int>() // - on<atom("plus"), int, int>()
...@@ -34,38 +34,38 @@ void blocking_math_fun() { ...@@ -34,38 +34,38 @@ void blocking_math_fun() {
).until(gref(done)); ).until(gref(done));
} }
void calculator() { void calculator(untyped_actor* self) {
// execute this behavior until actor terminates // execute this behavior until actor terminates
become ( self->become (
on(atom("plus"), arg_match) >> [](int a, int b) { on(atom("plus"), arg_match) >> [](int a, int b) {
return make_cow_tuple(atom("result"), a + b); return make_cow_tuple(atom("result"), a + b);
}, },
on(atom("minus"), arg_match) >> [](int a, int b) { on(atom("minus"), arg_match) >> [](int a, int b) {
return make_cow_tuple(atom("result"), a - b); return make_cow_tuple(atom("result"), a - b);
}, },
on(atom("quit")) >> [] { on(atom("quit")) >> [=] {
// terminate actor with normal exit reason // terminate actor with normal exit reason
self->quit(); self->quit();
} }
); );
} }
void tester(const actor_ptr& testee) { void tester(untyped_actor* self, const actor& testee) {
self->link_to(testee); self->link_to(testee);
// will be invoked if we receive an unexpected response message // will be invoked if we receive an unexpected response message
self->on_sync_failure([] { self->on_sync_failure([=] {
aout << "AUT (actor under test) failed" << endl; aout << "AUT (actor under test) failed" << endl;
self->quit(exit_reason::user_shutdown); self->quit(exit_reason::user_shutdown);
}); });
// first test: 2 + 1 = 3 // first test: 2 + 1 = 3
sync_send(testee, atom("plus"), 2, 1).then( self->sync_send(testee, atom("plus"), 2, 1).then(
on(atom("result"), 3) >> [=] { on(atom("result"), 3) >> [=] {
// second test: 2 - 1 = 1 // second test: 2 - 1 = 1
sync_send(testee, atom("minus"), 2, 1).then( self->sync_send(testee, atom("minus"), 2, 1).then(
on(atom("result"), 1) >> [=] { on(atom("result"), 1) >> [=] {
// both tests succeeded // both tests succeeded
aout << "AUT (actor under test) seems to be ok" << endl; aout << "AUT (actor under test) seems to be ok" << endl;
send(testee, atom("quit")); self->send(testee, atom("quit"));
} }
); );
} }
...@@ -78,4 +78,3 @@ int main() { ...@@ -78,4 +78,3 @@ int main() {
shutdown(); shutdown();
return 0; return 0;
} }
...@@ -53,20 +53,20 @@ void draw_kirby(const animation_step& animation) { ...@@ -53,20 +53,20 @@ void draw_kirby(const animation_step& animation) {
} }
// uses a message-based loop to iterate over all animation steps // uses a message-based loop to iterate over all animation steps
void dancing_kirby() { void dancing_kirby(untyped_actor* self) {
// let's get it started // let's get it started
send(self, atom("Step"), size_t{0}); self->send(self, atom("Step"), size_t{0});
become ( self->become (
on(atom("Step"), array_size(animation_steps)) >> [] { on(atom("Step"), array_size(animation_steps)) >> [=] {
// we've printed all animation steps (done) // we've printed all animation steps (done)
cout << endl; cout << endl;
self->quit(); self->quit();
}, },
on(atom("Step"), arg_match) >> [](size_t step) { on(atom("Step"), arg_match) >> [=](size_t step) {
// print given step // print given step
draw_kirby(animation_steps[step]); draw_kirby(animation_steps[step]);
// animate next step in 150ms // animate next step in 150ms
delayed_send(self, std::chrono::milliseconds(150), atom("Step"), step + 1); self->delayed_send(self, std::chrono::milliseconds(150), atom("Step"), step + 1);
} }
); );
} }
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
...@@ -25,7 +25,7 @@ ...@@ -25,7 +25,7 @@
* * * *
* You should have received a copy of the GNU Lesser General Public License * * You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. * * along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/ \******************************************************************************/
#include <utility> #include <utility>
...@@ -39,6 +39,8 @@ namespace cppa { ...@@ -39,6 +39,8 @@ namespace cppa {
actor::actor(abstract_actor* ptr) : m_ops(ptr) { } actor::actor(abstract_actor* ptr) : m_ops(ptr) { }
actor::actor(const invalid_actor_t&) : m_ops(nullptr) { }
void actor::enqueue(const message_header& hdr, any_tuple msg) const { void actor::enqueue(const message_header& hdr, any_tuple msg) const {
if (m_ops.m_ptr) m_ops.m_ptr->enqueue(hdr, std::move(msg)); if (m_ops.m_ptr) m_ops.m_ptr->enqueue(hdr, std::move(msg));
} }
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
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