Commit ad53b9f9 authored by Dominik Charousset's avatar Dominik Charousset

Use new get_or API for scheduler setup

parent 324d1b7c
......@@ -304,12 +304,12 @@ public:
// -- scheduling parameters --------------------------------------------------
atom_value scheduler_policy;
size_t scheduler_max_threads;
size_t scheduler_max_throughput;
bool scheduler_enable_profiling;
size_t scheduler_profiling_ms_resolution;
std::string scheduler_profiling_output_file;
atom_value scheduler_policy CAF_DEPRECATED;
size_t scheduler_max_threads CAF_DEPRECATED;
size_t scheduler_max_throughput CAF_DEPRECATED;
bool scheduler_enable_profiling CAF_DEPRECATED;
size_t scheduler_profiling_ms_resolution CAF_DEPRECATED;
std::string scheduler_profiling_output_file CAF_DEPRECATED;
// -- work-stealing parameters -----------------------------------------------
......
......@@ -38,6 +38,7 @@
#include "caf/message.hpp"
#include "caf/node_id.hpp"
#include "caf/behavior.hpp"
#include "caf/defaults.hpp"
#include "caf/duration.hpp"
#include "caf/expected.hpp"
#include "caf/exec_main.hpp"
......
......@@ -56,6 +56,11 @@ public:
config_option_adder& add_us(size_t& ref, const char* name,
const char* description);
/// For backward compatibility only. Do not use for new code!
/// @private
config_option_adder& add_ms(size_t& ref, const char* name,
const char* description);
private:
// -- properties -------------------------------------------------------------
......
......@@ -40,6 +40,7 @@ extern const timespan credit_round_interval;
namespace scheduler {
extern const atom_value policy;
extern const char* profiling_output_file;
extern const size_t max_threads;
extern const size_t max_throughput;
extern const timespan profiling_resolution;
......
......@@ -79,6 +79,12 @@ config_option make_us_resolution_config_option(size_t& storage,
const char* name,
const char* description);
// Reads timespans, but stores an integer representing millisecond resolution.
config_option make_ms_resolution_config_option(size_t& storage,
const char* category,
const char* name,
const char* description);
// -- specializations for common types -----------------------------------------
#define CAF_SPECIALIZE_MAKE_CONFIG_OPTION(type) \
......
......@@ -38,13 +38,11 @@
#include <unordered_map>
#include "caf/actor_system_config.hpp"
#include "caf/scheduler/coordinator.hpp"
#include "caf/defaults.hpp"
#include "caf/logger.hpp"
#include "caf/policy/profiled.hpp"
#include "caf/policy/work_stealing.hpp"
#include "caf/logger.hpp"
#include "caf/scheduler/coordinator.hpp"
namespace caf {
namespace scheduler {
......@@ -178,14 +176,19 @@ public:
}
void init(actor_system_config& cfg) override {
namespace sr = defaults::scheduler;
super::init(cfg);
file_.open(cfg.scheduler_profiling_output_file);
auto fname = get_or(cfg, "scheduler.profiling-output-file",
sr::profiling_output_file);
file_.open(fname);
if (!file_)
std::cerr << R"([WARNING] could not open file ")"
<< cfg.scheduler_profiling_output_file
<< fname
<< R"(" (no profiler output will be generated))"
<< std::endl;
resolution_ = msec{cfg.scheduler_profiling_ms_resolution};
auto res = get_or(cfg, "scheduler.profiling-resolution",
sr::profiling_resolution);
resolution_ = std::chrono::duration_cast<msec>(res);
}
void start() override {
......
......@@ -27,22 +27,20 @@
#include <unordered_map>
#include <condition_variable>
#include "caf/send.hpp"
#include "caf/actor_ostream.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/after.hpp"
#include "caf/others.hpp"
#include "caf/defaults.hpp"
#include "caf/duration.hpp"
#include "caf/actor_system.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/actor_ostream.hpp"
#include "caf/system_messages.hpp"
#include "caf/logger.hpp"
#include "caf/others.hpp"
#include "caf/policy/work_stealing.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/scheduler/coordinator.hpp"
#include "caf/policy/work_stealing.hpp"
#include "caf/logger.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/send.hpp"
#include "caf/system_messages.hpp"
namespace caf {
namespace scheduler {
......@@ -251,8 +249,9 @@ void abstract_coordinator::start() {
}
void abstract_coordinator::init(actor_system_config& cfg) {
max_throughput_ = cfg.scheduler_max_throughput;
num_workers_ = cfg.scheduler_max_threads;
namespace sr = defaults::scheduler;
max_throughput_ = get_or(cfg, "scheduler.max-throughput", sr::max_throughput);
num_workers_ = get_or(cfg, "scheduler.max-threads", sr::max_threads);
}
actor_system::module::id_t abstract_coordinator::id() const {
......
......@@ -239,16 +239,18 @@ actor_system::actor_system(actor_system_config& cfg)
profiled_sharing = 0x0102
};
sched_conf sc = stealing;
if (cfg.scheduler_policy == atom("sharing"))
namespace sr = defaults::scheduler;
auto sr_policy = get_or(cfg, "scheduler.policy", sr::policy);
if (sr_policy == atom("sharing"))
sc = sharing;
else if (cfg.scheduler_policy == atom("testing"))
else if (sr_policy == atom("testing"))
sc = testing;
else if (cfg.scheduler_policy != atom("stealing"))
std::cerr << "[WARNING] " << deep_to_string(cfg.scheduler_policy)
else if (sr_policy != atom("stealing"))
std::cerr << "[WARNING] " << deep_to_string(sr_policy)
<< " is an unrecognized scheduler pollicy, "
"falling back to 'stealing' (i.e. work-stealing)"
<< std::endl;
if (cfg.scheduler_enable_profiling)
if (get_or(cfg, "scheduler.enable-profiling", false))
sc = static_cast<sched_conf>(sc | profiled);
switch (sc) {
default: // any invalid configuration falls back to work stealing
......
......@@ -60,11 +60,15 @@ actor_system_config::actor_system_config()
streaming_desired_batch_complexity_us = 50;
streaming_max_batch_delay_us = 50000;
streaming_credit_round_interval_us = 100000;
scheduler_policy = atom("stealing");
scheduler_max_threads = std::max(std::thread::hardware_concurrency(), 4u);
scheduler_max_throughput = std::numeric_limits<size_t>::max();
namespace sr = defaults::scheduler;
auto to_ms = [](timespan x) {
return static_cast<size_t>(x.count() / 1000000);
};
scheduler_policy = sr::policy;
scheduler_max_threads = sr::max_threads;
scheduler_max_throughput = sr::max_throughput;
scheduler_enable_profiling = false;
scheduler_profiling_ms_resolution = 100;
scheduler_profiling_ms_resolution = to_ms(sr::profiling_resolution);
namespace ws = defaults::work_stealing;
auto to_us = [](timespan x) {
return static_cast<size_t>(x.count() / 1000);
......@@ -112,7 +116,9 @@ actor_system_config::actor_system_config()
"sets the maximum number of messages an actor consumes before yielding")
.add(scheduler_enable_profiling, "enable-profiling",
"enables or disables profiler output")
.add(scheduler_profiling_ms_resolution, "profiling-ms-resolution",
.add_ms(scheduler_profiling_ms_resolution, "profiling-ms-resolution",
"deprecated (use profiling-resolution instead)")
.add_ms(scheduler_profiling_ms_resolution, "profiling-resolution",
"sets the rate in ms in which the profiler collects data")
.add(scheduler_profiling_output_file, "profiling-output-file",
"sets the output file for the profiler");
......
......@@ -35,13 +35,18 @@ config_option_adder& config_option_adder::add_neg(bool& ref, const char* name,
name, description));
}
config_option_adder& config_option_adder::add_us(size_t& ref,
const char* name,
config_option_adder& config_option_adder::add_us(size_t& ref, const char* name,
const char* description) {
return add_impl(make_us_resolution_config_option(ref, category_,
name, description));
}
config_option_adder& config_option_adder::add_ms(size_t& ref, const char* name,
const char* description) {
return add_impl(make_ms_resolution_config_option(ref, category_,
name, description));
}
config_option_adder& config_option_adder::add_impl(config_option&& opt) {
xs_.add(std::move(opt));
return *this;
......
......@@ -52,6 +52,7 @@ const timespan credit_round_interval = ms(10);
namespace scheduler {
const atom_value policy = atom("stealing");
const char* profiling_output_file = "";
const size_t max_threads = std::max(std::thread::hardware_concurrency(), 4u);
const size_t max_throughput = std::numeric_limits<size_t>::max();
const timespan profiling_resolution = ms(100);
......
......@@ -55,7 +55,7 @@ void await_all_locals_down(actor_system& sys, std::initializer_list<actor> xs) {
ys.push_back(x);
}
// Don't block when using the test coordinator.
if (sys.config().scheduler_policy != atom("testing"))
if (atom("testing") != get_or(sys.config(), "scheduler.policy", atom("")))
self->wait_for(ys);
}
......
......@@ -94,7 +94,7 @@ meta_state bool_meta{bool_check, bool_store, bool_get,
meta_state bool_neg_meta{bool_check, bool_store_neg, bool_get_neg,
detail::type_name<bool>()};
meta_state entangled_meta{
meta_state us_res_meta{
[](const config_value& x) -> error {
if (holds_alternative<timespan>(x))
return none;
......@@ -111,6 +111,23 @@ meta_state entangled_meta{
detail::type_name<timespan>()
};
meta_state ms_res_meta{
[](const config_value& x) -> error {
if (holds_alternative<timespan>(x))
return none;
return make_error(pec::type_mismatch);
},
[](void* ptr, const config_value& x) {
*static_cast<size_t*>(ptr) = get<timespan>(x).count() / 1000000;
},
[](const void* ptr) -> config_value {
auto ival = static_cast<int64_t>(*static_cast<const size_t*>(ptr));
timespan val{ival / 1000000};
return config_value{val};
},
detail::type_name<timespan>()
};
DEFAULT_META(atom_value);
DEFAULT_META(size_t);
......@@ -129,7 +146,14 @@ config_option make_us_resolution_config_option(size_t& storage,
const char* category,
const char* name,
const char* description) {
return {category, name, description, &entangled_meta, &storage};
return {category, name, description, &us_res_meta, &storage};
}
config_option make_ms_resolution_config_option(size_t& storage,
const char* category,
const char* name,
const char* description) {
return {category, name, description, &ms_res_meta, &storage};
}
DEFAULT_MAKE_IMPL(atom_value)
......
......@@ -118,7 +118,7 @@ behavior tester(event_based_actor* self, const actor& aut) {
struct config : actor_system_config {
config() {
scheduler_policy = atom("testing");
set("scheduler.policy", atom("testing"));
}
};
......
......@@ -34,7 +34,7 @@ namespace {
struct fixture {
fixture() {
cfg.scheduler_policy = caf::atom("testing");
cfg.set("scheduler.policy", atom("testing"));
}
void add(logger::field_type kind) {
......
......@@ -265,7 +265,7 @@ behavior ping_multiplexed3(ping_actor* self, bool* had_timeout,
struct config : actor_system_config {
config() {
scheduler_policy = atom("testing");
set("scheduler.policy", atom("testing"));
}
};
......
......@@ -72,7 +72,7 @@ timer::behavior_type timer_impl2(timer::pointer self) {
struct config : actor_system_config {
config() {
scheduler_policy = atom("testing");
set("scheduler.policy", atom("testing"));
}
};
......
......@@ -117,7 +117,8 @@ CAF_TEST_FIXTURE_SCOPE(counting_hook, fixture<counting_thread_hook>)
CAF_TEST(counting_system_without_actor) {
assumed_init_calls = 1;
assumed_thread_count = cfg.scheduler_max_threads;
assumed_thread_count = get_or(cfg, "scheduler.max-threads",
defaults::scheduler::max_threads);
auto& sched = sys.scheduler();
if (sched.detaches_utility_actors())
assumed_thread_count += sched.num_utility_actors();
......@@ -125,7 +126,9 @@ CAF_TEST(counting_system_without_actor) {
CAF_TEST(counting_system_with_actor) {
assumed_init_calls = 1;
assumed_thread_count = cfg.scheduler_max_threads + 1;
assumed_thread_count = get_or(cfg, "scheduler.max-threads",
defaults::scheduler::max_threads)
+ 1;
auto& sched = sys.scheduler();
if (sched.detaches_utility_actors())
assumed_thread_count += sched.num_utility_actors();
......
......@@ -504,6 +504,9 @@ private:
/// Sequential ids for handles of datagram servants
int64_t servant_ids_;
/// Maximum messages per resume run.
size_t max_throughput_;
};
inline connection_handle conn_hdl_from_socket(native_socket fd) {
......
......@@ -292,7 +292,8 @@ namespace network {
epollfd_(invalid_native_socket),
shadow_(1),
pipe_reader_(*this),
servant_ids_(0) {
servant_ids_(0),
max_throughput_(0 ){
init();
epollfd_ = epoll_create1(EPOLL_CLOEXEC);
if (epollfd_ == -1) {
......@@ -843,6 +844,9 @@ void default_multiplexer::init() {
CAF_CRITICAL("WSAStartup failed");
}
# endif
namespace sr = defaults::scheduler;
max_throughput_ = get_or(system().config(), "scheduler.max-throughput",
sr::max_throughput);
}
bool default_multiplexer::poll_once(bool block) {
......@@ -870,8 +874,7 @@ bool default_multiplexer::poll_once(bool block) {
void default_multiplexer::resume(intrusive_ptr<resumable> ptr) {
CAF_LOG_TRACE("");
auto mt = system().config().scheduler_max_throughput;
switch (ptr->resume(this, mt)) {
switch (ptr->resume(this, max_throughput_)) {
case resumable::resume_later:
// Delay resumable until next cycle.
internally_posted_.emplace_back(ptr.release(), false);
......
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