Commit 71b20538 authored by Dominik Charousset's avatar Dominik Charousset

Properly split scheduler implementation

parent 863f2977
......@@ -11,6 +11,7 @@ file(GLOB LIBCAF_CORE_HDRS "caf/*.hpp"
"caf/detail/*.hpp"
"caf/policy/*.hpp"
"caf/mixin/*.hpp"
"caf/scheduler/*.hpp"
"cppa/*.hpp")
# list cpp files excluding platform-dependent files
......@@ -18,6 +19,7 @@ set (LIBCAF_CORE_SRCS
src/abstract_actor.cpp
src/abstract_channel.cpp
src/abstract_group.cpp
src/abstract_coordinator.cpp
src/actor.cpp
src/actor_addr.cpp
src/actor_companion.cpp
......@@ -62,8 +64,8 @@ set (LIBCAF_CORE_SRCS
src/response_promise.cpp
src/resumable.cpp
src/ripemd_160.cpp
src/scheduler.cpp
src/scoped_actor.cpp
src/set_scheduler.cpp
src/serializer.cpp
src/shared_spinlock.cpp
src/shutdown.cpp
......
......@@ -23,7 +23,9 @@
#include <atomic>
#include "caf/message.hpp"
#include "caf/scheduler.hpp"
#include "caf/execution_unit.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/detail/singletons.hpp"
#include "caf/detail/single_reader_queue.hpp"
......
......@@ -29,7 +29,6 @@
#include "caf/config.hpp"
#include "caf/extend.hpp"
#include "caf/behavior.hpp"
#include "caf/scheduler.hpp"
#include "caf/policy/resume_policy.hpp"
......
......@@ -199,10 +199,12 @@ class work_stealing {
template <class Worker>
void after_resume(Worker* self) {
// give others the opportunity to steal from us
/*
if (d(self).private_queue.size() > 1 && d(self).exposed_queue.empty()) {
d(self).exposed_queue.push_back(d(self).private_queue.front());
d(self).private_queue.pop_front();
}
*/
}
template <class Worker, class UnaryFunction>
......
This diff is collapsed.
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* 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 LICENCE_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_SCHEDULER_ABSTRACT_COORDINATOR_HPP
#define CAF_SCHEDULER_ABSTRACT_COORDINATOR_HPP
#include <chrono>
#include <atomic>
#include <cstddef>
#include "caf/fwd.hpp"
#include "caf/atom.hpp"
#include "caf/actor.hpp"
#include "caf/channel.hpp"
#include "caf/message.hpp"
#include "caf/duration.hpp"
#include "caf/actor_addr.hpp"
namespace caf {
namespace scheduler {
/**
* A coordinator creates the workers, manages delayed sends and
* the central printer instance for {@link aout}. It also forwards
* sends from detached workers or non-actor threads to randomly
* chosen workers.
*/
class abstract_coordinator {
public:
friend class detail::singletons;
explicit abstract_coordinator(size_t num_worker_threads);
virtual ~abstract_coordinator();
/**
* Returns a handle to the central printing actor.
*/
inline actor printer() const {
return m_printer;
}
/**
* Puts `what` into the queue of a randomly chosen worker.
*/
virtual void enqueue(resumable* what) = 0;
template <class Duration, class... Data>
void delayed_send(Duration rel_time, actor_addr from, channel to,
message_id mid, message data) {
m_timer->enqueue(invalid_actor_addr, message_id::invalid,
make_message(atom("_Send"), duration{rel_time},
std::move(from), std::move(to), mid,
std::move(data)),
nullptr);
}
inline size_t num_workers() const {
return m_num_workers;
}
protected:
abstract_coordinator();
virtual void initialize();
virtual void stop() = 0;
void stop_actors();
// Creates a default instance.
static abstract_coordinator* create_singleton();
inline void dispose() {
delete this;
}
actor m_timer;
actor m_printer;
// ID of the worker receiving the next enqueue
std::atomic<size_t> m_next_worker;
size_t m_num_workers;
};
} // namespace scheduler
} // namespace caf
#endif // CAF_SCHEDULER_ABSTRACT_COORDINATOR_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* 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 LICENCE_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_SCHEDULER_COORDINATOR_HPP
#define CAF_SCHEDULER_COORDINATOR_HPP
#include <thread>
#include <limits>
#include "caf/scheduler/worker.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
namespace caf {
namespace scheduler {
/**
* Policy-based implementation of the abstract coordinator base class.
*/
template <class Policy>
class coordinator : public abstract_coordinator {
public:
using super = abstract_coordinator;
using policy_data = typename Policy::coordinator_data;
coordinator(size_t nw = std::thread::hardware_concurrency(),
size_t mt = std::numeric_limits<size_t>::max())
: super(nw),
m_max_throughput(mt) {
// nop
}
using worker_type = worker<Policy>;
worker_type* worker_by_id(size_t id) {//override {
return &m_workers[id];
}
policy_data& data() {
return m_data;
}
protected:
void initialize() override {
super::initialize();
// create & start workers
m_workers.resize(num_workers());
for (size_t i = 0; i < num_workers(); ++i) {
m_workers[i].start(i, this, m_max_throughput);
}
}
void stop() override {
// perform cleanup code of base classe
CAF_LOG_TRACE("");
// shutdown workers
class shutdown_helper : public resumable {
public:
void attach_to_scheduler() override {
// nop
}
void detach_from_scheduler() override {
// nop
}
resumable::resume_result resume(execution_unit* ptr, size_t) override {
CAF_LOG_DEBUG("shutdown_helper::resume => shutdown worker");
CAF_REQUIRE(ptr != nullptr);
std::unique_lock<std::mutex> guard(mtx);
last_worker = ptr;
cv.notify_all();
return resumable::shutdown_execution_unit;
}
shutdown_helper() : last_worker(nullptr) {
// nop
}
std::mutex mtx;
std::condition_variable cv;
execution_unit* last_worker;
};
shutdown_helper sh;
std::vector<worker_type*> alive_workers;
auto num = num_workers();
for (size_t i = 0; i < num; ++i) {
alive_workers.push_back(worker_by_id(i));
}
CAF_LOG_DEBUG("enqueue shutdown_helper into each worker");
while (!alive_workers.empty()) {
alive_workers.back()->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);
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();
}
// 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_central_resumable(this, f);
}
void enqueue(resumable* ptr) {
m_policy.central_enqueue(this, ptr);
}
private:
// usually of size std::thread::hardware_concurrency()
std::vector<worker_type> m_workers;
// policy-specific data
policy_data m_data;
// instance of our policy object
Policy m_policy;
// number of messages each actor is allowed to consume per resume
size_t m_max_throughput;
};
} // namespace scheduler
} // namespace caf
#endif // CAF_SCHEDULER_COORDINATOR_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* 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 LICENCE_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_SCHEDULER_WORKER_HPP
#define CAF_SCHEDULER_WORKER_HPP
#include <cstddef>
#include "caf/execution_unit.hpp"
#include "caf/detail/logging.hpp"
#include "caf/detail/producer_consumer_list.hpp"
namespace caf {
namespace scheduler {
template <class Policy>
class coordinator;
/**
* Policy-based implementation of the abstract worker base class.
*/
template <class Policy>
class worker : public execution_unit {
public:
worker(const worker&) = delete;
worker& operator=(const worker&) = delete;
worker() = default;
worker(worker&& other) {
*this = std::move(other); // delegate to move assignment operator
}
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;
}
using coordinator_type = coordinator<Policy>;
using job_ptr = resumable*;
using job_queue = detail::producer_consumer_list<resumable>;
using policy_data = typename Policy::worker_data;
/**
* Enqueues a new job to the worker's queue from an external
* source, i.e., from any other thread.
*/
void external_enqueue(job_ptr job) {
CAF_REQUIRE(job != nullptr);
CAF_LOG_TRACE("id = " << id() << " actor id " << id_of(job));
m_policy.external_enqueue(this, job);
}
/**
* 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 job) override {
CAF_REQUIRE(job != nullptr);
CAF_LOG_TRACE("id = " << id() << " actor id " << id_of(job));
m_policy.internal_enqueue(this, job);
}
// go on a raid in quest for a shiny new job
job_ptr raid() {
auto result = m_policy.raid(this);
CAF_LOG_DEBUG_IF(result, "got actor with id " << id_of(result));
return result;
}
coordinator_type* parent() {
return m_parent;
}
size_t id() const {
return m_id;
}
std::thread& get_thread() {
return m_this_thread;
}
void detach_all() {
CAF_LOG_TRACE("");
m_policy.foreach_resumable(this, [](resumable* job) {
job->detach_from_scheduler();
});
}
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)
: nullptr;
return aptr ? aptr->id() : 0;
}
policy_data& data() {
return m_data;
}
size_t max_throughput() {
return m_max_throughput;
}
private:
void run() {
CAF_LOG_TRACE("worker with ID " << m_id);
// scheduling loop
for (;;) {
auto job = m_policy.dequeue(this);
CAF_REQUIRE(job != nullptr);
CAF_LOG_DEBUG("resume actor " << id_of(job));
CAF_PUSH_AID_FROM_PTR(dynamic_cast<abstract_actor*>(job));
switch (job->resume(this, m_max_throughput)) {
case resumable::resume_later: {
m_policy.resume_job_later(this, job);
break;
}
case resumable::done: {
job->detach_from_scheduler();
break;
}
case resumable::awaiting_message: {
// resumable will be enqueued again later
break;
}
case resumable::shutdown_execution_unit: {
m_policy.before_shutdown(this);
return;
}
}
m_policy.after_resume(this);
}
}
// number of messages each actor is allowed to consume per resume
size_t m_max_throughput;
// the worker's thread
std::thread m_this_thread;
// the worker's ID received from scheduler
size_t m_id;
// pointer to central coordinator
coordinator_type* m_parent;
// policy-specific data
policy_data m_data;
// instance of our policy object
Policy m_policy;
};
} // namespace scheduler
} // namespace caf
#endif // CAF_SCHEDULER_WORKER_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* 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 LICENCE_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_SET_SCHEDULER_HPP
#define CAF_SET_SCHEDULER_HPP
#include <thread>
#include <limits>
#include "caf/policy/work_stealing.hpp"
#include "caf/scheduler/worker.hpp"
#include "caf/scheduler/coordinator.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
namespace caf {
/**
* Sets a user-defined scheduler.
* @note This function must be used before actor is spawned. Dynamically
* changing the scheduler at runtime is not supported.
* @throws std::logic_error if a scheduler is already defined
*/
void set_scheduler(scheduler::abstract_coordinator* ptr);
/**
* Sets a user-defined scheduler using given policies. The scheduler
* is instantiated with `nw` number of workers and allows each actor
* to consume up to `max_throughput` per resume (must be > 0).
* @note This function must be used before actor is spawned. Dynamically
* changing the scheduler at runtime is not supported.
* @throws std::logic_error if a scheduler is already defined
* @throws std::invalid_argument if `max_throughput == 0`
*/
template <class Policy = policy::work_stealing>
void set_scheduler(size_t nw = std::thread::hardware_concurrency(),
size_t max_throughput = std::numeric_limits<size_t>::max()) {
if (max_throughput == 0) {
throw std::invalid_argument("max_throughput must not be 0");
}
set_scheduler(new scheduler::coordinator<Policy>(nw, max_throughput));
}
} // namespace caf
#endif // CAF_SET_SCHEDULER_HPP
......@@ -22,7 +22,6 @@
#include <type_traits>
#include "caf/scheduler.hpp"
#include "caf/spawn_fwd.hpp"
#include "caf/typed_actor.hpp"
#include "caf/spawn_options.hpp"
......
......@@ -17,6 +17,8 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/scheduler/abstract_coordinator.hpp"
#include <thread>
#include <atomic>
#include <chrono>
......@@ -25,18 +27,16 @@
#include "caf/on.hpp"
#include "caf/send.hpp"
#include "caf/spawn.hpp"
#include "caf/anything.hpp"
#include "caf/to_string.hpp"
#include "caf/scheduler.hpp"
#include "caf/local_actor.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/system_messages.hpp"
#include "caf/policy/no_resume.hpp"
#include "caf/policy/no_scheduling.hpp"
#include "caf/policy/actor_policies.hpp"
#include "caf/policy/nestable_invoke.hpp"
#include "caf/policy/not_prioritizing.hpp"
#include "caf/scheduler/coordinator.hpp"
#include "caf/policy/work_stealing.hpp"
#include "caf/detail/logging.hpp"
#include "caf/detail/proper_actor.hpp"
......@@ -79,7 +79,8 @@ inline void insert_dmsg(Map& storage, const duration& d, Ts&&... vs) {
}
class timer_actor final : public detail::proper_actor<blocking_actor,
timer_actor_policies> {
timer_actor_policies>,
public spawn_as_is {
public:
inline unique_mailbox_element_pointer dequeue() {
await_data();
......@@ -212,31 +213,31 @@ abstract_coordinator* abstract_coordinator::create_singleton() {
}
void abstract_coordinator::initialize() {
// launch threads of utility actors
auto ptr = m_timer.get();
m_timer_thread = std::thread([ptr] { ptr->act(); });
m_printer_thread = std::thread{printer_loop, m_printer.get()};
// launch utility actors
m_timer = spawn<timer_actor, hidden + detached + blocking_api>();
m_printer = spawn<hidden + detached + blocking_api>(printer_loop);
}
void abstract_coordinator::stop_actors() {
CAF_LOG_TRACE("");
scoped_actor self(true);
self->monitor(m_timer);
self->monitor(m_printer);
self->send_exit(m_timer, exit_reason::user_shutdown);
self->send_exit(m_printer, exit_reason::user_shutdown);
int i = 0;
self->receive_for(i, 2)(
[](const down_msg&) {
// nop
}
);
}
abstract_coordinator::abstract_coordinator(size_t nw)
: m_timer(new timer_actor),
m_printer(true),
m_next_worker(0),
: m_next_worker(0),
m_num_workers(nw) {
// nop
}
actor abstract_coordinator::printer() const {
return m_printer.get();
}
} // namespace scheduler
void set_scheduler(scheduler::abstract_coordinator* impl) {
if (!detail::singletons::set_scheduling_coordinator(impl)) {
delete impl;
throw std::logic_error("scheduler already defined");
}
}
} // namespace caf
......@@ -17,11 +17,13 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/actor_ostream.hpp"
#include "caf/all.hpp"
#include "caf/scheduler.hpp"
#include "caf/local_actor.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/actor_ostream.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/detail/singletons.hpp"
......
......@@ -24,7 +24,6 @@
#include "caf/atom.hpp"
#include "caf/to_string.hpp"
#include "caf/message.hpp"
#include "caf/scheduler.hpp"
#include "caf/actor_proxy.hpp"
#include "caf/exit_reason.hpp"
......
......@@ -18,7 +18,6 @@
******************************************************************************/
#include "caf/exception.hpp"
#include "caf/scheduler.hpp"
#include "caf/blocking_actor.hpp"
#include "caf/detail/logging.hpp"
......
......@@ -17,7 +17,6 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/scheduler.hpp"
#include "caf/message_id.hpp"
#include "caf/event_based_actor.hpp"
......
......@@ -20,7 +20,6 @@
#include <string>
#include "caf/all.hpp"
#include "caf/atom.hpp"
#include "caf/scheduler.hpp"
#include "caf/actor_cast.hpp"
#include "caf/local_actor.hpp"
#include "caf/default_attachable.hpp"
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* 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 LICENCE_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 <stdexcept>
#include "caf/set_scheduler.hpp"
#include "caf/detail/singletons.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
namespace caf {
void set_scheduler(scheduler::abstract_coordinator* impl) {
if (!detail::singletons::set_scheduling_coordinator(impl)) {
delete impl;
throw std::logic_error("scheduler already defined");
}
}
} // namespace caf
......@@ -20,10 +20,11 @@
#include <atomic>
#include "caf/message.hpp"
#include "caf/scheduler.hpp"
#include "caf/exception.hpp"
#include "caf/local_actor.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/detail/logging.hpp"
#include "caf/detail/singletons.hpp"
#include "caf/detail/group_manager.hpp"
......
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