Commit 08ba7997 authored by Dominik Charousset's avatar Dominik Charousset

Merge pull request #438 from Lingxi-Li/topic/cleanup_scheduler_code

Cleanup scheduling-related code
parents 90a9c015 c6160a26
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* 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. *
******************************************************************************/
#ifndef CAF_POLICY_UNPROFILED_HPP
#define CAF_POLICY_UNPROFILED_HPP
namespace caf {
namespace policy {
/// This class is intended to be used as a base class for actual polices.
/// It provides a default empty implementation for the customization points.
/// By deriving from it, actual policy classes only need to implement/override
/// the customization points they need. This class also serves as a place to
/// factor common utilities for implementing actual policies.
class unprofiled {
public:
virtual ~unprofiled() = default;
/// Policy-specific data fields for the coordinator.
struct coordinator_data { };
/// Policy-specific data fields for the worker.
struct worker_data { };
/// Performs cleanup action before a shutdown takes place.
template <class Worker>
void before_shutdown(Worker*) {
// nop
}
/// Called immediately before resuming an actor.
template <class Worker>
void before_resume(Worker*, resumable*) {
// nop
}
/// Called whenever an actor has been resumed. This function can
/// prepare some fields before the next resume operation takes place
/// or perform cleanup actions between to actor runs.
template <class Worker>
void after_resume(Worker*, resumable*) {
// nop
}
/// Called whenever an actor has completed a job.
template <class Worker>
void after_completion(Worker*, resumable*) {
// nop
}
protected:
// Convenience function to access the data field.
template <class WorkerOrCoordinator>
static auto d(WorkerOrCoordinator* self) -> decltype(self->data()) {
return self->data();
}
};
} // namespace policy
} // namespace caf
#endif // CAF_POLICY_UNPROFILED_HPP
...@@ -26,37 +26,23 @@ ...@@ -26,37 +26,23 @@
#include <condition_variable> #include <condition_variable>
#include "caf/resumable.hpp" #include "caf/resumable.hpp"
#include "caf/policy/unprofiled.hpp"
namespace caf { namespace caf {
namespace policy { namespace policy {
/// @extends scheduler_policy /// @extends scheduler_policy
class work_sharing { class work_sharing : public unprofiled {
public: public:
// A thead-safe queue implementation. // A thread-safe queue implementation.
using queue_type = std::list<resumable*>; using queue_type = std::list<resumable*>;
struct coordinator_data { struct coordinator_data {
queue_type queue; queue_type queue;
std::mutex lock; std::mutex lock;
std::condition_variable cv; std::condition_variable cv;
inline coordinator_data() {
// nop
}
}; };
struct worker_data {
inline worker_data() {
// nop
}
};
// Convenience function to access the data field.
template <class WorkerOrCoordinator>
static auto d(WorkerOrCoordinator* self) -> decltype(self->data()) {
return self->data();
}
template <class Coordinator> template <class Coordinator>
void enqueue(Coordinator* self, resumable* job) { void enqueue(Coordinator* self, resumable* job) {
queue_type l; queue_type l;
...@@ -78,12 +64,7 @@ public: ...@@ -78,12 +64,7 @@ public:
template <class Worker> template <class Worker>
void internal_enqueue(Worker* self, resumable* job) { void internal_enqueue(Worker* self, resumable* job) {
auto& parent_data = d(self->parent()); enqueue(self->parent(), job);
queue_type l;
l.push_back(job);
std::unique_lock<std::mutex> guard(parent_data.lock);
parent_data.queue.splice(parent_data.queue.begin(), l);
parent_data.cv.notify_one();
} }
template <class Worker> template <class Worker>
...@@ -97,54 +78,33 @@ public: ...@@ -97,54 +78,33 @@ public:
resumable* dequeue(Worker* self) { resumable* dequeue(Worker* self) {
auto& parent_data = d(self->parent()); auto& parent_data = d(self->parent());
std::unique_lock<std::mutex> guard(parent_data.lock); std::unique_lock<std::mutex> guard(parent_data.lock);
while (parent_data.queue.empty()) { parent_data.cv.wait(guard, [&] { return ! parent_data.queue.empty(); });
parent_data.cv.wait(guard);
}
resumable* job = parent_data.queue.front(); resumable* job = parent_data.queue.front();
parent_data.queue.pop_front(); parent_data.queue.pop_front();
return job; return job;
} }
template <class Worker> template <class Worker, class UnaryFunction>
void before_shutdown(Worker*) { void foreach_resumable(Worker*, UnaryFunction) {
// nop
}
template <class Worker>
void before_resume(Worker*, resumable*) {
// nop
}
template <class Worker>
void after_resume(Worker*, resumable*) {
// nop
}
template <class Worker>
void after_completion(Worker*, resumable*) {
// nop // nop
} }
template <class Worker, class UnaryFunction> template <class Coordinator, class UnaryFunction>
void foreach_resumable(Worker* self, UnaryFunction f) { void foreach_central_resumable(Coordinator* self, UnaryFunction f) {
auto& queue = d(self).queue;
auto next = [&]() -> resumable* { auto next = [&]() -> resumable* {
if (d(self->parent()).queue.empty()) { if (queue.empty()) {
return nullptr; return nullptr;
} }
auto front = d(self->parent()).queue.front(); auto front = queue.front();
d(self->parent()).queue.pop_front(); queue.pop_front();
return front; return front;
}; };
std::unique_lock<std::mutex> guard(d(self->parent()).lock); std::unique_lock<std::mutex> guard(d(self).lock);
for (auto job = next(); job != nullptr; job = next()) { for (auto job = next(); job != nullptr; job = next()) {
f(job); f(job);
} }
} }
template <class Coordinator, class UnaryFunction>
void foreach_central_resumable(Coordinator*, UnaryFunction) {
// nop
}
}; };
} // namespace policy } // namespace policy
......
...@@ -27,6 +27,7 @@ ...@@ -27,6 +27,7 @@
#include <cstddef> #include <cstddef>
#include "caf/resumable.hpp" #include "caf/resumable.hpp"
#include "caf/policy/unprofiled.hpp"
#include "caf/detail/double_ended_queue.hpp" #include "caf/detail/double_ended_queue.hpp"
...@@ -49,9 +50,9 @@ namespace policy { ...@@ -49,9 +50,9 @@ namespace policy {
/// [1] http://dl.acm.org/citation.cfm?doid=2398857.2384639 /// [1] http://dl.acm.org/citation.cfm?doid=2398857.2384639
/// ///
/// @extends scheduler_policy /// @extends scheduler_policy
class work_stealing { class work_stealing : public unprofiled {
public: public:
// A thead-safe queue implementation. // A thread-safe queue implementation.
using queue_type = detail::double_ended_queue<resumable>; using queue_type = detail::double_ended_queue<resumable>;
// 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.
...@@ -67,22 +68,14 @@ public: ...@@ -67,22 +68,14 @@ public:
// 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.
queue_type queue; queue_type queue;
// needed by our engine
std::random_device rdevice;
// needed to generate pseudo random numbers // needed to generate pseudo random numbers
std::default_random_engine rengine; std::default_random_engine rengine;
// initialize random engine // initialize random engine
inline worker_data() : rdevice(), rengine(rdevice()) { inline worker_data() : rengine(std::random_device{}()) {
// nop // nop
} }
}; };
// Convenience function to access the data field.
template <class WorkerOrCoordinator>
static auto d(WorkerOrCoordinator* self) -> decltype(self->data()) {
return self->data();
}
// 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) {
...@@ -167,26 +160,6 @@ public: ...@@ -167,26 +160,6 @@ public:
return nullptr; return nullptr;
} }
template <class Worker>
void before_shutdown(Worker*) {
// nop
}
template <class Worker>
void before_resume(Worker*, resumable*) {
// nop
}
template <class Worker>
void after_resume(Worker*, resumable*) {
// nop
}
template <class Worker>
void after_completion(Worker*, resumable*) {
// nop
}
template <class Worker, class UnaryFunction> template <class Worker, class UnaryFunction>
void foreach_resumable(Worker* self, UnaryFunction f) { void foreach_resumable(Worker* self, UnaryFunction f) {
auto next = [&] { return d(self).queue.take_head(); }; auto next = [&] { return d(self).queue.take_head(); };
......
...@@ -82,8 +82,6 @@ public: ...@@ -82,8 +82,6 @@ public:
void* subtype_ptr() override; void* subtype_ptr() override;
protected: protected:
abstract_coordinator();
void stop_actors(); void stop_actors();
// ID of the worker receiving the next enqueue // ID of the worker receiving the next enqueue
......
...@@ -59,7 +59,7 @@ protected: ...@@ -59,7 +59,7 @@ protected:
void start() override { void start() override {
// initialize workers vector // initialize workers vector
auto num = num_workers(); auto num = num_workers();
//workers_.resize(num); workers_.reserve(num);
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, max_throughput_));
// start all workers now that all workers have been initialized // start all workers now that all workers have been initialized
......
...@@ -91,13 +91,6 @@ public: ...@@ -91,13 +91,6 @@ public:
return this_thread_; return this_thread_;
} }
void detach_all() {
CAF_LOG_TRACE("");
policy_.foreach_resumable(this, [](resumable* job) {
intrusive_ptr_release(job);
});
}
actor_id id_of(resumable* ptr) { actor_id id_of(resumable* ptr) {
abstract_actor* dptr = ptr ? dynamic_cast<abstract_actor*>(ptr) abstract_actor* dptr = ptr ? dynamic_cast<abstract_actor*>(ptr)
: nullptr; : nullptr;
...@@ -124,25 +117,23 @@ private: ...@@ -124,25 +117,23 @@ private:
CAF_LOG_DEBUG("resume actor:" << CAF_ARG(id_of(job))); CAF_LOG_DEBUG("resume actor:" << CAF_ARG(id_of(job)));
CAF_PUSH_AID_FROM_PTR(dynamic_cast<abstract_actor*>(job)); CAF_PUSH_AID_FROM_PTR(dynamic_cast<abstract_actor*>(job));
policy_.before_resume(this, job); policy_.before_resume(this, job);
switch (job->resume(this, max_throughput_)) { auto res = job->resume(this, max_throughput_);
case resumable::resume_later: {
policy_.after_resume(this, job); policy_.after_resume(this, job);
switch (res) {
case resumable::resume_later: {
policy_.resume_job_later(this, job); policy_.resume_job_later(this, job);
break; break;
} }
case resumable::done: { case resumable::done: {
policy_.after_resume(this, job);
policy_.after_completion(this, job); policy_.after_completion(this, job);
intrusive_ptr_release(job); intrusive_ptr_release(job);
break; break;
} }
case resumable::awaiting_message: { case resumable::awaiting_message: {
// resumable will be enqueued again later // resumable will be enqueued again later
policy_.after_resume(this, job);
break; break;
} }
case resumable::shutdown_execution_unit: { case resumable::shutdown_execution_unit: {
policy_.after_resume(this, job);
policy_.after_completion(this, job); policy_.after_completion(this, job);
policy_.before_shutdown(this); policy_.before_shutdown(this);
return; return;
......
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