Commit b93be04a authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Simplify bookkeeping credit assignment

parent 5ea96794
......@@ -21,8 +21,10 @@
#include <cstddef>
#include <cstdint>
#include "caf/actor_clock.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/downstream_msg.hpp"
#include "caf/meta/type_name.hpp"
#include "caf/stream_aborter.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_priority.hpp"
......@@ -30,8 +32,6 @@
#include "caf/timestamp.hpp"
#include "caf/upstream_msg.hpp"
#include "caf/meta/type_name.hpp"
namespace caf {
/// State for a path to an upstream actor (source).
......@@ -70,9 +70,6 @@ public:
/// Amount of credit we assign sources after receiving `open`.
static constexpr int initial_credit = 50;
/// Keep track of measurements for the last X batches.
static constexpr size_t stats_sampling_size = 16;
/// Stores statistics for measuring complexity of incoming batches.
struct stats_t {
/// Wraps a time measurement for a single processed batch.
......@@ -91,24 +88,35 @@ public:
int32_t items_per_batch;
};
stats_t();
/// Total number of elements in all processed batches.
int64_t num_elements;
/// Stores `stats_sampling_size` measurements in a ring.
std::vector<measurement> measurements;
/// Elapsed time for processing all elements of all batches.
timespan processing_time;
/// Current position in `measurements`
size_t ring_iter;
stats_t();
/// Returns the maximum number of items this actor could handle for given
/// cycle length with a minimum of 1.
calculation_result calculate(timespan cycle, timespan desired_complexity);
/// Stores a new measurement in the ring buffer.
/// Adds a measurement to this statistic.
void store(measurement x);
/// Resets this statistic.
void reset();
};
/// Summarizes how many elements we processed during the last cycle and how
/// much time we spent processing those elements.
stats_t stats;
/// Stores the time point of the last credit decision for this source.
actor_clock::time_point last_credit_decision;
/// Stores the time point of the last credit decision for this source.
actor_clock::time_point next_credit_decision;
/// Constructs a path for given handle and stream ID.
inbound_path(stream_manager_ptr mgr_ptr, stream_slots id,
strong_actor_ptr ptr);
......@@ -138,7 +146,8 @@ public:
/// @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,
int32_t max_downstream_capacity, timespan cycle,
int32_t max_downstream_capacity,
actor_clock::time_point now, timespan cycle,
timespan desired_batch_complexity);
/// Returns whether the path received no input since last emitting
......@@ -155,6 +164,9 @@ public:
static void emit_irregular_shutdown(local_actor* self, stream_slots slots,
const strong_actor_ptr& hdl,
error reason);
private:
actor_clock& clock();
};
/// @relates inbound_path
......
......@@ -25,9 +25,8 @@
namespace caf {
inbound_path::stats_t::stats_t() : ring_iter(0) {
measurement x{0, timespan{0}};
measurements.resize(stats_sampling_size, x);
inbound_path::stats_t::stats_t() : num_elements(0), processing_time(0) {
// nop
}
auto inbound_path::stats_t::calculate(timespan c, timespan d)
......@@ -37,12 +36,7 @@ auto inbound_path::stats_t::calculate(timespan c, timespan d)
// instead of C.
// We compute our values in 64-bit for more precision before truncating to a
// 32-bit integer type at the end.
int64_t total_ns = 0;
int64_t total_items = 0;
for (auto& x : measurements) {
total_ns += x.calculation_time.count();
total_items += x.batch_size;
}
int64_t total_ns = processing_time.count();
if (total_ns == 0)
return {1, 1};
/// Helper for truncating a 64-bit integer to a 32-bit integer with a minimum
......@@ -57,13 +51,18 @@ auto inbound_path::stats_t::calculate(timespan c, timespan d)
};
// Instead of C * (N / t) we calculate (C * N) / t to avoid double conversion
// and rounding errors.
return {clamp((c.count() * total_items) / total_ns),
clamp((d.count() * total_items) / total_ns)};
return {clamp((c.count() * num_elements) / total_ns),
clamp((d.count() * num_elements) / total_ns)};
}
void inbound_path::stats_t::store(measurement x) {
measurements[ring_iter] = x;
ring_iter = (ring_iter + 1) % stats_sampling_size;
num_elements += x.batch_size;
processing_time += x.calculation_time;
}
void inbound_path::stats_t::reset() {
num_elements = 0;
processing_time = timespan{0};
}
inbound_path::inbound_path(stream_manager_ptr mgr_ptr, stream_slots id,
......@@ -85,14 +84,21 @@ inbound_path::~inbound_path() {
void inbound_path::handle(downstream_msg::batch& x) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(x));
auto& clk = clock();
auto batch_size = x.xs_size;
assigned_credit -= batch_size;
last_batch_id = x.id;
auto& clock = mgr->self()->clock();
auto t0 = clock.now();
auto t0 = clk.now();
if (assigned_credit <= batch_size) {
assigned_credit = 0;
// Do not log a message when "running out of credit" for the first batch
// that can easily consume the initial credit in one shot.
} else {
assigned_credit -= batch_size;
CAF_ASSERT(assigned_credit >= 0);
}
mgr->handle(this, x);
auto t1 = clock.now();
auto dt = clock.difference(atom("batch"), batch_size, t0, t1);
auto t1 = clk.now();
auto dt = clk.difference(atom("batch"), batch_size, t0, t1);
stats.store({batch_size, dt});
mgr->push();
}
......@@ -101,6 +107,7 @@ 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 = mgr->acquire_credit(this, initial_credit);
CAF_ASSERT(assigned_credit >= 0);
// Make sure we receive errors from this point on.
stream_aborter::add(hdl, self->address(), slots.receiver,
stream_aborter::source_aborter);
......@@ -109,16 +116,28 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) {
make<upstream_msg::ack_open>(
slots.invert(), self->address(), std::move(rebind_from),
self->ctrl(), assigned_credit, desired_batch_size));
last_credit_decision = clock().now();
}
void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items,
int32_t max_downstream_capacity,
timespan cycle, timespan complexity) {
actor_clock::time_point now, timespan cycle,
timespan complexity) {
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);
// Update timestamps.
last_credit_decision = now;
next_credit_decision = now + cycle;
// Short-circuit if we didn't receive anything during the last cycle.
if (stats.num_elements == 0)
return;
auto x = stats.calculate(cycle, complexity);
std::cout << "stats = " << stats.num_elements << "/"
<< deep_to_string(stats.processing_time) << " => "
<< x.max_throughput << "/" << x.items_per_batch << std::endl;
stats.reset();
// 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);
......@@ -128,6 +147,7 @@ void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items,
// Compute the amount of credit we grant in this round.
auto credit = std::min(std::max(max_capacity - assigned_credit, 0),
max_new_credit);
CAF_ASSERT(credit >= 0);
// 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())
......@@ -135,8 +155,10 @@ void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items,
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)
if (credit > 0) {
assigned_credit += credit;
CAF_ASSERT(assigned_credit >= 0);
}
desired_batch_size = static_cast<int32_t>(x.items_per_batch);
unsafe_send_as(self, hdl,
make<upstream_msg::ack_batch>(slots.invert(), self->address(),
......@@ -177,4 +199,9 @@ void inbound_path::emit_irregular_shutdown(local_actor* self,
make<upstream_msg::forced_drop>(slots.invert(), self->address(),
std::move(reason)));
}
actor_clock& inbound_path::clock() {
return mgr->self()->clock();
}
} // namespace caf
......@@ -1168,7 +1168,7 @@ scheduled_actor::advance_streams(actor_clock::time_point now) {
auto inptr = kvp.second.policy().handler.get();
auto bs = static_cast<int32_t>(kvp.second.total_task_size());
inptr->emit_ack_batch(this, bs, inptr->mgr->out().max_capacity(),
cycle, bc);
now, cycle, bc);
}
}
return stream_ticks_.next_timeout(now, {max_batch_delay_ticks_,
......
......@@ -149,6 +149,7 @@ void stream_manager::advance() {
CAF_LOG_TRACE("");
// Try to emit more credit.
if (!inbound_paths_.empty()) {
auto now = self_->clock().now();
auto& cfg = self_->system().config();
auto bc = cfg.stream_desired_batch_complexity;
auto interval = cfg.stream_credit_round_interval;
......@@ -159,7 +160,8 @@ 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, out().max_capacity(), interval, bc);
inptr->emit_ack_batch(self_, bs, out().max_capacity(), now, interval,
bc);
}
}
}
......
......@@ -39,12 +39,6 @@ namespace {
struct fixture {
inbound_path::stats_t x;
size_t sampling_size = inbound_path::stats_sampling_size;
fixture() {
CAF_CHECK_EQUAL(x.measurements.size(), sampling_size);
CAF_CHECK_EQUAL(sampling_size % 2, 0u);
}
void calculate(int32_t total_items, int32_t total_time) {
int32_t c = 1000;
......@@ -86,26 +80,11 @@ CAF_TEST(one_store) {
}
CAF_TEST(multiple_stores) {
CAF_MESSAGE("store a measurement: (50, 500ns), (60, 400ns), (40, 600ns)");
CAF_MESSAGE("store measurements: (50, 500ns), (60, 400ns), (40, 600ns)");
store(50, 500);
store(40, 600);
store(60, 400);
calculate(150, 1500);
}
CAF_TEST(overriding_stores) {
CAF_MESSAGE("fill measurements with (100, 1000ns)");
for (size_t i = 0; i < sampling_size; ++i)
store(100, 1000);
calculate(100, 1000);
CAF_MESSAGE("override first half of the measurements with (10, 1000ns)");
for (size_t i = 0; i < sampling_size / 2; ++i)
store(10, 1000);
calculate(55, 1000);
CAF_MESSAGE("override second half of the measurements with (10, 1000ns)");
for (size_t i = 0; i < sampling_size / 2; ++i)
store(10, 1000);
calculate(10, 1000);
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -342,7 +342,8 @@ 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, 30, cycle, desired_batch_complexity);
inptr->emit_ack_batch(this, bs, 30, now(), 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