Commit d84104c8 authored by Dominik Charousset's avatar Dominik Charousset

Add getter for tick count instead of storing it

parent d896d712
...@@ -270,7 +270,7 @@ public: ...@@ -270,7 +270,7 @@ public:
/// Greatest common divisor of `streaming_max_batch_delay_us` and /// Greatest common divisor of `streaming_max_batch_delay_us` and
/// `streaming_credit_round_interval_us`. /// `streaming_credit_round_interval_us`.
size_t streaming_tick_duration_us; size_t streaming_tick_duration_us() const noexcept;
// -- scheduling parameters -------------------------------------------------- // -- scheduling parameters --------------------------------------------------
......
...@@ -115,7 +115,6 @@ actor_system_config::actor_system_config() ...@@ -115,7 +115,6 @@ actor_system_config::actor_system_config()
streaming_desired_batch_complexity_us = 50; streaming_desired_batch_complexity_us = 50;
streaming_max_batch_delay_us = 50000; streaming_max_batch_delay_us = 50000;
streaming_credit_round_interval_us = 100000; streaming_credit_round_interval_us = 100000;
streaming_tick_duration_us = 50000;
scheduler_policy = atom("stealing"); scheduler_policy = atom("stealing");
scheduler_max_threads = std::max(std::thread::hardware_concurrency(), 4u); scheduler_max_threads = std::max(std::thread::hardware_concurrency(), 4u);
scheduler_max_throughput = std::numeric_limits<size_t>::max(); scheduler_max_throughput = std::numeric_limits<size_t>::max();
...@@ -474,9 +473,6 @@ actor_system_config& actor_system_config::parse(message& args, ...@@ -474,9 +473,6 @@ actor_system_config& actor_system_config::parse(message& args,
cout << x.name() << "=" << x.to_string() << endl; cout << x.name() << "=" << x.to_string() << endl;
}); });
} }
// Set dependent parameters.
streaming_tick_duration_us = detail::gcd(streaming_max_batch_delay_us,
streaming_credit_round_interval_us);
return *this; return *this;
} }
...@@ -504,6 +500,11 @@ actor_system_config& actor_system_config::set(const char* cn, config_value cv) { ...@@ -504,6 +500,11 @@ actor_system_config& actor_system_config::set(const char* cn, config_value cv) {
return *this; return *this;
} }
size_t actor_system_config::streaming_tick_duration_us() const noexcept {
return caf::detail::gcd(streaming_credit_round_interval_us,
streaming_max_batch_delay_us);
}
std::string actor_system_config::render_sec(uint8_t x, atom_value, std::string actor_system_config::render_sec(uint8_t x, atom_value,
const message& xs) { const message& xs) {
auto tmp = static_cast<sec>(x); auto tmp = static_cast<sec>(x);
......
...@@ -111,12 +111,16 @@ scheduled_actor::scheduled_actor(actor_config& cfg) ...@@ -111,12 +111,16 @@ scheduled_actor::scheduled_actor(actor_config& cfg)
# endif // CAF_NO_EXCEPTIONS # endif // CAF_NO_EXCEPTIONS
{ {
auto& sys_cfg = home_system().config(); auto& sys_cfg = home_system().config();
auto interval = sys_cfg.streaming_tick_duration_us; auto interval = sys_cfg.streaming_tick_duration_us();
CAF_ASSERT(interval != 0); CAF_ASSERT(interval != 0);
stream_ticks_.interval(std::chrono::microseconds(interval)); stream_ticks_.interval(std::chrono::microseconds(interval));
CAF_ASSERT(sys_cfg.streaming_max_batch_delay_us != 0);
max_batch_delay_ticks_ = sys_cfg.streaming_max_batch_delay_us / interval; max_batch_delay_ticks_ = sys_cfg.streaming_max_batch_delay_us / interval;
CAF_ASSERT(max_batch_delay_ticks_ != 0);
CAF_ASSERT(sys_cfg.streaming_credit_round_interval_us != 0);
credit_round_ticks_ = sys_cfg.streaming_credit_round_interval_us / interval; credit_round_ticks_ = sys_cfg.streaming_credit_round_interval_us / interval;
CAF_LOG_DEBUG(CAF_ARG(max_batch_delay_ticks_) CAF_ASSERT(credit_round_ticks_ != 0);
CAF_LOG_DEBUG(CAF_ARG(interval) << CAF_ARG(max_batch_delay_ticks_)
<< CAF_ARG(credit_round_ticks_)); << CAF_ARG(credit_round_ticks_));
} }
......
...@@ -541,9 +541,8 @@ struct test_coordinator_fixture { ...@@ -541,9 +541,8 @@ struct test_coordinator_fixture {
sched.clock().time_per_unit.emplace(caf::atom("batch"), sched.clock().time_per_unit.emplace(caf::atom("batch"),
caf::timespan{1000}); caf::timespan{1000});
// Compute reasonable step size. // Compute reasonable step size.
auto cycle_count = caf::detail::gcd(credit_round_interval.count(), auto cycle_us = cfg.streaming_tick_duration_us();
max_batch_delay.count()); streaming_cycle = caf::timespan{us_t{cycle_us}};
streaming_cycle = caf::timespan{cycle_count};
// Make sure the current time isn't 0. // Make sure the current time isn't 0.
sched.clock().current_time += streaming_cycle; sched.clock().current_time += streaming_cycle;
} }
......
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