Commit 656d0bd2 authored by Dominik Charousset's avatar Dominik Charousset

Use timespans for stream parameters

parent eabbc579
...@@ -264,44 +264,21 @@ public: ...@@ -264,44 +264,21 @@ public:
/// Credentials for connecting to the bootstrap node. /// Credentials for connecting to the bootstrap node.
std::string bootstrap_node; std::string bootstrap_node;
// -- streaming parameters --------------------------------------------------- // -- stream parameters ------------------------------------------------------
/// @private /// @private
size_t streaming_desired_batch_complexity_us; timespan stream_desired_batch_complexity;
/// @private /// @private
size_t streaming_max_batch_delay_us; timespan stream_max_batch_delay;
/// @private /// @private
size_t streaming_credit_round_interval_us; timespan stream_credit_round_interval;
/// @private /// @private
size_t streaming_tick_duration_us() const noexcept; timespan stream_tick_duration() const noexcept;
/// Returns the greatest common divisor of `streaming_max_batch_delay` and timespan streaming_credit_round_interval() const noexcept CAF_DEPRECATED;
/// `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 -------------------------------------------------- // -- scheduling parameters --------------------------------------------------
......
...@@ -29,7 +29,7 @@ ...@@ -29,7 +29,7 @@
namespace caf { namespace caf {
namespace defaults { namespace defaults {
namespace streaming { namespace stream {
extern const timespan desired_batch_complexity; extern const timespan desired_batch_complexity;
extern const timespan max_batch_delay; extern const timespan max_batch_delay;
......
...@@ -58,9 +58,9 @@ actor_system_config::actor_system_config() ...@@ -58,9 +58,9 @@ actor_system_config::actor_system_config()
add_message_type_impl<std::vector<atom_value>>("std::vector<@atom>"); add_message_type_impl<std::vector<atom_value>>("std::vector<@atom>");
add_message_type_impl<std::vector<message>>("std::vector<@message>"); add_message_type_impl<std::vector<message>>("std::vector<@message>");
// (1) hard-coded defaults // (1) hard-coded defaults
streaming_desired_batch_complexity_us = 50; stream_desired_batch_complexity = defaults::stream::desired_batch_complexity;
streaming_max_batch_delay_us = 50000; stream_max_batch_delay = defaults::stream::max_batch_delay;
streaming_credit_round_interval_us = 100000; stream_credit_round_interval = defaults::stream::credit_round_interval;
namespace sr = defaults::scheduler; namespace sr = defaults::scheduler;
auto to_ms = [](timespan x) { auto to_ms = [](timespan x) {
return static_cast<size_t>(x.count() / 1000000); return static_cast<size_t>(x.count() / 1000000);
...@@ -101,13 +101,13 @@ actor_system_config::actor_system_config() ...@@ -101,13 +101,13 @@ actor_system_config::actor_system_config()
.add<bool>("help,h?", "print help and exit") .add<bool>("help,h?", "print help and exit")
.add<bool>("long-help", "print all help options and exit") .add<bool>("long-help", "print all help options and exit")
.add<bool>("dump-config", "print configuration in INI format and exit"); .add<bool>("dump-config", "print configuration in INI format and exit");
opt_group{custom_options_, "streaming"} opt_group{custom_options_, "stream"}
.add(streaming_desired_batch_complexity_us, "desired-batch-complexity-us", .add(stream_desired_batch_complexity, "desired-batch-complexity",
"sets the desired timespan for a single batch") "sets the desired timespan for a single batch")
.add(streaming_max_batch_delay_us, "max-batch-delay-us", .add(stream_max_batch_delay, "max-batch-delay",
"sets the maximum delay for sending underfull batches in microseconds") "sets the maximum delay for sending underfull batches")
.add(streaming_credit_round_interval_us, "credit-round-interval-us", .add(stream_credit_round_interval, "credit-round-interval",
"sets the length of credit intervals in microseconds"); "sets the length of credit intervals");
opt_group{custom_options_, "scheduler"} opt_group{custom_options_, "scheduler"}
.add(scheduler_policy, "policy", .add(scheduler_policy, "policy",
"sets the scheduling policy to either 'stealing' (default) or 'sharing'") "sets the scheduling policy to either 'stealing' (default) or 'sharing'")
...@@ -393,9 +393,14 @@ actor_system_config& actor_system_config::set_impl(string_view name, ...@@ -393,9 +393,14 @@ actor_system_config& actor_system_config::set_impl(string_view name,
return *this; return *this;
} }
size_t actor_system_config::streaming_tick_duration_us() const noexcept { timespan actor_system_config::stream_tick_duration() const noexcept {
return caf::detail::gcd(streaming_credit_round_interval_us, auto ns_count = caf::detail::gcd(stream_credit_round_interval.count(),
streaming_max_batch_delay_us); stream_max_batch_delay.count());
return timespan{ns_count};
}
timespan actor_system_config::streaming_credit_round_interval() const noexcept {
return stream_credit_round_interval;
} }
std::string actor_system_config::render_sec(uint8_t x, atom_value, std::string actor_system_config::render_sec(uint8_t x, atom_value,
......
...@@ -19,6 +19,7 @@ ...@@ -19,6 +19,7 @@
#include "caf/defaults.hpp" #include "caf/defaults.hpp"
#include <algorithm> #include <algorithm>
#include <chrono>
#include <limits> #include <limits>
#include <thread> #include <thread>
...@@ -41,7 +42,7 @@ constexpr caf::timespan ms(ms_t::rep x) { ...@@ -41,7 +42,7 @@ constexpr caf::timespan ms(ms_t::rep x) {
namespace caf { namespace caf {
namespace defaults { namespace defaults {
namespace streaming { namespace stream {
const timespan desired_batch_complexity = us(50); const timespan desired_batch_complexity = us(50);
const timespan max_batch_delay = ms(5); const timespan max_batch_delay = ms(5);
......
...@@ -111,15 +111,17 @@ scheduled_actor::scheduled_actor(actor_config& cfg) ...@@ -111,15 +111,17 @@ 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.stream_tick_duration();
CAF_ASSERT(interval != 0); CAF_ASSERT(interval.count() > 0);
stream_ticks_.interval(std::chrono::microseconds(interval)); stream_ticks_.interval(interval);
CAF_ASSERT(sys_cfg.streaming_max_batch_delay_us != 0); CAF_ASSERT(sys_cfg.stream_max_batch_delay.count() > 0);
max_batch_delay_ticks_ = sys_cfg.streaming_max_batch_delay_us / interval; max_batch_delay_ticks_ = sys_cfg.stream_max_batch_delay.count()
CAF_ASSERT(max_batch_delay_ticks_ != 0); / interval.count();
CAF_ASSERT(sys_cfg.streaming_credit_round_interval_us != 0); CAF_ASSERT(max_batch_delay_ticks_ > 0);
credit_round_ticks_ = sys_cfg.streaming_credit_round_interval_us / interval; CAF_ASSERT(sys_cfg.streaming_credit_round_interval_us > 0);
CAF_ASSERT(credit_round_ticks_ != 0); credit_round_ticks_ = sys_cfg.stream_credit_round_interval.count()
/ interval.count();
CAF_ASSERT(credit_round_ticks_ > 0);
CAF_LOG_DEBUG(CAF_ARG(interval) << CAF_ARG(max_batch_delay_ticks_) CAF_LOG_DEBUG(CAF_ARG(interval) << CAF_ARG(max_batch_delay_ticks_)
<< CAF_ARG(credit_round_ticks_)); << CAF_ARG(credit_round_ticks_));
} }
...@@ -1103,8 +1105,7 @@ scheduled_actor::advance_streams(actor_clock::time_point now) { ...@@ -1103,8 +1105,7 @@ scheduled_actor::advance_streams(actor_clock::time_point now) {
CAF_LOG_DEBUG("new credit round"); CAF_LOG_DEBUG("new credit round");
auto cycle = stream_ticks_.interval(); auto cycle = stream_ticks_.interval();
cycle *= static_cast<decltype(cycle)::rep>(credit_round_ticks_); cycle *= static_cast<decltype(cycle)::rep>(credit_round_ticks_);
auto bc_us = home_system().config().streaming_desired_batch_complexity_us; auto bc = home_system().config().stream_desired_batch_complexity;
auto bc = std::chrono::microseconds(bc_us);
auto& qs = get<2>(mailbox_.queue().queues()).queues(); auto& qs = get<2>(mailbox_.queue().queues()).queues();
for (auto& kvp : qs) { for (auto& kvp : qs) {
auto inptr = kvp.second.policy().handler.get(); auto inptr = kvp.second.policy().handler.get();
......
...@@ -121,10 +121,9 @@ void stream_manager::advance() { ...@@ -121,10 +121,9 @@ void stream_manager::advance() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// Try to emit more credit. // Try to emit more credit.
if (!inbound_paths_.empty()) { if (!inbound_paths_.empty()) {
using std::chrono::microseconds;
auto& cfg = self_->system().config(); auto& cfg = self_->system().config();
microseconds bc{cfg.streaming_desired_batch_complexity_us}; auto bc = cfg.stream_desired_batch_complexity;
microseconds interval{cfg.streaming_credit_round_interval_us}; auto interval = cfg.stream_credit_round_interval;
auto& mbox = self_->mailbox(); auto& mbox = self_->mailbox();
auto& qs = get<2>(mbox.queue().queues()).queues(); auto& qs = get<2>(mbox.queue().queues()).queues();
// Iterate all queues for inbound traffic. // Iterate all queues for inbound traffic.
......
...@@ -213,7 +213,7 @@ TESTEE(doubler) { ...@@ -213,7 +213,7 @@ TESTEE(doubler) {
struct fixture : test_coordinator_fixture<> { struct fixture : test_coordinator_fixture<> {
void tick() { void tick() {
advance_time(cfg.streaming_credit_round_interval()); advance_time(cfg.stream_credit_round_interval);
} }
}; };
......
...@@ -567,7 +567,7 @@ public: ...@@ -567,7 +567,7 @@ public:
caf::timespan{1000}); caf::timespan{1000});
// Make sure the current time isn't 0. // Make sure the current time isn't 0.
sched.clock().current_time += std::chrono::hours(1); sched.clock().current_time += std::chrono::hours(1);
credit_round_interval = cfg.streaming_credit_round_interval(); credit_round_interval = cfg.stream_credit_round_interval;
} }
virtual ~test_coordinator_fixture() { virtual ~test_coordinator_fixture() {
......
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