Add a C++11 ThreadPool implementation in LLVM

[mehdi_amini]

Summary

This is a very simple implementation of a thread pool using C++11
thread. It accepts any std::function<void()> for asynchronous
execution. Individual task can be synchronize using the returned
future, or the client can block on the full queue completion.

In case LLVM is configured with Threading disabled, it falls back
to sequential execution using std::async with launch:deferred.

This is intended to support parallelism for ThinLTO processing in
linker plugin, but is generic enough to be for any other uses.

[mehdi_amini]

Fix some comments

[tejohnson]

Thanks for contributing this. A few questions and a few minor nits.

include/llvm/Support/thread.h

46

Is this change needed? No other change here.

lib/Support/ThreadPool.cpp

45

Might want to note that the order of operations is important here. The increment of ActiveThreads needs to happen before the pop() or the CompletionCondition in wait() might trigger early.

49

Typo: grabbed

51

Do we need to get a lock before modifying ActiveThreads (and use the same lock to guard the wait on CompletionCondition further down? According to http://en.cppreference.com/w/cpp/thread/condition_variable:

Even if the shared variable is atomic, it must be modified under the mutex in order to correctly publish the modification to the waiting thread.

53

nit: ThreadPool::wait()

87

Do we need to get QueueLock before modifying EnableFlag? Same comment as earlier about ActiveThread (According to http://en.cppreference.com/w/cpp/thread/condition_variable:

Even if the shared variable is atomic, it must be modified under the mutex in order to correctly publish the modification to the waiting thread.

)

118

nit: ThreadPool::wait()

unittests/Support/ThreadPool.cpp

32

This is expected to be 0 because the sleep in the threads is expected to delay their execution sufficiently? It seems like that's how all of the tests work. I wonder if that will always be enough to guarantee that the first assert executes before the first thread increments. I.e could the main thread ever get delayed under some circumstance, and a child thread update first? Wondering if the pre-wait() asserts are really necessary.

Why are some of the sleeps for 50ms and some for 500ms?

[mehdi_amini]

Update following Teresa's comments.

[mehdi_amini]

Thanks for the review!

include/llvm/Support/thread.h

46

Yes it is needed for std::forward, the unittests can't be compiled with LLVM_ENABLE_THREADS=OFF without this.

lib/Support/ThreadPool.cpp

45

done

49

oops

51

I tried to be too clever :(

Added another lock

53

done

unittests/Support/ThreadPool.cpp

32

Yes, with LLVM_ENABLE_THREADS=OFF it is deterministic, but not with real threads.
I tried to improve it using added std::this_thread::yield(), even if it still theoretically possible to break the test, it should be unlikely enough to not happen in practice.

[tejohnson]

Thanks! Will change my gold-plugin patch to use the thread pool once this is in.

lib/Support/ThreadPool.cpp

46

Nit: typo "popping"

140

Is this just for completeness? It doesn't look like this flag is used when ! LLVM_ENABLE_THREADS.

unittests/Support/ThreadPool.cpp

33

Ok, that seems reasonable. We can reevaluate if the test behaves flakily.

[mehdi_amini]

r255444

lib/Support/ThreadPool.cpp

140

It is to silent a compiler warning about "member variable not used" when compiling with LLVM_ENABLE_THREAD=OFF

[tejohnson]

Here's the fix I mentioned on IRC:

diff --git a/include/llvm/Support/ThreadPool.h b/include/llvm/Support/ThreadPool.h
index c388c07..4b29392 100644

• a/include/llvm/Support/ThreadPool.h

+++ b/include/llvm/Support/ThreadPool.h
@@ -50,7 +50,7 @@ public:

template <typename Function, typename... Args>
inline std::shared_future<void> async(Function &&F, Args &&... ArgList) {
  auto Task =

• std::bind(std::forward<Function>(F), std::forward<Args...>(ArgList...));

+ std::bind(std::forward<Function>(F), std::forward<Args>(ArgList)...);

  return asyncImpl(Task);
}

diff --git a/unittests/Support/ThreadPool.cpp b/unittests/Support/ThreadPool.cpp
index d36341e..5457cdc 100644

• a/unittests/Support/ThreadPool.cpp

+++ b/unittests/Support/ThreadPool.cpp
@@ -44,6 +44,20 @@ TEST(ThreadPoolTest, AsyncBarrier) {

ASSERT_EQ(5, checked_in);

}

+static void TestFunc(std::atomic_int &checked_in, int i) { checked_in += i; }
+
+TEST(ThreadPoolTest, AsyncBarrierArgs) {
+ // Test that async works with a function requiring multiple parameters.
+ std::atomic_int checked_in{0};
+
+ ThreadPool Pool;
+ for (size_t i = 0; i < 5; ++i) {
+ Pool.async(TestFunc, std::ref(checked_in), i);
+ }
+ Pool.wait();
+ ASSERT_EQ(10, checked_in);
+}
+
TEST(ThreadPoolTest, Async) {

ThreadPool Pool;

[mehdi_amini]

recommitted r255593, trying to please MSVC.

Teresa: your patch LGTM, feel free to commit on top of it.

[loladiro]

This breaks the LLVM build with mingw-w64, which doesn't have C++11 threading support as far as I know. It would be nice if the threading-disabled version didn't use any of that stuff (mutexes, futures, etc) such that LLVM can be built with mingw-w64.

[mehdi_amini]

There is no inclusion of <thread> when threading is disabled right now. However #include <future> is necessary even without threading.
If some toolchains can't provide std::future and std::async (with std::launch::deferred) then someone needs to provide a patch to implement a include/llvm/Support/future.h on the model of include/llvm/Support/future.h. I'll be happy to review.

include/llvm/Support/ThreadPool.h

+//===-- llvm/Support/ThreadPool.h - A ThreadPool implementation -*- C++ -*-===//
+//
+//                     The LLVM Compiler Infrastructure
+//
+// This file is distributed under the University of Illinois Open Source
+// License. See LICENSE.TXT for details.
+//
+//===----------------------------------------------------------------------===//
+//
+// This file defines a crude C++11 based thread pool.
+//
+//===----------------------------------------------------------------------===//
+
+#ifndef LLVM_SUPPORT_THREAD_POOL_H
+#define LLVM_SUPPORT_THREAD_POOL_H
+
+#include "llvm/Support/thread.h"
+
+#include <condition_variable>
+#include <functional>
+#include <future>
+#include <memory>
+#include <mutex>
+#include <queue>
+#include <utility>
+
+namespace llvm {
+
+/// A ThreadPool for asynchronous parallel execution on a defined number of
+/// threads.
+///
+/// The pool keeps a vector of threads alive, waiting on a condition variable
+/// for some work to become available.
+class ThreadPool {
+public:
+  using TaskTy = std::function<void()>;
+
+  /// Construct a pool with the number of core available on the system (or
+  /// whatever the value returned by std::thread::hardware_concurrency() is).
+  ThreadPool();
+
+  /// Construct a pool of \p ThreadCount threads
+  ThreadPool(unsigned ThreadCount);
+
+  /// Blocking destructor: the pool will wait for all the threads to complete.
+  ~ThreadPool();
+
+  /// Asynchronous submission of a task to the pool. The returned future can be
+  /// used to wait for the task to finish and is *non-blocking* on destruction.
+  template <typename Function, typename... Args>
+  inline std::shared_future<void> async(Function &&F, Args &&... ArgList) {
+    auto Task =
+        std::bind(std::forward<Function>(F), std::forward<Args...>(ArgList...));
+    return asyncImpl(Task);
+  }
+
+  /// Asynchronous submission of a task to the pool. The returned future can be
+  /// used to wait for the task to finish and is *non-blocking* on destruction.
+  template <typename Function>
+  inline std::shared_future<void> async(Function &&F) {
+    return asyncImpl(F);
+  }
+
+  /// Blocking wait for all the threads to complete and the queue to be empty.
+  /// It is an error to try to add new tasks while blocking on this call.
+  void wait();
+
+private:
+  /// Asynchronous submission of a task to the pool. The returned future can be
+  /// used to wait for the task to finish and is *non-blocking* on destruction.
+  std::shared_future<void> asyncImpl(TaskTy f);
+
+  /// Threads in flight
+  std::vector<llvm::thread> Threads;
+
+  /// Tasks waiting for execution in the pool.
+  std::queue<std::packaged_task<void()>> Tasks;
+
+  /// Locking and signaling for accessing the Tasks queue.
+  std::mutex QueueLock;
+  std::condition_variable QueueCondition;
+
+  /// Signaling for job completion
+  std::condition_variable CompletionCondition;
+
+  /// Keep track of the number of thread actually busy
+  std::atomic<unsigned> ActiveThreads;
+
+  /// Signal for the destruction of the pool, asking thread to exit.
+  std::atomic<bool> EnableFlag;
+};
+}
+
+#endif // LLVM_SUPPORT_THREAD_POOL_H

include/llvm/Support/thread.h

 #endif
 
 namespace llvm {
 typedef std::thread thread;
 }
 
 #else // !LLVM_ENABLE_THREADS
 
+#include <utility>
+
 namespace llvm {
 
 struct thread {
   thread() {}
   thread(thread &&other) {}
   template <class Function, class... Args>
   explicit thread(Function &&f, Args &&... args) {
     f(std::forward<Args>(args)...);

lib/Support/CMakeLists.txt

add_llvm_library(LLVMSupport
   StreamingMemoryObject.cpp
   StringExtras.cpp
   StringMap.cpp
   StringPool.cpp
   StringSaver.cpp
   StringRef.cpp
   SystemUtils.cpp
   TargetParser.cpp
+  ThreadPool.cpp
   Timer.cpp
   ToolOutputFile.cpp
   Triple.cpp
   Twine.cpp
   Unicode.cpp
   YAMLParser.cpp
   YAMLTraits.cpp
   raw_os_ostream.cpp

lib/Support/ThreadPool.cpp

+//==-- llvm/Support/ThreadPool.cpp - A ThreadPool implementation -*- C++ -*-==//
+//
+//                     The LLVM Compiler Infrastructure
+//
+// This file is distributed under the University of Illinois Open Source
+// License. See LICENSE.TXT for details.
+//
+//===----------------------------------------------------------------------===//
+//
+// This file implements a crude C++11 based thread pool.
+//
+//===----------------------------------------------------------------------===//
+
+#include "llvm/Support/ThreadPool.h"
+
+#include "llvm/Config/llvm-config.h"
+#include "llvm/Support/raw_ostream.h"
+
+using namespace llvm;
+
+#if LLVM_ENABLE_THREADS
+
+// Default to std::thread::hardware_concurrency
+ThreadPool::ThreadPool() : ThreadPool(std::thread::hardware_concurrency()) {}
+
+ThreadPool::ThreadPool(unsigned ThreadCount)
+    : ActiveThreads(0), EnableFlag(true) {
+  // Create ThreadCount threads that will spin waiting for tasks to be queued
+  Threads.reserve(ThreadCount);
+  for (unsigned ThreadID = 0; ThreadID < ThreadCount; ++ThreadID) {
+    Threads.emplace_back([&] {
+      while (true) {
+        std::packaged_task<void()> Task;
+        {
+          std::unique_lock<std::mutex> LockGuard(QueueLock);
+          // Wait for tasks to be pushed in the queue
+          QueueCondition.wait(LockGuard,
+                              [&] { return !EnableFlag || !Tasks.empty(); });
+          // Exit condition
+          if (!EnableFlag && Tasks.empty())
+            return;
+          // Yeah, we have a task, grab it and release the lock on the queue
+          ++ActiveThreads;
+          Task = std::move(Tasks.front());
+          Tasks.pop();
+        }
+        // Run the task we just grabed
+        Task();
+        --ActiveThreads;
+
+        // Notify task completion, in case someone waits on Pool::wait()
+        CompletionCondition.notify_all();
+      }
+    });
+  }
+}
+
+void ThreadPool::wait() {
+  // Wait for all thread to complete and the queue to be empty
+  std::mutex CompletionLock;
+  std::unique_lock<std::mutex> LockGuard(CompletionLock);
+  CompletionCondition.wait(LockGuard,
+                           [&] { return Tasks.empty() && !ActiveThreads; });
+}
+
+std::shared_future<void> ThreadPool::asyncImpl(TaskTy Task) {
+  /// Wrap the Task in a packaged_task to return a future object.
+  std::packaged_task<void()> PackagedTask(std::move(Task));
+  std::future<void> Future = PackagedTask.get_future();
+  {
+    // Lock the queue and push the new task
+    std::unique_lock<std::mutex> LockGuard(QueueLock);
+
+    // Don't allow enqueueing after disabling the pool
+    assert(EnableFlag && "Queuing a thread during ThreadPool destruction");
+
+    Tasks.push(std::move(PackagedTask));
+  }
+  QueueCondition.notify_one();
+  return Future.share();
+}
+
+// the destructor joins all threads
+ThreadPool::~ThreadPool() {
+  EnableFlag = false;
+  QueueCondition.notify_all();
+  for (auto &Worker : Threads)
+    Worker.join();
+}
+
+#else // LLVM_ENABLE_THREADS Disabled
+
+ThreadPool::ThreadPool() : ThreadPool(0) {}
+
+// the constructor just launches some amount of Threads
+ThreadPool::ThreadPool(unsigned ThreadCount)
+    : ActiveThreads(0), EnableFlag(true) {
+  if (ThreadCount) {
+    errs() << "Warning: request a ThreadPool with " << ThreadCount
+           << " threads, but LLVM_ENABLE_THREADS has been turned off\n";
+  }
+}
+
+void ThreadPool::wait() {
+  // Sequential implementation running the tasks
+  while (!Tasks.empty()) {
+    auto Task = std::move(Tasks.front());
+    Tasks.pop();
+    Task();
+  }
+}
+
+std::shared_future<void> ThreadPool::asyncImpl(TaskTy Task) {
+  // Get a Future with launch::deferred execution using std::async
+  auto Future = std::async(std::launch::deferred, std::move(Task)).share();
+  // Wrap the future so that both Pool::wait() can operate and the returned
+  // future can be sync'ed on.
+  std::packaged_task<void()> PackagedTask([Future]() { Future.get(); });
+  Tasks.push(std::move(PackagedTask));
+  return Future;
+}
+
+ThreadPool::~ThreadPool() { wait(); }
+
+#endif

unittests/Support/CMakeLists.txt

add_llvm_unittest(SupportTests
   ScaledNumberTest.cpp
   SourceMgrTest.cpp
   SpecialCaseListTest.cpp
   StreamingMemoryObject.cpp
   StringPool.cpp
   SwapByteOrderTest.cpp
   TargetRegistry.cpp
   ThreadLocalTest.cpp
+  ThreadPool.cpp
   TimeValueTest.cpp
   TrailingObjectsTest.cpp
   UnicodeTest.cpp
   YAMLIOTest.cpp
   YAMLParserTest.cpp
   formatted_raw_ostream_test.cpp
   raw_ostream_test.cpp
   raw_pwrite_stream_test.cpp
   )
 
 # ManagedStatic.cpp uses <pthread>.
 if(LLVM_ENABLE_THREADS AND HAVE_LIBPTHREAD)
   target_link_libraries(SupportTests pthread)
 endif()

unittests/Support/ThreadPool.cpp

+//========- unittests/Support/ThreadPools.cpp - ThreadPools.h tests --========//
+//
+//                     The LLVM Compiler Infrastructure
+//
+// This file is distributed under the University of Illinois Open Source
+// License. See LICENSE.TXT for details.
+//
+//===----------------------------------------------------------------------===//
+
+#include "llvm/Support/ThreadPool.h"
+
+#include "llvm/ADT/STLExtras.h"
+
+#include "gtest/gtest.h"
+
+using namespace llvm;
+using namespace std::chrono;
+
+TEST(ThreadPoolTest, AsyncBarrier) {
+  // test that async & barrier work together properly.
+
+  std::atomic_int checked_in{0};
+
+  ThreadPool Pool;
+  for (size_t i = 0; i < 5; ++i) {
+    Pool.async([&checked_in, i] {
+      std::this_thread::sleep_for(milliseconds(500));
+      ++checked_in;
+    });
+  }
+  ASSERT_EQ(0, checked_in);
+  Pool.wait();
+  ASSERT_EQ(5, checked_in);
+}
+
+TEST(ThreadPoolTest, Async) {
+  ThreadPool Pool;
+  std::atomic_int i{0};
+  // sleep here just to ensure that the not-equal is correct.
+  Pool.async([&i] {
+    std::this_thread::sleep_for(milliseconds(50));
+    ++i;
+  });
+  Pool.async([&i] { ++i; });
+  ASSERT_NE(2, i.load());
+  Pool.wait();
+  ASSERT_EQ(2, i.load());
+}
+
+TEST(ThreadPoolTest, GetFuture) {
+  ThreadPool Pool;
+  std::atomic_int i{0};
+  // sleep here just to ensure that the not-equal is correct.
+  Pool.async([&i] {
+    std::this_thread::sleep_for(milliseconds(50));
+    ++i;
+  });
+  // Force the future using get()
+  Pool.async([&i] { ++i; }).get();
+  ASSERT_NE(2, i.load());
+  Pool.wait();
+  ASSERT_EQ(2, i.load());
+}
+
+TEST(ThreadPoolTest, PoolDestruction) {
+  // Test that we are waiting on destruction
+  std::atomic_int checked_in{0};
+
+  {
+    ThreadPool Pool;
+    for (size_t i = 0; i < 5; ++i) {
+      Pool.async([&checked_in, i] {
+        std::this_thread::sleep_for(milliseconds(500));
+        ++checked_in;
+      });
+    }
+    ASSERT_EQ(0, checked_in);
+  }
+  ASSERT_EQ(5, checked_in);
+}