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

Add scaffold for auto-adjustment of batch sizes

parent 1b95f6d5
......@@ -23,6 +23,7 @@
#include <cstdint>
#include "caf/actor_control_block.hpp"
#include "caf/downstream_msg.hpp"
#include "caf/stream_aborter.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_priority.hpp"
......@@ -48,11 +49,14 @@ public:
/// Points to the manager responsible for incoming traffic.
stream_manager_ptr mgr;
/// Handle to the source.
strong_actor_ptr hdl;
/// Stores slot IDs for sender (hdl) and receiver (self).
stream_slots slots;
/// Handle to the source.
strong_actor_ptr hdl;
/// Amount of credit we have signaled upstream.
int32_t assigned_credit;
/// Priority of incoming batches from this source.
stream_priority prio;
......@@ -63,12 +67,6 @@ public:
/// ID of the last received batch.
int64_t last_batch_id;
/// Amount of credit we have signaled upstream.
long assigned_credit;
/// Ideal size for individual batches.
long desired_batch_size;
/// Stores whether the source actor is failsafe, i.e., allows the runtime to
/// redeploy it on failure.
bool redeployable;
......@@ -77,6 +75,40 @@ public:
/// whether the destructor sends `close` or `forced_close` messages.
error shutdown_reason;
/// Keep track of measurements for the last 10 batches.
static constexpr size_t stats_sampling_size = 16;
/// Stores statistics for measuring complexity of incoming batches.
struct stats_t {
struct measurement {
long batch_size;
std::chrono::nanoseconds calculation_time;
};
struct calculation_result {
long max_throughput;
long items_per_batch;
};
stats_t();
/// Stores `stats_sampling_size` measurements in a ring.
std::vector<measurement> measurements;
/// Current position in `measurements`
size_t ring_iter;
/// Returns the maximum number of items this actor could handle for given
/// cycle length.
calculation_result calculate(std::chrono::nanoseconds cycle,
std::chrono::nanoseconds desired_complexity);
/// Stores a new measurement in the ring buffer.
void store(measurement x);
};
stats_t stats;
/// Constructs a path for given handle and stream ID.
inbound_path(stream_manager_ptr mgr_ptr, stream_slots id,
strong_actor_ptr ptr);
......@@ -84,15 +116,24 @@ public:
~inbound_path();
/// Updates `last_batch_id` and `assigned_credit`.
void handle_batch(long batch_size, int64_t batch_id);
//void handle_batch(long batch_size, int64_t batch_id);
void handle(downstream_msg::batch& x);
/// Emits a `stream_msg::ack_batch` on this path and sets `assigned_credit`
/// to `initial_demand`.
void emit_ack_open(local_actor* self, actor_addr rebind_from, long initial_demand,
void emit_ack_open(local_actor* self, actor_addr rebind_from,
bool is_redeployable);
/// Sends a `stream_msg::ack_batch`.
void emit_ack_batch(local_actor* self, long new_demand);
/// Sends a `stream_msg::ack_batch` with credit for the next round. Credit is
/// calculated as `max_queue_size - (assigned_credit - queued_items)`, whereas
/// `max_queue_size` is `2 * ...`.
/// @param self Points to the parent actor.
/// @param queued_items Counts the accumulated size of all batches that are
/// currently waiting in the mailbox.
void emit_ack_batch(local_actor* self, long queued_items,
std::chrono::nanoseconds cycle,
std::chrono::nanoseconds desired_batch_complexity);
/// Sends a `stream_msg::close` on this path.
void emit_regular_shutdown(local_actor* self);
......
......@@ -25,16 +25,47 @@
namespace caf {
inbound_path::stats_t::stats_t() {
// we initially assume one item = 1us
// TODO: put constant in some header
measurement x{50, std::chrono::microseconds(50)};
measurements.resize(stats_sampling_size, x);
}
auto inbound_path::stats_t::calculate(std::chrono::nanoseconds c,
std::chrono::nanoseconds d)
-> calculation_result {
// 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
// instead of C.
long total_ns = 0;
long total_items = 0;
for (auto& x : measurements) {
total_ns += static_cast<long>(x.calculation_time.count());
total_items += x.batch_size;
}
// Instead of C * (N / t) we calculate N * (C / t) to avoid imprecisions due
// to double rounding for very small numbers.
auto ct = static_cast<double>(c.count()) / total_ns;
auto dt = static_cast<double>(d.count()) / total_ns;
return {std::max(static_cast<long>(total_items * ct), 1l),
std::max(static_cast<long>(total_items * dt), 1l)};
}
void inbound_path::stats_t::store(measurement x) {
measurements[ring_iter] = x;
ring_iter = (ring_iter + 1) % stats_sampling_size;
}
inbound_path::inbound_path(stream_manager_ptr mgr_ptr, stream_slots id,
strong_actor_ptr ptr)
: mgr(std::move(mgr_ptr)),
slots(id),
hdl(std::move(ptr)),
slots(id),
assigned_credit(0),
prio(stream_priority::normal),
last_acked_batch_id(0),
last_batch_id(0),
assigned_credit(0),
desired_batch_size(50), // TODO: at least put default in some header
redeployable(false) {
mgr->register_input_path(this);
}
......@@ -43,30 +74,39 @@ inbound_path::~inbound_path() {
mgr->deregister_input_path(this);
}
void inbound_path::handle_batch(long batch_size, int64_t batch_id) {
assigned_credit -= batch_size;
last_batch_id = batch_id;
void inbound_path::handle(downstream_msg::batch& x) {
assigned_credit -= x.xs_size;
last_batch_id = x.id;
// TODO: add time-measurement functions to local actor and store taken
// measurement in stats
mgr->handle(this, x);
mgr->push();
}
void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from,
long initial_demand, bool is_redeployable) {
assigned_credit = initial_demand;
bool is_redeployable) {
assigned_credit = 50; // TODO: put constant in some header
redeployable = is_redeployable;
auto batch_size = static_cast<int32_t>(desired_batch_size);
int32_t desired_batch_size = 50; // TODO: put constant in some header
unsafe_send_as(self, hdl,
make<upstream_msg::ack_open>(
slots, self->address(), std::move(rebind_from), self->ctrl(),
static_cast<int32_t>(initial_demand), batch_size,
is_redeployable));
slots.invert(), self->address(), std::move(rebind_from),
self->ctrl(), static_cast<int32_t>(assigned_credit),
desired_batch_size, is_redeployable));
}
void inbound_path::emit_ack_batch(local_actor* self, long new_demand) {
last_acked_batch_id = last_batch_id;
assigned_credit += new_demand;
auto batch_size = static_cast<int32_t>(desired_batch_size);
void inbound_path::emit_ack_batch(local_actor* self, long queued_items,
std::chrono::nanoseconds cycle,
std::chrono::nanoseconds complexity){
auto x = stats.calculate(cycle, complexity);
auto credit = std::max(x.max_throughput - (assigned_credit - queued_items),
0l);
auto batch_size = static_cast<int32_t>(x.items_per_batch);
if (credit != 0)
assigned_credit += credit;
unsafe_send_as(self, hdl,
make<upstream_msg::ack_batch>(slots, self->address(),
static_cast<int32_t>(new_demand),
make<upstream_msg::ack_batch>(slots.invert(), self->address(),
static_cast<int32_t>(credit),
batch_size, last_batch_id));
}
......@@ -75,18 +115,18 @@ void inbound_path::emit_regular_shutdown(local_actor* self) {
}
void inbound_path::emit_regular_shutdown(local_actor* self, error reason) {
unsafe_send_as(
self, hdl,
make<upstream_msg::forced_drop>(slots, self->address(), std::move(reason)));
unsafe_send_as(self, hdl,
make<upstream_msg::forced_drop>(
slots.invert(), self->address(), std::move(reason)));
}
void inbound_path::emit_irregular_shutdown(local_actor* self,
stream_slots slots,
const strong_actor_ptr& hdl,
error reason) {
unsafe_send_as(
self, hdl,
make<upstream_msg::forced_drop>(slots, self->address(), std::move(reason)));
unsafe_send_as(self, hdl,
make<upstream_msg::forced_drop>(
slots.invert(), self->address(), std::move(reason)));
}
} // namespace caf
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