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
|
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#pragma once
#include <cstddef>
#include "caf/detail/double_ended_queue.hpp"
#include "caf/detail/set_thread_name.hpp"
#include "caf/execution_unit.hpp"
#include "caf/logger.hpp"
#include "caf/resumable.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,
const policy_data& init, size_t throughput)
: execution_unit(&worker_parent->system()),
max_throughput_(throughput),
id_(worker_id),
parent_(worker_parent),
data_(init) {
// nop
}
void start() {
CAF_ASSERT(this_thread_.get_id() == std::thread::id{});
auto this_worker = this;
this_thread_ = std::thread{[this_worker] {
CAF_SET_LOGGER_SYS(&this_worker->system());
detail::set_thread_name("caf.multiplexer");
this_worker->system().thread_started();
this_worker->run();
this_worker->system().thread_terminates();
}};
}
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);
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);
policy_.internal_enqueue(this, job);
}
coordinator_ptr parent() {
return parent_;
}
size_t id() const {
return id_;
}
std::thread& get_thread() {
return this_thread_;
}
actor_id id_of(resumable* ptr) {
abstract_actor* dptr = ptr != nullptr ? dynamic_cast<abstract_actor*>(ptr)
: nullptr;
return dptr != nullptr ? dptr->id() : 0;
}
policy_data& data() {
return data_;
}
size_t max_throughput() {
return max_throughput_;
}
private:
void run() {
CAF_SET_LOGGER_SYS(&system());
// scheduling loop
for (;;) {
auto job = policy_.dequeue(this);
CAF_ASSERT(job != nullptr);
CAF_ASSERT(job->subtype() != resumable::io_actor);
CAF_PUSH_AID_FROM_PTR(dynamic_cast<abstract_actor*>(job));
policy_.before_resume(this, job);
auto res = job->resume(this, max_throughput_);
policy_.after_resume(this, job);
switch (res) {
case resumable::resume_later: {
// keep reference to this actor, as it remains in the "loop"
policy_.resume_job_later(this, job);
break;
}
case resumable::done: {
policy_.after_completion(this, job);
intrusive_ptr_release(job);
break;
}
case resumable::awaiting_message: {
// resumable will maybe be enqueued again later, deref it for now
intrusive_ptr_release(job);
break;
}
case resumable::shutdown_execution_unit: {
policy_.after_completion(this, job);
policy_.before_shutdown(this);
return;
}
}
}
}
// number of messages each actor is allowed to consume per resume
size_t max_throughput_;
// the worker's thread
std::thread this_thread_;
// the worker's ID received from scheduler
size_t id_;
// pointer to central coordinator
coordinator_ptr parent_;
// policy-specific data
policy_data data_;
// instance of our policy object
Policy policy_;
};
} // namespace scheduler
} // namespace caf
|
