Commit 144c66b0 authored by Dominik Charousset's avatar Dominik Charousset

Integrate review feedback

parent 10a861f3
...@@ -735,10 +735,10 @@ protected: ...@@ -735,10 +735,10 @@ 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. /// Caches 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. /// Caches metric objects for outbound stream traffic.
outbound_stream_metrics_map outbound_stream_metrics_; outbound_stream_metrics_map outbound_stream_metrics_;
#ifdef CAF_ENABLE_EXCEPTIONS #ifdef CAF_ENABLE_EXCEPTIONS
......
...@@ -258,7 +258,7 @@ auto make_actor_metric_families(telemetry::metric_registry& reg) { ...@@ -258,7 +258,7 @@ auto make_actor_metric_families(telemetry::metric_registry& reg) {
reg.counter_family("caf.actor.stream", "processed-elements", reg.counter_family("caf.actor.stream", "processed-elements",
{"name", "type"}, {"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", "type"}, {"name", "type"},
"Number of buffered stream elements from upstream."), "Number of buffered stream elements from upstream."),
reg.counter_family( reg.counter_family(
......
...@@ -57,9 +57,7 @@ auto downstream_messages::quantum(const nested_queue_type& q, ...@@ -57,9 +57,7 @@ auto downstream_messages::quantum(const nested_queue_type& q,
} }
void downstream_messages::cleanup(nested_queue_type& sub_queue) noexcept { void downstream_messages::cleanup(nested_queue_type& sub_queue) noexcept {
auto& handler = sub_queue.policy().handler; if (auto handler = sub_queue.policy().handler.get())
if (!handler)
return;
if (auto input_buffer_size = handler->metrics.input_buffer_size) if (auto input_buffer_size = handler->metrics.input_buffer_size)
input_buffer_size->dec(sub_queue.total_task_size()); input_buffer_size->dec(sub_queue.total_task_size());
} }
......
...@@ -989,12 +989,11 @@ void scheduled_actor::erase_inbound_path_later(stream_slot slot, error reason) { ...@@ -989,12 +989,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);
} }
} }
......
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