LLVM 19.0.0git
ThreadPool.h
Go to the documentation of this file.
1//===-- llvm/Support/ThreadPool.h - A ThreadPool implementation -*- C++ -*-===//
2//
3// Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
4// See https://llvm.org/LICENSE.txt for license information.
5// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
6//
7//===----------------------------------------------------------------------===//
8//
9// This file defines a crude C++11 based thread pool.
10//
11//===----------------------------------------------------------------------===//
12
13#ifndef LLVM_SUPPORT_THREADPOOL_H
14#define LLVM_SUPPORT_THREADPOOL_H
15
16#include "llvm/ADT/DenseMap.h"
17#include "llvm/Config/llvm-config.h"
20#include "llvm/Support/thread.h"
21
22#include <future>
23
24#include <condition_variable>
25#include <deque>
26#include <functional>
27#include <memory>
28#include <mutex>
29#include <utility>
30
31namespace llvm {
32
33class ThreadPoolTaskGroup;
34
35/// A ThreadPool for asynchronous parallel execution on a defined number of
36/// threads.
37///
38/// The pool keeps a vector of threads alive, waiting on a condition variable
39/// for some work to become available.
40///
41/// It is possible to reuse one thread pool for different groups of tasks
42/// by grouping tasks using ThreadPoolTaskGroup. All tasks are processed using
43/// the same queue, but it is possible to wait only for a specific group of
44/// tasks to finish.
45///
46/// It is also possible for worker threads to submit new tasks and wait for
47/// them. Note that this may result in a deadlock in cases such as when a task
48/// (directly or indirectly) tries to wait for its own completion, or when all
49/// available threads are used up by tasks waiting for a task that has no thread
50/// left to run on (this includes waiting on the returned future). It should be
51/// generally safe to wait() for a group as long as groups do not form a cycle.
53public:
54 /// Construct a pool using the hardware strategy \p S for mapping hardware
55 /// execution resources (threads, cores, CPUs)
56 /// Defaults to using the maximum execution resources in the system, but
57 /// accounting for the affinity mask.
59
60 /// Blocking destructor: the pool will wait for all the threads to complete.
62
63 /// Asynchronous submission of a task to the pool. The returned future can be
64 /// used to wait for the task to finish and is *non-blocking* on destruction.
65 template <typename Function, typename... Args>
66 auto async(Function &&F, Args &&...ArgList) {
67 auto Task =
68 std::bind(std::forward<Function>(F), std::forward<Args>(ArgList)...);
69 return async(std::move(Task));
70 }
71
72 /// Overload, task will be in the given task group.
73 template <typename Function, typename... Args>
74 auto async(ThreadPoolTaskGroup &Group, Function &&F, Args &&...ArgList) {
75 auto Task =
76 std::bind(std::forward<Function>(F), std::forward<Args>(ArgList)...);
77 return async(Group, std::move(Task));
78 }
79
80 /// Asynchronous submission of a task to the pool. The returned future can be
81 /// used to wait for the task to finish and is *non-blocking* on destruction.
82 template <typename Func>
83 auto async(Func &&F) -> std::shared_future<decltype(F())> {
84 return asyncImpl(std::function<decltype(F())()>(std::forward<Func>(F)),
85 nullptr);
86 }
87
88 template <typename Func>
89 auto async(ThreadPoolTaskGroup &Group, Func &&F)
90 -> std::shared_future<decltype(F())> {
91 return asyncImpl(std::function<decltype(F())()>(std::forward<Func>(F)),
92 &Group);
93 }
94
95 /// Blocking wait for all the threads to complete and the queue to be empty.
96 /// It is an error to try to add new tasks while blocking on this call.
97 /// Calling wait() from a task would deadlock waiting for itself.
98 void wait();
99
100 /// Blocking wait for only all the threads in the given group to complete.
101 /// It is possible to wait even inside a task, but waiting (directly or
102 /// indirectly) on itself will deadlock. If called from a task running on a
103 /// worker thread, the call may process pending tasks while waiting in order
104 /// not to waste the thread.
105 void wait(ThreadPoolTaskGroup &Group);
106
107 // Returns the maximum number of worker threads in the pool, not the current
108 // number of threads!
109 unsigned getMaxConcurrency() const { return MaxThreadCount; }
110
111 // TODO: misleading legacy name warning!
112 LLVM_DEPRECATED("Use getMaxConcurrency instead", "getMaxConcurrency")
113 unsigned getThreadCount() const { return MaxThreadCount; }
114
115 /// Returns true if the current thread is a worker thread of this thread pool.
116 bool isWorkerThread() const;
117
118private:
119 /// Helpers to create a promise and a callable wrapper of \p Task that sets
120 /// the result of the promise. Returns the callable and a future to access the
121 /// result.
122 template <typename ResTy>
123 static std::pair<std::function<void()>, std::future<ResTy>>
124 createTaskAndFuture(std::function<ResTy()> Task) {
125 std::shared_ptr<std::promise<ResTy>> Promise =
126 std::make_shared<std::promise<ResTy>>();
127 auto F = Promise->get_future();
128 return {
129 [Promise = std::move(Promise), Task]() { Promise->set_value(Task()); },
130 std::move(F)};
131 }
132 static std::pair<std::function<void()>, std::future<void>>
133 createTaskAndFuture(std::function<void()> Task) {
134 std::shared_ptr<std::promise<void>> Promise =
135 std::make_shared<std::promise<void>>();
136 auto F = Promise->get_future();
137 return {[Promise = std::move(Promise), Task]() {
138 Task();
139 Promise->set_value();
140 },
141 std::move(F)};
142 }
143
144 /// Returns true if all tasks in the given group have finished (nullptr means
145 /// all tasks regardless of their group). QueueLock must be locked.
146 bool workCompletedUnlocked(ThreadPoolTaskGroup *Group) const;
147
148 /// Asynchronous submission of a task to the pool. The returned future can be
149 /// used to wait for the task to finish and is *non-blocking* on destruction.
150 template <typename ResTy>
151 std::shared_future<ResTy> asyncImpl(std::function<ResTy()> Task,
152 ThreadPoolTaskGroup *Group) {
153
154#if LLVM_ENABLE_THREADS
155 /// Wrap the Task in a std::function<void()> that sets the result of the
156 /// corresponding future.
157 auto R = createTaskAndFuture(Task);
158
159 int requestedThreads;
160 {
161 // Lock the queue and push the new task
162 std::unique_lock<std::mutex> LockGuard(QueueLock);
163
164 // Don't allow enqueueing after disabling the pool
165 assert(EnableFlag && "Queuing a thread during ThreadPool destruction");
166 Tasks.emplace_back(std::make_pair(std::move(R.first), Group));
167 requestedThreads = ActiveThreads + Tasks.size();
168 }
169 QueueCondition.notify_one();
170 grow(requestedThreads);
171 return R.second.share();
172
173#else // LLVM_ENABLE_THREADS Disabled
174
175 // Get a Future with launch::deferred execution using std::async
176 auto Future = std::async(std::launch::deferred, std::move(Task)).share();
177 // Wrap the future so that both ThreadPool::wait() can operate and the
178 // returned future can be sync'ed on.
179 Tasks.emplace_back(std::make_pair([Future]() { Future.get(); }, Group));
180 return Future;
181#endif
182 }
183
184#if LLVM_ENABLE_THREADS
185 // Grow to ensure that we have at least `requested` Threads, but do not go
186 // over MaxThreadCount.
187 void grow(int requested);
188
189 void processTasks(ThreadPoolTaskGroup *WaitingForGroup);
190#endif
191
192 /// Threads in flight
193 std::vector<llvm::thread> Threads;
194 /// Lock protecting access to the Threads vector.
195 mutable llvm::sys::RWMutex ThreadsLock;
196
197 /// Tasks waiting for execution in the pool.
198 std::deque<std::pair<std::function<void()>, ThreadPoolTaskGroup *>> Tasks;
199
200 /// Locking and signaling for accessing the Tasks queue.
201 std::mutex QueueLock;
202 std::condition_variable QueueCondition;
203
204 /// Signaling for job completion (all tasks or all tasks in a group).
205 std::condition_variable CompletionCondition;
206
207 /// Keep track of the number of thread actually busy
208 unsigned ActiveThreads = 0;
209 /// Number of threads active for tasks in the given group (only non-zero).
210 DenseMap<ThreadPoolTaskGroup *, unsigned> ActiveGroups;
211
212#if LLVM_ENABLE_THREADS // avoids warning for unused variable
213 /// Signal for the destruction of the pool, asking thread to exit.
214 bool EnableFlag = true;
215#endif
216
217 const ThreadPoolStrategy Strategy;
218
219 /// Maximum number of threads to potentially grow this pool to.
220 const unsigned MaxThreadCount;
221};
222
223/// A group of tasks to be run on a thread pool. Thread pool tasks in different
224/// groups can run on the same threadpool but can be waited for separately.
225/// It is even possible for tasks of one group to submit and wait for tasks
226/// of another group, as long as this does not form a loop.
228public:
229 /// The ThreadPool argument is the thread pool to forward calls to.
230 ThreadPoolTaskGroup(ThreadPool &Pool) : Pool(Pool) {}
231
232 /// Blocking destructor: will wait for all the tasks in the group to complete
233 /// by calling ThreadPool::wait().
235
236 /// Calls ThreadPool::async() for this group.
237 template <typename Function, typename... Args>
238 inline auto async(Function &&F, Args &&...ArgList) {
239 return Pool.async(*this, std::forward<Function>(F),
240 std::forward<Args>(ArgList)...);
241 }
242
243 /// Calls ThreadPool::wait() for this group.
244 void wait() { Pool.wait(*this); }
245
246private:
247 ThreadPool &Pool;
248};
249
250} // namespace llvm
251
252#endif // LLVM_SUPPORT_THREADPOOL_H
#define LLVM_DEPRECATED(MSG, FIX)
Definition: Compiler.h:157
This file defines the DenseMap class.
#define F(x, y, z)
Definition: MD5.cpp:55
assert(ImpDefSCC.getReg()==AMDGPU::SCC &&ImpDefSCC.isDef())
This tells how a thread pool will be used.
Definition: Threading.h:116
A group of tasks to be run on a thread pool.
Definition: ThreadPool.h:227
auto async(Function &&F, Args &&...ArgList)
Calls ThreadPool::async() for this group.
Definition: ThreadPool.h:238
ThreadPoolTaskGroup(ThreadPool &Pool)
The ThreadPool argument is the thread pool to forward calls to.
Definition: ThreadPool.h:230
void wait()
Calls ThreadPool::wait() for this group.
Definition: ThreadPool.h:244
~ThreadPoolTaskGroup()
Blocking destructor: will wait for all the tasks in the group to complete by calling ThreadPool::wait...
Definition: ThreadPool.h:234
A ThreadPool for asynchronous parallel execution on a defined number of threads.
Definition: ThreadPool.h:52
void wait()
Blocking wait for all the threads to complete and the queue to be empty.
Definition: ThreadPool.cpp:202
~ThreadPool()
Blocking destructor: the pool will wait for all the threads to complete.
Definition: ThreadPool.cpp:221
bool isWorkerThread() const
Returns true if the current thread is a worker thread of this thread pool.
Definition: ThreadPool.cpp:217
unsigned getMaxConcurrency() const
Definition: ThreadPool.h:109
unsigned getThreadCount() const
Definition: ThreadPool.h:113
auto async(ThreadPoolTaskGroup &Group, Function &&F, Args &&...ArgList)
Overload, task will be in the given task group.
Definition: ThreadPool.h:74
auto async(Function &&F, Args &&...ArgList)
Asynchronous submission of a task to the pool.
Definition: ThreadPool.h:66
auto async(Func &&F) -> std::shared_future< decltype(F())>
Asynchronous submission of a task to the pool.
Definition: ThreadPool.h:83
auto async(ThreadPoolTaskGroup &Group, Func &&F) -> std::shared_future< decltype(F())>
Definition: ThreadPool.h:89
This is an optimization pass for GlobalISel generic memory operations.
Definition: AddressRanges.h:18
ThreadPoolStrategy hardware_concurrency(unsigned ThreadCount=0)
Returns a default thread strategy where all available hardware resources are to be used,...
Definition: Threading.h:185