Commit fed4e6c5 authored by Dominik Charousset's avatar Dominik Charousset

Collect metrics for inbound stream traffic

parent 206ba3f2
...@@ -196,35 +196,35 @@ public: ...@@ -196,35 +196,35 @@ public:
/// of the actor). /// of the actor).
struct actor_metric_families_t { struct actor_metric_families_t {
/// Samples how long the actor needs to process messages. /// Samples how long the actor needs to process messages.
telemetry::dbl_histogram_family* processing_time_family = nullptr; telemetry::dbl_histogram_family* processing_time = nullptr;
/// Samples how long a message waits in the mailbox before the actor /// Samples how long a message waits in the mailbox before the actor
/// processes it. /// processes it.
telemetry::dbl_histogram_family* mailbox_time_family = nullptr; telemetry::dbl_histogram_family* mailbox_time = nullptr;
/// Counts how many messages are currently waiting in the mailbox. /// Counts how many messages are currently waiting in the mailbox.
telemetry::int_gauge_family* mailbox_size_family = nullptr; telemetry::int_gauge_family* mailbox_size = nullptr;
struct { struct {
// -- inbound ------------------------------------------------------------ // -- inbound ------------------------------------------------------------
/// Counts the total number of processed stream elements from upstream. /// Counts the total number of processed stream elements from upstream.
telemetry::int_counter_family* processed_elements_family = nullptr; telemetry::int_counter_family* processed_elements = nullptr;
/// Tracks how many stream elements from upstream are currently buffered. /// Tracks how many stream elements from upstream are currently buffered.
telemetry::int_gauge_family* input_buffer_size_family = nullptr; telemetry::int_gauge_family* input_buffer_size = nullptr;
// -- outbound ----------------------------------------------------------- // -- outbound -----------------------------------------------------------
/// Counts the total number of elements that have been pushed downstream. /// Counts the total number of elements that have been pushed downstream.
telemetry::int_counter_family* pushed_elements_family = nullptr; telemetry::int_counter_family* pushed_elements = nullptr;
/// Counts the total number of batches that have been pushed downstream. /// Counts the total number of batches that have been pushed downstream.
telemetry::int_counter_family* pushed_batches_family = nullptr; telemetry::int_counter_family* pushed_batches = nullptr;
/// Tracks how many stream elements are currently waiting in the output /// Tracks how many stream elements are currently waiting in the output
/// buffer due to insufficient credit. /// buffer due to insufficient credit.
telemetry::int_gauge_family* output_buffer_size_family = nullptr; telemetry::int_gauge_family* output_buffer_size = nullptr;
} }
/// Wraps streaming-related actor metric families. /// Wraps streaming-related actor metric families.
......
...@@ -35,39 +35,45 @@ ...@@ -35,39 +35,45 @@
namespace caf { namespace caf {
/// Stream messages that travel downstream, i.e., batches and close messages. /// Transmits stream data.
struct CAF_CORE_EXPORT downstream_msg : tag::boxing_type { struct downstream_msg_batch {
// -- nested types ----------------------------------------------------------- /// Allows the testing DSL to unbox this type automagically.
using outer_type = downstream_msg;
/// Size of the type-erased vector<T> (used credit).
int32_t xs_size;
/// A type-erased vector<T> containing the elements of the batch.
message xs;
/// ID of this batch (ascending numbering).
int64_t id;
};
/// Transmits stream data. /// Orderly shuts down a stream after receiving an ACK for the last batch.
struct batch { struct downstream_msg_close {
/// Allows the testing DSL to unbox this type automagically. /// Allows the testing DSL to unbox this type automagically.
using outer_type = downstream_msg; using outer_type = downstream_msg;
};
/// Size of the type-erased vector<T> (used credit). /// Propagates a fatal error from sources to sinks.
int32_t xs_size; struct downstream_msg_forced_close {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = downstream_msg;
/// A type-erased vector<T> containing the elements of the batch. /// Reason for shutting down the stream.
message xs; error reason;
};
/// ID of this batch (ascending numbering). /// Stream messages that travel downstream, i.e., batches and close messages.
int64_t id; struct CAF_CORE_EXPORT downstream_msg : tag::boxing_type {
}; // -- nested types -----------------------------------------------------------
/// Orderly shuts down a stream after receiving an ACK for the last batch. using batch = downstream_msg_batch;
struct close {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = downstream_msg;
};
/// Propagates a fatal error from sources to sinks. using close = downstream_msg_close;
struct forced_close {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = downstream_msg;
/// Reason for shutting down the stream. using forced_close = downstream_msg_forced_close;
error reason;
};
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
......
...@@ -154,6 +154,9 @@ class stateful_actor; ...@@ -154,6 +154,9 @@ class stateful_actor;
struct down_msg; struct down_msg;
struct downstream_msg; struct downstream_msg;
struct downstream_msg_batch;
struct downstream_msg_close;
struct downstream_msg_forced_close;
struct exit_msg; struct exit_msg;
struct group_down_msg; struct group_down_msg;
struct illegal_message_element; struct illegal_message_element;
......
...@@ -31,6 +31,8 @@ ...@@ -31,6 +31,8 @@
#include "caf/stream_manager.hpp" #include "caf/stream_manager.hpp"
#include "caf/stream_priority.hpp" #include "caf/stream_priority.hpp"
#include "caf/stream_slot.hpp" #include "caf/stream_slot.hpp"
#include "caf/telemetry/counter.hpp"
#include "caf/telemetry/gauge.hpp"
#include "caf/timestamp.hpp" #include "caf/timestamp.hpp"
#include "caf/upstream_msg.hpp" #include "caf/upstream_msg.hpp"
...@@ -54,6 +56,12 @@ public: ...@@ -54,6 +56,12 @@ public:
/// Stores slot IDs for sender (hdl) and receiver (self). /// Stores slot IDs for sender (hdl) and receiver (self).
stream_slots slots; stream_slots slots;
/// Optionally stores pointers to telemetry objects.
struct metrics_t {
telemetry::int_counter* processed_elements;
telemetry::int_gauge* input_buffer_size;
} metrics;
/// Stores the last computed desired batch size. /// Stores the last computed desired batch size.
int32_t desired_batch_size = 0; int32_t desired_batch_size = 0;
......
...@@ -55,6 +55,11 @@ public: ...@@ -55,6 +55,11 @@ public:
// nop // nop
} }
~wdrr_dynamic_multiplexed_queue() noexcept {
for (auto& kvp : qs_)
policy_.cleanup(kvp.second);
}
policy_type& policy() noexcept { policy_type& policy() noexcept {
return policy_; return policy_;
} }
...@@ -64,10 +69,8 @@ public: ...@@ -64,10 +69,8 @@ public:
} }
bool push_back(mapped_type* ptr) noexcept { bool push_back(mapped_type* ptr) noexcept {
auto i = qs_.find(policy_.id_of(*ptr)); if (auto i = qs_.find(policy_.id_of(*ptr)); i != qs_.end()) {
if (i != qs_.end()) { return policy_.push_back(i->second, ptr);
i->second.push_back(ptr);
return true;
} else { } else {
typename unique_pointer::deleter_type d; typename unique_pointer::deleter_type d;
d(ptr); d(ptr);
...@@ -135,8 +138,12 @@ public: ...@@ -135,8 +138,12 @@ public:
/// Erases all keys previously marked via `erase_later`. /// Erases all keys previously marked via `erase_later`.
void cleanup() { void cleanup() {
if (!erase_list_.empty()) { if (!erase_list_.empty()) {
for (auto& k : erase_list_) for (auto& k : erase_list_) {
qs_.erase(k); if (auto i = qs_.find(k); i != qs_.end()) {
policy_.cleanup(i->second);
qs_.erase(i);
}
}
erase_list_.clear(); erase_list_.clear();
} }
} }
...@@ -191,9 +198,8 @@ public: ...@@ -191,9 +198,8 @@ public:
} }
void lifo_append(pointer ptr) noexcept { void lifo_append(pointer ptr) noexcept {
auto i = qs_.find(policy_.id_of(*ptr)); if (auto i = qs_.find(policy_.id_of(*ptr)); i != qs_.end()) {
if (i != qs_.end()) { policy_.lifo_append(i->second, ptr);
i->second.lifo_append(ptr);
} else { } else {
typename unique_pointer::deleter_type d; typename unique_pointer::deleter_type d;
d(ptr); d(ptr);
...@@ -202,7 +208,7 @@ public: ...@@ -202,7 +208,7 @@ public:
void stop_lifo_append() noexcept { void stop_lifo_append() noexcept {
for (auto& kvp : qs_) for (auto& kvp : qs_)
kvp.second.stop_lifo_append(); policy_.stop_lifo_append(kvp.second);
} }
private: private:
......
...@@ -50,6 +50,18 @@ public: ...@@ -50,6 +50,18 @@ public:
using handler_type = std::unique_ptr<inbound_path>; using handler_type = std::unique_ptr<inbound_path>;
static task_size_type task_size(const downstream_msg_batch& x) noexcept;
static constexpr task_size_type
task_size(const downstream_msg_close&) noexcept {
return 1;
}
static constexpr task_size_type
task_size(const downstream_msg_forced_close&) noexcept {
return 1;
}
static task_size_type task_size(const mailbox_element& x) noexcept; static task_size_type task_size(const mailbox_element& x) noexcept;
// -- constructors, destructors, and assignment operators ------------------ // -- constructors, destructors, and assignment operators ------------------
...@@ -72,6 +84,8 @@ public: ...@@ -72,6 +84,8 @@ public:
// -- member variables ----------------------------------------------------- // -- member variables -----------------------------------------------------
handler_type handler; handler_type handler;
size_t bulk_inserted_size = 0;
}; };
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -82,6 +96,8 @@ public: ...@@ -82,6 +96,8 @@ public:
using deficit_type = size_t; using deficit_type = size_t;
using pointer = mapped_type*;
using unique_pointer = mailbox_element_ptr; using unique_pointer = mailbox_element_ptr;
using key_type = stream_slot; using key_type = stream_slot;
...@@ -116,6 +132,17 @@ public: ...@@ -116,6 +132,17 @@ public:
static task_size_type task_size(const mailbox_element&) noexcept { static task_size_type task_size(const mailbox_element&) noexcept {
return 1; return 1;
} }
// -- required functions for wdrr_dynamic_multiplexed_queue ------------------
static void cleanup(nested_queue_type&) noexcept;
static bool push_back(nested_queue_type& sub_queue,
pointer ptr) noexcept;
static void lifo_append(nested_queue_type& sub_queue, pointer ptr) noexcept;
static void stop_lifo_append(nested_queue_type& sub_queue) noexcept;
}; };
} // namespace caf::policy } // namespace caf::policy
...@@ -119,6 +119,10 @@ public: ...@@ -119,6 +119,10 @@ public:
/// Base type. /// Base type.
using super = local_actor; using super = local_actor;
/// Maps types to metric objects for inbound stream traffic.
using inbound_stream_metrics_map
= std::unordered_map<type_id_t, inbound_stream_metrics_t>;
/// Maps slot IDs to stream managers. /// Maps slot IDs to stream managers.
using stream_manager_map = std::map<stream_slot, stream_manager_ptr>; using stream_manager_map = std::map<stream_slot, stream_manager_ptr>;
...@@ -331,6 +335,10 @@ public: ...@@ -331,6 +335,10 @@ public:
return pending_stream_managers_; return pending_stream_managers_;
} }
// -- actor metrics ----------------------------------------------------------
inbound_stream_metrics_t inbound_stream_metrics(type_id_t type);
// -- event handlers --------------------------------------------------------- // -- event handlers ---------------------------------------------------------
/// Sets a custom handler for unexpected messages. /// Sets a custom handler for unexpected messages.
...@@ -721,6 +729,9 @@ protected: ...@@ -721,6 +729,9 @@ protected:
/// Pointer to a private thread object associated with a detached actor. /// Pointer to a private thread object associated with a detached actor.
detail::private_thread* private_thread_; detail::private_thread* private_thread_;
/// Catche metric objects for inbound stream traffic.
inbound_stream_metrics_map inbound_stream_metrics_;
#ifdef CAF_ENABLE_EXCEPTIONS #ifdef CAF_ENABLE_EXCEPTIONS
/// Customization point for setting a default exception callback. /// Customization point for setting a default exception callback.
exception_handler exception_handler_; exception_handler exception_handler_;
......
...@@ -256,18 +256,18 @@ auto make_actor_metric_families(telemetry::metric_registry& reg) { ...@@ -256,18 +256,18 @@ auto make_actor_metric_families(telemetry::metric_registry& reg) {
"Number of messages in the mailbox."), "Number of messages in the mailbox."),
{ {
reg.counter_family("caf.actor.stream", "processed-elements", reg.counter_family("caf.actor.stream", "processed-elements",
{"name", "slot"}, {"name", "type"},
"Number of processed stream elements from upstream."), "Number of processed stream elements from upstream."),
reg.gauge_family("caf.actor.stream", "input_buffer_size", reg.gauge_family("caf.actor.stream", "input_buffer_size",
{"name", "slot"}, {"name", "type"},
"Number of buffered stream elements from upstream."), "Number of buffered stream elements from upstream."),
reg.counter_family( reg.counter_family(
"caf.actor.stream", "pushed-elements", {"name", "slot"}, "caf.actor.stream", "pushed-elements", {"name", "type"},
"Number of elements that have been pushed downstream."), "Number of elements that have been pushed downstream."),
reg.counter_family("caf.actor.stream", "pushed-batches", {"name", "slot"}, reg.counter_family("caf.actor.stream", "pushed-batches", {"name", "type"},
"Number of batches that have been pushed downstream."), "Number of batches that have been pushed downstream."),
reg.gauge_family("caf.actor.stream", "output-buffer-size", reg.gauge_family("caf.actor.stream", "output-buffer-size",
{"name", "slot"}, {"name", "type"},
"Number of buffered output stream elements."), "Number of buffered output stream elements."),
}, },
}; };
......
...@@ -72,21 +72,25 @@ inbound_path::inbound_path(stream_manager_ptr mgr_ptr, stream_slots id, ...@@ -72,21 +72,25 @@ inbound_path::inbound_path(stream_manager_ptr mgr_ptr, stream_slots id,
strong_actor_ptr ptr, strong_actor_ptr ptr,
[[maybe_unused]] type_id_t in_type) [[maybe_unused]] type_id_t in_type)
: mgr(std::move(mgr_ptr)), hdl(std::move(ptr)), slots(id) { : mgr(std::move(mgr_ptr)), hdl(std::move(ptr)), slots(id) {
auto self = mgr->self();
auto [processed_elements, input_buffer_size]
= self->inbound_stream_metrics(in_type);
metrics = metrics_t{processed_elements, input_buffer_size};
mgr->register_input_path(this); mgr->register_input_path(this);
CAF_STREAM_LOG_DEBUG(mgr->self()->name() CAF_STREAM_LOG_DEBUG(self->name()
<< "opens input stream with element type" << "opens input stream with element type"
<< detail::global_meta_object(in_type)->type_name << detail::global_meta_object(in_type)->type_name
<< "at slot" << id.receiver << "from" << hdl); << "at slot" << id.receiver << "from" << hdl);
if (auto str = get_if<std::string>(&self()->system().config(), if (auto str = get_if<std::string>(&self->system().config(),
"caf.stream.credit-policy")) { "caf.stream.credit-policy")) {
if (*str == "testing") if (*str == "testing")
controller_.reset(new detail::test_credit_controller(self())); controller_.reset(new detail::test_credit_controller(self));
else if (*str == "size") else if (*str == "size")
controller_.reset(new detail::size_based_credit_controller(self())); controller_.reset(new detail::size_based_credit_controller(self));
else else
controller_.reset(new detail::complexity_based_credit_controller(self())); controller_.reset(new detail::complexity_based_credit_controller(self));
} else { } else {
controller_.reset(new detail::complexity_based_credit_controller(self())); controller_.reset(new detail::complexity_based_credit_controller(self));
} }
} }
......
...@@ -58,9 +58,9 @@ local_actor::metrics_t make_instance_metrics(local_actor* self) { ...@@ -58,9 +58,9 @@ local_actor::metrics_t make_instance_metrics(local_actor* self) {
const auto& families = sys.actor_metric_families(); const auto& families = sys.actor_metric_families();
string_view sv{name, strlen(name)}; string_view sv{name, strlen(name)};
return { return {
families.processing_time_family->get_or_add({{"name", sv}}), families.processing_time->get_or_add({{"name", sv}}),
families.mailbox_time_family->get_or_add({{"name", sv}}), families.mailbox_time->get_or_add({{"name", sv}}),
families.mailbox_size_family->get_or_add({{"name", sv}}), families.mailbox_size->get_or_add({{"name", sv}}),
}; };
} }
......
...@@ -25,29 +25,18 @@ ...@@ -25,29 +25,18 @@
namespace caf::policy { namespace caf::policy {
namespace { auto downstream_messages::nested::task_size(
const downstream_msg_batch& batch) noexcept -> task_size_type {
class task_size_calculator { CAF_ASSERT(batch.xs_size > 0);
public: return batch.xs_size;
using size_type = downstream_messages::nested::task_size_type; }
inline size_type operator()(const downstream_msg::batch& x) const noexcept {
CAF_ASSERT(x.xs_size > 0);
return static_cast<size_type>(x.xs_size);
}
template <class T>
size_type operator()(const T&) const noexcept {
return 1;
}
};
} // namespace
auto downstream_messages::nested::task_size(const mailbox_element& x) noexcept auto downstream_messages::nested::task_size(const mailbox_element& x) noexcept
-> task_size_type { -> task_size_type {
task_size_calculator f; CAF_ASSERT(x.mid.is_downstream_message());
return visit(f, x.content().get_as<downstream_msg>(0).content); CAF_ASSERT(x.payload.match_elements<downstream_msg>());
auto f = [](auto& content) { return task_size(content); };
return visit(f, x.payload.get_as<downstream_msg>(0).content);
} }
auto downstream_messages::id_of(mailbox_element& x) noexcept -> key_type { auto downstream_messages::id_of(mailbox_element& x) noexcept -> key_type {
...@@ -67,4 +56,49 @@ auto downstream_messages::quantum(const nested_queue_type& q, ...@@ -67,4 +56,49 @@ auto downstream_messages::quantum(const nested_queue_type& q,
return x * static_cast<deficit_type>(q.policy().handler->desired_batch_size); return x * static_cast<deficit_type>(q.policy().handler->desired_batch_size);
} }
void downstream_messages::cleanup(nested_queue_type& sub_queue) noexcept {
auto& handler = sub_queue.policy().handler;
if (!handler)
return;
if (auto input_buffer_size = handler->metrics.input_buffer_size)
input_buffer_size->dec(sub_queue.total_task_size());
}
bool downstream_messages::push_back(nested_queue_type& sub_queue,
mapped_type* ptr) noexcept {
if (auto handler = sub_queue.policy().handler.get()) {
if (auto input_buffer_size = handler->metrics.input_buffer_size)
input_buffer_size->inc(sub_queue.policy().task_size(*ptr));
return sub_queue.push_back(ptr);
} else {
unique_pointer::deleter_type d;
d(ptr);
return false;
}
}
void downstream_messages::lifo_append(nested_queue_type& sub_queue,
pointer ptr) noexcept {
auto& sub_policy = sub_queue.policy();
if (sub_policy.handler) {
sub_policy.bulk_inserted_size += sub_policy.task_size(*ptr);
sub_queue.lifo_append(ptr);
} else {
unique_pointer::deleter_type d;
d(ptr);
}
}
void downstream_messages::stop_lifo_append(
nested_queue_type& sub_queue) noexcept {
auto& sub_policy = sub_queue.policy();
if (sub_policy.bulk_inserted_size > 0) {
CAF_ASSERT(sub_policy.handler != nullptr);
if (auto input_buffer_size = sub_policy.handler->metrics.input_buffer_size)
input_buffer_size->inc(sub_policy.bulk_inserted_size);
sub_policy.bulk_inserted_size = 0;
sub_queue.stop_lifo_append();
}
}
} // namespace caf::policy } // namespace caf::policy
...@@ -22,6 +22,7 @@ ...@@ -22,6 +22,7 @@
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/detail/default_invoke_result_visitor.hpp" #include "caf/detail/default_invoke_result_visitor.hpp"
#include "caf/detail/meta_object.hpp"
#include "caf/detail/private_thread.hpp" #include "caf/detail/private_thread.hpp"
#include "caf/detail/sync_request_bouncer.hpp" #include "caf/detail/sync_request_bouncer.hpp"
#include "caf/inbound_path.hpp" #include "caf/inbound_path.hpp"
...@@ -319,52 +320,6 @@ scheduled_actor::mailbox_visitor::operator()(size_t, upstream_queue&, ...@@ -319,52 +320,6 @@ scheduled_actor::mailbox_visitor::operator()(size_t, upstream_queue&,
}); });
} }
namespace {
// TODO: replace with generic lambda when switching to C++14
struct downstream_msg_visitor {
scheduled_actor* selfptr;
scheduled_actor::downstream_queue& qs_ref;
policy::downstream_messages::nested_queue_type& q_ref;
downstream_msg& dm;
template <class T>
intrusive::task_result operator()(T& x) {
CAF_LOG_TRACE(CAF_ARG(x));
auto& inptr = q_ref.policy().handler;
if (inptr == nullptr)
return intrusive::task_result::stop;
// Do *not* store a reference here since we potentially reset `inptr`.
auto mgr = inptr->mgr;
inptr->handle(x);
// The sender slot can be 0. This is the case for forced_close or
// forced_drop messages from stream aborters.
CAF_ASSERT(
inptr->slots == dm.slots
|| (dm.slots.sender == 0 && dm.slots.receiver == inptr->slots.receiver));
if constexpr (std::is_same<T, downstream_msg::close>::value
|| std::is_same<T, downstream_msg::forced_close>::value) {
inptr.reset();
qs_ref.erase_later(dm.slots.receiver);
selfptr->erase_stream_manager(dm.slots.receiver);
if (mgr->done()) {
CAF_LOG_DEBUG("path is done receiving and closes its manager");
selfptr->erase_stream_manager(mgr);
mgr->stop();
}
return intrusive::task_result::stop;
} else if (mgr->done()) {
CAF_LOG_DEBUG("path is done receiving and closes its manager");
selfptr->erase_stream_manager(mgr);
mgr->stop();
return intrusive::task_result::stop;
}
return intrusive::task_result::resume;
}
};
} // namespace
intrusive::task_result scheduled_actor::mailbox_visitor::operator()( intrusive::task_result scheduled_actor::mailbox_visitor::operator()(
size_t, downstream_queue& qs, stream_slot, size_t, downstream_queue& qs, stream_slot,
policy::downstream_messages::nested_queue_type& q, mailbox_element& x) { policy::downstream_messages::nested_queue_type& q, mailbox_element& x) {
...@@ -375,7 +330,48 @@ intrusive::task_result scheduled_actor::mailbox_visitor::operator()( ...@@ -375,7 +330,48 @@ intrusive::task_result scheduled_actor::mailbox_visitor::operator()(
CAF_BEFORE_PROCESSING(self, x); CAF_BEFORE_PROCESSING(self, x);
CAF_ASSERT(x.content().match_elements<downstream_msg>()); CAF_ASSERT(x.content().match_elements<downstream_msg>());
auto& dm = x.content().get_mutable_as<downstream_msg>(0); auto& dm = x.content().get_mutable_as<downstream_msg>(0);
downstream_msg_visitor f{self, qs, q, dm}; auto f = [&, this](auto& content) {
using content_type = std::decay_t<decltype(content)>;
auto& inptr = q.policy().handler;
if (inptr == nullptr)
return intrusive::task_result::stop;
if (auto processed_elements = inptr->metrics.processed_elements) {
auto num_elements = q.policy().task_size(content);
auto input_buffer_size = inptr->metrics.input_buffer_size;
CAF_ASSERT(input_buffer_size != nullptr);
processed_elements->inc(num_elements);
input_buffer_size->dec(num_elements);
}
// Do *not* store a reference here since we potentially reset `inptr`.
auto mgr = inptr->mgr;
inptr->handle(content);
// The sender slot can be 0. This is the case for forced_close or
// forced_drop messages from stream aborters.
CAF_ASSERT(inptr->slots == dm.slots
|| (dm.slots.sender == 0
&& dm.slots.receiver == inptr->slots.receiver));
if constexpr (std::is_same<content_type, downstream_msg::close>::value
|| std::is_same<content_type,
downstream_msg::forced_close>::value) {
if (auto input_buffer_size = inptr->metrics.input_buffer_size)
input_buffer_size->dec(q.total_task_size());
inptr.reset();
qs.erase_later(dm.slots.receiver);
self->erase_stream_manager(dm.slots.receiver);
if (mgr->done()) {
CAF_LOG_DEBUG("path is done receiving and closes its manager");
self->erase_stream_manager(mgr);
mgr->stop();
}
return intrusive::task_result::stop;
} else if (mgr->done()) {
CAF_LOG_DEBUG("path is done receiving and closes its manager");
self->erase_stream_manager(mgr);
mgr->stop();
return intrusive::task_result::stop;
}
return intrusive::task_result::resume;
};
auto res = visit(f, dm.content); auto res = visit(f, dm.content);
CAF_AFTER_PROCESSING(self, invoke_message_result::consumed); CAF_AFTER_PROCESSING(self, invoke_message_result::consumed);
return ++handled_msgs < max_throughput ? res return ++handled_msgs < max_throughput ? res
...@@ -502,6 +498,28 @@ void scheduled_actor::quit(error x) { ...@@ -502,6 +498,28 @@ void scheduled_actor::quit(error x) {
} }
} }
// -- actor metrics ------------------------------------------------------------
auto scheduled_actor::inbound_stream_metrics(type_id_t type)
-> inbound_stream_metrics_t {
if (!has_metrics_enabled())
return {nullptr, nullptr};
if (auto i = inbound_stream_metrics_.find(type);
i != inbound_stream_metrics_.end())
return i->second;
auto actor_name_cstr = name();
auto actor_name = string_view{actor_name_cstr, strlen(actor_name_cstr)};
auto tname_cstr = detail::global_meta_object(type)->type_name;
auto tname = string_view{tname_cstr, strlen(tname_cstr)};
auto fs = system().actor_metric_families().stream;
inbound_stream_metrics_t result{
fs.processed_elements->get_or_add({{"name", actor_name}, {"type", tname}}),
fs.input_buffer_size->get_or_add({{"name", actor_name}, {"type", tname}}),
};
inbound_stream_metrics_.emplace(type, result);
return result;
}
// -- timeout management ------------------------------------------------------- // -- timeout management -------------------------------------------------------
uint64_t scheduled_actor::set_receive_timeout(actor_clock::time_point x) { uint64_t scheduled_actor::set_receive_timeout(actor_clock::time_point x) {
...@@ -951,10 +969,11 @@ void scheduled_actor::erase_inbound_path_later(stream_slot slot, error reason) { ...@@ -951,10 +969,11 @@ void scheduled_actor::erase_inbound_path_later(stream_slot slot, error reason) {
if (i != e) { if (i != e) {
auto& path = i->second.policy().handler; auto& path = i->second.policy().handler;
if (path != nullptr) { if (path != nullptr) {
if (reason == none) if (reason == none) {
path->emit_regular_shutdown(this); path->emit_regular_shutdown(this);
else } else {
path->emit_irregular_shutdown(this, std::move(reason)); path->emit_irregular_shutdown(this, std::move(reason));
}
} }
q.erase_later(slot); q.erase_later(slot);
} }
......
...@@ -98,6 +98,14 @@ struct inode_policy { ...@@ -98,6 +98,14 @@ struct inode_policy {
return enable_priorities && *q.policy().queue_id == 0 ? 2 * x : x; 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; bool enable_priorities = false;
}; };
......
...@@ -250,14 +250,34 @@ struct dmsg_queue_policy : policy_base { ...@@ -250,14 +250,34 @@ struct dmsg_queue_policy : policy_base {
} }
template <class Queue> template <class Queue>
static inline bool enabled(const Queue&) { static inline bool enabled(const Queue&) noexcept {
return true; return true;
} }
template <class Queue> template <class Queue>
deficit_type quantum(const Queue&, deficit_type x) { deficit_type quantum(const Queue&, deficit_type x) noexcept {
return x; return x;
} }
template <class Queue>
static void cleanup(Queue&) noexcept {
// nop
}
template <class Queue, class T>
static void lifo_append(Queue& q, T* ptr) noexcept {
q.lifo_append(ptr);
}
template <class Queue>
static void stop_lifo_append(Queue& q) noexcept {
q.stop_lifo_append();
}
template <class Queue, class T>
static bool push_back(Queue& q, T* ptr) noexcept {
return q.push_back(ptr);
}
}; };
using dmsg_queue = wdrr_dynamic_multiplexed_queue<dmsg_queue_policy>; using dmsg_queue = wdrr_dynamic_multiplexed_queue<dmsg_queue_policy>;
......
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