Commit 148243da authored by Dominik Charousset's avatar Dominik Charousset

Expose streaming parameters as timespans

parent 31df841a
......@@ -263,19 +263,43 @@ public:
// -- streaming parameters ---------------------------------------------------
/// Maximum delay for sending underfull batches.
/// @private
size_t streaming_desired_batch_complexity_us;
/// Maximum delay for sending underfull batches.
/// @private
size_t streaming_max_batch_delay_us;
/// Timespan between two credit rounds.
/// @private
size_t streaming_credit_round_interval_us;
/// Greatest common divisor of `streaming_max_batch_delay_us` and
/// `streaming_credit_round_interval_us`.
/// @private
size_t streaming_tick_duration_us() const noexcept;
/// Returns the greatest common divisor of `streaming_max_batch_delay` and
/// `streaming_credit_round_interval`.
inline timespan streaming_tick_duration() const noexcept {
using us_t = std::chrono::microseconds;
return timespan{us_t{streaming_tick_duration_us()}};
}
/// Returns the desired timespan in a sink or stage for processing a single
/// batch.
inline timespan streaming_desired_batch_complexity() const noexcept {
using us_t = std::chrono::microseconds;
return timespan{us_t{streaming_desired_batch_complexity_us}};
}
/// Returns the maximum delay for sending underfull batches.
inline timespan streaming_max_batch_delay() const noexcept {
using us_t = std::chrono::microseconds;
return timespan{us_t{streaming_max_batch_delay_us}};
}
// Returns the desired timespan between two credit rounds.
inline timespan streaming_credit_round_interval() const noexcept {
using us_t = std::chrono::microseconds;
return timespan{us_t{streaming_credit_round_interval_us}};
}
// -- scheduling parameters --------------------------------------------------
atom_value scheduler_policy;
......
......@@ -556,8 +556,8 @@ struct test_coordinator_fixture {
.set("scheduler.policy", caf::atom("testing"))),
self(sys, true),
sched(dynamic_cast<scheduler_type&>(sys.scheduler())),
credit_round_interval(us_t{cfg.streaming_credit_round_interval_us}),
max_batch_delay(us_t{cfg.streaming_max_batch_delay_us}) {
credit_round_interval(cfg.streaming_credit_round_interval()),
max_batch_delay(cfg.streaming_max_batch_delay()) {
// Configure the clock to measure each batch item with 1us.
sched.clock().time_per_unit.emplace(caf::atom("batch"),
caf::timespan{1000});
......
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