Commit f6ca2450 authored by Dominik Charousset's avatar Dominik Charousset

Move cache to `single_reader_queue` / mailbox

By moving the cache to the mailbox itself, `mailbox.count()` properly counts
*all* messages and `mailbox.empty()` works as expected (as opposed to: is
pretty much useless in priority-aware actors).
parent 57ca0479
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_INTRUSIVE_LIST_HPP
#define CAF_DETAIL_INTRUSIVE_LIST_HPP
#include <memory>
#include <iterator>
namespace caf {
namespace detail {
template <class T, class Delete = std::default_delete<T>>
class intrusive_partitioned_list {
public:
using value_type = T;
using pointer = value_type*;
using deleter_type = Delete;
struct iterator : std::iterator<std::bidirectional_iterator_tag, value_type> {
pointer ptr;
iterator(pointer init = nullptr) : ptr(init) {
// nop
}
iterator(const iterator&) = default;
iterator& operator=(const iterator&) = default;
iterator& operator++() {
ptr = ptr->next;
return *this;
}
iterator operator++(int) {
iterator res = *this;
ptr = ptr->next;
return res;
}
iterator& operator--() {
ptr = ptr->prev;
return *this;
}
iterator operator--(int) {
iterator res = *this;
ptr = ptr->prev;
return res;
}
value_type& operator*() {
return *ptr;
}
pointer operator->() {
return ptr;
}
bool operator==(const iterator& other) const {
return ptr == other.ptr;
}
bool operator!=(const iterator& other) const {
return ptr != other.ptr;
}
};
intrusive_partitioned_list() {
m_head.next = &m_separator;
m_separator.prev = &m_head;
m_separator.next = &m_tail;
m_tail.prev = &m_separator;
}
iterator first_begin() {
return m_head.next;
}
iterator first_end() {
return &m_separator;
}
iterator second_begin() {
return m_separator.next;
}
iterator second_end() {
return &m_tail;
}
iterator insert(iterator next, pointer val) {
auto prev = next->prev;
val->prev = prev;
val->next = next.ptr;
prev->next = val;
next->prev = val;
return val;
}
bool first_empty() {
return first_begin() == first_end();
}
bool second_empty() {
return second_begin() == second_end();
}
bool empty() {
return first_empty() && second_empty();
}
pointer take(iterator pos) {
auto res = pos.ptr;
auto next = res->next;
auto prev = res->prev;
prev->next = next;
next->prev = prev;
return res;
}
iterator erase(iterator pos) {
auto next = pos->next;
m_delete(take(pos));
return next;
}
pointer take_first_front() {
return take(first_begin());
}
pointer take_second_front() {
return take(second_begin());
}
value_type& first_front() {
return *first_begin();
}
value_type& second_front() {
return *second_begin();
}
void pop_first_front() {
erase(first_begin());
}
void pop_second_front() {
erase(second_begin());
}
void push_first_back(pointer val) {
insert(first_end(), val);
}
void push_second_back(pointer val) {
insert(second_end(), val);
}
size_t count(iterator first, iterator last, size_t max_count) {
size_t result = 0;
while (first != last && result < max_count) {
++first;
++result;
}
return result;
}
size_t count(size_t max_count = std::numeric_limits<size_t>::max()) {
auto r1 = count(first_begin(), first_end(), max_count);
return r1 + count(second_begin(), second_end(), max_count - r1);
}
private:
value_type m_head;
value_type m_separator;
value_type m_tail;
deleter_type m_delete;
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_INTRUSIVE_LIST_HPP
......@@ -28,6 +28,7 @@
#include "caf/detail/logging.hpp"
#include "caf/policy/invoke_policy.hpp"
#include "caf/policy/scheduling_policy.hpp"
namespace caf {
......@@ -39,9 +40,9 @@ namespace detail {
template <class Base, class Derived, class Policies>
class proper_actor_base : public Policies::resume_policy::template
mixin<Base, Derived> {
public:
using super = typename Policies::resume_policy::template mixin<Base, Derived>;
public:
template <class... Ts>
proper_actor_base(Ts&&... args) : super(std::forward<Ts>(args)...) {
CAF_REQUIRE(this->is_registered() == false);
......@@ -51,20 +52,12 @@ class proper_actor_base : public Policies::resume_policy::template
using timeout_type = typename Policies::scheduling_policy::timeout_type;
void enqueue(const actor_addr& sender,
message_id mid,
message msg,
execution_unit* eu) override {
CAF_PUSH_AID(dptr()->id());
/*
CAF_LOG_DEBUG(CAF_TARG(sender, to_string)
<< ", " << CAF_MARG(mid, integer_value) << ", "
<< CAF_TARG(msg, to_string));
*/
void enqueue(const actor_addr& sender, message_id mid,
message msg, execution_unit* eu) override {
scheduling_policy().enqueue(dptr(), sender, mid, msg, eu);
}
inline void launch(bool hide, bool lazy, execution_unit* eu) {
void launch(bool hide, bool lazy, execution_unit* eu) {
CAF_LOG_TRACE("");
this->is_registered(!hide);
this->scheduling_policy().launch(this, eu, lazy);
......@@ -87,43 +80,16 @@ class proper_actor_base : public Policies::resume_policy::template
// member functions from priority policy
inline unique_mailbox_element_pointer next_message() {
unique_mailbox_element_pointer next_message() {
return priority_policy().next_message(dptr());
}
inline bool has_next_message() {
bool has_next_message() {
return priority_policy().has_next_message(dptr());
}
inline void push_to_cache(unique_mailbox_element_pointer ptr) {
priority_policy().push_to_cache(std::move(ptr));
}
using cache_iterator = typename Policies::priority_policy::cache_iterator;
inline bool cache_empty() {
return priority_policy().cache_empty();
}
inline cache_iterator cache_begin() {
return priority_policy().cache_begin();
}
inline cache_iterator cache_end() {
return priority_policy().cache_end();
}
inline unique_mailbox_element_pointer cache_take_first() {
return priority_policy().cache_take_first();
}
template <class Iterator>
inline void cache_prepend(Iterator first, Iterator last) {
priority_policy().cache_prepend(first, last);
}
inline void cache_erase(cache_iterator iter) {
priority_policy().cache_erase(iter);
void push_to_cache(unique_mailbox_element_pointer ptr) {
priority_policy().push_to_cache(dptr(), std::move(ptr));
}
// member functions from resume policy
......@@ -133,31 +99,30 @@ class proper_actor_base : public Policies::resume_policy::template
// member functions from invoke policy
template <class PartialFunctionOrBehavior>
inline bool invoke_message(unique_mailbox_element_pointer& ptr,
policy::invoke_message_result invoke_message(mailbox_element& me,
PartialFunctionOrBehavior& fun,
message_id awaited_response) {
return invoke_policy().invoke_message(dptr(), ptr, fun,
awaited_response);
return invoke_policy().invoke_message(dptr(), me, fun, awaited_response);
}
protected:
inline typename Policies::scheduling_policy& scheduling_policy() {
typename Policies::scheduling_policy& scheduling_policy() {
return m_policies.get_scheduling_policy();
}
inline typename Policies::priority_policy& priority_policy() {
typename Policies::priority_policy& priority_policy() {
return m_policies.get_priority_policy();
}
inline typename Policies::resume_policy& resume_policy() {
typename Policies::resume_policy& resume_policy() {
return m_policies.get_resume_policy();
}
inline typename Policies::invoke_policy& invoke_policy() {
typename Policies::invoke_policy& invoke_policy() {
return m_policies.get_invoke_policy();
}
inline Derived* dptr() {
Derived* dptr() {
return static_cast<Derived*>(this);
}
......@@ -172,12 +137,12 @@ template <class Base, class Policies,
class proper_actor
: public proper_actor_base<Base, proper_actor<Base, Policies, false>,
Policies> {
using super = proper_actor_base<Base, proper_actor, Policies>;
public:
static_assert(std::is_base_of<local_actor, Base>::value,
"Base is not derived from local_actor");
using super = proper_actor_base<Base, proper_actor, Policies>;
template <class... Ts>
proper_actor(Ts&&... args) : super(std::forward<Ts>(args)...) {
// nop
......@@ -185,30 +150,16 @@ class proper_actor
// required by event_based_resume::mixin::resume
bool invoke_message(unique_mailbox_element_pointer& ptr) {
policy::invoke_message_result invoke_message(mailbox_element& me) {
CAF_LOG_TRACE("");
auto bhvr = this->bhvr_stack().back();
auto mid = this->bhvr_stack().back_id();
return this->invoke_policy().invoke_message(this, ptr, bhvr, mid);
return this->invoke_policy().invoke_message(this, me, bhvr, mid);
}
bool invoke_message_from_cache() {
CAF_LOG_TRACE("");
auto bhvr = this->bhvr_stack().back();
auto mid = this->bhvr_stack().back_id();
auto e = this->cache_end();
CAF_LOG_DEBUG(std::distance(this->cache_begin(), e)
<< " elements in cache");
for (auto i = this->cache_begin(); i != e; ++i) {
auto im = this->invoke_policy().invoke_message(this, *i, bhvr, mid);
if (im || !*i) {
this->cache_erase(i);
if (im) return true;
// start anew, because we have invalidated our iterators now
return invoke_message_from_cache();
}
}
return false;
return this->priority_policy().invoke_from_cache(this);
}
};
......@@ -226,11 +177,11 @@ class proper_actor<Base, Policies, true>
// 'import' optional blocking member functions from policies
inline void await_data() {
void await_data() {
this->scheduling_policy().await_data(this);
}
inline void await_ready() {
void await_ready() {
this->resume_policy().await_ready(this);
}
......@@ -238,22 +189,8 @@ class proper_actor<Base, Policies, true>
void dequeue_response(behavior& bhvr, message_id mid) override {
// try to dequeue from cache first
if (!this->cache_empty()) {
std::vector<unique_mailbox_element_pointer> tmp_vec;
auto restore_cache = [&] {
if (!tmp_vec.empty()) {
this->cache_prepend(tmp_vec.begin(), tmp_vec.end());
}
};
while (!this->cache_empty()) {
auto tmp = this->cache_take_first();
if (this->invoke_message(tmp, bhvr, mid)) {
restore_cache();
if (this->priority_policy().invoke_from_cache(this, bhvr, mid)) {
return;
} else
tmp_vec.push_back(std::move(tmp));
}
restore_cache();
}
bool timeout_valid = false;
uint32_t timeout_id;
......@@ -275,15 +212,17 @@ class proper_actor<Base, Policies, true>
});
// read incoming messages
for (;;) {
auto msg = this->next_message();
if (!msg) {
this->await_ready();
} else {
if (this->invoke_message(msg, bhvr, mid)) {
// we're done
auto msg = this->next_message();
switch (this->invoke_message(*msg, bhvr, mid)) {
case policy::im_success:
return;
}
if (msg) this->push_to_cache(std::move(msg));
case policy::im_skipped:
this->push_to_cache(std::move(msg));
break;
default:
// delete msg
break;
}
}
}
......@@ -307,7 +246,7 @@ class proper_actor<Base, Policies, true>
return tid;
}
inline void handle_timeout(behavior& bhvr, uint32_t timeout_id) {
void handle_timeout(behavior& bhvr, uint32_t timeout_id) {
auto e = m_pending_timeouts.end();
auto i = std::find(m_pending_timeouts.begin(), e, timeout_id);
CAF_LOG_WARNING_IF(i == e, "ignored unexpected timeout");
......@@ -318,24 +257,24 @@ class proper_actor<Base, Policies, true>
}
// required by nestable invoke policy
inline void pop_timeout() {
void pop_timeout() {
m_pending_timeouts.pop_back();
}
// required by nestable invoke policy;
// adds a dummy timeout to the pending timeouts to prevent
// nestable invokes to trigger an inactive timeout
inline void push_timeout() {
void push_timeout() {
m_pending_timeouts.push_back(++m_next_timeout_id);
}
inline bool waits_for_timeout(uint32_t timeout_id) const {
bool waits_for_timeout(uint32_t timeout_id) const {
auto e = m_pending_timeouts.end();
auto i = std::find(m_pending_timeouts.begin(), e, timeout_id);
return i != e;
}
inline bool is_active_timeout(uint32_t tid) const {
bool is_active_timeout(uint32_t tid) const {
return !m_pending_timeouts.empty() && m_pending_timeouts.back() == tid;
}
......
......@@ -21,6 +21,7 @@
#define CAF_SINGLE_READER_QUEUE_HPP
#include <list>
#include <deque>
#include <mutex>
#include <atomic>
#include <memory>
......@@ -29,6 +30,8 @@
#include "caf/config.hpp"
#include "caf/detail/intrusive_partitioned_list.hpp"
namespace caf {
namespace detail {
......@@ -63,6 +66,9 @@ class single_reader_queue {
public:
using value_type = T;
using pointer = value_type*;
using deleter_type = Delete;
using unique_pointer = std::unique_ptr<value_type, deleter_type>;
using cache_type = intrusive_partitioned_list<value_type, deleter_type>;
/**
* Tries to dequeue a new element from the mailbox.
......@@ -115,7 +121,7 @@ class single_reader_queue {
*/
bool empty() {
CAF_REQUIRE(!closed());
return m_head == nullptr && is_dummy(m_stack.load());
return m_cache.empty() && m_head == nullptr && is_dummy(m_stack.load());
}
/**
......@@ -194,7 +200,10 @@ class single_reader_queue {
}
size_t count(size_t max_count = std::numeric_limits<size_t>::max()) {
size_t res = 0;
size_t res = m_cache.count(max_count);
if (res >= max_count) {
return res;
}
fetch_new_data();
auto ptr = m_head;
while (ptr && res < max_count) {
......@@ -204,6 +213,15 @@ class single_reader_queue {
return res;
}
// note: the cache is intended to be used by the owner, the queue itself
// never accesses the cache other than for counting;
// the first partition of the cache is meant to be used to store and
// sort messages that were not processed yet, while the second
// partition is meant to store skipped messages
cache_type& cache() {
return m_cache;
}
/**************************************************************************
* support for synchronized access *
**************************************************************************/
......@@ -260,7 +278,8 @@ class single_reader_queue {
// accessed only by the owner
pointer m_head;
Delete m_delete;
deleter_type m_delete;
intrusive_partitioned_list<value_type, deleter_type> m_cache;
// atomically sets m_stack back and enqueues all elements to the cache
bool fetch_new_data(pointer end_ptr) {
......
......@@ -37,18 +37,17 @@ class local_actor;
class mailbox_element : public extend<memory_managed>::
with<mixin::memory_cached> {
friend class local_actor;
friend class detail::memory;
public:
mailbox_element* next; // intrusive next pointer
mailbox_element* prev; // intrusive previous pointer
bool marked; // denotes if this node is currently processed
actor_addr sender;
message_id mid;
message msg; // 'content field'
mailbox_element();
mailbox_element(actor_addr sender, message_id id, message data);
~mailbox_element();
mailbox_element(mailbox_element&&) = delete;
......@@ -62,12 +61,9 @@ class mailbox_element : public extend<memory_managed>::
std::forward<T>(data));
}
private:
mailbox_element() = default;
mailbox_element(actor_addr sender, message_id id, message data);
inline bool is_high_priority() const {
return mid.is_high_priority();
}
};
using unique_mailbox_element_pointer =
......
......@@ -30,6 +30,7 @@
#include "caf/extend.hpp"
#include "caf/behavior.hpp"
#include "caf/policy/invoke_policy.hpp"
#include "caf/policy/resume_policy.hpp"
#include "caf/detail/logging.hpp"
......@@ -124,7 +125,8 @@ class event_based_resume {
for (size_t i = 0; i < max_throughput; ++i) {
auto ptr = d->next_message();
if (ptr) {
if (d->invoke_message(ptr)) {
switch (d->invoke_message(*ptr)) {
case policy::im_success:
d->bhvr_stack().cleanup();
++handled_msgs;
if (actor_done()) {
......@@ -140,12 +142,13 @@ class event_based_resume {
return resume_result::done;
}
}
}
// add ptr to cache if invoke_message
// did not reset it (i.e. skipped, but not dropped)
if (ptr) {
CAF_LOG_DEBUG("add message to cache");
break;
case policy::im_skipped:
d->push_to_cache(std::move(ptr));
break;
case policy::im_dropped:
// destroy msg
break;
}
} else {
CAF_LOG_DEBUG("no more element in mailbox; going to block");
......
......@@ -49,11 +49,10 @@ enum receive_policy_flag {
rp_sequential
};
enum handle_message_result {
hm_skip_msg,
hm_drop_msg,
hm_cache_msg,
hm_msg_handled
enum invoke_message_result {
im_success,
im_skipped,
im_dropped
};
template <receive_policy_flag X>
......@@ -79,29 +78,89 @@ class invoke_policy {
sync_response // a synchronous response
};
/**
* @note `node_ptr`.release() is called whenever a message was
* handled or dropped.
*/
// the workflow of invoke_message (im) is as follows:
// - should_skip? if yes: return im_skipped
// - msg is ordinary message? if yes:
// - begin(...) -> prepares a self for message handling
// - self could process message?
// - yes: cleanup()
// - no: revert(...) -> set self back to state it had before begin()
template <class Actor, class Fun>
bool invoke_message(Actor* self, unique_mailbox_element_pointer& node_ptr,
invoke_message_result invoke_message(Actor* self, mailbox_element& node,
Fun& fun, message_id awaited_response) {
if (!node_ptr) {
return false;
}
CAF_LOG_TRACE("");
switch (handle_message(self, node_ptr.get(), fun, awaited_response)) {
case hm_msg_handled:
node_ptr.reset();
return true;
case hm_cache_msg:
case hm_skip_msg:
// do not reset ptr (not handled => cache or skip)
return false;
default: // drop message
node_ptr.reset();
return false;
bool handle_sync_failure_on_mismatch = true;
if (dptr()->hm_should_skip(node)) {
return im_skipped;
}
switch (this->filter_msg(self, node)) {
case msg_type::normal_exit:
CAF_LOG_DEBUG("dropped normal exit signal");
return im_dropped;
case msg_type::expired_sync_response:
CAF_LOG_DEBUG("dropped expired sync response");
return im_dropped;
case msg_type::expired_timeout:
CAF_LOG_DEBUG("dropped expired timeout message");
return im_dropped;
case msg_type::inactive_timeout:
CAF_LOG_DEBUG("skipped inactive timeout message");
return im_skipped;
case msg_type::non_normal_exit:
CAF_LOG_DEBUG("handled non-normal exit signal");
// this message was handled
// by calling self->quit(...)
return im_success;
case msg_type::timeout: {
CAF_LOG_DEBUG("handle timeout message");
auto& tm = node.msg.get_as<timeout_msg>(0);
self->handle_timeout(fun, tm.timeout_id);
if (awaited_response.valid()) {
self->mark_arrived(awaited_response);
self->remove_handler(awaited_response);
}
return im_success;
}
case msg_type::timeout_response:
handle_sync_failure_on_mismatch = false;
CAF_ANNOTATE_FALLTHROUGH;
case msg_type::sync_response:
CAF_LOG_DEBUG("handle as synchronous response: "
<< CAF_TARG(node.msg, to_string) << ", "
<< CAF_MARG(node.mid, integer_value) << ", "
<< CAF_MARG(awaited_response, integer_value));
if (awaited_response.valid() && node.mid == awaited_response) {
auto previous_node = dptr()->hm_begin(self, node);
auto res = invoke_fun(self, node.msg, node.mid, fun);
if (!res && handle_sync_failure_on_mismatch) {
CAF_LOG_WARNING("sync failure occured in actor "
<< "with ID " << self->id());
self->handle_sync_failure();
}
self->mark_arrived(awaited_response);
self->remove_handler(awaited_response);
dptr()->hm_cleanup(self, previous_node);
return im_success;
}
return im_skipped;
case msg_type::ordinary:
if (!awaited_response.valid()) {
auto previous_node = dptr()->hm_begin(self, node);
auto res = invoke_fun(self, node.msg, node.mid, fun);
if (res) {
dptr()->hm_cleanup(self, previous_node);
return im_success;
}
// no match (restore self members)
dptr()->hm_revert(self, previous_node);
}
CAF_LOG_DEBUG_IF(awaited_response.valid(),
"ignored message; await response: "
<< awaited_response.integer_value());
return im_skipped;
}
// should be unreachable
CAF_CRITICAL("invalid message type");
}
using nestable = typename rp_flag<rp_nestable>::type;
......@@ -185,91 +244,6 @@ class invoke_policy {
return res;
}
// the workflow of handle_message (hm) is as follows:
// - should_skip? if yes: return hm_skip_msg
// - msg is ordinary message? if yes:
// - begin(...) -> prepares a self for message handling
// - self could process message?
// - yes: cleanup()
// - no: revert(...) -> set self back to state it had before begin()
template <class Actor, class Fun>
handle_message_result handle_message(Actor* self, mailbox_element* node,
Fun& fun, message_id awaited_response) {
bool handle_sync_failure_on_mismatch = true;
if (dptr()->hm_should_skip(node)) {
return hm_skip_msg;
}
switch (this->filter_msg(self, node)) {
case msg_type::normal_exit:
CAF_LOG_DEBUG("dropped normal exit signal");
return hm_drop_msg;
case msg_type::expired_sync_response:
CAF_LOG_DEBUG("dropped expired sync response");
return hm_drop_msg;
case msg_type::expired_timeout:
CAF_LOG_DEBUG("dropped expired timeout message");
return hm_drop_msg;
case msg_type::inactive_timeout:
CAF_LOG_DEBUG("skipped inactive timeout message");
return hm_skip_msg;
case msg_type::non_normal_exit:
CAF_LOG_DEBUG("handled non-normal exit signal");
// this message was handled
// by calling self->quit(...)
return hm_msg_handled;
case msg_type::timeout: {
CAF_LOG_DEBUG("handle timeout message");
auto& tm = node->msg.get_as<timeout_msg>(0);
self->handle_timeout(fun, tm.timeout_id);
if (awaited_response.valid()) {
self->mark_arrived(awaited_response);
self->remove_handler(awaited_response);
}
return hm_msg_handled;
}
case msg_type::timeout_response:
handle_sync_failure_on_mismatch = false;
CAF_ANNOTATE_FALLTHROUGH;
case msg_type::sync_response:
CAF_LOG_DEBUG("handle as synchronous response: "
<< CAF_TARG(node->msg, to_string) << ", "
<< CAF_MARG(node->mid, integer_value) << ", "
<< CAF_MARG(awaited_response, integer_value));
if (awaited_response.valid() && node->mid == awaited_response) {
auto previous_node = dptr()->hm_begin(self, node);
auto res = invoke_fun(self, node->msg, node->mid, fun);
if (!res && handle_sync_failure_on_mismatch) {
CAF_LOG_WARNING("sync failure occured in actor "
<< "with ID " << self->id());
self->handle_sync_failure();
}
self->mark_arrived(awaited_response);
self->remove_handler(awaited_response);
dptr()->hm_cleanup(self, previous_node);
return hm_msg_handled;
}
return hm_cache_msg;
case msg_type::ordinary:
if (!awaited_response.valid()) {
auto previous_node = dptr()->hm_begin(self, node);
auto res = invoke_fun(self, node->msg, node->mid, fun);
if (res) {
dptr()->hm_cleanup(self, previous_node);
return hm_msg_handled;
}
// no match (restore self members)
dptr()->hm_revert(self, previous_node);
}
CAF_LOG_DEBUG_IF(awaited_response.valid(),
"ignored message; await response: "
<< awaited_response.integer_value());
return hm_cache_msg;
}
// should be unreachable
CAF_CRITICAL("invalid message type");
}
protected:
Derived* dptr() {
return static_cast<Derived*>(this);
......@@ -285,9 +259,9 @@ class invoke_policy {
// - expired synchronous response messages
template <class Actor>
msg_type filter_msg(Actor* self, mailbox_element* node) {
const message& msg = node->msg;
auto mid = node->mid;
msg_type filter_msg(Actor* self, mailbox_element& node) {
const message& msg = node.msg;
auto mid = node.mid;
if (msg.size() == 1) {
if (msg.match_element<exit_msg>(0)) {
auto& em = msg.get_as<exit_msg>(0);
......
......@@ -17,8 +17,8 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_THREADED_HPP
#define CAF_THREADED_HPP
#ifndef CAF_POLICY_NESTABLE_INVOKE_HPP
#define CAF_POLICY_NESTABLE_INVOKE_HPP
#include <mutex>
#include <chrono>
......@@ -37,28 +37,28 @@ namespace policy {
class nestable_invoke : public invoke_policy<nestable_invoke> {
public:
inline bool hm_should_skip(mailbox_element* node) {
return node->marked;
inline bool hm_should_skip(mailbox_element& node) {
return node.marked;
}
template <class Actor>
inline mailbox_element* hm_begin(Actor* self, mailbox_element* node) {
mailbox_element* hm_begin(Actor* self, mailbox_element& node) {
auto previous = self->current_node();
self->current_node(node);
self->current_node(&node);
self->push_timeout();
node->marked = true;
node.marked = true;
return previous;
}
template <class Actor>
inline void hm_cleanup(Actor* self, mailbox_element* previous) {
void hm_cleanup(Actor* self, mailbox_element* previous) {
self->current_node()->marked = false;
self->current_node(previous);
self->pop_timeout();
}
template <class Actor>
inline void hm_revert(Actor* self, mailbox_element* previous) {
void hm_revert(Actor* self, mailbox_element* previous) {
// same operation for blocking, i.e., nestable, invoke
hm_cleanup(self, previous);
}
......@@ -67,4 +67,4 @@ class nestable_invoke : public invoke_policy<nestable_invoke> {
} // namespace policy
} // namespace caf
#endif // CAF_THREADED_HPP
#endif // CAF_POLICY_NESTABLE_INVOKE_HPP
......@@ -17,78 +17,61 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef NOT_PRIORITIZING_HPP
#define NOT_PRIORITIZING_HPP
#ifndef CAF_POLICY_NOT_PRIORITIZING_HPP
#define CAF_POLICY_NOT_PRIORITIZING_HPP
#include <deque>
#include <iterator>
#include "caf/mailbox_element.hpp"
#include "caf/policy/invoke_policy.hpp"
#include "caf/policy/priority_policy.hpp"
namespace caf {
namespace policy {
class not_prioritizing {
public:
using cache_type = std::deque<unique_mailbox_element_pointer>;
using cache_iterator = cache_type::iterator;
template <class Actor>
unique_mailbox_element_pointer next_message(Actor* self) {
return unique_mailbox_element_pointer{self->mailbox().try_pop()};
}
template <class Actor>
inline bool has_next_message(Actor* self) {
bool has_next_message(Actor* self) {
return self->mailbox().can_fetch_more();
}
inline void push_to_cache(unique_mailbox_element_pointer ptr) {
m_cache.push_back(std::move(ptr));
}
inline cache_iterator cache_begin() {
return m_cache.begin();
}
inline cache_iterator cache_end() {
return m_cache.end();
}
inline void cache_erase(cache_iterator iter) {
m_cache.erase(iter);
template <class Actor>
void push_to_cache(Actor* self, unique_mailbox_element_pointer ptr) {
self->mailbox().cache().push_second_back(ptr.release());
}
inline bool cache_empty() const {
return m_cache.empty();
template <class Actor, class... Ts>
bool invoke_from_cache(Actor* self, Ts&... args) {
auto& cache = self->mailbox().cache();
auto i = cache.second_begin();
auto e = cache.second_end();
CAF_LOG_DEBUG(std::distance(i, e) << " elements in cache");
while (i != e) {
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;
}
inline unique_mailbox_element_pointer cache_take_first() {
auto tmp = std::move(m_cache.front());
m_cache.erase(m_cache.begin());
return std::move(tmp);
}
template <class Iterator>
inline void cache_prepend(Iterator first, Iterator last) {
auto mi = std::make_move_iterator(first);
auto me = std::make_move_iterator(last);
m_cache.insert(m_cache.begin(), mi, me);
return false;
}
private:
cache_type m_cache;
};
} // namespace policy
} // namespace caf
#endif // NOT_PRIORITIZING_HPP
#endif // CAF_POLICY_NOT_PRIORITIZING_HPP
......@@ -24,102 +24,78 @@
#include "caf/mailbox_element.hpp"
#include "caf/message_priority.hpp"
#include "caf/policy/not_prioritizing.hpp"
#include "caf/detail/single_reader_queue.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
namespace caf {
namespace policy {
// this policy partitions the mailbox into four segments:
// <-------- !was_skipped --------> | <-------- was_skipped -------->
// <-- high prio --><-- low prio -->|<-- high prio --><-- low prio -->
class prioritizing {
public:
using cache_type = std::list<unique_mailbox_element_pointer>;
using cache_iterator = cache_type::iterator;
using pointer = mailbox_element*;
using mailbox_type = detail::single_reader_queue<mailbox_element,
detail::disposer>;
using cache_type = mailbox_type::cache_type;
using iterator = cache_type::iterator;
template <class Actor>
unique_mailbox_element_pointer next_message(Actor* self) {
if (!m_high.empty()) return take_first(m_high);
// read whole mailbox
unique_mailbox_element_pointer tmp{self->mailbox().try_pop()};
typename Actor::mailbox_type::unique_pointer next_message(Actor* self) {
auto& cache = self->mailbox().cache();
auto i = cache.first_begin();
auto e = cache.first_end();
if (i == e || !i->is_high_priority()) {
// insert points for high priority
auto hp_pos = i;
// read whole mailbox at once
auto tmp = self->mailbox().try_pop();
while (tmp) {
if (tmp->mid.is_high_priority())
m_high.push_back(std::move(tmp));
else
m_low.push_back(std::move(tmp));
tmp.reset(self->mailbox().try_pop());
cache.insert(tmp->is_high_priority() ? hp_pos : e, tmp);
// adjust high priority insert point on first low prio element insert
if (hp_pos == e && !tmp->is_high_priority()) {
--hp_pos;
}
if (!m_high.empty()) return take_first(m_high);
if (!m_low.empty()) return take_first(m_low);
return unique_mailbox_element_pointer{};
tmp = self->mailbox().try_pop();
}
template <class Actor>
inline bool has_next_message(Actor* self) {
return !m_high.empty() || !m_low.empty() ||
self->mailbox().can_fetch_more();
}
inline void push_to_cache(unique_mailbox_element_pointer ptr) {
if (ptr->mid.is_high_priority()) {
// insert before first element with low priority
m_cache.insert(cache_low_begin(), std::move(ptr));
} else
m_cache.push_back(std::move(ptr));
typename Actor::mailbox_type::unique_pointer result;
if (!cache.first_empty()) {
result.reset(cache.take_first_front());
}
inline cache_iterator cache_begin() { return m_cache.begin(); }
inline cache_iterator cache_end() { return m_cache.begin(); }
inline void cache_erase(cache_iterator iter) { m_cache.erase(iter); }
inline bool cache_empty() const { return m_cache.empty(); }
inline unique_mailbox_element_pointer cache_take_first() {
return take_first(m_cache);
return result;
}
template <class Iterator>
inline void cache_prepend(Iterator first, Iterator last) {
using std::make_move_iterator;
// split input range between high and low priority messages
cache_type high;
cache_type low;
for (; first != last; ++first) {
if ((*first)->mid.is_high_priority())
high.push_back(std::move(*first));
else
low.push_back(std::move(*first));
}
// prepend high priority messages
m_cache.insert(m_cache.begin(), make_move_iterator(high.begin()),
make_move_iterator(high.end()));
// insert low priority messages after high priority messages
m_cache.insert(cache_low_begin(), make_move_iterator(low.begin()),
make_move_iterator(low.end()));
template <class Actor>
bool has_next_message(Actor* self) {
auto& mbox = self->mailbox();
auto& cache = mbox.cache();
return !cache.first_empty() || mbox.can_fetch_more();
}
private:
cache_iterator cache_low_begin() {
// insert before first element with low priority
return std::find_if(m_cache.begin(), m_cache.end(),
[](const unique_mailbox_element_pointer& e) {
return !e->mid.is_high_priority();
});
template <class Actor>
void push_to_cache(Actor* self, unique_mailbox_element_pointer ptr) {
auto high_prio = [](const mailbox_element& val) {
return val.is_high_priority();
};
auto& cache = self->mailbox().cache();
auto e = cache.second_end();
auto i = ptr->is_high_priority()
? std::partition_point(cache.second_begin(), e, high_prio)
: e;
cache.insert(i, ptr.release());
}
inline unique_mailbox_element_pointer take_first(cache_type& from) {
auto tmp = std::move(from.front());
from.erase(from.begin());
return std::move(tmp);
template <class Actor, class... Ts>
bool invoke_from_cache(Actor* self, Ts&... args) {
// same as in not prioritzing policy
not_prioritizing np;
return np.invoke_from_cache(self, args...);
}
cache_type m_cache;
cache_type m_high;
cache_type m_low;
};
} // namespace policy
......
......@@ -20,25 +20,19 @@
#ifndef CAF_PRIORITY_POLICY_HPP
#define CAF_PRIORITY_POLICY_HPP
namespace caf {
class mailbox_element;
} // namespace caf
#include "caf/mailbox_element.hpp"
namespace caf {
namespace policy {
/**
* The priority_policy *concept* class. Please note that this
* class is **not** implemented. It merely explains all
* required member functions.
* The priority_policy *concept* class. Please note that this class is **not**
* implemented. It merely explains all required member functions.
*/
class priority_policy {
public:
/**
* 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>
unique_mailbox_element_pointer next_message(Actor* self);
......@@ -49,18 +43,18 @@ class priority_policy {
template <class Actor>
bool has_next_message(Actor* self);
void push_to_cache(unique_mailbox_element_pointer ptr);
using cache_type = std::vector<unique_mailbox_element_pointer>;
using cache_iterator = cache_type::iterator;
cache_iterator cache_begin();
cache_iterator cache_end();
void cache_erase(cache_iterator iter);
/**
* Stores the message in a cache for later retrieval.
*/
template <class Actor>
void push_to_cache(Actor* self, unique_mailbox_element_pointer ptr);
/**
* Removes the first element from the cache matching predicate `p`.
* @returns `true` if an element was removed, `false` otherwise.
*/
template <class Actor, class Predicate>
bool invoke_from_cache(Actor* self, Predicate p);
};
} // namespace policy
......
......@@ -34,13 +34,13 @@ namespace policy {
*/
class sequential_invoke : public invoke_policy<sequential_invoke> {
public:
inline bool hm_should_skip(mailbox_element*) {
inline bool hm_should_skip(mailbox_element&) {
return false;
}
template <class Actor>
mailbox_element* hm_begin(Actor* self, mailbox_element* node) {
self->current_node(node);
mailbox_element* hm_begin(Actor* self, mailbox_element& node) {
self->current_node(&node);
return nullptr;
}
......
......@@ -53,7 +53,11 @@ constexpr struct pop_aid_log_event_t {
struct log_event {
log_event* next;
log_event* prev;
std::string msg;
log_event(std::string logmsg = "") : msg(std::move(logmsg)) {
// nop
}
};
#ifndef CAF_LOG_LEVEL
......@@ -75,7 +79,7 @@ class logging_impl : public logging {
void stop() {
log("TRACE", "logging", "run", __FILE__, __LINE__, "EXIT");
// an empty string means: shut down
m_queue.synchronized_enqueue(m_queue_mtx, m_queue_cv, new log_event{0, ""});
m_queue.synchronized_enqueue(m_queue_mtx, m_queue_cv, new log_event{""});
m_thread.join();
}
......@@ -123,7 +127,7 @@ class logging_impl : public logging {
<< class_name << " " << function_name << " " << file_name << ":"
<< line_num << " " << msg << std::endl;
m_queue.synchronized_enqueue(m_queue_mtx, m_queue_cv,
new log_event{nullptr, line.str()});
new log_event{line.str()});
}
private:
......
......@@ -21,12 +21,22 @@
namespace caf {
mailbox_element::mailbox_element()
: next(nullptr),
prev(nullptr),
marked(false) {
// nop
}
mailbox_element::mailbox_element(actor_addr arg0, message_id arg1, message arg2)
: next(nullptr),
prev(nullptr),
marked(false),
sender(std::move(arg0)),
mid(arg1),
msg(std::move(arg2)) {}
msg(std::move(arg2)) {
// nop
}
mailbox_element::~mailbox_element() {
// nop
......
......@@ -358,10 +358,10 @@ class broker : public extend<local_actor>::
std::map<accept_handle, doorman_pointer> m_doormen;
std::map<connection_handle, scribe_pointer> m_scribes;
policy::not_prioritizing m_priority_policy;
policy::sequential_invoke m_invoke_policy;
middleman& m_mm;
std::deque<unique_mailbox_element_pointer> m_cache;
};
class broker::functor_based : public extend<broker>::
......
......@@ -172,8 +172,8 @@ void broker::invoke_message(const actor_addr& sender, message_id mid,
try {
auto bhvr = bhvr_stack().back();
auto bid = bhvr_stack().back_id();
switch (m_invoke_policy.handle_message(this, &m_dummy_node, bhvr, bid)) {
case policy::hm_msg_handled: {
switch (m_invoke_policy.invoke_message(this, m_dummy_node, bhvr, bid)) {
case policy::im_success: {
CAF_LOG_DEBUG("handle_message returned hm_msg_handled");
while (!bhvr_stack().empty()
&& planned_exit_reason() == exit_reason::not_exited
......@@ -182,15 +182,14 @@ void broker::invoke_message(const actor_addr& sender, message_id mid,
}
break;
}
case policy::hm_drop_msg:
case policy::im_dropped:
CAF_LOG_DEBUG("handle_message returned hm_drop_msg");
break;
case policy::hm_skip_msg:
case policy::hm_cache_msg: {
case policy::im_skipped: {
CAF_LOG_DEBUG("handle_message returned hm_skip_msg or hm_cache_msg");
auto e = mailbox_element::create(sender, bid,
std::move(m_dummy_node.msg));
m_priority_policy.push_to_cache(unique_mailbox_element_pointer{e});
m_cache.push_back(unique_mailbox_element_pointer{e});
break;
}
}
......@@ -227,17 +226,20 @@ bool broker::invoke_message_from_cache() {
CAF_LOG_TRACE("");
auto bhvr = bhvr_stack().back();
auto mid = bhvr_stack().back_id();
auto e = m_priority_policy.cache_end();
CAF_LOG_DEBUG(std::distance(m_priority_policy.cache_begin(), e)
<< " elements in cache");
for (auto i = m_priority_policy.cache_begin(); i != e; ++i) {
auto res = m_invoke_policy.invoke_message(this, *i, bhvr, mid);
if (res || !*i) {
m_priority_policy.cache_erase(i);
if (res){
auto i = m_cache.begin();
auto e = m_cache.end();
CAF_LOG_DEBUG(std::distance(i, e) << " elements in cache");
while (i != e) {
switch (m_invoke_policy.invoke_message(this, *(i->get()), bhvr, mid)) {
case policy::im_success:
m_cache.erase(i);
return true;
}
return invoke_message_from_cache();
case policy::im_skipped:
++i;
break;
case policy::im_dropped:
i = m_cache.erase(i);
break;
}
}
return false;
......
......@@ -21,6 +21,7 @@
#include "test.hpp"
#include "caf/policy/prioritizing.hpp"
#include "caf/detail/single_reader_queue.hpp"
using std::begin;
......@@ -32,10 +33,17 @@ size_t s_iint_instances = 0;
struct iint {
iint* next;
iint* prev;
int value;
inline iint(int val = 0) : next(nullptr), value(val) { ++s_iint_instances; }
~iint() { --s_iint_instances; }
bool is_high_priority() const {
return value % 2 == 0;
}
iint(int val = 0) : next(nullptr), prev(nullptr), value(val) {
++s_iint_instances;
}
~iint() {
--s_iint_instances;
}
};
inline bool operator==(const iint& lhs, const iint& rhs) {
......@@ -48,27 +56,48 @@ inline bool operator==(int lhs, const iint& rhs) { return lhs == rhs.value; }
using iint_queue = caf::detail::single_reader_queue<iint>;
struct pseudo_actor {
using mailbox_type = caf::detail::single_reader_queue<iint>;
mailbox_type mbox;
mailbox_type& mailbox() {
return mbox;
}
};
int main() {
CAF_TEST(test_intrusive_containers);
caf::detail::single_reader_queue<iint> q;
q.enqueue(new iint(1));
q.enqueue(new iint(2));
q.enqueue(new iint(3));
CAF_CHECK_EQUAL(3, s_iint_instances);
auto x = q.try_pop();
CAF_CHECK_EQUAL(x->value, 1);
delete x;
x = q.try_pop();
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));
CAF_CHECK_EQUAL(baseline + 4, s_iint_instances);
caf::policy::prioritizing policy;
// first "high priority message"
CAF_CHECK_EQUAL(self.mbox.count(), 4);
CAF_CHECK_EQUAL(policy.has_next_message(&self), true);
auto x = policy.next_message(&self);
CAF_CHECK_EQUAL(x->value, 2);
delete x;
x = q.try_pop();
// second "high priority message"
CAF_CHECK_EQUAL(self.mbox.count(), 3);
CAF_CHECK_EQUAL(policy.has_next_message(&self), true);
x = policy.next_message(&self);
CAF_CHECK_EQUAL(x->value, 4);
// first "low priority message"
CAF_CHECK_EQUAL(self.mbox.count(), 2);
CAF_CHECK_EQUAL(policy.has_next_message(&self), true);
x = policy.next_message(&self);
CAF_CHECK_EQUAL(x->value, 1);
// first "low priority message"
CAF_CHECK_EQUAL(self.mbox.count(), 1);
CAF_CHECK_EQUAL(policy.has_next_message(&self), true);
x = policy.next_message(&self);
CAF_CHECK_EQUAL(x->value, 3);
delete x;
x = q.try_pop();
CAF_CHECK(x == nullptr);
x.reset();
// back to baseline
CAF_CHECK_EQUAL(self.mbox.count(), 0);
CAF_CHECK_EQUAL(baseline, s_iint_instances);
return CAF_TEST_RESULT();
}
......@@ -296,6 +296,7 @@ simple_mirror::~simple_mirror() {
behavior simple_mirror::make_behavior() {
return {
others() >> [=] {
CAF_CHECKPOINT();
return last_dequeued();
}
};
......
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