Commit 6f071ab1 authored by Dominik Charousset's avatar Dominik Charousset

Integrate review feedback

parent 26a0016e
...@@ -82,10 +82,10 @@ public: ...@@ -82,10 +82,10 @@ public:
/// Returns the threshold for when we may give extra credit to a source /// Returns the threshold for when we may give extra credit to a source
/// during a cycle. /// during a cycle.
/// @returns A negative value if the controller never grants bridge credit, /// @returns Zero or a negative value if the controller never grants bridge
/// otherwise the threshold for calling `compute_bridge` to generate /// credit, otherwise the threshold for calling `compute_bridge` to
/// additional credit. /// generate additional credit.
virtual int32_t low_threshold() const noexcept; virtual int32_t threshold() const noexcept;
private: private:
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
......
...@@ -31,13 +31,21 @@ public: ...@@ -31,13 +31,21 @@ public:
using super = credit_controller; using super = credit_controller;
// -- constants --------------------------------------------------------------
/// Stores how many elements we buffer at most after the handshake.
int32_t initial_buffer_size = 50;
/// Stores how many elements we allow per batch after the handshake.
int32_t initial_batch_size = 10;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit complexity_based_credit_controller(scheduled_actor* self); explicit complexity_based_credit_controller(scheduled_actor* self);
~complexity_based_credit_controller() override; ~complexity_based_credit_controller() override;
// -- implementation of virtual functions ------------------------------------ // -- overrides --------------------------------------------------------------
void before_processing(downstream_msg::batch& x) override; void before_processing(downstream_msg::batch& x) override;
......
...@@ -20,8 +20,6 @@ ...@@ -20,8 +20,6 @@
#include "caf/credit_controller.hpp" #include "caf/credit_controller.hpp"
#include <chrono>
namespace caf::detail { namespace caf::detail {
/// A credit controller that estimates the bytes required to store incoming /// A credit controller that estimates the bytes required to store incoming
...@@ -32,9 +30,19 @@ public: ...@@ -32,9 +30,19 @@ public:
using super = credit_controller; using super = credit_controller;
using clock_type = std::chrono::steady_clock; // -- constants --------------------------------------------------------------
/// Configures at what buffer level we grant bridge credit (0 to 1).
static constexpr float buffer_threshold = 0.75f;
/// Configures how many samples we require for recalculating buffer sizes.
static constexpr int32_t min_samples = 10;
/// Stores how many elements we buffer at most after the handshake.
int32_t initial_buffer_size = 10;
using time_point = clock_type::time_point; /// Stores how many elements we allow per batch after the handshake.
int32_t initial_batch_size = 2;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
...@@ -42,7 +50,7 @@ public: ...@@ -42,7 +50,7 @@ public:
~size_based_credit_controller() override; ~size_based_credit_controller() override;
// -- implementation of virtual functions ------------------------------------ // -- overrides --------------------------------------------------------------
void before_processing(downstream_msg::batch& x) override; void before_processing(downstream_msg::batch& x) override;
...@@ -54,7 +62,7 @@ public: ...@@ -54,7 +62,7 @@ public:
assignment compute_bridge() override; assignment compute_bridge() override;
int32_t low_threshold() const noexcept override; int32_t threshold() const noexcept override;
private: private:
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
...@@ -63,10 +71,10 @@ private: ...@@ -63,10 +71,10 @@ private:
int64_t num_batches_ = 0; int64_t num_batches_ = 0;
/// Stores how many elements the buffer should hold at most. /// Stores how many elements the buffer should hold at most.
int32_t buffer_size_ = 10; int32_t buffer_size_ = initial_buffer_size;
/// Stores how many elements each batch should contain. /// Stores how many elements each batch should contain.
int32_t batch_size_ = 2; int32_t batch_size_ = initial_batch_size;
/// Configures how many bytes we store in total. /// Configures how many bytes we store in total.
int32_t buffer_capacity_; int32_t buffer_capacity_;
...@@ -75,11 +83,11 @@ private: ...@@ -75,11 +83,11 @@ private:
int32_t bytes_per_batch_; int32_t bytes_per_batch_;
/// Stores how many elements we have sampled during the current cycle. /// Stores how many elements we have sampled during the current cycle.
int64_t sampled_elements_ = 0; int32_t sampled_elements_ = 0;
/// Stores approximately how many bytes the sampled elements require when /// Stores approximately how many bytes the sampled elements require when
/// serialized. /// serialized.
int64_t sampled_total_size_ = 0; int32_t sampled_total_size_ = 0;
/// Counter for keeping track of when to sample a batch. /// Counter for keeping track of when to sample a batch.
int32_t sample_counter_ = 0; int32_t sample_counter_ = 0;
......
...@@ -36,7 +36,7 @@ public: ...@@ -36,7 +36,7 @@ public:
~test_credit_controller() override; ~test_credit_controller() override;
// -- implementation of virtual functions ------------------------------------ // -- overrides --------------------------------------------------------------
void before_processing(downstream_msg::batch& x) override; void before_processing(downstream_msg::batch& x) override;
......
...@@ -46,7 +46,7 @@ void impl::after_processing(downstream_msg::batch&) { ...@@ -46,7 +46,7 @@ void impl::after_processing(downstream_msg::batch&) {
} }
credit_controller::assignment impl::compute_initial() { credit_controller::assignment impl::compute_initial() {
return {50, 10}; return {initial_buffer_size, initial_batch_size};
} }
credit_controller::assignment credit_controller::assignment
......
...@@ -32,7 +32,7 @@ credit_controller::assignment credit_controller::compute_bridge() { ...@@ -32,7 +32,7 @@ credit_controller::assignment credit_controller::compute_bridge() {
return {0, 0}; return {0, 0};
} }
int32_t credit_controller::low_threshold() const noexcept { int32_t credit_controller::threshold() const noexcept {
return -1; return -1;
} }
......
...@@ -111,7 +111,7 @@ void inbound_path::handle(downstream_msg::batch& x) { ...@@ -111,7 +111,7 @@ void inbound_path::handle(downstream_msg::batch& x) {
assigned_credit -= batch_size; assigned_credit -= batch_size;
CAF_ASSERT(assigned_credit >= 0); CAF_ASSERT(assigned_credit >= 0);
} }
auto threshold = controller_->low_threshold(); auto threshold = controller_->threshold();
if (threshold >= 0 && assigned_credit <= threshold) if (threshold >= 0 && assigned_credit <= threshold)
caf::emit_ack_batch(*this, controller_->compute_bridge()); caf::emit_ack_batch(*this, controller_->compute_bridge());
controller_->before_processing(x); controller_->before_processing(x);
......
...@@ -59,7 +59,7 @@ credit_controller::assignment impl::compute_initial() { ...@@ -59,7 +59,7 @@ credit_controller::assignment impl::compute_initial() {
} }
credit_controller::assignment impl::compute(timespan, int32_t) { credit_controller::assignment impl::compute(timespan, int32_t) {
if (sampled_elements_ > 9) { if (sampled_elements_ >= min_samples) {
// Helper for truncating a 64-bit integer to a 32-bit integer with a // Helper for truncating a 64-bit integer to a 32-bit integer with a
// minimum value of 1. // minimum value of 1.
auto clamp_i32 = [](int64_t x) -> int32_t { auto clamp_i32 = [](int64_t x) -> int32_t {
...@@ -77,8 +77,8 @@ credit_controller::assignment impl::compute(timespan, int32_t) { ...@@ -77,8 +77,8 @@ credit_controller::assignment impl::compute(timespan, int32_t) {
// Reset bookkeeping state. // Reset bookkeeping state.
sampled_elements_ = 0; sampled_elements_ = 0;
sampled_total_size_ = 0; sampled_total_size_ = 0;
// Ideally, we sample 10 batches per cycle. // Adjust the sample rate to reach min_samples in the next cycle.
sample_rate_ = clamp_i32(num_batches_ / 10); sample_rate_ = clamp_i32(num_batches_ / min_samples);
if (sample_counter_ >= sample_rate_) if (sample_counter_ >= sample_rate_)
sample_counter_ = 0; sample_counter_ = 0;
num_batches_ = 0; num_batches_ = 0;
...@@ -92,8 +92,8 @@ credit_controller::assignment impl::compute_bridge() { ...@@ -92,8 +92,8 @@ credit_controller::assignment impl::compute_bridge() {
return {buffer_size_, batch_size_}; return {buffer_size_, batch_size_};
} }
int32_t impl::low_threshold() const noexcept { int32_t impl::threshold() const noexcept {
return static_cast<int32_t>(buffer_size_ * .75f); return static_cast<int32_t>(buffer_size_ * buffer_threshold);
} }
} // namespace caf::detail } // 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