Commit 795f8ad6 authored by Dominik Charousset's avatar Dominik Charousset

Add a new interface for mailboxes

parent dabbb358
...@@ -70,6 +70,7 @@ caf_add_component( ...@@ -70,6 +70,7 @@ caf_add_component(
caf/abstract_actor.cpp caf/abstract_actor.cpp
caf/abstract_channel.cpp caf/abstract_channel.cpp
caf/abstract_group.cpp caf/abstract_group.cpp
caf/abstract_mailbox.cpp
caf/action.cpp caf/action.cpp
caf/actor.cpp caf/actor.cpp
caf/actor_addr.cpp caf/actor_addr.cpp
...@@ -111,6 +112,8 @@ caf_add_component( ...@@ -111,6 +112,8 @@ caf_add_component(
caf/detail/behavior_stack.cpp caf/detail/behavior_stack.cpp
caf/detail/blocking_behavior.cpp caf/detail/blocking_behavior.cpp
caf/detail/config_consumer.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/get_process_id.cpp
caf/detail/glob_match.cpp caf/detail/glob_match.cpp
caf/detail/group_tunnel.cpp caf/detail/group_tunnel.cpp
...@@ -156,6 +159,7 @@ caf_add_component( ...@@ -156,6 +159,7 @@ caf_add_component(
caf/hash/fnv.test.cpp caf/hash/fnv.test.cpp
caf/hash/sha1.cpp caf/hash/sha1.cpp
caf/init_global_meta_objects.cpp caf/init_global_meta_objects.cpp
caf/intrusive/stack.test.cpp
caf/ipv4_address.cpp caf/ipv4_address.cpp
caf/ipv4_endpoint.cpp caf/ipv4_endpoint.cpp
caf/ipv4_subnet.cpp caf/ipv4_subnet.cpp
......
// 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/fwd.hpp"
#include "caf/intrusive/inbox_result.hpp"
#include "caf/mailbox_element.hpp"
namespace caf {
/// The base class for all mailbox implementations.
class 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;
/// Called by the testing DSL to peek at the next element in the mailbox.
/// @returns A pointer to the next mailbox element or `nullptr` if the
/// mailbox is empty or does not support peeking.
/// @note Only the owning actor is allowed to call this function.
virtual mailbox_element* peek(message_id id = make_message_id()) = 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() = 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() = 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;
/// 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;
};
} // 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() {
return inbox_.closed();
}
bool default_mailbox::blocked() {
return inbox_.blocked();
}
bool default_mailbox::try_block() {
return cached() == 0 && inbox_.try_block();
}
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;
do {
auto next = head->next;
auto phead = promote(head);
if (phead->mid.is_urgent_message())
urgent_queue_.lifo_append(phead);
else
normal_queue_.lifo_append(phead);
head = next;
} while (head != nullptr);
urgent_queue_.stop_lifo_append();
normal_queue_.stop_lifo_append();
return true;
}
} // 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/intrusive/fifo_inbox.hpp"
#include "caf/intrusive/task_queue.hpp"
#include "caf/intrusive/wdrr_fixed_multiplexed_queue.hpp"
#include "caf/policy/categorized.hpp"
#include "caf/policy/normal_messages.hpp"
#include "caf/policy/urgent_messages.hpp"
namespace caf::detail {
class default_mailbox : public abstract_mailbox {
public:
struct policy {
using mapped_type = mailbox_element;
using task_size_type = size_t;
using deficit_type = size_t;
using unique_pointer = mailbox_element_ptr;
static size_t task_size(const mapped_type&) noexcept {
return 1;
}
};
default_mailbox() = default;
default_mailbox(const default_mailbox&) = delete;
default_mailbox& operator=(const default_mailbox&) = delete;
intrusive::inbox_result push_back(mailbox_element_ptr ptr) override;
void push_front(mailbox_element_ptr ptr) override;
mailbox_element* peek(message_id id) override;
mailbox_element_ptr pop_front() override;
bool closed() override;
bool blocked() override;
bool try_block() override;
size_t close(const error&) override;
size_t size() override;
private:
/// Returns the total number of elements stored in the queues.
size_t cached() const noexcept {
return urgent_queue_.total_task_size() + normal_queue_.total_task_size();
}
/// Tries to fetch more messages from the LIFO inbox.
bool fetch_more();
/// Stores incoming messages in LIFO order.
alignas(CAF_CACHE_LINE_SIZE) intrusive::lifo_inbox<policy> inbox_;
/// Stores urgent messages in FIFO order.
intrusive::task_queue<policy> urgent_queue_;
/// Stores normal messages in FIFO order.
intrusive::task_queue<policy> normal_queue_;
};
} // 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()
...@@ -160,7 +160,7 @@ public: ...@@ -160,7 +160,7 @@ public:
/// `f` must take ownership of the passed pointer. /// `f` must take ownership of the passed pointer.
/// @warning Call only from the reader (owner). /// @warning Call only from the reader (owner).
template <class F> template <class F>
void close(F& f) noexcept(noexcept(f(std::declval<pointer>()))) { void close(F&& f) noexcept(noexcept(f(std::declval<pointer>()))) {
node_pointer ptr = take_head(stack_closed_tag()); node_pointer ptr = take_head(stack_closed_tag());
while (ptr != nullptr) { while (ptr != nullptr) {
auto next = ptr->next; auto next = ptr->next;
......
// 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
namespace caf::intrusive {
template <class T>
class stack {
public:
using pointer = T*;
stack() = default;
stack(const stack&) = delete;
stack& operator=(const stack&) = delete;
~stack() {
while (head_) {
auto next = static_cast<pointer>(head_->next);
delete head_;
head_ = next;
}
}
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()
...@@ -52,7 +52,11 @@ public: ...@@ -52,7 +52,11 @@ public:
// -- constructors, destructors, and assignment operators ------------------- // -- constructors, destructors, and assignment operators -------------------
task_queue(policy_type p) task_queue() : old_last_(nullptr), new_head_(nullptr) {
init();
}
explicit task_queue(policy_type p)
: old_last_(nullptr), new_head_(nullptr), policy_(std::move(p)) { : old_last_(nullptr), new_head_(nullptr), policy_(std::move(p)) {
init(); init();
} }
...@@ -190,6 +194,24 @@ public: ...@@ -190,6 +194,24 @@ public:
return result; return result;
} }
/// Removes the first element from the queue and returns it.
unique_pointer pop_front() {
unique_pointer result;
if (!empty()) {
auto ptr = promote(head_.next);
auto ts = policy_.task_size(*ptr);
CAF_ASSERT(ts > 0);
total_task_size_ -= ts;
head_.next = ptr->next;
if (total_task_size_ == 0) {
CAF_ASSERT(head_.next == &(tail_));
tail_.next = &(head_);
}
result.reset(ptr);
}
return result;
}
// -- iterator access -------------------------------------------------------- // -- iterator access --------------------------------------------------------
/// Returns an iterator to the dummy before the first element. /// Returns an iterator to the dummy before the first element.
...@@ -260,6 +282,21 @@ public: ...@@ -260,6 +282,21 @@ public:
return push_back(new value_type(std::forward<Ts>(xs)...)); return push_back(new value_type(std::forward<Ts>(xs)...));
} }
void push_front(pointer ptr) noexcept {
CAF_ASSERT(ptr != nullptr);
if (empty()) {
push_back(ptr);
return;
}
ptr->next = head_.next;
head_.next = ptr;
inc_total_task_size(*ptr);
}
void push_front(unique_pointer ptr) noexcept {
push_front(ptr.release());
}
/// Transfers all element from `other` to the front of this queue. /// Transfers all element from `other` to the front of this queue.
template <class Container> template <class Container>
void prepend(Container& other) { void prepend(Container& other) {
...@@ -328,16 +365,22 @@ public: ...@@ -328,16 +365,22 @@ public:
total_task_size_ = 0; total_task_size_ = 0;
} }
/// Deletes all elements. template <class F>
/// @warning leaves the queue in an invalid state until calling `init` again. void drain(F fn) {
/// @private
void deinit() noexcept {
for (auto i = head_.next; i != &tail_;) { for (auto i = head_.next; i != &tail_;) {
auto ptr = i; auto ptr = i;
i = i->next; i = i->next;
typename unique_pointer::deleter_type d; fn(promote(ptr));
d(promote(ptr)); }
init();
} }
/// Deletes all elements.
/// @warning leaves the queue in an invalid state until calling `init` again.
/// @private
void deinit() noexcept {
typename unique_pointer::deleter_type fn;
drain(fn);
} }
protected: protected:
......
...@@ -118,7 +118,6 @@ scheduled_actor::batch_forwarder::~batch_forwarder() { ...@@ -118,7 +118,6 @@ scheduled_actor::batch_forwarder::~batch_forwarder() {
scheduled_actor::scheduled_actor(actor_config& cfg) scheduled_actor::scheduled_actor(actor_config& cfg)
: super(cfg), : super(cfg),
mailbox_(unit, unit, unit),
default_handler_(print_and_drop), default_handler_(print_and_drop),
error_handler_(default_error_handler), error_handler_(default_error_handler),
down_handler_(default_down_handler), down_handler_(default_down_handler),
...@@ -182,15 +181,9 @@ bool scheduled_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) { ...@@ -182,15 +181,9 @@ bool scheduled_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) {
} }
mailbox_element* scheduled_actor::peek_at_next_mailbox_element() { mailbox_element* scheduled_actor::peek_at_next_mailbox_element() {
if (mailbox().closed() || mailbox().blocked()) { return mailbox().peek(awaited_responses_.empty()
return nullptr; ? make_message_id()
} else if (awaited_responses_.empty()) { : awaited_responses_.begin()->first);
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);
}
} }
// -- overridden functions of local_actor -------------------------------------- // -- overridden functions of local_actor --------------------------------------
...@@ -238,19 +231,11 @@ bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) { ...@@ -238,19 +231,11 @@ bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) {
run_actions(); run_actions();
} }
// Clear mailbox. // Clear mailbox.
if (!mailbox_.closed()) { if (!mailbox().closed()) {
mailbox_.close(); unstash();
get_normal_queue().flush_cache(); auto dropped = mailbox().close(fail_state);
get_urgent_queue().flush_cache(); if (dropped > 0 && metrics_.mailbox_size)
detail::sync_request_bouncer bounce{fail_state}; metrics_.mailbox_size->dec(static_cast<int64_t>(dropped));
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;
}
} }
// Dispatch to parent's `cleanup` function. // Dispatch to parent's `cleanup` function.
return super::cleanup(std::move(fail_state), host); return super::cleanup(std::move(fail_state), host);
...@@ -277,66 +262,54 @@ resumable::resume_result scheduled_actor::resume(execution_unit* ctx, ...@@ -277,66 +262,54 @@ resumable::resume_result scheduled_actor::resume(execution_unit* ctx,
if (!activate(ctx)) if (!activate(ctx))
return resumable::done; return resumable::done;
size_t consumed = 0; 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 = [&] { auto reset_timeouts_if_needed = [&] {
// Set a new receive timeout if we called our behavior at least once. // Set a new receive timeout if we called our behavior at least once.
if (consumed > 0) if (consumed > 0)
set_receive_timeout(); set_receive_timeout();
}; };
// Callback for handling urgent and normal messages. mailbox_element_ptr ptr;
auto handle_async = [this, max_throughput, &consumed](mailbox_element& x) { while (consumed < max_throughput) {
return run_with_metrics(x, [this, max_throughput, &consumed, &x] { auto ptr = mailbox().pop_front();
switch (reactivate(x)) { if (!ptr) {
if (mailbox().try_block()) {
reset_timeouts_if_needed();
CAF_LOG_DEBUG("mailbox empty: await new messages");
return resumable::awaiting_message;
}
continue; // Interrupted by a new message, try again.
}
auto res = run_with_metrics(*ptr, [this, &ptr, &consumed] {
switch (reactivate(*ptr)) {
case activation_result::success:
++consumed;
unstash();
return intrusive::task_result::resume;
case activation_result::terminated: case activation_result::terminated:
return intrusive::task_result::stop; 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: case activation_result::skipped:
stash_.push(ptr.release());
return intrusive::task_result::skip; return intrusive::task_result::skip;
default: default: // drop
return intrusive::task_result::resume; return intrusive::task_result::resume;
} }
}); });
}; if (res == intrusive::task_result::stop)
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 {
reset_timeouts_if_needed();
if (mailbox().try_block())
return resumable::awaiting_message;
CAF_LOG_DEBUG("mailbox().try_block() returned false");
}
CAF_LOG_DEBUG("check for shutdown");
if (finalize())
return resumable::done; return resumable::done;
} }
CAF_LOG_DEBUG("max throughput reached");
reset_timeouts_if_needed(); reset_timeouts_if_needed();
if (mailbox().try_block()) if (mailbox().try_block()) {
CAF_LOG_DEBUG("mailbox empty: await new messages");
return resumable::awaiting_message; return resumable::awaiting_message;
}
// time's up // time's up
CAF_LOG_DEBUG("max throughput reached: resume later");
return resumable::resume_later; return resumable::resume_later;
} }
...@@ -890,25 +863,7 @@ bool scheduled_actor::finalize() { ...@@ -890,25 +863,7 @@ bool scheduled_actor::finalize() {
} }
void scheduled_actor::push_to_cache(mailbox_element_ptr ptr) { void scheduled_actor::push_to_cache(mailbox_element_ptr ptr) {
using namespace intrusive; stash_.push(ptr.release());
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());
} }
disposable scheduled_actor::run_scheduled(timestamp when, action what) { disposable scheduled_actor::run_scheduled(timestamp when, action what) {
...@@ -986,6 +941,7 @@ void scheduled_actor::deregister_stream(uint64_t stream_id) { ...@@ -986,6 +941,7 @@ void scheduled_actor::deregister_stream(uint64_t stream_id) {
} }
void scheduled_actor::run_actions() { void scheduled_actor::run_actions() {
CAF_LOG_TRACE("");
delayed_actions_this_run_ = 0; delayed_actions_this_run_ = 0;
if (!actions_.empty()) { if (!actions_.empty()) {
// Note: can't use iterators here since actions may add to the vector. // Note: can't use iterators here since actions may add to the vector.
...@@ -1020,4 +976,9 @@ void scheduled_actor::try_push_stream(uint64_t local_id) { ...@@ -1020,4 +976,9 @@ void scheduled_actor::try_push_stream(uint64_t local_id) {
i->second->push(); i->second->push();
} }
void scheduled_actor::unstash() {
while (auto stashed = stash_.pop())
mailbox().push_front(mailbox_element_ptr{stashed});
}
} // namespace caf } // namespace caf
...@@ -10,12 +10,14 @@ ...@@ -10,12 +10,14 @@
# include <exception> # include <exception>
#endif // CAF_ENABLE_EXCEPTIONS #endif // CAF_ENABLE_EXCEPTIONS
#include "caf/abstract_mailbox.hpp"
#include "caf/action.hpp" #include "caf/action.hpp"
#include "caf/actor_traits.hpp" #include "caf/actor_traits.hpp"
#include "caf/async/fwd.hpp" #include "caf/async/fwd.hpp"
#include "caf/cow_string.hpp" #include "caf/cow_string.hpp"
#include "caf/detail/behavior_stack.hpp" #include "caf/detail/behavior_stack.hpp"
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/detail/default_mailbox.hpp"
#include "caf/detail/stream_bridge.hpp" #include "caf/detail/stream_bridge.hpp"
#include "caf/disposable.hpp" #include "caf/disposable.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
...@@ -26,11 +28,7 @@ ...@@ -26,11 +28,7 @@
#include "caf/flow/multicaster.hpp" #include "caf/flow/multicaster.hpp"
#include "caf/flow/observer.hpp" #include "caf/flow/observer.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/intrusive/drr_cached_queue.hpp" #include "caf/intrusive/stack.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/invoke_message_result.hpp" #include "caf/invoke_message_result.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
...@@ -39,9 +37,6 @@ ...@@ -39,9 +37,6 @@
#include "caf/mixin/sender.hpp" #include "caf/mixin/sender.hpp"
#include "caf/no_stages.hpp" #include "caf/no_stages.hpp"
#include "caf/policy/arg.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/response_handle.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/telemetry/timer.hpp" #include "caf/telemetry/timer.hpp"
...@@ -118,26 +113,6 @@ public: ...@@ -118,26 +113,6 @@ public:
/// Base type. /// Base type.
using super = local_actor; 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>>; using batch_op_ptr = intrusive_ptr<flow::op::base<async::batch>>;
struct stream_source_state { struct stream_source_state {
...@@ -145,12 +120,8 @@ public: ...@@ -145,12 +120,8 @@ public:
size_t max_items_per_batch; 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. /// A queue optimized for single-reader-many-writers.
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>; using mailbox_type = detail::default_mailbox;
/// The message ID of an outstanding response with its callback. /// The message ID of an outstanding response with its callback.
using pending_response = std::pair<const message_id, behavior>; using pending_response = std::pair<const message_id, behavior>;
...@@ -269,7 +240,7 @@ public: ...@@ -269,7 +240,7 @@ public:
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
/// Returns the queue for storing incoming messages. /// Returns the queue for storing incoming messages.
mailbox_type& mailbox() noexcept { abstract_mailbox& mailbox() noexcept {
return mailbox_; return mailbox_;
} }
...@@ -443,12 +414,6 @@ public: ...@@ -443,12 +414,6 @@ public:
/// Pushes `ptr` to the cache of the default queue. /// Pushes `ptr` to the cache of the default queue.
void push_to_cache(mailbox_element_ptr ptr); 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 ---------------------------------------------------------- // -- caf::flow API ----------------------------------------------------------
steady_time_point steady_time() override; steady_time_point steady_time() override;
...@@ -762,6 +727,9 @@ protected: ...@@ -762,6 +727,9 @@ protected:
private: private:
// -- utilities for instrumenting actors ------------------------------------- // -- utilities for instrumenting actors -------------------------------------
/// Places all messages from the `stash_` back into the mailbox.
void unstash();
template <class F> template <class F>
intrusive::task_result run_with_metrics(mailbox_element& x, F body) { intrusive::task_result run_with_metrics(mailbox_element& x, F body) {
if (metrics_.mailbox_time) { if (metrics_.mailbox_time) {
...@@ -852,6 +820,9 @@ private: ...@@ -852,6 +820,9 @@ private:
/// This is to make sure that actor does not terminate because it thinks it's /// This is to make sure that actor does not terminate because it thinks it's
/// done before processing the delayed action. /// done before processing the delayed action.
behavior delay_bhvr_; behavior delay_bhvr_;
/// Stashes skipped messages until the actor processes the next message.
intrusive::stack<mailbox_element> stash_;
}; };
} // namespace caf } // namespace caf
...@@ -48,7 +48,7 @@ public: ...@@ -48,7 +48,7 @@ public:
/// Peeks into the mailbox of `next_job<scheduled_actor>()`. /// Peeks into the mailbox of `next_job<scheduled_actor>()`.
template <class... Ts> template <class... Ts>
decltype(auto) peek() { 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); CAF_ASSERT(ptr != nullptr);
if (auto view = make_const_typed_message_view<Ts...>(ptr->payload)) { if (auto view = make_const_typed_message_view<Ts...>(ptr->payload)) {
if constexpr (sizeof...(Ts) == 1) if constexpr (sizeof...(Ts) == 1)
......
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