Commit 3ccf2d6e authored by Dominik Charousset's avatar Dominik Charousset

Remove propagation of downstream buffer sizes

Stream stages should adjust their buffers according to local
configuration rather than using buffer sizes of the connected sinks. In
particular, since stages usually perform little work and as a result can
run with small buffers.
parent 10a861f3
...@@ -94,18 +94,6 @@ public: ...@@ -94,18 +94,6 @@ public:
return i != state_map_.end() ? i->second.buf.size() : 0u; return 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) {
......
...@@ -70,10 +70,7 @@ public: ...@@ -70,10 +70,7 @@ public:
/// Assigs new credit to the source after a cycle ends. /// Assigs new credit to the source after a cycle ends.
/// @param cycle Duration of a cycle. /// @param cycle Duration of a cycle.
/// @param max_downstream_credit Maximum downstream capacity as reported by virtual assignment compute(timespan cycle) = 0;
/// the downstream manager. Controllers may use
/// this capacity as an upper bound.
virtual assignment compute(timespan cycle, int32_t max_downstream_credit) = 0;
// -- virtual functions ------------------------------------------------------ // -- virtual functions ------------------------------------------------------
......
...@@ -53,7 +53,7 @@ public: ...@@ -53,7 +53,7 @@ public:
assignment compute_initial() override; assignment compute_initial() override;
assignment compute(timespan cycle, int32_t) override; assignment compute(timespan cycle) override;
private: private:
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
......
...@@ -58,7 +58,7 @@ public: ...@@ -58,7 +58,7 @@ public:
assignment compute_initial() override; assignment compute_initial() override;
assignment compute(timespan cycle, int32_t) override; assignment compute(timespan cycle) override;
assignment compute_bridge() override; assignment compute_bridge() override;
......
...@@ -44,7 +44,7 @@ public: ...@@ -44,7 +44,7 @@ public:
assignment compute_initial() override; assignment compute_initial() override;
assignment compute(timespan cycle, int32_t) override; assignment compute(timespan cycle) override;
private: private:
/// Total number of elements in all processed batches in the current cycle. /// Total number of elements in all processed batches in the current cycle.
......
...@@ -193,9 +193,6 @@ public: ...@@ -193,9 +193,6 @@ 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;
......
...@@ -165,9 +165,6 @@ public: ...@@ -165,9 +165,6 @@ 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
......
...@@ -75,10 +75,6 @@ struct CAF_CORE_EXPORT upstream_msg : tag::boxing_type { ...@@ -75,10 +75,6 @@ struct CAF_CORE_EXPORT 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
...@@ -169,7 +165,7 @@ template <class Inspector> ...@@ -169,7 +165,7 @@ template <class Inspector>
typename Inspector::result_type typename Inspector::result_type
inspect(Inspector& f, upstream_msg::ack_batch& x) { inspect(Inspector& f, upstream_msg::ack_batch& x) {
return f(meta::type_name("ack_batch"), x.new_capacity, x.desired_batch_size, return f(meta::type_name("ack_batch"), x.new_capacity, x.desired_batch_size,
x.acknowledged_id, x.max_capacity); x.acknowledged_id);
} }
/// @relates upstream_msg::drop /// @relates upstream_msg::drop
......
...@@ -50,7 +50,7 @@ credit_controller::assignment impl::compute_initial() { ...@@ -50,7 +50,7 @@ credit_controller::assignment impl::compute_initial() {
} }
credit_controller::assignment credit_controller::assignment
impl::compute(timespan cycle, int32_t downstream_capacity) { impl::compute(timespan cycle) {
// Max throughput = C * (N / t), where C = cycle length, N = measured items, // Max throughput = C * (N / t), where C = cycle length, N = measured items,
// and t = measured time. Desired batch size is the same formula with D // and t = measured time. Desired batch size is the same formula with D
// (desired complexity) instead of C. We compute our values in 64-bit for // (desired complexity) instead of C. We compute our values in 64-bit for
...@@ -71,9 +71,8 @@ impl::compute(timespan cycle, int32_t downstream_capacity) { ...@@ -71,9 +71,8 @@ impl::compute(timespan cycle, int32_t downstream_capacity) {
// Instead of C * (N / t) we calculate (C * N) / t to avoid double conversion // Instead of C * (N / t) we calculate (C * N) / t to avoid double conversion
// and rounding errors. // and rounding errors.
assignment result; assignment result;
// Give enough credit to last 2 cycles, but don't exceed downstream capacity. // Give enough credit to last 2 cycles.
result.credit = 2 * clamp((cycle.count() * num_elements_) / total_ns); result.credit = 2 * clamp((cycle.count() * num_elements_) / total_ns);
result.credit = std::min(result.credit, downstream_capacity);
result.batch_size = clamp((complexity_.count() * num_elements_) / total_ns); result.batch_size = clamp((complexity_.count() * num_elements_) / total_ns);
// Reset state and return. // Reset state and return.
num_elements_ = 0; num_elements_ = 0;
......
...@@ -199,10 +199,6 @@ size_t downstream_manager::buffered(stream_slot) const noexcept { ...@@ -199,10 +199,6 @@ 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;
......
...@@ -46,10 +46,10 @@ void emit_ack_batch(inbound_path& path, credit_controller::assignment x, ...@@ -46,10 +46,10 @@ void emit_ack_batch(inbound_path& path, credit_controller::assignment x,
auto guard = detail::make_scope_guard([&] { auto guard = detail::make_scope_guard([&] {
if (!force_ack_msg || path.up_to_date()) if (!force_ack_msg || path.up_to_date())
return; return;
unsafe_send_as(path.self(), path.hdl, unsafe_send_as(
make<upstream_msg::ack_batch>( path.self(), path.hdl,
path.slots.invert(), path.self()->address(), 0, make<upstream_msg::ack_batch>(path.slots.invert(), path.self()->address(),
x.batch_size, path.last_batch_id, x.credit)); 0, x.batch_size, path.last_batch_id));
path.last_acked_batch_id = path.last_batch_id; path.last_acked_batch_id = path.last_batch_id;
}); });
if (x.credit <= used) if (x.credit <= used)
...@@ -61,7 +61,7 @@ void emit_ack_batch(inbound_path& path, credit_controller::assignment x, ...@@ -61,7 +61,7 @@ void emit_ack_batch(inbound_path& path, credit_controller::assignment x,
unsafe_send_as(path.self(), path.hdl, unsafe_send_as(path.self(), path.hdl,
make<upstream_msg::ack_batch>( make<upstream_msg::ack_batch>(
path.slots.invert(), path.self()->address(), new_credit, path.slots.invert(), path.self()->address(), new_credit,
x.batch_size, path.last_batch_id, x.credit)); x.batch_size, path.last_batch_id));
path.last_acked_batch_id = path.last_batch_id; path.last_acked_batch_id = path.last_batch_id;
path.assigned_credit += new_credit; path.assigned_credit += new_credit;
} }
...@@ -148,9 +148,7 @@ void inbound_path::emit_ack_batch(local_actor*, int32_t, ...@@ -148,9 +148,7 @@ void inbound_path::emit_ack_batch(local_actor*, int32_t,
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(cycle)); CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(cycle));
last_credit_decision = now; last_credit_decision = now;
next_credit_decision = now + cycle; next_credit_decision = now + cycle;
auto max_capacity = static_cast<int32_t>(mgr->out().max_capacity()); caf::emit_ack_batch(*this, controller_->compute(cycle), force_ack);
caf::emit_ack_batch(*this, controller_->compute(cycle, max_capacity),
force_ack);
} }
bool inbound_path::up_to_date() { bool inbound_path::up_to_date() {
......
...@@ -40,7 +40,6 @@ outbound_path::outbound_path(stream_slot sender_slot, ...@@ -40,7 +40,6 @@ 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
} }
......
...@@ -312,7 +312,9 @@ scheduled_actor::mailbox_visitor::operator()(size_t, upstream_queue&, ...@@ -312,7 +312,9 @@ scheduled_actor::mailbox_visitor::operator()(size_t, upstream_queue&,
CAF_LOG_RECEIVE_EVENT((&x)); CAF_LOG_RECEIVE_EVENT((&x));
CAF_BEFORE_PROCESSING(self, x); CAF_BEFORE_PROCESSING(self, x);
auto& um = x.content().get_mutable_as<upstream_msg>(0); auto& um = x.content().get_mutable_as<upstream_msg>(0);
upstream_msg_visitor f{self, um}; auto f = [&](auto& content) {
self->handle_upstream_msg(um.slots, um.sender, content);
};
visit(f, um.content); visit(f, um.content);
CAF_AFTER_PROCESSING(self, invoke_message_result::consumed); CAF_AFTER_PROCESSING(self, invoke_message_result::consumed);
return ++handled_msgs < max_throughput ? intrusive::task_result::resume return ++handled_msgs < max_throughput ? intrusive::task_result::resume
......
...@@ -58,7 +58,7 @@ credit_controller::assignment impl::compute_initial() { ...@@ -58,7 +58,7 @@ credit_controller::assignment impl::compute_initial() {
return {buffer_size_, batch_size_}; return {buffer_size_, batch_size_};
} }
credit_controller::assignment impl::compute(timespan, int32_t) { credit_controller::assignment impl::compute(timespan) {
if (sampled_elements_ >= min_samples) { if (sampled_elements_ >= min_samples) {
// Helper for truncating a 64-bit integer to a 32-bit integer with a // Helper for truncating a 64-bit integer to a 32-bit integer with a
// minimum value of 1. // minimum value of 1.
...@@ -93,7 +93,8 @@ credit_controller::assignment impl::compute_bridge() { ...@@ -93,7 +93,8 @@ credit_controller::assignment impl::compute_bridge() {
} }
int32_t impl::threshold() const noexcept { int32_t impl::threshold() const noexcept {
return static_cast<int32_t>(buffer_size_ * buffer_threshold); return buffer_size_-1;
// return static_cast<int32_t>(buffer_size_ * buffer_threshold);
} }
} // namespace caf::detail } // namespace caf::detail
...@@ -99,9 +99,7 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) { ...@@ -99,9 +99,7 @@ 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.
......
...@@ -41,8 +41,7 @@ credit_controller::assignment test_credit_controller::compute_initial() { ...@@ -41,8 +41,7 @@ credit_controller::assignment test_credit_controller::compute_initial() {
return {50, 50}; return {50, 50};
} }
credit_controller::assignment credit_controller::assignment test_credit_controller::compute(timespan cycle) {
test_credit_controller::compute(timespan cycle, int32_t) {
auto& cfg = self()->system().config(); auto& cfg = self()->system().config();
auto complexity = cfg.stream_desired_batch_complexity; auto complexity = cfg.stream_desired_batch_complexity;
// Max throughput = C * (N / t), where C = cycle length, N = measured items, // Max throughput = C * (N / t), where C = cycle length, N = measured items,
......
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