Commit 489d0bc9 authored by Dominik Charousset's avatar Dominik Charousset

Merge invoke policies, remove `single_timeout`

parent c6c25889
...@@ -51,6 +51,7 @@ set (LIBCAF_CORE_SRCS ...@@ -51,6 +51,7 @@ set (LIBCAF_CORE_SRCS
src/group_manager.cpp src/group_manager.cpp
src/local_actor.cpp src/local_actor.cpp
src/logging.cpp src/logging.cpp
src/mailbox_based_actor.cpp
src/mailbox_element.cpp src/mailbox_element.cpp
src/memory.cpp src/memory.cpp
src/memory_managed.cpp src/memory_managed.cpp
......
...@@ -29,11 +29,11 @@ ...@@ -29,11 +29,11 @@
#include "caf/typed_actor.hpp" #include "caf/typed_actor.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
#include "caf/response_handle.hpp" #include "caf/response_handle.hpp"
#include "caf/mailbox_based_actor.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
#include "caf/mixin/sync_sender.hpp" #include "caf/mixin/sync_sender.hpp"
#include "caf/mixin/mailbox_based.hpp"
namespace caf { namespace caf {
...@@ -43,9 +43,8 @@ namespace caf { ...@@ -43,9 +43,8 @@ namespace caf {
* @extends local_actor * @extends local_actor
*/ */
class blocking_actor class blocking_actor
: public extend<local_actor, blocking_actor>:: : public extend<mailbox_based_actor, blocking_actor>::
with<mixin::mailbox_based, with<mixin::sync_sender<blocking_response_handle_tag>::impl> {
mixin::sync_sender<blocking_response_handle_tag>::impl> {
public: public:
class functor_based; class functor_based;
......
...@@ -55,7 +55,7 @@ class proper_actor_base : public Policies::resume_policy::template ...@@ -55,7 +55,7 @@ class proper_actor_base : public Policies::resume_policy::template
void enqueue(const actor_addr& sender, message_id mid, void enqueue(const actor_addr& sender, message_id mid,
message msg, execution_unit* eu) override { message msg, execution_unit* eu) override {
auto d = dptr(); auto d = dptr();
scheduling_policy().enqueue(d, d->new_mailbox_element(sender, mid, scheduling_policy().enqueue(d, mailbox_element::make(sender, mid,
std::move(msg)), std::move(msg)),
eu); eu);
} }
...@@ -125,7 +125,7 @@ class proper_actor_base : public Policies::resume_policy::template ...@@ -125,7 +125,7 @@ class proper_actor_base : public Policies::resume_policy::template
return m_policies.get_resume_policy(); return m_policies.get_resume_policy();
} }
typename Policies::invoke_policy& invoke_policy() { typename policy::invoke_policy& invoke_policy() {
return m_policies.get_invoke_policy(); return m_policies.get_invoke_policy();
} }
...@@ -246,8 +246,6 @@ class proper_actor<Base, Policies, true> ...@@ -246,8 +246,6 @@ class proper_actor<Base, Policies, true>
// immediately enqueue timeout message if duration == 0s // immediately enqueue timeout message if duration == 0s
this->enqueue(this->address(), invalid_message_id, this->enqueue(this->address(), invalid_message_id,
std::move(msg), this->host()); std::move(msg), this->host());
// auto e = this->new_mailbox_element(this, std::move(msg));
// this->m_mailbox.enqueue(e);
} else { } else {
this->delayed_send(this, d, std::move(msg)); this->delayed_send(this, d, std::move(msg));
} }
...@@ -265,14 +263,13 @@ class proper_actor<Base, Policies, true> ...@@ -265,14 +263,13 @@ class proper_actor<Base, Policies, true>
} }
} }
// required by nestable invoke policy // required by invoke policy
void pop_timeout() { void pop_timeout() {
m_pending_timeouts.pop_back(); m_pending_timeouts.pop_back();
} }
// required by nestable invoke policy; // required by invoke policy; adds a dummy timeout to the pending
// adds a dummy timeout to the pending timeouts to prevent // timeouts to prevent invokes to trigger an inactive timeout
// nestable invokes to trigger an inactive timeout
void push_timeout() { void push_timeout() {
m_pending_timeouts.push_back(++m_next_timeout_id); m_pending_timeouts.push_back(++m_next_timeout_id);
} }
......
...@@ -26,9 +26,9 @@ ...@@ -26,9 +26,9 @@
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/response_handle.hpp" #include "caf/response_handle.hpp"
#include "caf/mailbox_based_actor.hpp"
#include "caf/mixin/sync_sender.hpp" #include "caf/mixin/sync_sender.hpp"
#include "caf/mixin/mailbox_based.hpp"
#include "caf/mixin/functor_based.hpp" #include "caf/mixin/functor_based.hpp"
#include "caf/mixin/behavior_stack_based.hpp" #include "caf/mixin/behavior_stack_based.hpp"
...@@ -44,9 +44,8 @@ namespace caf { ...@@ -44,9 +44,8 @@ namespace caf {
* @extends local_actor * @extends local_actor
*/ */
class event_based_actor class event_based_actor
: public extend<local_actor, event_based_actor>:: : public extend<mailbox_based_actor, event_based_actor>::
with<mixin::mailbox_based, with<mixin::behavior_stack_based<behavior>::impl,
mixin::behavior_stack_based<behavior>::impl,
mixin::sync_sender<nonblocking_response_handle_tag>::impl> { mixin::sync_sender<nonblocking_response_handle_tag>::impl> {
public: public:
/** /**
......
...@@ -50,6 +50,7 @@ class mailbox_element; ...@@ -50,6 +50,7 @@ class mailbox_element;
class message_handler; class message_handler;
class uniform_type_info; class uniform_type_info;
class event_based_actor; class event_based_actor;
class mailbox_based_actor;
class forwarding_actor_proxy; class forwarding_actor_proxy;
// structs // structs
......
...@@ -60,8 +60,6 @@ class sync_handle_helper; ...@@ -60,8 +60,6 @@ class sync_handle_helper;
/** /**
* Base class for local running actors. * Base class for local running actors.
* @warning Instances of `local_actor` start with a reference count of 1
* @extends abstract_actor
*/ */
class local_actor : public abstract_actor { class local_actor : public abstract_actor {
public: public:
...@@ -515,11 +513,6 @@ class local_actor : public abstract_actor { ...@@ -515,11 +513,6 @@ class local_actor : public abstract_actor {
void cleanup(uint32_t reason); void cleanup(uint32_t reason);
template <class... Ts>
inline mailbox_element_ptr new_mailbox_element(Ts&&... args) {
return mailbox_element::make(std::forward<Ts>(args)...);
}
protected: protected:
// 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;
......
...@@ -17,54 +17,39 @@ ...@@ -17,54 +17,39 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_MIXIN_MAILBOX_BASED_HPP #ifndef CAF_MAILBOX_BASED_ACTOR_HPP
#define CAF_MIXIN_MAILBOX_BASED_HPP #define CAF_MAILBOX_BASED_ACTOR_HPP
#include <type_traits> #include <type_traits>
#include "caf/local_actor.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
#include "caf/detail/sync_request_bouncer.hpp" #include "caf/detail/sync_request_bouncer.hpp"
#include "caf/detail/single_reader_queue.hpp" #include "caf/detail/single_reader_queue.hpp"
namespace caf { namespace caf {
namespace mixin {
template <class Base, class Subtype> /**
class mailbox_based : public Base { * Base class for local running actors using a mailbox.
*/
class mailbox_based_actor : public local_actor {
public: public:
using del = detail::disposer; using del = detail::disposer;
using mailbox_type = detail::single_reader_queue<mailbox_element, del>; using mailbox_type = detail::single_reader_queue<mailbox_element, del>;
~mailbox_based() { ~mailbox_based_actor();
if (!m_mailbox.closed()) {
detail::sync_request_bouncer f{this->exit_reason()};
m_mailbox.close(f);
}
}
void cleanup(uint32_t reason) { void cleanup(uint32_t reason);
detail::sync_request_bouncer f{reason};
m_mailbox.close(f);
Base::cleanup(reason);
}
mailbox_type& mailbox() { inline mailbox_type& mailbox() {
return m_mailbox; return m_mailbox;
} }
protected: protected:
using combined_type = mailbox_based;
template <class... Ts>
mailbox_based(Ts&&... args) : Base(std::forward<Ts>(args)...) {
// nop
}
mailbox_type m_mailbox; mailbox_type m_mailbox;
}; };
} // namespace mixin
} // namespace caf } // namespace caf
#endif // CAF_MIXIN_MAILBOX_BASED_HPP #endif // CAF_MAILBOX_BASED_ACTOR_HPP
...@@ -37,28 +37,23 @@ ...@@ -37,28 +37,23 @@
#include "caf/actor_companion.hpp" #include "caf/actor_companion.hpp"
#include "caf/message_handler.hpp" #include "caf/message_handler.hpp"
#include "caf/policy/sequential_invoke.hpp"
namespace caf { namespace caf {
namespace mixin { namespace mixin {
template<typename Base, int EventId = static_cast<int>(QEvent::User + 31337)> template<typename Base, int EventId = static_cast<int>(QEvent::User + 31337)>
class actor_widget : public Base { class actor_widget : public Base {
public: public:
typedef typename actor_companion::message_pointer message_pointer; typedef typename actor_companion::message_pointer message_pointer;
struct event_type : public QEvent { struct event_type : public QEvent {
message_pointer mptr; message_pointer mptr;
event_type(message_pointer ptr) event_type(message_pointer ptr)
: QEvent(static_cast<QEvent::Type>(EventId)), mptr(std::move(ptr)) { } : QEvent(static_cast<QEvent::Type>(EventId)), mptr(std::move(ptr)) {
// nop
}
}; };
template<typename... Ts> template <typename... Ts>
actor_widget(Ts&&... args) : Base(std::forward<Ts>(args)...) { actor_widget(Ts&&... args) : Base(std::forward<Ts>(args)...) {
m_companion.reset(detail::memory::create<actor_companion>()); m_companion.reset(detail::memory::create<actor_companion>());
m_companion->on_enqueue([=](message_pointer ptr) { m_companion->on_enqueue([=](message_pointer ptr) {
...@@ -66,7 +61,7 @@ class actor_widget : public Base { ...@@ -66,7 +61,7 @@ class actor_widget : public Base {
}); });
} }
template<typename T> template <typename T>
void set_message_handler(T pfun) { void set_message_handler(T pfun) {
m_companion->become(pfun(m_companion.get())); m_companion->become(pfun(m_companion.get()));
} }
...@@ -75,8 +70,7 @@ class actor_widget : public Base { ...@@ -75,8 +70,7 @@ class actor_widget : public Base {
if (event->type() == static_cast<QEvent::Type>(EventId)) { if (event->type() == static_cast<QEvent::Type>(EventId)) {
auto ptr = dynamic_cast<event_type*>(event); auto ptr = dynamic_cast<event_type*>(event);
if (ptr) { if (ptr) {
m_invoke.invoke_message(m_companion.get(), m_invoke.invoke_message(m_companion.get(), ptr->mptr,
ptr->mptr,
m_companion->bhvr_stack().back(), m_companion->bhvr_stack().back(),
m_companion->bhvr_stack().back_id()); m_companion->bhvr_stack().back_id());
return true; return true;
...@@ -90,10 +84,8 @@ class actor_widget : public Base { ...@@ -90,10 +84,8 @@ class actor_widget : public Base {
} }
private: private:
policy::invoke_policy m_invoke;
policy::sequential_invoke m_invoke;
actor_companion_ptr m_companion; actor_companion_ptr m_companion;
}; };
} // namespace mixin } // namespace mixin
......
...@@ -25,51 +25,47 @@ ...@@ -25,51 +25,47 @@
#include "caf/behavior_policy.hpp" #include "caf/behavior_policy.hpp"
#include "caf/response_handle.hpp" #include "caf/response_handle.hpp"
#include "caf/mixin/single_timeout.hpp"
#include "caf/detail/behavior_stack.hpp" #include "caf/detail/behavior_stack.hpp"
namespace caf { namespace caf {
namespace mixin { namespace mixin {
template <class Base, class Subtype, class BehaviorType> template <class Base, class Subtype, class BehaviorType>
class behavior_stack_based_impl : public single_timeout<Base, Subtype> { class behavior_stack_based_impl : public Base {
using super = single_timeout<Base, Subtype>;
public: public:
// types and constructors // types and constructors
using behavior_type = BehaviorType; using behavior_type = BehaviorType;
using combined_type = behavior_stack_based_impl; using combined_type = behavior_stack_based_impl;
using response_handle_type = response_handle<behavior_stack_based_impl, using response_handle_type = response_handle<behavior_stack_based_impl,
message, message,
nonblocking_response_handle_tag>; nonblocking_response_handle_tag>;
template <class... Ts> template <class... Ts>
behavior_stack_based_impl(Ts&&... vs) behavior_stack_based_impl(Ts&&... vs)
: super(std::forward<Ts>(vs)...) {} : Base(std::forward<Ts>(vs)...),
m_timeout_id(0) {
// nop
}
/************************************************************************** /****************************************************************************
* become() member function family * * become() member function family *
**************************************************************************/ ****************************************************************************/
void become(behavior_type bhvr) { do_become(std::move(bhvr), true); } void become(behavior_type bhvr) {
do_become(std::move(bhvr), true);
}
inline void become(const keep_behavior_t&, behavior_type bhvr) { void become(const keep_behavior_t&, behavior_type bhvr) {
do_become(std::move(bhvr), false); do_become(std::move(bhvr), false);
} }
template <class T, class... Ts> template <class T, class... Ts>
inline typename std::enable_if< typename std::enable_if<
!std::is_same<keep_behavior_t, typename std::decay<T>::type>::value, !std::is_same<keep_behavior_t, typename std::decay<T>::type>::value,
void>::type void
>::type
become(T&& arg, Ts&&... args) { become(T&& arg, Ts&&... args) {
do_become( do_become(behavior_type{std::forward<T>(arg), std::forward<Ts>(args)...},
behavior_type{std::forward<T>(arg), std::forward<Ts>(args)...},
true); true);
} }
...@@ -78,15 +74,19 @@ class behavior_stack_based_impl : public single_timeout<Base, Subtype> { ...@@ -78,15 +74,19 @@ class behavior_stack_based_impl : public single_timeout<Base, Subtype> {
do_become(behavior_type{std::forward<Ts>(args)...}, false); do_become(behavior_type{std::forward<Ts>(args)...}, false);
} }
inline void unbecome() { m_bhvr_stack.pop_async_back(); } void unbecome() {
m_bhvr_stack.pop_async_back();
}
/************************************************************************** /****************************************************************************
* convenience member function for stack manipulation * * convenience member function for stack manipulation *
**************************************************************************/ ****************************************************************************/
inline bool has_behavior() const { return m_bhvr_stack.empty() == false; } bool has_behavior() const {
return m_bhvr_stack.empty() == false;
}
inline behavior& get_behavior() { behavior& get_behavior() {
CAF_REQUIRE(m_bhvr_stack.empty() == false); CAF_REQUIRE(m_bhvr_stack.empty() == false);
return m_bhvr_stack.back(); return m_bhvr_stack.back();
} }
...@@ -95,13 +95,48 @@ class behavior_stack_based_impl : public single_timeout<Base, Subtype> { ...@@ -95,13 +95,48 @@ class behavior_stack_based_impl : public single_timeout<Base, Subtype> {
return m_bhvr_stack.sync_handler(msg_id); return m_bhvr_stack.sync_handler(msg_id);
} }
inline void remove_handler(message_id mid) { m_bhvr_stack.erase(mid); } void remove_handler(message_id mid) {
m_bhvr_stack.erase(mid);
}
inline detail::behavior_stack& bhvr_stack() { return m_bhvr_stack; } detail::behavior_stack& bhvr_stack() {
return m_bhvr_stack;
}
/************************************************************************** /****************************************************************************
* extended timeout handling (handle_timeout mem fun) * * timeout handling *
**************************************************************************/ ****************************************************************************/
void request_timeout(const duration& d) {
if (d.valid()) {
this->has_timeout(true);
auto tid = ++m_timeout_id;
auto msg = make_message(timeout_msg{tid});
if (d.is_zero()) {
// immediately enqueue timeout message if duration == 0s
this->enqueue(this->address(), invalid_message_id,
std::move(msg), this->host());
} else
this->delayed_send(this, d, std::move(msg));
} else
this->has_timeout(false);
}
bool waits_for_timeout(uint32_t timeout_id) const {
return this->has_timeout() && m_timeout_id == timeout_id;
}
bool is_active_timeout(uint32_t tid) const {
return waits_for_timeout(tid);
}
uint32_t active_timeout_id() const {
return m_timeout_id;
}
void reset_timeout() {
this->has_timeout(false);
}
void handle_timeout(behavior& bhvr, uint32_t timeout_id) { void handle_timeout(behavior& bhvr, uint32_t timeout_id) {
if (this->is_active_timeout(timeout_id)) { if (this->is_active_timeout(timeout_id)) {
...@@ -117,31 +152,34 @@ class behavior_stack_based_impl : public single_timeout<Base, Subtype> { ...@@ -117,31 +152,34 @@ class behavior_stack_based_impl : public single_timeout<Base, Subtype> {
} }
private: private:
void do_become(behavior_type bhvr, bool discard_old) { void do_become(behavior_type bhvr, bool discard_old) {
if (discard_old) this->m_bhvr_stack.pop_async_back(); if (discard_old) this->m_bhvr_stack.pop_async_back();
// since we know we extend single_timeout, we can be sure
// request_timeout simply resets the timeout when it's invalid // request_timeout simply resets the timeout when it's invalid
this->request_timeout(bhvr.timeout()); this->request_timeout(bhvr.timeout());
this->m_bhvr_stack.push_back(std::move(unbox(bhvr))); this->m_bhvr_stack.push_back(std::move(unbox(bhvr)));
} }
static inline behavior& unbox(behavior& arg) { return arg; } static behavior& unbox(behavior& arg) {
return arg;
}
template <class... Ts> template <class... Ts>
static inline behavior& unbox(typed_behavior<Ts...>& arg) { static behavior& unbox(typed_behavior<Ts...>& arg) {
return arg.unbox(); return arg.unbox();
} }
// utility for getting a pointer-to-derived-type // utility for getting a pointer-to-derived-type
Subtype* dptr() { return static_cast<Subtype*>(this); } Subtype* dptr() {
return static_cast<Subtype*>(this);
}
// utility for getting a const pointer-to-derived-type // utility for getting a const pointer-to-derived-type
const Subtype* dptr() const { return static_cast<const Subtype*>(this); } const Subtype* dptr() const {
return static_cast<const Subtype*>(this);
}
// allows actors to keep previous behaviors and enables unbecome()
detail::behavior_stack m_bhvr_stack; detail::behavior_stack m_bhvr_stack;
uint32_t m_timeout_id;
}; };
/** /**
...@@ -150,22 +188,19 @@ class behavior_stack_based_impl : public single_timeout<Base, Subtype> { ...@@ -150,22 +188,19 @@ class behavior_stack_based_impl : public single_timeout<Base, Subtype> {
*/ */
template <class BehaviorType> template <class BehaviorType>
class behavior_stack_based { class behavior_stack_based {
public: public:
template <class Base, class Subtype> template <class Base, class Subtype>
class impl : public behavior_stack_based_impl<Base, Subtype, BehaviorType> { class impl : public behavior_stack_based_impl<Base, Subtype, BehaviorType> {
using super = behavior_stack_based_impl<Base, Subtype, BehaviorType>;
public: public:
using super = behavior_stack_based_impl<Base, Subtype, BehaviorType>;
using combined_type = impl; using combined_type = impl;
template <class... Ts> template <class... Ts>
impl(Ts&&... args) impl(Ts&&... args)
: super(std::forward<Ts>(args)...) {} : super(std::forward<Ts>(args)...) {
// nop
}
}; };
}; };
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_MIXIN_SINGLE_TIMEOUT_HPP
#define CAF_MIXIN_SINGLE_TIMEOUT_HPP
#include "caf/message.hpp"
#include "caf/duration.hpp"
#include "caf/system_messages.hpp"
namespace caf {
namespace mixin {
/**
* Mixin for actors using a non-nestable message processing.
*/
template <class Base, class Subtype>
class single_timeout : public Base {
public:
using super = Base;
using combined_type = single_timeout;
template <class... Ts>
single_timeout(Ts&&... args)
: super(std::forward<Ts>(args)...),
m_timeout_id(0) {
// nop
}
void request_timeout(const duration& d) {
if (d.valid()) {
this->has_timeout(true);
auto tid = ++m_timeout_id;
auto msg = make_message(timeout_msg{tid});
if (d.is_zero()) {
// immediately enqueue timeout message if duration == 0s
this->enqueue(this->address(), invalid_message_id,
std::move(msg), this->host());
} else
this->delayed_send(this, d, std::move(msg));
} else
this->has_timeout(false);
}
bool waits_for_timeout(uint32_t timeout_id) const {
return this->has_timeout() && m_timeout_id == timeout_id;
}
bool is_active_timeout(uint32_t tid) const {
return waits_for_timeout(tid);
}
uint32_t active_timeout_id() const {
return m_timeout_id;
}
void reset_timeout() {
this->has_timeout(false);
}
protected:
uint32_t m_timeout_id;
};
} // namespace mixin
} // namespace caf
#endif // CAF_MIXIN_SINGLE_TIMEOUT_HPP
...@@ -20,16 +20,15 @@ ...@@ -20,16 +20,15 @@
#ifndef CAF_POLICY_POLICIES_HPP #ifndef CAF_POLICY_POLICIES_HPP
#define CAF_POLICY_POLICIES_HPP #define CAF_POLICY_POLICIES_HPP
#include "caf/policy/invoke_policy.hpp"
namespace caf { namespace caf {
namespace policy { namespace policy {
/** /**
* A container for actor-related policies. * A container for actor-related policies.
*/ */
template <class SchedulingPolicy, template <class SchedulingPolicy, class PriorityPolicy, class ResumePolicy>
class PriorityPolicy,
class ResumePolicy,
class InvokePolicy>
class actor_policies { class actor_policies {
public: public:
...@@ -40,8 +39,6 @@ class actor_policies { ...@@ -40,8 +39,6 @@ class actor_policies {
using resume_policy = ResumePolicy; using resume_policy = ResumePolicy;
using invoke_policy = InvokePolicy;
inline scheduling_policy& get_scheduling_policy() { inline scheduling_policy& get_scheduling_policy() {
return m_scheduling_policy; return m_scheduling_policy;
} }
......
...@@ -32,7 +32,6 @@ ...@@ -32,7 +32,6 @@
#include "caf/exit_reason.hpp" #include "caf/exit_reason.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
#include "caf/system_messages.hpp" #include "caf/system_messages.hpp"
#include "caf/message_handler.hpp"
#include "caf/response_promise.hpp" #include "caf/response_promise.hpp"
#include "caf/detail/memory.hpp" #include "caf/detail/memory.hpp"
...@@ -42,28 +41,15 @@ ...@@ -42,28 +41,15 @@
namespace caf { namespace caf {
namespace policy { namespace policy {
enum receive_policy_flag {
// receives can be nested
rp_nestable,
// receives are guaranteed to be sequential
rp_sequential
};
enum invoke_message_result { enum invoke_message_result {
im_success, im_success,
im_skipped, im_skipped,
im_dropped im_dropped
}; };
template <receive_policy_flag X>
struct rp_flag {
using type = std::integral_constant<receive_policy_flag, X>;
};
/** /**
* Base class for invoke policies. * Base class for invoke policies.
*/ */
template <class Derived>
class invoke_policy { class invoke_policy {
public: public:
enum class msg_type { enum class msg_type {
...@@ -85,14 +71,12 @@ class invoke_policy { ...@@ -85,14 +71,12 @@ class invoke_policy {
// - self could process message? // - self could process message?
// - yes: cleanup() // - yes: cleanup()
// - no: revert(...) -> set self back to state it had before begin() // - no: revert(...) -> set self back to state it had before begin()
template <class Actor, class Fun> template <class Actor>
invoke_message_result invoke_message(Actor* self, mailbox_element_ptr& node, invoke_message_result invoke_message(Actor* self, mailbox_element_ptr& node,
Fun& fun, message_id awaited_response) { behavior& fun,
message_id awaited_response) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
bool handle_sync_failure_on_mismatch = true; bool handle_sync_failure_on_mismatch = true;
if (dptr()->hm_should_skip(*node)) {
return im_skipped;
}
switch (this->filter_msg(self, *node)) { switch (this->filter_msg(self, *node)) {
case msg_type::normal_exit: case msg_type::normal_exit:
CAF_LOG_DEBUG("dropped normal exit signal"); CAF_LOG_DEBUG("dropped normal exit signal");
...@@ -130,7 +114,7 @@ class invoke_policy { ...@@ -130,7 +114,7 @@ class invoke_policy {
<< CAF_MARG(node->mid, integer_value) << ", " << CAF_MARG(node->mid, integer_value) << ", "
<< CAF_MARG(awaited_response, integer_value)); << CAF_MARG(awaited_response, integer_value));
if (awaited_response.valid() && node->mid == awaited_response) { if (awaited_response.valid() && node->mid == awaited_response) {
dptr()->hm_begin(self, node); node.swap(self->current_element());
auto res = invoke_fun(self, fun); auto res = invoke_fun(self, fun);
if (!res && handle_sync_failure_on_mismatch) { if (!res && handle_sync_failure_on_mismatch) {
CAF_LOG_WARNING("sync failure occured in actor " CAF_LOG_WARNING("sync failure occured in actor "
...@@ -139,19 +123,18 @@ class invoke_policy { ...@@ -139,19 +123,18 @@ class invoke_policy {
} }
self->mark_arrived(awaited_response); self->mark_arrived(awaited_response);
self->remove_handler(awaited_response); self->remove_handler(awaited_response);
dptr()->hm_cleanup(self, node); node.swap(self->current_element());
return im_success; return im_success;
} }
return im_skipped; return im_skipped;
case msg_type::ordinary: case msg_type::ordinary:
if (!awaited_response.valid()) { if (!awaited_response.valid()) {
dptr()->hm_begin(self, node); node.swap(self->current_element());
auto res = invoke_fun(self, fun); auto res = invoke_fun(self, fun);
dptr()->hm_cleanup(self, node); node.swap(self->current_element());
if (res) { if (res) {
return im_success; return im_success;
} }
// no match (restore self members)
} }
CAF_LOG_DEBUG_IF(awaited_response.valid(), CAF_LOG_DEBUG_IF(awaited_response.valid(),
"ignored message; await response: " "ignored message; await response: "
...@@ -162,10 +145,6 @@ class invoke_policy { ...@@ -162,10 +145,6 @@ class invoke_policy {
CAF_CRITICAL("invalid message type"); CAF_CRITICAL("invalid message type");
} }
using nestable = typename rp_flag<rp_nestable>::type;
using sequential = typename rp_flag<rp_sequential>::type;
template <class Actor> template <class Actor>
response_promise fetch_response_promise(Actor* cl, int) { response_promise fetch_response_promise(Actor* cl, int) {
return cl->make_response_promise(); return cl->make_response_promise();
...@@ -243,20 +222,10 @@ class invoke_policy { ...@@ -243,20 +222,10 @@ class invoke_policy {
return res; return res;
} }
protected:
Derived* dptr() {
return static_cast<Derived*>(this);
}
private: private:
void handle_timeout(message_handler&) {
CAF_CRITICAL("handle_timeout(message_handler&)");
}
// identifies 'special' messages that should not be processed normally: // identifies 'special' messages that should not be processed normally:
// - system messages such as EXIT (if self doesn't trap exits) and TIMEOUT // - system messages such as EXIT (if self doesn't trap exits) and TIMEOUT
// - expired synchronous response messages // - expired synchronous response messages
template <class Actor> template <class Actor>
msg_type filter_msg(Actor* self, mailbox_element& node) { msg_type filter_msg(Actor* self, mailbox_element& node) {
const message& msg = node.msg; const message& msg = node.msg;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_POLICY_NESTABLE_INVOKE_HPP
#define CAF_POLICY_NESTABLE_INVOKE_HPP
#include <mutex>
#include <chrono>
#include <condition_variable>
#include "caf/exit_reason.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/detail/single_reader_queue.hpp"
#include "caf/policy/invoke_policy.hpp"
namespace caf {
namespace policy {
class nestable_invoke : public invoke_policy<nestable_invoke> {
public:
inline bool hm_should_skip(mailbox_element& node) {
return node.marked;
}
template <class Actor>
void hm_begin(Actor* self, mailbox_element_ptr& node) {
node->marked = true;
node.swap(self->current_element());
}
template <class Actor>
void hm_cleanup(Actor* self, mailbox_element_ptr& node) {
auto& ref = self->current_element();
if (ref) {
ref->marked = false;
}
ref.swap(node);
}
};
} // namespace policy
} // namespace caf
#endif // CAF_POLICY_NESTABLE_INVOKE_HPP
...@@ -31,9 +31,7 @@ ...@@ -31,9 +31,7 @@
#include "caf/policy/prioritizing.hpp" #include "caf/policy/prioritizing.hpp"
#include "caf/policy/no_scheduling.hpp" #include "caf/policy/no_scheduling.hpp"
#include "caf/policy/actor_policies.hpp" #include "caf/policy/actor_policies.hpp"
#include "caf/policy/nestable_invoke.hpp"
#include "caf/policy/not_prioritizing.hpp" #include "caf/policy/not_prioritizing.hpp"
#include "caf/policy/sequential_invoke.hpp"
#include "caf/policy/event_based_resume.hpp" #include "caf/policy/event_based_resume.hpp"
#include "caf/policy/cooperative_scheduling.hpp" #include "caf/policy/cooperative_scheduling.hpp"
...@@ -88,18 +86,11 @@ intrusive_ptr<C> spawn_impl(execution_unit* host, ...@@ -88,18 +86,11 @@ intrusive_ptr<C> spawn_impl(execution_unit* host,
policy::no_resume, policy::no_resume,
policy::event_based_resume policy::event_based_resume
>::type; >::type;
using invoke_policy =
typename std::conditional<
has_blocking_api_flag(Os),
policy::nestable_invoke,
policy::sequential_invoke
>::type;
using policy_token = using policy_token =
policy::actor_policies< policy::actor_policies<
scheduling_policy, scheduling_policy,
priority_policy, priority_policy,
resume_policy, resume_policy
invoke_policy
>; >;
using actor_impl = using actor_impl =
typename std::conditional< typename std::conditional<
......
...@@ -23,9 +23,9 @@ ...@@ -23,9 +23,9 @@
#include "caf/replies_to.hpp" #include "caf/replies_to.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/typed_behavior.hpp" #include "caf/typed_behavior.hpp"
#include "caf/mailbox_based_actor.hpp"
#include "caf/mixin/sync_sender.hpp" #include "caf/mixin/sync_sender.hpp"
#include "caf/mixin/mailbox_based.hpp"
#include "caf/mixin/behavior_stack_based.hpp" #include "caf/mixin/behavior_stack_based.hpp"
namespace caf { namespace caf {
...@@ -39,9 +39,8 @@ namespace caf { ...@@ -39,9 +39,8 @@ namespace caf {
*/ */
template <class... Rs> template <class... Rs>
class typed_event_based_actor : public class typed_event_based_actor : public
extend<local_actor, typed_event_based_actor<Rs...>>::template extend<mailbox_based_actor, typed_event_based_actor<Rs...>>::template
with<mixin::mailbox_based, with<mixin::behavior_stack_based<typed_behavior<Rs...>>::template impl,
mixin::behavior_stack_based<typed_behavior<Rs...>>::template impl,
mixin::sync_sender<nonblocking_response_handle_tag>::template impl> { mixin::sync_sender<nonblocking_response_handle_tag>::template impl> {
public: public:
using signatures = detail::type_list<Rs...>; using signatures = detail::type_list<Rs...>;
......
...@@ -54,8 +54,7 @@ using hrc = std::chrono::high_resolution_clock; ...@@ -54,8 +54,7 @@ using hrc = std::chrono::high_resolution_clock;
using timer_actor_policies = policy::actor_policies<policy::no_scheduling, using timer_actor_policies = policy::actor_policies<policy::no_scheduling,
policy::not_prioritizing, policy::not_prioritizing,
policy::no_resume, policy::no_resume>;
policy::nestable_invoke>;
struct delayed_msg { struct delayed_msg {
actor_addr from; actor_addr from;
......
...@@ -17,39 +17,21 @@ ...@@ -17,39 +17,21 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_THREADLESS_HPP #include "caf/mailbox_based_actor.hpp"
#define CAF_THREADLESS_HPP
#include "caf/atom.hpp"
#include "caf/behavior.hpp"
#include "caf/duration.hpp"
#include "caf/policy/invoke_policy.hpp"
namespace caf { namespace caf {
namespace policy {
/**
* An actor that is scheduled or otherwise managed.
*/
class sequential_invoke : public invoke_policy<sequential_invoke> {
public:
inline bool hm_should_skip(mailbox_element&) {
return false;
}
template <class Actor> mailbox_based_actor::~mailbox_based_actor() {
void hm_begin(Actor* self, mailbox_element_ptr& node) { if (!m_mailbox.closed()) {
node.swap(self->current_element()); detail::sync_request_bouncer f{this->exit_reason()};
m_mailbox.close(f);
} }
}
template <class Actor> void mailbox_based_actor::cleanup(uint32_t reason) {
void hm_cleanup(Actor* self, mailbox_element_ptr& node) { detail::sync_request_bouncer f{reason};
node.swap(self->current_element()); m_mailbox.close(f);
} local_actor::cleanup(reason);
}; }
} // namespace policy
} // namespace caf } // namespace caf
#endif // CAF_THREADLESS_HPP
...@@ -22,7 +22,6 @@ ...@@ -22,7 +22,6 @@
#include "caf/policy/no_resume.hpp" #include "caf/policy/no_resume.hpp"
#include "caf/policy/no_scheduling.hpp" #include "caf/policy/no_scheduling.hpp"
#include "caf/policy/actor_policies.hpp" #include "caf/policy/actor_policies.hpp"
#include "caf/policy/nestable_invoke.hpp"
#include "caf/policy/not_prioritizing.hpp" #include "caf/policy/not_prioritizing.hpp"
#include "caf/detail/singletons.hpp" #include "caf/detail/singletons.hpp"
...@@ -41,8 +40,7 @@ struct impl : blocking_actor { ...@@ -41,8 +40,7 @@ struct impl : blocking_actor {
blocking_actor* alloc() { blocking_actor* alloc() {
using namespace policy; using namespace policy;
using policies = actor_policies<no_scheduling, not_prioritizing, using policies = actor_policies<no_scheduling, not_prioritizing, no_resume>;
no_resume, nestable_invoke>;
return new detail::proper_actor<impl, policies>; return new detail::proper_actor<impl, policies>;
} }
......
...@@ -30,8 +30,6 @@ ...@@ -30,8 +30,6 @@
#include "caf/mixin/functor_based.hpp" #include "caf/mixin/functor_based.hpp"
#include "caf/mixin/behavior_stack_based.hpp" #include "caf/mixin/behavior_stack_based.hpp"
#include "caf/policy/sequential_invoke.hpp"
#include "caf/detail/intrusive_partitioned_list.hpp" #include "caf/detail/intrusive_partitioned_list.hpp"
#include "caf/io/fwd.hpp" #include "caf/io/fwd.hpp"
...@@ -367,7 +365,7 @@ class broker : public extend<local_actor>:: ...@@ -367,7 +365,7 @@ class broker : public extend<local_actor>::
std::map<accept_handle, doorman_pointer> m_doormen; std::map<accept_handle, doorman_pointer> m_doormen;
std::map<connection_handle, scribe_pointer> m_scribes; std::map<connection_handle, scribe_pointer> m_scribes;
policy::sequential_invoke m_invoke_policy; policy::invoke_policy m_invoke_policy;
middleman& m_mm; middleman& m_mm;
detail::intrusive_partitioned_list<mailbox_element, detail::disposer> m_cache; detail::intrusive_partitioned_list<mailbox_element, detail::disposer> m_cache;
......
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