Commit e0e7e4da authored by Dominik Charousset's avatar Dominik Charousset

Make work stealing configurable, closes #464

parent 0aa52c85
...@@ -7,22 +7,39 @@ ...@@ -7,22 +7,39 @@
policy='stealing' policy='stealing'
; configures whether the scheduler generates profiling output ; configures whether the scheduler generates profiling output
enable-profiling=false enable-profiling=false
; can be overriden to force a fixed number of threads ; forces a fixed number of threads if set
max-threads=<number of cores> max-threads=<number of cores>
; configures the maximum number of messages actors can consume in one run ; maximum number of messages actors can consume in one run
max-throughput=<infinite> max-throughput=<infinite>
; measurement resolution in milliseconds (only if profiling is enabled) ; measurement resolution in milliseconds (only if profiling is enabled)
profiling-ms-resolution=100 profiling-ms-resolution=100
; output file for profiler data (only if profiling is enabled) ; output file for profiler data (only if profiling is enabled)
profiling-output-file="/dev/null" profiling-output-file="/dev/null"
; when using 'stealing' as scheduler policy
[work-stealing]
; number of zero-sleep-interval polling attempts
aggressive-poll-attempts=100
; frequency of steal attempts during aggressive polling
aggressive-steal-interval=10
; number of moderately aggressive polling attempts
moderate-poll-attempts=500
; frequency of steal attempts during moderate polling
moderate-steal-interval=5
; sleep interval in microseconds between poll attempts
moderate-sleep-duration=50
; frequency of steal attempts during relaxed polling
relaxed-steal-interval=1
; sleep interval in microseconds between poll attempts
relaxed-sleep-duration=10000
; when loading io::middleman ; when loading io::middleman
[middleman] [middleman]
; configures whether MMs try to span a full mesh ; configures whether MMs try to span a full mesh
enable-automatic-connections=false enable-automatic-connections=false
; accepted alternative: 'asio' (only when compiling CAF with ASIO) ; accepted alternative: 'asio' (only when compiling CAF with ASIO)
network-backend='default' network-backend='default'
; sets the maximum number of consecutive I/O reads per broker ; maximum number of consecutive I/O reads per broker
max-consecutive-reads=50 max-consecutive-reads=50
; heartbeat message interval in ms (0 disables heartbeating) ; heartbeat message interval in ms (0 disables heartbeating)
heartbeat-interval=0 heartbeat-interval=0
......
...@@ -196,6 +196,15 @@ public: ...@@ -196,6 +196,15 @@ public:
size_t scheduler_profiling_ms_resolution; size_t scheduler_profiling_ms_resolution;
std::string scheduler_profiling_output_file; std::string scheduler_profiling_output_file;
// Config parameters for work-stealing strategy
size_t work_stealing_aggressive_poll_attempts;
size_t work_stealing_aggressive_steal_interval;
size_t work_stealing_moderate_poll_attempts;
size_t work_stealing_moderate_steal_interval;
size_t work_stealing_moderate_sleep_duration_us;
size_t work_stealing_relaxed_steal_interval;
size_t work_stealing_relaxed_sleep_duration_us;
// Config parameters of middleman. // Config parameters of middleman.
atom_value middleman_network_backend; atom_value middleman_network_backend;
bool middleman_enable_automatic_connections; bool middleman_enable_automatic_connections;
......
...@@ -27,6 +27,8 @@ ...@@ -27,6 +27,8 @@
#include <cstddef> #include <cstddef>
#include "caf/resumable.hpp" #include "caf/resumable.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/policy/unprofiled.hpp" #include "caf/policy/unprofiled.hpp"
#include "caf/detail/double_ended_queue.hpp" #include "caf/detail/double_ended_queue.hpp"
...@@ -43,6 +45,16 @@ public: ...@@ -43,6 +45,16 @@ public:
// A thread-safe queue implementation. // A thread-safe queue implementation.
using queue_type = detail::double_ended_queue<resumable>; using queue_type = detail::double_ended_queue<resumable>;
using usec = std::chrono::microseconds;
// configuration for aggressive/moderate/relaxed poll strategies.
struct poll_strategy {
size_t attempts;
size_t step_size;
size_t steal_interval;
usec sleep_duration;
};
// The coordinator has only a counter for round-robin enqueue to its workers. // The coordinator has only a counter for round-robin enqueue to its workers.
struct coordinator_data { struct coordinator_data {
inline explicit coordinator_data(scheduler::abstract_coordinator*) inline explicit coordinator_data(scheduler::abstract_coordinator*)
...@@ -59,7 +71,17 @@ public: ...@@ -59,7 +71,17 @@ public:
: rengine(std::random_device{}()), : rengine(std::random_device{}()),
// no need to worry about wrap-around; if `p->num_workers() < 2`, // no need to worry about wrap-around; if `p->num_workers() < 2`,
// `uniform` will not be used anyway // `uniform` will not be used anyway
uniform(0, p->num_workers() - 2) { uniform(0, p->num_workers() - 2),
strategies{
{p->system().config().work_stealing_aggressive_poll_attempts, 1,
p->system().config().work_stealing_aggressive_steal_interval,
usec{0}},
{p->system().config().work_stealing_moderate_poll_attempts, 1,
p->system().config().work_stealing_moderate_steal_interval,
usec{p->system().config().work_stealing_moderate_sleep_duration_us}},
{1, 0, p->system().config().work_stealing_relaxed_steal_interval,
usec{p->system().config().work_stealing_relaxed_sleep_duration_us}}
} {
// nop // nop
} }
...@@ -69,6 +91,7 @@ public: ...@@ -69,6 +91,7 @@ public:
// needed to generate pseudo random numbers // needed to generate pseudo random numbers
std::default_random_engine rengine; std::default_random_engine rengine;
std::uniform_int_distribution<size_t> uniform; std::uniform_int_distribution<size_t> uniform;
poll_strategy strategies[3];
}; };
// Goes on a raid in quest for a shiny new job. // Goes on a raid in quest for a shiny new job.
...@@ -119,35 +142,21 @@ public: ...@@ -119,35 +142,21 @@ public:
// on and poll every 10 ms; this strategy strives to minimize the // on and poll every 10 ms; this strategy strives to minimize the
// downside of "busy waiting", which still performs much better than a // downside of "busy waiting", which still performs much better than a
// "signalizing" implementation based on mutexes and conition variables // "signalizing" implementation based on mutexes and conition variables
struct poll_strategy { auto& strategies = d(self).strategies;
size_t attempts;
size_t step_size;
size_t steal_interval;
std::chrono::microseconds sleep_duration;
};
poll_strategy strategies[3] = {
// aggressive polling (100x) without sleep interval
{100, 1, 10, std::chrono::microseconds{0}},
// moderate polling (500x) with 50 us sleep interval
{500, 1, 5, std::chrono::microseconds{50}},
// relaxed polling (infinite attempts) with 10 ms sleep interval
{101, 0, 1, std::chrono::microseconds{10000}}
};
resumable* job = nullptr; resumable* job = nullptr;
for (auto& strat : strategies) { for (auto& strat : strategies) {
for (size_t i = 0; i < strat.attempts; i += strat.step_size) { for (size_t i = 0; i < strat.attempts; i += strat.step_size) {
job = d(self).queue.take_head(); job = d(self).queue.take_head();
if (job) { if (job)
return job; return job;
}
// try to steal every X poll attempts // try to steal every X poll attempts
if ((i % strat.steal_interval) == 0) { if ((i % strat.steal_interval) == 0) {
job = try_steal(self); job = try_steal(self);
if (job) { if (job)
return job; return job;
}
} }
std::this_thread::sleep_for(strat.sleep_duration); if (strat.sleep_duration.count() > 0)
std::this_thread::sleep_for(strat.sleep_duration);
} }
} }
// unreachable, because the last strategy loops // unreachable, because the last strategy loops
......
...@@ -87,6 +87,13 @@ actor_system_config::actor_system_config() ...@@ -87,6 +87,13 @@ actor_system_config::actor_system_config()
scheduler_max_throughput = std::numeric_limits<size_t>::max(); scheduler_max_throughput = std::numeric_limits<size_t>::max();
scheduler_enable_profiling = false; scheduler_enable_profiling = false;
scheduler_profiling_ms_resolution = 100; scheduler_profiling_ms_resolution = 100;
work_stealing_aggressive_poll_attempts = 100;
work_stealing_aggressive_steal_interval = 10;
work_stealing_moderate_poll_attempts = 500;
work_stealing_moderate_steal_interval = 5;
work_stealing_moderate_sleep_duration_us = 50;
work_stealing_relaxed_steal_interval = 1;
work_stealing_relaxed_sleep_duration_us = 10000;
middleman_network_backend = atom("default"); middleman_network_backend = atom("default");
middleman_enable_automatic_connections = false; middleman_enable_automatic_connections = false;
middleman_max_consecutive_reads = 50; middleman_max_consecutive_reads = 50;
...@@ -106,6 +113,21 @@ actor_system_config::actor_system_config() ...@@ -106,6 +113,21 @@ actor_system_config::actor_system_config()
"sets the rate in ms in which the profiler collects data") "sets the rate in ms in which the profiler collects data")
.add(scheduler_profiling_output_file, "profiling-output-file", .add(scheduler_profiling_output_file, "profiling-output-file",
"sets the output file for the profiler"); "sets the output file for the profiler");
opt_group(options_, "work-stealing")
.add(work_stealing_aggressive_poll_attempts, "aggressive-poll-attempts",
"sets the number of zero-sleep-interval polling attempts")
.add(work_stealing_aggressive_steal_interval, "aggressive-steal-interval",
"sets the frequency of steal attempts during aggressive polling")
.add(work_stealing_moderate_poll_attempts, "moderate-poll-attempts",
"sets the number of moderately aggressive polling attempts")
.add(work_stealing_moderate_steal_interval, "moderate-steal-interval",
"sets the frequency of steal attempts during moderate polling")
.add(work_stealing_moderate_sleep_duration_us, "moderate-sleep-duration",
"sets the sleep interval between poll attempts during moderate polling")
.add(work_stealing_relaxed_steal_interval, "relaxed-steal-interval",
"sets the frequency of steal attempts during relaxed polling")
.add(work_stealing_relaxed_sleep_duration_us, "relaxed-sleep-duration",
"sets the sleep interval between poll attempts during relaxed polling");
opt_group{options_, "middleman"} opt_group{options_, "middleman"}
.add(middleman_network_backend, "network-backend", .add(middleman_network_backend, "network-backend",
"sets the network backend to either 'default' or 'asio' (if available)") "sets the network backend to either 'default' or 'asio' (if available)")
......
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