Commit 9d4614f3 authored by Dominik Charousset's avatar Dominik Charousset

Allow stream managers to restrict credit

parent 757fcca4
......@@ -134,7 +134,7 @@ public:
bool idle() const noexcept override {
// Same as `stream_stage<...>`::idle().
return out_.stalled() || (out_.clean() && this->inbound_paths_up_to_date());
return out_.stalled() || (out_.clean() && this->inbound_paths_idle());
}
downstream_manager_type& out() override {
......
......@@ -58,6 +58,10 @@ public:
CAF_LOG_ERROR("received unexpected batch type (dropped)");
}
int32_t acquire_credit(inbound_path* path, int32_t desired) override {
return driver_.acquire_credit(path, desired);
}
bool congested() const noexcept override {
return driver_.congested();
}
......
......@@ -77,6 +77,10 @@ public:
return this->out_.capacity() == 0;
}
int32_t acquire_credit(inbound_path* path, int32_t desired) override {
return driver_.acquire_credit(path, desired);
}
protected:
void finalize(const error& reason) override {
driver_.finalize(reason);
......
......@@ -79,13 +79,14 @@ public:
bool force_underfull) {
CAF_LOG_TRACE(CAF_ARG(force_underfull));
using type = detail::decay_t<decltype(*i)>;
// Signed Desired Batch Size.
// Ship full batches.
while (std::distance(i, e) >= desired_batch_size) {
std::vector<type> tmp{std::make_move_iterator(i),
std::make_move_iterator(i + desired_batch_size)};
emit_batch(self, desired_batch_size, make_message(std::move(tmp)));
i += desired_batch_size;
}
// Ship underful batch only if `force_underful` is set.
if (i != e && force_underfull) {
std::vector<type> tmp{std::make_move_iterator(i),
std::make_move_iterator(e)};
......
......@@ -148,15 +148,20 @@ public:
/// Returns the inbound paths at slot `x`.
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;
/// Queries whether all inbound paths are up-to-date and have non-zero
/// credit. A sink is idle if this function returns `true`.
bool inbound_paths_idle() const noexcept;
/// Returns the parent actor.
inline scheduled_actor* self() {
return self_;
}
/// Acquires credit on an inbound path. The calculated credit to fill our
/// queue fro two cycles is `desired`, but the manager is allowed to return
/// any non-negative value.
virtual int32_t acquire_credit(inbound_path* path, int32_t desired);
/// Creates an outbound path to the current sender without any type checking.
/// @pre `out().terminal() == false`
/// @private
......
......@@ -51,7 +51,7 @@ public:
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();
return this->inbound_paths_idle();
}
downstream_manager& out() override {
......
......@@ -66,6 +66,14 @@ public:
virtual bool congested() const noexcept {
return false;
}
/// Acquires credit on an inbound path. The calculated credit to fill our
/// queue fro two cycles is `desired`, but the driver is allowed to return
/// any non-negative value.
virtual int32_t acquire_credit(inbound_path* path, int32_t desired) {
CAF_IGNORE_UNUSED(path);
return desired;
}
};
} // namespace caf
......
......@@ -67,6 +67,14 @@ public:
virtual void finalize(const error&) {
// nop
}
/// Acquires credit on an inbound path. The calculated credit to fill our
/// queue fro two cycles is `desired`, but the driver is allowed to return
/// any non-negative value.
virtual int32_t acquire_credit(inbound_path* path, int32_t desired) {
CAF_IGNORE_UNUSED(path);
return desired;
}
};
} // namespace caf
......
......@@ -90,7 +90,7 @@ void inbound_path::handle(downstream_msg::batch& x) {
void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(rebind_from));
// Update state.
assigned_credit = initial_credit;
assigned_credit = mgr->acquire_credit(this, initial_credit);
// Make sure we receive errors from this point on.
stream_aborter::add(hdl, self->address(), slots.receiver,
stream_aborter::source_aborter);
......@@ -98,23 +98,25 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) {
unsafe_send_as(self, hdl,
make<upstream_msg::ack_open>(
slots.invert(), self->address(), std::move(rebind_from),
self->ctrl(), static_cast<int32_t>(assigned_credit),
desired_batch_size));
self->ctrl(), assigned_credit, desired_batch_size));
}
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 (up_to_date())
return;
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 + queued_items),
0l);
// The manager can restrict or adjust the amount of credit.
credit = mgr->acquire_credit(this, credit);
if (credit == 0 && up_to_date()) {
return;
}
desired_batch_size = static_cast<int32_t>(x.items_per_batch);
if (credit != 0)
if (credit > 0)
assigned_credit += credit;
CAF_LOG_DEBUG(CAF_ARG(credit) << CAF_ARG(desired_batch_size));
unsafe_send_as(self, hdl,
......
......@@ -1098,6 +1098,7 @@ scheduled_actor::advance_streams(actor_clock::time_point now) {
}
// Fill up credit on each input path.
if ((bitmask & 0x02) != 0) {
CAF_LOG_DEBUG("new credit round");
auto cycle = stream_ticks_.interval();
cycle *= static_cast<decltype(cycle)::rep>(credit_round_ticks_);
auto bc_us = home_system().config().streaming_desired_batch_complexity_us;
......
......@@ -187,9 +187,15 @@ inbound_path* stream_manager::get_inbound_path(stream_slot x) const noexcept {
}
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);
bool stream_manager::inbound_paths_idle() const noexcept {
auto f = [](inbound_path* x) {
return x->up_to_date() && x->assigned_credit > 0;
};
return std::all_of(inbound_paths_.begin(), inbound_paths_.end(), f);
}
int32_t stream_manager::acquire_credit(inbound_path*, int32_t desired) {
return desired;
}
stream_slot stream_manager::add_unchecked_outbound_path_impl(response_promise& rp,
......
......@@ -184,7 +184,7 @@ public:
}
bool idle() const noexcept override {
return inbound_paths_up_to_date() && out_.stalled();
return inbound_paths_idle() && out_.stalled();
}
void handle(inbound_path*, downstream_msg::batch& batch) override {
......
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