Commit 2e0be448 authored by Dominik Charousset's avatar Dominik Charousset

Remove obsolete DRR classes

parent fedb0a17
......@@ -310,13 +310,8 @@ caf_add_component(
function_view
handles
hash.sha1
intrusive.drr_cached_queue
intrusive.drr_queue
intrusive.fifo_inbox
intrusive.lifo_inbox
intrusive.task_queue
intrusive.wdrr_dynamic_multiplexed_queue
intrusive.wdrr_fixed_multiplexed_queue
intrusive_ptr
ipv4_address
ipv4_endpoint
......@@ -345,7 +340,6 @@ caf_add_component(
node_id
optional
or_else
policy.categorized
policy.select_all
policy.select_any
request_timeout
......
......@@ -26,7 +26,6 @@
#include "caf/mixin/subscriber.hpp"
#include "caf/none.hpp"
#include "caf/policy/arg.hpp"
#include "caf/policy/categorized.hpp"
#include "caf/send.hpp"
#include "caf/typed_actor.hpp"
......
......@@ -5,11 +5,8 @@
#pragma once
#include "caf/abstract_mailbox.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/intrusive/lifo_inbox.hpp"
#include "caf/intrusive/task_queue.hpp"
#include "caf/policy/categorized.hpp"
#include "caf/policy/normal_messages.hpp"
#include "caf/policy/urgent_messages.hpp"
namespace caf::detail {
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/config.hpp"
#include "caf/intrusive/new_round_result.hpp"
#include "caf/intrusive/task_queue.hpp"
#include "caf/intrusive/task_result.hpp"
#include <limits>
#include <utility>
namespace caf::intrusive {
/// A Deficit Round Robin queue with an internal cache for allowing skipping
/// consumers.
template <class Policy>
class drr_cached_queue { // Note that we do *not* inherit from
// task_queue<Policy>, because the cached queue can no
// longer offer iterator access.
public:
// -- member types ----------------------------------------------------------
using policy_type = Policy;
using value_type = typename policy_type::mapped_type;
using node_type = typename value_type::node_type;
using node_pointer = node_type*;
using pointer = value_type*;
using unique_pointer = typename policy_type::unique_pointer;
using deficit_type = typename policy_type::deficit_type;
using task_size_type = typename policy_type::task_size_type;
using list_type = task_queue<policy_type>;
using cache_type = task_queue<policy_type>;
// -- constructors, destructors, and assignment operators -------------------
drr_cached_queue(policy_type p)
: list_(p), deficit_(0), cache_(std::move(p)) {
// nop
}
drr_cached_queue(drr_cached_queue&& other)
: list_(std::move(other.list_)),
deficit_(other.deficit_),
cache_(std::move(other.cache_)) {
// nop
}
drr_cached_queue& operator=(drr_cached_queue&& other) {
list_ = std::move(other.list_);
deficit_ = other.deficit_;
cache_ = std::move(other.cache_);
return *this;
}
// -- observers -------------------------------------------------------------
/// Returns the policy object.
policy_type& policy() noexcept {
return list_.policy();
}
/// Returns the policy object.
const policy_type& policy() const noexcept {
return list_.policy();
}
deficit_type deficit() const {
return deficit_;
}
/// Returns the accumulated size of all stored tasks in the list, i.e., tasks
/// that are not in the cache.
task_size_type total_task_size() const {
return list_.total_task_size();
}
/// Returns whether the queue has no uncached tasks.
bool empty() const noexcept {
return total_task_size() == 0;
}
/// Peeks at the first element of the list.
pointer peek() noexcept {
return list_.peek();
}
/// Applies `f` to each element in the queue, excluding cached elements.
template <class F>
void peek_all(F f) const {
list_.peek_all(f);
}
/// Tries to find the next element in the queue (excluding cached elements)
/// that matches the given predicate.
template <class Predicate>
pointer find_if(Predicate pred) {
return list_.find_if(pred);
}
// -- modifiers -------------------------------------------------------------
/// Removes all elements from the queue.
void clear() {
list_.clear();
cache_.clear();
}
void inc_deficit(deficit_type x) noexcept {
if (!list_.empty())
deficit_ += x;
}
void flush_cache() noexcept {
list_.prepend(cache_);
}
/// @private
template <class T>
void inc_total_task_size(T&& x) noexcept {
list_.inc_total_task_size(std::forward<T>(x));
}
/// @private
template <class T>
void dec_total_task_size(T&& x) noexcept {
list_.dec_total_task_size(std::forward<T>(x));
}
/// Takes the first element out of the queue if the deficit allows it and
/// returns the element.
/// @private
unique_pointer next() noexcept {
return list_.next(deficit_);
}
/// Takes the first element out of the queue (after flushing the cache) and
/// returns it, ignoring the deficit count.
unique_pointer take_front() noexcept {
flush_cache();
if (!list_.empty()) {
// Don't modify the deficit counter.
auto dummy_deficit = std::numeric_limits<deficit_type>::max();
return list_.next(dummy_deficit);
}
return nullptr;
}
/// Consumes items from the queue until the queue is empty, there is not
/// enough deficit to dequeue the next task or the consumer returns `stop`.
/// @returns `true` if `f` consumed at least one item.
template <class F>
bool consume(F& f) noexcept(noexcept(f(std::declval<value_type&>()))) {
return new_round(0, f).consumed_items > 0;
}
/// Run a new round with `quantum`, dispatching all tasks to `consumer`.
template <class F>
new_round_result new_round(deficit_type quantum, F& consumer) noexcept(
noexcept(consumer(std::declval<value_type&>()))) {
if (list_.empty())
return {0, false};
deficit_ += quantum;
auto ptr = next();
if (ptr == nullptr)
return {0, false};
size_t consumed = 0;
do {
auto consumer_res = consumer(*ptr);
switch (consumer_res) {
case task_result::skip:
// Fix deficit counter since we didn't actually use it.
deficit_ += policy().task_size(*ptr);
// Push the unconsumed item to the cache.
cache_.push_back(ptr.release());
if (list_.empty()) {
deficit_ = 0;
return {consumed, false};
}
break;
case task_result::resume:
++consumed;
flush_cache();
if (list_.empty()) {
deficit_ = 0;
return {consumed, false};
}
break;
default:
++consumed;
flush_cache();
if (list_.empty())
deficit_ = 0;
return {consumed, consumer_res == task_result::stop_all};
}
ptr = next();
} while (ptr != nullptr);
return {consumed, false};
}
cache_type& cache() noexcept {
return cache_;
}
list_type& items() noexcept {
return list_;
}
// -- insertion --------------------------------------------------------------
/// Appends `ptr` to the queue.
/// @pre `ptr != nullptr`
bool push_back(pointer ptr) noexcept {
return list_.push_back(ptr);
}
/// Appends `ptr` to the queue.
/// @pre `ptr != nullptr`
bool push_back(unique_pointer ptr) noexcept {
return push_back(ptr.release());
}
/// Creates a new element from `xs...` and appends it.
template <class... Ts>
bool emplace_back(Ts&&... xs) {
return push_back(new value_type(std::forward<Ts>(xs)...));
}
/// @private
void lifo_append(node_pointer ptr) {
list_.lifo_append(ptr);
}
/// @private
void stop_lifo_append() {
list_.stop_lifo_append();
}
private:
// -- member variables ------------------------------------------------------
/// Stores current (unskipped) items.
list_type list_;
/// Stores the deficit on this queue.
deficit_type deficit_ = 0;
/// Stores previously skipped items.
cache_type cache_;
};
} // namespace caf::intrusive
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/intrusive/new_round_result.hpp"
#include "caf/intrusive/task_queue.hpp"
#include "caf/intrusive/task_result.hpp"
#include <utility>
namespace caf::intrusive {
/// A Deficit Round Robin queue.
template <class Policy>
class drr_queue : public task_queue<Policy> {
public:
// -- member types -----------------------------------------------------------
using super = task_queue<Policy>;
using typename super::policy_type;
using typename super::unique_pointer;
using typename super::value_type;
using deficit_type = typename policy_type::deficit_type;
// -- constructors, destructors, and assignment operators --------------------
drr_queue(policy_type p) : super(std::move(p)), deficit_(0) {
// nop
}
drr_queue(drr_queue&& other) : super(std::move(other)), deficit_(0) {
// nop
}
drr_queue& operator=(drr_queue&& other) {
super::operator=(std::move(other));
return *this;
}
// -- observers --------------------------------------------------------------
deficit_type deficit() const {
return deficit_;
}
// -- modifiers --------------------------------------------------------------
void inc_deficit(deficit_type x) noexcept {
if (!super::empty())
deficit_ += x;
}
void flush_cache() const noexcept {
// nop
}
/// Consumes items from the queue until the queue is empty or there is not
/// enough deficit to dequeue the next task.
/// @returns `true` if `f` consumed at least one item.
template <class F>
bool consume(F& f) noexcept(noexcept(f(std::declval<value_type&>()))) {
auto res = new_round(0, f);
return res.consumed_items;
}
/// Takes the first element out of the queue if the deficit allows it and
/// returns the element.
/// @private
unique_pointer next() noexcept {
return super::next(deficit_);
}
/// Run a new round with `quantum`, dispatching all tasks to `consumer`.
/// @returns `true` if at least one item was consumed, `false` otherwise.
template <class F>
new_round_result new_round(deficit_type quantum, F& consumer) {
size_t consumed = 0;
if (!super::empty()) {
deficit_ += quantum;
auto ptr = next();
if (ptr == nullptr)
return {0, false};
do {
++consumed;
switch (consumer(*ptr)) {
default:
break;
case task_result::stop:
return {consumed, false};
case task_result::stop_all:
return {consumed, true};
}
ptr = next();
} while (ptr != nullptr);
return {consumed, false};
}
return {consumed, false};
}
private:
// -- member variables -------------------------------------------------------
/// Stores the deficit on this queue.
deficit_type deficit_ = 0;
};
} // namespace caf::intrusive
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/config.hpp"
#include "caf/intrusive/inbox_result.hpp"
#include "caf/intrusive/lifo_inbox.hpp"
#include "caf/intrusive/new_round_result.hpp"
#include <atomic>
#include <condition_variable> // std::cv_status
#include <deque>
#include <limits>
#include <list>
#include <memory>
#include <mutex>
namespace caf::intrusive {
/// A FIFO inbox that combines an efficient thread-safe LIFO inbox with a FIFO
/// queue for re-ordering incoming messages.
template <class Policy>
class fifo_inbox {
public:
// -- member types -----------------------------------------------------------
using policy_type = Policy;
using queue_type = typename policy_type::queue_type;
using deficit_type = typename policy_type::deficit_type;
using value_type = typename policy_type::mapped_type;
using lifo_inbox_type = lifo_inbox<policy_type>;
using pointer = value_type*;
using unique_pointer = typename queue_type::unique_pointer;
using node_pointer = typename value_type::node_pointer;
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
fifo_inbox(Ts&&... xs) : queue_(std::forward<Ts>(xs)...) {
// nop
}
// -- queue and stack status functions ---------------------------------------
/// Returns an approximation of the current size.
size_t size() noexcept {
fetch_more();
return queue_.total_task_size();
}
/// Queries whether the inbox is empty.
bool empty() const noexcept {
return queue_.empty() && inbox_.empty();
}
/// Queries whether this inbox is closed.
bool closed() const noexcept {
return inbox_.closed();
}
/// Queries whether this has been marked as blocked, i.e.,
/// the owner of the list is waiting for new data.
bool blocked() const noexcept {
return inbox_.blocked();
}
/// Appends `ptr` to the inbox.
inbox_result push_back(pointer ptr) noexcept {
return inbox_.push_front(ptr);
}
/// Appends `ptr` to the inbox.
inbox_result push_back(unique_pointer ptr) noexcept {
return push_back(ptr.release());
}
template <class... Ts>
inbox_result emplace_back(Ts&&... xs) {
return push_back(new value_type(std::forward<Ts>(xs)...));
}
// -- backwards compatibility ------------------------------------------------
/// @cond PRIVATE
inbox_result enqueue(pointer ptr) noexcept {
return static_cast<inbox_result>(inbox_.push_front(ptr));
}
size_t count() noexcept {
return size();
}
size_t count(size_t) noexcept {
return size();
}
/// @endcond
// -- queue management -------------------------------------------------------
void flush_cache() noexcept {
queue_.flush_cache();
}
/// Tries to get more items from the inbox.
bool fetch_more() {
node_pointer head = inbox_.take_head();
if (head == nullptr)
return false;
do {
auto next = head->next;
queue_.lifo_append(lifo_inbox_type::promote(head));
head = next;
} while (head != nullptr);
queue_.stop_lifo_append();
return true;
}
/// Tries to set this queue from `empty` to `blocked`.
bool try_block() {
return queue_.empty() && inbox_.try_block();
}
/// Tries to set this queue from `blocked` to `empty`.
bool try_unblock() {
return inbox_.try_unblock();
}
/// Closes this inbox and moves all elements to the queue.
/// @warning Call only from the reader (owner).
void close() {
auto f = [&](pointer x) { queue_.lifo_append(x); };
inbox_.close(f);
queue_.stop_lifo_append();
}
/// Run a new round with `quantum`, dispatching all tasks to `consumer`.
template <class F>
new_round_result new_round(deficit_type quantum, F& consumer) {
fetch_more();
return queue_.new_round(quantum, consumer);
}
pointer peek() noexcept {
fetch_more();
return queue_.peek();
}
/// Tries to find an element in the queue that matches the given predicate.
template <class Predicate>
pointer find_if(Predicate pred) {
fetch_more();
return queue_.find_if(pred);
}
queue_type& queue() noexcept {
return queue_;
}
// -- synchronized access ----------------------------------------------------
template <class Mutex, class CondVar>
bool synchronized_push_back(Mutex& mtx, CondVar& cv, pointer ptr) {
return inbox_.synchronized_push_front(mtx, cv, ptr);
}
template <class Mutex, class CondVar>
bool synchronized_push_back(Mutex& mtx, CondVar& cv, unique_pointer ptr) {
return synchronized_push_back(mtx, cv, ptr.release());
}
template <class Mutex, class CondVar, class... Ts>
bool synchronized_emplace_back(Mutex& mtx, CondVar& cv, Ts&&... xs) {
return synchronized_push_back(mtx, cv,
new value_type(std::forward<Ts>(xs)...));
}
template <class Mutex, class CondVar>
void synchronized_await(Mutex& mtx, CondVar& cv) {
if (queue_.empty()) {
inbox_.synchronized_await(mtx, cv);
fetch_more();
}
}
template <class Mutex, class CondVar, class TimePoint>
bool synchronized_await(Mutex& mtx, CondVar& cv, const TimePoint& timeout) {
if (!queue_.empty())
return true;
if (inbox_.synchronized_await(mtx, cv, timeout)) {
fetch_more();
return true;
}
return false;
}
private:
// -- member variables -------------------------------------------------------
/// Thread-safe LIFO inbox.
lifo_inbox_type inbox_;
/// User-facing queue that is constantly resupplied from the inbox.
queue_type queue_;
};
} // namespace caf::intrusive
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/fwd.hpp"
#include <type_traits>
namespace caf::intrusive {
/// Returns the state of a consumer from `new_round`.
struct new_round_result {
/// Denotes whether the consumer accepted at least one element.
size_t consumed_items;
/// Denotes whether the consumer returned `task_result::stop_all`.
bool stop_all;
};
constexpr bool operator==(new_round_result x, new_round_result y) {
return x.consumed_items == y.consumed_items && x.stop_all == y.stop_all;
}
constexpr bool operator!=(new_round_result x, new_round_result y) {
return !(x == y);
}
template <class Inspector>
bool inspect(Inspector& f, new_round_result& x) {
return f.object(x).fields(f.field("consumed_items", x.consumed_items),
f.field("stop_all", x.stop_all));
}
} // namespace caf::intrusive
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/type_traits.hpp"
#include "caf/intrusive/new_round_result.hpp"
#include "caf/intrusive/task_result.hpp"
#include <utility>
namespace caf::intrusive {
/// A work queue that internally multiplexes any number of DRR queues.
template <class Policy>
class wdrr_dynamic_multiplexed_queue {
public:
using policy_type = Policy;
using deficit_type = typename policy_type::deficit_type;
using mapped_type = typename policy_type::mapped_type;
using key_type = typename policy_type::key_type;
using pointer = mapped_type*;
using unique_pointer = typename policy_type::unique_pointer;
using task_size_type = typename policy_type::task_size_type;
using queue_map_type = typename policy_type::queue_map_type;
using queue_type = typename queue_map_type::mapped_type;
template <class... Ps>
wdrr_dynamic_multiplexed_queue(policy_type p) : policy_(p) {
// nop
}
~wdrr_dynamic_multiplexed_queue() noexcept {
for (auto& kvp : qs_)
policy_.cleanup(kvp.second);
}
policy_type& policy() noexcept {
return policy_;
}
const policy_type& policy() const noexcept {
return policy_;
}
bool push_back(mapped_type* ptr) noexcept {
if (auto i = qs_.find(policy_.id_of(*ptr)); i != qs_.end()) {
return policy_.push_back(i->second, ptr);
} else {
typename unique_pointer::deleter_type d;
d(ptr);
return false;
}
}
bool push_back(unique_pointer ptr) noexcept {
return push_back(ptr.release());
}
template <class... Ts>
bool emplace_back(Ts&&... xs) {
return push_back(new mapped_type(std::forward<Ts>(xs)...));
}
/// @private
template <class F>
struct new_round_helper {
const key_type& k;
queue_type& q;
F& f;
template <class... Ts>
auto operator()(Ts&&... xs)
-> decltype((std::declval<F&>())(std::declval<const key_type&>(),
std::declval<queue_type&>(),
std::forward<Ts>(xs)...)) {
return f(k, q, std::forward<Ts>(xs)...);
}
};
void inc_deficit(deficit_type x) noexcept {
for (auto& kvp : qs_) {
auto& q = kvp.second;
q.inc_deficit(policy_.quantum(q, x));
}
}
/// Run a new round with `quantum`, dispatching all tasks to `consumer`.
/// @returns `true` if at least one item was consumed, `false` otherwise.
template <class F>
new_round_result new_round(deficit_type quantum, F& f) {
size_t consumed = 0;
bool stopped = false;
for (auto& kvp : qs_) {
if (policy_.enabled(kvp.second)) {
auto& q = kvp.second;
if (!stopped) {
new_round_helper<F> g{kvp.first, q, f};
auto res = q.new_round(policy_.quantum(q, quantum), g);
consumed += res.consumed_items;
if (res.stop_all)
stopped = true;
} else {
// Always increase deficit, even if a previous queue stopped the
// consumer preemptively.
q.inc_deficit(policy_.quantum(q, quantum));
}
}
}
cleanup();
return {consumed, stopped};
}
/// Erases all keys previously marked via `erase_later`.
void cleanup() {
if (!erase_list_.empty()) {
for (auto& k : erase_list_) {
if (auto i = qs_.find(k); i != qs_.end()) {
policy_.cleanup(i->second);
qs_.erase(i);
}
}
erase_list_.clear();
}
}
/// Marks the key `k` for erasing from the map later.
void erase_later(key_type k) {
erase_list_.emplace_back(std::move(k));
}
pointer peek() noexcept {
for (auto& kvp : qs_) {
auto ptr = kvp.second.peek();
if (ptr != nullptr)
return ptr;
}
return nullptr;
}
/// Applies `f` to each element in the queue.
template <class F>
void peek_all(F f) const {
for (auto& kvp : qs_)
kvp.second.peek_all(f);
}
/// Tries to find an element in the queue that matches the given predicate.
template <class Predicate>
pointer find_if(Predicate pred) {
for (auto& kvp : qs_)
if (auto ptr = kvp.second.find_if(pred))
return ptr;
return nullptr;
}
/// Returns `true` if all queues are empty, `false` otherwise.
bool empty() const noexcept {
return total_task_size() == 0;
}
void flush_cache() noexcept {
for (auto& kvp : qs_)
kvp.second.flush_cache();
}
task_size_type total_task_size() const noexcept {
task_size_type result = 0;
for (auto& kvp : qs_)
if (policy_.enabled(kvp.second))
result += kvp.second.total_task_size();
return result;
}
/// Returns the tuple containing all nested queues.
queue_map_type& queues() noexcept {
return qs_;
}
/// Returns the tuple containing all nested queues.
const queue_map_type& queues() const noexcept {
return qs_;
}
void lifo_append(pointer ptr) noexcept {
if (auto i = qs_.find(policy_.id_of(*ptr)); i != qs_.end()) {
policy_.lifo_append(i->second, ptr);
} else {
typename unique_pointer::deleter_type d;
d(ptr);
}
}
void stop_lifo_append() noexcept {
for (auto& kvp : qs_)
policy_.stop_lifo_append(kvp.second);
}
private:
// -- member variables -------------------------------------------------------
/// All queues.
queue_map_type qs_;
/// Policy object for adjusting a quantum based on queue weight etc.
Policy policy_;
/// List of keys that are marked erased.
std::vector<key_type> erase_list_;
};
} // namespace caf::intrusive
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/type_traits.hpp"
#include "caf/intrusive/new_round_result.hpp"
#include "caf/intrusive/task_result.hpp"
#include <tuple>
#include <type_traits>
#include <utility>
namespace caf::intrusive {
/// A work queue that internally multiplexes any number of DRR queues.
template <class Policy, class Q, class... Qs>
class wdrr_fixed_multiplexed_queue {
public:
using tuple_type = std::tuple<Q, Qs...>;
using policy_type = Policy;
using deficit_type = typename policy_type::deficit_type;
using value_type = typename policy_type::mapped_type;
using pointer = value_type*;
using unique_pointer = typename policy_type::unique_pointer;
using task_size_type = typename policy_type::task_size_type;
template <size_t I>
using index = std::integral_constant<size_t, I>;
static constexpr size_t num_queues = sizeof...(Qs) + 1;
wdrr_fixed_multiplexed_queue(policy_type p0, typename Q::policy_type p1,
typename Qs::policy_type... ps)
: qs_(std::move(p1), std::move(ps)...), policy_(std::move(p0)) {
// nop
}
policy_type& policy() noexcept {
return policy_;
}
const policy_type& policy() const noexcept {
return policy_;
}
bool push_back(value_type* ptr) noexcept {
return push_back_recursion<0>(policy_.id_of(*ptr), ptr);
}
bool push_back(unique_pointer ptr) noexcept {
return push_back(ptr.release());
}
template <class... Ts>
bool emplace_back(Ts&&... xs) {
return push_back(new value_type(std::forward<Ts>(xs)...));
}
void inc_deficit(deficit_type x) noexcept {
inc_deficit_recursion<0>(x);
}
/// Run a new round with `quantum`, dispatching all tasks to `consumer`.
/// @returns `true` if at least one item was consumed, `false` otherwise.
template <class F>
new_round_result new_round(deficit_type quantum, F& f) {
return new_round_recursion<0>(quantum, f);
}
pointer peek() noexcept {
return peek_recursion<0>();
}
/// Tries to find an element in the queue that matches the given predicate.
template <class Predicate>
pointer find_if(Predicate pred) {
return find_if_recursion<0>(pred);
}
/// Applies `f` to each element in the queue.
template <class F>
void peek_all(F f) const {
return peek_all_recursion<0>(f);
}
/// Returns `true` if all queues are empty, `false` otherwise.
bool empty() const noexcept {
return total_task_size() == 0;
}
void flush_cache() noexcept {
flush_cache_recursion<0>();
}
task_size_type total_task_size() const noexcept {
return total_task_size_recursion<0>();
}
/// Returns the tuple containing all nested queues.
tuple_type& queues() noexcept {
return qs_;
}
/// Returns the tuple containing all nested queues.
const tuple_type& queues() const noexcept {
return qs_;
}
void lifo_append(pointer ptr) noexcept {
lifo_append_recursion<0>(policy_.id_of(*ptr), ptr);
}
void stop_lifo_append() noexcept {
stop_lifo_append_recursion<0>();
}
private:
// -- recursive helper functions ---------------------------------------------
// TODO: a lot of this code could be vastly simplified by using C++14 generic
// lambdas and simple utility to dispatch on the tuple index. Consider
// to revisite this code once we switch to C++14.
template <size_t I>
detail::enable_if_t<I == num_queues, bool>
push_back_recursion(size_t, value_type*) noexcept {
return false;
}
template <size_t I>
detail::enable_if_t<I != num_queues, bool>
push_back_recursion(size_t pos, value_type* ptr) noexcept {
if (pos == I) {
auto& q = std::get<I>(qs_);
return q.push_back(ptr);
}
return push_back_recursion<I + 1>(pos, ptr);
}
template <size_t I, class Queue, class F>
struct new_round_recursion_helper {
Queue& q;
F& f;
template <class... Ts>
auto operator()(Ts&&... xs)
-> decltype((std::declval<F&>())(std::declval<index<I>>(),
std::declval<Queue&>(),
std::forward<Ts>(xs)...)) {
index<I> id;
return f(id, q, std::forward<Ts>(xs)...);
}
};
template <size_t I>
detail::enable_if_t<I == num_queues>
inc_deficit_recursion(deficit_type) noexcept {
// end of recursion
}
template <size_t I>
detail::enable_if_t<I != num_queues>
inc_deficit_recursion(deficit_type quantum) noexcept {
auto& q = std::get<I>(qs_);
q.inc_deficit(policy_.quantum(q, quantum));
inc_deficit_recursion<I + 1>(quantum);
}
template <size_t I, class F>
detail::enable_if_t<I == num_queues, new_round_result>
new_round_recursion(deficit_type, F&) noexcept {
return {0, false};
}
template <size_t I, class F>
detail::enable_if_t<I != num_queues, new_round_result>
new_round_recursion(deficit_type quantum, F& f) {
auto& q = std::get<I>(qs_);
using q_type = typename std::decay<decltype(q)>::type;
new_round_recursion_helper<I, q_type, F> g{q, f};
auto res = q.new_round(policy_.quantum(q, quantum), g);
if (res.stop_all) {
// Always increase deficit, even if a previous queue stopped the
// consumer preemptively.
inc_deficit_recursion<I + 1>(quantum);
return res;
}
auto sub = new_round_recursion<I + 1>(quantum, f);
return {res.consumed_items + sub.consumed_items, sub.stop_all};
}
template <size_t I>
detail::enable_if_t<I == num_queues, pointer> peek_recursion() noexcept {
return nullptr;
}
template <size_t I>
detail::enable_if_t<I != num_queues, pointer> peek_recursion() noexcept {
auto ptr = std::get<I>(qs_).peek();
if (ptr != nullptr)
return ptr;
return peek_recursion<I + 1>();
}
template <size_t I, class Predicate>
detail::enable_if_t<I == num_queues, pointer> find_if_recursion(Predicate) {
return nullptr;
}
template <size_t I, class Predicate>
detail::enable_if_t<I != num_queues, pointer>
find_if_recursion(Predicate pred) {
if (auto ptr = std::get<I>(qs_).find_if(pred))
return ptr;
else
return find_if_recursion<I + 1>(std::move(pred));
}
template <size_t I, class F>
detail::enable_if_t<I == num_queues> peek_all_recursion(F&) const {
// nop
}
template <size_t I, class F>
detail::enable_if_t<I != num_queues> peek_all_recursion(F& f) const {
std::get<I>(qs_).peek_all(f);
peek_all_recursion<I + 1>(f);
}
template <size_t I>
detail::enable_if_t<I == num_queues> flush_cache_recursion() noexcept {
// nop
}
template <size_t I>
detail::enable_if_t<I != num_queues> flush_cache_recursion() noexcept {
std::get<I>(qs_).flush_cache();
flush_cache_recursion<I + 1>();
}
template <size_t I>
detail::enable_if_t<I == num_queues, task_size_type>
total_task_size_recursion() const noexcept {
return 0;
}
template <size_t I>
detail::enable_if_t<I != num_queues, task_size_type>
total_task_size_recursion() const noexcept {
return std::get<I>(qs_).total_task_size()
+ total_task_size_recursion<I + 1>();
}
template <size_t I>
detail::enable_if_t<I == num_queues>
lifo_append_recursion(size_t, pointer) noexcept {
// nop
}
template <size_t I>
detail::enable_if_t<I != num_queues>
lifo_append_recursion(size_t i, pointer ptr) noexcept {
if (i == I)
std::get<I>(qs_).lifo_append(ptr);
else
lifo_append_recursion<I + 1>(i, ptr);
}
template <size_t I>
detail::enable_if_t<I == num_queues> stop_lifo_append_recursion() noexcept {
// nop
}
template <size_t I>
detail::enable_if_t<I != num_queues> stop_lifo_append_recursion() noexcept {
std::get<I>(qs_).stop_lifo_append();
stop_lifo_append_recursion<I + 1>();
}
// -- member variables -------------------------------------------------------
/// All queues.
tuple_type qs_;
/// Policy object for adjusting a quantum based on queue weight etc.
Policy policy_;
};
} // namespace caf::intrusive
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/core_export.hpp"
#include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/message_priority.hpp"
#include "caf/policy/normal_messages.hpp"
#include "caf/policy/urgent_messages.hpp"
#include "caf/unit.hpp"
namespace caf::policy {
/// Configures a cached WDRR fixed multiplexed queue for dispatching to four
/// nested queue (one for each message category type).
class CAF_CORE_EXPORT categorized {
public:
// -- member types -----------------------------------------------------------
using mapped_type = mailbox_element;
using task_size_type = size_t;
using deficit_type = size_t;
using unique_pointer = mailbox_element_ptr;
// -- constructors, destructors, and assignment operators --------------------
categorized() = default;
categorized(const categorized&) = default;
categorized& operator=(const categorized&) = default;
constexpr categorized(unit_t) {
// nop
}
// -- interface required by wdrr_fixed_multiplexed_queue ---------------------
template <template <class> class Queue>
static deficit_type
quantum(const Queue<urgent_messages>&, deficit_type x) noexcept {
// Allow actors to consume twice as many urgent as normal messages per
// credit round.
return x + x;
}
template <class Queue>
static deficit_type quantum(const Queue&, deficit_type x) noexcept {
return x;
}
static size_t id_of(const mailbox_element& x) noexcept {
return static_cast<size_t>(x.mid.category());
}
};
} // namespace caf::policy
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/core_export.hpp"
#include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/unit.hpp"
namespace caf::policy {
/// Configures a cached DRR queue for holding asynchronous messages with
/// default priority.
class CAF_CORE_EXPORT normal_messages {
public:
// -- member types -----------------------------------------------------------
using mapped_type = mailbox_element;
using task_size_type = size_t;
using deficit_type = size_t;
using unique_pointer = mailbox_element_ptr;
// -- constructors, destructors, and assignment operators --------------------
normal_messages() = default;
normal_messages(const normal_messages&) = default;
normal_messages& operator=(const normal_messages&) = default;
constexpr normal_messages(unit_t) {
// nop
}
// -- interface required by drr_queue ----------------------------------------
static task_size_type task_size(const mailbox_element&) noexcept {
return 1;
}
};
} // namespace caf::policy
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/core_export.hpp"
#include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/unit.hpp"
namespace caf::policy {
/// Configures a cached DRR queue for holding asynchronous messages with
/// default priority.
class CAF_CORE_EXPORT urgent_messages {
public:
// -- member types -----------------------------------------------------------
using mapped_type = mailbox_element;
using task_size_type = size_t;
using deficit_type = size_t;
using unique_pointer = mailbox_element_ptr;
// -- constructors, destructors, and assignment operators --------------------
urgent_messages() = default;
urgent_messages(const urgent_messages&) = default;
urgent_messages& operator=(const urgent_messages&) = default;
constexpr urgent_messages(unit_t) {
// nop
}
// -- interface required by drr_queue ----------------------------------------
static task_size_type task_size(const mailbox_element&) noexcept {
return 1;
}
};
} // namespace caf::policy
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE intrusive.drr_cached_queue
#include "caf/intrusive/drr_cached_queue.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "core-test.hpp"
#include <memory>
using namespace caf;
using namespace caf::intrusive;
namespace {
struct inode : singly_linked<inode> {
int value;
inode(int x = 0) : value(x) {
// nop
}
};
std::string to_string(const inode& x) {
return std::to_string(x.value);
}
struct inode_policy {
using mapped_type = inode;
using task_size_type = int;
using deficit_type = int;
using deleter_type = std::default_delete<mapped_type>;
using unique_pointer = std::unique_ptr<mapped_type, deleter_type>;
static inline task_size_type task_size(const mapped_type&) noexcept {
return 1;
}
};
using queue_type = drr_cached_queue<inode_policy>;
struct fixture {
inode_policy policy;
queue_type queue{policy};
template <class Queue>
void fill(Queue&) {
// nop
}
template <class Queue, class T, class... Ts>
void fill(Queue& q, T x, Ts... xs) {
q.emplace_back(x);
fill(q, xs...);
}
};
auto make_new_round_result(size_t consumed_items, bool stop_all) {
return new_round_result{consumed_items, stop_all};
}
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
CAF_TEST(default_constructed) {
CAF_REQUIRE_EQUAL(queue.empty(), true);
CAF_REQUIRE_EQUAL(queue.deficit(), 0);
CAF_REQUIRE_EQUAL(queue.total_task_size(), 0);
CAF_REQUIRE_EQUAL(queue.peek(), nullptr);
}
CAF_TEST(new_round) {
// Define a function object for consuming even numbers.
std::string fseq;
auto f = [&](inode& x) -> task_result {
if ((x.value & 0x01) == 1)
return task_result::skip;
fseq += to_string(x);
return task_result::resume;
};
// Define a function object for consuming odd numbers.
std::string gseq;
auto g = [&](inode& x) -> task_result {
if ((x.value & 0x01) == 0)
return task_result::skip;
gseq += to_string(x);
return task_result::resume;
};
fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9);
// Allow f to consume 2, 4, and 6.
auto round_result = queue.new_round(3, f);
CHECK_EQ(round_result, make_new_round_result(3, false));
CHECK_EQ(fseq, "246");
CHECK_EQ(queue.deficit(), 0);
// Allow g to consume 1, 3, 5, and 7.
round_result = queue.new_round(4, g);
CHECK_EQ(round_result, make_new_round_result(4, false));
CHECK_EQ(gseq, "1357");
CHECK_EQ(queue.deficit(), 0);
}
CAF_TEST(skipping) {
// Define a function object for consuming even numbers.
std::string seq;
auto f = [&](inode& x) -> task_result {
if ((x.value & 0x01) == 1)
return task_result::skip;
seq += to_string(x);
return task_result::resume;
};
MESSAGE("make a round on an empty queue");
CHECK_EQ(queue.new_round(10, f), make_new_round_result(0, false));
MESSAGE("make a round on a queue with only odd numbers (skip all)");
fill(queue, 1, 3, 5);
CHECK_EQ(queue.new_round(10, f), make_new_round_result(0, false));
MESSAGE("make a round on a queue with an even number at the front");
fill(queue, 2);
CHECK_EQ(queue.new_round(10, f), make_new_round_result(1, false));
CHECK_EQ(seq, "2");
MESSAGE("make a round on a queue with an even number in between");
fill(queue, 7, 9, 4, 11, 13);
CHECK_EQ(queue.new_round(10, f), make_new_round_result(1, false));
CHECK_EQ(seq, "24");
MESSAGE("make a round on a queue with an even number at the back");
fill(queue, 15, 17, 6);
CHECK_EQ(queue.new_round(10, f), make_new_round_result(1, false));
CHECK_EQ(seq, "246");
}
CAF_TEST(take_front) {
std::string seq;
fill(queue, 1, 2, 3, 4, 5, 6);
auto f = [&](inode& x) {
seq += to_string(x);
return task_result::resume;
};
CHECK_EQ(queue.deficit(), 0);
while (!queue.empty()) {
auto ptr = queue.take_front();
f(*ptr);
}
CHECK_EQ(seq, "123456");
fill(queue, 5, 4, 3, 2, 1);
while (!queue.empty()) {
auto ptr = queue.take_front();
f(*ptr);
}
CHECK_EQ(seq, "12345654321");
CHECK_EQ(queue.deficit(), 0);
}
CAF_TEST(alternating_consumer) {
using fun_type = std::function<task_result(inode&)>;
fun_type f;
fun_type g;
fun_type* selected = &f;
// Define a function object for consuming even numbers.
std::string seq;
f = [&](inode& x) -> task_result {
if ((x.value & 0x01) == 1)
return task_result::skip;
seq += to_string(x);
selected = &g;
return task_result::resume;
};
// Define a function object for consuming odd numbers.
g = [&](inode& x) -> task_result {
if ((x.value & 0x01) == 0)
return task_result::skip;
seq += to_string(x);
selected = &f;
return task_result::resume;
};
/// Define a function object that alternates between f and g.
auto h = [&](inode& x) { return (*selected)(x); };
// Fill and consume queue, h leaves 9 in the cache since it reads (odd, even)
// sequences and no odd value to read after 7 is available.
fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9);
auto round_result = queue.new_round(1000, h);
CHECK_EQ(round_result, make_new_round_result(8, false));
CHECK_EQ(seq, "21436587");
CHECK_EQ(queue.deficit(), 0);
CHECK_EQ(deep_to_string(queue.cache()), "[9]");
}
CAF_TEST(peek_all) {
auto queue_to_string = [&] {
std::string str;
auto peek_fun = [&](const inode& x) {
if (!str.empty())
str += ", ";
str += std::to_string(x.value);
};
queue.peek_all(peek_fun);
return str;
};
CHECK_EQ(queue_to_string(), "");
queue.emplace_back(2);
CHECK_EQ(queue_to_string(), "2");
queue.cache().emplace_back(1);
CHECK_EQ(queue_to_string(), "2");
queue.emplace_back(3);
CHECK_EQ(queue_to_string(), "2, 3");
queue.flush_cache();
CHECK_EQ(queue_to_string(), "1, 2, 3");
}
CAF_TEST(to_string) {
CHECK_EQ(deep_to_string(queue.items()), "[]");
fill(queue, 3, 4);
CHECK_EQ(deep_to_string(queue.items()), "[3, 4]");
fill(queue.cache(), 1, 2);
CHECK_EQ(deep_to_string(queue.items()), "[3, 4]");
queue.flush_cache();
CHECK_EQ(deep_to_string(queue.items()), "[1, 2, 3, 4]");
}
END_FIXTURE_SCOPE()
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE intrusive.drr_queue
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "core-test.hpp"
#include <memory>
using namespace caf;
using namespace caf::intrusive;
namespace {
struct inode : singly_linked<inode> {
int value;
inode(int x = 0) : value(x) {
// nop
}
};
std::string to_string(const inode& x) {
return std::to_string(x.value);
}
struct inode_policy {
using mapped_type = inode;
using task_size_type = int;
using deficit_type = int;
using deleter_type = std::default_delete<mapped_type>;
using unique_pointer = std::unique_ptr<mapped_type, deleter_type>;
static inline task_size_type task_size(const mapped_type& x) {
return x.value;
}
};
using queue_type = drr_queue<inode_policy>;
struct fixture {
inode_policy policy;
queue_type queue{policy};
void fill(queue_type&) {
// nop
}
template <class T, class... Ts>
void fill(queue_type& q, T x, Ts... xs) {
q.emplace_back(x);
fill(q, xs...);
}
};
auto make_new_round_result(size_t consumed_items, bool stop_all) {
return new_round_result{consumed_items, stop_all};
}
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
CAF_TEST(default_constructed) {
CAF_REQUIRE_EQUAL(queue.empty(), true);
CAF_REQUIRE_EQUAL(queue.deficit(), 0);
CAF_REQUIRE_EQUAL(queue.total_task_size(), 0);
CAF_REQUIRE_EQUAL(queue.peek(), nullptr);
CAF_REQUIRE_EQUAL(queue.next(), nullptr);
CAF_REQUIRE_EQUAL(queue.begin(), queue.end());
}
CAF_TEST(inc_deficit) {
// Increasing the deficit does nothing as long as the queue is empty.
queue.inc_deficit(100);
CAF_REQUIRE_EQUAL(queue.deficit(), 0);
// Increasing the deficit must work on non-empty queues.
fill(queue, 1);
queue.inc_deficit(100);
CAF_REQUIRE_EQUAL(queue.deficit(), 100);
// Deficit must drop back down to 0 once the queue becomes empty.
queue.next();
CAF_REQUIRE_EQUAL(queue.deficit(), 0);
}
CAF_TEST(new_round) {
std::string seq;
fill(queue, 1, 2, 3, 4, 5, 6);
auto f = [&](inode& x) {
seq += to_string(x);
return task_result::resume;
};
// Allow f to consume 1, 2, and 3 with a leftover deficit of 1.
auto round_result = queue.new_round(7, f);
CHECK_EQ(round_result, make_new_round_result(3, false));
CHECK_EQ(seq, "123");
CHECK_EQ(queue.deficit(), 1);
// Allow f to consume 4 and 5 with a leftover deficit of 0.
round_result = queue.new_round(8, f);
CHECK_EQ(round_result, make_new_round_result(2, false));
CHECK_EQ(seq, "12345");
CHECK_EQ(queue.deficit(), 0);
// Allow f to consume 6 with a leftover deficit of 0 (queue is empty).
round_result = queue.new_round(1000, f);
CHECK_EQ(round_result, make_new_round_result(1, false));
CHECK_EQ(seq, "123456");
CHECK_EQ(queue.deficit(), 0);
// new_round on an empty queue does nothing.
round_result = queue.new_round(1000, f);
CHECK_EQ(round_result, make_new_round_result(0, false));
CHECK_EQ(seq, "123456");
CHECK_EQ(queue.deficit(), 0);
}
CAF_TEST(next) {
std::string seq;
fill(queue, 1, 2, 3, 4, 5, 6);
auto f = [&](inode& x) {
seq += to_string(x);
return task_result::resume;
};
auto take = [&] {
queue.flush_cache();
queue.inc_deficit(queue.peek()->value);
return queue.next();
};
while (!queue.empty()) {
auto ptr = take();
f(*ptr);
}
CHECK_EQ(seq, "123456");
fill(queue, 5, 4, 3, 2, 1);
while (!queue.empty()) {
auto ptr = take();
f(*ptr);
}
CHECK_EQ(seq, "12345654321");
CHECK_EQ(queue.deficit(), 0);
}
CAF_TEST(peek_all) {
auto queue_to_string = [&] {
std::string str;
auto peek_fun = [&](const inode& x) {
if (!str.empty())
str += ", ";
str += std::to_string(x.value);
};
queue.peek_all(peek_fun);
return str;
};
CHECK_EQ(queue_to_string(), "");
queue.emplace_back(1);
CHECK_EQ(queue_to_string(), "1");
queue.emplace_back(2);
CHECK_EQ(queue_to_string(), "1, 2");
queue.emplace_back(3);
CHECK_EQ(queue_to_string(), "1, 2, 3");
queue.emplace_back(4);
CHECK_EQ(queue_to_string(), "1, 2, 3, 4");
}
CAF_TEST(to_string) {
CHECK_EQ(deep_to_string(queue), "[]");
fill(queue, 1, 2, 3, 4);
CHECK_EQ(deep_to_string(queue), "[1, 2, 3, 4]");
}
END_FIXTURE_SCOPE()
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE intrusive.fifo_inbox
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "core-test.hpp"
#include <memory>
using namespace caf;
using namespace caf::intrusive;
namespace {
struct inode : singly_linked<inode> {
int value;
inode(int x = 0) : value(x) {
// nop
}
};
std::string to_string(const inode& x) {
return std::to_string(x.value);
}
struct inode_policy {
using mapped_type = inode;
using task_size_type = int;
using deficit_type = int;
using deleter_type = std::default_delete<mapped_type>;
using unique_pointer = std::unique_ptr<mapped_type, deleter_type>;
using queue_type = drr_queue<inode_policy>;
static constexpr task_size_type task_size(const inode&) noexcept {
return 1;
}
};
using inbox_type = fifo_inbox<inode_policy>;
struct fixture {
inode_policy policy;
inbox_type inbox{policy};
void fill(inbox_type&) {
// nop
}
template <class T, class... Ts>
void fill(inbox_type& i, T x, Ts... xs) {
i.emplace_back(x);
fill(i, xs...);
}
std::string fetch() {
std::string result;
auto f = [&](inode& x) {
result += to_string(x);
return task_result::resume;
};
inbox.new_round(1000, f);
return result;
}
std::string close_and_fetch() {
std::string result;
auto f = [&](inode& x) {
result += to_string(x);
return task_result::resume;
};
inbox.close();
inbox.queue().new_round(1000, f);
return result;
}
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
CAF_TEST(default_constructed) {
CAF_REQUIRE_EQUAL(inbox.empty(), true);
}
CAF_TEST(push_front) {
fill(inbox, 1, 2, 3);
CAF_REQUIRE_EQUAL(close_and_fetch(), "123");
CAF_REQUIRE_EQUAL(inbox.closed(), true);
}
CAF_TEST(push_after_close) {
inbox.close();
auto res = inbox.push_back(new inode(0));
CAF_REQUIRE_EQUAL(res, inbox_result::queue_closed);
}
CAF_TEST(unblock) {
CAF_REQUIRE_EQUAL(inbox.try_block(), true);
auto res = inbox.push_back(new inode(0));
CAF_REQUIRE_EQUAL(res, inbox_result::unblocked_reader);
res = inbox.push_back(new inode(1));
CAF_REQUIRE_EQUAL(res, inbox_result::success);
CAF_REQUIRE_EQUAL(close_and_fetch(), "01");
}
CAF_TEST(await) {
std::mutex mx;
std::condition_variable cv;
std::thread t{[&] { inbox.synchronized_emplace_back(mx, cv, 1); }};
inbox.synchronized_await(mx, cv);
CAF_REQUIRE_EQUAL(close_and_fetch(), "1");
t.join();
}
CAF_TEST(timed_await) {
std::mutex mx;
std::condition_variable cv;
auto tout = std::chrono::system_clock::now();
tout += std::chrono::microseconds(1);
auto res = inbox.synchronized_await(mx, cv, tout);
CAF_REQUIRE_EQUAL(res, false);
fill(inbox, 1);
res = inbox.synchronized_await(mx, cv, tout);
CAF_REQUIRE_EQUAL(res, true);
CHECK_EQ(fetch(), "1");
tout += std::chrono::hours(1000);
std::thread t{[&] { inbox.synchronized_emplace_back(mx, cv, 2); }};
res = inbox.synchronized_await(mx, cv, tout);
CAF_REQUIRE_EQUAL(res, true);
CAF_REQUIRE_EQUAL(close_and_fetch(), "2");
t.join();
}
END_FIXTURE_SCOPE()
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE intrusive.wdrr_dynamic_multiplexed_queue
#include "caf/intrusive/wdrr_dynamic_multiplexed_queue.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "core-test.hpp"
#include <memory>
using namespace caf;
using namespace caf::intrusive;
namespace {
struct inode : singly_linked<inode> {
int value;
inode(int x = 0) noexcept : value(x) {
// nop
}
};
std::string to_string(const inode& x) {
return std::to_string(x.value);
}
struct nested_inode_policy {
using mapped_type = inode;
using task_size_type = int;
using deficit_type = int;
using deleter_type = std::default_delete<mapped_type>;
using unique_pointer = std::unique_ptr<mapped_type, deleter_type>;
static task_size_type task_size(const mapped_type&) noexcept {
return 1;
}
std::unique_ptr<int> queue_id;
nested_inode_policy(int i) : queue_id(new int(i)) {
// nop
}
nested_inode_policy(nested_inode_policy&&) noexcept = default;
nested_inode_policy& operator=(nested_inode_policy&&) noexcept = default;
};
struct inode_policy {
using mapped_type = inode;
using key_type = int;
using task_size_type = int;
using deficit_type = int;
using deleter_type = std::default_delete<mapped_type>;
using unique_pointer = std::unique_ptr<mapped_type, deleter_type>;
using queue_type = drr_queue<nested_inode_policy>;
using queue_map_type = std::map<key_type, queue_type>;
static key_type id_of(const inode& x) noexcept {
return x.value % 3;
}
static bool enabled(const queue_type&) noexcept {
return true;
}
deficit_type quantum(const queue_type& q, deficit_type x) noexcept {
return enable_priorities && *q.policy().queue_id == 0 ? 2 * x : x;
}
static void cleanup(queue_type&) noexcept {
// nop
}
static bool push_back(queue_type& q, mapped_type* ptr) noexcept {
return q.push_back(ptr);
}
bool enable_priorities = false;
};
using queue_type = wdrr_dynamic_multiplexed_queue<inode_policy>;
using nested_queue_type = inode_policy::queue_type;
struct fixture {
inode_policy policy;
queue_type queue{policy};
int fill(queue_type&) {
return 0;
}
template <class T, class... Ts>
int fill(queue_type& q, T x, Ts... xs) {
return (q.emplace_back(x) ? 1 : 0) + fill(q, xs...);
}
std::string fetch(int quantum) {
std::string result;
auto f = [&](int id, nested_queue_type& q, inode& x) {
CHECK_EQ(id, *q.policy().queue_id);
if (!result.empty())
result += ',';
result += to_string(id);
result += ':';
result += to_string(x);
return task_result::resume;
};
queue.new_round(quantum, f);
return result;
}
void make_queues() {
for (int i = 0; i < 3; ++i)
queue.queues().emplace(i, nested_inode_policy{i});
}
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
CAF_TEST(default_constructed) {
CAF_REQUIRE_EQUAL(queue.empty(), true);
}
CAF_TEST(dropping) {
CAF_REQUIRE_EQUAL(queue.empty(), true);
CAF_REQUIRE_EQUAL(fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9, 12), 0);
CAF_REQUIRE_EQUAL(queue.empty(), true);
}
CAF_TEST(new_round) {
make_queues();
fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9, 12);
CAF_REQUIRE_EQUAL(queue.empty(), false);
CHECK_EQ(fetch(1), "0:3,1:1,2:2");
CAF_REQUIRE_EQUAL(queue.empty(), false);
CHECK_EQ(fetch(9), "0:6,0:9,0:12,1:4,1:7,2:5,2:8");
CAF_REQUIRE_EQUAL(queue.empty(), true);
}
CAF_TEST(priorities) {
make_queues();
queue.policy().enable_priorities = true;
fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9);
// Allow f to consume 2 items from the high priority and 1 item otherwise.
CHECK_EQ(fetch(1), "0:3,0:6,1:1,2:2");
CAF_REQUIRE_EQUAL(queue.empty(), false);
// Drain the high-priority queue with one item left per other queue.
CHECK_EQ(fetch(1), "0:9,1:4,2:5");
CAF_REQUIRE_EQUAL(queue.empty(), false);
// Drain queue.
CHECK_EQ(fetch(1000), "1:7,2:8");
CAF_REQUIRE_EQUAL(queue.empty(), true);
}
CAF_TEST(peek_all) {
auto queue_to_string = [&] {
std::string str;
auto peek_fun = [&](const inode& x) {
if (!str.empty())
str += ", ";
str += std::to_string(x.value);
};
queue.peek_all(peek_fun);
return str;
};
make_queues();
CHECK_EQ(queue_to_string(), "");
queue.emplace_back(1);
CHECK_EQ(queue_to_string(), "1");
queue.emplace_back(2);
CHECK_EQ(queue_to_string(), "1, 2");
queue.emplace_back(3);
// Lists are iterated in order and 3 is stored in the first queue for
// `x mod 3 == 0` values.
CHECK_EQ(queue_to_string(), "3, 1, 2");
queue.emplace_back(4);
CHECK_EQ(queue_to_string(), "3, 1, 4, 2");
}
END_FIXTURE_SCOPE()
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE intrusive.wdrr_fixed_multiplexed_queue
#include "caf/intrusive/wdrr_fixed_multiplexed_queue.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "core-test.hpp"
#include <memory>
using namespace caf;
using namespace caf::intrusive;
namespace {
struct inode : singly_linked<inode> {
int value;
inode(int x = 0) : value(x) {
// nop
}
};
std::string to_string(const inode& x) {
return std::to_string(x.value);
}
class high_prio_queue;
struct inode_policy {
using mapped_type = inode;
using task_size_type = int;
using deficit_type = int;
using deleter_type = std::default_delete<mapped_type>;
using unique_pointer = std::unique_ptr<mapped_type, deleter_type>;
static inline task_size_type task_size(const mapped_type&) {
return 1;
}
static inline size_t id_of(const inode& x) {
return x.value % 3;
}
template <class Queue>
deficit_type quantum(const Queue&, deficit_type x) {
return x;
}
deficit_type quantum(const high_prio_queue&, deficit_type x) {
return enable_priorities ? 2 * x : x;
}
bool enable_priorities = false;
};
class high_prio_queue : public drr_queue<inode_policy> {
public:
using super = drr_queue<inode_policy>;
using super::super;
};
using nested_queue_type = drr_queue<inode_policy>;
using queue_type
= wdrr_fixed_multiplexed_queue<inode_policy, high_prio_queue,
nested_queue_type, nested_queue_type>;
struct fetch_helper {
std::string result;
template <size_t I, class Queue>
task_result
operator()(std::integral_constant<size_t, I>, const Queue&, inode& x) {
if (!result.empty())
result += ',';
result += std::to_string(I);
result += ':';
result += to_string(x);
return task_result::resume;
}
};
struct fixture {
inode_policy policy;
queue_type queue{policy, policy, policy, policy};
void fill(queue_type&) {
// nop
}
template <class T, class... Ts>
void fill(queue_type& q, T x, Ts... xs) {
q.emplace_back(x);
fill(q, xs...);
}
std::string fetch(int quantum) {
std::string result;
auto f = [&](size_t id, drr_queue<inode_policy>&, inode& x) {
if (!result.empty())
result += ',';
result += std::to_string(id);
result += ':';
result += to_string(x);
return task_result::resume;
};
queue.new_round(quantum, f);
return result;
}
};
auto make_new_round_result(size_t consumed_items, bool stop_all) {
return new_round_result{consumed_items, stop_all};
}
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
CAF_TEST(default_constructed) {
CAF_REQUIRE_EQUAL(queue.empty(), true);
}
CAF_TEST(new_round) {
fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9, 12);
// Allow f to consume 2 items per nested queue.
fetch_helper f;
auto round_result = queue.new_round(2, f);
CHECK_EQ(round_result, make_new_round_result(6, false));
CHECK_EQ(f.result, "0:3,0:6,1:1,1:4,2:2,2:5");
CAF_REQUIRE_EQUAL(queue.empty(), false);
// Allow f to consume one more item from each queue.
f.result.clear();
round_result = queue.new_round(1, f);
CHECK_EQ(round_result, make_new_round_result(3, false));
CHECK_EQ(f.result, "0:9,1:7,2:8");
CAF_REQUIRE_EQUAL(queue.empty(), false);
// Allow f to consume the remainder, i.e., 12.
f.result.clear();
round_result = queue.new_round(1000, f);
CHECK_EQ(round_result, make_new_round_result(1, false));
CHECK_EQ(f.result, "0:12");
CAF_REQUIRE_EQUAL(queue.empty(), true);
}
CAF_TEST(priorities) {
queue.policy().enable_priorities = true;
fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9);
// Allow f to consume 2 items from the high priority and 1 item otherwise.
CHECK_EQ(fetch(1), "0:3,0:6,1:1,2:2");
CAF_REQUIRE_EQUAL(queue.empty(), false);
// Drain the high-priority queue with one item left per other queue.
CHECK_EQ(fetch(1), "0:9,1:4,2:5");
CAF_REQUIRE_EQUAL(queue.empty(), false);
// Drain queue.
CHECK_EQ(fetch(1000), "1:7,2:8");
CAF_REQUIRE_EQUAL(queue.empty(), true);
}
CAF_TEST(peek_all) {
auto queue_to_string = [&] {
std::string str;
auto peek_fun = [&](const inode& x) {
if (!str.empty())
str += ", ";
str += std::to_string(x.value);
};
queue.peek_all(peek_fun);
return str;
};
CHECK_EQ(queue_to_string(), "");
queue.emplace_back(1);
CHECK_EQ(queue_to_string(), "1");
queue.emplace_back(2);
CHECK_EQ(queue_to_string(), "1, 2");
queue.emplace_back(3);
// Lists are iterated in order and 3 is stored in the first queue for
// `x mod 3 == 0` values.
CHECK_EQ(queue_to_string(), "3, 1, 2");
queue.emplace_back(4);
CHECK_EQ(queue_to_string(), "3, 1, 4, 2");
}
END_FIXTURE_SCOPE()
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE policy.categorized
#include "caf/policy/categorized.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/intrusive/wdrr_dynamic_multiplexed_queue.hpp"
#include "caf/intrusive/wdrr_fixed_multiplexed_queue.hpp"
#include "caf/policy/normal_messages.hpp"
#include "caf/policy/urgent_messages.hpp"
#include "caf/unit.hpp"
#include "core-test.hpp"
using namespace caf;
namespace {
using urgent_queue = intrusive::drr_queue<policy::urgent_messages>;
using normal_queue = intrusive::drr_queue<policy::normal_messages>;
struct mailbox_policy {
using deficit_type = size_t;
using mapped_type = mailbox_element;
using unique_pointer = mailbox_element_ptr;
using queue_type
= intrusive::wdrr_fixed_multiplexed_queue<policy::categorized, urgent_queue,
normal_queue>;
};
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
struct fixture {};
struct consumer {
std::vector<int> ints;
template <class Key, class Queue>
intrusive::task_result
operator()(const Key&, const Queue&, const mailbox_element& x) {
if (!x.content().match_elements<int>())
CAF_FAIL("unexpected message: " << x.content());
ints.emplace_back(x.content().get_as<int>(0));
return intrusive::task_result::resume;
}
template <class Key, class Queue, class... Ts>
intrusive::task_result operator()(const Key&, const Queue&, const Ts&...) {
CAF_FAIL("unexpected message type"); // << typeid(Ts).name());
return intrusive::task_result::resume;
}
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
CAF_TEST(priorities) {
mailbox_type mbox{unit, unit, unit};
mbox.push_back(make_mailbox_element(nullptr, make_message_id(), {}, 123));
mbox.push_back(make_mailbox_element(
nullptr, make_message_id(message_priority::high), {}, 456));
consumer f;
mbox.new_round(1000, f);
CHECK_EQ(f.ints, std::vector<int>({456, 123}));
}
END_FIXTURE_SCOPE()
......@@ -18,7 +18,6 @@
#include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/none.hpp"
#include "caf/policy/normal_messages.hpp"
#include "caf/unordered_flat_map.hpp"
namespace caf::net {
......
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