303 lines · cpp
1//===- llvm/Support/Parallel.cpp - Parallel algorithms --------------------===//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#include "llvm/Support/Parallel.h"10#include "llvm/ADT/ScopeExit.h"11#include "llvm/Config/llvm-config.h"12#include "llvm/Support/ExponentialBackoff.h"13#include "llvm/Support/Jobserver.h"14#include "llvm/Support/ManagedStatic.h"15#include "llvm/Support/Threading.h"16 17#include <atomic>18#include <future>19#include <memory>20#include <mutex>21#include <thread>22#include <vector>23 24llvm::ThreadPoolStrategy llvm::parallel::strategy;25 26namespace llvm {27namespace parallel {28#if LLVM_ENABLE_THREADS29 30#ifdef _WIN3231static thread_local unsigned threadIndex = UINT_MAX;32 33unsigned getThreadIndex() { GET_THREAD_INDEX_IMPL; }34#else35thread_local unsigned threadIndex = UINT_MAX;36#endif37 38namespace detail {39 40namespace {41 42/// An abstract class that takes closures and runs them asynchronously.43class Executor {44public:45 virtual ~Executor() = default;46 virtual void add(std::function<void()> func) = 0;47 virtual size_t getThreadCount() const = 0;48 49 static Executor *getDefaultExecutor();50};51 52/// An implementation of an Executor that runs closures on a thread pool53/// in filo order.54class ThreadPoolExecutor : public Executor {55public:56 explicit ThreadPoolExecutor(ThreadPoolStrategy S) {57 if (S.UseJobserver)58 TheJobserver = JobserverClient::getInstance();59 60 ThreadCount = S.compute_thread_count();61 // Spawn all but one of the threads in another thread as spawning threads62 // can take a while.63 Threads.reserve(ThreadCount);64 Threads.resize(1);65 std::lock_guard<std::mutex> Lock(Mutex);66 // Use operator[] before creating the thread to avoid data race in .size()67 // in 'safe libc++' mode.68 auto &Thread0 = Threads[0];69 Thread0 = std::thread([this, S] {70 for (unsigned I = 1; I < ThreadCount; ++I) {71 Threads.emplace_back([this, S, I] { work(S, I); });72 if (Stop)73 break;74 }75 ThreadsCreated.set_value();76 work(S, 0);77 });78 }79 80 // To make sure the thread pool executor can only be created with a parallel81 // strategy.82 ThreadPoolExecutor() = delete;83 84 void stop() {85 {86 std::lock_guard<std::mutex> Lock(Mutex);87 if (Stop)88 return;89 Stop = true;90 }91 Cond.notify_all();92 ThreadsCreated.get_future().wait();93 }94 95 ~ThreadPoolExecutor() override {96 stop();97 std::thread::id CurrentThreadId = std::this_thread::get_id();98 for (std::thread &T : Threads)99 if (T.get_id() == CurrentThreadId)100 T.detach();101 else102 T.join();103 }104 105 struct Creator {106 static void *call() { return new ThreadPoolExecutor(strategy); }107 };108 struct Deleter {109 static void call(void *Ptr) { ((ThreadPoolExecutor *)Ptr)->stop(); }110 };111 112 void add(std::function<void()> F) override {113 {114 std::lock_guard<std::mutex> Lock(Mutex);115 WorkStack.push_back(std::move(F));116 }117 Cond.notify_one();118 }119 120 size_t getThreadCount() const override { return ThreadCount; }121 122private:123 void work(ThreadPoolStrategy S, unsigned ThreadID) {124 threadIndex = ThreadID;125 S.apply_thread_strategy(ThreadID);126 // Note on jobserver deadlock avoidance:127 // GNU Make grants each invoked process one implicit job slot. Our128 // JobserverClient models this by returning an implicit JobSlot on the129 // first successful tryAcquire() in a process. This guarantees forward130 // progress without requiring a dedicated "always-on" thread here.131 132 static thread_local std::unique_ptr<ExponentialBackoff> Backoff;133 134 while (true) {135 if (TheJobserver) {136 // Jobserver-mode scheduling:137 // - Acquire one job slot (with exponential backoff to avoid busy-wait).138 // - While holding the slot, drain and run tasks from the local queue.139 // - Release the slot when the queue is empty or when shutting down.140 // Rationale: Holding a slot amortizes acquire/release overhead over141 // multiple tasks and avoids requeue/yield churn, while still enforcing142 // the jobserver’s global concurrency limit. With K available slots,143 // up to K workers run tasks in parallel; within each worker tasks run144 // sequentially until the local queue is empty.145 ExponentialBackoff Backoff(std::chrono::hours(24));146 JobSlot Slot;147 do {148 if (Stop)149 return;150 Slot = TheJobserver->tryAcquire();151 if (Slot.isValid())152 break;153 } while (Backoff.waitForNextAttempt());154 155 auto SlotReleaser = llvm::make_scope_exit(156 [&] { TheJobserver->release(std::move(Slot)); });157 158 while (true) {159 std::function<void()> Task;160 {161 std::unique_lock<std::mutex> Lock(Mutex);162 Cond.wait(Lock, [&] { return Stop || !WorkStack.empty(); });163 if (Stop && WorkStack.empty())164 return;165 if (WorkStack.empty())166 break;167 Task = std::move(WorkStack.back());168 WorkStack.pop_back();169 }170 Task();171 }172 } else {173 std::unique_lock<std::mutex> Lock(Mutex);174 Cond.wait(Lock, [&] { return Stop || !WorkStack.empty(); });175 if (Stop)176 break;177 auto Task = std::move(WorkStack.back());178 WorkStack.pop_back();179 Lock.unlock();180 Task();181 }182 }183 }184 185 std::atomic<bool> Stop{false};186 std::vector<std::function<void()>> WorkStack;187 std::mutex Mutex;188 std::condition_variable Cond;189 std::promise<void> ThreadsCreated;190 std::vector<std::thread> Threads;191 unsigned ThreadCount;192 193 JobserverClient *TheJobserver = nullptr;194};195 196Executor *Executor::getDefaultExecutor() {197#ifdef _WIN32198 // The ManagedStatic enables the ThreadPoolExecutor to be stopped via199 // llvm_shutdown() which allows a "clean" fast exit, e.g. via _exit(). This200 // stops the thread pool and waits for any worker thread creation to complete201 // but does not wait for the threads to finish. The wait for worker thread202 // creation to complete is important as it prevents intermittent crashes on203 // Windows due to a race condition between thread creation and process exit.204 //205 // The ThreadPoolExecutor will only be destroyed when the static unique_ptr to206 // it is destroyed, i.e. in a normal full exit. The ThreadPoolExecutor207 // destructor ensures it has been stopped and waits for worker threads to208 // finish. The wait is important as it prevents intermittent crashes on209 // Windows when the process is doing a full exit.210 //211 // The Windows crashes appear to only occur with the MSVC static runtimes and212 // are more frequent with the debug static runtime.213 //214 // This also prevents intermittent deadlocks on exit with the MinGW runtime.215 216 static ManagedStatic<ThreadPoolExecutor, ThreadPoolExecutor::Creator,217 ThreadPoolExecutor::Deleter>218 ManagedExec;219 static std::unique_ptr<ThreadPoolExecutor> Exec(&(*ManagedExec));220 return Exec.get();221#else222 // ManagedStatic is not desired on other platforms. When `Exec` is destroyed223 // by llvm_shutdown(), worker threads will clean up and invoke TLS224 // destructors. This can lead to race conditions if other threads attempt to225 // access TLS objects that have already been destroyed.226 static ThreadPoolExecutor Exec(strategy);227 return &Exec;228#endif229}230} // namespace231} // namespace detail232 233size_t getThreadCount() {234 return detail::Executor::getDefaultExecutor()->getThreadCount();235}236#endif237 238// Latch::sync() called by the dtor may cause one thread to block. If is a dead239// lock if all threads in the default executor are blocked. To prevent the dead240// lock, only allow the root TaskGroup to run tasks parallelly. In the scenario241// of nested parallel_for_each(), only the outermost one runs parallelly.242TaskGroup::TaskGroup()243#if LLVM_ENABLE_THREADS244 : Parallel((parallel::strategy.ThreadsRequested != 1) &&245 (threadIndex == UINT_MAX)) {}246#else247 : Parallel(false) {}248#endif249TaskGroup::~TaskGroup() {250 // We must ensure that all the workloads have finished before decrementing the251 // instances count.252 L.sync();253}254 255void TaskGroup::spawn(std::function<void()> F) {256#if LLVM_ENABLE_THREADS257 if (Parallel) {258 L.inc();259 detail::Executor::getDefaultExecutor()->add([&, F = std::move(F)] {260 F();261 L.dec();262 });263 return;264 }265#endif266 F();267}268 269} // namespace parallel270} // namespace llvm271 272void llvm::parallelFor(size_t Begin, size_t End,273 llvm::function_ref<void(size_t)> Fn) {274#if LLVM_ENABLE_THREADS275 if (parallel::strategy.ThreadsRequested != 1) {276 auto NumItems = End - Begin;277 // Limit the number of tasks to MaxTasksPerGroup to limit job scheduling278 // overhead on large inputs.279 auto TaskSize = NumItems / parallel::detail::MaxTasksPerGroup;280 if (TaskSize == 0)281 TaskSize = 1;282 283 parallel::TaskGroup TG;284 for (; Begin + TaskSize < End; Begin += TaskSize) {285 TG.spawn([=, &Fn] {286 for (size_t I = Begin, E = Begin + TaskSize; I != E; ++I)287 Fn(I);288 });289 }290 if (Begin != End) {291 TG.spawn([=, &Fn] {292 for (size_t I = Begin; I != End; ++I)293 Fn(I);294 });295 }296 return;297 }298#endif299 300 for (; Begin != End; ++Begin)301 Fn(Begin);302}303