Commit 062f96cd authored by Dominik Charousset's avatar Dominik Charousset

Compute credit on 64-bit before clamping to 32-bit

parent f8099464
...@@ -82,7 +82,7 @@ public: ...@@ -82,7 +82,7 @@ public:
/// Wraps a time measurement for a single processed batch. /// Wraps a time measurement for a single processed batch.
struct measurement { struct measurement {
/// Number of items in the batch. /// Number of items in the batch.
long batch_size; int32_t batch_size;
/// Elapsed time for processing all elements of the batch. /// Elapsed time for processing all elements of the batch.
timespan calculation_time; timespan calculation_time;
}; };
...@@ -90,9 +90,9 @@ public: ...@@ -90,9 +90,9 @@ public:
/// Wraps the resulf of `stats_t::calculate()`. /// Wraps the resulf of `stats_t::calculate()`.
struct calculation_result { struct calculation_result {
/// Number of items per credit cycle. /// Number of items per credit cycle.
long max_throughput; int32_t max_throughput;
/// Number of items per batch to reach the desired batch complexity. /// Number of items per batch to reach the desired batch complexity.
long items_per_batch; int32_t items_per_batch;
}; };
stats_t(); stats_t();
...@@ -141,7 +141,7 @@ public: ...@@ -141,7 +141,7 @@ 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, long queued_items, timespan cycle, void emit_ack_batch(local_actor* self, int32_t queued_items, 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
......
...@@ -35,20 +35,30 @@ auto inbound_path::stats_t::calculate(timespan c, timespan d) ...@@ -35,20 +35,30 @@ auto inbound_path::stats_t::calculate(timespan c, timespan d)
// 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
// instead of C. // instead of C.
long total_ns = 0; // We compute our values in 64-bit for more precision before truncating to a
long total_items = 0; // 32-bit integer type at the end.
int64_t total_ns = 0;
int64_t total_items = 0;
for (auto& x : measurements) { for (auto& x : measurements) {
total_ns += static_cast<long>(x.calculation_time.count()); total_ns += x.calculation_time.count();
total_items += x.batch_size; total_items += x.batch_size;
} }
if (total_ns == 0) if (total_ns == 0)
return {1, 1}; return {1, 1};
/// Helper for truncating a 64-bit integer to a 32-bit integer with a minimum
/// value of 1.
auto clamp = [](int64_t x) -> int32_t {
static constexpr auto upper_bound = std::numeric_limits<int32_t>::max();
if (x > upper_bound)
return upper_bound;
if (x <= 0)
return 1;
return static_cast<int32_t>(x);
};
// 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.
auto cl = static_cast<long>(c.count()); return {clamp((c.count() * total_items) / total_ns),
auto dl = static_cast<long>(d.count()); clamp((d.count() * total_items) / total_ns)};
return {std::max((cl * total_items) / total_ns, 1l),
std::max((dl * total_items) / total_ns, 1l)};
} }
void inbound_path::stats_t::store(measurement x) { void inbound_path::stats_t::store(measurement x) {
...@@ -101,7 +111,7 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) { ...@@ -101,7 +111,7 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) {
self->ctrl(), assigned_credit, desired_batch_size)); self->ctrl(), assigned_credit, desired_batch_size));
} }
void inbound_path::emit_ack_batch(local_actor* self, long queued_items, void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items,
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(cycle)
<< CAF_ARG(complexity)); << CAF_ARG(complexity));
...@@ -109,7 +119,7 @@ void inbound_path::emit_ack_batch(local_actor* self, long queued_items, ...@@ -109,7 +119,7 @@ void inbound_path::emit_ack_batch(local_actor* self, long queued_items,
// Hand out enough credit to fill our queue for 2 cycles. // Hand out enough credit to fill our queue for 2 cycles.
auto credit = std::max((x.max_throughput * 2) auto credit = std::max((x.max_throughput * 2)
- (assigned_credit + queued_items), - (assigned_credit + queued_items),
0l); 0);
// The manager can restrict or adjust the amount of credit. // The manager can restrict or adjust the amount of credit.
credit = mgr->acquire_credit(this, credit); credit = mgr->acquire_credit(this, credit);
if (credit == 0 && up_to_date()) { if (credit == 0 && up_to_date()) {
......
...@@ -46,13 +46,13 @@ struct fixture { ...@@ -46,13 +46,13 @@ struct fixture {
CAF_CHECK_EQUAL(sampling_size % 2, 0u); CAF_CHECK_EQUAL(sampling_size % 2, 0u);
} }
void calculate(long total_items, long total_time) { void calculate(int32_t total_items, int32_t total_time) {
long c = 1000; int32_t c = 1000;
long d = 100; int32_t d = 100;
long n = total_items; int32_t n = total_items;
long t = total_time; int32_t t = total_time;
long m = t > 0 ? std::max((c * n) / t, 1l) : 1l; int32_t m = t > 0 ? std::max((c * n) / t, 1) : 1;
long b = t > 0 ? std::max((d * n) / t, 1l) : 1l; int32_t b = t > 0 ? std::max((d * n) / t, 1) : 1;
print("with a cycle C = %ldns, desied complexity D = %ld,", c, d); print("with a cycle C = %ldns, desied complexity D = %ld,", c, d);
print("number of items N = %ld, and time delta t = %ld:", n, t); print("number of items N = %ld, and time delta t = %ld:", n, t);
print("- throughput M = max(C * N / t, 1) = max(%ld * %ld / %ld, 1) = %ld", print("- throughput M = max(C * N / t, 1) = max(%ld * %ld / %ld, 1) = %ld",
...@@ -64,7 +64,7 @@ struct fixture { ...@@ -64,7 +64,7 @@ struct fixture {
CAF_CHECK_EQUAL(cr.max_throughput, m); CAF_CHECK_EQUAL(cr.max_throughput, m);
} }
void store(long batch_size, long calculation_time_ns) { void store(int32_t batch_size, int32_t calculation_time_ns) {
inbound_path::stats_t::measurement m{batch_size, inbound_path::stats_t::measurement m{batch_size,
timespan{calculation_time_ns}}; timespan{calculation_time_ns}};
x.store(m); x.store(m);
......
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