Commit fedb0a17 authored by Dominik Charousset's avatar Dominik Charousset

Use the new mailbox type in actor shells

parent a3cfcbd2
......@@ -35,12 +35,12 @@ public:
/// Checks whether the mailbox has been closed.
/// @note Only the owning actor is allowed to call this function.
virtual bool closed() = 0;
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() = 0;
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.
......
......@@ -367,7 +367,11 @@ size_t blocking_actor::attach_functor(const strong_actor_ptr& ptr) {
bool blocking_actor::cleanup(error&& fail_state, execution_unit* host) {
if (!mailbox_.closed()) {
unstash();
mailbox_.close(fail_state);
auto dropped = mailbox_.close(fail_state);
while (dropped > 0 && getf(abstract_actor::collects_metrics_flag)) {
auto val = static_cast<int64_t>(dropped);
metrics_.mailbox_size->dec(val);
}
}
// Dispatch to parent's `cleanup` function.
return super::cleanup(std::move(fail_state), host);
......
......@@ -17,9 +17,6 @@
#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/stack.hpp"
#include "caf/is_timeout_or_catch_all.hpp"
#include "caf/local_actor.hpp"
......@@ -57,12 +54,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>;
/// Absolute timeout type.
using timeout_type = std::chrono::high_resolution_clock::time_point;
......
......@@ -50,11 +50,11 @@ mailbox_element_ptr default_mailbox::pop_front() {
}
}
bool default_mailbox::closed() {
bool default_mailbox::closed() const noexcept {
return inbox_.closed();
}
bool default_mailbox::blocked() {
bool default_mailbox::blocked() const noexcept {
return inbox_.blocked();
}
......
......@@ -7,7 +7,6 @@
#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"
......@@ -44,9 +43,9 @@ public:
mailbox_element_ptr pop_front() override;
bool closed() override;
bool closed() const noexcept override;
bool blocked() override;
bool blocked() const noexcept override;
bool try_block() override;
......
......@@ -10,6 +10,7 @@
#include "caf/flow/subscription.hpp"
#include <algorithm>
#include <deque>
#include <memory>
#include <tuple>
#include <type_traits>
......
......@@ -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"
......
......@@ -29,6 +29,7 @@
#include "caf/flow/observer.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/stack.hpp"
#include "caf/intrusive/task_result.hpp"
#include "caf/invoke_message_result.hpp"
#include "caf/local_actor.hpp"
#include "caf/logger.hpp"
......
......@@ -21,7 +21,7 @@ namespace caf::net {
abstract_actor_shell::abstract_actor_shell(actor_config& cfg,
async::execution_context_ptr loop)
: super(cfg), mailbox_(policy::normal_messages{}), loop_(loop) {
: super(cfg), loop_(loop) {
mailbox_.try_block();
resume_ = make_action([this] {
for (;;) {
......@@ -50,14 +50,8 @@ void abstract_actor_shell::quit(error reason) {
// -- mailbox access -----------------------------------------------------------
mailbox_element_ptr abstract_actor_shell::next_message() {
if (!mailbox_.blocked()) {
mailbox_.fetch_more();
auto& q = mailbox_.queue();
if (q.total_task_size() > 0) {
q.inc_deficit(1);
return q.next();
}
}
if (!mailbox_.blocked())
return mailbox_.pop_front();
return nullptr;
}
......@@ -153,7 +147,7 @@ bool abstract_actor_shell::enqueue(mailbox_element_ptr ptr, execution_unit*) {
}
mailbox_element* abstract_actor_shell::peek_at_next_mailbox_element() {
return mailbox().closed() || mailbox().blocked() ? nullptr : mailbox().peek();
return mailbox_.peek(make_message_id());
}
// -- overridden functions of local_actor --------------------------------------
......@@ -170,16 +164,11 @@ bool abstract_actor_shell::cleanup(error&& fail_state, execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(fail_state));
// Clear mailbox.
if (!mailbox_.closed()) {
mailbox_.close();
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 dropped = mailbox_.close(fail_state);
while (dropped > 0 && 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;
}
}
// Detach from owner.
{
......
......@@ -6,14 +6,14 @@
#include "caf/net/fwd.hpp"
#include "caf/abstract_mailbox.hpp"
#include "caf/actor_traits.hpp"
#include "caf/async/execution_context.hpp"
#include "caf/callback.hpp"
#include "caf/detail/default_mailbox.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/extend.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/local_actor.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp"
......@@ -30,18 +30,6 @@ public:
using super = local_actor;
struct mailbox_policy {
using queue_type = intrusive::drr_queue<policy::normal_messages>;
using deficit_type = policy::normal_messages::deficit_type;
using mapped_type = policy::normal_messages::mapped_type;
using unique_pointer = policy::normal_messages::unique_pointer;
};
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
using fallback_handler_sig = result<message>(abstract_actor_shell*, message&);
using fallback_handler = unique_callback_ptr<fallback_handler_sig>;
......@@ -82,7 +70,7 @@ public:
// -- mailbox access ---------------------------------------------------------
auto& mailbox() noexcept {
abstract_mailbox& mailbox() noexcept {
return mailbox_;
}
......@@ -120,7 +108,7 @@ protected:
private:
/// Stores incoming actor messages.
mailbox_type mailbox_;
detail::default_mailbox mailbox_;
/// Guards access to loop_.
std::mutex loop_mtx_;
......
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