Commit 6a65f5d0 authored by Dominik Charousset's avatar Dominik Charousset

streamlined init & shutdown of scheduler

parent 76eceb2e
...@@ -47,7 +47,8 @@ class resumable { ...@@ -47,7 +47,8 @@ class resumable {
enum resume_result { enum resume_result {
resume_later, resume_later,
done done,
shutdown_execution_unit
}; };
resumable(); resumable();
......
...@@ -60,7 +60,92 @@ namespace detail { class singleton_manager; } ...@@ -60,7 +60,92 @@ namespace detail { class singleton_manager; }
namespace scheduler { namespace scheduler {
class worker; class coordinator;
/**
* @brief A work-stealing scheduling worker.
*
* The work-stealing implementation of libcppa minimizes access to the
* synchronized queue. The reasoning behind this design decision is that
* it has been shown that stealing actually is very rare for workloads [1].
* Hence, implementations should focus on the performance in
* the non-stealing case. For this reason, each worker has an exposed
* job queue that can be accessed by the central scheduler instance as
* well as other workers, but it also has a private job list it is
* currently working on. To account for the load balancing aspect, each
* worker makes sure that at least one job is left in its exposed queue
* to allow other workers to steal it.
*
* [1] http://dl.acm.org/citation.cfm?doid=2398857.2384639
*/
class worker : public execution_unit {
friend class coordinator;
public:
worker() = default;
worker(worker&&);
worker& operator=(worker&&);
worker(const worker&) = delete;
worker& operator=(const worker&) = delete;
typedef resumable* job_ptr;
typedef util::producer_consumer_list<resumable> job_queue;
/**
* @brief Attempt to steal an element from the exposed job queue.
*/
job_ptr try_steal();
/**
* @brief Enqueues a new job to the worker's queue from an external
* source, i.e., from any other thread.
*/
void external_enqueue(job_ptr);
/**
* @brief Enqueues a new job to the worker's queue from an internal
* source, i.e., a job that is currently executed by
* this worker.
* @warning Must not be called from other threads.
*/
void exec_later(job_ptr) override;
private:
void start(size_t id, coordinator* parent); // called from coordinator
void run(); // work loop
job_ptr raid(); // go on a raid in quest for a shiny new job
// this queue is exposed to others, i.e., other workers
// may attempt to steal jobs from it and the central scheduling
// unit can push new jobs to the queue
job_queue m_exposed_queue;
// internal job stack
std::vector<job_ptr> m_job_list;
// the worker's thread
std::thread m_this_thread;
// the worker's ID received from scheduler
size_t m_id;
// the ID of the last victim we stole from
size_t m_last_victim;
coordinator* m_parent;
};
/** /**
* @brief Central scheduling interface. * @brief Central scheduling interface.
...@@ -110,8 +195,8 @@ class coordinator { ...@@ -110,8 +195,8 @@ class coordinator {
return static_cast<unsigned>(m_workers.size()); return static_cast<unsigned>(m_workers.size());
} }
inline worker* worker_by_id(size_t id) const { inline worker& worker_by_id(size_t id) {
return m_workers[id].get(); return m_workers[id];
} }
private: private:
...@@ -136,87 +221,7 @@ class coordinator { ...@@ -136,87 +221,7 @@ class coordinator {
std::atomic<size_t> m_next_worker; std::atomic<size_t> m_next_worker;
// vector of size std::thread::hardware_concurrency() // vector of size std::thread::hardware_concurrency()
std::vector<std::unique_ptr<worker>> m_workers; std::vector<worker> m_workers;
};
/**
* @brief A work-stealing scheduling worker.
*
* The work-stealing implementation of libcppa minimizes access to the
* synchronized queue. The reasoning behind this design decision is that
* it has been shown that stealing actually is very rare for workloads [1].
* Hence, implementations should focus on the performance in
* the non-stealing case. For this reason, each worker has an exposed
* job queue that can be accessed by the central scheduler instance as
* well as other workers, but it also has a private job list it is
* currently working on. To account for the load balancing aspect, each
* worker makes sure that at least one job is left in its exposed queue
* to allow other workers to steal it.
*
* [1] http://dl.acm.org/citation.cfm?doid=2398857.2384639
*/
class worker : public execution_unit {
friend class coordinator;
friend class coordinator::shutdown_helper;
public:
typedef resumable* job_ptr;
typedef util::producer_consumer_list<resumable> job_queue;
worker(size_t id, coordinator* parent);
/**
* @brief Attempt to steal an element from the exposed job queue.
*/
job_ptr try_steal();
/**
* @brief Enqueues a new job to the worker's queue from an external
* source, i.e., from any other thread.
*/
void external_enqueue(job_ptr);
/**
* @brief Enqueues a new job to the worker's queue from an internal
* source, i.e., a job that is currently executed by
* this worker.
* @warning Must not be called from other threads.
*/
void exec_later(job_ptr) override;
private:
void start(); // called from the scheduler
void run(); // work loop
job_ptr raid(); // go on a raid in quest for a new shiny job
bool m_running;
// this queue is exposed to others, i.e., other workers
// may attempt to steal jobs from it and the central scheduling
// unit can push new jobs to the queue
job_queue m_exposed_queue;
// internal job stack
std::vector<job_ptr> m_job_list;
// the worker's thread
std::thread m_this_thread;
// the worker's ID received from scheduler
size_t m_id;
// the ID of the last victim we stole from
size_t m_last_victim;
coordinator* m_parent;
}; };
......
...@@ -240,14 +240,12 @@ class coordinator::shutdown_helper : public resumable { ...@@ -240,14 +240,12 @@ class coordinator::shutdown_helper : public resumable {
void detach_from_scheduler() override { } void detach_from_scheduler() override { }
resumable::resume_result resume(detail::cs_thread*, execution_unit* ptr) { resumable::resume_result resume(detail::cs_thread*, execution_unit* ptr) {
auto w = dynamic_cast<worker*>(ptr);
CPPA_REQUIRE(w != nullptr);
CPPA_LOG_DEBUG("shutdown_helper::resume => shutdown worker"); CPPA_LOG_DEBUG("shutdown_helper::resume => shutdown worker");
w->m_running = false; last_worker = dynamic_cast<worker*>(ptr);
CPPA_REQUIRE(last_worker != nullptr);
std::unique_lock<std::mutex> guard(mtx); std::unique_lock<std::mutex> guard(mtx);
last_worker = w;
cv.notify_all(); cv.notify_all();
return resumable::resume_later; return resumable::shutdown_execution_unit;
} }
shutdown_helper() : last_worker(nullptr) { } shutdown_helper() : last_worker(nullptr) { }
...@@ -265,20 +263,21 @@ void coordinator::initialize() { ...@@ -265,20 +263,21 @@ void coordinator::initialize() {
ptr->act(); ptr->act();
}}; }};
m_printer_thread = std::thread{printer_loop, m_printer.get()}; m_printer_thread = std::thread{printer_loop, m_printer.get()};
// create workers // create & start workers
size_t hc = std::thread::hardware_concurrency(); auto hwc = static_cast<size_t>(std::thread::hardware_concurrency());
for (size_t i = 0; i < hc; ++i) { m_workers.resize(hwc);
m_workers.emplace_back(new worker(i, this)); for (size_t i = 0; i < hwc; ++i) {
m_workers[i].start(i, this);
} }
// start all workers
for (auto& w : m_workers) w->start();
} }
void coordinator::destroy() { void coordinator::destroy() {
CPPA_LOG_TRACE("");
// shutdown workers // shutdown workers
shutdown_helper sh; shutdown_helper sh;
std::vector<worker*> alive_workers; std::vector<worker*> alive_workers;
for (auto& w : m_workers) alive_workers.push_back(w.get()); for (auto& w : m_workers) alive_workers.push_back(&w);
CPPA_LOG_DEBUG("enqueue shutdown_helper into each worker");
while (!alive_workers.empty()) { while (!alive_workers.empty()) {
alive_workers.back()->external_enqueue(&sh); alive_workers.back()->external_enqueue(&sh);
// since jobs can be stolen, we cannot assume that we have // since jobs can be stolen, we cannot assume that we have
...@@ -294,13 +293,23 @@ void coordinator::destroy() { ...@@ -294,13 +293,23 @@ void coordinator::destroy() {
alive_workers.erase(i); alive_workers.erase(i);
} }
// shutdown utility actors // shutdown utility actors
CPPA_LOG_DEBUG("send 'DIE' messages to timer & printer");
auto msg = make_any_tuple(atom("DIE")); auto msg = make_any_tuple(atom("DIE"));
m_timer->enqueue({invalid_actor_addr, nullptr}, msg, nullptr); m_timer->enqueue({invalid_actor_addr, nullptr}, msg, nullptr);
m_printer->enqueue({invalid_actor_addr, nullptr}, msg, nullptr); m_printer->enqueue({invalid_actor_addr, nullptr}, msg, nullptr);
CPPA_LOG_DEBUG("join threads of utility actors");
m_timer_thread.join(); m_timer_thread.join();
m_printer_thread.join(); m_printer_thread.join();
// join each worker thread for good manners // join each worker thread for good manners
for (auto& w : m_workers) w->m_this_thread.join(); CPPA_LOG_DEBUG("join threads of workers");
for (auto& w : m_workers) w.m_this_thread.join();
CPPA_LOG_DEBUG("detach all resumables from all workers");
for (auto& w : m_workers) {
auto next = [&] { return w.m_exposed_queue.try_pop(); };
for (auto job = next(); job != nullptr; job = next()) {
job->detach_from_scheduler();
}
}
// cleanup // cleanup
delete this; delete this;
} }
...@@ -321,7 +330,7 @@ actor coordinator::printer() const { ...@@ -321,7 +330,7 @@ actor coordinator::printer() const {
void coordinator::enqueue(resumable* what) { void coordinator::enqueue(resumable* what) {
size_t nw = m_next_worker++; size_t nw = m_next_worker++;
m_workers[nw % m_workers.size()]->external_enqueue(what); m_workers[nw % m_workers.size()].external_enqueue(what);
} }
/****************************************************************************** /******************************************************************************
...@@ -331,11 +340,30 @@ void coordinator::enqueue(resumable* what) { ...@@ -331,11 +340,30 @@ void coordinator::enqueue(resumable* what) {
#define CPPA_LOG_DEBUG_WORKER(msg) \ #define CPPA_LOG_DEBUG_WORKER(msg) \
CPPA_LOG_DEBUG("worker " << m_id << ": " << msg) CPPA_LOG_DEBUG("worker " << m_id << ": " << msg)
worker::worker(size_t id, coordinator* parent) worker::worker(worker&& other) {
: m_running(true), m_id(id), m_last_victim(id), m_parent(parent) { } *this = std::move(other); // delegate to move assignment operator
}
worker& worker::operator=(worker&& other) {
// cannot be moved once m_this_thread is up and running
auto running = [](std::thread& t) {
return t.get_id() != std::thread::id{};
};
if (running(m_this_thread) || running(other.m_this_thread)) {
throw std::runtime_error("running workers cannot be moved");
}
m_job_list = std::move(other.m_job_list);
auto next = [&] { return other.m_exposed_queue.try_pop(); };
for (auto j = next(); j != nullptr; j = next()) {
m_exposed_queue.push_back(j);
}
return *this;
}
void worker::start() { void worker::start(size_t id, coordinator* parent) {
m_id = id;
m_last_victim = id;
m_parent = parent;
auto this_worker = this; auto this_worker = this;
m_this_thread = std::thread{[this_worker] { m_this_thread = std::thread{[this_worker] {
this_worker->run(); this_worker->run();
...@@ -412,12 +440,26 @@ void worker::run() { ...@@ -412,12 +440,26 @@ void worker::run() {
} }
}; };
// scheduling loop // scheduling loop
while (m_running) { for (;;) {
local_poll() || aggressive_poll() || moderate_poll() || relaxed_poll(); local_poll() || aggressive_poll() || moderate_poll() || relaxed_poll();
CPPA_PUSH_AID_FROM_PTR(dynamic_cast<abstract_actor*>(job)); CPPA_PUSH_AID_FROM_PTR(dynamic_cast<abstract_actor*>(job));
if (job->resume(&fself, this) == resumable::done) { switch (job->resume(&fself, this)) {
// was attached in policy::cooperative_scheduling::launch case resumable::done: {
job->detach_from_scheduler(); job->detach_from_scheduler();
break;
}
case resumable::resume_later: {
break;
}
case resumable::shutdown_execution_unit: {
// give others the opportunity to steal unfinished jobs
for (auto ptr : m_job_list) m_exposed_queue.push_back(ptr);
m_job_list.clear();
return;
}
default: {
CPPA_CRITICAL("job->resume returned illegal result");
}
} }
job = nullptr; job = nullptr;
// give others the opportunity to steal from us // give others the opportunity to steal from us
...@@ -444,7 +486,7 @@ worker::job_ptr worker::raid() { ...@@ -444,7 +486,7 @@ worker::job_ptr worker::raid() {
for (size_t i = 0; i < n; ++i) { for (size_t i = 0; i < n; ++i) {
m_last_victim = next(m_last_victim) % n; m_last_victim = next(m_last_victim) % n;
if (m_last_victim != m_id) { if (m_last_victim != m_id) {
auto job = m_parent->worker_by_id(m_last_victim)->try_steal(); auto job = m_parent->worker_by_id(m_last_victim).try_steal();
if (job) { if (job) {
CPPA_LOG_DEBUG_WORKER("successfully stolen a job from " CPPA_LOG_DEBUG_WORKER("successfully stolen a job from "
<< m_last_victim); << m_last_victim);
......
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