Commit f44e39f0 authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Add intrusive queues and inboxes

The queue types implement Deficit Round Robin (DRR) work flows that form
the basis for the new mailbox implementation.
parent 2cbd028b
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_DRR_CACHED_QUEUE_HPP
#define CAF_INTRUSIVE_DRR_CACHED_QUEUE_HPP
#include <utility>
#include "caf/intrusive/task_queue.hpp"
#include "caf/intrusive/task_result.hpp"
namespace caf {
namespace intrusive {
/// A Deficit Round Robin queue with an internal cache for allowing skipping
/// consumers.
template <class Policy>
class drr_cached_queue : public task_queue<Policy> {
public:
// -- member types ----------------------------------------------------------
using super = task_queue<Policy>;
using typename super::policy_type;
using typename super::deleter_type;
using typename super::unique_pointer;
using typename super::value_type;
using cache_type = task_queue<Policy>;
using deficit_type = typename policy_type::deficit_type;
using task_size_type = typename policy_type::task_size_type;
// -- constructors, destructors, and assignment operators -------------------
drr_cached_queue(const policy_type& p) : super(p), deficit_(0), cache_(p) {
// nop
}
drr_cached_queue(drr_cached_queue&& other)
: super(std::move(other)),
deficit_(0) {
// nop
}
drr_cached_queue& operator=(drr_cached_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() noexcept {
super::prepend(cache_);
}
/// Takes the first element out of the queue if the deficit allows it and
/// returns the element.
unique_pointer take_front() noexcept {
super::prepend(cache_);
unique_pointer result;
if (!super::empty()) {
auto ptr = promote(super::head_.next);
auto ts = super::policy_.task_size(*ptr);
CAF_ASSERT(ts > 0);
if (ts <= deficit_) {
deficit_ -= ts;
super::total_task_size_ -= ts;
super::head_.next = ptr->next;
if (super::total_task_size_ == 0) {
CAF_ASSERT(super::head_.next == &(super::tail_));
deficit_ = 0;
super::tail_.next = &(super::head_);
}
result.reset(ptr);
}
}
return result;
}
/// 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&>()))) {
long consumed = 0;
if (!cache_.empty())
super::prepend(cache_);
if (!super::empty()) {
auto ptr = promote(super::head_.next);
auto ts = super::policy_.task_size(*ptr);
while (ts <= deficit_) {
auto next = ptr->next;
super::total_task_size_ -= ts;
// Make sure the queue is in a consistent state before calling `f` in
// case `f` recursively calls consume.
super::head_.next = next;
if (super::total_task_size_ == 0) {
CAF_ASSERT(super::head_.next == &(super::tail_));
super::tail_.next = &(super::head_);
}
// Always decrease the deficit_ counter, again because `f` is allowed
// to call consume again.
deficit_ -= ts;
auto res = f(*ptr);
if (res == task_result::skip) {
// Fix deficit and push the unconsumed item to the cache.
deficit_ += ts;
cache_.push_back(ptr);
if (super::empty()) {
deficit_ = 0;
return consumed != 0;
}
} else {
deleter_type d;
d(ptr);
++consumed;
if (!cache_.empty())
super::prepend(cache_);
if (super::empty()) {
deficit_ = 0;
return consumed != 0;
}
if (res == task_result::stop)
return consumed != 0;
}
// Next iteration.
ptr = promote(super::head_.next);
ts = super::policy_.task_size(*ptr);
}
}
return consumed != 0;
}
/// Run a new round with `quantum`, dispatching all tasks to `consumer`.
template <class F>
bool new_round(deficit_type quantum, F& consumer)
noexcept(noexcept(consumer(std::declval<value_type&>()))) {
if (!super::empty()) {
deficit_ += quantum;
return consume(consumer);
}
return false;
}
cache_type& cache() noexcept {
return cache_;
}
private:
// -- member variables ------------------------------------------------------
/// Stores the deficit on this queue.
deficit_type deficit_ = 0;
/// Stores previously skipped items.
cache_type cache_;
};
} // namespace intrusive
} // namespace caf
#endif // CAF_INTRUSIVE_DRR_CACHED_QUEUE_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_DRR_QUEUE_HPP
#define CAF_INTRUSIVE_DRR_QUEUE_HPP
#include <utility>
#include "caf/intrusive/task_queue.hpp"
namespace caf {
namespace 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(const policy_type& p) : super(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
}
/// Takes the first element out of the queue if the deficit allows it and
/// returns the element.
unique_pointer take_front() noexcept {
unique_pointer result;
if (!super::empty()) {
auto ptr = promote(super::head_.next);
auto ts = super::policy_.task_size(*ptr);
CAF_ASSERT(ts > 0);
if (ts <= deficit_) {
deficit_ -= ts;
super::total_task_size_ -= ts;
super::head_.next = ptr->next;
if (super::total_task_size_ == 0) {
CAF_ASSERT(super::head_.next == &(super::tail_));
deficit_ = 0;
super::tail_.next = &(super::head_);
}
result.reset(ptr);
}
}
return result;
}
/// 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 ptr = take_front();
if (ptr == nullptr)
return false;
do {
f(*ptr);
ptr = take_front();
} while (ptr != nullptr);
return true;
}
/// 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>
bool new_round(deficit_type quantum, F& consumer) {
if (!super::empty()) {
deficit_ += quantum;
return consume(consumer);
}
return false;
}
private:
// -- member variables -------------------------------------------------------
/// Stores the deficit on this queue.
deficit_type deficit_ = 0;
};
} // namespace intrusive
} // namespace caf
#endif // CAF_INTRUSIVE_DRR_QUEUE_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_FIFO_INBOX_HPP
#define CAF_INTRUSIVE_FIFO_INBOX_HPP
#include "caf/config.hpp"
#include <list>
#include <deque>
#include <mutex>
#include <atomic>
#include <memory>
#include <limits>
#include <condition_variable> // std::cv_status
#include "caf/config.hpp"
#include "caf/intrusive/inbox_result.hpp"
#include "caf/intrusive/lifo_inbox.hpp"
namespace caf {
namespace 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 value_type = typename policy_type::mapped_type;
using deleter_type = typename policy_type::deleter_type;
using lifo_inbox_type = lifo_inbox<policy_type>;
using pointer = value_type*;
using unique_pointer = typename policy_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 count(size_t = 0) noexcept CAF_DEPRECATED {
return size();
}
/// 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();
}
inbox_result push_back(pointer ptr) noexcept {
return inbox_.push_front(ptr);
}
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)...));
}
// -- 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(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>
bool new_round(typename policy_type::deficit_type quantum, F& consumer) {
fetch_more();
return queue_.new_round(quantum, consumer);
}
pointer peek() noexcept {
fetch_more();
return queue_.peek();
}
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 intrusive
} // namespace caf
#endif // CAF_INTRUSIVE_FIFO_INBOX_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_FORWARD_ITERATOR_HPP
#define CAF_INTRUSIVE_FORWARD_ITERATOR_HPP
#include <cstddef>
#include <iterator>
#include <type_traits>
namespace caf {
namespace intrusive {
// A forward iterator for intrusive lists.
template <class T>
class forward_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::forward_iterator_tag;
// -- member variables -------------------------------------------------------
node_pointer ptr;
// -- constructors, destructors, and assignment operators --------------------
constexpr forward_iterator(node_pointer init = nullptr) : ptr(init) {
// nop
}
forward_iterator(const forward_iterator&) = default;
forward_iterator& operator=(const forward_iterator&) = default;
// -- convenience functions --------------------------------------------------
forward_iterator next() {
return ptr->next;
}
// -- operators --------------------------------------------------------------
forward_iterator& operator++() {
ptr = promote(ptr->next);
return *this;
}
forward_iterator operator++(int) {
forward_iterator res = *this;
ptr = promote(ptr->next);
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 forward_iterator
template <class T>
bool operator==(const forward_iterator<T>& x,
const forward_iterator<T>& y) {
return x.ptr == y.ptr;
}
/// @relates forward_iterator
template <class T>
bool operator!=(const forward_iterator<T>& x,
const forward_iterator<T>& y) {
return x.ptr != y.ptr;
}
} // namespace intrusive
} // namespace caf
#endif // CAF_INTRUSIVE_FORWARD_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_INBOX_RESULT_HPP
#define CAF_INTRUSIVE_INBOX_RESULT_HPP
namespace caf {
namespace intrusive {
/// Communicates the state of a LIFO or FIFO inbox after pushing to it.
enum class inbox_result {
/// Indicates that the enqueue operation succeeded and
/// the reader is ready to receive the data.
success,
/// Indicates that the enqueue operation succeeded and
/// the reader is currently blocked, i.e., needs to be re-scheduled.
unblocked_reader,
/// Indicates that the enqueue operation failed because the
/// queue has been closed by the reader.
queue_closed
};
} // namespace intrusive
} // namespace caf
#endif // CAF_INTRUSIVE_INBOX_RESULT_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_LIFO_INBOX_HPP
#define CAF_INTRUSIVE_LIFO_INBOX_HPP
#include <atomic>
#include <condition_variable>
#include <mutex>
#include "caf/config.hpp"
#include "caf/intrusive/inbox_result.hpp"
namespace caf {
namespace intrusive {
/// An intrusive, thread-safe LIFO queue implementation for a single reader
/// with any number of writers.
template <class Policy>
class lifo_inbox {
public:
using policy_type = Policy;
using value_type = typename policy_type::mapped_type;
using pointer = value_type*;
using node_type = typename value_type::node_type;
using node_pointer = node_type*;
using deleter_type = typename policy_type::deleter_type;
using unique_pointer = typename policy_type::unique_pointer;
/// Tries to enqueue a new element to the inbox.
/// @threadsafe
inbox_result push_front(pointer new_element) noexcept {
CAF_ASSERT(new_element != nullptr);
pointer e = stack_.load();
auto eof = stack_closed_tag();
auto blk = reader_blocked_tag();
while (e != eof) {
// A tag is never part of a non-empty list.
new_element->next = e != blk ? e : nullptr;
if (stack_.compare_exchange_strong(e, new_element))
return e == reader_blocked_tag() ? inbox_result::unblocked_reader
: inbox_result::success;
// Continue with new value of `e`.
}
// The queue has been closed, drop messages.
deleter_type d;
d(new_element);
return inbox_result::queue_closed;
}
/// Tries to enqueue a new element to the inbox.
/// @threadsafe
inbox_result push_front(unique_pointer x) noexcept {
return push_front(x.release());
}
/// Tries to enqueue a new element to the mailbox.
/// @threadsafe
template <class... Ts>
inbox_result emplace_front(Ts&&... xs) {
return push_front(new value_type(std::forward<Ts>(xs)...));
}
/// Queries whether this queue is empty.
/// @pre `!closed() && !blocked()`
bool empty() const noexcept {
CAF_ASSERT(!closed());
CAF_ASSERT(!blocked());
return stack_.load() == stack_empty_tag();
}
/// Queries whether this has been closed.
bool closed() const noexcept {
return stack_.load() == stack_closed_tag();
}
/// 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 stack_.load() == reader_blocked_tag();
}
/// Tries to set this queue from `empty` to `blocked`.
bool try_block() noexcept {
auto e = stack_empty_tag();
return stack_.compare_exchange_strong(e, reader_blocked_tag());
}
/// Tries to set this queue from `blocked` to `empty`.
bool try_unblock() noexcept {
auto e = reader_blocked_tag();
return stack_.compare_exchange_strong(e, stack_empty_tag());
}
/// Sets the head to `new_head` and returns the previous head if the queue
/// was not empty.
pointer take_head(pointer new_head) noexcept {
// This member function should only be used to transition to closed or
// empty.
CAF_ASSERT(new_head == stack_closed_tag()
|| new_head == stack_empty_tag());
// Must not be called on a closed queue.
pointer e = stack_.load();
CAF_ASSERT(e != stack_closed_tag());
// We don't assert these conditions again since only the owner is allowed
// to call this member function, i.e., there's never a race on `take_head`.
while (e != new_head) {
if (stack_.compare_exchange_weak(e, new_head)) {
CAF_ASSERT(e != stack_closed_tag());
if (is_empty_or_blocked_tag(e)) {
// Sanity check: going from empty/blocked to closed.
CAF_ASSERT(new_head == stack_closed_tag());
return nullptr;
}
return e;
}
// Next iteration.
}
return nullptr;
}
/// Sets the head to `stack_empty_tag()` and returns the previous head if
/// the queue was not empty.
pointer take_head() noexcept {
return take_head(stack_empty_tag());
}
/// Closes this queue and deletes all remaining elements.
/// @warning Call only from the reader (owner).
void close() noexcept {
deleter_type d;
static_assert(noexcept(d(std::declval<pointer>())),
"deleter is not noexcept");
close(d);
}
/// Closes this queue and applies `f` to each pointer. The function object
/// `f` must take ownership of the passed pointer.
/// @warning Call only from the reader (owner).
template <class F>
void close(F& f) noexcept(noexcept(f(std::declval<pointer>()))) {
node_pointer ptr = take_head(stack_closed_tag());
while (ptr != nullptr) {
auto next = ptr->next;
f(promote(ptr));
ptr = next;
}
}
lifo_inbox() noexcept {
stack_ = stack_empty_tag();
}
~lifo_inbox() noexcept {
if (!closed())
close();
}
// -- synchronized access ---------------------------------------------------
template <class Mutex, class CondVar>
bool synchronized_push_front(Mutex& mtx, CondVar& cv, pointer ptr) {
switch (push_front(ptr)) {
default:
// enqueued message to a running actor's mailbox
return true;
case inbox_result::unblocked_reader: {
std::unique_lock<Mutex> guard(mtx);
cv.notify_one();
return true;
}
case inbox_result::queue_closed:
// actor no longer alive
return false;
}
}
template <class Mutex, class CondVar>
bool synchronized_push_front(Mutex& mtx, CondVar& cv, unique_pointer ptr) {
return synchronized_push_front(mtx, cv, ptr.relase());
}
template <class Mutex, class CondVar, class... Ts>
bool synchronized_emplace_front(Mutex& mtx, CondVar& cv, Ts&&... xs) {
return synchronized_push_front(mtx, cv,
new value_type(std::forward<Ts>(xs)...));
}
template <class Mutex, class CondVar>
void synchronized_await(Mutex& mtx, CondVar& cv) {
CAF_ASSERT(!closed());
if (try_block()) {
std::unique_lock<Mutex> guard(mtx);
while (blocked())
cv.wait(guard);
}
}
template <class Mutex, class CondVar, class TimePoint>
bool synchronized_await(Mutex& mtx, CondVar& cv, const TimePoint& timeout) {
CAF_ASSERT(!closed());
if (try_block()) {
std::unique_lock<Mutex> guard(mtx);
while (blocked()) {
if (cv.wait_until(guard, timeout) == std::cv_status::timeout) {
// if we're unable to set the queue from blocked to empty,
// than there's a new element in the list
return !try_unblock();
}
}
}
return true;
}
private:
static constexpr pointer stack_empty_tag() {
// We are *never* going to dereference the returned pointer. It is only
// used as indicator wheter this queue is empty or not.
return static_cast<pointer>(nullptr);
}
pointer stack_closed_tag() const noexcept {
// We are *never* going to dereference the returned pointer. It is only
// used as indicator wheter this queue is closed or not.
return reinterpret_cast<pointer>(reinterpret_cast<intptr_t>(this) + 1);
}
pointer reader_blocked_tag() const noexcept {
// We are *never* going to dereference the returned pointer. It is only
// used as indicator wheter the owner of the queue is currently waiting for
// new messages.
return reinterpret_cast<pointer>(const_cast<lifo_inbox*>(this));
}
bool is_empty_or_blocked_tag(pointer x) const noexcept {
return x == stack_empty_tag() || x == reader_blocked_tag();
}
// -- member variables ------------------------------------------------------
std::atomic<pointer> stack_;
};
} // namespace intrusive
} // namespace caf
#endif // CAF_INTRUSIVE_LIFO_INBOX_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_SINGLY_LINKED_HPP
#define CAF_INTRUSIVE_SINGLY_LINKED_HPP
namespace caf {
namespace intrusive {
/// Intrusive base for singly linked types that allows queues to use `T` with
/// dummy nodes.
template <class T>
struct singly_linked {
// -- member types -----------------------------------------------------------
/// The type for dummy nodes in singly linked lists.
using node_type = singly_linked<T>;
/// Type of the pointer connecting two singly linked nodes.
using node_pointer = node_type*;
// -- constructors, destructors, and assignment operators --------------------
singly_linked(node_pointer n = nullptr) : next(n) {
// nop
}
// -- member variables -------------------------------------------------------
/// Intrusive pointer to the next element.
node_pointer next;
};
/// Casts a node type to its value type.
template <class T>
T* promote(singly_linked<T>* ptr) {
return static_cast<T*>(ptr);
}
/// Casts a node type to its value type.
template <class T>
const T* promote(const singly_linked<T>* ptr) {
return static_cast<const T*>(ptr);
}
} // namespace intrusive
} // namespace caf
#endif // CAF_INTRUSIVE_SINGLY_LINKED_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_TASK_QUEUE_HPP
#define CAF_INTRUSIVE_TASK_QUEUE_HPP
#include <utility>
#include "caf/intrusive/forward_iterator.hpp"
namespace caf {
namespace intrusive {
/// A singly-linked FIFO queue for storing tasks of varying size. This queue is
/// used as a base type for concrete task abstractions such as `drr_queue` and
/// therefore has no dequeue functions.
template <class Policy>
class task_queue {
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 const_pointer = const value_type*;
using reference = value_type&;
using const_reference = const value_type&;
using deleter_type = typename policy_type::deleter_type;
using unique_pointer = typename policy_type::unique_pointer;
using task_size_type = typename policy_type::task_size_type;
using iterator = forward_iterator<value_type>;
// -- constructors, destructors, and assignment operators -------------------
task_queue(const policy_type& p) : old_last_(nullptr), policy_(p) {
init();
}
task_queue(task_queue&& other) : task_queue(other.policy()) {
if (other.empty()) {
init();
} else {
head_.next = other.head_.next;
tail_.next = other.tail_.next;
tail_.next->next = &tail_;
total_task_size_ = other.total_task_size_;
other.init();
}
}
task_queue& operator=(task_queue&& other) {
deinit();
if (other.empty()) {
init();
} else {
head_.next = other.head_.next;
tail_.next = other.tail_.next;
tail_.next->next = &tail_;
total_task_size_ = other.total_task_size_;
other.init();
}
return *this;
}
virtual ~task_queue() {
deinit();
}
// -- observers -------------------------------------------------------------
/// Returns the policy object.
const policy_type& policy() const noexcept {
return policy_;
}
/// Returns the accumulated size of all stored tasks.
task_size_type total_task_size() const noexcept {
return total_task_size_;
}
/// Returns whether the queue has no elements.
bool empty() const noexcept {
return total_task_size() == 0;
}
/// Peeks at the first element in the queue. Returns `nullptr` if the queue is
/// empty.
pointer peek() noexcept {
auto ptr = head_.next;
return ptr != &tail_ ? promote(ptr) : nullptr;
}
// -- modifiers -------------------------------------------------------------
/// Removes all elements from the queue.
void clear() {
deinit();
init();
}
/// Appends `ptr` to the queue.
/// @pre `ptr != nullptr`
void push_back(pointer ptr) noexcept {
CAF_ASSERT(ptr != nullptr);
tail_.next->next = ptr;
tail_.next = ptr;
ptr->next = &tail_;
inc_total_task_size(*ptr);
}
/// Appends `ptr` to the queue.
/// @pre `ptr != nullptr`
void push_back(unique_pointer ptr) noexcept {
CAF_ASSERT(ptr != nullptr);
push_back(ptr.release());
}
/// Creates a new element from `xs...` and appends it.
template <class... Ts>
void emplace_back(Ts&&... xs) {
push_back(new value_type(std::forward<Ts>(xs)...));
}
/// @private
void inc_total_task_size(task_size_type x) noexcept {
total_task_size_ += x;
}
/// @private
void inc_total_task_size(const value_type& x) noexcept {
inc_total_task_size(policy_.task_size(x));
}
/// Returns an iterator to the dummy before the first element.
iterator before_begin() noexcept {
return &head_;
}
/// Returns an iterator to the dummy before the first element.
iterator begin() noexcept {
return head_.next;
}
/// Returns a pointer to the dummy past the last element.
iterator end() noexcept {
return &tail_;
}
/// Returns a pointer to the first element.
pointer front() noexcept {
return promote(head_.next);
}
/// Returns a pointer to the last element.
pointer back() noexcept {
return promote(tail_.next);
}
/// Transfers all element from `other` to the front of this queue.
template <class Container>
void prepend(Container& other) {
if (other.empty())
return;
if (empty()) {
*this = std::move(other);
return;
}
other.back()->next = head_.next;
head_.next = other.front();
total_task_size_ += other.total_task_size();
other.init();
}
/// Transfers all element from `other` to the back of this queue.
template <class Container>
void append(Container& other) {
if (other.empty())
return;
if (empty()) {
*this = std::move(other);
return;
}
back()->next = other.front();
other.back()->next = &tail_;
tail_.next = other.back();
total_task_size_ += other.total_task_size();
other.init();
}
/// Allows to insert a LIFO-ordered sequence at the end of the queue by
/// calling this member function multiple times. Converts the order to FIFO
/// on the fly.
/// @warning leaves the queue in an invalid state until calling
/// `stop_lifo_append`.
/// @private
void lifo_append(node_pointer ptr) {
if (old_last_ == nullptr) {
old_last_ = back();
push_back(promote(ptr));
} else {
ptr->next = new_head_;
inc_total_task_size(*promote(ptr));
}
new_head_ = ptr;
}
/// Restores a consistent state of the queue after calling `lifo_append`.
/// @private
void stop_lifo_append() {
if (old_last_ != nullptr) {
CAF_ASSERT(new_head_ != nullptr);
old_last_->next = new_head_;
old_last_ = nullptr;
}
}
// -- construction and destruction helper -----------------------------------
/// Restores a consistent, empty state.
/// @private
void init() noexcept {
head_.next = &tail_;
tail_.next = &head_;
total_task_size_ = 0;
}
/// Deletes all elements.
/// @warning leaves the queue in an invalid state until calling `init` again.
/// @private
void deinit() noexcept {
for (auto i = head_.next; i != &tail_;) {
auto ptr = i;
i = i->next;
deleter_type d;
d(promote(ptr));
}
}
protected:
// -- member variables ------------------------------------------------------
/// node element pointing to the first element.
node_type head_;
/// node element pointing past the last element.
node_type tail_;
/// Stores the total size of all items in the queue.
task_size_type total_task_size_ = 0;
/// Used for LIFO -> FIFO insertion.
node_pointer old_last_;
/// Used for LIFO -> FIFO insertion.
node_pointer new_head_;
/// Manipulates instances of T.
policy_type policy_;
};
} // namespace intrusive
} // namespace caf
#endif // CAF_INTRUSIVE_TASK_QUEUE_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_TASK_RESULT_HPP
#define CAF_INTRUSIVE_TASK_RESULT_HPP
#include <type_traits>
#include "caf/fwd.hpp"
namespace caf {
namespace intrusive {
/// Communicates the state of a consumer to a task queue.
enum class task_result {
/// The consumer processed the task and is ready to receive the next one.
resume,
/// The consumer skipped the task and is ready to receive the next one.
skip,
/// The consumer processed the task but does not accept further tasks.
stop
};
} // namespace intrusive
} // namespace caf
#endif // CAF_INTRUSIVE_TASK_RESULT_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_WDRR_FIXED_MULTIPLEXED_QUEUE_HPP
#define CAF_INTRUSIVE_WDRR_FIXED_MULTIPLEXED_QUEUE_HPP
#include <tuple>
#include <utility>
#include "caf/detail/type_traits.hpp"
namespace caf {
namespace 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 mapped_type = typename policy_type::mapped_type;
using pointer = mapped_type*;
using unique_pointer = typename policy_type::unique_pointer;
using task_size_type = typename policy_type::task_size_type;
static constexpr size_t num_queues = sizeof...(Qs) + 1;
template <class... Ps>
wdrr_fixed_multiplexed_queue(policy_type p0,
typename Q::policy_type p1, Ps... ps)
: qs_(std::move(p1), std::forward<Ps>(ps)...),
policy_(std::move(p0)) {
// nop
}
policy_type& policy() noexcept {
return policy_;
}
const policy_type& policy() const noexcept {
return policy_;
}
void push_back(mapped_type* ptr) noexcept {
push_back_recursion<0>(policy_.id_of(*ptr), ptr);
}
void push_back(unique_pointer ptr) noexcept {
push_back(ptr.release());
}
template <class... Ts>
void emplace_back(Ts&&... xs) {
push_back(new mapped_type(std::forward<Ts>(xs)...));
}
/// 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>
bool new_round(long quantum,
F& f) noexcept(noexcept(f(std::declval<mapped_type&>()))) {
return new_round_recursion<0>(quantum, f) != 0;
}
pointer peek() noexcept {
return peek_recursion<0>();
}
/// 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 ---------------------------------------------
template <size_t I>
detail::enable_if_t<I == num_queues>
push_back_recursion(size_t, mapped_type*) noexcept {
// nop
}
template <size_t I>
detail::enable_if_t<I != num_queues>
push_back_recursion(size_t pos, mapped_type* ptr) noexcept {
if (pos == I) {
auto& q = std::get<I>(qs_);
q.push_back(ptr);
} else {
push_back_recursion<I + 1>(pos, ptr);
}
}
template <size_t I, class F>
detail::enable_if_t<I == num_queues, int>
new_round_recursion(deficit_type, F&) noexcept {
return 0;
}
template <size_t I, class F>
detail::enable_if_t<I != num_queues, int>
new_round_recursion(deficit_type quantum,
F& f) noexcept(noexcept(f(std::declval<mapped_type&>()))) {
auto& q = std::get<I>(qs_);
if (q.new_round(policy_.quantum(q, quantum), f))
return 1 + new_round_recursion<I + 1>(quantum, f);
return 0 + new_round_recursion<I + 1>(quantum, f);
}
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>
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 intrusive
} // namespace caf
#endif // CAF_INTRUSIVE_WDRR_FIXED_MULTIPLEXED_QUEUE_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. *
******************************************************************************/
#define CAF_SUITE drr_cached_queue
#include <memory>
#include "caf/test/unit_test.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "caf/intrusive/drr_cached_queue.hpp"
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};
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...);
}
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(drr_cached_queue_tests, 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.begin(), queue.end());
CAF_REQUIRE_EQUAL(queue.before_begin()->next, queue.end().ptr);
}
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);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(fseq, "246");
CAF_CHECK_EQUAL(queue.deficit(), 0);
// Allow g to consume 1, 3, 5, and 7.
round_result = queue.new_round(4, g);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(gseq, "1357");
CAF_CHECK_EQUAL(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);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(seq, "21436587");
CAF_CHECK_EQUAL(queue.deficit(), 0);
CAF_CHECK_EQUAL(deep_to_string(queue.cache()), "[9]");
}
CAF_TEST_FIXTURE_SCOPE_END()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#define CAF_SUITE drr_queue
#include <memory>
#include "caf/test/unit_test.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "caf/intrusive/drr_queue.hpp"
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...);
}
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(drr_queue_tests, 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.take_front(), nullptr);
CAF_REQUIRE_EQUAL(queue.begin(), queue.end());
CAF_REQUIRE_EQUAL(queue.before_begin()->next, queue.end().ptr);
}
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.take_front();
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);
};
// Allow f to consume 1, 2, and 3 with a leftover deficit of 1.
auto round_result = queue.new_round(7, f);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(seq, "123");
CAF_CHECK_EQUAL(queue.deficit(), 1);
// Allow f to consume 4 and 5 with a leftover deficit of 0.
round_result = queue.new_round(8, f);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(seq, "12345");
CAF_CHECK_EQUAL(queue.deficit(), 0);
// Allow f to consume 6 with a leftover deficit of 0 (queue is empty).
round_result = queue.new_round(1000, f);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(seq, "123456");
CAF_CHECK_EQUAL(queue.deficit(), 0);
// new_round on an empty queue does nothing.
round_result = queue.new_round(1000, f);
CAF_CHECK_EQUAL(round_result, false);
CAF_CHECK_EQUAL(seq, "123456");
CAF_CHECK_EQUAL(queue.deficit(), 0);
}
CAF_TEST_FIXTURE_SCOPE_END()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#define CAF_SUITE fifo_inbox
#include <memory>
#include "caf/test/unit_test.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/intrusive/singly_linked.hpp"
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);
};
inbox.new_round(1000, f);
return result;
}
std::string close_and_fetch() {
std::string result;
auto f = [&](inode& x) {
result += to_string(x);
};
inbox.close();
inbox.queue().new_round(1000, f);
return result;
}
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(fifo_inbox_tests, 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);
CAF_CHECK_EQUAL(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();
}
CAF_TEST_FIXTURE_SCOPE_END()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#define CAF_SUITE lifo_inbox
#include <memory>
#include "caf/test/unit_test.hpp"
#include "caf/intrusive/lifo_inbox.hpp"
#include "caf/intrusive/singly_linked.hpp"
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 inbox_type = lifo_inbox<inode_policy>;
struct fixture {
inode_policy policy;
inbox_type inbox;
void fill(inbox_type&) {
// nop
}
template <class T, class... Ts>
void fill(inbox_type& i, T x, Ts... xs) {
i.emplace_front(x);
fill(i, xs...);
}
std::string fetch() {
std::string result;
inode_policy::unique_pointer ptr{inbox.take_head()};
while (ptr != nullptr) {
auto next = ptr->next;
result += to_string(*ptr);
ptr.reset(promote(next));
}
return result;
}
std::string close_and_fetch() {
std::string result;
auto f = [&](inode* x) {
result += to_string(*x);
delete x;
};
inbox.close(f);
return result;
}
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(lifo_inbox_tests, 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(), "321");
CAF_REQUIRE_EQUAL(inbox.closed(), true);
}
CAF_TEST(push_after_close) {
inbox.close();
auto res = inbox.push_front(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_front(new inode(1));
CAF_REQUIRE_EQUAL(res, inbox_result::unblocked_reader);
res = inbox.push_front(new inode(2));
CAF_REQUIRE_EQUAL(res, inbox_result::success);
CAF_REQUIRE_EQUAL(close_and_fetch(), "21");
}
CAF_TEST(await) {
std::mutex mx;
std::condition_variable cv;
std::thread t{[&] {
inbox.synchronized_emplace_front(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);
CAF_CHECK_EQUAL(fetch(), "1");
tout += std::chrono::hours(1000);
std::thread t{[&] {
inbox.synchronized_emplace_front(mx, cv, 2);
}};
res = inbox.synchronized_await(mx, cv, tout);
CAF_REQUIRE_EQUAL(res, true);
CAF_REQUIRE_EQUAL(close_and_fetch(), "2");
t.join();
}
CAF_TEST_FIXTURE_SCOPE_END()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#define CAF_SUITE task_queue
#include <memory>
#include "caf/test/unit_test.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "caf/intrusive/task_queue.hpp"
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 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 = task_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...);
}
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(task_queue_tests, fixture)
CAF_TEST(default_constructed) {
CAF_REQUIRE_EQUAL(queue.empty(), true);
CAF_REQUIRE_EQUAL(queue.total_task_size(), 0);
CAF_REQUIRE_EQUAL(queue.peek(), nullptr);
CAF_REQUIRE_EQUAL(queue.begin(), queue.end());
CAF_REQUIRE_EQUAL(queue.before_begin()->next, queue.end().ptr);
}
CAF_TEST(push_back) {
queue.emplace_back(1);
queue.push_back(inode_policy::unique_pointer{new inode(2)});
queue.push_back(new inode(3));
CAF_REQUIRE_EQUAL(deep_to_string(queue), "[1, 2, 3]");
}
CAF_TEST(lifo_conversion) {
queue.lifo_append(new inode(3));
queue.lifo_append(new inode(2));
queue.lifo_append(new inode(1));
queue.stop_lifo_append();
CAF_REQUIRE_EQUAL(deep_to_string(queue), "[1, 2, 3]");
}
CAF_TEST(move_construct) {
fill(queue, 1, 2, 3);
queue_type q2 = std::move(queue);
CAF_REQUIRE_EQUAL(queue.empty(), true);
CAF_REQUIRE_EQUAL(q2.empty(), false);
CAF_REQUIRE_EQUAL(deep_to_string(q2), "[1, 2, 3]");
}
CAF_TEST(move_assign) {
queue_type q2{policy};
fill(q2, 1, 2, 3);
queue = std::move(q2);
CAF_REQUIRE_EQUAL(q2.empty(), true);
CAF_REQUIRE_EQUAL(queue.empty(), false);
CAF_REQUIRE_EQUAL(deep_to_string(queue), "[1, 2, 3]");
}
CAF_TEST(append) {
queue_type q2{policy};
fill(queue, 1, 2, 3);
fill(q2, 4, 5, 6);
queue.append(q2);
CAF_REQUIRE_EQUAL(q2.empty(), true);
CAF_REQUIRE_EQUAL(queue.empty(), false);
CAF_REQUIRE_EQUAL(deep_to_string(queue), "[1, 2, 3, 4, 5, 6]");
}
CAF_TEST(prepend) {
queue_type q2{policy};
fill(queue, 1, 2, 3);
fill(q2, 4, 5, 6);
queue.prepend(q2);
CAF_REQUIRE_EQUAL(q2.empty(), true);
CAF_REQUIRE_EQUAL(queue.empty(), false);
CAF_REQUIRE_EQUAL(deep_to_string(queue), "[4, 5, 6, 1, 2, 3]");
}
CAF_TEST(peek) {
CAF_CHECK_EQUAL(queue.peek(), nullptr);
fill(queue, 1, 2, 3);
CAF_CHECK_EQUAL(queue.peek()->value, 1);
}
CAF_TEST(task_size) {
fill(queue, 1, 2, 3);
CAF_CHECK_EQUAL(queue.total_task_size(), 6);
fill(queue, 4, 5);
CAF_CHECK_EQUAL(queue.total_task_size(), 15);
queue.clear();
CAF_CHECK_EQUAL(queue.total_task_size(), 0);
}
CAF_TEST_FIXTURE_SCOPE_END()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#define CAF_SUITE wdrr_fixed_multiplexed_queue
#include <memory>
#include "caf/test/unit_test.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "caf/intrusive/wdrr_fixed_multiplexed_queue.hpp"
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 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 seq;
std::function<void (inode&)> f;
fixture() {
f = [&](inode& x) {
if (!seq.empty())
seq += ',';
seq += to_string(x);
};
}
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(wdrr_fixed_multiplexed_queue_tests, 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.
auto round_result = queue.new_round(2, f);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(seq, "3,6,1,4,2,5");
CAF_REQUIRE_EQUAL(queue.empty(), false);
// Allow f to consume one more item from each queue.
seq.clear();
round_result = queue.new_round(1, f);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(seq, "9,7,8");
CAF_REQUIRE_EQUAL(queue.empty(), false);
// Allow f to consume the remainder, i.e., 12.
seq.clear();
round_result = queue.new_round(1000, f);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(seq, "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.
auto round_result = queue.new_round(1, f);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(seq, "3,6,1,2");
CAF_REQUIRE_EQUAL(queue.empty(), false);
// Drain the high-priority queue with one item left per other queue.
seq.clear();
round_result = queue.new_round(1, f);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(seq, "9,4,5");
CAF_REQUIRE_EQUAL(queue.empty(), false);
// Drain queue.
seq.clear();
round_result = queue.new_round(1000, f);
CAF_CHECK_EQUAL(round_result, true);
CAF_CHECK_EQUAL(seq, "7,8");
CAF_REQUIRE_EQUAL(queue.empty(), true);
}
CAF_TEST_FIXTURE_SCOPE_END()
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