1//==-- llvm/Support/ThreadPool.cpp - 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//
10// This file implements a crude C++11 based thread pool.
11//
12//===----------------------------------------------------------------------===//
13
14#include "llvm/Support/ThreadPool.h"
15
16#include "llvm/Config/llvm-config.h"
17
18#include "llvm/ADT/ScopeExit.h"
19#include "llvm/Support/ExponentialBackoff.h"
20#include "llvm/Support/FormatVariadic.h"
21#include "llvm/Support/Threading.h"
22#include "llvm/Support/raw_ostream.h"
23
24using namespace llvm;
25
26ThreadPoolInterface::~ThreadPoolInterface() = default;
27
28// A note on thread groups: Tasks are by default in no group (represented
29// by nullptr ThreadPoolTaskGroup pointer in the Tasks queue) and functionality
30// here normally works on all tasks regardless of their group (functions
31// in that case receive nullptr ThreadPoolTaskGroup pointer as argument).
32// A task in a group has a pointer to that ThreadPoolTaskGroup in the Tasks
33// queue, and functions called to work only on tasks from one group take that
34// pointer.
35
36#if LLVM_ENABLE_THREADS
37
38StdThreadPool::StdThreadPool(ThreadPoolStrategy S)
39 : Strategy(S), MaxThreadCount(S.compute_thread_count()) {
40 if (Strategy.UseJobserver)
41 TheJobserver = JobserverClient::getInstance();
42}
43
44void StdThreadPool::grow(int requested) {
45 llvm::sys::ScopedWriter LockGuard(ThreadsLock);
46 if (Threads.size() >= MaxThreadCount)
47 return; // Already hit the max thread pool size.
48 int newThreadCount = std::min<int>(a: requested, b: MaxThreadCount);
49 while (static_cast<int>(Threads.size()) < newThreadCount) {
50 int ThreadID = Threads.size();
51 Threads.emplace_back(args: [this, ThreadID] {
52 set_thread_name(formatv(Fmt: "llvm-worker-{0}", Vals: ThreadID));
53 Strategy.apply_thread_strategy(ThreadPoolNum: ThreadID);
54 // Note on jobserver deadlock avoidance:
55 // GNU Make grants each invoked process one implicit job slot.
56 // JobserverClient::tryAcquire() returns that implicit slot on the first
57 // successful call in a process, ensuring forward progress without a
58 // dedicated "always-on" thread.
59 if (TheJobserver)
60 processTasksWithJobserver();
61 else
62 processTasks(WaitingForGroup: nullptr);
63 });
64 }
65}
66
67#ifndef NDEBUG
68// The group of the tasks run by the current thread.
69static LLVM_THREAD_LOCAL std::vector<ThreadPoolTaskGroup *>
70 *CurrentThreadTaskGroups = nullptr;
71#endif
72
73// WaitingForGroup == nullptr means all tasks regardless of their group.
74void StdThreadPool::processTasks(ThreadPoolTaskGroup *WaitingForGroup) {
75 while (true) {
76 llvm::unique_function<void()> Task;
77 ThreadPoolTaskGroup *GroupOfTask;
78 {
79 std::unique_lock<std::mutex> LockGuard(QueueLock);
80 bool workCompletedForGroup = false; // Result of workCompletedUnlocked()
81 // Wait for tasks to be pushed in the queue
82 QueueCondition.wait(lock&: LockGuard, p: [&] {
83 return !EnableFlag || !Tasks.empty() ||
84 (WaitingForGroup != nullptr &&
85 (workCompletedForGroup =
86 workCompletedUnlocked(Group: WaitingForGroup)));
87 });
88 // Exit condition
89 if (!EnableFlag && Tasks.empty())
90 return;
91 if (WaitingForGroup != nullptr && workCompletedForGroup)
92 return;
93 // Yeah, we have a task, grab it and release the lock on the queue
94
95 // We first need to signal that we are active before popping the queue
96 // in order for wait() to properly detect that even if the queue is
97 // empty, there is still a task in flight.
98 ++ActiveThreads;
99 Task = std::move(Tasks.front().first);
100 GroupOfTask = Tasks.front().second;
101 // Need to count active threads in each group separately, ActiveThreads
102 // would never be 0 if waiting for another group inside a wait.
103 if (GroupOfTask != nullptr)
104 ++ActiveGroups[GroupOfTask]; // Increment or set to 1 if new item
105 Tasks.pop_front();
106 }
107#ifndef NDEBUG
108 if (CurrentThreadTaskGroups == nullptr)
109 CurrentThreadTaskGroups = new std::vector<ThreadPoolTaskGroup *>;
110 CurrentThreadTaskGroups->push_back(GroupOfTask);
111#endif
112
113 // Run the task we just grabbed. This also destroys the task once run to
114 // release any resources held by it through RAII captured objects.
115 //
116 // It is particularly important to do this here so that we're not holding
117 // any lock and any further operations on the thread or `ThreadPool` take
118 // place here, at the same point as the task itself is executed.
119 std::exchange(obj&: Task, new_val: {})();
120
121#ifndef NDEBUG
122 CurrentThreadTaskGroups->pop_back();
123 if (CurrentThreadTaskGroups->empty()) {
124 delete CurrentThreadTaskGroups;
125 CurrentThreadTaskGroups = nullptr;
126 }
127#endif
128
129 bool Notify;
130 bool NotifyGroup;
131 {
132 // Adjust `ActiveThreads`, in case someone waits on StdThreadPool::wait()
133 std::lock_guard<std::mutex> LockGuard(QueueLock);
134 --ActiveThreads;
135 if (GroupOfTask != nullptr) {
136 auto A = ActiveGroups.find(Val: GroupOfTask);
137 if (--(A->second) == 0)
138 ActiveGroups.erase(I: A);
139 }
140 Notify = workCompletedUnlocked(Group: GroupOfTask);
141 NotifyGroup = GroupOfTask != nullptr && Notify;
142 }
143 // Notify task completion if this is the last active thread, in case
144 // someone waits on StdThreadPool::wait().
145 if (Notify)
146 CompletionCondition.notify_all();
147 // If this was a task in a group, notify also threads waiting for tasks
148 // in this function on QueueCondition, to make a recursive wait() return
149 // after the group it's been waiting for has finished.
150 if (NotifyGroup)
151 QueueCondition.notify_all();
152 }
153}
154
155/// Main loop for worker threads when using a jobserver.
156/// This function uses a two-level queue; it first acquires a job slot from the
157/// external jobserver, then retrieves a task from the internal queue.
158/// This allows the thread pool to cooperate with build systems like `make -j`.
159void StdThreadPool::processTasksWithJobserver() {
160 while (true) {
161 // Acquire a job slot from the external jobserver.
162 // This polls for a slot and yields the thread to avoid a high-CPU wait.
163 JobSlot Slot;
164 // The timeout for the backoff can be very long, as the shutdown
165 // is checked on each iteration. The sleep duration is capped by MaxWait
166 // in ExponentialBackoff, so shutdown latency is not a problem.
167 ExponentialBackoff Backoff(std::chrono::hours(24));
168 bool AcquiredToken = false;
169 do {
170 // Return if the thread pool is shutting down.
171 {
172 std::unique_lock<std::mutex> LockGuard(QueueLock);
173 if (!EnableFlag)
174 return;
175 }
176
177 Slot = TheJobserver->tryAcquire();
178 if (Slot.isValid()) {
179 AcquiredToken = true;
180 break;
181 }
182 } while (Backoff.waitForNextAttempt());
183
184 if (!AcquiredToken) {
185 // This is practically unreachable with a 24h timeout and indicates a
186 // deeper problem if hit.
187 report_fatal_error(reason: "Timed out waiting for jobserver token.");
188 }
189
190 // `llvm::scope_exit` guarantees the job slot is released, even if the
191 // task throws or we exit early. This prevents deadlocking the build.
192 llvm::scope_exit SlotReleaser(
193 [&] { TheJobserver->release(Slot: std::move(Slot)); });
194
195 // While we hold a job slot, process tasks from the internal queue.
196 while (true) {
197 llvm::unique_function<void()> Task;
198 ThreadPoolTaskGroup *GroupOfTask = nullptr;
199
200 {
201 std::unique_lock<std::mutex> LockGuard(QueueLock);
202
203 // Wait until a task is available or the pool is shutting down.
204 QueueCondition.wait(lock&: LockGuard,
205 p: [&] { return !EnableFlag || !Tasks.empty(); });
206
207 // If shutting down and the queue is empty, the thread can terminate.
208 if (!EnableFlag && Tasks.empty())
209 return;
210
211 // If the queue is empty, we're done processing tasks for now.
212 // Break the inner loop to release the job slot.
213 if (Tasks.empty())
214 break;
215
216 // A task is available. Mark it as active before releasing the lock
217 // to prevent race conditions with `wait()`.
218 ++ActiveThreads;
219 Task = std::move(Tasks.front().first);
220 GroupOfTask = Tasks.front().second;
221 if (GroupOfTask != nullptr)
222 ++ActiveGroups[GroupOfTask];
223 Tasks.pop_front();
224 } // The queue lock is released.
225
226 // Run the task. The job slot remains acquired during execution.
227 Task();
228
229 // The task has finished. Update the active count and notify any waiters.
230 {
231 std::lock_guard<std::mutex> LockGuard(QueueLock);
232 --ActiveThreads;
233 if (GroupOfTask != nullptr) {
234 auto A = ActiveGroups.find(Val: GroupOfTask);
235 if (--(A->second) == 0)
236 ActiveGroups.erase(I: A);
237 }
238 // If all tasks are complete, notify any waiting threads.
239 if (workCompletedUnlocked(Group: nullptr))
240 CompletionCondition.notify_all();
241 }
242 }
243 }
244}
245bool StdThreadPool::workCompletedUnlocked(ThreadPoolTaskGroup *Group) const {
246 if (Group == nullptr)
247 return !ActiveThreads && Tasks.empty();
248 return ActiveGroups.count(Val: Group) == 0 &&
249 !llvm::is_contained(Range: llvm::make_second_range(c: Tasks), Element: Group);
250}
251
252void StdThreadPool::wait() {
253 assert(!isWorkerThread()); // Would deadlock waiting for itself.
254 // Wait for all threads to complete and the queue to be empty
255 std::unique_lock<std::mutex> LockGuard(QueueLock);
256 CompletionCondition.wait(lock&: LockGuard,
257 p: [&] { return workCompletedUnlocked(Group: nullptr); });
258}
259
260void StdThreadPool::wait(ThreadPoolTaskGroup &Group) {
261 // Wait for all threads in the group to complete.
262 if (!isWorkerThread()) {
263 std::unique_lock<std::mutex> LockGuard(QueueLock);
264 CompletionCondition.wait(lock&: LockGuard,
265 p: [&] { return workCompletedUnlocked(Group: &Group); });
266 return;
267 }
268 // Make sure to not deadlock waiting for oneself.
269 assert(CurrentThreadTaskGroups == nullptr ||
270 !llvm::is_contained(*CurrentThreadTaskGroups, &Group));
271 // Handle the case of recursive call from another task in a different group,
272 // in which case process tasks while waiting to keep the thread busy and avoid
273 // possible deadlock.
274 processTasks(WaitingForGroup: &Group);
275}
276
277bool StdThreadPool::isWorkerThread() const {
278 llvm::sys::ScopedReader LockGuard(ThreadsLock);
279 llvm::thread::id CurrentThreadId = llvm::this_thread::get_id();
280 for (const llvm::thread &Thread : Threads)
281 if (CurrentThreadId == Thread.get_id())
282 return true;
283 return false;
284}
285
286// The destructor joins all threads, waiting for completion.
287StdThreadPool::~StdThreadPool() {
288 {
289 std::unique_lock<std::mutex> LockGuard(QueueLock);
290 EnableFlag = false;
291 }
292 QueueCondition.notify_all();
293 llvm::sys::ScopedReader LockGuard(ThreadsLock);
294 for (auto &Worker : Threads)
295 Worker.join();
296}
297
298#endif // LLVM_ENABLE_THREADS Disabled
299
300// No threads are launched, issue a warning if ThreadCount is not 0
301SingleThreadExecutor::SingleThreadExecutor(ThreadPoolStrategy S) {
302 int ThreadCount = S.compute_thread_count();
303 if (ThreadCount != 1) {
304 errs() << "Warning: request a ThreadPool with " << ThreadCount
305 << " threads, but LLVM_ENABLE_THREADS has been turned off\n";
306 }
307}
308
309void SingleThreadExecutor::wait() {
310 // Sequential implementation running the tasks
311 while (!Tasks.empty()) {
312 auto Task = std::move(Tasks.front().first);
313 Tasks.pop_front();
314 Task();
315 }
316}
317
318void SingleThreadExecutor::wait(ThreadPoolTaskGroup &) {
319 // Simply wait for all, this works even if recursive (the running task
320 // is already removed from the queue).
321 wait();
322}
323
324bool SingleThreadExecutor::isWorkerThread() const {
325 report_fatal_error(reason: "LLVM compiled without multithreading");
326}
327
328SingleThreadExecutor::~SingleThreadExecutor() { wait(); }
329