Commit 6984277f authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'issue/800'

Close #800.
parents a317d135 1377a01d
...@@ -81,6 +81,18 @@ public: ...@@ -81,6 +81,18 @@ public:
+ (i != state_map_.end() ? i->second.buf.size() : 0u); + (i != state_map_.end() ? i->second.buf.size() : 0u);
} }
int32_t max_capacity() const noexcept override {
// The maximum capacity is limited by the slowest downstream path.
auto result = std::numeric_limits<int32_t>::max();
for (auto& kvp : this->paths_) {
auto mc = kvp.second->max_capacity;
// max_capacity is 0 if and only if we didn't receive an ack_batch yet.
if (mc > 0)
result = std::min(result, mc);
}
return result;
}
/// Sets the filter for `slot` to `filter`. Inserts a new element if `slot` /// Sets the filter for `slot` to `filter`. Inserts a new element if `slot`
/// is a new path. /// is a new path.
void set_filter(stream_slot slot, filter_type new_filter) { void set_filter(stream_slot slot, filter_type new_filter) {
......
...@@ -192,6 +192,9 @@ public: ...@@ -192,6 +192,9 @@ public:
/// Queries an estimate of the size of the output buffer for `slot`. /// Queries an estimate of the size of the output buffer for `slot`.
virtual size_t buffered(stream_slot slot) const noexcept; virtual size_t buffered(stream_slot slot) const noexcept;
/// Computes the maximum available downstream capacity.
virtual int32_t max_capacity() const noexcept;
/// Queries whether the manager cannot make any progress, because its buffer /// Queries whether the manager cannot make any progress, because its buffer
/// is full and no more credit is available. /// is full and no more credit is available.
bool stalled() const noexcept; bool stalled() const noexcept;
......
...@@ -137,7 +137,8 @@ public: ...@@ -137,7 +137,8 @@ public:
/// waiting in the mailbox. /// waiting in the mailbox.
/// @param cycle Time between credit rounds. /// @param cycle Time between credit rounds.
/// @param desired_batch_complexity Desired processing time per batch. /// @param desired_batch_complexity Desired processing time per batch.
void emit_ack_batch(local_actor* self, int32_t queued_items, timespan cycle, void emit_ack_batch(local_actor* self, int32_t queued_items,
int32_t max_downstream_capacity, timespan cycle,
timespan desired_batch_complexity); timespan desired_batch_complexity);
/// Returns whether the path received no input since last emitting /// Returns whether the path received no input since last emitting
......
...@@ -170,6 +170,9 @@ public: ...@@ -170,6 +170,9 @@ public:
/// ACKs, i.e., receiving an ACK with a higher ID is not an error. /// ACKs, i.e., receiving an ACK with a higher ID is not an error.
int64_t next_ack_id; int64_t next_ack_id;
/// Stores the maximum capacity of the downstream actor.
int32_t max_capacity;
/// Stores whether an outbound path is marked for removal. The /// Stores whether an outbound path is marked for removal. The
/// `downstream_manger` no longer sends new batches to a closing path, but /// `downstream_manger` no longer sends new batches to a closing path, but
/// buffered batches are still shipped. The owning `stream_manager` removes /// buffered batches are still shipped. The owning `stream_manager` removes
......
...@@ -28,6 +28,7 @@ ...@@ -28,6 +28,7 @@
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/stream_priority.hpp" #include "caf/stream_priority.hpp"
#include "caf/stream_slot.hpp" #include "caf/stream_slot.hpp"
#include "caf/timespan.hpp"
#include "caf/variant.hpp" #include "caf/variant.hpp"
#include "caf/tag/boxing_type.hpp" #include "caf/tag/boxing_type.hpp"
...@@ -76,6 +77,10 @@ struct upstream_msg : tag::boxing_type { ...@@ -76,6 +77,10 @@ struct upstream_msg : tag::boxing_type {
/// Cumulative ack ID. /// Cumulative ack ID.
int64_t acknowledged_id; int64_t acknowledged_id;
/// Maximum capacity on this path. Stages can consider this metric for
/// downstream actors when calculating their own maximum capactiy.
int32_t max_capacity;
}; };
/// Asks the source to discard any remaining credit and close this path /// Asks the source to discard any remaining credit and close this path
......
...@@ -199,6 +199,10 @@ size_t downstream_manager::buffered(stream_slot) const noexcept { ...@@ -199,6 +199,10 @@ size_t downstream_manager::buffered(stream_slot) const noexcept {
return 0; return 0;
} }
int32_t downstream_manager::max_capacity() const noexcept {
return std::numeric_limits<int32_t>::max();
}
bool downstream_manager::stalled() const noexcept { bool downstream_manager::stalled() const noexcept {
auto no_credit = [](const outbound_path& x) { auto no_credit = [](const outbound_path& x) {
return x.open_credit == 0; return x.open_credit == 0;
......
...@@ -112,29 +112,37 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) { ...@@ -112,29 +112,37 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) {
} }
void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items, void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items,
int32_t max_downstream_capacity,
timespan cycle, timespan complexity) { timespan cycle, timespan complexity) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(queued_items) << CAF_ARG(cycle) CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(queued_items)
<< CAF_ARG(max_downstream_capacity) << CAF_ARG(cycle)
<< CAF_ARG(complexity)); << CAF_ARG(complexity));
CAF_IGNORE_UNUSED(queued_items); CAF_IGNORE_UNUSED(queued_items);
auto x = stats.calculate(cycle, complexity); auto x = stats.calculate(cycle, complexity);
// Hand out enough credit to fill our queue for 2 cycles. // Hand out enough credit to fill our queue for 2 cycles but never exceed
auto credit = std::max((x.max_throughput * 2) - assigned_credit, 0); // the downstream capacity.
auto max_capacity = std::min(x.max_throughput * 2, max_downstream_capacity);
CAF_ASSERT(max_capacity > 0);
// Protect against overflow on `assigned_credit`. // Protect against overflow on `assigned_credit`.
auto max_new_credit = std::numeric_limits<int32_t>::max() - assigned_credit; auto max_new_credit = std::numeric_limits<int32_t>::max() - assigned_credit;
// Compute the amount of credit we grant in this round.
auto credit = std::min(std::max(max_capacity - assigned_credit, 0),
max_new_credit);
// The manager can restrict or adjust the amount of credit. // The manager can restrict or adjust the amount of credit.
credit = std::min(mgr->acquire_credit(this, credit), max_new_credit); credit = std::min(mgr->acquire_credit(this, credit), max_new_credit);
if (credit == 0 && up_to_date()) if (credit == 0 && up_to_date())
return; return;
CAF_LOG_DEBUG(CAF_ARG(assigned_credit) << CAF_ARG(max_capacity)
<< CAF_ARG(queued_items) << CAF_ARG(credit)
<< CAF_ARG(desired_batch_size));
if (credit > 0) if (credit > 0)
assigned_credit += credit; assigned_credit += credit;
desired_batch_size = static_cast<int32_t>(x.items_per_batch); desired_batch_size = static_cast<int32_t>(x.items_per_batch);
CAF_LOG_DEBUG(CAF_ARG(credit) << CAF_ARG(desired_batch_size));
unsafe_send_as(self, hdl, unsafe_send_as(self, hdl,
make<upstream_msg::ack_batch>(slots.invert(), make<upstream_msg::ack_batch>(slots.invert(), self->address(),
self->address(),
static_cast<int32_t>(credit), static_cast<int32_t>(credit),
desired_batch_size, desired_batch_size,
last_batch_id)); last_batch_id, max_capacity));
last_acked_batch_id = last_batch_id; last_acked_batch_id = last_batch_id;
} }
......
...@@ -40,6 +40,7 @@ outbound_path::outbound_path(stream_slot sender_slot, ...@@ -40,6 +40,7 @@ outbound_path::outbound_path(stream_slot sender_slot,
open_credit(0), open_credit(0),
desired_batch_size(50), desired_batch_size(50),
next_ack_id(1), next_ack_id(1),
max_capacity(0),
closing(false) { closing(false) {
// nop // nop
} }
......
...@@ -1170,7 +1170,8 @@ scheduled_actor::advance_streams(actor_clock::time_point now) { ...@@ -1170,7 +1170,8 @@ scheduled_actor::advance_streams(actor_clock::time_point now) {
for (auto& kvp : qs) { for (auto& kvp : qs) {
auto inptr = kvp.second.policy().handler.get(); auto inptr = kvp.second.policy().handler.get();
auto bs = static_cast<int32_t>(kvp.second.total_task_size()); auto bs = static_cast<int32_t>(kvp.second.total_task_size());
inptr->emit_ack_batch(this, bs, cycle, bc); inptr->emit_ack_batch(this, bs, inptr->mgr->out().max_capacity(),
cycle, bc);
} }
} }
return stream_ticks_.next_timeout(now, {max_batch_delay_ticks_, return stream_ticks_.next_timeout(now, {max_batch_delay_ticks_,
......
...@@ -102,7 +102,9 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) { ...@@ -102,7 +102,9 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) {
auto path = out().path(slots.receiver); auto path = out().path(slots.receiver);
if (path != nullptr) { if (path != nullptr) {
path->open_credit += x.new_capacity; path->open_credit += x.new_capacity;
path->max_capacity = x.max_capacity;
CAF_ASSERT(path->open_credit >= 0); CAF_ASSERT(path->open_credit >= 0);
CAF_ASSERT(path->max_capacity >= 0);
path->set_desired_batch_size(x.desired_batch_size); path->set_desired_batch_size(x.desired_batch_size);
path->next_ack_id = x.acknowledged_id + 1; path->next_ack_id = x.acknowledged_id + 1;
// Gravefully remove path after receiving its final ACK. // Gravefully remove path after receiving its final ACK.
...@@ -158,7 +160,7 @@ void stream_manager::advance() { ...@@ -158,7 +160,7 @@ void stream_manager::advance() {
// Ignore inbound paths of other managers. // Ignore inbound paths of other managers.
if (inptr->mgr.get() == this) { if (inptr->mgr.get() == this) {
auto bs = static_cast<int32_t>(kvp.second.total_task_size()); auto bs = static_cast<int32_t>(kvp.second.total_task_size());
inptr->emit_ack_batch(self_, bs, interval, bc); inptr->emit_ack_batch(self_, bs, out().max_capacity(), interval, bc);
} }
} }
} }
......
...@@ -340,7 +340,7 @@ public: ...@@ -340,7 +340,7 @@ public:
for (auto& kvp : qs) { for (auto& kvp : qs) {
auto inptr = kvp.second.policy().handler.get(); auto inptr = kvp.second.policy().handler.get();
auto bs = static_cast<int32_t>(kvp.second.total_task_size()); auto bs = static_cast<int32_t>(kvp.second.total_task_size());
inptr->emit_ack_batch(this, bs, cycle, desired_batch_complexity); inptr->emit_ack_batch(this, bs, 30, cycle, desired_batch_complexity);
} }
} }
}; };
......
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