Commit 9869d7e0 authored by Dominik Charousset's avatar Dominik Charousset

Add inbound_path argument to congested

By passing the inbound path to the function, we enable stream managers
to fine-tune streaming logic if inbound paths are coupled to some but
not all outbound paths.
parent fdb812e9
...@@ -257,6 +257,16 @@ public: ...@@ -257,6 +257,16 @@ public:
return result; return result;
} }
size_t buffered(stream_slot slot) const noexcept override {
// We don't know which nested manager stores this path. Only one will give a
// valid answer, though. Everyone else always responds with 0. Hence, we can
// simply call all managers and sum up the results.
size_t result = 0;
for (auto ptr : ptrs_)
result += ptr->buffered(slot);
return result;
}
void clear_paths() override { void clear_paths() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
for (auto ptr : ptrs_) for (auto ptr : ptrs_)
......
...@@ -99,7 +99,7 @@ public: ...@@ -99,7 +99,7 @@ public:
/// Returns true if the handler is not able to process any further batches /// Returns true if the handler is not able to process any further batches
/// since it is unable to make progress sending on its own. /// since it is unable to make progress sending on its own.
virtual bool congested() const noexcept; virtual bool congested(const inbound_path& path) const noexcept;
/// Sends a handshake to `dest`. /// Sends a handshake to `dest`.
/// @pre `dest != nullptr` /// @pre `dest != nullptr`
......
...@@ -45,9 +45,10 @@ auto downstream_messages::id_of(mailbox_element& x) noexcept -> key_type { ...@@ -45,9 +45,10 @@ auto downstream_messages::id_of(mailbox_element& x) noexcept -> key_type {
} }
bool downstream_messages::enabled(const nested_queue_type& q) noexcept { bool downstream_messages::enabled(const nested_queue_type& q) noexcept {
auto congested = q.policy().handler->mgr->congested(); auto path = q.policy().handler.get();
CAF_LOG_DEBUG_IF(congested, "path is congested:" << CAF_ARG2( auto congested = path->mgr->congested(*path);
"slot", q.policy().handler->slots.receiver)); CAF_LOG_DEBUG_IF(congested, "path is congested:"
<< CAF_ARG2("slot", path->slots.receiver));
return !congested; return !congested;
} }
......
...@@ -154,7 +154,9 @@ void stream_manager::push() { ...@@ -154,7 +154,9 @@ void stream_manager::push() {
} while (generate_messages()); } while (generate_messages());
} }
bool stream_manager::congested() const noexcept { bool stream_manager::congested(const inbound_path&) const noexcept {
// By default, we assume all inbound paths write to the same output. Hence,
// the answer is independent of who is asking.
return out().capacity() == 0; return out().capacity() == 0;
} }
......
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