Commit 59c6e837 authored by Sebastian Woelke's avatar Sebastian Woelke

Implement NUMA-aware work stealing scheduler

parent ee702b88
...@@ -277,6 +277,15 @@ endif(CAF_ENABLE_ADDRESS_SANITIZER) ...@@ -277,6 +277,15 @@ endif(CAF_ENABLE_ADDRESS_SANITIZER)
if(NOT APPLE AND NOT WIN32) if(NOT APPLE AND NOT WIN32)
set(EXTRA_FLAGS "${EXTRA_FLAGS} -pthread") set(EXTRA_FLAGS "${EXTRA_FLAGS} -pthread")
endif() endif()
# load hwloc library if not told otherwise
if(NOT CAF_NO_HWLOC)
find_package(Hwloc)
if(Hwloc_FOUND)
set(LD_FLAGS ${LD_FLAGS} ${Hwloc_LIBRARIES})
else()
set(CAF_NO_HWLOC yes)
endif()
endif()
# -fPIC generates warnings on MinGW and Cygwin plus extra setup steps needed on MinGW # -fPIC generates warnings on MinGW and Cygwin plus extra setup steps needed on MinGW
if(MINGW) if(MINGW)
add_definitions(-D_WIN32_WINNT=0x0600) add_definitions(-D_WIN32_WINNT=0x0600)
...@@ -345,6 +354,7 @@ macro(to_int_value name) ...@@ -345,6 +354,7 @@ macro(to_int_value name)
endif() endif()
endmacro() endmacro()
to_int_value(CAF_NO_HWLOC)
to_int_value(CAF_LOG_LEVEL) to_int_value(CAF_LOG_LEVEL)
to_int_value(CAF_NO_EXCEPTIONS) to_int_value(CAF_NO_EXCEPTIONS)
to_int_value(CAF_NO_MEM_MANAGEMENT) to_int_value(CAF_NO_MEM_MANAGEMENT)
...@@ -496,6 +506,7 @@ endmacro() ...@@ -496,6 +506,7 @@ endmacro()
# build core and I/O library # build core and I/O library
add_caf_lib(core) add_caf_lib(core)
set(CAF_CORE_LIBRARY libcaf_core_shared)
add_optional_caf_lib(io) add_optional_caf_lib(io)
# build opencl library if not told otherwise and OpenCL package was found # build opencl library if not told otherwise and OpenCL package was found
...@@ -509,16 +520,6 @@ if(NOT CAF_NO_OPENCL) ...@@ -509,16 +520,6 @@ if(NOT CAF_NO_OPENCL)
endif() endif()
endif() endif()
# load hwloc library if not told otherwise
if(NOT CAF_NO_HWLOC)
find_package(Hwloc)
if(Hwloc_FOUND)
add_optional_caf_lib(hwloc)
else()
set(CAF_NO_HWLOC yes)
endif()
endif()
# build Python binding if not being told otherwise # build Python binding if not being told otherwise
if(NOT CAF_NO_PYTHON AND EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/libcaf_python/CMakeLists.txt") if(NOT CAF_NO_PYTHON AND EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/libcaf_python/CMakeLists.txt")
add_subdirectory(libcaf_python) add_subdirectory(libcaf_python)
......
...@@ -42,7 +42,7 @@ ...@@ -42,7 +42,7 @@
#define CAF_NO_EXCEPTIONS #define CAF_NO_EXCEPTIONS
#endif #endif
#if @CAF_NO_HWLOC@ != -1 #if @CAF_NO_HWLOC_INT@ != -1
#define CAF_NO_HWLOC #define CAF_NO_HWLOC
#endif #endif
......
...@@ -25,6 +25,7 @@ endmacro() ...@@ -25,6 +25,7 @@ endmacro()
# introductionary applications # introductionary applications
add(. aout) add(. aout)
add(. hello_world) add(. hello_world)
add(. fibonacci)
# basic message passing primitives # basic message passing primitives
add(message_passing cell) add(message_passing cell)
......
...@@ -33,13 +33,12 @@ void hello_world(event_based_actor* self, const actor& buddy) { ...@@ -33,13 +33,12 @@ void hello_world(event_based_actor* self, const actor& buddy) {
); );
} }
int main() { void caf_main(actor_system& system) {
// our CAF environment
actor_system_config cfg;
actor_system system{cfg};
// create a new actor that calls 'mirror()' // create a new actor that calls 'mirror()'
auto mirror_actor = system.spawn(mirror); auto mirror_actor = system.spawn(mirror);
// create another actor that calls 'hello_world(mirror_actor)'; // create another actor that calls 'hello_world(mirror_actor)';
system.spawn(hello_world, mirror_actor); system.spawn(hello_world, mirror_actor);
// system will wait until both actors are destroyed before leaving main // system will wait until both actors are destroyed before leaving main
} }
CAF_MAIN()
...@@ -75,6 +75,7 @@ set (LIBCAF_CORE_SRCS ...@@ -75,6 +75,7 @@ set (LIBCAF_CORE_SRCS
src/message_view.cpp src/message_view.cpp
src/monitorable_actor.cpp src/monitorable_actor.cpp
src/node_id.cpp src/node_id.cpp
src/numa_aware_work_stealing.cpp
src/outgoing_stream_multiplexer.cpp src/outgoing_stream_multiplexer.cpp
src/parse_ini.cpp src/parse_ini.cpp
src/private_thread.cpp src/private_thread.cpp
......
This diff is collapsed.
...@@ -31,15 +31,28 @@ namespace policy { ...@@ -31,15 +31,28 @@ namespace policy {
class scheduler_policy { class scheduler_policy {
public: public:
/// Policy-specific data fields for the coordinator. /// Policy-specific data fields for the coordinator.
template <class Worker>
struct coordinator_data { struct coordinator_data {
explicit coordinator_data(scheduler::abstract_coordinator*); explicit coordinator_data(scheduler::abstract_coordinator*);
/// Stores the pointer of all workes handeld by the scheduler
std::vector<std::unique_ptr<Worker>> workers;
}; };
/// Policy-specific data fields for the worker. /// Policy-specific data fields for the worker.
template <class Worker>
struct worker_data { struct worker_data {
explicit worker_data(scheduler::abstract_coordinator*); explicit worker_data(scheduler::abstract_coordinator*);
}; };
/// Create x workers.
template <class Coordinator, class Worker>
void create_workers(Coordinator* self, size_t num_workers,
size_t throughput);
/// Initalize worker thread.
template <class Worker>
void init_worker_thread(Worker* self);
/// Enqueues a new job to coordinator. /// Enqueues a new job to coordinator.
template <class Coordinator> template <class Coordinator>
void central_enqueue(Coordinator* self, resumable* job); void central_enqueue(Coordinator* self, resumable* job);
......
...@@ -40,22 +40,41 @@ public: ...@@ -40,22 +40,41 @@ public:
~work_sharing() override; ~work_sharing() override;
template <class Worker>
struct coordinator_data { struct coordinator_data {
inline explicit coordinator_data(scheduler::abstract_coordinator*) { explicit coordinator_data(scheduler::abstract_coordinator*) {
// nop // nop
} }
queue_type queue; queue_type queue;
std::mutex lock; std::mutex lock;
std::condition_variable cv; std::condition_variable cv;
std::vector<std::unique_ptr<Worker>> workers;
}; };
template <class Worker>
struct worker_data { struct worker_data {
inline explicit worker_data(scheduler::abstract_coordinator*) { explicit worker_data(scheduler::abstract_coordinator*) {
// nop // nop
} }
}; };
// Create x workers.
template <class Coordinator, class Worker>
void create_workers(Coordinator* self, size_t num_workers,
size_t throughput) {
for (size_t i = 0; i < num_workers; ++i) {
std::unique_ptr<Worker> worker(new Worker(i, self, throughput));
d(self).workers.emplace_back(std::move(worker));
}
}
/// Initalize worker thread.
template <class Worker>
void init_worker_thread(Worker*) {
// nop
}
template <class Coordinator> template <class Coordinator>
void enqueue(Coordinator* self, resumable* job) { void enqueue(Coordinator* self, resumable* job) {
queue_type l; queue_type l;
......
...@@ -56,32 +56,42 @@ public: ...@@ -56,32 +56,42 @@ public:
}; };
// 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.
template <class Worker>
struct coordinator_data { struct coordinator_data {
inline explicit coordinator_data(scheduler::abstract_coordinator*) explicit coordinator_data(scheduler::abstract_coordinator*)
: next_worker(0) { : next_worker(0) {
// nop // nop
} }
std::atomic<size_t> next_worker; std::atomic<size_t> next_worker;
std::vector<std::unique_ptr<Worker>> workers;
}; };
static std::vector<poll_strategy>
get_poll_strategies(scheduler::abstract_coordinator* p) {
auto& x = p->system().config();
std::vector<poll_strategy> res {
{x.work_stealing_aggressive_poll_attempts, 1,
x.work_stealing_aggressive_steal_interval,
usec{0}},
{x.work_stealing_moderate_poll_attempts, 1,
x.work_stealing_moderate_steal_interval,
usec{x.work_stealing_moderate_sleep_duration_us}},
{1, 0, x.work_stealing_relaxed_steal_interval,
usec {p->system().config().work_stealing_relaxed_sleep_duration_us}}
};
return res;
}
// Holds job job queue of a worker and a random number generator. // Holds job job queue of a worker and a random number generator.
template <class Worker>
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{}()), : 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{ , strategies(get_poll_strategies(p)) {
{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
} }
...@@ -91,9 +101,25 @@ public: ...@@ -91,9 +101,25 @@ 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::vector<poll_strategy> strategies;
}; };
// Create x workers.
template <class Coordinator, class Worker>
void create_workers(Coordinator* self, size_t num_workers,
size_t throughput) {
for (size_t i = 0; i < num_workers; ++i) {
std::unique_ptr<Worker> worker(new Worker(i, self, throughput));
d(self).workers.emplace_back(std::move(worker));
}
}
/// Initalize worker thread.
template <class Worker>
void init_worker_thread(Worker*) {
// nop
}
// Goes on a raid in quest for a shiny new job. // Goes on a raid in quest for a shiny new job.
template <class Worker> template <class Worker>
resumable* try_steal(Worker* self) { resumable* try_steal(Worker* self) {
...@@ -104,15 +130,16 @@ public: ...@@ -104,15 +130,16 @@ public:
} }
// roll the dice to pick a victim other than ourselves // roll the dice to pick a victim other than ourselves
auto victim = d(self).uniform(d(self).rengine); auto victim = d(self).uniform(d(self).rengine);
// in this case the worker id is similar to the worker index
if (victim == self->id()) if (victim == self->id())
victim = p->num_workers() - 1; victim = p->num_workers() - 1;
// steal oldest element from the victim's queue // steal oldest element from the victim's queue
return d(p->worker_by_id(victim)).queue.take_tail(); return d(p->worker_by_idx(victim)).queue.take_tail();
} }
template <class Coordinator> template <class Coordinator>
void central_enqueue(Coordinator* self, resumable* job) { void central_enqueue(Coordinator* self, resumable* job) {
auto w = self->worker_by_id(d(self).next_worker++ % self->num_workers()); auto w = self->worker_by_idx(d(self).next_worker++ % self->num_workers());
w->external_enqueue(job); w->external_enqueue(job);
} }
......
...@@ -39,16 +39,17 @@ class coordinator : public abstract_coordinator { ...@@ -39,16 +39,17 @@ class coordinator : public abstract_coordinator {
public: public:
using super = abstract_coordinator; using super = abstract_coordinator;
using policy_data = typename Policy::coordinator_data; using worker_type = worker<Policy>;
using policy_data = typename Policy::template coordinator_data<worker_type>;
coordinator(actor_system& sys) : super(sys), data_(this) { coordinator(actor_system& sys) : super(sys), data_(this) {
// nop // nop
} }
using worker_type = worker<Policy>; /// return a worker by its vector index
/// (this is not necessarily the id of the worker)
worker_type* worker_by_id(size_t x) { worker_type* worker_by_idx(size_t x) {
return workers_[x].get(); return data_.workers[x].get();
} }
policy_data& data() { policy_data& data() {
...@@ -58,12 +59,9 @@ public: ...@@ -58,12 +59,9 @@ public:
protected: protected:
void start() override { void start() override {
// initialize workers vector // initialize workers vector
auto num = num_workers(); policy_.template create_workers<coordinator<Policy>, worker_type>(this, num_workers(), max_throughput_);
workers_.reserve(num);
for (size_t i = 0; i < num; ++i)
workers_.emplace_back(new worker_type(i, this, max_throughput_));
// start all workers now that all workers have been initialized // start all workers now that all workers have been initialized
for (auto& w : workers_) for (auto& w : data_.workers)
w->start(); w->start();
// run remaining startup code // run remaining startup code
super::start(); super::start();
...@@ -99,7 +97,7 @@ protected: ...@@ -99,7 +97,7 @@ protected:
std::set<worker_type*> alive_workers; std::set<worker_type*> alive_workers;
auto num = num_workers(); auto num = num_workers();
for (size_t i = 0; i < num; ++i) { for (size_t i = 0; i < num; ++i) {
alive_workers.insert(worker_by_id(i)); alive_workers.insert(worker_by_idx(i));
sh.ref(); // make sure reference count is high enough sh.ref(); // make sure reference count is high enough
} }
while (!alive_workers.empty()) { while (!alive_workers.empty()) {
...@@ -116,12 +114,12 @@ protected: ...@@ -116,12 +114,12 @@ protected:
// shutdown utility actors // shutdown utility actors
stop_actors(); stop_actors();
// wait until all workers are done // wait until all workers are done
for (auto& w : workers_) { for (auto& w : data_.workers) {
w->get_thread().join(); w->get_thread().join();
} }
// run cleanup code for each resumable // run cleanup code for each resumable
auto f = &abstract_coordinator::cleanup_and_release; auto f = &abstract_coordinator::cleanup_and_release;
for (auto& w : workers_) for (auto& w : data_.workers)
policy_.foreach_resumable(w.get(), f); policy_.foreach_resumable(w.get(), f);
policy_.foreach_central_resumable(this, f); policy_.foreach_central_resumable(this, f);
} }
...@@ -131,8 +129,6 @@ protected: ...@@ -131,8 +129,6 @@ protected:
} }
private: private:
// usually of size std::thread::hardware_concurrency()
std::vector<std::unique_ptr<worker_type>> workers_;
// policy-specific data // policy-specific data
policy_data data_; policy_data data_;
// instance of our policy object // instance of our policy object
......
...@@ -40,7 +40,7 @@ class worker : public execution_unit { ...@@ -40,7 +40,7 @@ class worker : public execution_unit {
public: public:
using job_ptr = resumable*; using job_ptr = resumable*;
using coordinator_ptr = coordinator<Policy>*; using coordinator_ptr = coordinator<Policy>*;
using policy_data = typename Policy::worker_data; using policy_data = typename Policy::template worker_data<worker<Policy>>;
worker(size_t worker_id, coordinator_ptr worker_parent, size_t throughput) worker(size_t worker_id, coordinator_ptr worker_parent, size_t throughput)
: execution_unit(&worker_parent->system()), : execution_unit(&worker_parent->system()),
...@@ -106,6 +106,7 @@ public: ...@@ -106,6 +106,7 @@ public:
private: private:
void run() { void run() {
CAF_SET_LOGGER_SYS(&system()); CAF_SET_LOGGER_SYS(&system());
policy_.init_worker_thread(this);
// scheduling loop // scheduling loop
for (;;) { for (;;) {
auto job = policy_.dequeue(this); auto job = policy_.dequeue(this);
......
...@@ -29,6 +29,7 @@ ...@@ -29,6 +29,7 @@
#include "caf/policy/work_sharing.hpp" #include "caf/policy/work_sharing.hpp"
#include "caf/policy/work_stealing.hpp" #include "caf/policy/work_stealing.hpp"
#include "caf/policy/numa_aware_work_stealing.hpp"
#include "caf/scheduler/coordinator.hpp" #include "caf/scheduler/coordinator.hpp"
#include "caf/scheduler/test_coordinator.hpp" #include "caf/scheduler/test_coordinator.hpp"
...@@ -223,6 +224,7 @@ actor_system::actor_system(actor_system_config& cfg) ...@@ -223,6 +224,7 @@ actor_system::actor_system(actor_system_config& cfg)
using test = scheduler::test_coordinator; using test = scheduler::test_coordinator;
using share = scheduler::coordinator<policy::work_sharing>; using share = scheduler::coordinator<policy::work_sharing>;
using steal = scheduler::coordinator<policy::work_stealing>; using steal = scheduler::coordinator<policy::work_stealing>;
using numa_steal = scheduler::coordinator<policy::numa_aware_work_stealing>;
using profiled_share = scheduler::profiled_coordinator<policy::work_sharing>; using profiled_share = scheduler::profiled_coordinator<policy::work_sharing>;
using profiled_steal = scheduler::profiled_coordinator<policy::work_stealing>; using profiled_steal = scheduler::profiled_coordinator<policy::work_stealing>;
// set scheduler only if not explicitly loaded by user // set scheduler only if not explicitly loaded by user
...@@ -231,29 +233,37 @@ actor_system::actor_system(actor_system_config& cfg) ...@@ -231,29 +233,37 @@ actor_system::actor_system(actor_system_config& cfg)
stealing = 0x0001, stealing = 0x0001,
sharing = 0x0002, sharing = 0x0002,
testing = 0x0003, testing = 0x0003,
numa_stealing = 0x0004,
profiled = 0x0100, profiled = 0x0100,
profiled_stealing = 0x0101, profiled_stealing = 0x0101,
profiled_sharing = 0x0102 profiled_sharing = 0x0102
}; };
sched_conf sc = stealing; sched_conf sc = numa_stealing;
if (cfg.scheduler_policy == atom("sharing"))
if (cfg.scheduler_policy == atom("stealing"))
sc = stealing;
else if (cfg.scheduler_policy == atom("sharing"))
sc = sharing; sc = sharing;
else if (cfg.scheduler_policy == atom("testing")) else if (cfg.scheduler_policy == atom("testing"))
sc = testing; sc = testing;
else if (cfg.scheduler_policy != atom("stealing")) else if (cfg.scheduler_policy != atom("numa-steal"))
std::cerr << "[WARNING] " << deep_to_string(cfg.scheduler_policy) std::cerr
<< " is an unrecognized scheduler pollicy, " << "[WARNING] " << deep_to_string(cfg.scheduler_policy)
"falling back to 'stealing' (i.e. work-stealing)" << " is an unrecognized scheduler pollicy, "
<< std::endl; "falling back to 'numa-steal' (i.e. numa aware work-stealing)"
<< std::endl;
if (cfg.scheduler_enable_profiling) if (cfg.scheduler_enable_profiling)
sc = static_cast<sched_conf>(sc | profiled); sc = static_cast<sched_conf>(sc | profiled);
switch (sc) { switch (sc) {
default: // any invalid configuration falls back to work stealing default: // any invalid configuration falls back to numa work stealing
sched.reset(new steal(*this)); sched.reset(new numa_steal(*this));
break; break;
case sharing: case sharing:
sched.reset(new share(*this)); sched.reset(new share(*this));
break; break;
case stealing:
sched.reset(new steal(*this));
break;
case profiled_stealing: case profiled_stealing:
sched.reset(new profiled_steal(*this)); sched.reset(new profiled_steal(*this));
break; break;
......
...@@ -101,7 +101,7 @@ actor_system_config::actor_system_config() ...@@ -101,7 +101,7 @@ actor_system_config::actor_system_config()
slave_mode(false), slave_mode(false),
slave_mode_fun(nullptr) { slave_mode_fun(nullptr) {
// (1) hard-coded defaults // (1) hard-coded defaults
scheduler_policy = atom("stealing"); scheduler_policy = atom("numa-steal");
scheduler_max_threads = std::max(std::thread::hardware_concurrency(), scheduler_max_threads = std::max(std::thread::hardware_concurrency(),
unsigned{4}); unsigned{4});
scheduler_max_throughput = std::numeric_limits<size_t>::max(); scheduler_max_throughput = std::numeric_limits<size_t>::max();
...@@ -328,7 +328,7 @@ actor_system_config& actor_system_config::parse(message& args, ...@@ -328,7 +328,7 @@ actor_system_config& actor_system_config::parse(message& args,
atom("asio") atom("asio")
# endif # endif
}, middleman_network_backend, "middleman.network-backend"); }, middleman_network_backend, "middleman.network-backend");
verify_atom_opt({atom("stealing"), atom("sharing")}, verify_atom_opt({atom("stealing"), atom("sharing"), atom("numa-steal")},
scheduler_policy, "scheduler.policy "); scheduler_policy, "scheduler.policy ");
if (res.opts.count("caf#dump-config") != 0u) { if (res.opts.count("caf#dump-config") != 0u) {
cli_helptext_printed = true; cli_helptext_printed = true;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/policy/numa_aware_work_stealing.hpp"
namespace caf {
namespace policy {
numa_aware_work_stealing::~numa_aware_work_stealing() {
// nop
}
std::ostream& operator<<(std::ostream& s, const numa_aware_work_stealing::hwloc_bitmap_wrapper& w) {
char* tmp = nullptr;
hwloc_bitmap_asprintf(&tmp, w.get());
s << std::string(tmp);
free(tmp);
return s;
}
} // namespace policy
} // 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