Commit 66dba7e3 authored by Dominik Charousset's avatar Dominik Charousset

Simplify and tune size-based credit controller

parent fd14095d
......@@ -22,10 +22,10 @@
#include <chrono>
namespace caf {
namespace detail {
namespace caf::detail {
/// .
/// A credit controller that estimates the bytes required to store incoming
/// batches and constrains credit based on upper bounds for memory usage.
class size_based_credit_controller : public credit_controller {
public:
// -- member types -----------------------------------------------------------
......@@ -60,26 +60,13 @@ private:
// -- member variables -------------------------------------------------------
/// Total number of elements in all processed batches in the current cycle.
int64_t num_elements_ = 0;
/// Measured size of all sampled elements.
int64_t total_size_ = 0;
/// Elapsed time for processing all elements of all batches in the current
/// cycle.
timespan processing_time_;
/// Timestamp of the last call to `before_processing`.
time_point processing_begin_;
/// Stores the desired per-batch complexity.
timespan complexity_;
int64_t num_batches_ = 0;
/// Stores how many elements the buffer should hold at most.
int32_t buffer_size_ = 1;
int32_t buffer_size_ = 10;
/// Stores how many elements each batch should contain.
int32_t batch_size_ = 1;
int32_t batch_size_ = 2;
/// Configures how many bytes we store in total.
int32_t buffer_capacity_;
......@@ -87,15 +74,18 @@ private:
/// Configures how many bytes we transfer per batch.
int32_t bytes_per_batch_;
/// Stores the current write position in the ring buffer.
size_t ringbuf_pos_ = 0;
/// Stores how many elements we have sampled during the current cycle.
int64_t sampled_elements_ = 0;
/// Stores approximately how many bytes the sampled elements require when
/// serialized.
int64_t sampled_total_size_ = 0;
/// Stores how many valid entries the ring buffer holds.
size_t ringbuf_size_ = 0;
/// Counter for keeping track of when to sample a batch.
int32_t sample_counter_ = 0;
/// Records recent calculations for buffer and batch sizes.
std::array<std::pair<int32_t, int32_t>, 32> ringbuf_;
/// Configured how many batches we skip for the size sampling.
int32_t sample_rate_ = 50;
};
} // namespace detail
} // namespace caf
} // namespace caf::detail
......@@ -60,9 +60,6 @@ public:
/// Amount of credit we have signaled upstream.
int32_t assigned_credit = 0;
/// Upstream capacity for limiting the amount of credit we can give.
int32_t downstream_capacity = 0;
/// Priority of incoming batches from this source.
stream_priority prio = stream_priority::normal;
......
......@@ -25,8 +25,7 @@
// Safe us some typing and very ugly formatting.
#define impl complexity_based_credit_controller
namespace caf {
namespace detail {
namespace caf::detail {
impl::impl(scheduled_actor* self) : super(self) {
auto& cfg = self->system().config();
......@@ -80,5 +79,4 @@ credit_controller::assignment impl::compute(timespan cycle) {
return result;
}
} // namespace detail
} // namespace caf
} // namespace caf::detail
......@@ -53,8 +53,8 @@ const atom_value credit_policy = atom("complexity");
namespace size_policy {
const int32_t bytes_per_batch = 4 * 1024; // 4 kB
const int32_t buffer_capacity = 64 * 1024; // 64 kB
const int32_t bytes_per_batch = 1024; // 1 KB
const int32_t buffer_capacity = 64 * 1024; // 64 KB
} // namespace size_policy
......
......@@ -38,7 +38,9 @@ constexpr bool force_ack = true;
void emit_ack_batch(inbound_path& path, credit_controller::assignment x,
bool force_ack_msg = false) {
CAF_ASSERT(x.batch_size > 0);
auto& out = path.mgr->out();
path.desired_batch_size = x.batch_size;
auto downstream_capacity = out.max_capacity();
auto guard = detail::make_scope_guard([&] {
if (!force_ack_msg || path.up_to_date())
return;
......@@ -47,14 +49,14 @@ void emit_ack_batch(inbound_path& path, credit_controller::assignment x,
path.self()->address(), 0,
x.batch_size,
path.last_batch_id,
path.downstream_capacity));
downstream_capacity));
path.last_acked_batch_id = path.last_batch_id;
});
auto credit = std::min(x.credit, path.downstream_capacity);
if (credit <= path.assigned_credit)
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)
return;
auto new_credit = path.mgr->acquire_credit(&path,
credit - path.assigned_credit);
auto new_credit = path.mgr->acquire_credit(&path, max_credit - used_credit);
if (new_credit < 1)
return;
guard.disable();
......@@ -63,7 +65,7 @@ void emit_ack_batch(inbound_path& path, credit_controller::assignment x,
path.self()->address(),
new_credit, x.batch_size,
path.last_batch_id,
path.downstream_capacity));
downstream_capacity));
path.last_acked_batch_id = path.last_batch_id;
path.assigned_credit += new_credit;
}
......@@ -128,7 +130,6 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) {
// Update state.
auto initial = controller_->compute_initial();
assigned_credit = mgr->acquire_credit(this, initial.credit);
downstream_capacity = assigned_credit;
CAF_ASSERT(assigned_credit >= 0);
desired_batch_size = std::min(initial.batch_size, assigned_credit);
// Make sure we receive errors from this point on.
......@@ -146,7 +147,6 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) {
void inbound_path::emit_ack_batch(local_actor*, int32_t,
actor_clock::time_point now, timespan cycle) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(cycle));
downstream_capacity = mgr->out().max_capacity();
last_credit_decision = now;
next_credit_decision = now + cycle;
caf::emit_ack_batch(*this, controller_->compute(cycle), force_ack);
......
......@@ -27,12 +27,10 @@
// Safe us some typing and very ugly formatting.
#define impl size_based_credit_controller
namespace caf {
namespace detail {
namespace caf::detail {
impl::impl(scheduled_actor* self) : super(self) {
auto& cfg = self->system().config();
complexity_ = cfg.stream_desired_batch_complexity;
buffer_capacity_ = get_or(cfg, "stream.size-policy.buffer-capacity",
defaults::stream::size_policy::buffer_capacity);
bytes_per_batch_ = get_or(cfg, "stream.size-policy.bytes-per-batch",
......@@ -44,13 +42,16 @@ impl::~impl() {
}
void impl::before_processing(downstream_msg::batch& x) {
num_elements_ += x.xs_size;
total_size_ = serialized_size(self()->system(), x.xs);
processing_begin_ = clock_type::now();
if (++sample_counter_ == sample_rate_) {
sampled_elements_ += x.xs_size;
sampled_total_size_ += serialized_size(self()->system(), x.xs);
sample_counter_ = 0;
}
++num_batches_;
}
void impl::after_processing(downstream_msg::batch&) {
processing_time_ += clock_type::now() - processing_begin_;
// nop
}
credit_controller::assignment impl::compute_initial() {
......@@ -58,9 +59,7 @@ credit_controller::assignment impl::compute_initial() {
}
credit_controller::assignment impl::compute(timespan) {
// Update batch and buffer size if we have at least 10 new data points to
// work with.
if (num_elements_ > 9) {
if (sampled_elements_ > 0) {
// 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 {
......@@ -71,40 +70,18 @@ credit_controller::assignment impl::compute(timespan) {
return 1;
return static_cast<int32_t>(x);
};
// Calculate ideal batch size by complexity.
auto total_ns = processing_time_.count();
CAF_ASSERT(total_ns > 0);
auto size_by_complexity = clamp_i32((complexity_.count() * num_elements_)
/ total_ns);
// Calculate ideal batch size by size.
auto bytes_per_element = clamp_i32(total_size_ / num_elements_);
auto size_by_bytes = clamp_i32(bytes_per_batch_ / bytes_per_element);
// Always pick the smaller of the two in order to achieve a good tradeoff
// between size and computational complexity
auto batch_size = clamp_i32(std::min(size_by_bytes, size_by_complexity));
auto buffer_size = std::min(clamp_i32(buffer_capacity_ / bytes_per_element),
4 * batch_size_);
// Add a new entry to our ringbuffer and calculate batch and buffer sizes
// based on the average recorded sizes.
auto& kvp = ringbuf_[ringbuf_pos_];
kvp.first = buffer_size;
kvp.second = batch_size;
ringbuf_pos_ = (ringbuf_pos_ + 1) % ringbuf_.size();
if (ringbuf_size_ < ringbuf_.size())
++ringbuf_size_;
using int32_pair = std::pair<int32_t, int32_t>;
auto plus = [](int32_pair x, int32_pair y) {
return int32_pair{x.first + y.first, x.second + y.second};
};
auto psum = std::accumulate(ringbuf_.begin(),
ringbuf_.begin() + ringbuf_size_,
int32_pair{0, 0}, plus);
buffer_size_ = psum.first / ringbuf_size_;
batch_size_ = psum.second / ringbuf_size_;
auto bytes_per_element = clamp_i32(sampled_total_size_ / sampled_elements_);
batch_size_ = clamp_i32(bytes_per_batch_ / bytes_per_element);
buffer_size_ = clamp_i32(buffer_capacity_ / bytes_per_element);
// Reset bookkeeping state.
num_elements_ = 0;
total_size_ = 0;
processing_time_ = timespan{0};
sampled_elements_ = 0;
sampled_total_size_ = 0;
// Ideally, we sample 10 batches per cycle.
sample_rate_ = clamp_i32(num_batches_ / 10);
if (sample_counter_ >= sample_rate_)
sample_counter_ = 0;
num_batches_ = 0;
}
return {buffer_size_, batch_size_};
}
......@@ -116,8 +93,7 @@ credit_controller::assignment impl::compute_bridge() {
}
int32_t impl::low_threshold() const noexcept {
return buffer_size_ / 4;
return static_cast<int32_t>(buffer_size_ * .75f);
}
} // namespace detail
} // namespace caf
} // namespace caf::detail
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