Unverified Commit 20168887 authored by Joseph Noir's avatar Joseph Noir Committed by GitHub

Merge pull request #1137

Add telemetry for streams
parents de641b56 144c66b0
......@@ -196,14 +196,36 @@ public:
/// of the actor).
struct actor_metric_families_t {
/// 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
/// 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.
telemetry::int_gauge_family* mailbox_size_family = nullptr;
telemetry::int_gauge_family* mailbox_size = nullptr;
struct {
// -- inbound ------------------------------------------------------------
/// Counts the total number of processed stream elements from upstream.
telemetry::int_counter_family* processed_elements = nullptr;
/// Tracks how many stream elements from upstream are currently buffered.
telemetry::int_gauge_family* input_buffer_size = nullptr;
// -- outbound -----------------------------------------------------------
/// Counts the total number of elements that have been pushed downstream.
telemetry::int_counter_family* pushed_elements = nullptr;
/// Tracks how many stream elements are currently waiting in the output
/// buffer due to insufficient credit.
telemetry::int_gauge_family* output_buffer_size = nullptr;
}
/// Wraps streaming-related actor metric families.
stream;
};
/// @warning The system stores a reference to `cfg`, which means the
......
......@@ -58,12 +58,25 @@ public:
// -- constructors, destructors, and assignment operators --------------------
broadcast_downstream_manager(stream_manager* parent) : super(parent) {
broadcast_downstream_manager(stream_manager* parent)
: super(parent, type_id_v<T>) {
// nop
}
// -- properties -------------------------------------------------------------
template <class... Ts>
bool push_to(stream_slot slot, Ts&&... xs) {
auto old_size = buffered();
if (auto i = states().find(slot); i != states().end()) {
i->second.buf.emplace_back(std::forward<Ts>(xs)...);
auto new_size = buffered();
this->generated_messages(new_size - old_size);
return true;
}
return false;
}
size_t buffered() const noexcept override {
// We have a central buffer, but also an additional buffer at each path. We
// return the maximum size to reflect the current worst case.
......@@ -195,6 +208,7 @@ public:
/// Forces the manager flush its buffer to the individual path buffers.
void fan_out_flush() {
auto old_size = buffered();
auto& buf = this->buf_;
auto f = [&](typename map_type::value_type& x,
typename state_map_type::value_type& y) {
......@@ -203,8 +217,7 @@ public:
return;
// Push data from the global buffer to path buffers.
auto& st = y.second;
// TODO: replace with `if constexpr` when switching to C++17
if (std::is_same<select_type, detail::select_all>::value) {
if constexpr (std::is_same<select_type, detail::select_all>::value) {
st.buf.insert(st.buf.end(), buf.begin(), buf.end());
} else {
for (auto& piece : buf)
......@@ -214,6 +227,10 @@ public:
};
detail::zip_foreach(f, this->paths_.container(), state_map_.container());
buf.clear();
// We may drop messages due to filtering or because all paths are closed.
auto new_size = buffered();
CAF_ASSERT(old_size >= new_size);
this->dropped_messages(old_size - new_size);
}
protected:
......@@ -230,6 +247,7 @@ private:
CAF_ASSERT(this->paths_.size() <= state_map_.size());
if (this->paths_.empty())
return;
auto old_size = buffered();
// Calculate the chunk size, i.e., how many more items we can put to our
// caches at the most.
auto not_closing = [&](typename map_type::value_type& x,
......@@ -256,7 +274,6 @@ private:
detail::zip_foreach(g, this->paths_.container(), state_map_.container());
return;
}
auto chunk = this->get_chunk(chunk_size);
if (chunk.empty()) {
auto g = [&](typename map_type::value_type& x,
......@@ -288,6 +305,9 @@ private:
};
detail::zip_foreach(g, this->paths_.container(), state_map_.container());
}
auto new_size = buffered();
CAF_ASSERT(old_size >= new_size);
this->shipped_messages(old_size - new_size);
}
state_map_type state_map_;
......
......@@ -53,9 +53,15 @@ public:
// nop
}
buffered_downstream_manager(stream_manager* parent, type_id_t type)
: super(parent, type) {
// nop
}
template <class T0, class... Ts>
void push(T0&& x, Ts&&... xs) {
buf_.emplace_back(std::forward<T0>(x), std::forward<Ts>(xs)...);
this->generated_messages(1);
}
/// @pre `n <= buf_.size()`
......
......@@ -68,11 +68,14 @@ public:
CAF_LOG_DEBUG(CAF_ARG(hint));
if (hint == 0)
return false;
auto old_size = this->out_.buf().size();
downstream<typename Driver::output_type> ds{this->out_.buf()};
driver_.pull(ds, hint);
if (driver_.done())
at_end_ = true;
return hint != this->out_.capacity();
auto new_size = this->out_.buf().size();
this->out_.generated_messages(new_size - old_size);
return new_size != old_size;
}
protected:
......
......@@ -63,8 +63,11 @@ public:
CAF_LOG_TRACE(CAF_ARG(x));
using vec_type = std::vector<input_type>;
if (auto view = make_typed_message_view<vec_type>(x.xs)) {
auto old_size = this->out_.buf().size();
downstream<output_type> ds{this->out_.buf()};
driver_.process(ds, get<0>(view));
auto new_size = this->out_.buf().size();
this->out_.generated_messages(new_size - old_size);
return;
}
CAF_LOG_ERROR("received unexpected batch type (dropped)");
......
......@@ -25,6 +25,8 @@
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/downstream_manager.hpp"
#include "caf/outbound_path.hpp"
#include "caf/telemetry/counter.hpp"
#include "caf/telemetry/gauge.hpp"
namespace caf {
......@@ -40,10 +42,22 @@ public:
/// Maps slots to paths.
using map_type = detail::unordered_flat_map<stream_slot, unique_path_ptr>;
/// Optional metrics for outbound stream traffic.
struct metrics_t {
/// Counts the total number of elements that have been pushed downstream.
telemetry::int_counter* pushed_elements = nullptr;
/// Tracks how many stream elements are currently waiting in the output
/// buffer due to insufficient credit.
telemetry::int_gauge* output_buffer_size = nullptr;
};
// -- constructors, destructors, and assignment operators --------------------
explicit downstream_manager_base(stream_manager* parent);
downstream_manager_base(stream_manager* parent, type_id_t type);
~downstream_manager_base() override;
// -- properties -------------------------------------------------------------
......@@ -60,24 +74,45 @@ public:
size_t num_paths() const noexcept override;
bool
remove_path(stream_slot slots, error reason, bool silent) noexcept override;
bool remove_path(stream_slot slots, error reason,
bool silent) noexcept override;
path_ptr path(stream_slot slots) noexcept override;
void clear_paths() override;
// -- callbacks for actor metrics --------------------------------------------
void generated_messages(size_t num) {
if (num > 0 && metrics_.output_buffer_size)
metrics_.output_buffer_size->inc(static_cast<int64_t>(num));
}
void dropped_messages(size_t num) {
if (num > 0 && metrics_.output_buffer_size)
metrics_.output_buffer_size->dec(static_cast<int64_t>(num));
}
void shipped_messages(size_t num) {
if (num > 0 && metrics_.output_buffer_size) {
metrics_.output_buffer_size->dec(static_cast<int64_t>(num));
metrics_.pushed_elements->inc(static_cast<int64_t>(num));
}
}
protected:
bool insert_path(unique_path_ptr ptr) override;
void for_each_path_impl(path_visitor& f) override;
bool check_paths_impl(path_algorithm algo, path_predicate& pred) const
noexcept override;
bool check_paths_impl(path_algorithm algo,
path_predicate& pred) const noexcept override;
// -- member variables -------------------------------------------------------
map_type paths_;
metrics_t metrics_;
};
} // namespace caf
......@@ -35,39 +35,45 @@
namespace caf {
/// Stream messages that travel downstream, i.e., batches and close messages.
struct CAF_CORE_EXPORT downstream_msg : tag::boxing_type {
// -- nested types -----------------------------------------------------------
/// Transmits stream data.
struct downstream_msg_batch {
/// 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.
struct batch {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = downstream_msg;
/// Orderly shuts down a stream after receiving an ACK for the last batch.
struct downstream_msg_close {
/// 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;
/// Propagates a fatal error from sources to sinks.
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.
message xs;
/// Reason for shutting down the stream.
error reason;
};
/// ID of this batch (ascending numbering).
int64_t id;
};
/// Stream messages that travel downstream, i.e., batches and close messages.
struct CAF_CORE_EXPORT downstream_msg : tag::boxing_type {
// -- nested types -----------------------------------------------------------
/// Orderly shuts down a stream after receiving an ACK for the last batch.
struct close {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = downstream_msg;
};
using batch = downstream_msg_batch;
/// Propagates a fatal error from sources to sinks.
struct forced_close {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = downstream_msg;
using close = downstream_msg_close;
/// Reason for shutting down the stream.
error reason;
};
using forced_close = downstream_msg_forced_close;
// -- member types -----------------------------------------------------------
......
......@@ -154,6 +154,9 @@ class stateful_actor;
struct down_msg;
struct downstream_msg;
struct downstream_msg_batch;
struct downstream_msg_close;
struct downstream_msg_forced_close;
struct exit_msg;
struct group_down_msg;
struct illegal_message_element;
......
......@@ -31,6 +31,8 @@
#include "caf/stream_manager.hpp"
#include "caf/stream_priority.hpp"
#include "caf/stream_slot.hpp"
#include "caf/telemetry/counter.hpp"
#include "caf/telemetry/gauge.hpp"
#include "caf/timestamp.hpp"
#include "caf/upstream_msg.hpp"
......@@ -54,6 +56,12 @@ public:
/// Stores slot IDs for sender (hdl) and receiver (self).
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.
int32_t desired_batch_size = 0;
......
......@@ -55,6 +55,11 @@ public:
// nop
}
~wdrr_dynamic_multiplexed_queue() noexcept {
for (auto& kvp : qs_)
policy_.cleanup(kvp.second);
}
policy_type& policy() noexcept {
return policy_;
}
......@@ -64,10 +69,8 @@ public:
}
bool push_back(mapped_type* ptr) noexcept {
auto i = qs_.find(policy_.id_of(*ptr));
if (i != qs_.end()) {
i->second.push_back(ptr);
return true;
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);
......@@ -135,8 +138,12 @@ public:
/// Erases all keys previously marked via `erase_later`.
void cleanup() {
if (!erase_list_.empty()) {
for (auto& k : erase_list_)
qs_.erase(k);
for (auto& k : erase_list_) {
if (auto i = qs_.find(k); i != qs_.end()) {
policy_.cleanup(i->second);
qs_.erase(i);
}
}
erase_list_.clear();
}
}
......@@ -191,9 +198,8 @@ public:
}
void lifo_append(pointer ptr) noexcept {
auto i = qs_.find(policy_.id_of(*ptr));
if (i != qs_.end()) {
i->second.lifo_append(ptr);
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);
......@@ -202,7 +208,7 @@ public:
void stop_lifo_append() noexcept {
for (auto& kvp : qs_)
kvp.second.stop_lifo_append();
policy_.stop_lifo_append(kvp.second);
}
private:
......
......@@ -77,6 +77,27 @@ public:
telemetry::int_gauge* mailbox_size = nullptr;
};
/// Optional metrics for inbound stream traffic collected by individual actors
/// when configured to do so.
struct inbound_stream_metrics_t {
/// Counts the total number of processed stream elements from upstream.
telemetry::int_counter* processed_elements = nullptr;
/// Tracks how many stream elements from upstream are currently buffered.
telemetry::int_gauge* input_buffer_size = nullptr;
};
/// Optional metrics for outbound stream traffic collected by individual
/// actors when configured to do so.
struct outbound_stream_metrics_t {
/// Counts the total number of elements that have been pushed downstream.
telemetry::int_counter* pushed_elements = nullptr;
/// Tracks how many stream elements are currently waiting in the output
/// buffer due to insufficient credit.
telemetry::int_gauge* output_buffer_size = nullptr;
};
// -- constructors, destructors, and assignment operators --------------------
local_actor(actor_config& cfg);
......@@ -382,6 +403,11 @@ public:
return metrics_;
}
bool has_metrics_enabled() const noexcept {
// Either all fields are null or none is.
return metrics_.processing_time != nullptr;
}
template <class ActorHandle>
ActorHandle eval_opts(spawn_options opts, ActorHandle res) {
if (has_monitor_flag(opts))
......
......@@ -50,6 +50,18 @@ public:
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;
// -- constructors, destructors, and assignment operators ------------------
......@@ -72,6 +84,8 @@ public:
// -- member variables -----------------------------------------------------
handler_type handler;
size_t bulk_inserted_size = 0;
};
// -- member types -----------------------------------------------------------
......@@ -82,6 +96,8 @@ public:
using deficit_type = size_t;
using pointer = mapped_type*;
using unique_pointer = mailbox_element_ptr;
using key_type = stream_slot;
......@@ -96,8 +112,8 @@ public:
static bool enabled(const nested_queue_type& q) noexcept;
static deficit_type
quantum(const nested_queue_type& q, deficit_type x) noexcept;
static deficit_type quantum(const nested_queue_type& q,
deficit_type x) noexcept;
// -- constructors, destructors, and assignment operators --------------------
......@@ -116,6 +132,16 @@ public:
static task_size_type task_size(const mailbox_element&) noexcept {
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
......@@ -119,6 +119,14 @@ public:
/// Base type.
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 types to metric objects for outbound stream traffic.
using outbound_stream_metrics_map
= std::unordered_map<type_id_t, outbound_stream_metrics_t>;
/// Maps slot IDs to stream managers.
using stream_manager_map = std::map<stream_slot, stream_manager_ptr>;
......@@ -331,6 +339,12 @@ public:
return pending_stream_managers_;
}
// -- actor metrics ----------------------------------------------------------
inbound_stream_metrics_t inbound_stream_metrics(type_id_t type);
outbound_stream_metrics_t outbound_stream_metrics(type_id_t type);
// -- event handlers ---------------------------------------------------------
/// Sets a custom handler for unexpected messages.
......@@ -721,6 +735,12 @@ protected:
/// Pointer to a private thread object associated with a detached actor.
detail::private_thread* private_thread_;
/// Caches metric objects for inbound stream traffic.
inbound_stream_metrics_map inbound_stream_metrics_;
/// Caches metric objects for outbound stream traffic.
outbound_stream_metrics_map outbound_stream_metrics_;
#ifdef CAF_ENABLE_EXCEPTIONS
/// Customization point for setting a default exception callback.
exception_handler exception_handler_;
......
......@@ -254,6 +254,20 @@ auto make_actor_metric_families(telemetry::metric_registry& reg) {
"Time a message waits in the mailbox before processing.", "seconds"),
reg.gauge_family("caf.actor", "mailbox-size", {"name"},
"Number of messages in the mailbox."),
{
reg.counter_family("caf.actor.stream", "processed-elements",
{"name", "type"},
"Number of processed stream elements from upstream."),
reg.gauge_family("caf.actor.stream", "input-buffer-size",
{"name", "type"},
"Number of buffered stream elements from upstream."),
reg.counter_family(
"caf.actor.stream", "pushed-elements", {"name", "type"},
"Number of elements that have been pushed downstream."),
reg.gauge_family("caf.actor.stream", "output-buffer-size",
{"name", "type"},
"Number of buffered output stream elements."),
},
};
}
......
......@@ -22,6 +22,8 @@
#include "caf/logger.hpp"
#include "caf/outbound_path.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/stream_manager.hpp"
namespace caf {
......@@ -30,6 +32,14 @@ downstream_manager_base::downstream_manager_base(stream_manager* parent)
// nop
}
downstream_manager_base::downstream_manager_base(stream_manager* parent,
type_id_t type)
: super(parent) {
auto [pushed_elements, output_buffer_size]
= parent->self()->outbound_stream_metrics(type);
metrics_ = metrics_t{pushed_elements, output_buffer_size};
}
downstream_manager_base::~downstream_manager_base() {
// nop
}
......
......@@ -72,21 +72,25 @@ inbound_path::inbound_path(stream_manager_ptr mgr_ptr, stream_slots id,
strong_actor_ptr ptr,
[[maybe_unused]] type_id_t in_type)
: 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);
CAF_STREAM_LOG_DEBUG(mgr->self()->name()
CAF_STREAM_LOG_DEBUG(self->name()
<< "opens input stream with element type"
<< detail::global_meta_object(in_type)->type_name
<< "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")) {
if (*str == "testing")
controller_.reset(new detail::test_credit_controller(self()));
controller_.reset(new detail::test_credit_controller(self));
else if (*str == "size")
controller_.reset(new detail::size_based_credit_controller(self()));
controller_.reset(new detail::size_based_credit_controller(self));
else
controller_.reset(new detail::complexity_based_credit_controller(self()));
controller_.reset(new detail::complexity_based_credit_controller(self));
} 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) {
const auto& families = sys.actor_metric_families();
string_view sv{name, strlen(name)};
return {
families.processing_time_family->get_or_add({{"name", sv}}),
families.mailbox_time_family->get_or_add({{"name", sv}}),
families.mailbox_size_family->get_or_add({{"name", sv}}),
families.processing_time->get_or_add({{"name", sv}}),
families.mailbox_time->get_or_add({{"name", sv}}),
families.mailbox_size->get_or_add({{"name", sv}}),
};
}
......
......@@ -25,29 +25,18 @@
namespace caf::policy {
namespace {
class task_size_calculator {
public:
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 downstream_msg_batch& batch) noexcept -> task_size_type {
CAF_ASSERT(batch.xs_size > 0);
return batch.xs_size;
}
auto downstream_messages::nested::task_size(const mailbox_element& x) noexcept
-> task_size_type {
task_size_calculator f;
return visit(f, x.content().get_as<downstream_msg>(0).content);
CAF_ASSERT(x.mid.is_downstream_message());
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 {
......@@ -67,4 +56,47 @@ auto downstream_messages::quantum(const nested_queue_type& q,
return x * static_cast<deficit_type>(q.policy().handler->desired_batch_size);
}
void downstream_messages::cleanup(nested_queue_type& sub_queue) noexcept {
if (auto handler = sub_queue.policy().handler.get())
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
......@@ -22,6 +22,7 @@
#include "caf/actor_system_config.hpp"
#include "caf/config.hpp"
#include "caf/detail/default_invoke_result_visitor.hpp"
#include "caf/detail/meta_object.hpp"
#include "caf/detail/private_thread.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/inbound_path.hpp"
......@@ -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()(
size_t, downstream_queue& qs, stream_slot,
policy::downstream_messages::nested_queue_type& q, mailbox_element& x) {
......@@ -375,7 +330,48 @@ intrusive::task_result scheduled_actor::mailbox_visitor::operator()(
CAF_BEFORE_PROCESSING(self, x);
CAF_ASSERT(x.content().match_elements<downstream_msg>());
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);
CAF_AFTER_PROCESSING(self, invoke_message_result::consumed);
return ++handled_msgs < max_throughput ? res
......@@ -502,6 +498,48 @@ 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;
}
auto scheduled_actor::outbound_stream_metrics(type_id_t type)
-> outbound_stream_metrics_t {
if (!has_metrics_enabled())
return {nullptr, nullptr};
if (auto i = outbound_stream_metrics_.find(type);
i != outbound_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;
outbound_stream_metrics_t result{
fs.pushed_elements->get_or_add({{"name", actor_name}, {"type", tname}}),
fs.output_buffer_size->get_or_add({{"name", actor_name}, {"type", tname}}),
};
outbound_stream_metrics_.emplace(type, result);
return result;
}
// -- timeout management -------------------------------------------------------
uint64_t scheduled_actor::set_receive_timeout(actor_clock::time_point x) {
......
......@@ -98,6 +98,14 @@ struct inode_policy {
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;
};
......
......@@ -250,14 +250,34 @@ struct dmsg_queue_policy : policy_base {
}
template <class Queue>
static inline bool enabled(const Queue&) {
static inline bool enabled(const Queue&) noexcept {
return true;
}
template <class Queue>
deficit_type quantum(const Queue&, deficit_type x) {
deficit_type quantum(const Queue&, deficit_type x) noexcept {
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>;
......
......@@ -516,6 +516,26 @@ caf.mailbox-size
- **Type**: ``int_gauge``
- **Label dimensions**: name.
caf.stream.processed-elements
- Counts the total number of processed stream elements from upstream.
- **Type**: ``int_counter``
- **Label dimensions**: name, type.
caf.stream.input-buffer-size
- Tracks how many stream elements from upstream are currently buffered.
- **Type**: ``int_gauge``
- **Label dimensions**: name, type.
caf.stream.pushed-elements
- Counts the total number of elements that have been pushed downstream.
- **Type**: ``int_counter``
- **Label dimensions**: name, type.
caf.stream.output-buffer-size
- Tracks how many stream elements are currently waiting in the output buffer.
- **Type**: ``int_gauge``
- **Label dimensions**: name, type.
Exporting Metrics to Prometheus
-------------------------------
......
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