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
|
#pragma once
#include "base/assert.hpp"
#include "base/macros.hpp"
#include "base/thread_pool_delayed.hpp"
#include <algorithm>
#include <atomic>
#include <condition_variable>
#include <memory>
#include <mutex>
#include <unordered_set>
#include <utility>
#include <vector>
namespace dp
{
// This class MUST NOT run OpenGL-related tasks (which invoke OpenGL or contain any
// OpenGL data), use FR/BR threads for that.
class DrapeRoutine
{
public:
class Result
{
public:
void Wait()
{
if (m_isFinished)
return;
DrapeRoutine::Instance().Wait(m_id);
}
private:
friend class DrapeRoutine;
explicit Result(uint64_t id) : m_id(id), m_isFinished(false) {}
uint64_t Finish()
{
m_isFinished = true;
return m_id;
}
uint64_t const m_id;
std::atomic<bool> m_isFinished;
};
using ResultPtr = std::shared_ptr<Result>;
static void Init()
{
Instance();
}
static void Shutdown()
{
Instance().FinishAll();
}
template <typename Task>
static ResultPtr Run(Task && t)
{
ResultPtr result(new Result(Instance().GetNextId()));
bool const success = Instance().m_workerThread.Push([result, t = std::move(t)]() mutable
{
t();
Instance().Notify(result->Finish());
});
if (!success)
return {};
return result;
}
template <typename Task>
static ResultPtr RunDelayed(base::thread_pool::delayed::ThreadPool::Duration const & duration, Task && t)
{
ResultPtr result(new Result(Instance().GetNextId()));
bool const success = Instance().m_workerThread.PushDelayed(duration, [result, t = std::move(t)]() mutable
{
t();
Instance().Notify(result->Finish());
});
if (!success)
return {};
return result;
}
// Asynchronous execution for tasks when execution order matters.
template <typename Task>
static ResultPtr RunSequential(Task && t)
{
ResultPtr result(new Result(Instance().GetNextId()));
bool const success = Instance().m_sequentialWorkerThread.Push([result, t = std::move(t)]() mutable
{
t();
Instance().Notify(result->Finish());
});
if (!success)
return {};
return result;
}
private:
static DrapeRoutine & Instance()
{
static DrapeRoutine instance;
return instance;
}
DrapeRoutine() : m_workerThread(4 /* threads count */) {}
uint64_t GetNextId()
{
std::lock_guard<std::mutex> lock(m_mutex);
return m_counter++;
}
void Notify(uint64_t id)
{
std::lock_guard<std::mutex> lock(m_mutex);
m_finishedIds.insert(id);
m_condition.notify_all();
}
void Wait(uint64_t id)
{
std::unique_lock<std::mutex> lock(m_mutex);
if (m_finished)
return;
m_condition.wait(lock, [this, id]()
{
return m_finished || m_finishedIds.find(id) != m_finishedIds.end();
});
m_finishedIds.erase(id);
}
void FinishAll()
{
m_workerThread.ShutdownAndJoin();
m_sequentialWorkerThread.ShutdownAndJoin();
std::lock_guard<std::mutex> lock(m_mutex);
m_finished = true;
m_condition.notify_all();
}
std::unordered_set<uint64_t> m_finishedIds;
uint64_t m_counter = 0;
bool m_finished = false;
std::condition_variable m_condition;
std::mutex m_mutex;
base::thread_pool::delayed::ThreadPool m_workerThread;
base::thread_pool::delayed::ThreadPool m_sequentialWorkerThread;
};
// This is a helper class, which aggregates logic of waiting for active
// tasks completion. It must be used when we provide tasks completion
// before subsystem shutting down.
template <typename TaskType>
class ActiveTasks
{
struct ActiveTask
{
std::shared_ptr<TaskType> m_task;
DrapeRoutine::ResultPtr m_result;
ActiveTask(std::shared_ptr<TaskType> const & task,
DrapeRoutine::ResultPtr const & result)
: m_task(task)
, m_result(result)
{}
};
public:
~ActiveTasks()
{
FinishAll();
}
void Add(std::shared_ptr<TaskType> const & task,
DrapeRoutine::ResultPtr const & result)
{
ASSERT(task != nullptr, ());
ASSERT(result != nullptr, ());
std::lock_guard<std::mutex> lock(m_mutex);
m_tasks.emplace_back(task, result);
}
void Remove(std::shared_ptr<TaskType> const & task)
{
std::lock_guard<std::mutex> lock(m_mutex);
m_tasks.erase(std::remove_if(m_tasks.begin(), m_tasks.end(),
[task](ActiveTask const & t) { return t.m_task == task; }),
m_tasks.end());
}
void FinishAll()
{
// Move tasks to a temporary vector, because m_tasks
// can be modified during 'Cancel' calls.
std::vector<ActiveTask> tasks;
{
std::lock_guard<std::mutex> lock(m_mutex);
tasks.swap(m_tasks);
}
// Cancel all tasks.
for (auto & t : tasks)
t.m_task->Cancel();
// Wait for completion of unfinished tasks.
for (auto & t : tasks)
t.m_result->Wait();
}
private:
std::vector<ActiveTask> m_tasks;
std::mutex m_mutex;
};
} // namespace dp
|
