contrib/llvm/lib/Support/Parallel.cpp

   1 //===- llvm/Support/Parallel.cpp - Parallel algorithms --------------------===//
   2 //
   3 //                     The LLVM Compiler Infrastructure
   4 //
   5 // This file is distributed under the University of Illinois Open Source
   6 // License. See LICENSE.TXT for details.
   7 //
   8 //===----------------------------------------------------------------------===//
   9
  10 #include "llvm/Support/Parallel.h"
  11 #include "llvm/Config/llvm-config.h"
  12 #include "llvm/Support/Threading.h"
  13
  14 #include <atomic>
  15 #include <stack>
  16 #include <thread>
  17
  18 using namespace llvm;
  19
  20 namespace {
  21
  22 /// \brief An abstract class that takes closures and runs them asynchronously.
  23 class Executor {
  24 public:
  25   virtual ~Executor() = default;
  26   virtual void add(std::function<void()> func) = 0;
  27
  28   static Executor *getDefaultExecutor();
  29 };
  30
  31 #if !LLVM_ENABLE_THREADS
  32 class SyncExecutor : public Executor {
  33 public:
  34   virtual void add(std::function<void()> F) { F(); }
  35 };
  36
  37 Executor *Executor::getDefaultExecutor() {
  38   static SyncExecutor Exec;
  39   return &Exec;
  40 }
  41
  42 #elif defined(_MSC_VER)
  43 /// \brief An Executor that runs tasks via ConcRT.
  44 class ConcRTExecutor : public Executor {
  45   struct Taskish {
  46     Taskish(std::function<void()> Task) : Task(Task) {}
  47
  48     std::function<void()> Task;
  49
  50     static void run(void *P) {
  51       Taskish *Self = static_cast<Taskish *>(P);
  52       Self->Task();
  53       concurrency::Free(Self);
  54     }
  55   };
  56
  57 public:
  58   virtual void add(std::function<void()> F) {
  59     Concurrency::CurrentScheduler::ScheduleTask(
  60         Taskish::run, new (concurrency::Alloc(sizeof(Taskish))) Taskish(F));
  61   }
  62 };
  63
  64 Executor *Executor::getDefaultExecutor() {
  65   static ConcRTExecutor exec;
  66   return &exec;
  67 }
  68
  69 #else
  70 /// \brief An implementation of an Executor that runs closures on a thread pool
  71 ///   in filo order.
  72 class ThreadPoolExecutor : public Executor {
  73 public:
  74   explicit ThreadPoolExecutor(unsigned ThreadCount = hardware_concurrency())
  75       : Done(ThreadCount) {
  76     // Spawn all but one of the threads in another thread as spawning threads
  77     // can take a while.
  78     std::thread([&, ThreadCount] {
  79       for (size_t i = 1; i < ThreadCount; ++i) {
  80         std::thread([=] { work(); }).detach();
  81       }
  82       work();
  83     }).detach();
  84   }
  85
  86   ~ThreadPoolExecutor() override {
  87     std::unique_lock<std::mutex> Lock(Mutex);
  88     Stop = true;
  89     Lock.unlock();
  90     Cond.notify_all();
  91     // Wait for ~Latch.
  92   }
  93
  94   void add(std::function<void()> F) override {
  95     std::unique_lock<std::mutex> Lock(Mutex);
  96     WorkStack.push(F);
  97     Lock.unlock();
  98     Cond.notify_one();
  99   }
 100
 101 private:
 102   void work() {
 103     while (true) {
 104       std::unique_lock<std::mutex> Lock(Mutex);
 105       Cond.wait(Lock, [&] { return Stop || !WorkStack.empty(); });
 106       if (Stop)
 107         break;
 108       auto Task = WorkStack.top();
 109       WorkStack.pop();
 110       Lock.unlock();
 111       Task();
 112     }
 113     Done.dec();
 114   }
 115
 116   std::atomic<bool> Stop{false};
 117   std::stack<std::function<void()>> WorkStack;
 118   std::mutex Mutex;
 119   std::condition_variable Cond;
 120   parallel::detail::Latch Done;
 121 };
 122
 123 Executor *Executor::getDefaultExecutor() {
 124   static ThreadPoolExecutor exec;
 125   return &exec;
 126 }
 127 #endif
 128 }
 129
 130 #if LLVM_ENABLE_THREADS
 131 void parallel::detail::TaskGroup::spawn(std::function<void()> F) {
 132   L.inc();
 133   Executor::getDefaultExecutor()->add([&, F] {
 134     F();
 135     L.dec();
 136   });
 137 }
 138 #endif