Commit 57eaa064 authored by Dominik Charousset's avatar Dominik Charousset

Collect metrics for outbound stream traffic

parent fed4e6c5
...@@ -219,9 +219,6 @@ public: ...@@ -219,9 +219,6 @@ public:
/// 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 = nullptr; telemetry::int_counter_family* pushed_elements = nullptr;
/// Counts the total number of batches that have been pushed downstream.
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 = nullptr; telemetry::int_gauge_family* output_buffer_size = nullptr;
......
...@@ -58,12 +58,25 @@ public: ...@@ -58,12 +58,25 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- 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 // nop
} }
// -- properties ------------------------------------------------------------- // -- 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 { size_t buffered() const noexcept override {
// We have a central buffer, but also an additional buffer at each path. We // We have a central buffer, but also an additional buffer at each path. We
// return the maximum size to reflect the current worst case. // return the maximum size to reflect the current worst case.
...@@ -195,6 +208,7 @@ public: ...@@ -195,6 +208,7 @@ public:
/// Forces the manager flush its buffer to the individual path buffers. /// Forces the manager flush its buffer to the individual path buffers.
void fan_out_flush() { void fan_out_flush() {
auto old_size = buffered();
auto& buf = this->buf_; auto& buf = this->buf_;
auto f = [&](typename map_type::value_type& x, auto f = [&](typename map_type::value_type& x,
typename state_map_type::value_type& y) { typename state_map_type::value_type& y) {
...@@ -203,8 +217,7 @@ public: ...@@ -203,8 +217,7 @@ public:
return; return;
// Push data from the global buffer to path buffers. // Push data from the global buffer to path buffers.
auto& st = y.second; auto& st = y.second;
// TODO: replace with `if constexpr` when switching to C++17 if constexpr (std::is_same<select_type, detail::select_all>::value) {
if (std::is_same<select_type, detail::select_all>::value) {
st.buf.insert(st.buf.end(), buf.begin(), buf.end()); st.buf.insert(st.buf.end(), buf.begin(), buf.end());
} else { } else {
for (auto& piece : buf) for (auto& piece : buf)
...@@ -214,6 +227,10 @@ public: ...@@ -214,6 +227,10 @@ public:
}; };
detail::zip_foreach(f, this->paths_.container(), state_map_.container()); detail::zip_foreach(f, this->paths_.container(), state_map_.container());
buf.clear(); 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: protected:
...@@ -230,6 +247,7 @@ private: ...@@ -230,6 +247,7 @@ private:
CAF_ASSERT(this->paths_.size() <= state_map_.size()); CAF_ASSERT(this->paths_.size() <= state_map_.size());
if (this->paths_.empty()) if (this->paths_.empty())
return; return;
auto old_size = buffered();
// Calculate the chunk size, i.e., how many more items we can put to our // Calculate the chunk size, i.e., how many more items we can put to our
// caches at the most. // caches at the most.
auto not_closing = [&](typename map_type::value_type& x, auto not_closing = [&](typename map_type::value_type& x,
...@@ -256,7 +274,6 @@ private: ...@@ -256,7 +274,6 @@ private:
detail::zip_foreach(g, this->paths_.container(), state_map_.container()); detail::zip_foreach(g, this->paths_.container(), state_map_.container());
return; return;
} }
auto chunk = this->get_chunk(chunk_size); auto chunk = this->get_chunk(chunk_size);
if (chunk.empty()) { if (chunk.empty()) {
auto g = [&](typename map_type::value_type& x, auto g = [&](typename map_type::value_type& x,
...@@ -288,10 +305,14 @@ private: ...@@ -288,10 +305,14 @@ private:
}; };
detail::zip_foreach(g, this->paths_.container(), state_map_.container()); 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_; state_map_type state_map_;
select_type select_; select_type select_;
}; };
} // namespace caf } // namespace caf
...@@ -53,9 +53,15 @@ public: ...@@ -53,9 +53,15 @@ public:
// nop // nop
} }
buffered_downstream_manager(stream_manager* parent, type_id_t type)
: super(parent, type) {
// nop
}
template <class T0, class... Ts> template <class T0, class... Ts>
void push(T0&& x, Ts&&... xs) { void push(T0&& x, Ts&&... xs) {
buf_.emplace_back(std::forward<T0>(x), std::forward<Ts>(xs)...); buf_.emplace_back(std::forward<T0>(x), std::forward<Ts>(xs)...);
this->generated_messages(1);
} }
/// @pre `n <= buf_.size()` /// @pre `n <= buf_.size()`
......
...@@ -68,11 +68,14 @@ public: ...@@ -68,11 +68,14 @@ public:
CAF_LOG_DEBUG(CAF_ARG(hint)); CAF_LOG_DEBUG(CAF_ARG(hint));
if (hint == 0) if (hint == 0)
return false; return false;
auto old_size = this->out_.buf().size();
downstream<typename Driver::output_type> ds{this->out_.buf()}; downstream<typename Driver::output_type> ds{this->out_.buf()};
driver_.pull(ds, hint); driver_.pull(ds, hint);
if (driver_.done()) if (driver_.done())
at_end_ = true; 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: protected:
......
...@@ -63,8 +63,11 @@ public: ...@@ -63,8 +63,11 @@ public:
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
using vec_type = std::vector<input_type>; using vec_type = std::vector<input_type>;
if (auto view = make_typed_message_view<vec_type>(x.xs)) { 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()}; downstream<output_type> ds{this->out_.buf()};
driver_.process(ds, get<0>(view)); driver_.process(ds, get<0>(view));
auto new_size = this->out_.buf().size();
this->out_.generated_messages(new_size - old_size);
return; return;
} }
CAF_LOG_ERROR("received unexpected batch type (dropped)"); CAF_LOG_ERROR("received unexpected batch type (dropped)");
......
...@@ -25,6 +25,8 @@ ...@@ -25,6 +25,8 @@
#include "caf/detail/unordered_flat_map.hpp" #include "caf/detail/unordered_flat_map.hpp"
#include "caf/downstream_manager.hpp" #include "caf/downstream_manager.hpp"
#include "caf/outbound_path.hpp" #include "caf/outbound_path.hpp"
#include "caf/telemetry/counter.hpp"
#include "caf/telemetry/gauge.hpp"
namespace caf { namespace caf {
...@@ -40,10 +42,22 @@ public: ...@@ -40,10 +42,22 @@ public:
/// Maps slots to paths. /// Maps slots to paths.
using map_type = detail::unordered_flat_map<stream_slot, unique_path_ptr>; 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 -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit downstream_manager_base(stream_manager* parent); explicit downstream_manager_base(stream_manager* parent);
downstream_manager_base(stream_manager* parent, type_id_t type);
~downstream_manager_base() override; ~downstream_manager_base() override;
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
...@@ -60,24 +74,45 @@ public: ...@@ -60,24 +74,45 @@ public:
size_t num_paths() const noexcept override; size_t num_paths() const noexcept override;
bool bool remove_path(stream_slot slots, error reason,
remove_path(stream_slot slots, error reason, bool silent) noexcept override; bool silent) noexcept override;
path_ptr path(stream_slot slots) noexcept override; path_ptr path(stream_slot slots) noexcept override;
void clear_paths() 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: protected:
bool insert_path(unique_path_ptr ptr) override; bool insert_path(unique_path_ptr ptr) override;
void for_each_path_impl(path_visitor& f) override; void for_each_path_impl(path_visitor& f) override;
bool check_paths_impl(path_algorithm algo, path_predicate& pred) const bool check_paths_impl(path_algorithm algo,
noexcept override; path_predicate& pred) const noexcept override;
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
map_type paths_; map_type paths_;
metrics_t metrics_;
}; };
} // namespace caf } // namespace caf
...@@ -93,9 +93,6 @@ public: ...@@ -93,9 +93,6 @@ public:
/// 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* pushed_elements = nullptr; telemetry::int_counter* pushed_elements = nullptr;
/// Counts the total number of batches that have been pushed downstream.
telemetry::int_counter* 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* output_buffer_size = nullptr; telemetry::int_gauge* output_buffer_size = nullptr;
......
...@@ -123,6 +123,10 @@ public: ...@@ -123,6 +123,10 @@ public:
using inbound_stream_metrics_map using inbound_stream_metrics_map
= std::unordered_map<type_id_t, inbound_stream_metrics_t>; = 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. /// 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>;
...@@ -339,6 +343,8 @@ public: ...@@ -339,6 +343,8 @@ public:
inbound_stream_metrics_t inbound_stream_metrics(type_id_t type); inbound_stream_metrics_t inbound_stream_metrics(type_id_t type);
outbound_stream_metrics_t outbound_stream_metrics(type_id_t type);
// -- event handlers --------------------------------------------------------- // -- event handlers ---------------------------------------------------------
/// Sets a custom handler for unexpected messages. /// Sets a custom handler for unexpected messages.
...@@ -732,6 +738,9 @@ protected: ...@@ -732,6 +738,9 @@ protected:
/// Catche metric objects for inbound stream traffic. /// Catche metric objects for inbound stream traffic.
inbound_stream_metrics_map inbound_stream_metrics_; inbound_stream_metrics_map inbound_stream_metrics_;
/// Catche metric objects for outbound stream traffic.
outbound_stream_metrics_map outbound_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_;
......
...@@ -264,8 +264,6 @@ auto make_actor_metric_families(telemetry::metric_registry& reg) { ...@@ -264,8 +264,6 @@ auto make_actor_metric_families(telemetry::metric_registry& reg) {
reg.counter_family( reg.counter_family(
"caf.actor.stream", "pushed-elements", {"name", "type"}, "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", "type"},
"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", "type"}, {"name", "type"},
"Number of buffered output stream elements."), "Number of buffered output stream elements."),
......
...@@ -22,6 +22,8 @@ ...@@ -22,6 +22,8 @@
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/outbound_path.hpp" #include "caf/outbound_path.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/stream_manager.hpp"
namespace caf { namespace caf {
...@@ -30,6 +32,14 @@ downstream_manager_base::downstream_manager_base(stream_manager* parent) ...@@ -30,6 +32,14 @@ downstream_manager_base::downstream_manager_base(stream_manager* parent)
// nop // 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() { downstream_manager_base::~downstream_manager_base() {
// nop // nop
} }
......
...@@ -520,6 +520,26 @@ auto scheduled_actor::inbound_stream_metrics(type_id_t type) ...@@ -520,6 +520,26 @@ auto scheduled_actor::inbound_stream_metrics(type_id_t type)
return 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 ------------------------------------------------------- // -- 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) {
......
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