Unverified Commit 10f1f01f authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #1511

Make actor mailboxes configurable
parents 7081bef0 1960c5ef
......@@ -58,7 +58,6 @@ caf_add_component(
exit_reason
flow.op.state
intrusive.inbox_result
intrusive.task_result
invoke_message_result
message_priority
pec
......@@ -70,6 +69,7 @@ caf_add_component(
caf/abstract_actor.cpp
caf/abstract_channel.cpp
caf/abstract_group.cpp
caf/abstract_mailbox.cpp
caf/action.cpp
caf/actor.cpp
caf/actor_addr.cpp
......@@ -111,6 +111,8 @@ caf_add_component(
caf/detail/behavior_stack.cpp
caf/detail/blocking_behavior.cpp
caf/detail/config_consumer.cpp
caf/detail/default_mailbox.cpp
caf/detail/default_mailbox.test.cpp
caf/detail/get_process_id.cpp
caf/detail/glob_match.cpp
caf/detail/group_tunnel.cpp
......@@ -118,6 +120,8 @@ caf_add_component(
caf/detail/json.cpp
caf/detail/latch.cpp
caf/detail/local_group_module.cpp
caf/detail/mailbox_factory.cpp
caf/detail/mailbox_factory.test.cpp
caf/detail/message_builder_element.cpp
caf/detail/message_data.cpp
caf/detail/meta_object.cpp
......@@ -156,6 +160,9 @@ caf_add_component(
caf/hash/fnv.test.cpp
caf/hash/sha1.cpp
caf/init_global_meta_objects.cpp
caf/intrusive/lifo_inbox.test.cpp
caf/intrusive/linked_list.test.cpp
caf/intrusive/stack.test.cpp
caf/ipv4_address.cpp
caf/ipv4_endpoint.cpp
caf/ipv4_subnet.cpp
......@@ -306,13 +313,6 @@ 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
......@@ -341,7 +341,6 @@ caf_add_component(
node_id
optional
or_else
policy.categorized
policy.select_all
policy.select_any
request_timeout
......
// 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.
#include "caf/abstract_mailbox.hpp"
namespace caf {
abstract_mailbox::~abstract_mailbox() {
// nop
}
} // namespace caf
// 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/intrusive/inbox_result.hpp"
#include "caf/mailbox_element.hpp"
namespace caf {
/// The base class for all mailbox implementations.
class CAF_CORE_EXPORT abstract_mailbox {
public:
virtual ~abstract_mailbox();
/// Adds a new element to the mailbox.
/// @returns `inbox_result::success` if the element has been added to the
/// mailbox, `inbox_result::unblocked_reader` if the reader has been
/// unblocked, or `inbox_result::queue_closed` if the mailbox has
/// been closed.
/// @threadsafe
virtual intrusive::inbox_result push_back(mailbox_element_ptr ptr) = 0;
/// Adds a new element to the mailbox by putting it in front of the queue.
/// @note Only the owning actor is allowed to call this function.
virtual void push_front(mailbox_element_ptr ptr) = 0;
/// Removes the next element from the mailbox.
/// @returns The next element in the mailbox or `nullptr` if the mailbox is
/// empty.
/// @note Only the owning actor is allowed to call this function.
virtual mailbox_element_ptr pop_front() = 0;
/// Checks whether the mailbox has been closed.
/// @note Only the owning actor is allowed to call this function.
virtual bool closed() const noexcept = 0;
/// Checks whether the owner of this mailbox is currently waiting for new
/// messages.
/// @note Only the owning actor is allowed to call this function.
virtual bool blocked() const noexcept = 0;
/// Tries to put the mailbox in a blocked state.
/// @note Only the owning actor is allowed to call this function.
virtual bool try_block() = 0;
/// Tries to put the mailbox in an empty state from a blocked state.
/// @note Only the owning actor is allowed to call this function.
virtual bool try_unblock() = 0;
/// Closes the mailbox and discards all pending messages.
/// @returns The number of dropped messages.
/// @note Only the owning actor is allowed to call this function.
virtual size_t close(const error& reason) = 0;
/// Returns the number of pending messages.
/// @note Only the owning actor is allowed to call this function.
virtual size_t size() = 0;
/// Increases the reference count by one.
virtual void ref_mailbox() noexcept = 0;
/// Decreases the reference count by one and deletes this instance if the
/// reference count drops to zero.
virtual void deref_mailbox() noexcept = 0;
/// Checks whether the mailbox is empty.
bool empty() {
return size() == 0;
}
/// @private
/// @note Only used by the legacy test framework. Remove when dropping support
/// for it.
virtual mailbox_element* peek(message_id id) = 0;
};
} // namespace caf
......@@ -34,6 +34,7 @@ public:
int flags;
input_range<const group>* groups;
detail::unique_function<behavior(local_actor*)> init_fun;
detail::mailbox_factory* mbox_factory = nullptr;
// -- properties -------------------------------------------------------------
......
This diff is collapsed.
......@@ -4,6 +4,7 @@
#pragma once
#include "caf/abstract_mailbox.hpp"
#include "caf/actor_config.hpp"
#include "caf/actor_traits.hpp"
#include "caf/after.hpp"
......@@ -11,15 +12,12 @@
#include "caf/detail/apply_args.hpp"
#include "caf/detail/blocking_behavior.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/default_mailbox.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/extend.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_cached_queue.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/intrusive/stack.hpp"
#include "caf/is_timeout_or_catch_all.hpp"
#include "caf/local_actor.hpp"
#include "caf/mailbox_element.hpp"
......@@ -28,9 +26,6 @@
#include "caf/mixin/subscriber.hpp"
#include "caf/none.hpp"
#include "caf/policy/arg.hpp"
#include "caf/policy/categorized.hpp"
#include "caf/policy/normal_messages.hpp"
#include "caf/policy/urgent_messages.hpp"
#include "caf/send.hpp"
#include "caf/typed_actor.hpp"
......@@ -58,35 +53,6 @@ public:
/// Base type.
using super = extended_base;
/// Stores asynchronous messages with default priority.
using normal_queue = intrusive::drr_cached_queue<policy::normal_messages>;
/// Stores asynchronous messages with high priority.
using urgent_queue = intrusive::drr_cached_queue<policy::urgent_messages>;
/// Configures the FIFO inbox with two nested queues:
///
/// 1. Default asynchronous messages
/// 2. High-priority asynchronous 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,
normal_queue, urgent_queue>;
static constexpr size_t normal_queue_index = 0;
static constexpr size_t urgent_queue_index = 1;
};
/// A queue optimized for single-reader-many-writers.
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
/// Absolute timeout type.
using timeout_type = std::chrono::high_resolution_clock::time_point;
......@@ -193,23 +159,6 @@ public:
}
};
struct mailbox_visitor {
blocking_actor* self;
bool& done;
receive_cond& rcc;
message_id mid;
detail::blocking_behavior& bhvr;
// Dispatches messages with high and normal priority to the same handler.
template <class Queue>
intrusive::task_result operator()(size_t, Queue&, mailbox_element& x) {
return (*this)(x);
}
// Consumes `x`.
intrusive::task_result operator()(mailbox_element& x);
};
// -- constructors and destructors -------------------------------------------
blocking_actor(actor_config& cfg);
......@@ -344,7 +293,7 @@ public:
virtual mailbox_element_ptr dequeue();
/// Returns the queue for storing incoming messages.
mailbox_type& mailbox() {
abstract_mailbox& mailbox() {
return mailbox_;
}
/// @cond PRIVATE
......@@ -405,6 +354,8 @@ private:
size_t attach_functor(const strong_actor_ptr&);
void unstash();
template <class... Ts>
size_t attach_functor(const typed_actor<Ts...>& x) {
return attach_functor(actor_cast<strong_actor_ptr>(x));
......@@ -420,8 +371,11 @@ private:
// -- member variables -------------------------------------------------------
// used by both event-based and blocking actors
mailbox_type mailbox_;
/// Stores incoming messages.
detail::default_mailbox mailbox_;
/// Stashes skipped messages until the actor processes the next message.
intrusive::stack<mailbox_element> stash_;
};
} // namespace caf
// 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.
#include "caf/detail/default_mailbox.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/error.hpp"
#include "caf/message_id.hpp"
namespace caf::detail {
intrusive::inbox_result default_mailbox::push_back(mailbox_element_ptr ptr) {
return inbox_.push_front(ptr.release());
}
void default_mailbox::push_front(mailbox_element_ptr ptr) {
if (ptr->mid.is_urgent_message())
urgent_queue_.push_front(ptr.release());
else
normal_queue_.push_front(ptr.release());
}
mailbox_element* default_mailbox::peek(message_id id) {
if (inbox_.closed() || inbox_.blocked()) {
return nullptr;
}
fetch_more();
if (id.is_async()) {
if (!urgent_queue_.empty())
return urgent_queue_.front();
if (!normal_queue_.empty())
return normal_queue_.front();
return nullptr;
}
auto pred = [id](mailbox_element& x) { return x.mid == id; };
if (auto result = urgent_queue_.find_if(pred))
return result;
return normal_queue_.find_if(pred);
}
mailbox_element_ptr default_mailbox::pop_front() {
for (;;) {
if (auto result = urgent_queue_.pop_front())
return result;
if (auto result = normal_queue_.pop_front())
return result;
if (!fetch_more())
return nullptr;
}
}
bool default_mailbox::closed() const noexcept {
return inbox_.closed();
}
bool default_mailbox::blocked() const noexcept {
return inbox_.blocked();
}
bool default_mailbox::try_block() {
return cached() == 0 && inbox_.try_block();
}
bool default_mailbox::try_unblock() {
return inbox_.try_unblock();
}
size_t default_mailbox::close(const error& reason) {
size_t result = 0;
detail::sync_request_bouncer bounce{reason};
auto bounce_and_count = [&bounce, &result](mailbox_element* ptr) {
bounce(*ptr);
delete ptr;
++result;
};
urgent_queue_.drain(bounce_and_count);
normal_queue_.drain(bounce_and_count);
inbox_.close(bounce_and_count);
return result;
}
size_t default_mailbox::size() {
fetch_more();
return cached();
}
bool default_mailbox::fetch_more() {
using node_type = intrusive::singly_linked<mailbox_element>;
auto promote = [](node_type* ptr) {
return static_cast<mailbox_element*>(ptr);
};
auto* head = static_cast<node_type*>(inbox_.take_head());
if (head == nullptr)
return false;
auto urgent_insertion_point = urgent_queue_.before_end();
auto normal_insertion_point = normal_queue_.before_end();
do {
auto next = head->next;
auto phead = promote(head);
if (phead->mid.is_urgent_message())
urgent_queue_.insert_after(urgent_insertion_point, phead);
else
normal_queue_.insert_after(normal_insertion_point, phead);
head = next;
} while (head != nullptr);
return true;
}
void default_mailbox::ref_mailbox() noexcept {
++ref_count_;
}
void default_mailbox::deref_mailbox() noexcept {
if (--ref_count_ == 0)
delete this;
}
} // 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/abstract_mailbox.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/intrusive/lifo_inbox.hpp"
#include "caf/intrusive/linked_list.hpp"
#include <atomic>
#include <cstddef>
namespace caf::detail {
/// Our default mailbox implementation. Uses a LIFO inbox for storing incoming
/// messages and combines it with two FIFO caches for storing urgent and normal
/// messages.
class CAF_CORE_EXPORT default_mailbox : public abstract_mailbox {
public:
default_mailbox() noexcept : ref_count_(1) {
// nop
}
default_mailbox(const default_mailbox&) = delete;
default_mailbox& operator=(const default_mailbox&) = delete;
mailbox_element* peek(message_id id) override;
intrusive::inbox_result push_back(mailbox_element_ptr ptr) override;
void push_front(mailbox_element_ptr ptr) override;
mailbox_element_ptr pop_front() override;
bool closed() const noexcept override;
bool blocked() const noexcept override;
bool try_block() override;
bool try_unblock() override;
size_t close(const error&) override;
size_t size() override;
void ref_mailbox() noexcept override;
void deref_mailbox() noexcept override;
size_t ref_count() const noexcept {
return ref_count_.load();
}
private:
/// Returns the total number of elements stored in the queues.
size_t cached() const noexcept {
return urgent_queue_.size() + normal_queue_.size();
}
/// Tries to fetch more messages from the LIFO inbox.
bool fetch_more();
/// Stores urgent messages in FIFO order.
intrusive::linked_list<mailbox_element> urgent_queue_;
/// Stores normal messages in FIFO order.
intrusive::linked_list<mailbox_element> normal_queue_;
/// Stores incoming messages in LIFO order.
alignas(CAF_CACHE_LINE_SIZE) intrusive::lifo_inbox<mailbox_element> inbox_;
/// The intrusive reference count.
alignas(CAF_CACHE_LINE_SIZE) std::atomic<size_t> ref_count_;
};
} // 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.
#include "caf/detail/default_mailbox.hpp"
#include "caf/test/caf_test_main.hpp"
#include "caf/test/test.hpp"
using namespace caf;
using ires = intrusive::inbox_result;
template <message_priority P = message_priority::normal>
auto make_int_msg(int value) {
return make_mailbox_element(nullptr, make_message_id(P), {},
make_message(value));
}
TEST("a default-constructed mailbox is empty") {
detail::default_mailbox uut;
check(!uut.closed());
check(!uut.blocked());
check_eq(uut.size(), 0u);
check_eq(uut.pop_front(), nullptr);
check_eq(uut.size(), 0u);
check(uut.try_block());
check(!uut.try_block());
}
TEST("the first item unblocks the mailbox") {
detail::default_mailbox uut;
check(uut.try_block());
check_eq(uut.push_back(make_int_msg(1)), ires::unblocked_reader);
check_eq(uut.push_back(make_int_msg(2)), ires::success);
}
TEST("a closed mailbox no longer accepts new messages") {
detail::default_mailbox uut;
uut.close(error{});
check_eq(uut.push_back(make_int_msg(1)), ires::queue_closed);
}
TEST("urgent messages are processed first") {
detail::default_mailbox uut;
check_eq(uut.push_back(make_int_msg(1)), ires::success);
check_eq(uut.push_back(make_int_msg(2)), ires::success);
check_eq(uut.push_back(make_int_msg<message_priority::high>(3)),
ires::success);
std::vector<message> results;
for (auto ptr = uut.pop_front(); ptr != nullptr; ptr = uut.pop_front()) {
results.emplace_back(ptr->content());
}
if (check_eq(results.size(), 3u)) {
check_eq(results[0].get_as<int>(0), 3);
check_eq(results[1].get_as<int>(0), 1);
check_eq(results[2].get_as<int>(0), 2);
}
}
TEST("calling push_front inserts messages at the beginning") {
detail::default_mailbox uut;
check_eq(uut.push_back(make_int_msg(1)), ires::success);
check_eq(uut.push_back(make_int_msg(2)), ires::success);
uut.push_front(make_int_msg(3));
uut.push_front(make_int_msg(4));
std::vector<message> results;
for (auto ptr = uut.pop_front(); ptr != nullptr; ptr = uut.pop_front()) {
results.emplace_back(ptr->content());
}
if (check_eq(results.size(), 4u)) {
check_eq(results[0].get_as<int>(0), 4);
check_eq(results[1].get_as<int>(0), 3);
check_eq(results[2].get_as<int>(0), 1);
check_eq(results[3].get_as<int>(0), 2);
}
}
CAF_TEST_MAIN()
// 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.
#include "caf/detail/mailbox_factory.hpp"
namespace caf::detail {
mailbox_factory::~mailbox_factory() {
// nop
}
} // 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/detail/core_export.hpp"
#include "caf/fwd.hpp"
namespace caf::detail {
/// The base class for all mailbox implementations.
class CAF_CORE_EXPORT mailbox_factory {
public:
virtual ~mailbox_factory();
/// Creates a new mailbox for `owner`.
virtual abstract_mailbox* make(scheduled_actor* owner) = 0;
/// Creates a new mailbox for `owner`.
virtual abstract_mailbox* make(blocking_actor* owner) = 0;
};
} // 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.
#include "caf/detail/mailbox_factory.hpp"
#include "caf/test/caf_test_main.hpp"
#include "caf/test/test.hpp"
#include "caf/blocking_actor.hpp"
#include "caf/detail/default_mailbox.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/local_actor.hpp"
#include "caf/scheduler/test_coordinator.hpp"
using namespace caf;
class dummy_mailbox_factory : public detail::mailbox_factory {
public:
~dummy_mailbox_factory() override {
for (auto& kvp : mailboxes)
kvp.second->deref_mailbox();
}
abstract_mailbox* make(local_actor* owner) {
auto ptr = new detail::default_mailbox;
ptr->ref_mailbox();
mailboxes.emplace(owner->id(), ptr);
return ptr;
}
abstract_mailbox* make(scheduled_actor* owner) override {
return make(static_cast<local_actor*>(owner));
}
abstract_mailbox* make(blocking_actor* owner) override {
return make(static_cast<local_actor*>(owner));
}
std::map<actor_id, detail::default_mailbox*> mailboxes;
};
TEST("a mailbox factory creates mailboxes for actors") {
dummy_mailbox_factory factory;
actor_system_config cfg;
cfg.set("caf.scheduler.policy", "testing");
actor_system sys{cfg};
auto& sched = static_cast<scheduler::test_coordinator&>(sys.scheduler());
auto spawn_dummy = [&sys, &factory](auto fn) {
using impl_t = event_based_actor;
actor_config cfg{sys.dummy_execution_unit()};
cfg.init_fun.emplace([fn](local_actor* self) {
fn(static_cast<impl_t*>(self));
return behavior{};
});
cfg.mbox_factory = &factory;
auto res = make_actor<impl_t>(sys.next_actor_id(), sys.node(), &sys, cfg);
auto ptr = static_cast<impl_t*>(actor_cast<abstract_actor*>(res));
ptr->launch(cfg.host, false, false);
return res;
};
SECTION("spawning dummies creates mailboxes") {
auto initialized = std::make_shared<size_t>(0u);
auto dummy_impl = [this, &factory, initialized](event_based_actor* self) {
++*initialized;
auto* mbox = factory.mailboxes[self->id()];
check_eq(std::addressof(self->mailbox()), mbox);
check_eq(mbox->ref_count(), 2u);
};
check_eq(factory.mailboxes.size(), 0u);
auto dummy1 = spawn_dummy(dummy_impl).id();
check_eq(factory.mailboxes.size(), 1u);
auto dummy2 = spawn_dummy(dummy_impl).id();
check_eq(factory.mailboxes.size(), 2u);
sched.run();
check_eq(*initialized, 2u);
check_eq(factory.mailboxes[dummy1]->ref_count(), 1u);
check_eq(factory.mailboxes[dummy2]->ref_count(), 1u);
}
}
CAF_TEST_MAIN()
......@@ -15,10 +15,6 @@
namespace caf::detail {
sync_request_bouncer::sync_request_bouncer(error r) : rsn(std::move(r)) {
// nop
}
void sync_request_bouncer::operator()(const strong_actor_ptr& sender,
const message_id& mid) const {
if (sender && mid.is_request())
......@@ -27,10 +23,8 @@ void sync_request_bouncer::operator()(const strong_actor_ptr& sender,
nullptr);
}
intrusive::task_result
sync_request_bouncer::operator()(const mailbox_element& e) const {
void sync_request_bouncer::operator()(const mailbox_element& e) const {
(*this)(e.sender, e.mid);
return intrusive::task_result::resume;
}
} // namespace caf::detail
......@@ -7,28 +7,26 @@
#include "caf/detail/core_export.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/task_result.hpp"
#include <cstdint>
namespace caf::detail {
/// Drains a mailbox and sends an error message to each unhandled request.
/// Consumes mailbox elements and sends an error message for each request.
struct CAF_CORE_EXPORT sync_request_bouncer {
error rsn;
explicit sync_request_bouncer(error r);
// -- constructors, destructors, and assignment operators --------------------
explicit sync_request_bouncer(error r) : rsn(std::move(r)) {
// nop
}
// -- apply ------------------------------------------------------------------
void operator()(const strong_actor_ptr& sender, const message_id& mid) const;
intrusive::task_result operator()(const mailbox_element& e) const;
void operator()(const mailbox_element& e) const;
/// Unwrap WDRR queues. Nesting WDRR queues results in a Key/Queue prefix for
/// each layer of nesting.
template <class Key, class Queue, class... Ts>
intrusive::task_result
operator()(const Key&, const Queue&, const Ts&... xs) const {
(*this)(xs...);
return intrusive::task_result::resume;
}
// -- member variables -------------------------------------------------------
error rsn;
};
} // namespace caf::detail
......@@ -436,19 +436,6 @@ public:
static constexpr bool value = false;
};
template <class T>
class has_peek_all {
private:
template <class U>
static int
fun(const U*, decltype(std::declval<U&>().peek_all(unit))* = nullptr);
static char fun(const void*);
public:
static constexpr bool value = sizeof(fun(static_cast<T*>(nullptr))) > 1;
};
CAF_HAS_MEMBER_TRAIT(clear);
CAF_HAS_MEMBER_TRAIT(data);
CAF_HAS_MEMBER_TRAIT(make_behavior);
......
......@@ -126,6 +126,18 @@ public:
*this = unique_function{ptr};
}
template <class Fn>
void emplace(Fn fn) {
destroy();
if constexpr (std::is_convertible<Fn, raw_pointer>::value) {
holds_wrapper_ = false;
fptr_ = fn;
} else {
holds_wrapper_ = true;
new (&wptr_) wrapper_pointer{make_wrapper(std::move(fn))};
}
}
// -- properties -------------------------------------------------------------
bool is_nullptr() const noexcept {
......
......@@ -10,6 +10,7 @@
#include "caf/flow/subscription.hpp"
#include <algorithm>
#include <deque>
#include <memory>
#include <tuple>
#include <type_traits>
......
......@@ -74,6 +74,7 @@ class [[deprecated("use std::string_view instead")]] string_view;
class [[nodiscard]] error;
class abstract_actor;
class abstract_group;
class abstract_mailbox;
class action;
class actor;
class actor_addr;
......@@ -215,14 +216,6 @@ class fnv;
} // namespace hash
// -- intrusive containers -----------------------------------------------------
namespace intrusive {
enum class task_result;
} // namespace intrusive
// -- marker classes for mixins ------------------------------------------------
namespace mixin {
......@@ -275,6 +268,8 @@ using int_gauge_family = metric_family_impl<int_gauge>;
namespace detail {
class mailbox_factory;
template <class>
struct gauge_oracle;
......
// 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
......@@ -9,20 +9,19 @@
#include <atomic>
#include <condition_variable>
#include <memory>
#include <mutex>
namespace caf::intrusive {
/// An intrusive, thread-safe LIFO queue implementation for a single reader
/// with any number of writers.
template <class Policy>
template <class T>
class lifo_inbox {
public:
// -- member types -----------------------------------------------------------
using policy_type = Policy;
using value_type = typename policy_type::mapped_type;
using value_type = T;
using pointer = value_type*;
......@@ -30,7 +29,7 @@ public:
using node_pointer = node_type*;
using unique_pointer = typename policy_type::unique_pointer;
using unique_pointer = std::unique_ptr<value_type>;
using deleter_type = typename unique_pointer::deleter_type;
......@@ -146,21 +145,15 @@ public:
/// Closes this queue and deletes all remaining elements.
/// @warning Call only from the reader (owner).
void close() noexcept {
deleter_type d;
// We assume the node destructor to never throw. However, the following
// static assert fails. Unfortunately, std::default_delete's apply operator
// is not noexcept (event for types that have a noexcept destructor).
// static_assert(noexcept(d(std::declval<pointer>())),
// "deleter is not noexcept");
close(d);
void close() {
close(deleter_type{});
}
/// 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>()))) {
void close(F&& f) {
node_pointer ptr = take_head(stack_closed_tag());
while (ptr != nullptr) {
auto next = ptr->next;
......@@ -178,62 +171,6 @@ public:
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.release());
}
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
......
// 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.
#include "caf/intrusive/lifo_inbox.hpp"
#include "caf/test/caf_test_main.hpp"
#include "caf/test/test.hpp"
#include <memory>
#include <numeric>
using namespace caf;
using intrusive::inbox_result;
struct inode : intrusive::singly_linked<inode> {
explicit inode(int x = 0) : value(x) {
// nop
}
int value;
};
using inbox_type = intrusive::lifo_inbox<inode>;
std::string drain(inbox_type& xs) {
std::vector<int> tmp;
std::unique_ptr<inode> ptr{xs.take_head()};
while (ptr != nullptr) {
auto next = ptr->next;
tmp.push_back(ptr->value);
ptr.reset(inbox_type::promote(next));
}
return caf::deep_to_string(tmp);
}
TEST("a default default-constructed inbox is empty") {
inbox_type uut;
check(!uut.closed());
check(!uut.blocked());
check(uut.empty());
check_eq(uut.take_head(), nullptr);
}
TEST("push_front adds elements to the front of the inbox") {
inbox_type uut;
check_eq(uut.emplace_front(1), inbox_result::success);
check_eq(uut.emplace_front(2), inbox_result::success);
check_eq(uut.emplace_front(3), inbox_result::success);
check_eq(drain(uut), "[3, 2, 1]");
}
TEST("push_front discards elements if the inbox is closed") {
inbox_type uut;
uut.close();
check(uut.closed());
auto res = uut.emplace_front(0);
check_eq(res, inbox_result::queue_closed);
}
TEST("push_front unblocks a blocked reader") {
inbox_type uut;
check(uut.try_block());
check_eq(uut.emplace_front(1), inbox_result::unblocked_reader);
check_eq(uut.emplace_front(2), inbox_result::success);
check_eq(drain(uut), "[2, 1]");
}
CAF_TEST_MAIN()
// 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.
#include "caf/intrusive/linked_list.hpp"
#include "caf/test/caf_test_main.hpp"
#include "caf/test/test.hpp"
#include <memory>
#include <numeric>
using namespace caf;
struct inode : intrusive::singly_linked<inode> {
explicit inode(int x = 0) : value(x) {
// nop
}
int value;
};
std::string to_string(const inode& x) {
return std::to_string(x.value);
}
using list_type = intrusive::linked_list<inode>;
template <class... Ts>
void fill(list_type& xs, Ts... args) {
(xs.emplace_back(args), ...);
}
TEST("a default default-constructed list is empty") {
list_type uut;
check_eq(uut.empty(), true);
check_eq(uut.size(), 0u);
check_eq(uut.peek(), nullptr);
check_eq(uut.begin(), uut.end());
}
TEST("lists are convertible to strings") {
list_type uut;
check_eq(deep_to_string(uut), "[]");
fill(uut, 1, 2, 3, 4);
check_eq(deep_to_string(uut), "[1, 2, 3, 4]");
}
TEST("push_back adds elements to the back of the list") {
list_type uut;
uut.emplace_back(1);
uut.push_back(std::make_unique<inode>(2));
uut.push_back(new inode(3));
check_eq(deep_to_string(uut), "[1, 2, 3]");
}
TEST("push_front adds elements to the front of the list") {
list_type uut;
uut.emplace_front(1);
uut.push_front(std::make_unique<inode>(2));
uut.push_front(new inode(3));
check_eq(deep_to_string(uut), "[3, 2, 1]");
}
TEST("insert_after inserts elements after a given position") {
list_type uut;
uut.insert_after(uut.before_end(), new inode(1));
uut.insert_after(uut.before_end(), new inode(3));
uut.insert_after(uut.begin(), new inode(2));
uut.insert_after(uut.before_begin(), new inode(0));
check_eq(deep_to_string(uut), "[0, 1, 2, 3]");
}
TEST("lists are movable") {
list_type uut;
SECTION("move constructor") {
fill(uut, 1, 2, 3);
list_type q2 = std::move(uut);
check_eq(uut.empty(), true);
check_eq(q2.empty(), false);
check_eq(deep_to_string(q2), "[1, 2, 3]");
}
SECTION("move assignment operator") {
list_type q2;
fill(q2, 1, 2, 3);
uut = std::move(q2);
check_eq(q2.empty(), true);
check_eq(uut.empty(), false);
check_eq(deep_to_string(uut), "[1, 2, 3]");
}
}
TEST("peek returns a pointer to the first element without removing it") {
list_type uut;
check_eq(uut.peek(), nullptr);
fill(uut, 1, 2, 3);
check_eq(uut.peek()->value, 1);
}
TEST("the size of the list is the number of elements") {
list_type uut;
fill(uut, 1, 2, 3);
check_eq(uut.size(), 3u);
fill(uut, 4, 5);
check_eq(uut.size(), 5u);
}
TEST("calling clear removes all elements from a list") {
list_type uut;
fill(uut, 1, 2, 3);
check_eq(uut.size(), 3u);
uut.clear();
check_eq(uut.size(), 0u);
}
TEST("find_if selects an element from the list") {
list_type uut;
fill(uut, 1, 2, 3);
SECTION("find_if returns a pointer to the first matching element") {
auto ptr = uut.find_if([](const inode& x) { return x.value == 2; });
if (check(ptr != nullptr))
check_eq(ptr->value, 2);
}
SECTION("find_if returns nullptr if nothing matches") {
auto ptr = uut.find_if([](const inode& x) { return x.value == 4; });
check(ptr == nullptr);
}
}
TEST("lists allow iterator-based access") {
list_type uut;
fill(uut, 1, 2, 3);
// Mutable access via begin/end.
for (auto& x : uut)
x.value *= 2;
check_eq(uut.front()->value, 2);
check_eq(uut.back()->value, 6);
// Immutable access via cbegin/cend.
check_eq(std::accumulate(uut.cbegin(), uut.cend(), 0,
[](int acc, const inode& x) {
return acc + x.value;
}),
12);
}
TEST("pop_front removes the oldest element of a list and returns it") {
list_type uut;
fill(uut, 1, 2, 3);
check_eq(uut.pop_front()->value, 1);
if (check_eq(uut.size(), 2u))
check_eq(uut.pop_front()->value, 2);
if (check_eq(uut.size(), 1u))
check_eq(uut.pop_front()->value, 3);
check(uut.empty());
}
CAF_TEST_MAIN()
// 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
......@@ -20,14 +20,20 @@ struct singly_linked {
// -- constructors, destructors, and assignment operators --------------------
singly_linked(node_pointer n = nullptr) : next(n) {
singly_linked() noexcept = default;
explicit singly_linked(node_pointer n) noexcept : next(n) {
// nop
}
singly_linked(const singly_linked&) = delete;
singly_linked& operator=(const singly_linked&) = delete;
// -- member variables -------------------------------------------------------
/// Intrusive pointer to the next element.
node_pointer next;
node_pointer next = nullptr;
};
} // namespace caf::intrusive
......@@ -4,28 +4,46 @@
#pragma once
#include "caf/fwd.hpp"
namespace caf::intrusive {
#include <string>
#include <type_traits>
/// A simple stack implementation with singly-linked nodes.
template <class T>
class stack {
public:
using pointer = T*;
namespace caf::intrusive {
stack() = default;
/// 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.
/// Illegal for consumers of non-cached queues (non-cached queues treat
/// `skip` and `resume` in the same way).
skip,
/// The consumer processed the task but does not accept further tasks.
stop,
/// The consumer processed the task but does not accept further tasks and no
/// subsequent queue shall start a new round.
stop_all,
};
stack(const stack&) = delete;
stack& operator=(const stack&) = delete;
~stack() {
while (head_) {
auto next = static_cast<pointer>(head_->next);
delete head_;
head_ = next;
}
}
std::string to_string(task_result);
void push(pointer ptr) {
ptr->next = head_;
head_ = ptr;
}
pointer pop() {
auto result = head_;
if (result)
head_ = static_cast<pointer>(result->next);
return result;
}
bool empty() const noexcept {
return head_ == nullptr;
}
private:
pointer head_ = nullptr;
};
} // 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.
#include "caf/intrusive/stack.hpp"
#include "caf/test/caf_test_main.hpp"
#include "caf/test/test.hpp"
using namespace caf;
struct int_node : intrusive::singly_linked<int_node> {
explicit int_node(int x) : value(x) {
// nop
}
int value = 0;
};
using int_stack = intrusive::stack<int_node>;
void push(int_stack& xs, int x) {
xs.push(new int_node(x));
}
auto pop(int_stack& xs) {
if (xs.empty())
CAF_RAISE_ERROR("cannot pop from an empty stack");
auto ptr = std::unique_ptr<int_node>{xs.pop()};
return ptr->value;
}
TEST("a default-constructed stack is empty") {
int_stack uut;
check(uut.empty());
check_eq(uut.pop(), nullptr);
}
TEST("pushing values to a stack makes it non-empty") {
int_stack uut;
check(uut.empty());
push(uut, 1);
check(!uut.empty());
push(uut, 2);
check(!uut.empty());
push(uut, 3);
check(!uut.empty());
}
TEST("popping values from a stack returns the last pushed value") {
int_stack uut;
check(uut.empty());
push(uut, 1);
push(uut, 2);
push(uut, 3);
check(!uut.empty());
check_eq(pop(uut), 3);
check_eq(pop(uut), 2);
check_eq(pop(uut), 1);
check(uut.empty());
}
CAF_TEST_MAIN()
// 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
......@@ -13,7 +13,6 @@
#include "caf/detail/meta_object.hpp"
#include "caf/detail/pretty_type_name.hpp"
#include "caf/detail/set_thread_name.hpp"
#include "caf/intrusive/task_result.hpp"
#include "caf/local_actor.hpp"
#include "caf/message.hpp"
#include "caf/string_algorithms.hpp"
......
......@@ -15,9 +15,6 @@
#include "caf/detail/scope_guard.hpp"
#include "caf/detail/sync_ring_buffer.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "caf/ref_counted.hpp"
#include "caf/timestamp.hpp"
......
// 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
......@@ -10,6 +10,7 @@
#include "caf/config.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/default_invoke_result_visitor.hpp"
#include "caf/detail/mailbox_factory.hpp"
#include "caf/detail/meta_object.hpp"
#include "caf/detail/private_thread.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
......@@ -19,6 +20,8 @@
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/stream.hpp"
#include <new>
using namespace std::string_literals;
namespace caf {
......@@ -118,7 +121,6 @@ scheduled_actor::batch_forwarder::~batch_forwarder() {
scheduled_actor::scheduled_actor(actor_config& cfg)
: super(cfg),
mailbox_(unit, unit, unit),
default_handler_(print_and_drop),
error_handler_(default_error_handler),
down_handler_(default_down_handler),
......@@ -130,11 +132,18 @@ scheduled_actor::scheduled_actor(actor_config& cfg)
exception_handler_(default_exception_handler)
#endif // CAF_ENABLE_EXCEPTIONS
{
// nop
if (cfg.mbox_factory == nullptr)
mailbox_ = new (&default_mailbox_) detail::default_mailbox();
else
mailbox_ = cfg.mbox_factory->make(this);
}
scheduled_actor::~scheduled_actor() {
// nop
unstash();
if (mailbox_ == &default_mailbox_)
default_mailbox_.~default_mailbox();
else
mailbox_->deref_mailbox();
}
// -- overridden functions of abstract_actor -----------------------------------
......@@ -182,15 +191,9 @@ bool scheduled_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) {
}
mailbox_element* scheduled_actor::peek_at_next_mailbox_element() {
if (mailbox().closed() || mailbox().blocked()) {
return nullptr;
} else if (awaited_responses_.empty()) {
return mailbox().peek();
} else {
auto mid = awaited_responses_.begin()->first;
auto pred = [mid](mailbox_element& x) { return x.mid == mid; };
return mailbox().find_if(pred);
}
return mailbox().peek(awaited_responses_.empty()
? make_message_id()
: awaited_responses_.begin()->first);
}
// -- overridden functions of local_actor --------------------------------------
......@@ -238,19 +241,11 @@ bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) {
run_actions();
}
// Clear mailbox.
if (!mailbox_.closed()) {
mailbox_.close();
get_normal_queue().flush_cache();
get_urgent_queue().flush_cache();
detail::sync_request_bouncer bounce{fail_state};
auto dropped = mailbox_.queue().new_round(1000, bounce).consumed_items;
while (dropped > 0) {
if (getf(abstract_actor::collects_metrics_flag)) {
auto val = static_cast<int64_t>(dropped);
metrics_.mailbox_size->dec(val);
}
dropped = mailbox_.queue().new_round(1000, bounce).consumed_items;
}
if (!mailbox().closed()) {
unstash();
auto dropped = mailbox().close(fail_state);
if (dropped > 0 && metrics_.mailbox_size)
metrics_.mailbox_size->dec(static_cast<int64_t>(dropped));
}
// Dispatch to parent's `cleanup` function.
return super::cleanup(std::move(fail_state), host);
......@@ -277,66 +272,54 @@ resumable::resume_result scheduled_actor::resume(execution_unit* ctx,
if (!activate(ctx))
return resumable::done;
size_t consumed = 0;
auto guard = detail::make_scope_guard([this, &consumed] {
CAF_LOG_DEBUG("resume consumed" << consumed << "messages");
if (consumed > 0) {
auto val = static_cast<int64_t>(consumed);
home_system().base_metrics().processed_messages->inc(val);
}
});
auto reset_timeouts_if_needed = [&] {
// Set a new receive timeout if we called our behavior at least once.
if (consumed > 0)
set_receive_timeout();
};
// Callback for handling urgent and normal messages.
auto handle_async = [this, max_throughput, &consumed](mailbox_element& x) {
return run_with_metrics(x, [this, max_throughput, &consumed, &x] {
switch (reactivate(x)) {
case activation_result::terminated:
return intrusive::task_result::stop;
case activation_result::success:
return ++consumed < max_throughput ? intrusive::task_result::resume
: intrusive::task_result::stop_all;
case activation_result::skipped:
return intrusive::task_result::skip;
default:
return intrusive::task_result::resume;
}
});
};
mailbox_element_ptr ptr;
while (consumed < max_throughput) {
CAF_LOG_DEBUG("start new DRR round");
mailbox_.fetch_more();
auto prev = consumed; // Caches the value before processing more.
// TODO: maybe replace '3' with configurable / adaptive value?
static constexpr size_t quantum = 3;
// Dispatch urgent and normal (asynchronous) messages.
auto& hq = get_urgent_queue();
auto& nq = get_normal_queue();
if (hq.new_round(quantum * 3, handle_async).consumed_items > 0) {
// After matching any message, all caches must be re-evaluated.
nq.flush_cache();
}
if (nq.new_round(quantum, handle_async).consumed_items > 0) {
// After matching any message, all caches must be re-evaluated.
hq.flush_cache();
}
// Update metrics or try returning if the actor consumed nothing.
auto delta = consumed - prev;
CAF_LOG_DEBUG("consumed" << delta << "messages this round");
if (delta > 0) {
auto signed_val = static_cast<int64_t>(delta);
home_system().base_metrics().processed_messages->inc(signed_val);
} else {
auto ptr = mailbox().pop_front();
if (!ptr) {
if (mailbox().try_block()) {
reset_timeouts_if_needed();
if (mailbox().try_block())
CAF_LOG_DEBUG("mailbox empty: await new messages");
return resumable::awaiting_message;
CAF_LOG_DEBUG("mailbox().try_block() returned false");
}
CAF_LOG_DEBUG("check for shutdown");
if (finalize())
continue; // Interrupted by a new message, try again.
}
auto res = run_with_metrics(*ptr, [this, &ptr, &consumed] {
auto res = reactivate(*ptr);
switch (res) {
case activation_result::success:
++consumed;
unstash();
break;
case activation_result::skipped:
stash_.push(ptr.release());
break;
default: // drop
break;
}
return res;
});
if (res == activation_result::terminated)
return resumable::done;
}
CAF_LOG_DEBUG("max throughput reached");
reset_timeouts_if_needed();
if (mailbox().try_block())
if (mailbox().try_block()) {
CAF_LOG_DEBUG("mailbox empty: await new messages");
return resumable::awaiting_message;
}
// time's up
CAF_LOG_DEBUG("max throughput reached: resume later");
return resumable::resume_later;
}
......@@ -890,25 +873,7 @@ bool scheduled_actor::finalize() {
}
void scheduled_actor::push_to_cache(mailbox_element_ptr ptr) {
using namespace intrusive;
auto& p = mailbox_.queue().policy();
auto& qs = mailbox_.queue().queues();
auto push = [&ptr](auto& q) {
q.inc_total_task_size(q.policy().task_size(*ptr));
q.cache().push_back(ptr.release());
};
if (p.id_of(*ptr) == normal_queue_index)
push(std::get<normal_queue_index>(qs));
else
push(std::get<urgent_queue_index>(qs));
}
scheduled_actor::urgent_queue& scheduled_actor::get_urgent_queue() {
return get<urgent_queue_index>(mailbox_.queue().queues());
}
scheduled_actor::normal_queue& scheduled_actor::get_normal_queue() {
return get<normal_queue_index>(mailbox_.queue().queues());
stash_.push(ptr.release());
}
disposable scheduled_actor::run_scheduled(timestamp when, action what) {
......@@ -986,6 +951,7 @@ void scheduled_actor::deregister_stream(uint64_t stream_id) {
}
void scheduled_actor::run_actions() {
CAF_LOG_TRACE("");
delayed_actions_this_run_ = 0;
if (!actions_.empty()) {
// Note: can't use iterators here since actions may add to the vector.
......@@ -1020,4 +986,9 @@ void scheduled_actor::try_push_stream(uint64_t local_id) {
i->second->push();
}
void scheduled_actor::unstash() {
while (auto stashed = stash_.pop())
mailbox().push_front(mailbox_element_ptr{stashed});
}
} // namespace caf
......@@ -10,12 +10,14 @@
# include <exception>
#endif // CAF_ENABLE_EXCEPTIONS
#include "caf/abstract_mailbox.hpp"
#include "caf/action.hpp"
#include "caf/actor_traits.hpp"
#include "caf/async/fwd.hpp"
#include "caf/cow_string.hpp"
#include "caf/detail/behavior_stack.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/default_mailbox.hpp"
#include "caf/detail/stream_bridge.hpp"
#include "caf/disposable.hpp"
#include "caf/error.hpp"
......@@ -26,11 +28,7 @@
#include "caf/flow/multicaster.hpp"
#include "caf/flow/observer.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_cached_queue.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/intrusive/stack.hpp"
#include "caf/invoke_message_result.hpp"
#include "caf/local_actor.hpp"
#include "caf/logger.hpp"
......@@ -39,9 +37,6 @@
#include "caf/mixin/sender.hpp"
#include "caf/no_stages.hpp"
#include "caf/policy/arg.hpp"
#include "caf/policy/categorized.hpp"
#include "caf/policy/normal_messages.hpp"
#include "caf/policy/urgent_messages.hpp"
#include "caf/response_handle.hpp"
#include "caf/sec.hpp"
#include "caf/telemetry/timer.hpp"
......@@ -118,26 +113,6 @@ public:
/// Base type.
using super = local_actor;
/// Stores asynchronous messages with default priority.
using normal_queue = intrusive::drr_cached_queue<policy::normal_messages>;
/// Stores asynchronous messages with hifh priority.
using urgent_queue = intrusive::drr_cached_queue<policy::urgent_messages>;
/// Configures the FIFO inbox with two nested queues. One for high-priority
/// and one for normal priority 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 batch_op_ptr = intrusive_ptr<flow::op::base<async::batch>>;
struct stream_source_state {
......@@ -145,13 +120,6 @@ public:
size_t max_items_per_batch;
};
static constexpr size_t urgent_queue_index = 0;
static constexpr size_t normal_queue_index = 1;
/// A queue optimized for single-reader-many-writers.
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
/// The message ID of an outstanding response with its callback.
using pending_response = std::pair<const message_id, behavior>;
......@@ -269,8 +237,8 @@ public:
// -- properties -------------------------------------------------------------
/// Returns the queue for storing incoming messages.
mailbox_type& mailbox() noexcept {
return mailbox_;
abstract_mailbox& mailbox() noexcept {
return *mailbox_;
}
// -- event handlers ---------------------------------------------------------
......@@ -443,12 +411,6 @@ public:
/// Pushes `ptr` to the cache of the default queue.
void push_to_cache(mailbox_element_ptr ptr);
/// Returns the queue of the mailbox that stores high priority messages.
urgent_queue& get_urgent_queue();
/// Returns the default queue of the mailbox that stores ordinary messages.
normal_queue& get_normal_queue();
// -- caf::flow API ----------------------------------------------------------
steady_time_point steady_time() override;
......@@ -722,7 +684,7 @@ protected:
// -- member variables -------------------------------------------------------
/// Stores incoming messages.
mailbox_type mailbox_;
abstract_mailbox* mailbox_;
/// Stores user-defined callbacks for message handling.
detail::behavior_stack bhvr_stack_;
......@@ -762,13 +724,16 @@ protected:
private:
// -- utilities for instrumenting actors -------------------------------------
/// Places all messages from the `stash_` back into the mailbox.
void unstash();
template <class F>
intrusive::task_result run_with_metrics(mailbox_element& x, F body) {
activation_result run_with_metrics(mailbox_element& x, F body) {
if (metrics_.mailbox_time) {
auto t0 = std::chrono::steady_clock::now();
auto mbox_time = x.seconds_until(t0);
auto res = body();
if (res != intrusive::task_result::skip) {
if (res != activation_result::skipped) {
telemetry::timer::observe(metrics_.processing_time, t0);
metrics_.mailbox_time->observe(mbox_time);
metrics_.mailbox_size->dec();
......@@ -852,6 +817,15 @@ private:
/// This is to make sure that actor does not terminate because it thinks it's
/// done before processing the delayed action.
behavior delay_bhvr_;
/// Stashes skipped messages until the actor processes the next message.
intrusive::stack<mailbox_element> stash_;
union {
/// The default mailbox instance that we use if the user does not configure
/// a mailbox via the ::actor_config.
detail::default_mailbox default_mailbox_;
};
};
} // namespace caf
......@@ -48,7 +48,7 @@ public:
/// Peeks into the mailbox of `next_job<scheduled_actor>()`.
template <class... Ts>
decltype(auto) peek() {
auto ptr = next_job<scheduled_actor>().mailbox().peek();
auto ptr = next_job<scheduled_actor>().peek_at_next_mailbox_element();
CAF_ASSERT(ptr != nullptr);
if (auto view = make_const_typed_message_view<Ts...>(ptr->payload)) {
if constexpr (sizeof...(Ts) == 1)
......
// 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.lifo_inbox
#include "caf/intrusive/lifo_inbox.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 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(inbox_type::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
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(), "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);
CHECK_EQ(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();
}
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.task_queue
#include "caf/intrusive/task_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
}
};
[[maybe_unused]] 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
BEGIN_FIXTURE_SCOPE(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_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) {
CHECK_EQ(queue.peek(), nullptr);
fill(queue, 1, 2, 3);
CHECK_EQ(queue.peek()->value, 1);
}
CAF_TEST(task_size) {
fill(queue, 1, 2, 3);
CHECK_EQ(queue.total_task_size(), 6);
fill(queue, 4, 5);
CHECK_EQ(queue.total_task_size(), 15);
queue.clear();
CHECK_EQ(queue.total_task_size(), 0);
}
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.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()
This diff is collapsed.
This diff is collapsed.
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