1*0b57cec5SDimitry Andric //===- llvm/Support/Parallel.cpp - Parallel algorithms --------------------===//
2*0b57cec5SDimitry Andric //
3*0b57cec5SDimitry Andric // Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
4*0b57cec5SDimitry Andric // See https://llvm.org/LICENSE.txt for license information.
5*0b57cec5SDimitry Andric // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
6*0b57cec5SDimitry Andric //
7*0b57cec5SDimitry Andric //===----------------------------------------------------------------------===//
8*0b57cec5SDimitry Andric 
9*0b57cec5SDimitry Andric #include "llvm/Support/Parallel.h"
10*0b57cec5SDimitry Andric #include "llvm/Config/llvm-config.h"
11480093f4SDimitry Andric #include "llvm/Support/ManagedStatic.h"
12*0b57cec5SDimitry Andric 
13*0b57cec5SDimitry Andric #if LLVM_ENABLE_THREADS
14*0b57cec5SDimitry Andric 
15*0b57cec5SDimitry Andric #include "llvm/Support/Threading.h"
16*0b57cec5SDimitry Andric 
17*0b57cec5SDimitry Andric #include <atomic>
18480093f4SDimitry Andric #include <future>
19*0b57cec5SDimitry Andric #include <stack>
20*0b57cec5SDimitry Andric #include <thread>
21480093f4SDimitry Andric #include <vector>
22*0b57cec5SDimitry Andric 
23*0b57cec5SDimitry Andric namespace llvm {
24*0b57cec5SDimitry Andric namespace parallel {
25*0b57cec5SDimitry Andric namespace detail {
26*0b57cec5SDimitry Andric 
27*0b57cec5SDimitry Andric namespace {
28*0b57cec5SDimitry Andric 
29*0b57cec5SDimitry Andric /// An abstract class that takes closures and runs them asynchronously.
30*0b57cec5SDimitry Andric class Executor {
31*0b57cec5SDimitry Andric public:
32*0b57cec5SDimitry Andric   virtual ~Executor() = default;
33*0b57cec5SDimitry Andric   virtual void add(std::function<void()> func) = 0;
34*0b57cec5SDimitry Andric 
35*0b57cec5SDimitry Andric   static Executor *getDefaultExecutor();
36*0b57cec5SDimitry Andric };
37*0b57cec5SDimitry Andric 
38*0b57cec5SDimitry Andric /// An implementation of an Executor that runs closures on a thread pool
39*0b57cec5SDimitry Andric ///   in filo order.
40*0b57cec5SDimitry Andric class ThreadPoolExecutor : public Executor {
41*0b57cec5SDimitry Andric public:
42480093f4SDimitry Andric   explicit ThreadPoolExecutor(unsigned ThreadCount = hardware_concurrency()) {
43*0b57cec5SDimitry Andric     // Spawn all but one of the threads in another thread as spawning threads
44*0b57cec5SDimitry Andric     // can take a while.
45480093f4SDimitry Andric     Threads.reserve(ThreadCount);
46480093f4SDimitry Andric     Threads.resize(1);
47480093f4SDimitry Andric     std::lock_guard<std::mutex> Lock(Mutex);
48480093f4SDimitry Andric     Threads[0] = std::thread([&, ThreadCount] {
49480093f4SDimitry Andric       for (unsigned i = 1; i < ThreadCount; ++i) {
50480093f4SDimitry Andric         Threads.emplace_back([=] { work(); });
51480093f4SDimitry Andric         if (Stop)
52480093f4SDimitry Andric           break;
53*0b57cec5SDimitry Andric       }
54480093f4SDimitry Andric       ThreadsCreated.set_value();
55*0b57cec5SDimitry Andric       work();
56480093f4SDimitry Andric     });
57480093f4SDimitry Andric   }
58480093f4SDimitry Andric 
59480093f4SDimitry Andric   void stop() {
60480093f4SDimitry Andric     {
61480093f4SDimitry Andric       std::lock_guard<std::mutex> Lock(Mutex);
62480093f4SDimitry Andric       if (Stop)
63480093f4SDimitry Andric         return;
64480093f4SDimitry Andric       Stop = true;
65480093f4SDimitry Andric     }
66480093f4SDimitry Andric     Cond.notify_all();
67480093f4SDimitry Andric     ThreadsCreated.get_future().wait();
68*0b57cec5SDimitry Andric   }
69*0b57cec5SDimitry Andric 
70*0b57cec5SDimitry Andric   ~ThreadPoolExecutor() override {
71480093f4SDimitry Andric     stop();
72480093f4SDimitry Andric     std::thread::id CurrentThreadId = std::this_thread::get_id();
73480093f4SDimitry Andric     for (std::thread &T : Threads)
74480093f4SDimitry Andric       if (T.get_id() == CurrentThreadId)
75480093f4SDimitry Andric         T.detach();
76480093f4SDimitry Andric       else
77480093f4SDimitry Andric         T.join();
78*0b57cec5SDimitry Andric   }
79*0b57cec5SDimitry Andric 
80480093f4SDimitry Andric   struct Deleter {
81480093f4SDimitry Andric     static void call(void *Ptr) { ((ThreadPoolExecutor *)Ptr)->stop(); }
82480093f4SDimitry Andric   };
83480093f4SDimitry Andric 
84*0b57cec5SDimitry Andric   void add(std::function<void()> F) override {
85480093f4SDimitry Andric     {
86480093f4SDimitry Andric       std::lock_guard<std::mutex> Lock(Mutex);
87*0b57cec5SDimitry Andric       WorkStack.push(F);
88480093f4SDimitry Andric     }
89*0b57cec5SDimitry Andric     Cond.notify_one();
90*0b57cec5SDimitry Andric   }
91*0b57cec5SDimitry Andric 
92*0b57cec5SDimitry Andric private:
93*0b57cec5SDimitry Andric   void work() {
94*0b57cec5SDimitry Andric     while (true) {
95*0b57cec5SDimitry Andric       std::unique_lock<std::mutex> Lock(Mutex);
96*0b57cec5SDimitry Andric       Cond.wait(Lock, [&] { return Stop || !WorkStack.empty(); });
97*0b57cec5SDimitry Andric       if (Stop)
98*0b57cec5SDimitry Andric         break;
99*0b57cec5SDimitry Andric       auto Task = WorkStack.top();
100*0b57cec5SDimitry Andric       WorkStack.pop();
101*0b57cec5SDimitry Andric       Lock.unlock();
102*0b57cec5SDimitry Andric       Task();
103*0b57cec5SDimitry Andric     }
104*0b57cec5SDimitry Andric   }
105*0b57cec5SDimitry Andric 
106*0b57cec5SDimitry Andric   std::atomic<bool> Stop{false};
107*0b57cec5SDimitry Andric   std::stack<std::function<void()>> WorkStack;
108*0b57cec5SDimitry Andric   std::mutex Mutex;
109*0b57cec5SDimitry Andric   std::condition_variable Cond;
110480093f4SDimitry Andric   std::promise<void> ThreadsCreated;
111480093f4SDimitry Andric   std::vector<std::thread> Threads;
112*0b57cec5SDimitry Andric };
113*0b57cec5SDimitry Andric 
114*0b57cec5SDimitry Andric Executor *Executor::getDefaultExecutor() {
115480093f4SDimitry Andric   // The ManagedStatic enables the ThreadPoolExecutor to be stopped via
116480093f4SDimitry Andric   // llvm_shutdown() which allows a "clean" fast exit, e.g. via _exit(). This
117480093f4SDimitry Andric   // stops the thread pool and waits for any worker thread creation to complete
118480093f4SDimitry Andric   // but does not wait for the threads to finish. The wait for worker thread
119480093f4SDimitry Andric   // creation to complete is important as it prevents intermittent crashes on
120480093f4SDimitry Andric   // Windows due to a race condition between thread creation and process exit.
121480093f4SDimitry Andric   //
122480093f4SDimitry Andric   // The ThreadPoolExecutor will only be destroyed when the static unique_ptr to
123480093f4SDimitry Andric   // it is destroyed, i.e. in a normal full exit. The ThreadPoolExecutor
124480093f4SDimitry Andric   // destructor ensures it has been stopped and waits for worker threads to
125480093f4SDimitry Andric   // finish. The wait is important as it prevents intermittent crashes on
126480093f4SDimitry Andric   // Windows when the process is doing a full exit.
127480093f4SDimitry Andric   //
128480093f4SDimitry Andric   // The Windows crashes appear to only occur with the MSVC static runtimes and
129480093f4SDimitry Andric   // are more frequent with the debug static runtime.
130480093f4SDimitry Andric   //
131480093f4SDimitry Andric   // This also prevents intermittent deadlocks on exit with the MinGW runtime.
132480093f4SDimitry Andric   static ManagedStatic<ThreadPoolExecutor, object_creator<ThreadPoolExecutor>,
133480093f4SDimitry Andric                        ThreadPoolExecutor::Deleter>
134480093f4SDimitry Andric       ManagedExec;
135480093f4SDimitry Andric   static std::unique_ptr<ThreadPoolExecutor> Exec(&(*ManagedExec));
136480093f4SDimitry Andric   return Exec.get();
137*0b57cec5SDimitry Andric }
1388bcb0991SDimitry Andric } // namespace
139*0b57cec5SDimitry Andric 
140*0b57cec5SDimitry Andric static std::atomic<int> TaskGroupInstances;
141*0b57cec5SDimitry Andric 
142*0b57cec5SDimitry Andric // Latch::sync() called by the dtor may cause one thread to block. If is a dead
143*0b57cec5SDimitry Andric // lock if all threads in the default executor are blocked. To prevent the dead
144*0b57cec5SDimitry Andric // lock, only allow the first TaskGroup to run tasks parallelly. In the scenario
145*0b57cec5SDimitry Andric // of nested parallel_for_each(), only the outermost one runs parallelly.
146*0b57cec5SDimitry Andric TaskGroup::TaskGroup() : Parallel(TaskGroupInstances++ == 0) {}
147*0b57cec5SDimitry Andric TaskGroup::~TaskGroup() { --TaskGroupInstances; }
148*0b57cec5SDimitry Andric 
149*0b57cec5SDimitry Andric void TaskGroup::spawn(std::function<void()> F) {
150*0b57cec5SDimitry Andric   if (Parallel) {
151*0b57cec5SDimitry Andric     L.inc();
152*0b57cec5SDimitry Andric     Executor::getDefaultExecutor()->add([&, F] {
153*0b57cec5SDimitry Andric       F();
154*0b57cec5SDimitry Andric       L.dec();
155*0b57cec5SDimitry Andric     });
156*0b57cec5SDimitry Andric   } else {
157*0b57cec5SDimitry Andric     F();
158*0b57cec5SDimitry Andric   }
159*0b57cec5SDimitry Andric }
160*0b57cec5SDimitry Andric 
161*0b57cec5SDimitry Andric } // namespace detail
162*0b57cec5SDimitry Andric } // namespace parallel
163*0b57cec5SDimitry Andric } // namespace llvm
164*0b57cec5SDimitry Andric #endif // LLVM_ENABLE_THREADS
165