Commit c1502009 authored by Dominik Charousset's avatar Dominik Charousset

Add streaming-related parameters to system config

parent 6c924275
......@@ -233,29 +233,43 @@ public:
return *this;
}
/// Stores whether the help text for this config object was
/// printed. If set to `true`, the application should not use
/// this config object to initialize an `actor_system` and
/// return from `main` immediately.
/// Sets a config by using its INI name `config_name` to `config_value`.
actor_system_config& set(const char* cn, config_value cv);
// -- parser and CLI state ---------------------------------------------------
/// Stores whether the help text was printed. If set to `true`, the
/// application should not use this config to initialize an `actor_system`
/// and instead return from `main` immediately.
bool cli_helptext_printed;
/// Stores CLI arguments that were not consumed by CAF.
message args_remainder;
// -- caf-run parameters -----------------------------------------------------
/// Stores whether this node was started in slave mode.
bool slave_mode;
/// Stores the name of this node when started in slave mode.
/// Name of this node when started in slave mode.
std::string slave_name;
/// Stores credentials for connecting to the bootstrap node
/// when using the caf-run launcher.
/// Credentials for connecting to the bootstrap node.
std::string bootstrap_node;
/// Stores CLI arguments that were not consumed by CAF.
message args_remainder;
// -- streaming parameters ---------------------------------------------------
/// Sets a config by using its INI name `config_name` to `config_value`.
actor_system_config& set(const char* cn, config_value cv);
/// Maximum delay for sending underfull batches.
size_t streaming_max_batch_delay_us;
/// Timespan between two credit rounds.
size_t streaming_credit_round_interval_us;
/// Greatest common divisor of `streaming_max_batch_delay_us` and
/// `streaming_credit_round_interval_us`.
size_t streaming_tick_duration_us;
// -- config parameters of the scheduler -------------------------------------
// -- scheduling parameters --------------------------------------------------
atom_value scheduler_policy;
size_t scheduler_max_threads;
......@@ -264,7 +278,7 @@ public:
size_t scheduler_profiling_ms_resolution;
std::string scheduler_profiling_output_file;
// -- config parameters for work-stealing ------------------------------------
// -- work-stealing parameters -----------------------------------------------
size_t work_stealing_aggressive_poll_attempts;
size_t work_stealing_aggressive_steal_interval;
......@@ -274,7 +288,7 @@ public:
size_t work_stealing_relaxed_steal_interval;
size_t work_stealing_relaxed_sleep_duration_us;
// -- config parameters for the logger ---------------------------------------
// -- logger parameters ------------------------------------------------------
std::string logger_file_name;
std::string logger_file_format;
......@@ -289,7 +303,7 @@ public:
std::string& logger_filename CAF_DEPRECATED;
std::string& logger_filter CAF_DEPRECATED;
// -- config parameters of the middleman -------------------------------------
// -- middleman parameters ---------------------------------------------------
atom_value middleman_network_backend;
std::string middleman_app_identifier;
......@@ -303,11 +317,11 @@ public:
size_t middleman_cached_udp_buffers;
size_t middleman_max_pending_msgs;
// -- config parameters of the OpenCL module ---------------------------------
// -- OpenCL parameters ------------------------------------------------------
std::string opencl_device_ids;
// -- config parameters of the OpenSSL module ---------------------------------
// -- OpenSSL parameters -----------------------------------------------------
std::string openssl_certificate;
std::string openssl_key;
......
......@@ -25,6 +25,7 @@
#include "caf/message_builder.hpp"
#include "caf/detail/gcd.hpp"
#include "caf/detail/parse_ini.hpp"
namespace caf {
......@@ -111,9 +112,11 @@ actor_system_config::actor_system_config()
add_message_type_impl<std::vector<atom_value>>("std::vector<@atom>");
add_message_type_impl<std::vector<message>>("std::vector<@message>");
// (1) hard-coded defaults
streaming_max_batch_delay_us = 50000;
streaming_credit_round_interval_us = 100000;
streaming_tick_duration_us = 50000;
scheduler_policy = atom("stealing");
scheduler_max_threads = std::max(std::thread::hardware_concurrency(),
unsigned{4});
scheduler_max_threads = std::max(std::thread::hardware_concurrency(), 4u);
scheduler_max_throughput = std::numeric_limits<size_t>::max();
scheduler_enable_profiling = false;
scheduler_profiling_ms_resolution = 100;
......@@ -141,6 +144,11 @@ actor_system_config::actor_system_config()
middleman_cached_udp_buffers = 10;
middleman_max_pending_msgs = 10;
// fill our options vector for creating INI and CLI parsers
opt_group{options_, "streaming"}
.add(streaming_max_batch_delay_us, "max-batch-delay-us",
"sets the maximum delay for sending underfull batches in microseconds")
.add(streaming_credit_round_interval_us, "credit-cycle-interval-us",
"sets the length of credit intervals in microseconds");
opt_group{options_, "scheduler"}
.add(scheduler_policy, "policy",
"sets the scheduling policy to either 'stealing' (default) or 'sharing'")
......@@ -234,10 +242,13 @@ actor_system_config::actor_system_config()
actor_system_config::actor_system_config(actor_system_config&& other)
: cli_helptext_printed(other.cli_helptext_printed),
args_remainder(std::move(other.args_remainder)),
slave_mode(other.slave_mode),
slave_name(std::move(other.slave_name)),
bootstrap_node(std::move(other.bootstrap_node)),
args_remainder(std::move(other.args_remainder)),
streaming_max_batch_delay_us(other.streaming_max_batch_delay_us),
streaming_credit_round_interval_us(
other.streaming_credit_round_interval_us),
scheduler_policy(other.scheduler_policy),
scheduler_max_threads(other.scheduler_max_threads),
scheduler_max_throughput(other.scheduler_max_throughput),
......@@ -432,6 +443,7 @@ actor_system_config& actor_system_config::parse(message& args,
if (bootstrap_node.empty())
std::cerr << "running in slave mode without bootstrap node" << endl;
}
// Verify settings.
auto verify_atom_opt = [](std::initializer_list<atom_value> xs, atom_value& x,
const char* xname) {
if (std::find(xs.begin(), xs.end(), x) == xs.end()) {
......@@ -459,6 +471,9 @@ actor_system_config& actor_system_config::parse(message& args,
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;
}
......
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