-
Notifications
You must be signed in to change notification settings - Fork 41
Expand file tree
/
Copy pathasyncrpcqueue.cpp
More file actions
235 lines (203 loc) · 6.23 KB
/
Copy pathasyncrpcqueue.cpp
File metadata and controls
235 lines (203 loc) · 6.23 KB
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
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
// Copyright (c) 2016 The Zcash developers
#include <thread>
// Copyright (c) 2018-2022 The Flux Developers
// Distributed under the MIT software license, see the accompanying
// file COPYING or https://www.opensource.org/licenses/mit-license.php.
#include "asyncrpcqueue.h"
static std::atomic<size_t> workerCounter(0);
/**
* Static method to return the shared/default queue.
*/
shared_ptr<AsyncRPCQueue> AsyncRPCQueue::sharedInstance() {
// Thread-safe in C+11 and gcc 4.3
static shared_ptr<AsyncRPCQueue> q = std::make_shared<AsyncRPCQueue>();
return q;
}
AsyncRPCQueue::AsyncRPCQueue() : closed_(false), finish_(false) {
}
AsyncRPCQueue::~AsyncRPCQueue() {
closeAndWait(); // join on all worker threads
}
/**
* A worker will execute this method on a new thread
*/
void AsyncRPCQueue::run(size_t workerId) {
while (true) {
AsyncRPCOperationId key;
std::shared_ptr<AsyncRPCOperation> operation;
{
std::unique_lock<std::mutex> guard(lock_);
while (operation_id_queue_.empty() && !isClosed() && !isFinishing()) {
this->condition_.wait(guard);
}
// Exit if the queue is empty and we are finishing up
if (isFinishing() && operation_id_queue_.empty()) {
break;
}
// Exit if the queue is closing.
if (isClosed()) {
while (!operation_id_queue_.empty()) {
operation_id_queue_.pop();
}
break;
}
// Get operation id
key = operation_id_queue_.front();
operation_id_queue_.pop();
// Search operation map
AsyncRPCOperationMap::const_iterator iter = operation_map_.find(key);
if (iter != operation_map_.end()) {
operation = iter->second;
}
}
if (!operation) {
// cannot find operation in map, may have been removed
} else if (operation->isCancelled()) {
// skip cancelled operation
} else {
operation->main();
}
}
}
/**
* Add shared_ptr to operation.
*
* To retain polymorphic behaviour, i.e. main() method of derived classes is invoked,
* caller should create the shared_ptr like this:
*
* std::shared_ptr<AsyncRPCOperation> ptr(new MyCustomAsyncRPCOperation(params));
*
* Don't use std::make_shared<AsyncRPCOperation>().
*/
void AsyncRPCQueue::addOperation(const std::shared_ptr<AsyncRPCOperation> &ptrOperation) {
std::lock_guard<std::mutex> guard(lock_);
// Don't add if queue is closed or finishing
if (isClosed() || isFinishing()) {
return;
}
AsyncRPCOperationId id = ptrOperation->getId();
operation_map_.emplace(id, ptrOperation);
operation_id_queue_.push(id);
this->condition_.notify_one();
}
/**
* Return the operation for a given operation id.
*/
std::shared_ptr<AsyncRPCOperation> AsyncRPCQueue::getOperationForId(AsyncRPCOperationId id) const {
std::shared_ptr<AsyncRPCOperation> ptr;
std::lock_guard<std::mutex> guard(lock_);
AsyncRPCOperationMap::const_iterator iter = operation_map_.find(id);
if (iter != operation_map_.end()) {
ptr = iter->second;
}
return ptr;
}
/**
* Return the operation for a given operation id and then remove the operation from internal storage.
*/
std::shared_ptr<AsyncRPCOperation> AsyncRPCQueue::popOperationForId(AsyncRPCOperationId id) {
std::shared_ptr<AsyncRPCOperation> ptr = getOperationForId(id);
if (ptr) {
std::lock_guard<std::mutex> guard(lock_);
// Note: if the id still exists in the operationIdQueue, when it gets processed by a worker
// there will no operation in the map to execute, so nothing will happen.
operation_map_.erase(id);
}
return ptr;
}
/**
* Return true if the queue is closed to new operations.
*/
bool AsyncRPCQueue::isClosed() const {
return closed_.load();
}
/**
* Close the queue and cancel all existing operations
*/
void AsyncRPCQueue::close() {
closed_.store(true);
cancelAllOperations();
}
/**
* Return true if the queue is finishing up
*/
bool AsyncRPCQueue::isFinishing() const {
return finish_.load();
}
/**
* Close the queue but finish existing operations. Do not accept new operations.
*/
void AsyncRPCQueue::finish() {
finish_.store(true);
}
/**
* Call cancel() on all operations
*/
void AsyncRPCQueue::cancelAllOperations() {
std::lock_guard<std::mutex> guard(lock_);
for (auto key : operation_map_) {
key.second->cancel();
}
this->condition_.notify_all();
}
/**
* Return the number of operations in the queue
*/
size_t AsyncRPCQueue::getOperationCount() const {
std::lock_guard<std::mutex> guard(lock_);
return operation_id_queue_.size();
}
/**
* Spawn a worker thread
*/
void AsyncRPCQueue::addWorker() {
std::lock_guard<std::mutex> guard(lock_);
workers_.emplace_back( std::thread(&AsyncRPCQueue::run, this, ++workerCounter) );
}
/**
* Return the number of worker threads spawned by the queue
*/
size_t AsyncRPCQueue::getNumberOfWorkers() const {
std::lock_guard<std::mutex> guard(lock_);
return workers_.size();
}
/**
* Return a list of all known operation ids found in internal storage.
*/
std::vector<AsyncRPCOperationId> AsyncRPCQueue::getAllOperationIds() const {
std::lock_guard<std::mutex> guard(lock_);
std::vector<AsyncRPCOperationId> v;
for(auto & entry: operation_map_) {
v.push_back(entry.first);
}
return v;
}
/**
* Calling thread will close and wait for worker threads to join.
*/
void AsyncRPCQueue::closeAndWait() {
close();
wait_for_worker_threads();
}
/**
* Block current thread until all workers have finished their tasks.
*/
void AsyncRPCQueue::finishAndWait() {
finish();
wait_for_worker_threads();
}
/**
* Block current thread until all operations are finished or the queue has closed.
*/
void AsyncRPCQueue::wait_for_worker_threads() {
// Notify any workers who are waiting, so they see the updated queue state
{
std::lock_guard<std::mutex> guard(lock_);
this->condition_.notify_all();
}
for (std::thread & t : this->workers_) {
if (t.joinable()) {
t.join();
}
}
}