Commit 8dba0a98 authored by Dominik Charousset's avatar Dominik Charousset

Suppress stream timeouts for idle managers

Idle managers no longer trigger timeouts indefinitely. This new behavior
simplifies testing by terminating a run_dispatch_loop() once there is no
more activity. Also, this wastes fewer ressources overall by limiting
timeout messages in the system to what's actually needed.
parent 6bbac347
......@@ -144,6 +144,11 @@ public:
&& inbound_paths_.empty() && out_.clean();
}
bool idle() const noexcept override {
// Same as `stream_stage<...>`::idle().
return out_.stalled() || (out_.clean() && this->inbound_paths_up_to_date());
}
downstream_manager_type& out() override {
return out_;
}
......
......@@ -160,12 +160,16 @@ public:
/// desired batch size.
virtual void force_emit_batches();
/// Returns the currently available capacity for the output buffer.
/// Queries the currently available capacity for the output buffer.
virtual size_t capacity() const noexcept;
/// Returns the size of the output buffer.
/// Queries the size of the output buffer.
virtual size_t buffered() const noexcept;
/// Queries whether the manager cannot make any progress, because its buffer
/// is full and no more credit is available.
bool stalled() const noexcept;
/// Silently removes all paths.
virtual void clear_paths();
......
......@@ -147,6 +147,10 @@ public:
void emit_ack_batch(local_actor* self, long queued_items, timespan cycle,
timespan desired_batch_complexity);
/// Returns whether the path received no input since last emitting
/// `ack_batch`, i.e., `last_acked_batch_id == last_batch_id`.
bool up_to_date();
/// Sends a `stream_msg::close` on this path.
void emit_regular_shutdown(local_actor* self);
......
......@@ -100,10 +100,15 @@ public:
/// Returns the manager for downstream communication.
virtual downstream_manager& out() = 0;
/// Returns whether the stream has reached the end and can be discarded
/// Returns whether the manager has reached the end and can be discarded
/// safely.
virtual bool done() const = 0;
/// Returns whether the manager cannot make any progress on its own at the
/// moment. For example, a source is idle if it has filled its output buffer
/// and there isn't any credit left.
virtual bool idle() const noexcept = 0;
/// Advances time.
virtual void cycle_timeout(size_t cycle_nr);
......@@ -138,12 +143,16 @@ public:
}
/// Returns the list of inbound paths.
inline const inbound_paths_list& inbound_paths() const {
inline const inbound_paths_list& inbound_paths() const noexcept{
return inbound_paths_;
}
/// Returns the inbound paths at slot `x`.
inbound_path* get_inbound_path(stream_slot x) const;
inbound_path* get_inbound_path(stream_slot x) const noexcept;
/// Queries whether all inbound paths are up-to-date. A sink is idle if this
/// function returns `true`.
bool inbound_paths_up_to_date() const noexcept;
/// Returns the parent actor.
inline scheduled_actor* self() {
......
......@@ -19,8 +19,10 @@
#ifndef CAF_DETAIL_STREAM_SINK_HPP
#define CAF_DETAIL_STREAM_SINK_HPP
#include <algorithm>
#include <tuple>
#include "caf/inbound_path.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/stream_manager.hpp"
......@@ -47,6 +49,12 @@ public:
return !this->continuous() && this->inbound_paths_.empty();
}
bool idle() const noexcept override {
// A sink is idle if there's no pending batch and a new credit round would
// emit no `ack_batch` messages.
return this->inbound_paths_up_to_date();
}
downstream_manager& out() override {
return dummy_out_;
}
......
......@@ -43,6 +43,11 @@ public:
// nop
}
bool idle() const noexcept override {
// A source is idle if it can't make any progress on its downstream.
return out_.stalled();
}
DownstreamManager& out() override {
return out_;
}
......
......@@ -47,6 +47,13 @@ public:
// nop
}
bool idle() const noexcept override {
// A stage is idle if it can't make progress on its downstream manager or
// if it has no pending work at all.
auto& dm = this->out_;
return dm.stalled() || (dm.clean() && right_super::idle());
}
DownstreamManager& out() override {
return left_super::out();
}
......
......@@ -139,6 +139,13 @@ size_t downstream_manager::buffered() const noexcept {
return 0;
}
bool downstream_manager::stalled() const noexcept {
auto no_credit = [](const outbound_path& x) {
return x.open_credit == 0;
};
return capacity() == 0 && all_paths(no_credit);
}
void downstream_manager::clear_paths() {
// nop
}
......
......@@ -101,7 +101,7 @@ void inbound_path::emit_ack_batch(local_actor* self, long queued_items,
timespan cycle, timespan complexity) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(queued_items) << CAF_ARG(cycle)
<< CAF_ARG(complexity));
if (last_acked_batch_id == last_batch_id)
if (up_to_date())
return;
auto x = stats.calculate(cycle, complexity);
// Hand out enough credit to fill our queue for 2 cycles.
......@@ -113,12 +113,17 @@ void inbound_path::emit_ack_batch(local_actor* self, long queued_items,
assigned_credit += credit;
CAF_LOG_DEBUG(CAF_ARG(credit) << CAF_ARG(batch_size));
unsafe_send_as(self, hdl,
make<upstream_msg::ack_batch>(slots.invert(), self->address(),
make<upstream_msg::ack_batch>(slots.invert(),
self->address(),
static_cast<int32_t>(credit),
batch_size, last_batch_id));
last_acked_batch_id = last_batch_id;
}
bool inbound_path::up_to_date() {
return last_acked_batch_id == last_batch_id;
}
void inbound_path::emit_regular_shutdown(local_actor* self) {
CAF_LOG_TRACE(CAF_ARG(slots));
unsafe_send_as(self, hdl, make<upstream_msg::drop>(slots, self->address()));
......
......@@ -440,8 +440,23 @@ bool scheduled_actor::is_active_receive_timeout(uint64_t tid) const {
uint64_t scheduled_actor::set_stream_timeout(actor_clock::time_point x) {
CAF_LOG_TRACE(x);
// Do not request 'infinite' timeouts.
if (x == actor_clock::time_point::max())
return 0;
// Do not request a timeout if all streams are idle.
std::vector<stream_manager_ptr> mgrs;
for (auto& kvp : stream_managers_)
mgrs.emplace_back(kvp.second);
std::sort(mgrs.begin(), mgrs.end());
auto e = std::unique(mgrs.begin(), mgrs.end());
auto idle = [=](const stream_manager_ptr& x) {
return x->idle();
};
if (std::all_of(mgrs.begin(), e, idle)) {
CAF_LOG_DEBUG("suppress stream timeout");
return 0;
}
// Delegate call.
return set_timeout(stream_atom::value, x);
}
......
......@@ -175,7 +175,7 @@ void stream_manager::remove_input_path(stream_slot slot, error reason,
self_->erase_inbound_path_later(slot, std::move(reason));
}
inbound_path* stream_manager::get_inbound_path(stream_slot x) const {
inbound_path* stream_manager::get_inbound_path(stream_slot x) const noexcept {
auto pred = [=](inbound_path* ptr) {
return ptr->slots.receiver == x;
};
......@@ -184,6 +184,12 @@ inbound_path* stream_manager::get_inbound_path(stream_slot x) const {
return i != e ? *i : nullptr;
}
bool stream_manager::inbound_paths_up_to_date() const noexcept {
auto up_to_date = [](inbound_path* x) { return x->up_to_date(); };
return std::all_of(inbound_paths_.begin(), inbound_paths_.end(), up_to_date);
}
stream_slot stream_manager::add_unchecked_outbound_path_impl(response_promise& rp,
message handshake) {
CAF_LOG_TRACE(CAF_ARG(rp) << CAF_ARG(handshake));
......
......@@ -159,10 +159,7 @@ CAF_TEST(depth_3_pipeline_with_fork) {
sched.run();
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 1u);
auto predicate = [&] {
return st.stage->inbound_paths().empty() && st.stage->out().clean();
};
sched.run_dispatch_loop(predicate, streaming_cycle);
sched.run_dispatch_loop(streaming_cycle);
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 0u);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk1).state.x, 1275);
......@@ -186,10 +183,7 @@ CAF_TEST(depth_3_pipeline_with_join) {
sched.run();
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 1u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 2u);
auto predicate = [&] {
return st.stage->inbound_paths().empty() && st.stage->out().clean();
};
sched.run_dispatch_loop(predicate, streaming_cycle);
sched.run_dispatch_loop(streaming_cycle);
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 1u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 0u);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 2550);
......
......@@ -183,6 +183,10 @@ public:
&& out_.clean();
}
bool idle() const noexcept override {
return inbound_paths_up_to_date() && out_.stalled();
}
void handle(inbound_path*, downstream_msg::batch& batch) override {
using std::make_move_iterator;
using int_vec = std::vector<int>;
......
......@@ -306,8 +306,8 @@ CAF_TEST(depth_2_pipeline_500_items) {
CAF_MESSAGE("trigger timeouts");
sched.clock().current_time += credit_round_interval;
sched.dispatch();
expect((timeout_msg), from(snk).to(snk));
expect((timeout_msg), from(src).to(src));
allow((timeout_msg), from(snk).to(snk));
allow((timeout_msg), from(src).to(src));
CAF_MESSAGE("process ack_batch in source");
expect((upstream_msg::ack_batch), from(snk).to(src));
} while (!received<downstream_msg::close>(snk));
......@@ -370,9 +370,9 @@ CAF_TEST(depth_3_pipeline_50_items) {
auto next_cycle = [&] {
sched.clock().current_time += credit_round_interval;
sched.dispatch();
expect((timeout_msg), from(snk).to(snk));
expect((timeout_msg), from(stg).to(stg));
expect((timeout_msg), from(src).to(src));
allow((timeout_msg), from(snk).to(snk));
allow((timeout_msg), from(stg).to(stg));
allow((timeout_msg), from(src).to(src));
};
CAF_MESSAGE(CAF_ARG(self) << CAF_ARG(src) << CAF_ARG(stg) << CAF_ARG(snk));
CAF_MESSAGE("initiate stream handshake");
......
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