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 // This file implements a crude C++11 based thread pool. 10 // 11 //===----------------------------------------------------------------------===// 12 13 #include "llvm/Support/ThreadPool.h" 14 15 #include "llvm/Config/llvm-config.h" 16 #include "llvm/Support/Threading.h" 17 #include "llvm/Support/raw_ostream.h" 18 19 using namespace llvm; 20 21 #if LLVM_ENABLE_THREADS 22 23 ThreadPool::ThreadPool(ThreadPoolStrategy S) 24 : ThreadCount(S.compute_thread_count()) { 25 // Create ThreadCount threads that will loop forever, wait on QueueCondition 26 // for tasks to be queued or the Pool to be destroyed. 27 Threads.reserve(ThreadCount); 28 for (unsigned ThreadID = 0; ThreadID < ThreadCount; ++ThreadID) { 29 Threads.emplace_back([S, ThreadID, this] { 30 S.apply_thread_strategy(ThreadID); 31 while (true) { 32 PackagedTaskTy Task; 33 { 34 std::unique_lock<std::mutex> LockGuard(QueueLock); 35 // Wait for tasks to be pushed in the queue 36 QueueCondition.wait(LockGuard, 37 [&] { return !EnableFlag || !Tasks.empty(); }); 38 // Exit condition 39 if (!EnableFlag && Tasks.empty()) 40 return; 41 // Yeah, we have a task, grab it and release the lock on the queue 42 43 // We first need to signal that we are active before popping the queue 44 // in order for wait() to properly detect that even if the queue is 45 // empty, there is still a task in flight. 46 ++ActiveThreads; 47 Task = std::move(Tasks.front()); 48 Tasks.pop(); 49 } 50 // Run the task we just grabbed 51 Task(); 52 53 bool Notify; 54 { 55 // Adjust `ActiveThreads`, in case someone waits on ThreadPool::wait() 56 std::lock_guard<std::mutex> LockGuard(QueueLock); 57 --ActiveThreads; 58 Notify = workCompletedUnlocked(); 59 } 60 // Notify task completion if this is the last active thread, in case 61 // someone waits on ThreadPool::wait(). 62 if (Notify) 63 CompletionCondition.notify_all(); 64 } 65 }); 66 } 67 } 68 69 void ThreadPool::wait() { 70 // Wait for all threads to complete and the queue to be empty 71 std::unique_lock<std::mutex> LockGuard(QueueLock); 72 CompletionCondition.wait(LockGuard, [&] { return workCompletedUnlocked(); }); 73 } 74 75 std::shared_future<void> ThreadPool::asyncImpl(TaskTy Task) { 76 /// Wrap the Task in a packaged_task to return a future object. 77 PackagedTaskTy PackagedTask(std::move(Task)); 78 auto Future = PackagedTask.get_future(); 79 { 80 // Lock the queue and push the new task 81 std::unique_lock<std::mutex> LockGuard(QueueLock); 82 83 // Don't allow enqueueing after disabling the pool 84 assert(EnableFlag && "Queuing a thread during ThreadPool destruction"); 85 86 Tasks.push(std::move(PackagedTask)); 87 } 88 QueueCondition.notify_one(); 89 return Future.share(); 90 } 91 92 // The destructor joins all threads, waiting for completion. 93 ThreadPool::~ThreadPool() { 94 { 95 std::unique_lock<std::mutex> LockGuard(QueueLock); 96 EnableFlag = false; 97 } 98 QueueCondition.notify_all(); 99 for (auto &Worker : Threads) 100 Worker.join(); 101 } 102 103 #else // LLVM_ENABLE_THREADS Disabled 104 105 // No threads are launched, issue a warning if ThreadCount is not 0 106 ThreadPool::ThreadPool(ThreadPoolStrategy S) 107 : ThreadCount(S.compute_thread_count()) { 108 if (ThreadCount != 1) { 109 errs() << "Warning: request a ThreadPool with " << ThreadCount 110 << " threads, but LLVM_ENABLE_THREADS has been turned off\n"; 111 } 112 } 113 114 void ThreadPool::wait() { 115 // Sequential implementation running the tasks 116 while (!Tasks.empty()) { 117 auto Task = std::move(Tasks.front()); 118 Tasks.pop(); 119 Task(); 120 } 121 } 122 123 std::shared_future<void> ThreadPool::asyncImpl(TaskTy Task) { 124 // Get a Future with launch::deferred execution using std::async 125 auto Future = std::async(std::launch::deferred, std::move(Task)).share(); 126 // Wrap the future so that both ThreadPool::wait() can operate and the 127 // returned future can be sync'ed on. 128 PackagedTaskTy PackagedTask([Future]() { Future.get(); }); 129 Tasks.push(std::move(PackagedTask)); 130 return Future; 131 } 132 133 ThreadPool::~ThreadPool() { wait(); } 134 135 #endif 136