/usr/include/caf/scheduler/worker.hpp is in libcaf-dev 0.13.2-3.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 | /******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_SCHEDULER_WORKER_HPP
#define CAF_SCHEDULER_WORKER_HPP
#include <cstddef>
#include "caf/execution_unit.hpp"
#include "caf/detail/logging.hpp"
#include "caf/detail/double_ended_queue.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:
using job_ptr = resumable*;
using coordinator_ptr = coordinator<Policy>*;
using policy_data = typename Policy::worker_data;
worker(size_t worker_id, coordinator_ptr worker_parent, size_t throughput)
: m_max_throughput(throughput),
m_id(worker_id),
m_parent(worker_parent) {
// nop
}
void start() {
CAF_ASSERT(m_this_thread.get_id() == std::thread::id{});
auto this_worker = this;
m_this_thread = std::thread{[this_worker] {
CAF_LOGF_TRACE("id = " << this_worker->id());
this_worker->run();
}};
}
worker(const worker&) = delete;
worker& operator=(const worker&) = delete;
/**
* 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_ASSERT(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_ASSERT(job != nullptr);
CAF_LOG_TRACE("id = " << id() << " actor id " << id_of(job));
m_policy.internal_enqueue(this, job);
}
coordinator_ptr 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();
});
}
actor_id id_of(resumable* ptr) {
abstract_actor* dptr = ptr ? dynamic_cast<abstract_actor*>(ptr)
: nullptr;
return dptr ? dptr->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_ASSERT(job != nullptr);
CAF_LOG_DEBUG("resume actor " << id_of(job));
CAF_PUSH_AID_FROM_PTR(dynamic_cast<abstract_actor*>(job));
m_policy.before_resume(this, job);
switch (job->resume(this, m_max_throughput)) {
case resumable::resume_later: {
m_policy.after_resume(this, job);
m_policy.resume_job_later(this, job);
break;
}
case resumable::done: {
m_policy.after_resume(this, job);
m_policy.after_completion(this, job);
job->detach_from_scheduler();
break;
}
case resumable::awaiting_message: {
// resumable will be enqueued again later
m_policy.after_resume(this, job);
break;
}
case resumable::shutdown_execution_unit: {
m_policy.after_resume(this, job);
m_policy.after_completion(this, job);
m_policy.before_shutdown(this);
return;
}
}
}
}
// 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_ptr 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
|