[ThreadPool] Simplify the interface

[davide]

Summary

The returned std::future is unused.

[mehdi_amini]

What is the motivation? This is removing a feature. Even though it is not used at that time, it is tested and could find its use in the future.

[davide]

In D27159#607314, @mehdi_amini wrote:

What is the motivation? This is removing a feature. Even though it is not used at that time, it is tested and could find its use in the future.

This is pretty much dead code. If you're going to find a consumer using this feature, you can add it again. I plan to merge the lld and the llvm thread pool interface and this removes the dependence on <future> (and makes the merge slightly simpler).

[mehdi_amini]

Why is the dependency on future an issue?

[davide]

In D27159#607319, @mehdi_amini wrote:

Why is the dependency on future an issue?

Because it complicates the interface, for no reason (atm, as nobody uses it).
If you prefer the code to stay as is, fine, but we will keep having two different implementations in-tree.

[mehdi_amini]

If you prefer the code to stay as is, fine, but we will keep having two different implementations in-tree.

You're not explaining why it prevents the merge.

[mehdi_amini]

Maybe submitting another patch for the merge, and marking this one as blocking could help motivating this patch.

[davide]

In D27159#607321, @mehdi_amini wrote:

If you prefer the code to stay as is, fine, but we will keep having two different implementations in-tree.

You're not explaining why it prevents the merge.

It makes it harder for me. If it doesn't for you, feel free to go ahead. In general, I shouldn't explain why I want to remove dead code from the tree (as it's dead).

[mehdi_amini]

unittests/Support/ThreadPool.cpp

102

You need to make sure this works with LLVM_ENABLE_THREADS=OFF

[davide]

[49/49] Running the LLVM regression tests
Testing Time: 259.34s
  Expected Passes    : 18270
  Expected Failures  : 121
  Unsupported Tests  : 386

$ grep ENABLE_THREADS CMakeCache.txt
LIBCXXABI_ENABLE_THREADS:BOOL=OFF
//\n   This option may only be set to OFF when LIBCXX_ENABLE_THREADS=OFF.
LIBCXX_ENABLE_THREADS:BOOL=OFF
LIBCXX_ENABLE_THREAD_UNSAFE_C_FUNCTIONS:BOOL=ON
//\n   This option may only be set to ON when LIBCXX_ENABLE_THREADS=ON.
LLVM_ENABLE_THREADS:BOOL=OFF

[davide]

[49/49] Running the LLVM regression tests
Testing Time: 259.34s
  Expected Passes    : 18270
  Expected Failures  : 121
  Unsupported Tests  : 386

$ grep ENABLE_THREADS CMakeCache.txt
LIBCXXABI_ENABLE_THREADS:BOOL=OFF
//\n   This option may only be set to OFF when LIBCXX_ENABLE_THREADS=OFF.
LIBCXX_ENABLE_THREADS:BOOL=OFF
LIBCXX_ENABLE_THREAD_UNSAFE_C_FUNCTIONS:BOOL=ON
//\n   This option may only be set to ON when LIBCXX_ENABLE_THREADS=ON.
LLVM_ENABLE_THREADS:BOOL=OFF

unittests/Support/ThreadPool.cpp

102

I tried to run the testsuite with ENABLE_THREADS=OFF. Is there any other test you recommend to run or is this fine?

[davide]

So, @Bigcheese / @mehdi_amini , can this get in?

[davide]

Put this on hold until we have a new API. it's probably easier to replace it directly when @Bigcheese will have the new implementation in.

include/llvm/Support/ThreadPool.h

 //
 //===----------------------------------------------------------------------===//
 
 #ifndef LLVM_SUPPORT_THREAD_POOL_H
 #define LLVM_SUPPORT_THREAD_POOL_H
 
 #include "llvm/Support/thread.h"
 
-#ifdef _MSC_VER
-// concrt.h depends on eh.h for __uncaught_exception declaration
-// even if we disable exceptions.
-#include <eh.h>
-
-// Disable warnings from ppltasks.h transitively included by <future>.
-#pragma warning(push)
-#pragma warning(disable:4530)
-#pragma warning(disable:4062)
-#endif
-
-#include <future>
-
-#ifdef _MSC_VER
-#pragma warning(pop)
-#endif
-
 #include <atomic>
 #include <condition_variable>
 #include <functional>
 #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:
-#ifndef _MSC_VER
-  using VoidTy = void;
   using TaskTy = std::function<void()>;
-  using PackagedTaskTy = std::packaged_task<void()>;
-#else
-  // MSVC 2013 has a bug and can't use std::packaged_task<void()>;
-  // We force it to use bool(bool) instead.
-  using VoidTy = bool;
-  using TaskTy = std::function<bool(bool)>;
-  using PackagedTaskTy = std::packaged_task<bool(bool)>;
-#endif
 
   /// 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.
+  /// Asynchronous submission of a task to the pool.
   template <typename Function, typename... Args>
-  inline std::shared_future<VoidTy> async(Function &&F, Args &&... ArgList) {
+  inline void async(Function &&F, Args &&... ArgList) {
     auto Task =
         std::bind(std::forward<Function>(F), std::forward<Args>(ArgList)...);
-#ifndef _MSC_VER
-    return asyncImpl(std::move(Task));
-#else
-    // This lambda has to be marked mutable because MSVC 2013's std::bind call
-    // operator isn't const qualified.
-    return asyncImpl([Task](VoidTy) mutable -> VoidTy {
-      Task();
-      return VoidTy();
-    });
-#endif
+    asyncImpl(std::move(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.
+  /// Asynchronous submission of a task to the pool.
   template <typename Function>
-  inline std::shared_future<VoidTy> async(Function &&F) {
-#ifndef _MSC_VER
-    return asyncImpl(std::forward<Function>(F));
-#else
-    return asyncImpl([F] (VoidTy) -> VoidTy { F(); return VoidTy(); });
-#endif
-  }
+  inline void async(Function &&F) { asyncImpl([F] () { 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<VoidTy> asyncImpl(TaskTy F);
+  /// Asynchronous submission of a task to the pool.
+  void asyncImpl(TaskTy F);
 
   /// Threads in flight
   std::vector<llvm::thread> Threads;
 
   /// Tasks waiting for execution in the pool.
-  std::queue<PackagedTaskTy> Tasks;
+  std::queue<TaskTy> Tasks;
 
   /// Locking and signaling for accessing the Tasks queue.
   std::mutex QueueLock;
   std::condition_variable QueueCondition;
 
   /// Locking and signaling for job completion
   std::mutex CompletionLock;
   std::condition_variable CompletionCondition;

lib/Support/ThreadPool.cpp

 ThreadPool::ThreadPool(unsigned ThreadCount)
     : ActiveThreads(0), EnableFlag(true) {
   // Create ThreadCount threads that will loop forever, wait on QueueCondition
   // for tasks to be queued or the Pool to be destroyed.
   Threads.reserve(ThreadCount);
   for (unsigned ThreadID = 0; ThreadID < ThreadCount; ++ThreadID) {
     Threads.emplace_back([&] {
       while (true) {
-        PackagedTaskTy Task;
+        TaskTy 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
 
           // We first need to signal that we are active before popping the queue
           // in order for wait() to properly detect that even if the queue is
           // empty, there is still a task in flight.
           {
             ++ActiveThreads;
             std::unique_lock<std::mutex> LockGuard(CompletionLock);
           }
           Task = std::move(Tasks.front());
           Tasks.pop();
         }
         // Run the task we just grabbed
-#ifndef _MSC_VER
         Task();
-#else
-        Task(/* unused */ false);
-#endif
 
         {
           // Adjust `ActiveThreads`, in case someone waits on ThreadPool::wait()
           std::unique_lock<std::mutex> LockGuard(CompletionLock);
           --ActiveThreads;
         }
 
         // Notify task completion, in case someone waits on ThreadPool::wait()
         CompletionCondition.notify_all();
       }
     });
   }
 }
 
 void ThreadPool::wait() {
   // Wait for all threads to complete and the queue to be empty
   std::unique_lock<std::mutex> LockGuard(CompletionLock);
   // The order of the checks for ActiveThreads and Tasks.empty() matters because
   // any active threads might be modifying the Tasks queue, and this would be a
   // race.
   CompletionCondition.wait(LockGuard,
                            [&] { return !ActiveThreads && Tasks.empty(); });
 }
 
-std::shared_future<ThreadPool::VoidTy> ThreadPool::asyncImpl(TaskTy Task) {
-  /// Wrap the Task in a packaged_task to return a future object.
-  PackagedTaskTy PackagedTask(std::move(Task));
-  auto Future = PackagedTask.get_future();
+void ThreadPool::asyncImpl(TaskTy Task) {
   {
     // 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));
+    Tasks.push(std::move(Task));
   }
   QueueCondition.notify_one();
-  return Future.share();
 }
 
 // The destructor joins all threads, waiting for completion.
 ThreadPool::~ThreadPool() {
   {
     std::unique_lock<std::mutex> LockGuard(QueueLock);
     EnableFlag = false;
   }
ThreadPool::ThreadPool(unsigned ThreadCount)
   }
 }
 
 void ThreadPool::wait() {
   // Sequential implementation running the tasks
   while (!Tasks.empty()) {
     auto Task = std::move(Tasks.front());
     Tasks.pop();
-#ifndef _MSC_VER
-        Task();
-#else
-        Task(/* unused */ false);
-#endif
   }
 }
 
-std::shared_future<ThreadPool::VoidTy> ThreadPool::asyncImpl(TaskTy Task) {
-#ifndef _MSC_VER
-  // 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 ThreadPool::wait() can operate and the
-  // returned future can be sync'ed on.
-  PackagedTaskTy PackagedTask([Future]() { Future.get(); });
-#else
-  auto Future = std::async(std::launch::deferred, std::move(Task), false).share();
-  PackagedTaskTy PackagedTask([Future](bool) -> bool { Future.get(); return false; });
-#endif
-  Tasks.push(std::move(PackagedTask));
-  return Future;
+void ThreadPool::asyncImpl(TaskTy Task) {
+  Tasks.push(std::move(Task));
 }
 
 ThreadPool::~ThreadPool() {
   wait();
 }
 
 #endif

unittests/Support/ThreadPool.cpp

Pool.async([this, &checked_in, i] {
       waitForMainThread();
       ++checked_in;
     });
   }
   ASSERT_EQ(0, checked_in);
   setMainThreadReady();
   Pool.wait();
   ASSERT_EQ(5, checked_in);
 }
 
 static void TestFunc(std::atomic_int &checked_in, int i) { checked_in += i; }
 
 TEST_F(ThreadPoolTest, AsyncBarrierArgs) {
   CHECK_UNSUPPORTED();
   // Test that async works with a function requiring multiple parameters.
   std::atomic_int checked_in{0};
 
TEST_F(ThreadPoolTest, Async) {
   });
   Pool.async([&i] { ++i; });
   ASSERT_NE(2, i.load());
   setMainThreadReady();
   Pool.wait();
   ASSERT_EQ(2, i.load());
 }
 
-TEST_F(ThreadPoolTest, GetFuture) {
-  CHECK_UNSUPPORTED();
-  ThreadPool Pool{2};
-  std::atomic_int i{0};
-  Pool.async([this, &i] {
-    waitForMainThread();
-    ++i;
-  });
-  // Force the future using get()
-  Pool.async([&i] { ++i; }).get();
-  ASSERT_NE(2, i.load());
-  setMainThreadReady();
-  Pool.wait();
-  ASSERT_EQ(2, i.load());
-}
-
 TEST_F(ThreadPoolTest, PoolDestruction) {
   CHECK_UNSUPPORTED();
   // 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([this, &checked_in, i] {