Commit a3cfcbd2 authored by Dominik Charousset's avatar Dominik Charousset

Use the new mailbox type in blocking actors

parent 795f8ad6
...@@ -27,12 +27,6 @@ public: ...@@ -27,12 +27,6 @@ public:
/// @note Only the owning actor is allowed to call this function. /// @note Only the owning actor is allowed to call this function.
virtual void push_front(mailbox_element_ptr ptr) = 0; 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. /// Removes the next element from the mailbox.
/// @returns The next element in the mailbox or `nullptr` if the mailbox is /// @returns The next element in the mailbox or `nullptr` if the mailbox is
/// empty. /// empty.
...@@ -52,6 +46,10 @@ public: ...@@ -52,6 +46,10 @@ public:
/// @note Only the owning actor is allowed to call this function. /// @note Only the owning actor is allowed to call this function.
virtual bool try_block() = 0; 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. /// Closes the mailbox and discards all pending messages.
/// @returns The number of dropped messages. /// @returns The number of dropped messages.
/// @note Only the owning actor is allowed to call this function. /// @note Only the owning actor is allowed to call this function.
...@@ -60,6 +58,11 @@ public: ...@@ -60,6 +58,11 @@ public:
/// Returns the number of pending messages. /// Returns the number of pending messages.
/// @note Only the owning actor is allowed to call this function. /// @note Only the owning actor is allowed to call this function.
virtual size_t size() = 0; virtual size_t size() = 0;
/// Checks whether the mailbox is empty.
bool empty() {
return size() == 0;
}
}; };
} // namespace caf } // namespace caf
...@@ -41,8 +41,7 @@ bool blocking_actor::accept_one_cond::post() { ...@@ -41,8 +41,7 @@ bool blocking_actor::accept_one_cond::post() {
} }
blocking_actor::blocking_actor(actor_config& cfg) blocking_actor::blocking_actor(actor_config& cfg)
: super(cfg.add_flag(local_actor::is_blocking_flag)), : super(cfg.add_flag(local_actor::is_blocking_flag)) {
mailbox_(unit, unit, unit) {
// nop // nop
} }
...@@ -63,24 +62,32 @@ bool blocking_actor::enqueue(mailbox_element_ptr ptr, execution_unit*) { ...@@ -63,24 +62,32 @@ bool blocking_actor::enqueue(mailbox_element_ptr ptr, execution_unit*) {
metrics_.mailbox_size->inc(); metrics_.mailbox_size->inc();
} }
// returns false if mailbox has been closed // returns false if mailbox has been closed
if (!mailbox().synchronized_push_back(mtx_, cv_, std::move(ptr))) { switch (mailbox().push_back(std::move(ptr))) {
CAF_LOG_REJECT_EVENT(); case intrusive::inbox_result::queue_closed: {
home_system().base_metrics().rejected_messages->inc(); CAF_LOG_REJECT_EVENT();
if (collects_metrics) home_system().base_metrics().rejected_messages->inc();
metrics_.mailbox_size->dec(); if (collects_metrics)
if (mid.is_request()) { metrics_.mailbox_size->dec();
detail::sync_request_bouncer srb{exit_reason()}; if (mid.is_request()) {
srb(src, mid); detail::sync_request_bouncer srb{exit_reason()};
srb(src, mid);
}
return false;
} }
return false; case intrusive::inbox_result::unblocked_reader: {
} else { CAF_LOG_ACCEPT_EVENT(true);
CAF_LOG_ACCEPT_EVENT(false); std::unique_lock guard{mtx_};
return true; cv_.notify_one();
return true;
}
default:
CAF_LOG_ACCEPT_EVENT(false);
return true;
} }
} }
mailbox_element* blocking_actor::peek_at_next_mailbox_element() { mailbox_element* blocking_actor::peek_at_next_mailbox_element() {
return mailbox().closed() || mailbox().blocked() ? nullptr : mailbox().peek(); return mailbox_.peek(make_message_id());
} }
const char* blocking_actor::name() const { const char* blocking_actor::name() const {
...@@ -197,101 +204,37 @@ void blocking_actor::fail_state(error err) { ...@@ -197,101 +204,37 @@ void blocking_actor::fail_state(error err) {
fail_state_ = std::move(err); fail_state_ = std::move(err);
} }
intrusive::task_result void blocking_actor::receive_impl(receive_cond& rcc, message_id mid,
blocking_actor::mailbox_visitor::operator()(mailbox_element& x) { detail::blocking_behavior& bhvr) {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(mid));
CAF_LOG_RECEIVE_EVENT((&x)); unstash();
CAF_BEFORE_PROCESSING(self, x); // Convenience function for trying to consume a message.
// Wrap the actual body for the function. auto consume = [this, mid, &bhvr] {
auto body = [this, &x] { detail::default_invoke_result_visitor<blocking_actor> visitor{this};
auto check_if_done = [&]() -> intrusive::task_result { if (bhvr.nested(visitor, current_element_->content()))
// Stop consuming items when reaching the end of the user-defined receive return true;
// loop either via post or pre condition. auto sres = bhvr.fallback(current_element_->payload);
if (rcc.post() && rcc.pre())
return intrusive::task_result::resume;
done = true;
return intrusive::task_result::stop;
};
// Skip messages that don't match our message ID.
if (mid.is_response()) {
if (mid != x.mid) {
return intrusive::task_result::skip;
}
} else if (x.mid.is_response()) {
return intrusive::task_result::skip;
}
// Automatically unlink from actors after receiving an exit.
if (auto view = make_const_typed_message_view<exit_msg>(x.content()))
self->unlink_from(get<0>(view).source);
// Blocking actors can nest receives => push/pop `current_element_`
auto prev_element = self->current_element_;
self->current_element_ = &x;
auto g = detail::make_scope_guard(
[&] { self->current_element_ = prev_element; });
// Dispatch on x.
detail::default_invoke_result_visitor<blocking_actor> visitor{self};
if (bhvr.nested(visitor, x.content()))
return check_if_done();
// Blocking actors can have fallback handlers for catch-all rules.
auto sres = bhvr.fallback(self->current_element_->payload);
auto f = detail::make_overload( auto f = detail::make_overload(
[&](skip_t&) { [this, mid, &bhvr](skip_t&) {
// Response handlers must get re-invoked with an error when // Response handlers must get re-invoked with an error when
// receiving an unexpected message. // receiving an unexpected message.
if (mid.is_response()) { if (mid.is_response()) {
auto& x = *current_element_;
auto err = make_error(sec::unexpected_response, std::move(x.payload)); auto err = make_error(sec::unexpected_response, std::move(x.payload));
mailbox_element tmp{std::move(x.sender), x.mid, std::move(x.stages), mailbox_element tmp{std::move(x.sender), x.mid, std::move(x.stages),
make_message(std::move(err))}; make_message(std::move(err))};
self->current_element_ = &tmp; current_element_ = &tmp;
bhvr.nested(tmp.content()); bhvr.nested(tmp.content());
return check_if_done(); return true;
} }
return intrusive::task_result::skip; return false;
}, },
[&](auto& res) { [&visitor](auto& res) {
visitor(res); visitor(res);
return check_if_done(); return true;
}); });
return visit(f, sres); return visit(f, sres);
}; };
// Post-process the returned value from the function body.
if (!self->getf(abstract_actor::collects_metrics_flag)) {
auto result = body();
if (result == intrusive::task_result::skip) {
CAF_AFTER_PROCESSING(self, invoke_message_result::skipped);
CAF_LOG_SKIP_EVENT();
} else {
CAF_AFTER_PROCESSING(self, invoke_message_result::consumed);
CAF_LOG_FINALIZE_EVENT();
}
return result;
} else {
auto t0 = std::chrono::steady_clock::now();
auto mbox_time = x.seconds_until(t0);
auto result = body();
if (result == intrusive::task_result::skip) {
CAF_AFTER_PROCESSING(self, invoke_message_result::skipped);
CAF_LOG_SKIP_EVENT();
auto& builtins = self->builtin_metrics();
telemetry::timer::observe(builtins.processing_time, t0);
builtins.mailbox_time->observe(mbox_time);
builtins.mailbox_size->dec();
} else {
CAF_AFTER_PROCESSING(self, invoke_message_result::consumed);
CAF_LOG_FINALIZE_EVENT();
}
return result;
}
}
void blocking_actor::receive_impl(receive_cond& rcc, message_id mid,
detail::blocking_behavior& bhvr) {
CAF_LOG_TRACE(CAF_ARG(mid));
// Set to `true` by the visitor when done.
bool done = false;
// Make sure each receive sees all mailbox elements.
mailbox_visitor f{this, done, rcc, mid, bhvr};
mailbox().flush_cache();
// Check pre-condition once before entering the message consumption loop. The // Check pre-condition once before entering the message consumption loop. The
// consumer performs any future check on pre and post conditions via // consumer performs any future check on pre and post conditions via
// check_if_done. // check_if_done.
...@@ -299,7 +242,7 @@ void blocking_actor::receive_impl(receive_cond& rcc, message_id mid, ...@@ -299,7 +242,7 @@ void blocking_actor::receive_impl(receive_cond& rcc, message_id mid,
return; return;
// Read incoming messages for as long as the user's receive loop accepts more // Read incoming messages for as long as the user's receive loop accepts more
// messages. // messages.
do { for (;;) {
// Reset the timeout each iteration. // Reset the timeout each iteration.
auto rel_tout = bhvr.timeout(); auto rel_tout = bhvr.timeout();
if (rel_tout == infinite) { if (rel_tout == infinite) {
...@@ -316,28 +259,78 @@ void blocking_actor::receive_impl(receive_cond& rcc, message_id mid, ...@@ -316,28 +259,78 @@ void blocking_actor::receive_impl(receive_cond& rcc, message_id mid,
return; return;
} }
} }
mailbox_.new_round(3, f); // Fetch next message from our mailbox.
} while (!done); auto ptr = mailbox().pop_front();
auto t0 = std::chrono::steady_clock::now();
auto mbox_time = ptr->seconds_until(t0);
// Skip messages that don't match our message ID.
if (mid.is_response()) {
if (mid != ptr->mid) {
stash_.push(ptr.release());
continue;
}
} else if (ptr->mid.is_response()) {
stash_.push(ptr.release());
continue;
}
// Automatically unlink from actors after receiving an exit.
if (auto view = make_const_typed_message_view<exit_msg>(ptr->content()))
unlink_from(get<0>(view).source);
// Blocking actors can nest receives => push/pop `current_element_`
auto prev_element = current_element_;
current_element_ = ptr.get();
auto g = detail::make_scope_guard([&] { current_element_ = prev_element; });
// Dispatch on the current mailbox element.
if (consume()) {
unstash();
CAF_AFTER_PROCESSING(self, invoke_message_result::consumed);
CAF_LOG_FINALIZE_EVENT();
if (getf(abstract_actor::collects_metrics_flag)) {
auto& builtins = builtin_metrics();
telemetry::timer::observe(builtins.processing_time, t0);
builtins.mailbox_time->observe(mbox_time);
builtins.mailbox_size->dec();
}
// Check whether we are done.
if (!rcc.post() || !rcc.pre()) {
return;
}
continue;
}
// Message was skipped.
CAF_AFTER_PROCESSING(self, invoke_message_result::skipped);
CAF_LOG_SKIP_EVENT();
stash_.push(ptr.release());
}
} }
void blocking_actor::await_data() { void blocking_actor::await_data() {
mailbox().synchronized_await(mtx_, cv_); if (mailbox().try_block()) {
std::unique_lock guard{mtx_};
while (mailbox().blocked())
cv_.wait(guard);
}
} }
bool blocking_actor::await_data(timeout_type timeout) { bool blocking_actor::await_data(timeout_type timeout) {
return mailbox().synchronized_await(mtx_, cv_, timeout); if (mailbox().try_block()) {
std::unique_lock guard{mtx_};
while (mailbox().blocked()) {
if (cv_.wait_until(guard, timeout) == std::cv_status::timeout) {
// If we're unable to set the queue from blocked to empty, then there's
// a new element in the list.
return !mailbox().try_unblock();
}
}
}
return true;
} }
mailbox_element_ptr blocking_actor::dequeue() { mailbox_element_ptr blocking_actor::dequeue() {
mailbox().flush_cache(); if (auto ptr = mailbox().pop_front())
return ptr;
await_data(); await_data();
mailbox().fetch_more(); return mailbox().pop_front();
auto& qs = mailbox().queue().queues();
auto result = get<mailbox_policy::urgent_queue_index>(qs).take_front();
if (!result)
result = get<mailbox_policy::normal_queue_index>(qs).take_front();
CAF_ASSERT(result != nullptr);
return result;
} }
void blocking_actor::varargs_tup_receive(receive_cond& rcc, message_id mid, void blocking_actor::varargs_tup_receive(receive_cond& rcc, message_id mid,
...@@ -373,20 +366,16 @@ size_t blocking_actor::attach_functor(const strong_actor_ptr& ptr) { ...@@ -373,20 +366,16 @@ size_t blocking_actor::attach_functor(const strong_actor_ptr& ptr) {
bool blocking_actor::cleanup(error&& fail_state, execution_unit* host) { bool blocking_actor::cleanup(error&& fail_state, execution_unit* host) {
if (!mailbox_.closed()) { if (!mailbox_.closed()) {
mailbox_.close(); unstash();
// TODO: messages that are stuck in the cache can get lost mailbox_.close(fail_state);
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;
}
} }
// 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);
} }
void blocking_actor::unstash() {
while (auto stashed = stash_.pop())
mailbox().push_front(mailbox_element_ptr{stashed});
}
} // namespace caf } // namespace caf
...@@ -4,6 +4,7 @@ ...@@ -4,6 +4,7 @@
#pragma once #pragma once
#include "caf/abstract_mailbox.hpp"
#include "caf/actor_config.hpp" #include "caf/actor_config.hpp"
#include "caf/actor_traits.hpp" #include "caf/actor_traits.hpp"
#include "caf/after.hpp" #include "caf/after.hpp"
...@@ -11,6 +12,7 @@ ...@@ -11,6 +12,7 @@
#include "caf/detail/apply_args.hpp" #include "caf/detail/apply_args.hpp"
#include "caf/detail/blocking_behavior.hpp" #include "caf/detail/blocking_behavior.hpp"
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/detail/default_mailbox.hpp"
#include "caf/detail/type_list.hpp" #include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
#include "caf/extend.hpp" #include "caf/extend.hpp"
...@@ -18,8 +20,7 @@ ...@@ -18,8 +20,7 @@
#include "caf/intrusive/drr_cached_queue.hpp" #include "caf/intrusive/drr_cached_queue.hpp"
#include "caf/intrusive/drr_queue.hpp" #include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp" #include "caf/intrusive/fifo_inbox.hpp"
#include "caf/intrusive/wdrr_dynamic_multiplexed_queue.hpp" #include "caf/intrusive/stack.hpp"
#include "caf/intrusive/wdrr_fixed_multiplexed_queue.hpp"
#include "caf/is_timeout_or_catch_all.hpp" #include "caf/is_timeout_or_catch_all.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
...@@ -29,8 +30,6 @@ ...@@ -29,8 +30,6 @@
#include "caf/none.hpp" #include "caf/none.hpp"
#include "caf/policy/arg.hpp" #include "caf/policy/arg.hpp"
#include "caf/policy/categorized.hpp" #include "caf/policy/categorized.hpp"
#include "caf/policy/normal_messages.hpp"
#include "caf/policy/urgent_messages.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/typed_actor.hpp" #include "caf/typed_actor.hpp"
...@@ -64,29 +63,6 @@ public: ...@@ -64,29 +63,6 @@ public:
/// Stores asynchronous messages with high priority. /// Stores asynchronous messages with high priority.
using urgent_queue = intrusive::drr_cached_queue<policy::urgent_messages>; 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. /// Absolute timeout type.
using timeout_type = std::chrono::high_resolution_clock::time_point; using timeout_type = std::chrono::high_resolution_clock::time_point;
...@@ -193,23 +169,6 @@ public: ...@@ -193,23 +169,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 ------------------------------------------- // -- constructors and destructors -------------------------------------------
blocking_actor(actor_config& cfg); blocking_actor(actor_config& cfg);
...@@ -344,7 +303,7 @@ public: ...@@ -344,7 +303,7 @@ public:
virtual mailbox_element_ptr dequeue(); virtual mailbox_element_ptr dequeue();
/// Returns the queue for storing incoming messages. /// Returns the queue for storing incoming messages.
mailbox_type& mailbox() { abstract_mailbox& mailbox() {
return mailbox_; return mailbox_;
} }
/// @cond PRIVATE /// @cond PRIVATE
...@@ -405,6 +364,8 @@ private: ...@@ -405,6 +364,8 @@ private:
size_t attach_functor(const strong_actor_ptr&); size_t attach_functor(const strong_actor_ptr&);
void unstash();
template <class... Ts> template <class... Ts>
size_t attach_functor(const typed_actor<Ts...>& x) { size_t attach_functor(const typed_actor<Ts...>& x) {
return attach_functor(actor_cast<strong_actor_ptr>(x)); return attach_functor(actor_cast<strong_actor_ptr>(x));
...@@ -420,8 +381,11 @@ private: ...@@ -420,8 +381,11 @@ private:
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
// used by both event-based and blocking actors /// Stores incoming messages.
mailbox_type mailbox_; detail::default_mailbox mailbox_;
/// Stashes skipped messages until the actor processes the next message.
intrusive::stack<mailbox_element> stash_;
}; };
} // namespace caf } // namespace caf
...@@ -62,6 +62,10 @@ bool default_mailbox::try_block() { ...@@ -62,6 +62,10 @@ bool default_mailbox::try_block() {
return cached() == 0 && inbox_.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 default_mailbox::close(const error& reason) {
size_t result = 0; size_t result = 0;
detail::sync_request_bouncer bounce{reason}; detail::sync_request_bouncer bounce{reason};
......
...@@ -36,12 +36,12 @@ public: ...@@ -36,12 +36,12 @@ public:
default_mailbox& operator=(const default_mailbox&) = delete; default_mailbox& operator=(const default_mailbox&) = delete;
mailbox_element* peek(message_id id);
intrusive::inbox_result push_back(mailbox_element_ptr ptr) override; intrusive::inbox_result push_back(mailbox_element_ptr ptr) override;
void push_front(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; mailbox_element_ptr pop_front() override;
bool closed() override; bool closed() override;
...@@ -50,6 +50,8 @@ public: ...@@ -50,6 +50,8 @@ public:
bool try_block() override; bool try_block() override;
bool try_unblock() override;
size_t close(const error&) override; size_t close(const error&) override;
size_t size() override; size_t size() override;
......
...@@ -181,9 +181,9 @@ bool scheduled_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) { ...@@ -181,9 +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() {
return mailbox().peek(awaited_responses_.empty() return mailbox_.peek(awaited_responses_.empty()
? make_message_id() ? make_message_id()
: awaited_responses_.begin()->first); : awaited_responses_.begin()->first);
} }
// -- overridden functions of local_actor -------------------------------------- // -- overridden functions of local_actor --------------------------------------
......
...@@ -120,9 +120,6 @@ public: ...@@ -120,9 +120,6 @@ public:
size_t max_items_per_batch; size_t max_items_per_batch;
}; };
/// A queue optimized for single-reader-many-writers.
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>;
...@@ -687,7 +684,7 @@ protected: ...@@ -687,7 +684,7 @@ protected:
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// Stores incoming messages. /// Stores incoming messages.
mailbox_type mailbox_; detail::default_mailbox mailbox_;
/// Stores user-defined callbacks for message handling. /// Stores user-defined callbacks for message handling.
detail::behavior_stack bhvr_stack_; detail::behavior_stack bhvr_stack_;
......
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