Commit 2cbd028b authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Revise message_id and add intrusive containers

parent 1feb1109
...@@ -29,7 +29,7 @@ ...@@ -29,7 +29,7 @@
#include <limits> #include <limits>
#include <condition_variable> // std::cv_status #include <condition_variable> // std::cv_status
#include "caf/detail/intrusive_partitioned_list.hpp" #include "caf/intrusive/partitioned_list.hpp"
namespace caf { namespace caf {
namespace detail { namespace detail {
...@@ -54,10 +54,14 @@ template <class T, class Delete = std::default_delete<T>> ...@@ -54,10 +54,14 @@ template <class T, class Delete = std::default_delete<T>>
class single_reader_queue { class single_reader_queue {
public: public:
using value_type = T; using value_type = T;
using pointer = value_type*; using pointer = value_type*;
using deleter_type = Delete; using deleter_type = Delete;
using unique_pointer = std::unique_ptr<value_type, deleter_type>; using unique_pointer = std::unique_ptr<value_type, deleter_type>;
using cache_type = intrusive_partitioned_list<value_type, deleter_type>;
using cache_type = intrusive::partitioned_list<value_type, deleter_type>;
/// Tries to dequeue a new element from the mailbox. /// Tries to dequeue a new element from the mailbox.
/// @warning Call only from the reader (owner). /// @warning Call only from the reader (owner).
...@@ -162,12 +166,17 @@ public: ...@@ -162,12 +166,17 @@ public:
fetch_new_data(); fetch_new_data();
auto ptr = head_; auto ptr = head_;
while (ptr && res < max_count) { while (ptr && res < max_count) {
ptr = ptr->next; ptr = promote(ptr->next);
++res; ++res;
} }
return res; return res;
} }
void prepend(pointer ptr) {
ptr->next = head_;
head_ = ptr;
}
pointer peek() { pointer peek() {
if (head_ != nullptr || fetch_new_data()) if (head_ != nullptr || fetch_new_data())
return head_; return head_;
...@@ -195,7 +204,7 @@ public: ...@@ -195,7 +204,7 @@ public:
auto tail = old_head; auto tail = old_head;
while (tail->next != nullptr) while (tail->next != nullptr)
tail = tail->next; tail = tail->next;
// This gets new data from the stack and rewrite head_. // This gets new data from the stack and rewrites head_.
fetch_new_data(); fetch_new_data();
tail->next = head_; tail->next = head_;
head_ = old_head; head_ = old_head;
...@@ -265,6 +274,23 @@ public: ...@@ -265,6 +274,23 @@ public:
return true; return true;
} }
/// Take the stack pointer without re-ordering it to the cache.
pointer take_stack() {
auto end_ptr = stack_empty_dummy();
pointer e = stack_.load();
// must not be called on a closed queue
CAF_ASSERT(e != nullptr);
// fetching data while blocked is an error
CAF_ASSERT(e != reader_blocked_dummy());
// it's enough to check this once, since only the owner is allowed
// to close the queue and only the owner is allowed to call this
// member function
while (e != end_ptr)
if (stack_.compare_exchange_weak(e, end_ptr))
return e;
return nullptr;
}
private: private:
// exposed to "outside" access // exposed to "outside" access
std::atomic<pointer> stack_; std::atomic<pointer> stack_;
...@@ -272,7 +298,7 @@ private: ...@@ -272,7 +298,7 @@ private:
// accessed only by the owner // accessed only by the owner
pointer head_; pointer head_;
deleter_type delete_; deleter_type delete_;
intrusive_partitioned_list<value_type, deleter_type> cache_; intrusive::partitioned_list<value_type, deleter_type> cache_;
// atomically sets stack_ back and enqueues all elements to the cache // atomically sets stack_ back and enqueues all elements to the cache
bool fetch_new_data(pointer end_ptr) { bool fetch_new_data(pointer end_ptr) {
...@@ -299,7 +325,7 @@ private: ...@@ -299,7 +325,7 @@ private:
auto next = e->next; auto next = e->next;
e->next = head_; e->next = head_;
head_ = e; head_ = e;
e = next; e = promote(next);
} }
return true; return true;
} }
...@@ -315,7 +341,7 @@ private: ...@@ -315,7 +341,7 @@ private:
pointer take_head() { pointer take_head() {
if (head_ != nullptr || fetch_new_data()) { if (head_ != nullptr || fetch_new_data()) {
auto result = head_; auto result = head_;
head_ = head_->next; head_ = promote(head_->next);
return result; return result;
} }
return nullptr; return nullptr;
...@@ -324,7 +350,7 @@ private: ...@@ -324,7 +350,7 @@ private:
template <class F> template <class F>
void clear_cached_elements(const F& f) { void clear_cached_elements(const F& f) {
while (head_) { while (head_) {
auto next = head_->next; auto next = promote(head_->next);
f(*head_); f(*head_);
delete_(head_); delete_(head_);
head_ = next; head_ = next;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* 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_UNORDERED_FLAT_MAP_HPP
#define CAF_DETAIL_UNORDERED_FLAT_MAP_HPP
#include <vector>
#include <algorithm>
#include <stdexcept>
#include <functional>
#include "caf/detail/comparable.hpp"
namespace caf {
namespace detail {
/// A map abstraction with an unsorted `std::vector` providing `O(n)` lookup.
template <class Key, class T,
class Allocator = std::allocator<std::pair<Key, T>>>
class unordered_flat_map {
public:
// -- member types ----------------------------------------------------------
using key_type = Key;
using mapped_type = T;
using value_type = std::pair<Key, T>;
using vector_type = std::vector<value_type, Allocator>;
using allocator_type = typename vector_type::allocator_type;
using size_type = typename vector_type::size_type;
using difference_type = typename vector_type::difference_type;
using reference = typename vector_type::reference;
using const_reference = typename vector_type::const_reference;
using pointer = typename vector_type::pointer;
using const_pointer = typename vector_type::const_pointer;
using iterator = typename vector_type::iterator;
using const_iterator = typename vector_type::const_iterator;
using reverse_iterator = typename vector_type::reverse_iterator;
using const_reverse_iterator = typename vector_type::const_reverse_iterator;
// -- constructors, destructors, and assignment operators -------------------
unordered_flat_map() = default;
unordered_flat_map(std::initializer_list<value_type> l) : xs_(l) {
// nop
}
template <class InputIterator>
unordered_flat_map(InputIterator first, InputIterator last)
: xs_(first, last) {
// nop
}
// -- non-const begin iterators ---------------------------------------------
iterator begin() {
return xs_.begin();
}
reverse_iterator rbegin() {
return xs_.rbegin();
}
// -- const begin iterators -------------------------------------------------
const_iterator begin() const {
return xs_.begin();
}
const_iterator cbegin() const {
return xs_.cbegin();
}
const_reverse_iterator rbegin() const {
return xs_.rbegin();
}
// -- non-const end iterators -----------------------------------------------
iterator end() {
return xs_.end();
}
reverse_iterator rend() {
return xs_.rend();
}
// -- const end iterators ---------------------------------------------------
const_iterator end() const {
return xs_.end();
}
const_iterator cend() const {
return xs_.end();
}
const_reverse_iterator rend() const {
return xs_.rend();
}
// -- capacity --------------------------------------------------------------
bool empty() const {
return xs_.empty();
}
size_type size() const {
return xs_.size();
}
void reserve(size_type count) {
xs_.reserve(count);
}
void shrink_to_fit() {
xs_.shrink_to_fit();
}
// -- modifiers -------------------------------------------------------------
void clear() {
return xs_.clear();
}
std::pair<iterator, bool> insert(value_type x) {
return insert(end(), std::move(x));
}
std::pair<iterator, bool> insert(iterator hint, value_type x) {
return insert(static_cast<const_iterator>(hint), std::move(x));
}
std::pair<iterator, bool> insert(const_iterator hint, value_type x) {
auto i = find(x.first);
if (i == end())
return {xs_.insert(hint, std::move(x)), true};
return {i, false};
}
template <class InputIterator>
void insert(InputIterator first, InputIterator last) {
while (first != last)
insert(*first++);
}
template <class... Ts>
std::pair<iterator, bool> emplace(Ts&&... xs) {
return emplace_hint(end(), std::forward<Ts>(xs)...);
}
template <class... Ts>
std::pair<iterator, bool> emplace_hint(const_iterator hint, Ts&&... xs) {
return insert(hint, value_type(std::forward<Ts>(xs)...));
}
iterator erase(const_iterator i) {
return xs_.erase(i);
}
iterator erase(const_iterator first, const_iterator last) {
return xs_.erase(first, last);
}
size_type erase(const key_type& x) {
auto pred = [&](const value_type& y) {
return x == y.first;
};
auto i = std::remove_if(begin(), end(), pred);
if (i == end())
return 0;
erase(i);
return 1;
}
void swap(unordered_flat_map& other) {
xs_.swap(other);
}
// -- lookup ----------------------------------------------------------------
template <class K>
mapped_type& at(const K& key) {
auto i = find(key);
if (i == end())
throw std::out_of_range{"caf::detail::unordered_flat_map::at"};
return i->second;
}
template <class K>
const mapped_type& at(const K& key) const {
auto i = find(key);
if (i == end())
throw std::out_of_range{"caf::detail::unordered_flat_map::at"};
return i->second;
}
mapped_type& operator[](const key_type& key) {
auto i = find(key);
if (i != end())
return i->second;
return xs_.insert(i, value_type{key, mapped_type{}})->second;
}
template <class K>
iterator find(const K& x) {
auto pred = [&](const value_type& y) {
return x == y.first;
};
return std::find_if(xs_.begin(), xs_.end(), pred);
}
template <class K>
const_iterator find(const K& x) const {
auto pred = [&](const value_type& y) {
return x == y.first;
};
return std::find_if(xs_.begin(), xs_.end(), pred);
}
template <class K>
size_type count(const K& x) const {
return find(x) == end() ? 0 : 1;
}
private:
vector_type xs_;
};
template <class K, class T, class A>
bool operator==(const unordered_flat_map<K, T, A>& xs,
const unordered_flat_map<K, T, A>& ys) {
return xs.size() == ys.size() && std::equal(xs.begin(), xs.end(), ys.begin());
}
template <class K, class T, class A>
bool operator!=(const unordered_flat_map<K, T, A>& xs,
const unordered_flat_map<K, T, A>& ys) {
return !(xs == ys);
}
template <class K, class T, class A>
bool operator<(const unordered_flat_map<K, T, A>& xs,
const unordered_flat_map<K, T, A>& ys) {
return std::lexicographical_compare(xs.begin(), xs.end(),
ys.begin(), ys.end());
}
template <class K, class T, class A>
bool operator>=(const unordered_flat_map<K, T, A>& xs,
const unordered_flat_map<K, T, A>& ys) {
return !(xs < ys);
}
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_UNORDERED_FLAT_MAP_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* 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_INTRUSIVE_BIDIRECTIONAL_ITERATOR_HPP
#define CAF_INTRUSIVE_BIDIRECTIONAL_ITERATOR_HPP
#include <cstddef>
#include <iterator>
#include <type_traits>
#include "caf/intrusive/doubly_linked.hpp"
namespace caf {
namespace intrusive {
/// Intrusive base for doubly linked types that allows queues to use `T` with
/// node nodes.
template <class T>
class bidirectional_iterator {
public:
// -- member types -----------------------------------------------------------
using difference_type = std::ptrdiff_t;
using value_type = T;
using pointer = value_type*;
using const_pointer = const value_type*;
using reference = value_type&;
using const_reference = const value_type&;
using node_type =
typename std::conditional<
std::is_const<T>::value,
const typename T::node_type,
typename T::node_type
>::type;
using node_pointer = node_type*;
using iterator_category = std::bidirectional_iterator_tag;
// -- member variables -------------------------------------------------------
node_pointer ptr;
// -- constructors, destructors, and assignment operators --------------------
constexpr bidirectional_iterator(node_pointer init = nullptr) : ptr(init) {
// nop
}
bidirectional_iterator(const bidirectional_iterator&) = default;
bidirectional_iterator& operator=(const bidirectional_iterator&) = default;
// -- convenience functions --------------------------------------------------
bidirectional_iterator next() {
return ptr->next;
}
// -- operators --------------------------------------------------------------
bidirectional_iterator& operator++() {
ptr = promote(ptr->next);
return *this;
}
bidirectional_iterator operator++(int) {
bidirectional_iterator res = *this;
ptr = promote(ptr->next);
return res;
}
bidirectional_iterator& operator--() {
ptr = ptr->prev;
return *this;
}
bidirectional_iterator operator--(int) {
bidirectional_iterator res = *this;
ptr = promote(ptr->prev);
return res;
}
reference operator*() {
return *promote(ptr);
}
const_reference operator*() const {
return *promote(ptr);
}
pointer operator->() {
return promote(ptr);
}
const pointer operator->() const {
return promote(ptr);
}
};
/// @relates bidirectional_iterator
template <class T>
bool operator==(const bidirectional_iterator<T>& x,
const bidirectional_iterator<T>& y) {
return x.ptr == y.ptr;
}
/// @relates bidirectional_iterator
template <class T>
bool operator!=(const bidirectional_iterator<T>& x,
const bidirectional_iterator<T>& y) {
return x.ptr != y.ptr;
}
} // namespace intrusive
} // namespace caf
#endif // CAF_INTRUSIVE_BIDIRECTIONAL_ITERATOR_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* 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_INTRUSIVE_DOUBLY_LINKED_HPP
#define CAF_INTRUSIVE_DOUBLY_LINKED_HPP
namespace caf {
namespace intrusive {
/// Intrusive base for doubly linked types that allows queues to use `T` with
/// dummy nodes.
template <class T>
struct doubly_linked {
// -- member types -----------------------------------------------------------
/// The type for dummy nodes in doubly linked lists.
using node_type = doubly_linked<T>;
/// Type of the pointer connecting two doubly linked nodes.
using node_pointer = node_type*;
// -- constructors, destructors, and assignment operators --------------------
doubly_linked(node_pointer n = nullptr, node_pointer p = nullptr)
: next(n),
prev(p) {
// nop
}
// -- member variables -------------------------------------------------------
/// Intrusive pointer to the next element.
node_pointer next;
/// Intrusive pointer to the previous element.
node_pointer prev;
};
template <class T>
T* promote(doubly_linked<T>* ptr) {
return static_cast<T*>(ptr);
}
template <class T>
const T* promote(const doubly_linked<T>* ptr) {
return static_cast<const T*>(ptr);
}
} // namespace intrusive
} // namespace caf
#endif // CAF_INTRUSIVE_DOUBLY_LINKED_HPP
...@@ -16,8 +16,8 @@ ...@@ -16,8 +16,8 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_DETAIL_INTRUSIVE_PARTITIONED_LIST_HPP #ifndef CAF_INTRUSIVE_PARTITIONED_LIST_HPP
#define CAF_DETAIL_INTRUSIVE_PARTITIONED_LIST_HPP #define CAF_INTRUSIVE_PARTITIONED_LIST_HPP
#include "caf/config.hpp" #include "caf/config.hpp"
...@@ -28,8 +28,11 @@ ...@@ -28,8 +28,11 @@
#include "caf/behavior.hpp" #include "caf/behavior.hpp"
#include "caf/invoke_message_result.hpp" #include "caf/invoke_message_result.hpp"
#include "caf/intrusive/doubly_linked.hpp"
#include "caf/intrusive/bidirectional_iterator.hpp"
namespace caf { namespace caf {
namespace detail { namespace intrusive {
/// Describes a partitioned list of elements. The first part /// Describes a partitioned list of elements. The first part
/// of the list is described by the iterator pair `[begin, separator)`. /// of the list is described by the iterator pair `[begin, separator)`.
...@@ -37,82 +40,39 @@ namespace detail { ...@@ -37,82 +40,39 @@ namespace detail {
/// of the list to store previously skipped elements. Priority-aware actors /// of the list to store previously skipped elements. Priority-aware actors
/// also use the first half of the list to sort messages by priority. /// also use the first half of the list to sort messages by priority.
template <class T, class Delete = std::default_delete<T>> template <class T, class Delete = std::default_delete<T>>
class intrusive_partitioned_list { class partitioned_list {
public: public:
using value_type = T; // -- member types -----------------------------------------------------------
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) { using value_type = T;
iterator res = *this;
ptr = ptr->prev;
return res;
}
const value_type& operator*() const { using pointer = value_type*;
return *ptr;
}
value_type& operator*() { using node_type = typename value_type::node_type;
return *ptr;
}
pointer operator->() { using deleter_type = Delete;
return ptr;
}
bool operator==(const iterator& other) const { using iterator = bidirectional_iterator<T>;
return ptr == other.ptr;
}
bool operator!=(const iterator& other) const { using range = std::pair<iterator, iterator>;
return ptr != other.ptr;
}
iterator next() const { // -- constructors, destructors, and assignment operators --------------------
return ptr->next;
}
};
intrusive_partitioned_list() { partitioned_list() {
head_.next = &separator_; head_.next = &separator_;
separator_.prev = &head_; separator_.prev = &head_;
separator_.next = &tail_; separator_.next = &tail_;
tail_.prev = &separator_; tail_.prev = &separator_;
} }
~intrusive_partitioned_list() { ~partitioned_list() {
clear(); clear();
} }
// -- iterators --------------------------------------------------------------
iterator begin() { iterator begin() {
return head_.next; return promote(head_.next);
} }
iterator separator() { iterator separator() {
...@@ -120,15 +80,13 @@ public: ...@@ -120,15 +80,13 @@ public:
} }
iterator continuation() { iterator continuation() {
return separator_.next; return promote(separator_.next);
} }
iterator end() { iterator end() {
return &tail_; return &tail_;
} }
using range = std::pair<iterator, iterator>;
/// Returns the two iterator pairs describing the first and second part /// Returns the two iterator pairs describing the first and second part
/// of the partitioned list. /// of the partitioned list.
std::array<range, 2> ranges() { std::array<range, 2> ranges() {
...@@ -143,8 +101,8 @@ public: ...@@ -143,8 +101,8 @@ public:
while (i != e) { while (i != e) {
auto ptr = i.ptr; auto ptr = i.ptr;
++i; ++i;
f(*ptr); f(*promote(ptr));
delete_(ptr); delete_(promote(ptr));
} }
} }
if (head_.next != &separator_) { if (head_.next != &separator_) {
...@@ -181,13 +139,13 @@ public: ...@@ -181,13 +139,13 @@ public:
auto prev = res->prev; auto prev = res->prev;
prev->next = next; prev->next = next;
next->prev = prev; next->prev = prev;
return res; return promote(res);
} }
iterator erase(iterator pos) { iterator erase(iterator pos) {
auto next = pos->next; auto next = pos->next;
delete_(take(pos)); delete_(take(pos));
return next; return promote(next);
} }
size_t count(size_t max_count = std::numeric_limits<size_t>::max()) { size_t count(size_t max_count = std::numeric_limits<size_t>::max()) {
...@@ -200,13 +158,13 @@ public: ...@@ -200,13 +158,13 @@ public:
} }
private: private:
value_type head_; node_type head_;
value_type separator_; node_type separator_;
value_type tail_; node_type tail_;
deleter_type delete_; deleter_type delete_;
}; };
} // namespace detail } // namespace intrusive
} // namespace caf } // namespace caf
#endif // CAF_DETAIL_INTRUSIVE_PARTITIONED_LIST_HPP #endif // CAF_INTRUSIVE_PARTITIONED_LIST_HPP
...@@ -37,6 +37,8 @@ ...@@ -37,6 +37,8 @@
#include "caf/abstract_actor.hpp" #include "caf/abstract_actor.hpp"
#include "caf/deep_to_string.hpp" #include "caf/deep_to_string.hpp"
#include "caf/intrusive/doubly_linked.hpp"
#include "caf/detail/scope_guard.hpp" #include "caf/detail/scope_guard.hpp"
#include "caf/detail/shared_spinlock.hpp" #include "caf/detail/shared_spinlock.hpp"
#include "caf/detail/pretty_type_name.hpp" #include "caf/detail/pretty_type_name.hpp"
...@@ -65,27 +67,36 @@ public: ...@@ -65,27 +67,36 @@ public:
friend class actor_system; friend class actor_system;
/// Encapsulates a single logging event. /// Encapsulates a single logging event.
struct event { struct event : intrusive::doubly_linked<event> {
/// Intrusive pointer to the next logging event. event() = default;
event* next;
/// Intrusive pointer to the previous logging event. event(int lvl, const char* cat, const char* fun, const char* fn, int line,
event* prev; std::string msg, std::thread::id t, actor_id a, timestamp ts);
/// Level/priority of the event. /// Level/priority of the event.
int level; int level;
/// Name of the category (component) logging the event. /// Name of the category (component) logging the event.
const char* category_name; const char* category_name;
/// Name of the current context as reported by `__PRETTY_FUNCTION__`. /// Name of the current context as reported by `__PRETTY_FUNCTION__`.
const char* pretty_fun; const char* pretty_fun;
/// Name of the current file. /// Name of the current file.
const char* file_name; const char* file_name;
/// Current line in the file. /// Current line in the file.
int line_number; int line_number;
/// User-provided message. /// User-provided message.
std::string message; std::string message;
/// Thread ID of the caller. /// Thread ID of the caller.
std::thread::id tid; std::thread::id tid;
/// Actor ID of the caller. /// Actor ID of the caller.
actor_id aid; actor_id aid;
/// Timestamp of the event. /// Timestamp of the event.
timestamp tstamp; timestamp tstamp;
}; };
...@@ -343,8 +354,8 @@ inline caf::actor_id caf_set_aid_dummy() { return 0; } ...@@ -343,8 +354,8 @@ inline caf::actor_id caf_set_aid_dummy() { return 0; }
&& CAF_UNIFYN(caf_logger)->accepts(loglvl, component)) \ && CAF_UNIFYN(caf_logger)->accepts(loglvl, component)) \
CAF_UNIFYN(caf_logger) \ CAF_UNIFYN(caf_logger) \
->log(new ::caf::logger::event{ \ ->log(new ::caf::logger::event{ \
nullptr, nullptr, loglvl, component, CAF_PRETTY_FUN, __FILE__, \ loglvl, component, CAF_PRETTY_FUN, __FILE__, __LINE__, \
__LINE__, (::caf::logger::line_builder{} << message).get(), \ (::caf::logger::line_builder{} << message).get(), \
::std::this_thread::get_id(), \ ::std::this_thread::get_id(), \
CAF_UNIFYN(caf_logger)->thread_local_aid(), \ CAF_UNIFYN(caf_logger)->thread_local_aid(), \
::caf::make_timestamp()}); \ ::caf::make_timestamp()}); \
......
...@@ -31,6 +31,8 @@ ...@@ -31,6 +31,8 @@
#include "caf/type_erased_tuple.hpp" #include "caf/type_erased_tuple.hpp"
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
#include "caf/intrusive/doubly_linked.hpp"
#include "caf/meta/type_name.hpp" #include "caf/meta/type_name.hpp"
#include "caf/meta/omittable_if_empty.hpp" #include "caf/meta/omittable_if_empty.hpp"
...@@ -40,16 +42,12 @@ ...@@ -40,16 +42,12 @@
namespace caf { namespace caf {
class mailbox_element : public memory_managed, public message_view { class mailbox_element : public memory_managed,
public intrusive::doubly_linked<mailbox_element>,
public message_view {
public: public:
using forwarding_stack = std::vector<strong_actor_ptr>; using forwarding_stack = std::vector<strong_actor_ptr>;
/// Intrusive pointer to the next mailbox element.
mailbox_element* next;
/// Intrusive pointer to the previous mailbox element.
mailbox_element* prev;
/// Avoids multi-processing in blocking actors via flagging. /// Avoids multi-processing in blocking actors via flagging.
bool marked; bool marked;
...@@ -78,15 +76,15 @@ public: ...@@ -78,15 +76,15 @@ public:
const type_erased_tuple& content() const; const type_erased_tuple& content() const;
inline bool is_high_priority() const {
return mid.category() == message_id::urgent_message_category;
}
mailbox_element(mailbox_element&&) = delete; mailbox_element(mailbox_element&&) = delete;
mailbox_element(const mailbox_element&) = delete; mailbox_element(const mailbox_element&) = delete;
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;
inline bool is_high_priority() const {
return mid.is_high_priority();
}
protected: protected:
empty_type_erased_tuple dummy_; empty_type_erased_tuple dummy_;
}; };
......
...@@ -44,7 +44,7 @@ constexpr invalid_message_id_t invalid_message_id = invalid_message_id_t{}; ...@@ -44,7 +44,7 @@ constexpr invalid_message_id_t invalid_message_id = invalid_message_id_t{};
struct make_message_id_t; struct make_message_id_t;
/// Denotes whether a message is asynchronous or synchronous /// Denotes whether a message is asynchronous or synchronous.
/// @note Asynchronous messages always have an invalid message id. /// @note Asynchronous messages always have an invalid message id.
class message_id : detail::comparable<message_id> { class message_id : detail::comparable<message_id> {
public: public:
...@@ -54,80 +54,157 @@ public: ...@@ -54,80 +54,157 @@ public:
// -- constants ------------------------------------------------------------- // -- constants -------------------------------------------------------------
/// The first bit flags response messages.
static constexpr uint64_t response_flag_mask = 0x8000000000000000; static constexpr uint64_t response_flag_mask = 0x8000000000000000;
/// The second bit flags whether the actor already responded.
static constexpr uint64_t answered_flag_mask = 0x4000000000000000; static constexpr uint64_t answered_flag_mask = 0x4000000000000000;
static constexpr uint64_t high_prioity_flag_mask = 0x2000000000000000;
static constexpr uint64_t request_id_mask = 0x1FFFFFFFFFFFFFFF;
// The third and fourth bit are used to categorize messages.
static constexpr uint64_t category_flag_mask = 0x3000000000000000;
/// The trailing 60 bits are used for the actual ID.
static constexpr uint64_t request_id_mask = 0x0FFFFFFFFFFFFFFF;
/// Identifies one-to-one messages with normal priority.
static constexpr uint64_t default_message_category = 0; // 0b00
/// Identifies stream messages received from upstream actors, e.g.,
/// `stream_msg::batch`.
static constexpr uint64_t upstream_message_category = 1; // 0b01
/// Identifies stream messages received from downstream actors, e.g.,
/// `stream_msg::ack_batch`.
static constexpr uint64_t downstream_message_category = 2; // 0b10
/// Identifies one-to-one messages with high priority.
static constexpr uint64_t urgent_message_category = 3; // 0b11
/// Number of bits trailing the category.
static constexpr uint64_t category_offset = 60;
// -- constructors, destructors, and assignment operators -------------------
/// Constructs a message ID for asynchronous messages with normal priority.
constexpr message_id() : value_(0) { constexpr message_id() : value_(0) {
// nop // nop
} }
constexpr message_id(invalid_message_id_t) : value_(0) { /// Constructs a message ID for asynchronous messages with normal priority.
constexpr message_id(invalid_message_id_t) : message_id() {
// nop // nop
} }
message_id(message_id&&) = default;
message_id(const message_id&) = default; message_id(const message_id&) = default;
message_id& operator=(message_id&&) = default;
message_id& operator=(const message_id&) = default; message_id& operator=(const message_id&) = default;
inline message_id& operator++() { // -- observers -------------------------------------------------------------
++value_;
return *this;
}
/// Returns whether a message is asynchronous, i.e., neither a request nor a
/// stream message.
inline bool is_async() const { inline bool is_async() const {
return value_ == 0 || value_ == high_prioity_flag_mask; return value_ == 0
|| value_ == (urgent_message_category << category_offset);
} }
/// Returns whether a message is a response to a previously send request.
inline bool is_response() const { inline bool is_response() const {
return (value_ & response_flag_mask) != 0; return (value_ & response_flag_mask) != 0;
} }
/// Returns whether a message is a request.
inline bool is_request() const {
return valid() && !is_response();
}
/// Returns whether a message is tagged as answered by the receiving actor.
inline bool is_answered() const { inline bool is_answered() const {
return (value_ & answered_flag_mask) != 0; return (value_ & answered_flag_mask) != 0;
} }
inline bool is_high_priority() const { /// Returns the message category, i.e., one of `default_message_category`,
return (value_ & high_prioity_flag_mask) != 0; /// `upstream_message_category`, `downstream_message_category`, or
/// `urgent_message_category`.
inline uint64_t category() const {
return (value_ & category_flag_mask) >> category_offset;
} }
inline bool valid() const { /// Returns whether `category() == default_message_category`.
return (value_ & request_id_mask) != 0; inline bool is_default_message() const {
return (value_ & category_flag_mask) == 0;
} }
inline bool is_request() const { /// Returns whether `category() == urgent_message_category`.
return valid() && !is_response(); inline bool is_urgent_message() const {
return (value_ & category_flag_mask)
== (urgent_message_category << category_offset);
} }
/// Returns whether `category() == upstream_message_category`.
inline bool is_upstream_message() const {
return (value_ & category_flag_mask)
== (urgent_message_category << category_offset);
}
/// Returns whether `category() == downstream_message_category`.
inline bool is_downstream_message() const {
return (value_ & category_flag_mask)
== (urgent_message_category << category_offset);
}
/// Returns whether a message is a request or a response.
inline bool valid() const {
return (value_ & request_id_mask) != 0;
}
/// Returns a response ID for the current request or an asynchronous ID with
/// the same priority as this ID.
inline message_id response_id() const { inline message_id response_id() const {
return message_id{is_request() ? value_ | response_flag_mask : 0}; if (is_request())
return message_id{value_ | response_flag_mask};
return message_id{is_urgent_message()
? urgent_message_category << category_offset
: 0u};
} }
/// Extracts the request number part of this ID.
inline message_id request_id() const { inline message_id request_id() const {
return message_id(value_ & request_id_mask); return message_id{value_ & request_id_mask};
} }
/// Returns the same ID but with high instead of default message priority.
/// @pre `!is_upstream_message() && !is_downstream_message()`
inline message_id with_high_priority() const { inline message_id with_high_priority() const {
return message_id(value_ | high_prioity_flag_mask); return message_id{value_ | category_flag_mask}; // works because urgent == 0b11
} }
/// Returns the same ID but with high instead of default message priority.
inline message_id with_normal_priority() const { inline message_id with_normal_priority() const {
return message_id(value_ & ~high_prioity_flag_mask); return message_id{value_ & ~category_flag_mask}; // works because normal == 0b00
}
inline void mark_as_answered() {
value_ |= answered_flag_mask;
} }
/// Returns the "raw bytes" for this ID.
inline uint64_t integer_value() const { inline uint64_t integer_value() const {
return value_; return value_;
} }
long compare(const message_id& other) const { /// Returns a negative value if `*this < other`, zero if `*this == other`,
return (value_ == other.value_) ? 0 /// and a positive value otherwise.
: (value_ < other.value_ ? -1 : 1); inline int64_t compare(const message_id& other) const {
return static_cast<int64_t>(value_) - static_cast<int64_t>(other.value_);
}
// -- mutators --------------------------------------------------------------
inline message_id& operator++() {
++value_;
return *this;
}
/// Sets the flag for marking an incoming message as answered.
inline void mark_as_answered() {
value_ |= answered_flag_mask;
} }
template <class Inspector> template <class Inspector>
...@@ -142,7 +219,7 @@ public: ...@@ -142,7 +219,7 @@ public:
CAF_DEPRECATED_MSG("use make_message_id instead"); CAF_DEPRECATED_MSG("use make_message_id instead");
private: private:
constexpr message_id(uint64_t value) : value_(value) { constexpr explicit message_id(uint64_t value) : value_(value) {
// nop // nop
} }
...@@ -159,12 +236,11 @@ struct make_message_id_t { ...@@ -159,12 +236,11 @@ struct make_message_id_t {
} }
constexpr message_id operator()(uint64_t value = 0) const { constexpr message_id operator()(uint64_t value = 0) const {
return value; return message_id{value};
} }
constexpr message_id operator()(message_priority p) const { constexpr message_id operator()(message_priority p) const {
return p == message_priority::high ? message_id::high_prioity_flag_mask return message_id{static_cast<uint64_t>(p) << message_id::category_offset};
: 0u;
} }
}; };
......
...@@ -25,7 +25,7 @@ namespace caf { ...@@ -25,7 +25,7 @@ namespace caf {
enum class message_priority : uint32_t { enum class message_priority : uint32_t {
normal, normal,
high high = 3 // 0b11, see message_id.hpp why this is important
}; };
} // namespace caf } // namespace caf
......
...@@ -16,8 +16,8 @@ ...@@ -16,8 +16,8 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_ABSTRACT_EVENT_BASED_ACTOR_HPP #ifndef CAF_SCHEDULED_ACTOR_HPP
#define CAF_ABSTRACT_EVENT_BASED_ACTOR_HPP #define CAF_SCHEDULED_ACTOR_HPP
#include "caf/config.hpp" #include "caf/config.hpp"
...@@ -881,4 +881,4 @@ protected: ...@@ -881,4 +881,4 @@ protected:
} // namespace caf } // namespace caf
#endif // CAF_ABSTRACT_EVENT_BASED_ACTOR_HPP #endif // CAF_SCHEDULED_ACTOR_HPP
...@@ -112,6 +112,21 @@ inline logger* get_current_logger() { ...@@ -112,6 +112,21 @@ inline logger* get_current_logger() {
} // namespace <anonymous> } // namespace <anonymous>
logger::event::event(int lvl, const char* cat, const char* fun, const char* fn,
int line, std::string msg, std::thread::id t, actor_id a,
timestamp ts)
: level(lvl),
category_name(cat),
pretty_fun(fun),
file_name(fn),
line_number(line),
message(std::move(msg)),
tid(std::move(t)),
aid(a),
tstamp(ts) {
// nop
}
logger::line_builder::line_builder() : behind_arg_(false) { logger::line_builder::line_builder() : behind_arg_(false) {
// nop // nop
} }
...@@ -432,9 +447,7 @@ void logger::log_first_line() { ...@@ -432,9 +447,7 @@ void logger::log_first_line() {
msg += to_string(system_.config().logger_verbosity); msg += to_string(system_.config().logger_verbosity);
msg += ", node = "; msg += ", node = ";
msg += to_string(system_.node()); msg += to_string(system_.node());
event tmp{nullptr, event tmp{CAF_LOG_LEVEL_INFO,
nullptr,
CAF_LOG_LEVEL_INFO,
CAF_LOG_COMPONENT, CAF_LOG_COMPONENT,
CAF_PRETTY_FUN, CAF_PRETTY_FUN,
__FILE__, __FILE__,
...@@ -447,9 +460,7 @@ void logger::log_first_line() { ...@@ -447,9 +460,7 @@ void logger::log_first_line() {
} }
void logger::log_last_line() { void logger::log_last_line() {
event tmp{nullptr, event tmp{CAF_LOG_LEVEL_INFO,
nullptr,
CAF_LOG_LEVEL_INFO,
CAF_LOG_COMPONENT, CAF_LOG_COMPONENT,
CAF_PRETTY_FUN, CAF_PRETTY_FUN,
__FILE__, __FILE__,
......
...@@ -54,18 +54,13 @@ private: ...@@ -54,18 +54,13 @@ private:
} // namespace <anonymous> } // namespace <anonymous>
mailbox_element::mailbox_element() mailbox_element::mailbox_element() : marked(false) {
: next(nullptr),
prev(nullptr),
marked(false) {
// nop // nop
} }
mailbox_element::mailbox_element(strong_actor_ptr&& x, message_id y, mailbox_element::mailbox_element(strong_actor_ptr&& x, message_id y,
forwarding_stack&& z) forwarding_stack&& z)
: next(nullptr), : marked(false),
prev(nullptr),
marked(false),
sender(std::move(x)), sender(std::move(x)),
mid(y), mid(y),
stages(std::move(z)) { stages(std::move(z)) {
......
...@@ -77,6 +77,7 @@ void stream_gatherer_impl::close(message result) { ...@@ -77,6 +77,7 @@ void stream_gatherer_impl::close(message result) {
for (auto& path : paths_) for (auto& path : paths_)
stream_aborter::del(path->hdl, self_->address(), path->sid, aborter_type); stream_aborter::del(path->hdl, self_->address(), path->sid, aborter_type);
paths_.clear(); paths_.clear();
assignment_vec_.clear();
for (auto& listener : listeners_) for (auto& listener : listeners_)
listener.deliver(result); listener.deliver(result);
listeners_.clear(); listeners_.clear();
...@@ -88,6 +89,7 @@ void stream_gatherer_impl::abort(error reason) { ...@@ -88,6 +89,7 @@ void stream_gatherer_impl::abort(error reason) {
path->shutdown_reason = reason; path->shutdown_reason = reason;
} }
paths_.clear(); paths_.clear();
assignment_vec_.clear();
for (auto& listener : listeners_) for (auto& listener : listeners_)
listener.deliver(reason); listener.deliver(reason);
listeners_.clear(); listeners_.clear();
......
...@@ -111,8 +111,6 @@ CAF_TEST(rendering) { ...@@ -111,8 +111,6 @@ CAF_TEST(rendering) {
CAF_CHECK_EQUAL(render(logger::render_date, t0), t0_buf); CAF_CHECK_EQUAL(render(logger::render_date, t0), t0_buf);
// Rendering of events. // Rendering of events.
logger::event e{ logger::event e{
nullptr,
nullptr,
CAF_LOG_LEVEL_WARNING, CAF_LOG_LEVEL_WARNING,
"unit.test", "unit.test",
"void ns::foo::bar()", "void ns::foo::bar()",
......
...@@ -24,7 +24,8 @@ ...@@ -24,7 +24,8 @@
#include "caf/scheduled_actor.hpp" #include "caf/scheduled_actor.hpp"
#include "caf/prohibit_top_level_spawn_marker.hpp" #include "caf/prohibit_top_level_spawn_marker.hpp"
#include "caf/detail/intrusive_partitioned_list.hpp"
#include "caf/intrusive/partitioned_list.hpp"
#include "caf/io/fwd.hpp" #include "caf/io/fwd.hpp"
#include "caf/io/accept_handle.hpp" #include "caf/io/accept_handle.hpp"
...@@ -425,7 +426,7 @@ private: ...@@ -425,7 +426,7 @@ private:
scribe_map scribes_; scribe_map scribes_;
doorman_map doormen_; doorman_map doormen_;
datagram_servant_map datagram_servants_; datagram_servant_map datagram_servants_;
detail::intrusive_partitioned_list<mailbox_element, detail::disposer> cache_; intrusive::partitioned_list<mailbox_element, detail::disposer> cache_;
std::vector<char> dummy_wr_buf_; std::vector<char> dummy_wr_buf_;
}; };
......
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