329 lines · cpp
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-exception6//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 (represented29// by nullptr ThreadPoolTaskGroup pointer in the Tasks queue) and functionality30// here normally works on all tasks regardless of their group (functions31// in that case receive nullptr ThreadPoolTaskGroup pointer as argument).32// A task in a group has a pointer to that ThreadPoolTaskGroup in the Tasks33// queue, and functions called to work only on tasks from one group take that34// pointer.35 36#if LLVM_ENABLE_THREADS37 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>(requested, MaxThreadCount);49 while (static_cast<int>(Threads.size()) < newThreadCount) {50 int ThreadID = Threads.size();51 Threads.emplace_back([this, ThreadID] {52 set_thread_name(formatv("llvm-worker-{0}", ThreadID));53 Strategy.apply_thread_strategy(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 first57 // successful call in a process, ensuring forward progress without a58 // dedicated "always-on" thread.59 if (TheJobserver)60 processTasksWithJobserver();61 else62 processTasks(nullptr);63 });64 }65}66 67#ifndef NDEBUG68// The group of the tasks run by the current thread.69static LLVM_THREAD_LOCAL std::vector<ThreadPoolTaskGroup *>70 *CurrentThreadTaskGroups = nullptr;71#endif72 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 queue82 QueueCondition.wait(LockGuard, [&] {83 return !EnableFlag || !Tasks.empty() ||84 (WaitingForGroup != nullptr &&85 (workCompletedForGroup =86 workCompletedUnlocked(WaitingForGroup)));87 });88 // Exit condition89 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 queue94 95 // We first need to signal that we are active before popping the queue96 // in order for wait() to properly detect that even if the queue is97 // 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, ActiveThreads102 // 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 item105 Tasks.pop_front();106 }107#ifndef NDEBUG108 if (CurrentThreadTaskGroups == nullptr)109 CurrentThreadTaskGroups = new std::vector<ThreadPoolTaskGroup *>;110 CurrentThreadTaskGroups->push_back(GroupOfTask);111#endif112 113 // Run the task we just grabbed. This also destroys the task once run to114 // 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 holding117 // any lock and any further operations on the thread or `ThreadPool` take118 // place here, at the same point as the task itself is executed.119 std::exchange(Task, {})();120 121#ifndef NDEBUG122 CurrentThreadTaskGroups->pop_back();123 if (CurrentThreadTaskGroups->empty()) {124 delete CurrentThreadTaskGroups;125 CurrentThreadTaskGroups = nullptr;126 }127#endif128 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(GroupOfTask);137 if (--(A->second) == 0)138 ActiveGroups.erase(A);139 }140 Notify = workCompletedUnlocked(GroupOfTask);141 NotifyGroup = GroupOfTask != nullptr && Notify;142 }143 // Notify task completion if this is the last active thread, in case144 // 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 tasks148 // in this function on QueueCondition, to make a recursive wait() return149 // 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 the157/// 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 shutdown165 // is checked on each iteration. The sleep duration is capped by MaxWait166 // 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 a186 // deeper problem if hit.187 report_fatal_error("Timed out waiting for jobserver token.");188 }189 190 // `make_scope_exit` guarantees the job slot is released, even if the191 // task throws or we exit early. This prevents deadlocking the build.192 auto SlotReleaser =193 make_scope_exit([&] { TheJobserver->release(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(LockGuard,205 [&] { 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 lock217 // 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(GroupOfTask);235 if (--(A->second) == 0)236 ActiveGroups.erase(A);237 }238 // If all tasks are complete, notify any waiting threads.239 if (workCompletedUnlocked(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(Group) == 0 &&249 !llvm::is_contained(llvm::make_second_range(Tasks), 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 empty255 std::unique_lock<std::mutex> LockGuard(QueueLock);256 CompletionCondition.wait(LockGuard,257 [&] { return workCompletedUnlocked(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(LockGuard,265 [&] { return workCompletedUnlocked(&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 avoid273 // possible deadlock.274 processTasks(&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 Disabled299 300// No threads are launched, issue a warning if ThreadCount is not 0301SingleThreadExecutor::SingleThreadExecutor(ThreadPoolStrategy S) {302 int ThreadCount = S.compute_thread_count();303 if (ThreadCount != 1) {304 errs() << "Warning: request a ThreadPool with " << ThreadCount305 << " threads, but LLVM_ENABLE_THREADS has been turned off\n";306 }307}308 309void SingleThreadExecutor::wait() {310 // Sequential implementation running the tasks311 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 task320 // is already removed from the queue).321 wait();322}323 324bool SingleThreadExecutor::isWorkerThread() const {325 report_fatal_error("LLVM compiled without multithreading");326}327 328SingleThreadExecutor::~SingleThreadExecutor() { wait(); }329