Commit d72c0ec8 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'topic/reduce-memory-allocations-2'

parents a470e327 ba1daa54
...@@ -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
......
...@@ -45,6 +45,15 @@ class abstract_channel : public ref_counted { ...@@ -45,6 +45,15 @@ class abstract_channel : public ref_counted {
virtual void enqueue(const actor_addr& sender, message_id mid, virtual void enqueue(const actor_addr& sender, message_id mid,
message content, execution_unit* host) = 0; message content, execution_unit* host) = 0;
/**
* Enqueues a new message wrapped in a `mailbox_element` to the channel.
* This variant is used by actors whenever it is possible to allocate
* mailbox element and message on the same memory block and is thus
* more efficient. Non-actors use the default implementation which simply
* calls the pure virtual version.
*/
virtual void enqueue(mailbox_element_ptr what, execution_unit* host);
/** /**
* Returns the ID of the node this actor is running on. * Returns the ID of the node this actor is running on.
*/ */
......
...@@ -29,7 +29,7 @@ ...@@ -29,7 +29,7 @@
#include "caf/mixin/sync_sender.hpp" #include "caf/mixin/sync_sender.hpp"
#include "caf/mixin/behavior_stack_based.hpp" #include "caf/mixin/behavior_stack_based.hpp"
#include "caf/detail/memory.hpp" #include "caf/detail/disposer.hpp"
#include "caf/detail/shared_spinlock.hpp" #include "caf/detail/shared_spinlock.hpp"
namespace caf { namespace caf {
...@@ -39,16 +39,13 @@ namespace caf { ...@@ -39,16 +39,13 @@ namespace caf {
* callback to another object, thus serving as gateway to * callback to another object, thus serving as gateway to
* allow any object to interact with other actors. * allow any object to interact with other actors.
*/ */
class actor_companion : public extend<local_actor, actor_companion>:: class actor_companion
with<mixin::behavior_stack_based<behavior>::impl, : public extend<local_actor, actor_companion>::
mixin::sync_sender<nonblocking_response_handle_tag>::impl> { with<mixin::behavior_stack_based<behavior>::impl,
mixin::sync_sender<nonblocking_response_handle_tag>::impl> {
using lock_type = detail::shared_spinlock;
public: public:
using lock_type = detail::shared_spinlock;
using message_pointer = std::unique_ptr<mailbox_element, detail::disposer>; using message_pointer = std::unique_ptr<mailbox_element, detail::disposer>;
using enqueue_handler = std::function<void (message_pointer)>; using enqueue_handler = std::function<void (message_pointer)>;
/** /**
...@@ -64,16 +61,14 @@ class actor_companion : public extend<local_actor, actor_companion>:: ...@@ -64,16 +61,14 @@ class actor_companion : public extend<local_actor, actor_companion>::
void on_enqueue(enqueue_handler handler); void on_enqueue(enqueue_handler handler);
void enqueue(const actor_addr& sender, message_id mid, void enqueue(const actor_addr& sender, message_id mid,
message content, execution_unit* host) override; message content, execution_unit* host) override;
private: private:
// set by parent to define custom enqueue action // set by parent to define custom enqueue action
enqueue_handler m_on_enqueue; enqueue_handler m_on_enqueue;
// guards access to m_handler // guards access to m_handler
lock_type m_lock; lock_type m_lock;
}; };
/** /**
......
...@@ -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;
......
...@@ -17,54 +17,23 @@ ...@@ -17,54 +17,23 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_POLICY_NESTABLE_INVOKE_HPP #ifndef CAF_DETAIL_DISPOSER_HPP
#define CAF_POLICY_NESTABLE_INVOKE_HPP #define CAF_DETAIL_DISPOSER_HPP
#include <mutex> #include "caf/memory_managed.hpp"
#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 caf {
namespace policy { namespace detail {
class nestable_invoke : public invoke_policy<nestable_invoke> { class disposer {
public: public:
inline bool hm_should_skip(mailbox_element& node) { inline void operator()(memory_managed* ptr) const {
return node.marked; ptr->request_deletion();
}
template <class Actor>
mailbox_element* hm_begin(Actor* self, mailbox_element& node) {
auto previous = self->current_node();
self->current_node(&node);
self->push_timeout();
node.marked = true;
return previous;
}
template <class Actor>
void hm_cleanup(Actor* self, mailbox_element* previous) {
self->current_node()->marked = false;
self->current_node(previous);
self->pop_timeout();
}
template <class Actor>
void hm_revert(Actor* self, mailbox_element* previous) {
// same operation for blocking, i.e., nestable, invoke
hm_cleanup(self, previous);
} }
}; };
} // namespace policy } // namespace detail
} // namespace caf } // namespace caf
#endif // CAF_POLICY_NESTABLE_INVOKE_HPP #endif // CAF_DETAIL_DISPOSER_HPP
...@@ -17,45 +17,43 @@ ...@@ -17,45 +17,43 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_THREADLESS_HPP #ifndef CAF_DETAIL_EMBEDDED_HPP
#define CAF_THREADLESS_HPP #define CAF_DETAIL_EMBEDDED_HPP
#include "caf/atom.hpp" #include "caf/ref_counted.hpp"
#include "caf/behavior.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/duration.hpp"
#include "caf/policy/invoke_policy.hpp"
namespace caf { namespace caf {
namespace policy { namespace detail {
/** template <class Base>
* An actor that is scheduled or otherwise managed. class embedded final : public Base {
*/
class sequential_invoke : public invoke_policy<sequential_invoke> {
public: public:
inline bool hm_should_skip(mailbox_element&) { template <class... Vs>
return false; embedded(intrusive_ptr<ref_counted> storage, Vs&&... vs)
: Base(std::forward<Vs>(vs)...),
m_storage(std::move(storage)) {
// nop
} }
template <class Actor> ~embedded() {
mailbox_element* hm_begin(Actor* self, mailbox_element& node) { // nop
self->current_node(&node);
return nullptr;
} }
template <class Actor> void request_deletion() override {
void hm_cleanup(Actor* self, mailbox_element*) { intrusive_ptr<ref_counted> guard;
self->current_node(self->dummy_node()); guard.swap(m_storage);
// this code assumes that embedded is part of pair_storage<>,
// i.e., this object lives inside a union!
this->~embedded();
} }
template <class Actor> protected:
void hm_revert(Actor* self, mailbox_element*) { intrusive_ptr<ref_counted> m_storage;
self->current_node(self->dummy_node());
}
}; };
} // namespace policy } // namespace detail
} // namespace caf } // namespace caf
#endif // CAF_THREADLESS_HPP #endif // CAF_DETAIL_EMBEDDED_HPP
...@@ -17,12 +17,14 @@ ...@@ -17,12 +17,14 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_DETAIL_INTRUSIVE_LIST_HPP #ifndef CAF_DETAIL_INTRUSIVE_PARTITIONED_LIST_HPP
#define CAF_DETAIL_INTRUSIVE_LIST_HPP #define CAF_DETAIL_INTRUSIVE_PARTITIONED_LIST_HPP
#include <memory> #include <memory>
#include <iterator> #include <iterator>
#include "caf/policy/invoke_policy.hpp"
namespace caf { namespace caf {
namespace detail { namespace detail {
...@@ -173,6 +175,49 @@ class intrusive_partitioned_list { ...@@ -173,6 +175,49 @@ class intrusive_partitioned_list {
insert(second_end(), val); insert(second_end(), val);
} }
template <class Actor, class... Vs>
bool invoke(Actor* self, iterator first, iterator last, Vs&... vs) {
pointer prev = first->prev;
pointer next = first->next;
auto move_on = [&](bool first_valid) {
if (first_valid) {
prev = first.ptr;
}
first = next;
next = first->next;
};
while (first != last) {
std::unique_ptr<value_type, deleter_type> tmp{first.ptr};
// since this function can be called recursively during
// self->invoke_message(tmp, vs...), we have to remove the
// element from the list proactively and put it back in if
// it's safe, i.e., if invoke_message returned im_skipped
prev->next = next;
next->prev = prev;
switch (self->invoke_message(tmp, vs...)) {
case policy::im_dropped:
move_on(false);
break;
case policy::im_success:
return true;
case policy::im_skipped:
if (tmp) {
// re-integrate tmp and move on
prev->next = tmp.get();
next->prev = tmp.release();
move_on(true);
}
else {
// only happens if the user does something
// really, really stupid; check it nonetheless
move_on(false);
}
break;
}
}
return false;
}
size_t count(iterator first, iterator last, size_t max_count) { size_t count(iterator first, iterator last, size_t max_count) {
size_t result = 0; size_t result = 0;
while (first != last && result < max_count) { while (first != last && result < max_count) {
...@@ -197,4 +242,4 @@ class intrusive_partitioned_list { ...@@ -197,4 +242,4 @@ class intrusive_partitioned_list {
} // namespace detail } // namespace detail
} // namespace caf } // namespace caf
#endif // CAF_DETAIL_INTRUSIVE_LIST_HPP #endif // CAF_DETAIL_INTRUSIVE_PARTITIONED_LIST_HPP
...@@ -24,23 +24,26 @@ ...@@ -24,23 +24,26 @@
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/mixin/memory_cached.hpp"
#include "caf/detail/memory.hpp" #include "caf/detail/memory.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf { namespace caf {
namespace detail { namespace detail {
template <class T, class... Ts> template <class T, class... Ts>
typename std::enable_if<mixin::is_memory_cached<T>::value, typename std::enable_if<
intrusive_ptr<T>>::type detail::is_memory_cached<T>::value,
intrusive_ptr<T>
>::type
make_counted(Ts&&... args) { make_counted(Ts&&... args) {
return {detail::memory::create<T>(std::forward<Ts>(args)...)}; return {detail::memory::create<T>(std::forward<Ts>(args)...)};
} }
template <class T, class... Ts> template <class T, class... Ts>
typename std::enable_if<!mixin::is_memory_cached<T>::value, typename std::enable_if<
intrusive_ptr<T>>::type !detail::is_memory_cached<T>::value,
intrusive_ptr<T>
>::type
make_counted(Ts&&... args) { make_counted(Ts&&... args) {
return {new T(std::forward<Ts>(args)...)}; return {new T(std::forward<Ts>(args)...)};
} }
......
...@@ -29,6 +29,9 @@ ...@@ -29,6 +29,9 @@
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/detail/embedded.hpp"
#include "caf/detail/memory_cache_flag_type.hpp"
namespace caf { namespace caf {
class mailbox_element; class mailbox_element;
} // namespace caf } // namespace caf
...@@ -45,147 +48,133 @@ constexpr size_t s_max_elements = 20; // don't create > 20 elements ...@@ -45,147 +48,133 @@ constexpr size_t s_max_elements = 20; // don't create > 20 elements
} // namespace <anonymous> } // namespace <anonymous>
struct disposer { using embedded_storage = std::pair<intrusive_ptr<ref_counted>, void*>;
inline void operator()(memory_managed* ptr) const {
ptr->request_deletion();
}
};
class instance_wrapper {
public:
virtual ~instance_wrapper();
// calls the destructor
virtual void destroy() = 0;
// releases memory
virtual void deallocate() = 0;
};
class memory_cache { class memory_cache {
public: public:
virtual ~memory_cache(); virtual ~memory_cache();
virtual embedded_storage new_embedded_storage() = 0;
// calls dtor and either releases memory or re-uses it later
virtual void release_instance(void*) = 0;
virtual std::pair<instance_wrapper*, void*> new_instance() = 0;
// casts `ptr` to the derived type and returns it
virtual void* downcast(memory_managed* ptr) = 0;
}; };
class instance_wrapper;
template <class T> template <class T>
class basic_memory_cache; class basic_memory_cache;
#ifdef CAF_NO_MEM_MANAGEMENT #ifdef CAF_NO_MEM_MANAGEMENT
class memory { template <class T>
struct rc_storage : public ref_counted {
T instance;
template <class... Vs>
rc_storage(Vs&&... vs) : instance(this, std::forward<Vs>(vs)...) {
// nop
}
};
memory() = delete; template <class T>
T* unbox_rc_storage(T* ptr) {
return ptr;
}
template <class T>
T* unbox_rc_storage(rc_storage<T>* ptr) {
return &(ptr->instance);
}
class memory {
public: public:
memory() = delete;
// Allocates storage, initializes a new object, and returns the new instance. // Allocates storage, initializes a new object, and returns the new instance.
template <class T, class... Ts> template <class T, class... Ts>
static T* create(Ts&&... args) { static T* create(Ts&&... args) {
return new T(std::forward<Ts>(args)...); using embedded_t =
typename std::conditional<
T::memory_cache_flag == provides_embedding,
rc_storage<T>,
T
>::type;
return unbox_rc_storage(new embedded_t(std::forward<Ts>(args)...));
} }
static inline memory_cache* get_cache_map_entry(const std::type_info*) { static inline memory_cache* get_cache_map_entry(const std::type_info*) {
return nullptr; return nullptr;
} }
}; };
#else // CAF_NO_MEM_MANAGEMENT #else // CAF_NO_MEM_MANAGEMENT
template <class T> template <class T>
class basic_memory_cache : public memory_cache { class basic_memory_cache : public memory_cache {
public:
static constexpr size_t ne = s_alloc_size / sizeof(T); static constexpr size_t ne = s_alloc_size / sizeof(T);
static constexpr size_t ms = ne < s_min_elements ? s_min_elements : ne; static constexpr size_t ms = ne < s_min_elements ? s_min_elements : ne;
static constexpr size_t dsize = ms > s_max_elements ? s_max_elements : ms; static constexpr size_t dsize = ms > s_max_elements ? s_max_elements : ms;
struct wrapper : instance_wrapper { static_assert(dsize > 0, "dsize == 0");
ref_counted* parent;
union { using embedded_t =
T instance; typename std::conditional<
T::memory_cache_flag == needs_embedding,
embedded<T>,
T
>::type;
struct wrapper {
union {
embedded_t instance;
}; };
wrapper() : parent(nullptr) {}
~wrapper() {}
void destroy() { instance.~T(); }
void deallocate() { parent->deref(); }
wrapper() {
// nop
}
~wrapper() {
// nop
}
}; };
class storage : public ref_counted { class storage : public ref_counted {
public: public:
storage() : m_pos(0) {
storage() { // nop
for (auto& elem : data) {
// each instance has a reference to its parent
elem.parent = this;
ref(); // deref() is called in wrapper::deallocate
}
} }
using iterator = wrapper*; ~storage() {
// nop
}
iterator begin() { return data; } bool has_next() {
return m_pos < dsize;
}
iterator end() { return begin() + dsize; } embedded_t* next() {
return &(m_data[m_pos++].instance);
}
private: private:
size_t m_pos;
wrapper data[dsize]; wrapper m_data[dsize];
}; };
public: embedded_storage new_embedded_storage() override {
// allocate cache on-the-fly
std::vector<wrapper*> cached_elements; if (!m_cache) {
m_cache.reset(new storage);
basic_memory_cache() { cached_elements.reserve(dsize); } }
auto res = m_cache->next();
~basic_memory_cache() { if (m_cache->has_next()) {
for (auto e : cached_elements) e->deallocate(); return {m_cache, res};
}
void* downcast(memory_managed* ptr) { return static_cast<T*>(ptr); }
void release_instance(void* vptr) override {
CAF_REQUIRE(vptr != nullptr);
auto ptr = reinterpret_cast<T*>(vptr);
CAF_REQUIRE(ptr->outer_memory != nullptr);
auto wptr = static_cast<wrapper*>(ptr->outer_memory);
wptr->destroy();
wptr->deallocate();
}
std::pair<instance_wrapper*, void*> new_instance() override {
if (cached_elements.empty()) {
auto elements = new storage;
for (auto i = elements->begin(); i != elements->end(); ++i) {
cached_elements.push_back(i);
}
} }
wrapper* wptr = cached_elements.back(); // we got the last element out of the cache; pass the reference to the
cached_elements.pop_back(); // client to avoid pointless increase/decrease ops on the reference count
return std::make_pair(wptr, &(wptr->instance)); embedded_storage result;
result.first.reset(m_cache.release(), false);
result.second = res;
return result;
} }
private:
intrusive_ptr<storage> m_cache;
}; };
class memory { class memory {
...@@ -198,13 +187,19 @@ class memory { ...@@ -198,13 +187,19 @@ class memory {
public: public:
// Allocates storage, initializes a new object, and returns the new instance. // Allocates storage, initializes a new object, and returns the new instance.
template <class T, class... Ts> template <class T, class... Vs>
static T* create(Ts&&... args) { static T* create(Vs&&... vs) {
using embedded_t =
typename std::conditional<
T::memory_cache_flag == needs_embedding,
embedded<T>,
T
>::type;
auto mc = get_or_set_cache_map_entry<T>(); auto mc = get_or_set_cache_map_entry<T>();
auto p = mc->new_instance(); auto es = mc->new_embedded_storage();
auto result = new (p.second) T(std::forward<Ts>(args)...); auto ptr = reinterpret_cast<embedded_t*>(es.second);
result->outer_memory = p.first; new (ptr) embedded_t(std::move(es.first), std::forward<Vs>(vs)...);
return result; return ptr;
} }
static memory_cache* get_cache_map_entry(const std::type_info* tinf); static memory_cache* get_cache_map_entry(const std::type_info* tinf);
...@@ -212,7 +207,7 @@ class memory { ...@@ -212,7 +207,7 @@ class memory {
private: private:
static void add_cache_map_entry(const std::type_info* tinf, static void add_cache_map_entry(const std::type_info* tinf,
memory_cache* instance); memory_cache* instance);
template <class T> template <class T>
static inline memory_cache* get_or_set_cache_map_entry() { static inline memory_cache* get_or_set_cache_map_entry() {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_DETAIL_MEMORY_CACHE_FLAG_TYPE
#define CAF_DETAIL_MEMORY_CACHE_FLAG_TYPE
namespace caf {
namespace detail {
enum memory_cache_flag_type {
needs_embedding,
provides_embedding
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_MEMORY_CACHE_FLAG_TYPE
...@@ -17,69 +17,78 @@ ...@@ -17,69 +17,78 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_MIXIN_SINGLE_TIMEOUT_HPP #ifndef CAF_DETAIL_PAIR_STORAGE_HPP
#define CAF_MIXIN_SINGLE_TIMEOUT_HPP #define CAF_DETAIL_PAIR_STORAGE_HPP
#include "caf/message.hpp" #include "caf/extend.hpp"
#include "caf/duration.hpp" #include "caf/ref_counted.hpp"
#include "caf/system_messages.hpp"
#include "caf/detail/embedded.hpp"
#include "caf/detail/memory_cache_flag_type.hpp"
namespace caf { namespace caf {
namespace mixin { namespace detail {
/** // Reduces memory allocations by placing two independent types on one
* Mixin for actors using a non-nestable message processing. // memory block. Typical use case is to combine the content of a message
*/ // (tuple_vals) with its "context" (message ID and sender; mailbox_element).
template <class Base, class Subtype> //
class single_timeout : public Base { // pair_storage<mailbox_element, tuple_vals<Ts...>>:
public: //
using super = Base; // +-----------------------------------------------+
// | |
// | +------------+ |
// | | | intrusive_ptr | intrusive_ptr
// v v | |
// +------------+-------------------+---------------------+
// | refcount | mailbox_element | tuple_vals<Ts...> |
// +------------+-------------------+---------------------+
// ^ ^
// | |
// unique_ptr<mailbox_element, |
// detail::disposer> |
// |
// |
// intrusive_ptr<message_data>
using combined_type = single_timeout; template <class FirstType, class SecondType>
class pair_storage {
public:
union { embedded<FirstType> first; };
union { embedded<SecondType> second; };
template <class... Ts> template <class... Vs>
single_timeout(Ts&&... args) pair_storage(intrusive_ptr<ref_counted> storage,
: super(std::forward<Ts>(args)...), std::integral_constant<size_t, 0>, Vs&&... vs)
m_timeout_id(0) { : first(storage),
second(std::move(storage), std::forward<Vs>(vs)...) {
// nop // nop
} }
void request_timeout(const duration& d) { template <class V0, class... Vs>
if (d.valid()) { pair_storage(intrusive_ptr<ref_counted> storage,
this->has_timeout(true); std::integral_constant<size_t, 1>, V0&& v0, Vs&&... vs)
auto tid = ++m_timeout_id; : first(storage, std::forward<V0>(v0)),
auto msg = make_message(timeout_msg{tid}); second(std::move(storage), std::forward<Vs>(vs)...) {
if (d.is_zero()) { // nop
// 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 { template <class V0, class V1, class... Vs>
return m_timeout_id; pair_storage(intrusive_ptr<ref_counted> storage,
std::integral_constant<size_t, 2>, V0&& v0, V1&& v1, Vs&&... vs)
: first(storage, std::forward<V0>(v0), std::forward<V1>(v1)),
second(std::move(storage), std::forward<Vs>(vs)...) {
// nop
} }
void reset_timeout() { ~pair_storage() {
this->has_timeout(false); // nop
} }
protected: static constexpr auto memory_cache_flag = provides_embedding;
uint32_t m_timeout_id;
}; };
} // namespace mixin } // namespace detail
} // namespace caf } // namespace caf
#endif // CAF_MIXIN_SINGLE_TIMEOUT_HPP #endif // CAF_DETAIL_PAIR_STORAGE_HPP
...@@ -54,7 +54,14 @@ class proper_actor_base : public Policies::resume_policy::template ...@@ -54,7 +54,14 @@ 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 {
scheduling_policy().enqueue(dptr(), sender, mid, msg, eu); auto d = dptr();
scheduling_policy().enqueue(d, mailbox_element::make(sender, mid,
std::move(msg)),
eu);
}
void enqueue(mailbox_element_ptr what, execution_unit* eu) override {
scheduling_policy().enqueue(dptr(), std::move(what), eu);
} }
void launch(bool hide, bool lazy, execution_unit* eu) { void launch(bool hide, bool lazy, execution_unit* eu) {
...@@ -80,7 +87,7 @@ class proper_actor_base : public Policies::resume_policy::template ...@@ -80,7 +87,7 @@ class proper_actor_base : public Policies::resume_policy::template
// member functions from priority policy // member functions from priority policy
unique_mailbox_element_pointer next_message() { mailbox_element_ptr next_message() {
return priority_policy().next_message(dptr()); return priority_policy().next_message(dptr());
} }
...@@ -88,7 +95,7 @@ class proper_actor_base : public Policies::resume_policy::template ...@@ -88,7 +95,7 @@ class proper_actor_base : public Policies::resume_policy::template
return priority_policy().has_next_message(dptr()); return priority_policy().has_next_message(dptr());
} }
void push_to_cache(unique_mailbox_element_pointer ptr) { void push_to_cache(mailbox_element_ptr ptr) {
priority_policy().push_to_cache(dptr(), std::move(ptr)); priority_policy().push_to_cache(dptr(), std::move(ptr));
} }
...@@ -99,7 +106,7 @@ class proper_actor_base : public Policies::resume_policy::template ...@@ -99,7 +106,7 @@ class proper_actor_base : public Policies::resume_policy::template
// member functions from invoke policy // member functions from invoke policy
template <class PartialFunctionOrBehavior> template <class PartialFunctionOrBehavior>
policy::invoke_message_result invoke_message(mailbox_element& me, policy::invoke_message_result invoke_message(mailbox_element_ptr& me,
PartialFunctionOrBehavior& fun, PartialFunctionOrBehavior& fun,
message_id awaited_response) { message_id awaited_response) {
return invoke_policy().invoke_message(dptr(), me, fun, awaited_response); return invoke_policy().invoke_message(dptr(), me, fun, awaited_response);
...@@ -118,7 +125,7 @@ class proper_actor_base : public Policies::resume_policy::template ...@@ -118,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();
} }
...@@ -150,7 +157,7 @@ class proper_actor ...@@ -150,7 +157,7 @@ class proper_actor
// required by event_based_resume::mixin::resume // required by event_based_resume::mixin::resume
policy::invoke_message_result invoke_message(mailbox_element& me) { policy::invoke_message_result invoke_message(mailbox_element_ptr& me) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
auto bhvr = this->bhvr_stack().back(); auto bhvr = this->bhvr_stack().back();
auto mid = this->bhvr_stack().back_id(); auto mid = this->bhvr_stack().back_id();
...@@ -214,11 +221,13 @@ class proper_actor<Base, Policies, true> ...@@ -214,11 +221,13 @@ class proper_actor<Base, Policies, true>
for (;;) { for (;;) {
this->await_ready(); this->await_ready();
auto msg = this->next_message(); auto msg = this->next_message();
switch (this->invoke_message(*msg, bhvr, mid)) { switch (this->invoke_message(msg, bhvr, mid)) {
case policy::im_success: case policy::im_success:
return; return;
case policy::im_skipped: case policy::im_skipped:
this->push_to_cache(std::move(msg)); if (msg) {
this->push_to_cache(std::move(msg));
}
break; break;
default: default:
// delete msg // delete msg
...@@ -237,8 +246,6 @@ class proper_actor<Base, Policies, true> ...@@ -237,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));
} }
...@@ -256,14 +263,13 @@ class proper_actor<Base, Policies, true> ...@@ -256,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);
} }
......
...@@ -267,7 +267,7 @@ class has_static_type_name { ...@@ -267,7 +267,7 @@ class has_static_type_name {
private: private:
template <class U, template <class U,
class = typename std::enable_if< class = typename std::enable_if<
!std::is_member_pointer<decltype(&U::is_baz)>::value !std::is_member_pointer<decltype(&U::static_type_name)>::value
>::type> >::type>
static std::true_type sfinae_fun(int); static std::true_type sfinae_fun(int);
template <class> template <class>
...@@ -276,6 +276,22 @@ class has_static_type_name { ...@@ -276,6 +276,22 @@ class has_static_type_name {
static constexpr bool value = decltype(sfinae_fun<T>(0))::value; static constexpr bool value = decltype(sfinae_fun<T>(0))::value;
}; };
/**
* Checks whether `T::memory_cache_flag` exists.
*/
template <class T>
class is_memory_cached {
private:
template <class U, bool = U::memory_cache_flag>
static std::true_type check(int);
template <class>
static std::false_type check(...);
public:
static constexpr bool value = decltype(check<T>(0))::value;
};
/** /**
* Returns either `T` or `T::type` if `T` is an option. * Returns either `T` or `T::type` if `T` is an option.
*/ */
......
...@@ -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:
/** /**
......
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#ifndef CAF_FWD_HPP #ifndef CAF_FWD_HPP
#define CAF_FWD_HPP #define CAF_FWD_HPP
#include <memory>
#include <cstdint> #include <cstdint>
namespace caf { namespace caf {
...@@ -45,9 +46,11 @@ class execution_unit; ...@@ -45,9 +46,11 @@ class execution_unit;
class abstract_actor; class abstract_actor;
class abstract_group; class abstract_group;
class blocking_actor; class blocking_actor;
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
...@@ -90,6 +93,7 @@ namespace scheduler { ...@@ -90,6 +93,7 @@ namespace scheduler {
namespace detail { namespace detail {
class logging; class logging;
class disposer;
class singletons; class singletons;
class message_data; class message_data;
class group_manager; class group_manager;
...@@ -97,6 +101,8 @@ namespace detail { ...@@ -97,6 +101,8 @@ namespace detail {
class uniform_type_info_map; class uniform_type_info_map;
} // namespace detail } // namespace detail
using mailbox_element_ptr = std::unique_ptr<mailbox_element, detail::disposer>;
} // namespace caf } // namespace caf
#endif // CAF_FWD_HPP #endif // CAF_FWD_HPP
...@@ -46,12 +46,12 @@ ...@@ -46,12 +46,12 @@
#include "caf/message_priority.hpp" #include "caf/message_priority.hpp"
#include "caf/check_typed_input.hpp" #include "caf/check_typed_input.hpp"
#include "caf/mixin/memory_cached.hpp"
#include "caf/detail/logging.hpp" #include "caf/detail/logging.hpp"
#include "caf/detail/disposer.hpp"
#include "caf/detail/behavior_stack.hpp" #include "caf/detail/behavior_stack.hpp"
#include "caf/detail/typed_actor_util.hpp" #include "caf/detail/typed_actor_util.hpp"
#include "caf/detail/single_reader_queue.hpp" #include "caf/detail/single_reader_queue.hpp"
#include "caf/detail/memory_cache_flag_type.hpp"
namespace caf { namespace caf {
...@@ -60,18 +60,18 @@ class sync_handle_helper; ...@@ -60,18 +60,18 @@ 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 extend<abstract_actor>::with<mixin::memory_cached> { class local_actor : public abstract_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>;
static constexpr auto memory_cache_flag = detail::needs_embedding;
~local_actor(); ~local_actor();
/************************************************************************** /**************************************************************************
* spawn untyped actors * * spawn untyped actors *
**************************************************************************/ **************************************************************************/
template <class C, spawn_options Os = no_spawn_options, class... Ts> template <class C, spawn_options Os = no_spawn_options, class... Ts>
...@@ -107,7 +107,7 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> { ...@@ -107,7 +107,7 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
} }
/************************************************************************** /**************************************************************************
* spawn typed actors * * spawn typed actors *
**************************************************************************/ **************************************************************************/
template <class C, spawn_options Os = no_spawn_options, class... Ts> template <class C, spawn_options Os = no_spawn_options, class... Ts>
...@@ -142,35 +142,33 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> { ...@@ -142,35 +142,33 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
**************************************************************************/ **************************************************************************/
/** /**
* Sends `{what...} to `whom` using the priority `prio`. * Sends `{vs...} to `whom` using the priority `prio`.
*/ */
template <class... Ts> template <class... Vs>
inline void send(message_priority prio, const channel& whom, Ts&&... what) { inline void send(message_priority prio, const channel& whom, Vs&&... vs) {
static_assert(sizeof...(Ts) > 0, "sizeof...(Ts) == 0"); static_assert(sizeof...(Vs) > 0, "sizeof...(Vs) == 0");
send_impl(prio, whom, make_message(std::forward<Ts>(what)...)); fast_send(prio, whom, std::forward<Vs>(vs)...);
} }
/** /**
* Sends `{what...} to `whom`. * Sends `{vs...} to `whom` using normal priority.
*/ */
template <class... Ts> template <class... Vs>
inline void send(const channel& whom, Ts&&... what) { inline void send(const channel& whom, Vs&&... vs) {
static_assert(sizeof...(Ts) > 0, "sizeof...(Ts) == 0"); static_assert(sizeof...(Vs) > 0, "sizeof...(Vs) == 0");
send_impl(message_priority::normal, whom, fast_send(message_priority::normal, whom, std::forward<Vs>(vs)...);
make_message(std::forward<Ts>(what)...));
} }
/** /**
* Sends `{what...} to `whom`. * Sends `{what...} to `whom`.
*/ */
template <class... Rs, class... Ts> template <class... Rs, class... Vs>
void send(const typed_actor<Rs...>& whom, Ts... what) { void send(const typed_actor<Rs...>& whom, Vs&&... vs) {
check_typed_input(whom, check_typed_input(whom,
detail::type_list<typename detail::implicit_conversions< detail::type_list<typename detail::implicit_conversions<
typename std::decay<Ts>::type typename std::decay<Vs>::type
>::type...>{}); >::type...>{});
send_impl(message_priority::normal, actor{whom.m_ptr.get()}, fast_send(message_priority::normal, whom, std::forward<Vs>(vs)...);
make_message(std::forward<Ts>(what)...));
} }
/** /**
...@@ -201,7 +199,7 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> { ...@@ -201,7 +199,7 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
void delayed_send(message_priority prio, const channel& whom, void delayed_send(message_priority prio, const channel& whom,
const duration& rtime, Ts&&... args) { const duration& rtime, Ts&&... args) {
delayed_send_impl(prio, whom, rtime, delayed_send_impl(prio, whom, rtime,
make_message(std::forward<Ts>(args)...)); make_message(std::forward<Ts>(args)...));
} }
/** /**
...@@ -297,16 +295,12 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> { ...@@ -297,16 +295,12 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
* Returns the last message that was dequeued from the actor's mailbox. * Returns the last message that was dequeued from the actor's mailbox.
* @warning Only set during callback invocation. * @warning Only set during callback invocation.
*/ */
inline message& last_dequeued() { message& last_dequeued();
return m_current_node->msg;
}
/** /**
* Returns the address of the last sender of the last dequeued message. * Returns the address of the last sender of the last dequeued message.
*/ */
inline actor_addr& last_sender() { actor_addr& last_sender();
return m_current_node->sender;
}
/** /**
* Adds a unidirectional `monitor` to `whom`. * Adds a unidirectional `monitor` to `whom`.
...@@ -443,12 +437,8 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> { ...@@ -443,12 +437,8 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
return res; return res;
} }
inline void current_node(mailbox_element* ptr) { inline mailbox_element_ptr& current_element() {
this->m_current_node = ptr; return m_current_element;
}
inline mailbox_element* current_node() {
return this->m_current_node;
} }
inline message_id new_request_id() { inline message_id new_request_id() {
...@@ -495,7 +485,7 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> { ...@@ -495,7 +485,7 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
// returns 0 if last_dequeued() is an asynchronous or sync request message, // returns 0 if last_dequeued() is an asynchronous or sync request message,
// a response id generated from the request id otherwise // a response id generated from the request id otherwise
inline message_id get_response_id() { inline message_id get_response_id() {
auto mid = m_current_node->mid; auto mid = m_current_element->mid;
return (mid.is_request()) ? mid.response_id() : message_id(); return (mid.is_request()) ? mid.response_id() : message_id();
} }
...@@ -523,15 +513,6 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> { ...@@ -523,15 +513,6 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
void cleanup(uint32_t reason); void cleanup(uint32_t reason);
inline mailbox_element* dummy_node() {
return &m_dummy_node;
}
template <class... Ts>
inline mailbox_element* new_mailbox_element(Ts&&... args) {
return mailbox_element::create(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;
...@@ -539,12 +520,13 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> { ...@@ -539,12 +520,13 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
// identifies all IDs of sync messages waiting for a response // identifies all IDs of sync messages waiting for a response
std::forward_list<message_id> m_pending_responses; std::forward_list<message_id> m_pending_responses;
// "default value" for m_current_node // "default value" for m_current_element
mailbox_element m_dummy_node; actor_addr m_dummy_sender;
message m_dummy_message;
// points to m_dummy_node if no callback is currently invoked, // points to m_dummy_node if no callback is currently invoked,
// points to the node under processing otherwise // points to the node under processing otherwise
mailbox_element* m_current_node; mailbox_element_ptr m_current_element;
// set by quit // set by quit
uint32_t m_planned_exit_reason; uint32_t m_planned_exit_reason;
...@@ -552,10 +534,40 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> { ...@@ -552,10 +534,40 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
/** @endcond */ /** @endcond */
private: private:
void send_impl(message_priority prio, const channel& dest, message&& what); template <class... Vs>
void fast_send_impl(message_priority mp, abstract_channel* dest, Vs&&... vs) {
if (!dest) {
return;
}
dest->enqueue(mailbox_element::make_joint(address(), message_id::make(mp),
std::forward<Vs>(vs)...),
host());
}
template <class T, class V0, class... Vs>
typename std::enable_if<
!std::is_same<typename std::decay<V0>::type, message>::value
>::type
fast_send(message_priority mp, const T& dest, V0&& v0, Vs&&... vs) {
fast_send_impl(mp, actor_cast<abstract_channel*>(dest),
std::forward<V0>(v0), std::forward<Vs>(vs)...);
}
template <class T>
void fast_send(message_priority mp, const T& dest, message what) {
send_impl(mp, actor_cast<abstract_channel*>(dest), std::move(what));
}
void send_impl(message_priority prio, abstract_channel* dest, message&& what);
template <class T>
void send_impl(message_priority prio, const T& dest, message&& what) {
send_impl(prio, actor_cast<abstract_channel*>(dest), std::move(what));
}
void delayed_send_impl(message_priority prio, const channel& whom, void delayed_send_impl(message_priority prio, const channel& whom,
const duration& rtime, message data); const duration& rtime, message data);
using super = combined_type;
std::function<void()> m_sync_failure_handler; std::function<void()> m_sync_failure_handler;
std::function<void()> m_sync_timeout_handler; std::function<void()> m_sync_timeout_handler;
}; };
......
...@@ -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
...@@ -17,10 +17,10 @@ ...@@ -17,10 +17,10 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_RECURSIVE_QUEUE_NODE_HPP #ifndef CAF_MAILBOX_ELEMENT_HPP
#define CAF_RECURSIVE_QUEUE_NODE_HPP #define CAF_MAILBOX_ELEMENT_HPP
#include <cstdint> #include <cstddef>
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
...@@ -28,16 +28,20 @@ ...@@ -28,16 +28,20 @@
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/mixin/memory_cached.hpp" #include "caf/detail/memory.hpp"
#include "caf/detail/embedded.hpp"
#include "caf/detail/disposer.hpp"
#include "caf/detail/tuple_vals.hpp"
#include "caf/detail/pair_storage.hpp"
#include "caf/detail/message_data.hpp"
#include "caf/detail/memory_cache_flag_type.hpp"
// needs access to constructor + destructor to initialize m_dummy_node
namespace caf { namespace caf {
class local_actor; class mailbox_element : public memory_managed {
class mailbox_element : public extend<memory_managed>::
with<mixin::memory_cached> {
public: public:
static constexpr auto memory_cache_flag = detail::needs_embedding;
mailbox_element* next; // intrusive next pointer mailbox_element* next; // intrusive next pointer
mailbox_element* prev; // intrusive previous pointer mailbox_element* prev; // intrusive previous pointer
bool marked; // denotes if this node is currently processed bool marked; // denotes if this node is currently processed
...@@ -46,6 +50,7 @@ class mailbox_element : public extend<memory_managed>:: ...@@ -46,6 +50,7 @@ class mailbox_element : public extend<memory_managed>::
message msg; // 'content field' message msg; // 'content field'
mailbox_element(); mailbox_element();
mailbox_element(actor_addr sender, message_id id);
mailbox_element(actor_addr sender, message_id id, message data); mailbox_element(actor_addr sender, message_id id, message data);
~mailbox_element(); ~mailbox_element();
...@@ -55,10 +60,24 @@ class mailbox_element : public extend<memory_managed>:: ...@@ -55,10 +60,24 @@ class mailbox_element : public extend<memory_managed>::
mailbox_element& operator=(mailbox_element&&) = delete; mailbox_element& operator=(mailbox_element&&) = delete;
mailbox_element& operator=(const mailbox_element&) = delete; mailbox_element& operator=(const mailbox_element&) = delete;
template <class T> using unique_ptr = std::unique_ptr<mailbox_element, detail::disposer>;
static mailbox_element* create(actor_addr sender, message_id id, T&& data) {
return detail::memory::create<mailbox_element>(std::move(sender), id, static unique_ptr make(actor_addr sender, message_id id, message msg);
std::forward<T>(data));
template <class... Vs>
static unique_ptr make_joint(actor_addr sender, message_id id, Vs&&... vs) {
using value_storage =
detail::tuple_vals<
typename unbox_message_element<
typename detail::strip_and_convert<Vs>::type
>::type...
>;
std::integral_constant<size_t, 2> tk;
using storage = detail::pair_storage<mailbox_element, value_storage>;
auto ptr = detail::memory::create<storage>(tk, std::move(sender), id,
std::forward<Vs>(vs)...);
ptr->first.msg.reset(&(ptr->second));
return unique_ptr{&(ptr->first)};
} }
inline bool is_high_priority() const { inline bool is_high_priority() const {
...@@ -66,9 +85,8 @@ class mailbox_element : public extend<memory_managed>:: ...@@ -66,9 +85,8 @@ class mailbox_element : public extend<memory_managed>::
} }
}; };
using unique_mailbox_element_pointer = using mailbox_element_ptr = std::unique_ptr<mailbox_element, detail::disposer>;
std::unique_ptr<mailbox_element, detail::disposer>;
} // namespace caf } // namespace caf
#endif // CAF_RECURSIVE_QUEUE_NODE_HPP #endif // CAF_MAILBOX_ELEMENT_HPP
...@@ -22,10 +22,6 @@ ...@@ -22,10 +22,6 @@
namespace caf { namespace caf {
namespace detail {
struct disposer;
}
/** /**
* This base enables derived classes to enforce a different * This base enables derived classes to enforce a different
* allocation strategy than new/delete by providing a virtual * allocation strategy than new/delete by providing a virtual
...@@ -33,8 +29,6 @@ struct disposer; ...@@ -33,8 +29,6 @@ struct disposer;
*/ */
class memory_managed { class memory_managed {
public: public:
friend struct detail::disposer;
/** /**
* Default implementations calls `delete this, but can * Default implementations calls `delete this, but can
* be overriden in case deletion depends on some condition or * be overriden in case deletion depends on some condition or
......
...@@ -307,8 +307,6 @@ class message { ...@@ -307,8 +307,6 @@ class message {
explicit message(const data_ptr& vals); explicit message(const data_ptr& vals);
void reset();
inline data_ptr& vals() { inline data_ptr& vals() {
return m_vals; return m_vals;
} }
...@@ -329,6 +327,10 @@ class message { ...@@ -329,6 +327,10 @@ class message {
m_vals.detach(); m_vals.detach();
} }
void reset(raw_ptr new_ptr = nullptr);
void swap(message& other);
inline std::string tuple_type_names() const { inline std::string tuple_type_names() const {
return m_vals->tuple_type_names(); return m_vals->tuple_type_names();
} }
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include <cstdint> #include <cstdint>
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/message_priority.hpp"
#include "caf/detail/comparable.hpp" #include "caf/detail/comparable.hpp"
namespace caf { namespace caf {
...@@ -114,6 +115,11 @@ class message_id : detail::comparable<message_id> { ...@@ -114,6 +115,11 @@ class message_id : detail::comparable<message_id> {
return result; return result;
} }
static inline message_id make(message_priority prio) {
message_id res;
return prio == message_priority::high ? res.with_high_priority() : res;
}
long compare(const message_id& other) const { long compare(const message_id& other) const {
return (m_value == other.m_value) ? 0 return (m_value == other.m_value) ? 0
: (m_value < other.m_value ? -1 : 1); : (m_value < other.m_value ? -1 : 1);
......
...@@ -27,7 +27,6 @@ namespace caf { ...@@ -27,7 +27,6 @@ namespace caf {
enum class message_priority : uint32_t { enum class message_priority : uint32_t {
normal, normal,
high high
}; };
} // namespace caf } // namespace caf
......
...@@ -37,63 +37,55 @@ ...@@ -37,63 +37,55 @@
#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 {
message_pointer mptr;
struct event_type : public QEvent { event_type(message_pointer ptr)
: QEvent(static_cast<QEvent::Type>(EventId)), mptr(std::move(ptr)) {
message_pointer mptr; // nop
event_type(message_pointer ptr)
: QEvent(static_cast<QEvent::Type>(EventId)), mptr(std::move(ptr)) { }
};
template<typename... Ts>
actor_widget(Ts&&... args) : Base(std::forward<Ts>(args)...) {
m_companion.reset(detail::memory::create<actor_companion>());
m_companion->on_enqueue([=](message_pointer ptr) {
qApp->postEvent(this, new event_type(std::move(ptr)));
});
} }
};
template<typename T>
void set_message_handler(T pfun) { template <typename... Ts>
m_companion->become(pfun(m_companion.get())); actor_widget(Ts&&... args) : Base(std::forward<Ts>(args)...) {
} m_companion.reset(detail::memory::create<actor_companion>());
m_companion->on_enqueue([=](message_pointer ptr) {
virtual bool event(QEvent* event) { qApp->postEvent(this, new event_type(std::move(ptr)));
if (event->type() == static_cast<QEvent::Type>(EventId)) { });
auto ptr = dynamic_cast<event_type*>(event); }
if (ptr) {
m_invoke.invoke_message(m_companion.get(), template <typename T>
*ptr->mptr.get(), void set_message_handler(T pfun) {
m_companion->bhvr_stack().back(), m_companion->become(pfun(m_companion.get()));
m_companion->bhvr_stack().back_id()); }
return true;
} virtual bool event(QEvent* event) {
} if (event->type() == static_cast<QEvent::Type>(EventId)) {
return Base::event(event); auto ptr = dynamic_cast<event_type*>(event);
if (ptr) {
m_invoke.invoke_message(m_companion.get(), ptr->mptr,
m_companion->bhvr_stack().back(),
m_companion->bhvr_stack().back_id());
return true;
}
} }
return Base::event(event);
}
actor as_actor() const { actor as_actor() const {
return m_companion; return m_companion;
} }
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,52 +25,48 @@ ...@@ -25,52 +25,48 @@
#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);
} }
template <class... Ts> template <class... Ts>
...@@ -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
}
}; };
}; };
......
...@@ -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;
} }
......
...@@ -34,13 +34,11 @@ namespace caf { ...@@ -34,13 +34,11 @@ namespace caf {
namespace policy { namespace policy {
class cooperative_scheduling { class cooperative_scheduling {
public: public:
using timeout_type = int; using timeout_type = int;
template <class Actor> template <class Actor>
inline void launch(Actor* self, execution_unit* host, bool lazy) { void launch(Actor* self, execution_unit* host, bool lazy) {
// detached in scheduler::worker::run // detached in scheduler::worker::run
self->attach_to_scheduler(); self->attach_to_scheduler();
if (lazy) { if (lazy) {
...@@ -55,17 +53,17 @@ class cooperative_scheduling { ...@@ -55,17 +53,17 @@ class cooperative_scheduling {
} }
template <class Actor> template <class Actor>
void enqueue(Actor* self, const actor_addr& sender, message_id mid, void enqueue(Actor* self, mailbox_element_ptr ptr, execution_unit* eu) {
message& msg, execution_unit* eu) { auto mid = ptr->mid;
auto e = self->new_mailbox_element(sender, mid, std::move(msg)); auto sender = ptr->sender;
switch (self->mailbox().enqueue(e)) { switch (self->mailbox().enqueue(ptr.release())) {
case detail::enqueue_result::unblocked_reader: { case detail::enqueue_result::unblocked_reader: {
// re-schedule actor // re-schedule actor
if (eu) if (eu) {
eu->exec_later(self); eu->exec_later(self);
else } else {
detail::singletons::get_scheduling_coordinator()->enqueue( detail::singletons::get_scheduling_coordinator()->enqueue(self);
self); }
break; break;
} }
case detail::enqueue_result::queue_closed: { case detail::enqueue_result::queue_closed: {
...@@ -80,7 +78,6 @@ class cooperative_scheduling { ...@@ -80,7 +78,6 @@ class cooperative_scheduling {
break; break;
} }
} }
}; };
} // namespace policy } // namespace policy
......
...@@ -125,7 +125,7 @@ class event_based_resume { ...@@ -125,7 +125,7 @@ class event_based_resume {
for (size_t i = 0; i < max_throughput; ++i) { for (size_t i = 0; i < max_throughput; ++i) {
auto ptr = d->next_message(); auto ptr = d->next_message();
if (ptr) { if (ptr) {
switch (d->invoke_message(*ptr)) { switch (d->invoke_message(ptr)) {
case policy::im_success: case policy::im_success:
d->bhvr_stack().cleanup(); d->bhvr_stack().cleanup();
++handled_msgs; ++handled_msgs;
...@@ -144,6 +144,7 @@ class event_based_resume { ...@@ -144,6 +144,7 @@ class event_based_resume {
} }
break; break;
case policy::im_skipped: case policy::im_skipped:
CAF_REQUIRE(ptr != nullptr);
d->push_to_cache(std::move(ptr)); d->push_to_cache(std::move(ptr));
break; break;
case policy::im_dropped: case policy::im_dropped:
......
...@@ -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,15 +71,13 @@ class invoke_policy { ...@@ -85,15 +71,13 @@ 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& 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)) { switch (this->filter_msg(self, *node)) {
return im_skipped;
}
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");
return im_dropped; return im_dropped;
...@@ -113,7 +97,7 @@ class invoke_policy { ...@@ -113,7 +97,7 @@ class invoke_policy {
return im_success; return im_success;
case msg_type::timeout: { case msg_type::timeout: {
CAF_LOG_DEBUG("handle timeout message"); CAF_LOG_DEBUG("handle timeout message");
auto& tm = node.msg.get_as<timeout_msg>(0); auto& tm = node->msg.get_as<timeout_msg>(0);
self->handle_timeout(fun, tm.timeout_id); self->handle_timeout(fun, tm.timeout_id);
if (awaited_response.valid()) { if (awaited_response.valid()) {
self->mark_arrived(awaited_response); self->mark_arrived(awaited_response);
...@@ -126,12 +110,12 @@ class invoke_policy { ...@@ -126,12 +110,12 @@ class invoke_policy {
CAF_ANNOTATE_FALLTHROUGH; CAF_ANNOTATE_FALLTHROUGH;
case msg_type::sync_response: case msg_type::sync_response:
CAF_LOG_DEBUG("handle as synchronous response: " CAF_LOG_DEBUG("handle as synchronous response: "
<< CAF_TARG(node.msg, to_string) << ", " << CAF_TARG(node->msg, to_string) << ", "
<< 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) {
auto previous_node = dptr()->hm_begin(self, node); node.swap(self->current_element());
auto res = invoke_fun(self, node.msg, node.mid, 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 "
<< "with ID " << self->id()); << "with ID " << self->id());
...@@ -139,20 +123,18 @@ class invoke_policy { ...@@ -139,20 +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, previous_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()) {
auto previous_node = dptr()->hm_begin(self, node); node.swap(self->current_element());
auto res = invoke_fun(self, node.msg, node.mid, fun); auto res = invoke_fun(self, fun);
node.swap(self->current_element());
if (res) { if (res) {
dptr()->hm_cleanup(self, previous_node);
return im_success; return im_success;
} }
// no match (restore self members)
dptr()->hm_revert(self, previous_node);
} }
CAF_LOG_DEBUG_IF(awaited_response.valid(), CAF_LOG_DEBUG_IF(awaited_response.valid(),
"ignored message; await response: " "ignored message; await response: "
...@@ -163,10 +145,6 @@ class invoke_policy { ...@@ -163,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();
...@@ -180,14 +158,14 @@ class invoke_policy { ...@@ -180,14 +158,14 @@ class invoke_policy {
// - extracts response message from handler // - extracts response message from handler
// - returns true if fun was successfully invoked // - returns true if fun was successfully invoked
template <class Actor, class Fun, class MaybeResponseHdl = int> template <class Actor, class Fun, class MaybeResponseHdl = int>
optional<message> invoke_fun(Actor* self, message& msg, message_id& mid, optional<message> invoke_fun(Actor* self, Fun& fun,
Fun& fun,
MaybeResponseHdl hdl = MaybeResponseHdl{}) { MaybeResponseHdl hdl = MaybeResponseHdl{}) {
# ifdef CAF_LOG_LEVEL # ifdef CAF_LOG_LEVEL
auto msg_str = to_string(msg); auto msg_str = to_string(msg);
# endif # endif
CAF_LOG_TRACE(CAF_MARG(mid, integer_value) << ", msg = " << msg_str); CAF_LOG_TRACE(CAF_MARG(mid, integer_value) << ", msg = " << msg_str);
auto res = fun(msg); // might change mid auto mid = self->current_element()->mid;
auto res = fun(self->current_element()->msg);
CAF_LOG_DEBUG_IF(res, "actor did consume message: " << msg_str); CAF_LOG_DEBUG_IF(res, "actor did consume message: " << msg_str);
CAF_LOG_DEBUG_IF(!res, "actor did ignore message: " << msg_str); CAF_LOG_DEBUG_IF(!res, "actor did ignore message: " << msg_str);
if (!res) { if (!res) {
...@@ -244,20 +222,10 @@ class invoke_policy { ...@@ -244,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;
......
...@@ -52,11 +52,11 @@ class no_scheduling { ...@@ -52,11 +52,11 @@ class no_scheduling {
using timeout_type = std::chrono::high_resolution_clock::time_point; using timeout_type = std::chrono::high_resolution_clock::time_point;
template <class Actor> template <class Actor>
void enqueue(Actor* self, const actor_addr& sender, message_id mid, void enqueue(Actor* self, mailbox_element_ptr ptr, execution_unit*) {
message& msg, execution_unit*) { auto mid = ptr->mid;
auto ptr = self->new_mailbox_element(sender, mid, std::move(msg)); auto sender = ptr->sender;
// returns false if mailbox has been closed // returns false if mailbox has been closed
if (!self->mailbox().synchronized_enqueue(m_mtx, m_cv, ptr)) { if (!self->mailbox().synchronized_enqueue(m_mtx, m_cv, ptr.release())) {
if (mid.is_request()) { if (mid.is_request()) {
detail::sync_request_bouncer srb{self->exit_reason()}; detail::sync_request_bouncer srb{self->exit_reason()};
srb(sender, mid); srb(sender, mid);
......
...@@ -34,8 +34,10 @@ namespace policy { ...@@ -34,8 +34,10 @@ namespace policy {
class not_prioritizing { class not_prioritizing {
public: public:
template <class Actor> template <class Actor>
unique_mailbox_element_pointer next_message(Actor* self) { typename Actor::mailbox_type::unique_pointer next_message(Actor* self) {
return unique_mailbox_element_pointer{self->mailbox().try_pop()}; typename Actor::mailbox_type::unique_pointer result;
result.reset(self->mailbox().try_pop());
return result;
} }
template <class Actor> template <class Actor>
...@@ -44,30 +46,18 @@ class not_prioritizing { ...@@ -44,30 +46,18 @@ class not_prioritizing {
} }
template <class Actor> template <class Actor>
void push_to_cache(Actor* self, unique_mailbox_element_pointer ptr) { void push_to_cache(Actor* self,
typename Actor::mailbox_type::unique_pointer ptr) {
self->mailbox().cache().push_second_back(ptr.release()); self->mailbox().cache().push_second_back(ptr.release());
} }
template <class Actor, class... Ts> template <class Actor, class... Vs>
bool invoke_from_cache(Actor* self, Ts&... args) { bool invoke_from_cache(Actor* self, Vs&... args) {
auto& cache = self->mailbox().cache(); auto& cache = self->mailbox().cache();
auto i = cache.second_begin(); auto i = cache.second_begin();
auto e = cache.second_end(); auto e = cache.second_end();
CAF_LOG_DEBUG(std::distance(i, e) << " elements in cache"); CAF_LOG_DEBUG(std::distance(i, e) << " elements in cache");
while (i != e) { return cache.invoke(self, i, e, args...);
switch (self->invoke_message(*i, args...)) {
case im_dropped:
i = cache.erase(i);
break;
case im_success:
cache.erase(i);
return true;
default:
++i;
break;
}
}
return false;
} }
}; };
......
...@@ -78,8 +78,9 @@ class prioritizing { ...@@ -78,8 +78,9 @@ class prioritizing {
} }
template <class Actor> template <class Actor>
void push_to_cache(Actor* self, unique_mailbox_element_pointer ptr) { void push_to_cache(Actor* self,
auto high_prio = [](const mailbox_element& val) { typename Actor::mailbox_type::unique_pointer ptr) {
auto high_prio = [](const typename Actor::mailbox_type::value_type& val) {
return val.is_high_priority(); return val.is_high_priority();
}; };
auto& cache = self->mailbox().cache(); auto& cache = self->mailbox().cache();
......
...@@ -35,7 +35,7 @@ class priority_policy { ...@@ -35,7 +35,7 @@ class priority_policy {
* Returns the next message from the mailbox or `nullptr` if it's empty. * Returns the next message from the mailbox or `nullptr` if it's empty.
*/ */
template <class Actor> template <class Actor>
unique_mailbox_element_pointer next_message(Actor* self); mailbox_element_ptr next_message(Actor* self);
/** /**
* Queries whether the mailbox is not empty. * Queries whether the mailbox is not empty.
...@@ -47,7 +47,7 @@ class priority_policy { ...@@ -47,7 +47,7 @@ class priority_policy {
* Stores the message in a cache for later retrieval. * Stores the message in a cache for later retrieval.
*/ */
template <class Actor> template <class Actor>
void push_to_cache(Actor* self, unique_mailbox_element_pointer ptr); void push_to_cache(Actor* self, mailbox_element_ptr ptr);
/** /**
* Removes the first element from the cache matching predicate `p`. * Removes the first element from the cache matching predicate `p`.
......
...@@ -86,12 +86,11 @@ class scheduling_policy { ...@@ -86,12 +86,11 @@ class scheduling_policy {
timed_fetch_result fetch_messages(Actor* self, F cb, timeout_type abs_time); timed_fetch_result fetch_messages(Actor* self, F cb, timeout_type abs_time);
/** /**
* Enqueues given message to the actor's mailbox and take any * Enqueues given message to the actor's mailbox and takes any
* steps to resume the actor if it's currently blocked. * steps necessary to resume the actor if it's currently blocked.
*/ */
template <class Actor> template <class Actor>
void enqueue(Actor* self, const actor_addr& sender, message_id mid, void enqueue(Actor* self, mailbox_element_ptr ptr, execution_unit* host);
message& msg, execution_unit* host);
/** /**
* Starts the given actor either by launching a thread or enqueuing * Starts the given actor either by launching a thread or enqueuing
......
...@@ -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...>;
......
...@@ -18,6 +18,8 @@ ...@@ -18,6 +18,8 @@
******************************************************************************/ ******************************************************************************/
#include "caf/abstract_channel.hpp" #include "caf/abstract_channel.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/detail/singletons.hpp" #include "caf/detail/singletons.hpp"
namespace caf { namespace caf {
...@@ -44,6 +46,10 @@ abstract_channel::~abstract_channel() { ...@@ -44,6 +46,10 @@ abstract_channel::~abstract_channel() {
// nop // nop
} }
void abstract_channel::enqueue(mailbox_element_ptr what, execution_unit* host) {
enqueue(what->sender, what->mid, what->msg, host);
}
bool abstract_channel::is_remote() const { bool abstract_channel::is_remote() const {
return m_node != singletons::get_node_id(); return m_node != singletons::get_node_id();
} }
......
...@@ -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;
...@@ -76,27 +75,27 @@ inline void insert_dmsg(Map& storage, const duration& d, Ts&&... vs) { ...@@ -76,27 +75,27 @@ inline void insert_dmsg(Map& storage, const duration& d, Ts&&... vs) {
storage.insert(std::make_pair(std::move(tout), std::move(dmsg))); storage.insert(std::make_pair(std::move(tout), std::move(dmsg)));
} }
class timer_actor final : public detail::proper_actor<blocking_actor, class timer_actor : public detail::proper_actor<blocking_actor,
timer_actor_policies>, timer_actor_policies>,
public spawn_as_is { public spawn_as_is {
public: public:
inline unique_mailbox_element_pointer dequeue() { inline mailbox_element_ptr dequeue() {
await_data(); await_data();
return next_message(); return next_message();
} }
inline unique_mailbox_element_pointer try_dequeue(const hrc::time_point& tp) { inline mailbox_element_ptr try_dequeue(const hrc::time_point& tp) {
if (scheduling_policy().await_data(this, tp)) { if (scheduling_policy().await_data(this, tp)) {
return next_message(); return next_message();
} }
return unique_mailbox_element_pointer{}; return mailbox_element_ptr{};
} }
void act() override { void act() override {
trap_exit(true); trap_exit(true);
// setup & local variables // setup & local variables
bool done = false; bool done = false;
unique_mailbox_element_pointer msg_ptr; mailbox_element_ptr msg_ptr;
std::multimap<hrc::time_point, delayed_msg> messages; std::multimap<hrc::time_point, delayed_msg> messages;
// message handling rules // message handling rules
message_handler mfun{ message_handler mfun{
......
...@@ -32,9 +32,7 @@ namespace caf { ...@@ -32,9 +32,7 @@ namespace caf {
// later on in spawn(); this prevents subtle bugs that lead to segfaults, // later on in spawn(); this prevents subtle bugs that lead to segfaults,
// e.g., when calling address() in the ctor of a derived class // e.g., when calling address() in the ctor of a derived class
local_actor::local_actor() local_actor::local_actor()
: super(size_t{1}), : abstract_actor(size_t{1}),
m_dummy_node(),
m_current_node(&m_dummy_node),
m_planned_exit_reason(exit_reason::not_exited) { m_planned_exit_reason(exit_reason::not_exited) {
// nop // nop
} }
...@@ -43,6 +41,14 @@ local_actor::~local_actor() { ...@@ -43,6 +41,14 @@ local_actor::~local_actor() {
// nop // nop
} }
message& local_actor::last_dequeued() {
return m_current_element ? m_current_element->msg : m_dummy_message;
}
actor_addr& local_actor::last_sender() {
return m_current_element ? m_current_element->sender : m_dummy_sender;
}
void local_actor::monitor(const actor_addr& whom) { void local_actor::monitor(const actor_addr& whom) {
if (whom == invalid_actor_addr) { if (whom == invalid_actor_addr) {
return; return;
...@@ -100,11 +106,11 @@ std::vector<group> local_actor::joined_groups() const { ...@@ -100,11 +106,11 @@ std::vector<group> local_actor::joined_groups() const {
} }
void local_actor::reply_message(message&& what) { void local_actor::reply_message(message&& what) {
auto& whom = m_current_node->sender; auto& whom = m_current_element->sender;
if (!whom) { if (!whom) {
return; return;
} }
auto& mid = m_current_node->mid; auto& mid = m_current_element->mid;
if (mid.valid() == false || mid.is_response()) { if (mid.valid() == false || mid.is_response()) {
send(actor_cast<channel>(whom), std::move(what)); send(actor_cast<channel>(whom), std::move(what));
} else if (!mid.is_answered()) { } else if (!mid.is_answered()) {
...@@ -118,24 +124,19 @@ void local_actor::forward_message(const actor& dest, message_priority prio) { ...@@ -118,24 +124,19 @@ void local_actor::forward_message(const actor& dest, message_priority prio) {
if (!dest) { if (!dest) {
return; return;
} }
auto mid = (prio == message_priority::high) auto mid = m_current_element->mid;
? m_current_node->mid.with_high_priority() m_current_element->mid = prio == message_priority::high
: m_current_node->mid.with_normal_priority(); ? mid.with_high_priority()
dest->enqueue(m_current_node->sender, mid, m_current_node->msg, host()); : mid.with_normal_priority();
// treat this message as asynchronous message from now on dest->enqueue(std::move(m_current_element), host());
m_current_node->mid = invalid_message_id;
} }
void local_actor::send_impl(message_priority prio, const channel& dest, void local_actor::send_impl(message_priority prio, abstract_channel* dest,
message&& what) { message&& what) {
if (!dest) { if (!dest) {
return; return;
} }
message_id mid; dest->enqueue(address(), message_id::make(prio), std::move(what), host());
if (prio == message_priority::high) {
mid = mid.with_high_priority();
}
dest->enqueue(address(), mid, std::move(what), host());
} }
void local_actor::send_exit(const actor_addr& whom, uint32_t reason) { void local_actor::send_exit(const actor_addr& whom, uint32_t reason) {
...@@ -153,15 +154,18 @@ void local_actor::delayed_send_impl(message_priority prio, const channel& dest, ...@@ -153,15 +154,18 @@ void local_actor::delayed_send_impl(message_priority prio, const channel& dest,
} }
response_promise local_actor::make_response_promise() { response_promise local_actor::make_response_promise() {
auto n = m_current_node; auto& ptr = m_current_element;
response_promise result{address(), n->sender, n->mid.response_id()}; if (!ptr) {
n->mid.mark_as_answered(); return response_promise{};
}
response_promise result{address(), ptr->sender, ptr->mid.response_id()};
ptr->mid.mark_as_answered();
return result; return result;
} }
void local_actor::cleanup(uint32_t reason) { void local_actor::cleanup(uint32_t reason) {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(reason));
super::cleanup(reason); abstract_actor::cleanup(reason);
// tell registry we're done // tell registry we're done
is_registered(false); is_registered(false);
} }
......
...@@ -17,76 +17,21 @@ ...@@ -17,76 +17,21 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_MIXIN_MEMORY_CACHED_HPP #include "caf/mailbox_based_actor.hpp"
#define CAF_MIXIN_MEMORY_CACHED_HPP
#include <utility>
#include <type_traits>
#include "caf/detail/memory.hpp"
namespace caf { namespace caf {
namespace mixin {
/**
* This mixin adds all member functions and member variables needed
* by the memory management subsystem.
*/
template <class Base, class Subtype>
class memory_cached : public Base {
friend class detail::memory;
template <class>
friend class detail::basic_memory_cache;
protected: mailbox_based_actor::~mailbox_based_actor() {
if (!m_mailbox.closed()) {
using combined_type = memory_cached; detail::sync_request_bouncer f{this->exit_reason()};
m_mailbox.close(f);
public:
static constexpr bool is_memory_cached_type = true;
void request_deletion() override {
auto mc = detail::memory::get_cache_map_entry(&typeid(*this));
if (!mc) {
auto om = outer_memory;
if (om) {
om->destroy();
om->deallocate();
} else
delete this;
} else
mc->release_instance(mc->downcast(this));
} }
}
template <class... Ts> void mailbox_based_actor::cleanup(uint32_t reason) {
memory_cached(Ts&&... args) detail::sync_request_bouncer f{reason};
: Base(std::forward<Ts>(args)...), outer_memory(nullptr) {} m_mailbox.close(f);
local_actor::cleanup(reason);
private: }
detail::instance_wrapper* outer_memory;
};
template <class T>
struct is_memory_cached {
template <class U, bool = U::is_memory_cached_type>
static std::true_type check(int);
template <class>
static std::false_type check(...);
public:
static constexpr bool value = decltype(check<T>(0))::value;
};
} // namespace mixin
} // namespace caf } // namespace caf
#endif // CAF_MIXIN_MEMORY_CACHED_HPP
...@@ -28,6 +28,15 @@ mailbox_element::mailbox_element() ...@@ -28,6 +28,15 @@ mailbox_element::mailbox_element()
// nop // nop
} }
mailbox_element::mailbox_element(actor_addr arg0, message_id arg1)
: next(nullptr),
prev(nullptr),
marked(false),
sender(std::move(arg0)),
mid(arg1) {
// nop
}
mailbox_element::mailbox_element(actor_addr arg0, message_id arg1, message arg2) mailbox_element::mailbox_element(actor_addr arg0, message_id arg1, message arg2)
: next(nullptr), : next(nullptr),
prev(nullptr), prev(nullptr),
...@@ -42,4 +51,24 @@ mailbox_element::~mailbox_element() { ...@@ -42,4 +51,24 @@ mailbox_element::~mailbox_element() {
// nop // nop
} }
mailbox_element_ptr mailbox_element::make(actor_addr sender, message_id id,
message msg) {
auto ptr = detail::memory::create<mailbox_element>(std::move(sender), id,
std::move(msg));
return mailbox_element_ptr{ptr};
}
/*
mailbox_element::joint::joint(ref_counted* v0, actor_addr&& v1, message_id v2)
: embedded<mailbox_element>(v0, std::move(v1), v2) {
// nop
}
void mailbox_element::joint::request_deletion() {
sender = invalid_actor_addr;
msg.reset();
m_storage->deref();
}
*/
} // namespace caf } // namespace caf
...@@ -88,10 +88,6 @@ void memory::add_cache_map_entry(const type_info* tinf, ...@@ -88,10 +88,6 @@ void memory::add_cache_map_entry(const type_info* tinf,
cache[tinf].reset(instance); cache[tinf].reset(instance);
} }
instance_wrapper::~instance_wrapper() {
// nop
}
} // namespace detail } // namespace detail
} // namespace caf } // namespace caf
......
...@@ -46,8 +46,12 @@ message& message::operator=(message&& other) { ...@@ -46,8 +46,12 @@ message& message::operator=(message&& other) {
return *this; return *this;
} }
void message::reset() { void message::reset(raw_ptr new_ptr) {
m_vals.reset(); m_vals.reset(new_ptr);
}
void message::swap(message& other) {
m_vals.swap(other.m_vals);
} }
void* message::mutable_at(size_t p) { void* message::mutable_at(size_t p) {
......
...@@ -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,7 @@ ...@@ -30,8 +30,7 @@
#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/not_prioritizing.hpp" #include "caf/detail/intrusive_partitioned_list.hpp"
#include "caf/policy/sequential_invoke.hpp"
#include "caf/io/fwd.hpp" #include "caf/io/fwd.hpp"
#include "caf/io/accept_handle.hpp" #include "caf/io/accept_handle.hpp"
...@@ -274,10 +273,18 @@ class broker : public extend<local_actor>:: ...@@ -274,10 +273,18 @@ class broker : public extend<local_actor>::
accept_handle add_tcp_doorman(network::native_socket fd); accept_handle add_tcp_doorman(network::native_socket fd);
policy::invoke_message_result invoke_message(mailbox_element_ptr& msg,
behavior& bhvr,
message_id mid);
void invoke_message(mailbox_element_ptr& msg);
void invoke_message(const actor_addr& sender, message_id mid, message& msg); void invoke_message(const actor_addr& sender, message_id mid, message& msg);
void enqueue(const actor_addr&, message_id, message, void enqueue(const actor_addr&, message_id,
execution_unit*) override; message, execution_unit*) override;
void enqueue(mailbox_element_ptr, execution_unit*) override;
/** /**
* Closes all connections and acceptors. * Closes all connections and acceptors.
...@@ -358,10 +365,10 @@ class broker : public extend<local_actor>:: ...@@ -358,10 +365,10 @@ 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;
std::deque<unique_mailbox_element_pointer> m_cache; detail::intrusive_partitioned_list<mailbox_element, detail::disposer> m_cache;
}; };
class broker::functor_based : public extend<broker>:: class broker::functor_based : public extend<broker>::
......
...@@ -31,8 +31,6 @@ ...@@ -31,8 +31,6 @@
#include "caf/exception.hpp" #include "caf/exception.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/mixin/memory_cached.hpp"
#include "caf/io/fwd.hpp" #include "caf/io/fwd.hpp"
#include "caf/io/accept_handle.hpp" #include "caf/io/accept_handle.hpp"
#include "caf/io/receive_policy.hpp" #include "caf/io/receive_policy.hpp"
......
...@@ -34,9 +34,9 @@ ...@@ -34,9 +34,9 @@
#include "caf/io/network/protocol.hpp" #include "caf/io/network/protocol.hpp"
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
#include "caf/mixin/memory_cached.hpp"
#include "caf/detail/memory.hpp" #include "caf/detail/memory.hpp"
#include "caf/detail/disposer.hpp"
#include "caf/detail/memory_cache_flag_type.hpp"
namespace boost { namespace boost {
namespace asio { namespace asio {
...@@ -116,7 +116,8 @@ class multiplexer { ...@@ -116,7 +116,8 @@ class multiplexer {
/** /**
* Simple wrapper for runnables * Simple wrapper for runnables
*/ */
struct runnable : extend<memory_managed>::with<mixin::memory_cached> { struct runnable : memory_managed {
static constexpr auto memory_cache_flag = detail::needs_embedding;
virtual void run() = 0; virtual void run() = 0;
virtual ~runnable(); virtual ~runnable();
}; };
......
...@@ -130,49 +130,47 @@ void broker::doorman::io_failure(network::operation op) { ...@@ -130,49 +130,47 @@ void broker::doorman::io_failure(network::operation op) {
class broker::continuation { class broker::continuation {
public: public:
continuation(broker_ptr ptr, actor_addr from, message_id mid, message&& msg) continuation(broker_ptr bptr, mailbox_element_ptr mptr)
: m_self(std::move(ptr)), : m_self(std::move(bptr)),
m_from(from), m_ptr(std::move(mptr)) {
m_mid(mid),
m_data(std::move(msg)) {
// nop // nop
} }
inline void operator()() { inline void operator()() {
CAF_PUSH_AID(m_self->id()); CAF_PUSH_AID(m_self->id());
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
m_self->invoke_message(m_from, m_mid, m_data); m_self->invoke_message(m_ptr);
} }
private: private:
broker_ptr m_self; broker_ptr m_self;
actor_addr m_from; mailbox_element_ptr m_ptr;
message_id m_mid;
message m_data;
}; };
void broker::invoke_message(const actor_addr& sender, message_id mid, policy::invoke_message_result broker::invoke_message(mailbox_element_ptr& msg,
message& msg) { behavior& bhvr,
message_id mid) {
return m_invoke_policy.invoke_message(this, msg, bhvr, mid);
}
void broker::invoke_message(mailbox_element_ptr& ptr) {
CAF_LOG_TRACE(CAF_TARG(msg, to_string)); CAF_LOG_TRACE(CAF_TARG(msg, to_string));
if (planned_exit_reason() != exit_reason::not_exited if (planned_exit_reason() != exit_reason::not_exited
|| bhvr_stack().empty()) { || bhvr_stack().empty()) {
CAF_LOG_DEBUG("actor already finished execution" CAF_LOG_DEBUG("actor already finished execution"
<< ", planned_exit_reason = " << planned_exit_reason() << ", planned_exit_reason = " << planned_exit_reason()
<< ", bhvr_stack().empty() = " << bhvr_stack().empty()); << ", bhvr_stack().empty() = " << bhvr_stack().empty());
if (mid.valid()) { if (ptr->mid.valid()) {
detail::sync_request_bouncer srb{exit_reason()}; detail::sync_request_bouncer srb{exit_reason()};
srb(sender, mid); srb(ptr->sender, ptr->mid);
} }
return; return;
} }
// prepare actor for invocation of message handler // prepare actor for invocation of message handler
m_dummy_node.sender = sender;
m_dummy_node.mid = mid;
std::swap(msg, m_dummy_node.msg);
try { try {
auto bhvr = bhvr_stack().back(); auto bhvr = bhvr_stack().back();
auto bid = bhvr_stack().back_id(); auto bid = bhvr_stack().back_id();
switch (m_invoke_policy.invoke_message(this, m_dummy_node, bhvr, bid)) { switch (invoke_message(ptr, bhvr, bid)) {
case policy::im_success: { case policy::im_success: {
CAF_LOG_DEBUG("handle_message returned hm_msg_handled"); CAF_LOG_DEBUG("handle_message returned hm_msg_handled");
while (!bhvr_stack().empty() while (!bhvr_stack().empty()
...@@ -187,9 +185,9 @@ void broker::invoke_message(const actor_addr& sender, message_id mid, ...@@ -187,9 +185,9 @@ void broker::invoke_message(const actor_addr& sender, message_id mid,
break; break;
case policy::im_skipped: { case policy::im_skipped: {
CAF_LOG_DEBUG("handle_message returned hm_skip_msg or hm_cache_msg"); CAF_LOG_DEBUG("handle_message returned hm_skip_msg or hm_cache_msg");
auto e = mailbox_element::create(sender, bid, if (ptr) {
std::move(m_dummy_node.msg)); m_cache.push_second_back(ptr.release());
m_cache.push_back(unique_mailbox_element_pointer{e}); }
break; break;
} }
} }
...@@ -205,9 +203,6 @@ void broker::invoke_message(const actor_addr& sender, message_id mid, ...@@ -205,9 +203,6 @@ void broker::invoke_message(const actor_addr& sender, message_id mid,
CAF_LOG_ERROR("broker killed due to an unknown exception"); CAF_LOG_ERROR("broker killed due to an unknown exception");
quit(exit_reason::unhandled_exception); quit(exit_reason::unhandled_exception);
} }
// restore dummy node
m_dummy_node.sender = invalid_actor_addr;
std::swap(m_dummy_node.msg, msg);
// cleanup if needed // cleanup if needed
if (planned_exit_reason() != exit_reason::not_exited) { if (planned_exit_reason() != exit_reason::not_exited) {
cleanup(planned_exit_reason()); cleanup(planned_exit_reason());
...@@ -222,27 +217,23 @@ void broker::invoke_message(const actor_addr& sender, message_id mid, ...@@ -222,27 +217,23 @@ void broker::invoke_message(const actor_addr& sender, message_id mid,
} }
} }
void broker::invoke_message(const actor_addr& v0, message_id v1, message& v2) {
auto ptr = mailbox_element::make(v0, v1, message{});
ptr->msg.swap(v2);
invoke_message(ptr);
if (ptr) {
ptr->msg.swap(v2);
}
}
bool broker::invoke_message_from_cache() { bool broker::invoke_message_from_cache() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
auto bhvr = bhvr_stack().back(); auto bhvr = bhvr_stack().back();
auto mid = bhvr_stack().back_id(); auto bid = bhvr_stack().back_id();
auto i = m_cache.begin(); auto i = m_cache.second_begin();
auto e = m_cache.end(); auto e = m_cache.second_end();
CAF_LOG_DEBUG(std::distance(i, e) << " elements in cache"); CAF_LOG_DEBUG(std::distance(i, e) << " elements in cache");
while (i != e) { return m_cache.invoke(this, i, e, bhvr, bid);
switch (m_invoke_policy.invoke_message(this, *(i->get()), bhvr, mid)) {
case policy::im_success:
m_cache.erase(i);
return true;
case policy::im_skipped:
++i;
break;
case policy::im_dropped:
i = m_cache.erase(i);
break;
}
}
return false;
} }
void broker::write(connection_handle hdl, size_t bs, const void* buf) { void broker::write(connection_handle hdl, size_t bs, const void* buf) {
...@@ -252,10 +243,13 @@ void broker::write(connection_handle hdl, size_t bs, const void* buf) { ...@@ -252,10 +243,13 @@ void broker::write(connection_handle hdl, size_t bs, const void* buf) {
out.insert(out.end(), first, last); out.insert(out.end(), first, last);
} }
void broker::enqueue(mailbox_element_ptr ptr, execution_unit*) {
parent().backend().post(continuation{this, std::move(ptr)});
}
void broker::enqueue(const actor_addr& sender, message_id mid, message msg, void broker::enqueue(const actor_addr& sender, message_id mid, message msg,
execution_unit*) { execution_unit* eu) {
parent().backend().post(continuation{this, sender, enqueue(mailbox_element::make(sender, mid, std::move(msg)), eu);
mid, std::move(msg)});
} }
broker::broker() : m_mm(*middleman::instance()) { broker::broker() : m_mm(*middleman::instance()) {
......
...@@ -27,9 +27,13 @@ ...@@ -27,9 +27,13 @@
using std::begin; using std::begin;
using std::end; using std::end;
using namespace caf;
namespace { namespace {
size_t s_iint_instances = 0; size_t s_iint_instances = 0;
}
} // namespace <anonymous>
struct iint { struct iint {
iint* next; iint* next;
...@@ -54,18 +58,46 @@ inline bool operator==(const iint& lhs, int rhs) { return lhs.value == rhs; } ...@@ -54,18 +58,46 @@ inline bool operator==(const iint& lhs, int rhs) { return lhs.value == rhs; }
inline bool operator==(int lhs, const iint& rhs) { return lhs == rhs.value; } inline bool operator==(int lhs, const iint& rhs) { return lhs == rhs.value; }
using iint_queue = caf::detail::single_reader_queue<iint>; using iint_queue = detail::single_reader_queue<iint>;
struct pseudo_actor { struct pseudo_actor {
using mailbox_type = caf::detail::single_reader_queue<iint>; using mailbox_type = detail::single_reader_queue<iint>;
using uptr = mailbox_type::unique_pointer;
mailbox_type mbox; mailbox_type mbox;
mailbox_type& mailbox() { mailbox_type& mailbox() {
return mbox; return mbox;
} }
policy::invoke_message_result invoke_message(uptr& ptr, int i) {
if (ptr->value == 1) {
ptr.reset();
return policy::im_dropped;
}
if (ptr->value == i) {
// call reset on some of our messages
if (ptr->is_high_priority()) {
ptr.reset();
}
return policy::im_success;
}
return policy::im_skipped;
}
template <class Policy>
policy::invoke_message_result invoke_message(uptr& ptr, Policy& policy, int i,
std::vector<int>& remaining) {
auto res = invoke_message(ptr, i);
if (res == im_success && !remaining.empty()) {
auto next = remaining.front();
remaining.erase(remaining.begin());
policy.invoke_from_cache(this, policy, next, remaining);
}
return res;
}
}; };
int main() { void test_prioritizing_dequeue() {
CAF_TEST(test_intrusive_containers);
pseudo_actor self; pseudo_actor self;
auto baseline = s_iint_instances; auto baseline = s_iint_instances;
CAF_CHECK_EQUAL(baseline, 3); // "begin", "separator", and "end" CAF_CHECK_EQUAL(baseline, 3); // "begin", "separator", and "end"
...@@ -74,7 +106,7 @@ int main() { ...@@ -74,7 +106,7 @@ int main() {
self.mbox.enqueue(new iint(3)); self.mbox.enqueue(new iint(3));
self.mbox.enqueue(new iint(4)); self.mbox.enqueue(new iint(4));
CAF_CHECK_EQUAL(baseline + 4, s_iint_instances); CAF_CHECK_EQUAL(baseline + 4, s_iint_instances);
caf::policy::prioritizing policy; policy::prioritizing policy;
// first "high priority message" // first "high priority message"
CAF_CHECK_EQUAL(self.mbox.count(), 4); CAF_CHECK_EQUAL(self.mbox.count(), 4);
CAF_CHECK_EQUAL(policy.has_next_message(&self), true); CAF_CHECK_EQUAL(policy.has_next_message(&self), true);
...@@ -99,5 +131,80 @@ int main() { ...@@ -99,5 +131,80 @@ int main() {
// back to baseline // back to baseline
CAF_CHECK_EQUAL(self.mbox.count(), 0); CAF_CHECK_EQUAL(self.mbox.count(), 0);
CAF_CHECK_EQUAL(baseline, s_iint_instances); CAF_CHECK_EQUAL(baseline, s_iint_instances);
}
template <class Policy>
void test_invoke_from_cache() {
pseudo_actor self;
auto baseline = s_iint_instances;
CAF_CHECK_EQUAL(baseline, 3); // "begin", "separator", and "end"
self.mbox.enqueue(new iint(1));
self.mbox.enqueue(new iint(2));
self.mbox.enqueue(new iint(3));
self.mbox.enqueue(new iint(4));
self.mbox.enqueue(new iint(5));
Policy policy;
CAF_CHECK_EQUAL(self.mbox.count(), 5);
// fill cache
auto ptr = policy.next_message(&self);
while (ptr) {
policy.push_to_cache(&self, std::move(ptr));
ptr = policy.next_message(&self);
}
CAF_CHECK_EQUAL(self.mbox.count(), 5);
// dequeue 3, 2, 4, 1, note: 1 is dropped on first dequeue
int expected;
//std::vector<int> expected{3, 2, 4, 5};
size_t remaining = 4;
//for (auto& i : expected) {
CAF_CHECK(policy.invoke_from_cache(&self, expected = 3));
CAF_CHECK_EQUAL(self.mbox.count(), --remaining);
CAF_CHECK(policy.invoke_from_cache(&self, expected = 2));
CAF_CHECK_EQUAL(self.mbox.count(), --remaining);
CAF_CHECK(policy.invoke_from_cache(&self, expected = 4));
CAF_CHECK_EQUAL(self.mbox.count(), --remaining);
CAF_CHECK(policy.invoke_from_cache(&self, expected = 5));
CAF_CHECK_EQUAL(self.mbox.count(), --remaining);
//}
}
template <class Policy>
void test_recursive_invoke_from_cache() {
pseudo_actor self;
auto baseline = s_iint_instances;
CAF_CHECK_EQUAL(baseline, 3); // "begin", "separator", and "end"
self.mbox.enqueue(new iint(1));
self.mbox.enqueue(new iint(2));
self.mbox.enqueue(new iint(3));
self.mbox.enqueue(new iint(4));
self.mbox.enqueue(new iint(5));
Policy policy;
CAF_CHECK_EQUAL(self.mbox.count(), 5);
// fill cache
auto ptr = policy.next_message(&self);
while (ptr) {
policy.push_to_cache(&self, std::move(ptr));
ptr = policy.next_message(&self);
}
CAF_CHECK_EQUAL(self.mbox.count(), 5);
// dequeue 3, 2, 4, 1, note: 1 is dropped on first dequeue
std::vector<int> remaining{2, 4, 5};
int first = 3;
policy.invoke_from_cache(&self, policy, first, remaining);
CAF_CHECK_EQUAL(self.mbox.count(), 0);
}
int main() {
CAF_TEST(test_intrusive_containers);
CAF_CHECKPOINT();
test_prioritizing_dequeue();
CAF_PRINT("test_invoke_from_cache<policy::prioritizing>");
test_invoke_from_cache<policy::prioritizing>();
CAF_PRINT("test_invoke_from_cache<policy::not_prioritizing>");
test_invoke_from_cache<policy::not_prioritizing>();
CAF_PRINT("test_recursive_invoke_from_cache<policy::prioritizing>");
test_recursive_invoke_from_cache<policy::prioritizing>();
CAF_PRINT("test_recursive_invoke_from_cache<policy::not_prioritizing>");
test_recursive_invoke_from_cache<policy::not_prioritizing>();
return CAF_TEST_RESULT(); return CAF_TEST_RESULT();
} }
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