Commit a40a675c authored by Dominik Charousset's avatar Dominik Charousset

Fix handling of downstream capacity

parent 66dba7e3
......@@ -32,10 +32,10 @@ public:
/// Wraps an assignment of the controller to its source.
struct assignment {
/// Store how much credit we assign to the source.
/// Stores how much credit we assign to the source.
int32_t credit;
/// Store how many elements we demand per batch.
/// Stores how many elements we demand per batch.
int32_t batch_size;
};
......@@ -67,8 +67,12 @@ public:
/// @returns The initial credit for the new sources.
virtual assignment compute_initial() = 0;
/// Computes a credit assignment to the source after a cycle ends.
virtual assignment compute(timespan cycle) = 0;
/// Assigs new credit to the source after a cycle ends.
/// @param cycle Duration of a cycle.
/// @param max_downstream_credit Maximum downstream capacity as reported by
/// 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 ------------------------------------------------------
......
......@@ -45,7 +45,7 @@ public:
assignment compute_initial() override;
assignment compute(timespan cycle) override;
assignment compute(timespan cycle, int32_t) override;
private:
// -- member variables -------------------------------------------------------
......
......@@ -50,7 +50,7 @@ public:
assignment compute_initial() override;
assignment compute(timespan cycle) override;
assignment compute(timespan cycle, int32_t) override;
assignment compute_bridge() override;
......@@ -85,7 +85,7 @@ private:
int32_t sample_counter_ = 0;
/// Configured how many batches we skip for the size sampling.
int32_t sample_rate_ = 50;
int32_t sample_rate_ = 1;
};
} // namespace caf::detail
......@@ -44,7 +44,7 @@ public:
assignment compute_initial() override;
assignment compute(timespan cycle) override;
assignment compute(timespan cycle, int32_t) override;
private:
/// Total number of elements in all processed batches in the current cycle.
......
......@@ -49,7 +49,8 @@ credit_controller::assignment impl::compute_initial() {
return {50, 10};
}
credit_controller::assignment impl::compute(timespan cycle) {
credit_controller::assignment
impl::compute(timespan cycle, int32_t downstream_capacity) {
// 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
// (desired complexity) instead of C. We compute our values in 64-bit for
......@@ -70,8 +71,9 @@ credit_controller::assignment impl::compute(timespan cycle) {
// Instead of C * (N / t) we calculate (C * N) / t to avoid double conversion
// and rounding errors.
assignment result;
// Give enough credit to last 2 cycles.
// Give enough credit to last 2 cycles, but don't exceed downstream capacity.
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);
// Reset state and return.
num_elements_ = 0;
......
......@@ -53,7 +53,7 @@ const atom_value credit_policy = atom("complexity");
namespace size_policy {
const int32_t bytes_per_batch = 1024; // 1 KB
const int32_t bytes_per_batch = 2048; // 2 KB
const int32_t buffer_capacity = 64 * 1024; // 64 KB
} // namespace size_policy
......
......@@ -40,32 +40,27 @@ void emit_ack_batch(inbound_path& path, credit_controller::assignment x,
CAF_ASSERT(x.batch_size > 0);
auto& out = path.mgr->out();
path.desired_batch_size = x.batch_size;
auto downstream_capacity = out.max_capacity();
int32_t new_credit = 0;
auto used = static_cast<int32_t>(out.buffered()) + path.assigned_credit;
auto guard = detail::make_scope_guard([&] {
if (!force_ack_msg || path.up_to_date())
return;
unsafe_send_as(path.self(), path.hdl,
make<upstream_msg::ack_batch>(path.slots.invert(),
path.self()->address(), 0,
x.batch_size,
path.last_batch_id,
downstream_capacity));
make<upstream_msg::ack_batch>(
path.slots.invert(), path.self()->address(), 0,
x.batch_size, path.last_batch_id, x.credit));
path.last_acked_batch_id = path.last_batch_id;
});
auto max_credit = std::min(x.credit, downstream_capacity);
auto used_credit = static_cast<int32_t>(out.buffered()) + path.assigned_credit;
if (max_credit <= used_credit)
if (x.credit <= used)
return;
auto new_credit = path.mgr->acquire_credit(&path, max_credit - used_credit);
new_credit = path.mgr->acquire_credit(&path, x.credit - used);
if (new_credit < 1)
return;
guard.disable();
unsafe_send_as(path.self(), path.hdl,
make<upstream_msg::ack_batch>(path.slots.invert(),
path.self()->address(),
new_credit, x.batch_size,
path.last_batch_id,
downstream_capacity));
make<upstream_msg::ack_batch>(
path.slots.invert(), path.self()->address(), new_credit,
x.batch_size, path.last_batch_id, x.credit));
path.last_acked_batch_id = path.last_batch_id;
path.assigned_credit += new_credit;
}
......@@ -149,7 +144,9 @@ void inbound_path::emit_ack_batch(local_actor*, int32_t,
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(cycle));
last_credit_decision = now;
next_credit_decision = now + cycle;
caf::emit_ack_batch(*this, controller_->compute(cycle), force_ack);
auto max_capacity = static_cast<int32_t>(mgr->out().max_capacity());
caf::emit_ack_batch(*this, controller_->compute(cycle, max_capacity),
force_ack);
}
bool inbound_path::up_to_date() {
......
......@@ -58,8 +58,8 @@ credit_controller::assignment impl::compute_initial() {
return {buffer_size_, batch_size_};
}
credit_controller::assignment impl::compute(timespan) {
if (sampled_elements_ > 0) {
credit_controller::assignment impl::compute(timespan, int32_t) {
if (sampled_elements_ > 9) {
// Helper for truncating a 64-bit integer to a 32-bit integer with a
// minimum value of 1.
auto clamp_i32 = [](int64_t x) -> int32_t {
......
......@@ -41,7 +41,8 @@ credit_controller::assignment test_credit_controller::compute_initial() {
return {50, 50};
}
credit_controller::assignment test_credit_controller::compute(timespan cycle) {
credit_controller::assignment
test_credit_controller::compute(timespan cycle, int32_t) {
auto& cfg = self()->system().config();
auto complexity = cfg.stream_desired_batch_complexity;
// 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