Commit 1377a01d authored by Dominik Charousset's avatar Dominik Charousset

Consider downstream capacity for credit decisions

parent a317d135
......@@ -81,6 +81,18 @@ public:
+ (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`
/// is a new path.
void set_filter(stream_slot slot, filter_type new_filter) {
......
......@@ -192,6 +192,9 @@ public:
/// Queries an estimate of the size of the output buffer for `slot`.
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
/// is full and no more credit is available.
bool stalled() const noexcept;
......
......@@ -137,7 +137,8 @@ public:
/// waiting in the mailbox.
/// @param cycle Time between credit rounds.
/// @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);
/// Returns whether the path received no input since last emitting
......
......@@ -170,6 +170,9 @@ public:
/// ACKs, i.e., receiving an ACK with a higher ID is not an error.
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
/// `downstream_manger` no longer sends new batches to a closing path, but
/// buffered batches are still shipped. The owning `stream_manager` removes
......
......@@ -28,6 +28,7 @@
#include "caf/message.hpp"
#include "caf/stream_priority.hpp"
#include "caf/stream_slot.hpp"
#include "caf/timespan.hpp"
#include "caf/variant.hpp"
#include "caf/tag/boxing_type.hpp"
......@@ -76,6 +77,10 @@ struct upstream_msg : tag::boxing_type {
/// Cumulative ack 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
......
......@@ -199,6 +199,10 @@ size_t downstream_manager::buffered(stream_slot) const noexcept {
return 0;
}
int32_t downstream_manager::max_capacity() const noexcept {
return std::numeric_limits<int32_t>::max();
}
bool downstream_manager::stalled() const noexcept {
auto no_credit = [](const outbound_path& x) {
return x.open_credit == 0;
......
......@@ -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,
int32_t max_downstream_capacity,
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_IGNORE_UNUSED(queued_items);
auto x = stats.calculate(cycle, complexity);
// Hand out enough credit to fill our queue for 2 cycles.
auto credit = std::max((x.max_throughput * 2) - assigned_credit, 0);
// Hand out enough credit to fill our queue for 2 cycles but never exceed
// 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`.
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.
credit = std::min(mgr->acquire_credit(this, credit), max_new_credit);
if (credit == 0 && up_to_date())
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)
assigned_credit += credit;
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,
make<upstream_msg::ack_batch>(slots.invert(),
self->address(),
make<upstream_msg::ack_batch>(slots.invert(), self->address(),
static_cast<int32_t>(credit),
desired_batch_size,
last_batch_id));
last_batch_id, max_capacity));
last_acked_batch_id = last_batch_id;
}
......
......@@ -40,6 +40,7 @@ outbound_path::outbound_path(stream_slot sender_slot,
open_credit(0),
desired_batch_size(50),
next_ack_id(1),
max_capacity(0),
closing(false) {
// nop
}
......
......@@ -1170,7 +1170,8 @@ scheduled_actor::advance_streams(actor_clock::time_point now) {
for (auto& kvp : qs) {
auto inptr = kvp.second.policy().handler.get();
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_,
......
......@@ -102,7 +102,9 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) {
auto path = out().path(slots.receiver);
if (path != nullptr) {
path->open_credit += x.new_capacity;
path->max_capacity = x.max_capacity;
CAF_ASSERT(path->open_credit >= 0);
CAF_ASSERT(path->max_capacity >= 0);
path->set_desired_batch_size(x.desired_batch_size);
path->next_ack_id = x.acknowledged_id + 1;
// Gravefully remove path after receiving its final ACK.
......@@ -158,7 +160,7 @@ void stream_manager::advance() {
// Ignore inbound paths of other managers.
if (inptr->mgr.get() == this) {
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:
for (auto& kvp : qs) {
auto inptr = kvp.second.policy().handler.get();
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