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
|
//===--------------------- TaskPool.h ---------------------------*- C++ -*-===//
//
// The LLVM Compiler Infrastructure
//
// This file is distributed under the University of Illinois Open Source
// License. See LICENSE.TXT for details.
//
//===----------------------------------------------------------------------===//
#ifndef utility_TaskPool_h_
#define utility_TaskPool_h_
#include <functional> // for bind, function
#include <future>
#include <list>
#include <memory> // for make_shared
#include <mutex> // for mutex, unique_lock, condition_variable
#include <type_traits> // for forward, result_of, move
// Global TaskPool class for running tasks in parallel on a set of worker thread
// created the first
// time the task pool is used. The TaskPool provide no guarantee about the order
// the task will be run
// and about what tasks will run in parallel. None of the task added to the task
// pool should block
// on something (mutex, future, condition variable) what will be set only by the
// completion of an
// other task on the task pool as they may run on the same thread sequentally.
class TaskPool {
public:
// Add a new task to the task pool and return a std::future belonging to the
// newly created task.
// The caller of this function has to wait on the future for this task to
// complete.
template <typename F, typename... Args>
static std::future<typename std::result_of<F(Args...)>::type>
AddTask(F &&f, Args &&... args);
// Run all of the specified tasks on the task pool and wait until all of them
// are finished
// before returning. This method is intended to be used for small number tasks
// where listing
// them as function arguments is acceptable. For running large number of tasks
// you should use
// AddTask for each task and then call wait() on each returned future.
template <typename... T> static void RunTasks(T &&... tasks);
private:
TaskPool() = delete;
template <typename... T> struct RunTaskImpl;
static void AddTaskImpl(std::function<void()> &&task_fn);
};
// Wrapper class around the global TaskPool implementation to make it possible
// to create a set of
// tasks and then wait for the tasks to be completed by the
// WaitForNextCompletedTask call. This
// class should be used when WaitForNextCompletedTask is needed because this
// class add no other
// extra functionality to the TaskPool class and it have a very minor
// performance overhead.
template <typename T> // The return type of the tasks what will be added to this
// task runner
class TaskRunner {
public:
// Add a task to the task runner what will also add the task to the global
// TaskPool. The
// function doesn't return the std::future for the task because it will be
// supplied by the
// WaitForNextCompletedTask after the task is completed.
template <typename F, typename... Args> void AddTask(F &&f, Args &&... args);
// Wait for the next task in this task runner to finish and then return the
// std::future what
// belongs to the finished task. If there is no task in this task runner
// (neither pending nor
// comleted) then this function will return an invalid future. Usually this
// function should be
// called in a loop processing the results of the tasks until it returns an
// invalid std::future
// what means that all task in this task runner is completed.
std::future<T> WaitForNextCompletedTask();
// Convenience method to wait for all task in this TaskRunner to finish. Do
// NOT use this class
// just because of this method. Use TaskPool instead and wait for each
// std::future returned by
// AddTask in a loop.
void WaitForAllTasks();
private:
std::list<std::future<T>> m_ready;
std::list<std::future<T>> m_pending;
std::mutex m_mutex;
std::condition_variable m_cv;
};
template <typename F, typename... Args>
std::future<typename std::result_of<F(Args...)>::type>
TaskPool::AddTask(F &&f, Args &&... args) {
auto task_sp = std::make_shared<
std::packaged_task<typename std::result_of<F(Args...)>::type()>>(
std::bind(std::forward<F>(f), std::forward<Args>(args)...));
AddTaskImpl([task_sp]() { (*task_sp)(); });
return task_sp->get_future();
}
template <typename... T> void TaskPool::RunTasks(T &&... tasks) {
RunTaskImpl<T...>::Run(std::forward<T>(tasks)...);
}
template <typename Head, typename... Tail>
struct TaskPool::RunTaskImpl<Head, Tail...> {
static void Run(Head &&h, Tail &&... t) {
auto f = AddTask(std::forward<Head>(h));
RunTaskImpl<Tail...>::Run(std::forward<Tail>(t)...);
f.wait();
}
};
template <> struct TaskPool::RunTaskImpl<> {
static void Run() {}
};
template <typename T>
template <typename F, typename... Args>
void TaskRunner<T>::AddTask(F &&f, Args &&... args) {
std::unique_lock<std::mutex> lock(m_mutex);
auto it = m_pending.emplace(m_pending.end());
*it = std::move(TaskPool::AddTask(
[this, it](F f, Args... args) {
T &&r = f(std::forward<Args>(args)...);
std::unique_lock<std::mutex> lock(this->m_mutex);
this->m_ready.splice(this->m_ready.end(), this->m_pending, it);
lock.unlock();
this->m_cv.notify_one();
return r;
},
std::forward<F>(f), std::forward<Args>(args)...));
}
template <>
template <typename F, typename... Args>
void TaskRunner<void>::AddTask(F &&f, Args &&... args) {
std::unique_lock<std::mutex> lock(m_mutex);
auto it = m_pending.emplace(m_pending.end());
*it = std::move(TaskPool::AddTask(
[this, it](F f, Args... args) {
f(std::forward<Args>(args)...);
std::unique_lock<std::mutex> lock(this->m_mutex);
this->m_ready.emplace_back(std::move(*it));
this->m_pending.erase(it);
lock.unlock();
this->m_cv.notify_one();
},
std::forward<F>(f), std::forward<Args>(args)...));
}
template <typename T> std::future<T> TaskRunner<T>::WaitForNextCompletedTask() {
std::unique_lock<std::mutex> lock(m_mutex);
if (m_ready.empty() && m_pending.empty())
return std::future<T>(); // No more tasks
if (m_ready.empty())
m_cv.wait(lock, [this]() { return !this->m_ready.empty(); });
std::future<T> res = std::move(m_ready.front());
m_ready.pop_front();
lock.unlock();
res.wait();
return std::move(res);
}
template <typename T> void TaskRunner<T>::WaitForAllTasks() {
while (WaitForNextCompletedTask().valid())
;
}
#endif // #ifndef utility_TaskPool_h_
|
