Commit 3f470f6f authored by Dominik Charousset's avatar Dominik Charousset

Use new get_or API for work-stealing setup

parent c21f0426
...@@ -313,13 +313,13 @@ public: ...@@ -313,13 +313,13 @@ public:
// -- work-stealing parameters ----------------------------------------------- // -- work-stealing parameters -----------------------------------------------
size_t work_stealing_aggressive_poll_attempts; size_t work_stealing_aggressive_poll_attempts CAF_DEPRECATED;
size_t work_stealing_aggressive_steal_interval; size_t work_stealing_aggressive_steal_interval CAF_DEPRECATED;
size_t work_stealing_moderate_poll_attempts; size_t work_stealing_moderate_poll_attempts CAF_DEPRECATED;
size_t work_stealing_moderate_steal_interval; size_t work_stealing_moderate_steal_interval CAF_DEPRECATED;
size_t work_stealing_moderate_sleep_duration_us; size_t work_stealing_moderate_sleep_duration_us CAF_DEPRECATED;
size_t work_stealing_relaxed_steal_interval; size_t work_stealing_relaxed_steal_interval CAF_DEPRECATED;
size_t work_stealing_relaxed_sleep_duration_us; size_t work_stealing_relaxed_sleep_duration_us CAF_DEPRECATED;
// -- logger parameters ------------------------------------------------------ // -- logger parameters ------------------------------------------------------
......
...@@ -18,18 +18,18 @@ ...@@ -18,18 +18,18 @@
#pragma once #pragma once
#include <deque> #include <array>
#include <chrono> #include <chrono>
#include <thread>
#include <random>
#include <cstddef> #include <cstddef>
#include <deque>
#include <random>
#include <thread>
#include "caf/resumable.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/policy/unprofiled.hpp"
#include "caf/detail/double_ended_queue.hpp" #include "caf/detail/double_ended_queue.hpp"
#include "caf/policy/unprofiled.hpp"
#include "caf/resumable.hpp"
#include "caf/timespan.hpp"
namespace caf { namespace caf {
namespace policy { namespace policy {
...@@ -43,14 +43,12 @@ public: ...@@ -43,14 +43,12 @@ 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. // configuration for aggressive/moderate/relaxed poll strategies.
struct poll_strategy { struct poll_strategy {
size_t attempts; size_t attempts;
size_t step_size; size_t step_size;
size_t steal_interval; size_t steal_interval;
usec sleep_duration; timespan 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.
...@@ -65,23 +63,8 @@ public: ...@@ -65,23 +63,8 @@ public:
// Holds job job queue of a worker and a random number generator. // Holds job job queue of a worker and a random number generator.
struct worker_data { struct worker_data {
inline explicit worker_data(scheduler::abstract_coordinator* p) explicit worker_data(scheduler::abstract_coordinator* p);
: rengine(std::random_device{}()), worker_data(const worker_data& other);
// no need to worry about wrap-around; if `p->num_workers() < 2`,
// `uniform` will not be used anyway
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
}
// This queue is exposed to other workers that may attempt to steal jobs // This queue is exposed to other workers that may attempt to steal jobs
// from it and the central scheduling unit can push new jobs to the queue. // from it and the central scheduling unit can push new jobs to the queue.
...@@ -89,7 +72,7 @@ public: ...@@ -89,7 +72,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]; std::array<poll_strategy, 3> strategies;
}; };
// Goes on a raid in quest for a shiny new job. // Goes on a raid in quest for a shiny new job.
......
...@@ -65,6 +65,8 @@ public: ...@@ -65,6 +65,8 @@ public:
return system_; return system_;
} }
const actor_system_config& config() const;
inline size_t max_throughput() const { inline size_t max_throughput() const {
return max_throughput_; return max_throughput_;
} }
......
...@@ -61,19 +61,23 @@ public: ...@@ -61,19 +61,23 @@ public:
protected: protected:
void start() override { void start() override {
// initialize workers vector // Create initial state for all workers.
typename worker_type::policy_data init{this};
// Prepare workers vector.
auto num = num_workers(); auto num = num_workers();
workers_.reserve(num); workers_.reserve(num);
// Create worker instanes.
for (size_t i = 0; i < num; ++i) for (size_t i = 0; i < num; ++i)
workers_.emplace_back(new worker_type(i, this, max_throughput_)); workers_.emplace_back(new worker_type(i, this, init, max_throughput_));
// start all workers now that all workers have been initialized // Start all workers.
for (auto& w : workers_) for (auto& w : workers_)
w->start(); w->start();
// launch thread for dispatching timeouts and delayed messages // Launch an additional background thread for dispatching timeouts and
// delayed messages.
timer_ = std::thread{[&] { timer_ = std::thread{[&] {
clock_.run_dispatch_loop(); clock_.run_dispatch_loop();
}}; }};
// run remaining startup code // Run remaining startup code.
super::start(); super::start();
} }
......
...@@ -40,12 +40,13 @@ public: ...@@ -40,12 +40,13 @@ public:
using coordinator_ptr = coordinator<Policy>*; using coordinator_ptr = coordinator<Policy>*;
using policy_data = typename Policy::worker_data; using policy_data = typename Policy::worker_data;
worker(size_t worker_id, coordinator_ptr worker_parent, size_t throughput) worker(size_t worker_id, coordinator_ptr worker_parent,
const policy_data& init, size_t throughput)
: execution_unit(&worker_parent->system()), : execution_unit(&worker_parent->system()),
max_throughput_(throughput), max_throughput_(throughput),
id_(worker_id), id_(worker_id),
parent_(worker_parent), parent_(worker_parent),
data_(worker_parent) { data_(init) {
// nop // nop
} }
......
...@@ -235,6 +235,10 @@ public: ...@@ -235,6 +235,10 @@ public:
* implementation of coordinator * * implementation of coordinator *
******************************************************************************/ ******************************************************************************/
const actor_system_config& abstract_coordinator::config() const {
return system_.config();
}
bool abstract_coordinator::detaches_utility_actors() const { bool abstract_coordinator::detaches_utility_actors() const {
return true; return true;
} }
......
...@@ -18,6 +18,15 @@ ...@@ -18,6 +18,15 @@
#include "caf/policy/work_stealing.hpp" #include "caf/policy/work_stealing.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/config_value.hpp"
#include "caf/defaults.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#define CONFIG(str_name, var_name) \
get_or(p->config(), "work-stealing." str_name, \
defaults::work_stealing::var_name)
namespace caf { namespace caf {
namespace policy { namespace policy {
...@@ -25,5 +34,29 @@ work_stealing::~work_stealing() { ...@@ -25,5 +34,29 @@ work_stealing::~work_stealing() {
// nop // nop
} }
work_stealing::worker_data::worker_data(scheduler::abstract_coordinator* p)
: rengine(std::random_device{}()),
// no need to worry about wrap-around; if `p->num_workers() < 2`,
// `uniform` will not be used anyway
uniform(0, p->num_workers() - 2),
strategies{{
{CONFIG("aggressive-poll-attempts", aggressive_poll_attempts), 1,
CONFIG("aggressive-steal-interval", aggressive_steal_interval),
timespan{0}},
{CONFIG("moderate-poll-attempts", moderate_poll_attempts), 1,
CONFIG("moderate-steal-interval", moderate_steal_interval),
CONFIG("moderate-sleep-duration", moderate_sleep_duration)},
{1, 0, CONFIG("relaxed-steal-interval", relaxed_steal_interval),
CONFIG("relaxed-sleep-duration", relaxed_sleep_duration)}}} {
// nop
}
work_stealing::worker_data::worker_data(const worker_data& other)
: rengine(std::random_device{}()),
uniform(other.uniform),
strategies(other.strategies) {
// nop
}
} // namespace policy } // namespace policy
} // namespace caf } // namespace caf
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