Commit d2ba1d41 authored by Dominik Charousset's avatar Dominik Charousset

Make worker non-copyable and non-movable

Use workers via `unique_ptr` instead of using them as values. This gets rid of
the weird worker semantics regarding construction and assignment.
parent 61068dc7
......@@ -22,6 +22,7 @@
#include <thread>
#include <limits>
#include <memory>
#include <condition_variable>
#include "caf/scheduler/worker.hpp"
......@@ -50,7 +51,7 @@ class coordinator : public abstract_coordinator {
using worker_type = worker<Policy>;
worker_type* worker_by_id(size_t id) {//override {
return &m_workers[id];
return m_workers[id].get();
}
policy_data& data() {
......@@ -60,15 +61,19 @@ class coordinator : public abstract_coordinator {
protected:
void initialize() override {
super::initialize();
// create & start workers
// create workers
m_workers.resize(num_workers());
for (size_t i = 0; i < num_workers(); ++i) {
m_workers[i].start(i, this, m_max_throughput);
auto& ref = m_workers[i];
ref.reset(new worker_type(i, this, m_max_throughput));
}
// start all workers now that all workers have been initialized
for (auto& w : m_workers) {
w->start();
}
}
void stop() override {
// perform cleanup code of base classe
CAF_LOG_TRACE("");
// shutdown workers
class shutdown_helper : public resumable {
......@@ -94,38 +99,37 @@ class coordinator : public abstract_coordinator {
std::condition_variable cv;
execution_unit* last_worker;
};
// use a set to keep track of remaining workers
shutdown_helper sh;
std::vector<worker_type*> alive_workers;
std::set<worker_type*> alive_workers;
auto num = num_workers();
for (size_t i = 0; i < num; ++i) {
alive_workers.push_back(worker_by_id(i));
alive_workers.insert(worker_by_id(i));
}
CAF_LOG_DEBUG("enqueue shutdown_helper into each worker");
while (!alive_workers.empty()) {
alive_workers.back()->external_enqueue(&sh);
(*alive_workers.begin())->external_enqueue(&sh);
// since jobs can be stolen, we cannot assume that we have
// actually shut down the worker we've enqueued sh to
{ // lifetime scope of guard
std::unique_lock<std::mutex> guard(sh.mtx);
sh.cv.wait(guard, [&] { return sh.last_worker != nullptr; });
}
auto last = alive_workers.end();
auto i = std::find(alive_workers.begin(), last, sh.last_worker);
alive_workers.erase(static_cast<worker_type*>(sh.last_worker));
sh.last_worker = nullptr;
alive_workers.erase(i);
}
// shutdown utility actors
stop_actors();
// wait until all workers are done
for (auto& w : m_workers) {
w.get_thread().join();
w->get_thread().join();
}
// run cleanup code for each resumable
auto f = [](resumable* job) {
job->detach_from_scheduler();
};
for (auto& w : m_workers) {
m_policy.foreach_resumable(&w, f);
m_policy.foreach_resumable(w.get(), f);
}
m_policy.foreach_central_resumable(this, f);
}
......@@ -136,7 +140,7 @@ class coordinator : public abstract_coordinator {
private:
// usually of size std::thread::hardware_concurrency()
std::vector<worker_type> m_workers;
std::vector<std::unique_ptr<worker_type>> m_workers;
// policy-specific data
policy_data m_data;
// instance of our policy object
......
......@@ -39,35 +39,28 @@ class coordinator;
template <class Policy>
class worker : public execution_unit {
public:
worker(const worker&) = delete;
worker& operator=(const worker&) = delete;
worker() = default;
using job_ptr = resumable*;
using coordinator_ptr = coordinator<Policy>*;
using policy_data = typename Policy::worker_data;
worker(worker&& other) {
*this = std::move(other); // delegate to move assignment operator
worker(size_t id, coordinator_ptr parent, size_t max_throughput) {
m_max_throughput = max_throughput;
m_id = id;
m_parent = parent;
}
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_policy = std::move(other.m_policy);
m_policy = std::move(other.m_policy);
return *this;
void start() {
CAF_REQUIRE(m_this_thread.get_id() == std::thread::id{});
auto this_worker = this;
m_this_thread = std::thread{[this_worker] {
CAF_LOGC_TRACE("caf::scheduler::worker", "start$lambda",
"id = " << this_worker->id());
this_worker->run();
}};
}
using coordinator_type = coordinator<Policy>;
using job_ptr = resumable*;
using job_queue = detail::double_ended_queue<resumable>;
using policy_data = typename Policy::worker_data;
worker(const worker&) = delete;
worker& operator=(const worker&) = delete;
/**
* Enqueues a new job to the worker's queue from an external
......@@ -97,7 +90,7 @@ class worker : public execution_unit {
return result;
}
coordinator_type* parent() {
coordinator_ptr parent() {
return m_parent;
}
......@@ -116,22 +109,10 @@ class worker : public execution_unit {
});
}
void start(size_t id, coordinator_type* parent, size_t max_throughput) {
m_max_throughput = max_throughput;
m_id = id;
m_parent = parent;
auto this_worker = this;
m_this_thread = std::thread{[this_worker] {
CAF_LOGC_TRACE("caf::scheduler::worker", "start$lambda",
"id = " << this_worker->id());
this_worker->run();
}};
}
actor_id id_of(resumable* ptr) {
abstract_actor* aptr = ptr ? dynamic_cast<abstract_actor*>(ptr)
abstract_actor* dptr = ptr ? dynamic_cast<abstract_actor*>(ptr)
: nullptr;
return aptr ? aptr->id() : 0;
return dptr ? dptr->id() : 0;
}
policy_data& data() {
......@@ -179,7 +160,7 @@ class worker : public execution_unit {
// the worker's ID received from scheduler
size_t m_id;
// pointer to central coordinator
coordinator_type* m_parent;
coordinator_ptr m_parent;
// policy-specific data
policy_data m_data;
// instance of our policy object
......
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