Split the locking of the queue and the threads vector in the ThreadPool implementation

[mehdi_amini]

Summary

This allows to release the QueueLock early and create Thread
independently of the queue processing.

[mehdi_amini]

Minor fixes after testing

[mehdi_amini]

Replace max with min, obviously...

[Benoit]

llvm/include/llvm/Support/ThreadPool.h

44

is the performance benefit of avoiding some small mallocs worth the line of code?

llvm/lib/Support/ThreadPool.cpp

29–30

how about using a new variable instead of mutating requested?

[mehdi_amini]

llvm/include/llvm/Support/ThreadPool.h

44

Probably not, happy to leave it out, it does not really belong to this patch anyway I think.

llvm/include/llvm/Support/ThreadPool.h

 /// for some work to become available.
 class ThreadPool {
 public:
   /// Construct a pool using the hardware strategy \p S for mapping hardware
   /// execution resources (threads, cores, CPUs)
   /// Defaults to using the maximum execution resources in the system, but
   /// accounting for the affinity mask.
   ThreadPool(ThreadPoolStrategy S = hardware_concurrency());
 
   /// 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 auto async(Function &&F, Args &&...ArgList) {
     auto Task =
private:
   template <typename ResTy>
   std::shared_future<ResTy> asyncImpl(std::function<ResTy()> Task) {
 
 #if LLVM_ENABLE_THREADS
     /// Wrap the Task in a std::function<void()> that sets the result of the
     /// corresponding future.
     auto R = createTaskAndFuture(Task);
 
+    int requestedThreads;
     {
       // 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(R.first));
-      grow();
+      requestedThreads = ActiveThreads + Tasks.size();
     }
     QueueCondition.notify_one();
+    grow(requestedThreads);
     return R.second.share();
 
 #else // LLVM_ENABLE_THREADS Disabled
 
     // 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.
     Tasks.push([Future]() { Future.get(); });
     return Future;
 #endif
   }
 
 #if LLVM_ENABLE_THREADS
-  // Maybe create a new thread and add it to Threads.
-  //
-  // Requirements:
-  //   * this->QueueLock should be owned by the calling thread prior to
-  //     calling this function. It will neither lock it nor unlock it.
-  //     Calling this function without owning QueueLock would result in data
-  //     races as this function reads Tasks and ActiveThreads.
-  //   * this->Tasks should be populated with any pending tasks. This function
-  //     uses Tasks.size() to determine whether it needs to create a new thread.
-  //   * this->ActiveThreads should be up to date as it is also used to
-  //     determine whether to create a new thread.
-  void grow();
+  // Grow to ensure that we have at least `requested` Threads, but do not go
+  // over MaxThreadCount.
+  void grow(int requested);
 #endif
 
   /// Threads in flight
   std::vector<llvm::thread> Threads;
+  /// Lock protecting access to the Threads vector.
+  mutable std::mutex ThreadsLock;
 
   /// Tasks waiting for execution in the pool.
   std::queue<std::function<void()>> Tasks;
 
   /// Locking and signaling for accessing the Tasks queue.
-  mutable std::mutex QueueLock;
+  std::mutex QueueLock;
   std::condition_variable QueueCondition;
 
   /// Signaling for job completion
   std::condition_variable CompletionCondition;
 
   /// Keep track of the number of thread actually busy
   unsigned ActiveThreads = 0;
 

llvm/lib/Support/ThreadPool.cpp

 
 using namespace llvm;
 
 #if LLVM_ENABLE_THREADS
 
 ThreadPool::ThreadPool(ThreadPoolStrategy S)
     : Strategy(S), MaxThreadCount(S.compute_thread_count()) {}
 
-void ThreadPool::grow() {
+void ThreadPool::grow(int requested) {
+  std::unique_lock<std::mutex> LockGuard(ThreadsLock);
   if (Threads.size() >= MaxThreadCount)
     return; // Already hit the max thread pool size.
-  if (ActiveThreads + Tasks.size() <= Threads.size())
-    return; // We have enough threads for now.
+  int newThreadCount = std::min<int>(requested, MaxThreadCount);
+  while (static_cast<int>(Threads.size()) < newThreadCount) {
     int ThreadID = Threads.size();
     Threads.emplace_back([this, ThreadID] {
       Strategy.apply_thread_strategy(ThreadID);
       while (true) {
         std::function<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
 
           // 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;
           Task = std::move(Tasks.front());
           Tasks.pop();
         }
         // Run the task we just grabbed
         Task();
 
         bool Notify;
         {
           // Adjust `ActiveThreads`, in case someone waits on ThreadPool::wait()
           std::lock_guard<std::mutex> LockGuard(QueueLock);
           --ActiveThreads;
           Notify = workCompletedUnlocked();
         }
         // Notify task completion if this is the last active thread, in case
         // someone waits on ThreadPool::wait().
         if (Notify)
           CompletionCondition.notify_all();
       }
     });
   }
+}
 
 void ThreadPool::wait() {
   // Wait for all threads to complete and the queue to be empty
   std::unique_lock<std::mutex> LockGuard(QueueLock);
   CompletionCondition.wait(LockGuard, [&] { return workCompletedUnlocked(); });
 }
 
 bool ThreadPool::isWorkerThread() const {
-  std::unique_lock<std::mutex> LockGuard(QueueLock);
+  std::unique_lock<std::mutex> LockGuard(ThreadsLock);
   llvm::thread::id CurrentThreadId = llvm::this_thread::get_id();
   for (const llvm::thread &Thread : Threads)
     if (CurrentThreadId == Thread.get_id())
       return true;
   return false;
 }
 
 // The destructor joins all threads, waiting for completion.
 ThreadPool::~ThreadPool() {
   {
     std::unique_lock<std::mutex> LockGuard(QueueLock);
     EnableFlag = false;
   }
   QueueCondition.notify_all();
+  std::unique_lock<std::mutex> LockGuard(ThreadsLock);
   for (auto &Worker : Threads)
     Worker.join();
 }
 
 #else // LLVM_ENABLE_THREADS Disabled
 
 // No threads are launched, issue a warning if ThreadCount is not 0
 ThreadPool::ThreadPool(ThreadPoolStrategy S) : MaxThreadCount(1) {