lib/Support/Parallel.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-exception
   6 //
   7 //===----------------------------------------------------------------------===//
   8
   9 #include "llvm/Support/Parallel.h"
  10 #include "llvm/Config/llvm-config.h"
  11
  12 #if LLVM_ENABLE_THREADS
  13
  14 #include "llvm/Support/Threading.h"
  15
  16 #include <atomic>
  17 #include <stack>
  18 #include <thread>
  19
  20 namespace llvm {
  21 namespace parallel {
  22 namespace detail {
  23
  24 namespace {
  25
  26 /// An abstract class that takes closures and runs them asynchronously.
  27 class Executor {
  28 public:
  29   virtual ~Executor() = default;
  30   virtual void add(std::function<void()> func) = 0;
  31
  32   static Executor *getDefaultExecutor();
  33 };
  34
  35 /// An implementation of an Executor that runs closures on a thread pool
  36 ///   in filo order.
  37 class ThreadPoolExecutor : public Executor {
  38 public:
  39   explicit ThreadPoolExecutor(unsigned ThreadCount = hardware_concurrency())
  40       : Done(ThreadCount) {
  41     // Spawn all but one of the threads in another thread as spawning threads
  42     // can take a while.
  43     std::thread([&, ThreadCount] {
  44       for (size_t i = 1; i < ThreadCount; ++i) {
  45         std::thread([=] { work(); }).detach();
  46       }
  47       work();
  48     }).detach();
  49   }
  50
  51   ~ThreadPoolExecutor() override {
  52     std::unique_lock<std::mutex> Lock(Mutex);
  53     Stop = true;
  54     Lock.unlock();
  55     Cond.notify_all();
  56     // Wait for ~Latch.
  57   }
  58
  59   void add(std::function<void()> F) override {
  60     std::unique_lock<std::mutex> Lock(Mutex);
  61     WorkStack.push(F);
  62     Lock.unlock();
  63     Cond.notify_one();
  64   }
  65
  66 private:
  67   void work() {
  68     while (true) {
  69       std::unique_lock<std::mutex> Lock(Mutex);
  70       Cond.wait(Lock, [&] { return Stop || !WorkStack.empty(); });
  71       if (Stop)
  72         break;
  73       auto Task = WorkStack.top();
  74       WorkStack.pop();
  75       Lock.unlock();
  76       Task();
  77     }
  78     Done.dec();
  79   }
  80
  81   std::atomic<bool> Stop{false};
  82   std::stack<std::function<void()>> WorkStack;
  83   std::mutex Mutex;
  84   std::condition_variable Cond;
  85   parallel::detail::Latch Done;
  86 };
  87
  88 Executor *Executor::getDefaultExecutor() {
  89   static ThreadPoolExecutor exec;
  90   return &exec;
  91 }
  92 } // namespace
  93
  94 static std::atomic<int> TaskGroupInstances;
  95
  96 // Latch::sync() called by the dtor may cause one thread to block. If is a dead
  97 // lock if all threads in the default executor are blocked. To prevent the dead
  98 // lock, only allow the first TaskGroup to run tasks parallelly. In the scenario
  99 // of nested parallel_for_each(), only the outermost one runs parallelly.
 100 TaskGroup::TaskGroup() : Parallel(TaskGroupInstances++ == 0) {}
 101 TaskGroup::~TaskGroup() { --TaskGroupInstances; }
 102
 103 void TaskGroup::spawn(std::function<void()> F) {
 104   if (Parallel) {
 105     L.inc();
 106     Executor::getDefaultExecutor()->add([&, F] {
 107       F();
 108       L.dec();
 109     });
 110   } else {
 111     F();
 112   }
 113 }
 114
 115 } // namespace detail
 116 } // namespace parallel
 117 } // namespace llvm
 118 #endif // LLVM_ENABLE_THREADS