Commit b0dc9d4d authored by Dominik Charousset's avatar Dominik Charousset

preparations to replace mixin with policy design

this patch makes the first steps for replace the existing mixin-based design
with a more clean and simpler policy-based design for actors
parent 3b8cea71
......@@ -140,7 +140,7 @@ set(LIBCPPA_SRC
src/buffer.cpp
src/channel.cpp
src/common_actor_ops.cpp
src/context_switching_actor.cpp
src/context_switching_resume.cpp
src/continuable.cpp
src/decorated_tuple.cpp
src/actor_proxy.cpp
......
cppa/abstract_actor.hpp
cppa/abstract_channel.hpp
cppa/actor.hpp
cppa/actor_addressing.hpp
cppa/actor_companion_mixin.hpp
cppa/actor_addr.hpp
cppa/actor_namespace.hpp
cppa/actor_proxy.hpp
cppa/actor_state.hpp
cppa/announce.hpp
cppa/any_tuple.hpp
cppa/anything.hpp
......@@ -10,9 +13,12 @@ cppa/attachable.hpp
cppa/behavior.hpp
cppa/binary_deserializer.hpp
cppa/binary_serializer.hpp
cppa/blocking_actor.hpp
cppa/blocking_untyped_actor.hpp
cppa/channel.hpp
cppa/common_actor_ops.hpp
cppa/config.hpp
cppa/context_switching_actor.hpp
cppa/policy/context_switching_resume.hpp
cppa/cow_ptr.hpp
cppa/cow_tuple.hpp
cppa/cppa.hpp
......@@ -30,9 +36,7 @@ cppa/detail/default_uniform_type_info_impl.hpp
cppa/detail/demangle.hpp
cppa/detail/disablable_delete.hpp
cppa/detail/empty_tuple.hpp
cppa/detail/event_based_actor_factory.hpp
cppa/detail/fd_util.hpp
cppa/detail/fwd.hpp
cppa/detail/get_behavior.hpp
cppa/detail/group_manager.hpp
cppa/detail/handle.hpp
......@@ -48,8 +52,7 @@ cppa/detail/pair_member.hpp
cppa/detail/projection.hpp
cppa/detail/pseudo_tuple.hpp
cppa/detail/ptype_to_type.hpp
cppa/detail/receive_loop_helper.hpp
cppa/detail/receive_policy.hpp
cppa/detail/raw_access.hpp
cppa/detail/scheduled_actor_dummy.hpp
cppa/detail/serialize_tuple.hpp
cppa/detail/singleton_manager.hpp
......@@ -71,11 +74,10 @@ cppa/detail/uniform_type_info_map.hpp
cppa/detail/value_guard.hpp
cppa/detail/yield_interface.hpp
cppa/enable_weak_ptr.hpp
cppa/event_based_actor.hpp
cppa/policy/event_based_resume.hpp
cppa/exception.hpp
cppa/exit_reason.hpp
cppa/extend.hpp
cppa/factory.hpp
cppa/from_string.hpp
cppa/get.hpp
cppa/group.hpp
......@@ -89,12 +91,7 @@ cppa/io/broker.hpp
cppa/io/buffered_writing.hpp
cppa/io/connection_handle.hpp
cppa/io/continuable.hpp
cppa/io/default_actor_addressing.hpp
cppa/io/default_actor_proxy.hpp
cppa/io/default_message_queue.hpp
cppa/io/default_peer.hpp
cppa/io/default_peer_acceptor.hpp
cppa/io/default_protocol.hpp
cppa/io/event.hpp
cppa/io/input_stream.hpp
cppa/io/ipv4_acceptor.hpp
......@@ -102,7 +99,9 @@ cppa/io/ipv4_io_stream.hpp
cppa/io/middleman.hpp
cppa/io/middleman_event_handler.hpp
cppa/io/output_stream.hpp
cppa/io/protocol.hpp
cppa/io/peer.hpp
cppa/io/peer_acceptor.hpp
cppa/io/remote_actor_proxy.hpp
cppa/io/stream.hpp
cppa/local_actor.hpp
cppa/logging.hpp
......@@ -117,6 +116,8 @@ cppa/message_future.hpp
cppa/message_header.hpp
cppa/message_id.hpp
cppa/message_priority.hpp
cppa/node_id.hpp
cppa/nonblocking_actor.hpp
cppa/none.hpp
cppa/object.hpp
cppa/on.hpp
......@@ -132,39 +133,38 @@ cppa/opt.hpp
cppa/optional.hpp
cppa/optional_variant.hpp
cppa/partial_function.hpp
cppa/policy/invoke_policy_base.hpp
cppa/policy/nestable_invoke.hpp
cppa/policy/prioritizing.hpp
cppa/policy/sequential_invoke.hpp
cppa/primitive_type.hpp
cppa/primitive_variant.hpp
cppa/prioritizing.hpp
cppa/process_information.hpp
cppa/qtsupport/actor_widget_mixin.hpp
cppa/receive.hpp
cppa/ref_counted.hpp
cppa/replies_to.hpp
cppa/response_handle.hpp
cppa/sb_actor.hpp
cppa/scheduled_actor.hpp
cppa/scheduler.hpp
cppa/self.hpp
cppa/scoped_actor.hpp
cppa/send.hpp
cppa/serializer.hpp
cppa/singletons.hpp
cppa/spawn.hpp
cppa/spawn_options.hpp
cppa/stacked.hpp
cppa/stackless.hpp
cppa/thread_mapped_actor.hpp
cppa/threaded.hpp
cppa/threadless.hpp
cppa/timeout_definition.hpp
cppa/to_string.hpp
cppa/tpartial_function.hpp
cppa/tuple_cast.hpp
cppa/type_lookup_table.hpp
cppa/typed_actor.hpp
cppa/typed_actor_base.hpp
cppa/typed_actor_ptr.hpp
cppa/typed_behavior.hpp
cppa/uniform_type_info.hpp
cppa/unit.hpp
cppa/untyped_actor.hpp
cppa/util/abstract_uniform_type_info.hpp
cppa/util/algorithm.hpp
cppa/util/arg_match_t.hpp
......@@ -177,6 +177,7 @@ cppa/util/duration.hpp
cppa/util/fiber.hpp
cppa/util/get_mac_addresses.hpp
cppa/util/get_root_uuid.hpp
cppa/util/guard.hpp
cppa/util/int_list.hpp
cppa/util/left_or_right.hpp
cppa/util/limited_vector.hpp
......@@ -207,21 +208,22 @@ examples/opencl/proper_matrix.cpp
examples/opencl/simple_matrix.cpp
examples/qtsupport/chatwidget.cpp
examples/qtsupport/chatwidget.hpp
examples/qtsupport/chatwindow.ui
examples/qtsupport/qt_group_chat.cpp
examples/remote_actors/distributed_calculator.cpp
examples/remote_actors/group_chat.cpp
examples/remote_actors/group_server.cpp
examples/remote_actors/protobuf.scala
examples/remote_actors/protobuf_broker.cpp
examples/type_system/announce_1.cpp
examples/type_system/announce_2.cpp
examples/type_system/announce_3.cpp
examples/type_system/announce_4.cpp
examples/type_system/announce_5.cpp
src/abstract_actor.cpp
src/abstract_channel.cpp
src/abstract_tuple.cpp
src/actor.cpp
src/actor_addressing.cpp
src/actor_addr.cpp
src/actor_namespace.cpp
src/actor_proxy.cpp
src/actor_registry.cpp
src/algorithm.cpp
......@@ -235,19 +237,15 @@ src/binary_serializer.cpp
src/broker.cpp
src/buffer.cpp
src/channel.cpp
src/context_switching_actor.cpp
src/common_actor_ops.cpp
src/context_switching_resume.cpp
src/continuable.cpp
src/decorated_tuple.cpp
src/default_actor_addressing.cpp
src/default_actor_proxy.cpp
src/default_peer.cpp
src/default_peer_acceptor.cpp
src/default_protocol.cpp
src/demangle.cpp
src/deserializer.cpp
src/duration.cpp
src/empty_tuple.cpp
src/event_based_actor.cpp
src/event_based_resume.cpp
src/exception.cpp
src/exit_reason.cpp
src/factory.cpp
......@@ -270,6 +268,7 @@ src/middleman.cpp
src/middleman_event_handler.cpp
src/middleman_event_handler_epoll.cpp
src/middleman_event_handler_poll.cpp
src/node_id.cpp
src/object.cpp
src/object_array.cpp
src/on.cpp
......@@ -279,21 +278,24 @@ src/opencl/opencl_metainfo.cpp
src/opencl/program.cpp
src/opt.cpp
src/partial_function.cpp
src/peer.cpp
src/peer_acceptor.cpp
src/primitive_variant.cpp
src/process_information.cpp
src/protocol.cpp
src/ref_counted.cpp
src/remote_actor_proxy.cpp
src/response_handle.cpp
src/ripemd_160.cpp
src/scheduled_actor.cpp
src/scheduled_actor_dummy.cpp
src/scheduler.cpp
src/self.cpp
src/scoped_actor.cpp
src/send.cpp
src/serializer.cpp
src/shared_spinlock.cpp
src/singleton_manager.cpp
src/string_serialization.cpp
src/sync_request_bouncer.cpp
src/thread_mapped_actor.cpp
src/thread_pool_scheduler.cpp
src/to_uniform_name.cpp
......@@ -327,3 +329,5 @@ unit_testing/test_tuple.cpp
unit_testing/test_typed_spawn.cpp
unit_testing/test_uniform_type.cpp
unit_testing/test_yield_interface.cpp
cppa/detail/proper_actor.hpp
cppa/policy/no_resume.hpp
......@@ -39,18 +39,9 @@
#include "cppa/local_actor.hpp"
#include "cppa/mailbox_element.hpp"
#include "cppa/util/dptr.hpp"
#include "cppa/detail/receive_policy.hpp"
namespace cppa {
/**
* @brief An actor that uses the blocking API of @p libcppa and thus needs
* its own stack.
*/
template<class Base, class Subtype>
class stacked : public Base {
class blocking_actor : public local_actor {
friend class detail::receive_policy;
friend class detail::behavior_stack;
......
#ifndef PROPER_ACTOR_HPP
#define PROPER_ACTOR_HPP
#include "cppa/mailbox_element.hpp"
namespace cppa { namespace util { class fiber; } }
namespace cppa { namespace detail {
template<class Base,
class SchedulingPolicy,
class PriorityPolicy,
class ResumePolicy,
class InvokePolicy>
class proper_actor : public Base {
public:
template<typename... Ts>
proper_actor(Ts&&... args) : Base(std::forward<Ts>(args)) { }
void enqueue(const message_header& hdr, any_tuple msg) override {
m_scheduling_policy.enqueue(this, hdr, msg);
}
void resume(util::fiber* fb) override {
m_resume_policy.resume(this, fb);
}
inline void launch() {
m_resume_policy.launch(this);
}
inline mailbox_element* next_message() {
m_priority_policy.next_message(this);
}
void invoke_message(mailbox_element* msg) override {
m_invoke_policy.invoke(this, msg);
}
private:
SchedulingPolicy m_scheduling_policy;
PriorityPolicy m_priority_policy;
ResumePolicy m_resume_policy;
InvokePolicy m_invoke_policy;
};
} // namespace cppa::detail
#endif // PROPER_ACTOR_HPP
......@@ -309,16 +309,6 @@ class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_b
else quit(exit_reason::unhandled_sync_failure);
}
virtual void dequeue(behavior& bhvr);
inline void dequeue(behavior&& bhvr);
virtual void dequeue_response(behavior&, message_id);
inline void dequeue_response(behavior&&, message_id);
inline void do_unbecome();
local_actor(bool is_scheduled = false);
virtual bool initialized() const = 0;
......@@ -350,10 +340,6 @@ class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_b
//inline void do_become(const behavior& bhvr, bool discard_old);
const char* debug_name() const;
void debug_name(std::string str);
inline std::uint32_t planned_exit_reason() const;
inline void planned_exit_reason(std::uint32_t value);
......@@ -364,9 +350,6 @@ class local_actor : public extend<abstract_actor>::with<memory_cached, mailbox_b
void cleanup(std::uint32_t reason);
// used *only* when compiled in debug mode
union { std::string m_debug_name; };
// true if this actor receives EXIT messages as ordinary messages
bool m_trap_exit;
......@@ -443,10 +426,6 @@ inline actor_addr& local_actor::last_sender() {
return m_current_node->sender;
}
//inline void local_actor::do_unbecome() {
// m_bhvr_stack.pop_async_back();
//}
inline message_id local_actor::get_response_id() {
auto id = m_current_node->mid;
return (id.is_request()) ? id.response_id() : message_id();
......
......@@ -152,13 +152,6 @@ oss_wr operator<<(oss_wr&& lhs, T rhs) {
#define CPPA_DEBUG 3
#define CPPA_TRACE 4
#ifdef CPPA_DEBUG_MODE
# define CPPA_SET_DEBUG_NAME(strstr) \
self->debug_name((::cppa::oss_wr{} << strstr).str());
#else
# define CPPA_SET_DEBUG_NAME(unused)
#endif
#define CPPA_LVL_NAME0() "ERROR"
#define CPPA_LVL_NAME1() "WARN "
#define CPPA_LVL_NAME2() "INFO "
......
......@@ -48,7 +48,8 @@ class message_id : util::comparable<message_id> {
static constexpr std::uint64_t response_flag_mask = 0x8000000000000000;
static constexpr std::uint64_t answered_flag_mask = 0x4000000000000000;
static constexpr std::uint64_t request_id_mask = 0x3FFFFFFFFFFFFFFF;
static constexpr std::uint64_t high_prioity_flag_mask = 0x1000000000000000;
static constexpr std::uint64_t request_id_mask = 0x1FFFFFFFFFFFFFFF;
public:
......@@ -72,8 +73,12 @@ class message_id : util::comparable<message_id> {
return (m_value & answered_flag_mask) != 0;
}
inline bool is_high_priority() const {
return (m_value & high_prioity_flag_mask) != 0;
}
inline bool valid() const {
return m_value != 0;
return (m_value & request_id_mask) != 0;
}
inline bool is_request() const {
......
......@@ -31,50 +31,36 @@
#ifndef CPPA_CONTEXT_SWITCHING_ACTOR_HPP
#define CPPA_CONTEXT_SWITCHING_ACTOR_HPP
#include <stack>
#include "cppa/config.hpp"
#include "cppa/extend.hpp"
#include "cppa/stacked.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/detail/receive_policy.hpp"
#include "cppa/detail/behavior_stack.hpp"
#include "cppa/detail/yield_interface.hpp"
#include "cppa/mailbox_element.hpp"
namespace cppa {
namespace cppa { namespace policy {
/**
* @brief Context-switching actor implementation.
* @extends scheduled_actor
*/
class context_switching_actor : public extend<scheduled_actor, context_switching_actor>::with<stacked> {
friend class detail::behavior_stack;
friend class detail::receive_policy;
typedef combined_type super;
class context_switching_resume {
public:
/**
* @brief Creates a context-switching actor running @p fun.
*/
context_switching_actor(std::function<void()> fun);
//context_switching_resume(std::function<void()> fun);
resume_result resume(util::fiber* from);
scheduled_actor_type impl_type();
protected:
resume_result resume(abstract_actor*, util::fiber* from);
typedef std::chrono::high_resolution_clock::time_point timeout_type;
timeout_type init_timeout(const util::duration& rel_time);
mailbox_element* next_message();
mailbox_element* await_message();
inline mailbox_element* next_message(int) {
// we don't use the dummy element returned by init_timeout
return next_message();
}
mailbox_element* await_message(const timeout_type& abs_time);
int init_timeout(const util::duration& rel_time);
inline mailbox_element* try_pop() {
return m_mailbox.try_pop();
......@@ -90,6 +76,6 @@ class context_switching_actor : public extend<scheduled_actor, context_switching
};
} // namespace cppa
} } // namespace cppa::policy
#endif // CPPA_CONTEXT_SWITCHING_ACTOR_HPP
......@@ -43,13 +43,13 @@
#include "cppa/scheduled_actor.hpp"
#include "cppa/detail/receive_policy.hpp"
namespace cppa {
namespace cppa { namespace policy {
/**
* @brief Base class for all event-based actor implementations.
* @extends scheduled_actor
*/
class event_based_actor : public extend<scheduled_actor>::with<stackless> {
class event_based_resume {
friend class detail::receive_policy;
......@@ -57,7 +57,7 @@ class event_based_actor : public extend<scheduled_actor>::with<stackless> {
public:
resume_result resume(util::fiber*);
resume_result resume(untyped_actor*, util::fiber*);
/**
* @brief Initializes the actor.
......@@ -80,6 +80,6 @@ class event_based_actor : public extend<scheduled_actor>::with<stackless> {
};
} // namespace cppa
} } // namespace cppa::policy
#endif // CPPA_ABSTRACT_EVENT_BASED_ACTOR_HPP
......@@ -50,7 +50,7 @@
#include "cppa/util/scope_guard.hpp"
namespace cppa { namespace detail {
namespace cppa { namespace policy {
enum receive_policy_flag {
// receives can be nested
......@@ -62,7 +62,7 @@ enum receive_policy_flag {
template<receive_policy_flag X>
struct rp_flag { typedef std::integral_constant<receive_policy_flag, X> type; };
class receive_policy {
class invoke_policy_base {
public:
......@@ -280,72 +280,6 @@ class receive_policy {
return ordinary_message;
}
// the workflow of handle_message (hm) is as follows:
// - should_skip? if yes: return hm_skip_msg
// - msg is ordinary message? if yes:
// - begin(...) -> prepares a client for message handling
// - client could process message?
// - yes: cleanup()
// - no: revert(...) -> set client back to state it had before begin()
// workflow implementation for nestable receive policy
static inline bool hm_should_skip(pointer node, nestable) {
return node->marked;
}
template<class Client>
static inline pointer hm_begin(Client* client, pointer node, nestable) {
auto previous = client->m_current_node;
client->m_current_node = node;
client->push_timeout();
node->marked = true;
return previous;
}
template<class Client>
static inline void hm_cleanup(Client* client, pointer previous, nestable) {
client->m_current_node->marked = false;
client->m_current_node = previous;
}
template<class Client>
static inline void hm_revert(Client* client, pointer previous, nestable) {
client->m_current_node->marked = false;
client->m_current_node = previous;
client->pop_timeout();
}
// workflow implementation for sequential receive policy
static inline bool hm_should_skip(pointer, sequential) {
return false;
}
template<class Client>
static inline pointer hm_begin(Client* client, pointer node, sequential) {
auto previous = client->m_current_node;
client->m_current_node = node;
return previous;
}
template<class Client>
static inline void hm_cleanup(Client* client, pointer /*previous*/, sequential) {
client->m_current_node = &(client->m_dummy_node);
if (client->has_behavior()) {
client->request_timeout(client->get_behavior().timeout());
}
else client->reset_timeout();
}
template<class Client>
static inline void hm_revert(Client* client, pointer previous, sequential) {
client->m_current_node = previous;
}
public:
template<class Client>
......@@ -432,7 +366,13 @@ class receive_policy {
return none;
}
// workflow 'template'
// the workflow of handle_message (hm) is as follows:
// - should_skip? if yes: return hm_skip_msg
// - msg is ordinary message? if yes:
// - begin(...) -> prepares a client for message handling
// - client could process message?
// - yes: cleanup()
// - no: revert(...) -> set client back to state it had before begin()
template<class Client, class Fun, class Policy>
handle_message_result handle_message(Client* client,
......@@ -441,7 +381,7 @@ class receive_policy {
message_id awaited_response,
Policy policy ) {
bool handle_sync_failure_on_mismatch = true;
if (hm_should_skip(node, policy)) {
if (Derived::hm_should_skip(node, policy)) {
return hm_skip_msg;
}
switch (this->filter_msg(client, node)) {
......@@ -479,7 +419,7 @@ class receive_policy {
}
case sync_response: {
if (awaited_response.valid() && node->mid == awaited_response) {
auto previous_node = hm_begin(client, node, policy);
auto previous_node = Derived::hm_begin(client, node, policy);
auto res = invoke_fun(client,
node->msg,
node->mid,
......@@ -493,24 +433,24 @@ class receive_policy {
}
client->mark_arrived(awaited_response);
client->remove_handler(awaited_response);
hm_cleanup(client, previous_node, policy);
Derived::hm_cleanup(client, previous_node, policy);
return hm_msg_handled;
}
return hm_cache_msg;
}
case ordinary_message: {
if (!awaited_response.valid()) {
auto previous_node = hm_begin(client, node, policy);
auto previous_node = Derived::hm_begin(client, node, policy);
auto res = invoke_fun(client,
node->msg,
node->mid,
fun);
if (res) {
hm_cleanup(client, previous_node, policy);
Derived::hm_cleanup(client, previous_node, policy);
return hm_msg_handled;
}
// no match (restore client members)
hm_revert(client, previous_node, policy);
Derived::hm_revert(client, previous_node, policy);
}
return hm_cache_msg;
}
......
......@@ -44,23 +44,46 @@
#include "cppa/intrusive/single_reader_queue.hpp"
namespace cppa { namespace detail { class receive_policy; } }
#include "cppa/policy/invoke_policy_base.hpp"
namespace cppa {
namespace cppa { namespace detail { class receive_policy; } }
template<class Base, class Subtype>
class threaded : public Base {
namespace cppa { namespace policy {
friend class detail::receive_policy;
class nestable_invoke : public invoke_policy_base<nestable_invoke> {
typedef std::unique_lock<std::mutex> lock_type;
public:
static inline bool hm_should_skip(pointer) {
return false;
}
template<class Client>
static inline pointer hm_begin(Client* client, pointer node) {
auto previous = client->m_current_node;
client->m_current_node = node;
return previous;
}
template<class Client>
static inline void hm_cleanup(Client* client, pointer /*previous*/) {
client->m_current_node = &(client->m_dummy_node);
if (client->has_behavior()) {
client->request_timeout(client->get_behavior().timeout());
}
else client->reset_timeout();
}
template<class Client>
static inline void hm_revert(Client* client, pointer previous) {
client->m_current_node = previous;
}
typedef std::chrono::high_resolution_clock::time_point timeout_type;
template<typename... Ts>
threaded(Ts&&... args) : Base(std::forward<Ts>(args)...), m_initialized(false) { }
nestable_invoke() : m_initialized(false) { }
inline void reset_timeout() { }
......@@ -181,6 +204,6 @@ class threaded : public Base {
};
} // namespace cppa
} } // namespace cppa::policy
#endif // CPPA_THREADED_HPP
#ifndef NO_RESUME_HPP
#define NO_RESUME_HPP
namespace cppa { namespace policy {
class no_resume {
};
} // namespace cppa::policy
#endif // NO_RESUME_HPP
......@@ -39,8 +39,7 @@
namespace cppa {
template<class Base, class Subtype>
class prioritizing : public Base {
class prioritizing {
public:
......
......@@ -33,28 +33,47 @@
#include "cppa/atom.hpp"
#include "cppa/behavior.hpp"
#include "cppa/util/duration.hpp"
namespace cppa {
#include "cppa/policy/invoke_policy_base.hpp"
namespace cppa { namespace policy {
/**
* @brief An actor that is scheduled or otherwise managed.
*/
template<class Base, class Subtype>
class threadless : public Base {
class sequential_invoke : public invoke_policy_base<sequential_invoke> {
protected:
public:
typedef threadless combined_type;
sequential_invoke() : m_has_pending_tout(false), m_pending_tout(0) { }
public:
static inline bool hm_should_skip(pointer node) {
return node->marked;
}
static constexpr bool has_blocking_receive = false;
template<class Client>
static inline pointer hm_begin(Client* client, pointer node) {
auto previous = client->m_current_node;
client->m_current_node = node;
client->push_timeout();
node->marked = true;
return previous;
}
template<class Client>
static inline void hm_cleanup(Client* client, pointer previous) {
client->m_current_node->marked = false;
client->m_current_node = previous;
}
template<typename... Ts>
threadless(Ts&&... args) : Base(std::forward<Ts>(args)...)
, m_has_pending_tout(false)
, m_pending_tout(0) { }
template<class Client>
static inline void hm_revert(Client* client, pointer previous) {
client->m_current_node->marked = false;
client->m_current_node = previous;
client->pop_timeout();
}
inline void reset_timeout() {
if (m_has_pending_tout) {
......@@ -107,6 +126,6 @@ class threadless : public Base {
};
} // namespace cppa
} } // namespace cppa::policy
#endif // CPPA_THREADLESS_HPP
......@@ -41,7 +41,87 @@ namespace cppa {
/**
* @extends local_actor
*/
class untyped_actor : public extend<local_actor>::with<stackless> {
class untyped_actor : public local_actor {
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:
// allows actors to keep previous behaviors and enables unbecome()
detail::behavior_stack m_bhvr_stack;
};
......
......@@ -28,7 +28,7 @@
\******************************************************************************/
#include "cppa/context_switching_actor.hpp"
#include "cppa/policy/context_switching_resume.hpp"
#ifndef CPPA_DISABLE_CONTEXT_SWITCHING
#include <iostream>
......@@ -36,26 +36,28 @@
#include "cppa/cppa.hpp"
#include "cppa/self.hpp"
namespace cppa {
namespace cppa { namespace policy {
context_switching_actor::context_switching_actor(std::function<void()> fun)
/*
context_switching_resume::context_switching_resume(std::function<void()> fun)
: super(actor_state::ready, true)
, m_fiber(&context_switching_actor::trampoline, this) {
set_behavior(std::move(fun));
}
*/
auto context_switching_actor::init_timeout(const util::duration& tout) -> timeout_type {
int context_switching_actor::init_timeout(const util::duration& tout) {
// request explicit timeout message
request_timeout(tout);
return {};
return 0;
}
mailbox_element* context_switching_actor::await_message(const timeout_type&) {
// receives requested timeout message if timeout occured
mailbox_element* context_switching_actor::next_message(const timeout_type&) {
return await_message();
}
mailbox_element* context_switching_actor::await_message() {
mailbox_element* context_switching_actor::next_message() {
auto e = m_mailbox.try_pop();
while (e == nullptr) {
if (m_mailbox.can_fetch_more() == false) {
......@@ -73,7 +75,7 @@ mailbox_element* context_switching_actor::await_message() {
return e;
}
void context_switching_actor::trampoline(void* this_ptr) {
void context_switching_resume::trampoline(void* this_ptr) {
auto _this = reinterpret_cast<context_switching_actor*>(this_ptr);
bool cleanup_called = false;
try { _this->run(); }
......@@ -91,11 +93,7 @@ void context_switching_actor::trampoline(void* this_ptr) {
detail::yield(detail::yield_state::done);
}
scheduled_actor_type context_switching_actor::impl_type() {
return context_switching_impl;
}
resume_result context_switching_actor::resume(util::fiber* from, actor_ptr& next_job) {
resume_result context_switching_resume::resume(util::fiber* from) {
CPPA_LOGMF(CPPA_TRACE, this, "state = " << static_cast<int>(state()));
CPPA_REQUIRE(from != nullptr);
CPPA_REQUIRE(next_job == nullptr);
......@@ -140,7 +138,7 @@ resume_result context_switching_actor::resume(util::fiber* from, actor_ptr& next
}
}
} // namespace cppa
} } // namespace cppa::policy
#else // ifdef CPPA_DISABLE_CONTEXT_SWITCHING
......
......@@ -90,17 +90,17 @@ intrusive_ptr<event_based_actor> event_based_actor::from(std::function<void()> f
event_based_actor::event_based_actor(actor_state st) : super(st, true) { }
resume_result event_based_actor::resume(util::fiber*) {
CPPA_LOG_TRACE("id = " << id() << ", state = " << static_cast<int>(state()));
CPPA_REQUIRE( state() == actor_state::ready
|| state() == actor_state::pending);
resume_result event_based_resume::resume(local_actor* self, util::fiber*) {
CPPA_LOG_TRACE("id = " << self->id() << ", state = " << static_cast<int>(state()));
CPPA_REQUIRE( self->state() == actor_state::ready
|| self->state() == actor_state::pending);
auto done_cb = [&]() -> bool {
CPPA_LOG_TRACE("");
if (exit_reason() == exit_reason::not_exited) {
if (planned_exit_reason() == exit_reason::not_exited) {
planned_exit_reason(exit_reason::normal);
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);
}
on_exit();
self->on_exit();
if (!m_bhvr_stack.empty()) {
planned_exit_reason(exit_reason::not_exited);
return false; // on_exit did set a new behavior
......
......@@ -203,14 +203,6 @@ void local_actor::cleanup(std::uint32_t reason) {
super::cleanup(reason);
}
void local_actor::dequeue(behavior&) {
quit(exit_reason::unallowed_function_call);
}
void local_actor::dequeue_response(behavior&, message_id) {
quit(exit_reason::unallowed_function_call);
}
void local_actor::quit(std::uint32_t reason) {
CPPA_LOG_TRACE("reason = " << reason
<< ", class " << detail::demangle(typeid(*this)));
......
......@@ -349,7 +349,6 @@ middleman::~middleman() { }
void middleman_loop(middleman_impl* impl) {
# ifdef CPPA_LOG_LEVEL
auto mself = make_counted<thread_mapped_actor>();
CPPA_SET_DEBUG_NAME("middleman");
# endif
middleman_event_handler* handler = impl->m_handler.get();
CPPA_LOGF_TRACE("run middleman loop");
......
......@@ -58,28 +58,24 @@ size_t pongs() {
}
void ping(size_t num_pings) {
CPPA_SET_DEBUG_NAME("ping");
CPPA_LOGF_TRACE("num_pings = " << num_pings);
s_pongs = 0;
receive_loop(ping_behavior(num_pings));
}
void event_based_ping(size_t num_pings) {
CPPA_SET_DEBUG_NAME("event_based_ping");
CPPA_LOGF_TRACE("num_pings = " << num_pings);
s_pongs = 0;
become(ping_behavior(num_pings));
}
void pong(actor_ptr ping_actor) {
CPPA_SET_DEBUG_NAME("pong");
CPPA_LOGF_TRACE("ping_actor = " << to_string(ping_actor));
send(ping_actor, atom("pong"), 0); // kickoff
receive_loop(pong_behavior());
}
void event_based_pong(actor_ptr ping_actor) {
CPPA_SET_DEBUG_NAME("event_based_pong");
CPPA_LOGF_TRACE("ping_actor = " << to_string(ping_actor));
CPPA_REQUIRE(ping_actor != nullptr);
send(ping_actor, atom("pong"), 0); // kickoff
......
......@@ -65,7 +65,6 @@ void verbose_terminate() {
}
void set_default_test_settings() {
CPPA_SET_DEBUG_NAME("main");
set_terminate(verbose_terminate);
cout.unsetf(ios_base::unitbuf);
}
......
......@@ -21,7 +21,6 @@ typedef std::pair<std::string, std::string> string_pair;
typedef vector<actor_ptr> actor_vector;
void reflector() {
CPPA_SET_DEBUG_NAME("reflector" << self->id());
become (
others() >> [=] {
CPPA_LOGF_INFO("reflect and quit");
......@@ -96,7 +95,6 @@ void spawn5_server_impl(actor_ptr client, group_ptr grp) {
// receive seven reply messages (2 local, 5 remote)
void spawn5_server(actor_ptr client, bool inverted) {
CPPA_SET_DEBUG_NAME("spawn5_server");
if (!inverted) spawn5_server_impl(client, group::get("local", "foobar"));
else {
CPPA_LOGF_INFO("request group");
......@@ -109,7 +107,6 @@ void spawn5_server(actor_ptr client, bool inverted) {
}
void spawn5_client() {
CPPA_SET_DEBUG_NAME("spawn5_client");
become (
on(atom("GetGroup")) >> []() -> group_ptr {
CPPA_LOGF_INFO("received {'GetGroup'}");
......@@ -154,7 +151,6 @@ class client : public event_based_actor {
client(actor_ptr server) : m_server(std::move(server)) { }
void init() {
CPPA_SET_DEBUG_NAME("client");
spawn_ping();
}
......@@ -233,7 +229,6 @@ class server : public event_based_actor {
public:
void init() {
CPPA_SET_DEBUG_NAME("server");
await_spawn_ping();
}
......
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