Fix a race condition in support library ThreadPool

[jhen]

Summary

By running TSAN on the ThreadPool unit tests it was discovered that the
threads in the pool can pop tasks off the queue at the same time the "wait"
routine is trying to check if the task queue is empty. This patch fixes this
problem by checking for active threads in the waiter before checking
whether the queue is empty.

[jlebar]

Apologies for scope creep here. If you want to save the bigger comments for another patch, that's fine. But I think most of these need to be addressed one way or another (not necessarily by you, I suppose).

The atomic ActiveThreads niggles me; it may be worth cc'ing the original author here to make sure it's not a Chesterton's Fence. I think this code is much more correct, though. :)

include/llvm/Support/ThreadPool.h

130

Maybe worth adding comments here indicating which variables are guarded by the mutex, so we are less likely to have the bug we previously had? I'd also move the definition of Mutex and the condvars above whatever they protect.

It would also be nice to have a testcase which illustrates the bug (even if only under tsan), but I don't know how hard that would be.

lib/Support/ThreadPool.cpp

39

Hmm, this means that if there are tasks in the queue when we run the destructor, we skip them. That is, we only block until we complete the currently-running tasks.

At least this should be documented in the destructor comment; it's not clear to me from that comment that this is our behavior. But I'm also not sure it's desirable, as can leave us with futures that will never be completed. Specifically, if you have threads waiting on futures, and then you destroy the threadpool without calling wait() on the threadpool first, those waiting threads will never wake up.

Since the destructor has to wait for running threads to complete *anyway*, clearing out the whole task queue seems pretty reasonable to me.

47

The order is no longer relevant, as everything is guarded by a single mutex. I think we can nix this whole comment?

60

Not sure this comment is necessary; seems pretty clear what's going on.

88

This is interesting, as it means that you're not allowed to enqueue new tasks from a task T, if it's possible that T is still running when we destroy the threadpool. Seems like it's worth a comment at least. I think the right behavior probably is to explicitly not allow new tasks after we've begun shutting down -- maybe we return a null future? Can we do that?

Threadpool shutdown is always so complicated...

104

May be worth #ifndef NDEBUG asserting here that ActiveThreads is 0.

[jhen]

• Respond to jlebar's comments

include/llvm/Support/ThreadPool.h

130

Done for the comments and rearranging.

The current tests show the bug under tsan, but I can't think of a way to get it to reliably manifest itself otherwise.

lib/Support/ThreadPool.cpp

39

It looks to me like it's not actually leaving any tasks in the queue. The statement below is the only way the thread can terminate from within, and it checks for Tasks.empty() before returning. If there are still tasks in the queue, the thread will grab one and run it regardless of whether the EnableFlag is set.

88

I think this is OK because the only way to destroy the ThreadPool is to call its destructor. If anyone is calling the method to enqueue work after the destructor was called, then I would say that is a bug in their program, not in this ThreadPool, and I think this assert failure is appropriate. If a method is added one day to destroy the pool without calling its destructor, then I could see this being a problem, so I changed the name of the EnableFlag to InDestructorFlag to make anybody think twice about setting it if they are not in the destructor.

[mehdi_amini]

One lock is simpler to understand, however it adds unnecessary contention, which is what I was trying to avoid originally.

include/llvm/Support/ThreadPool.h

129

Is this related to the changes here? If not then split it out to a separate NFC patch, reducing the change in this patch to the bug it is actually fixing. Same for renaming QueueCondition to SubmissionCondition.

lib/Support/ThreadPool.cpp

79

Would just reversing the check solve the race?

[&] { return !ActiveThreads && Tasks.empty(); });

If not can you elaborate a little?

88–89

I agree with jlebar that this scenario is sketchy:

{ 
  ThreadPool Pool;
  Pool.async([&] () {
     if (random()) Pool.async(doSomething);
  });
}

Technically it means that the client has to insert an explicit wait if a task can enqueue:

{ 
  ThreadPool Pool;
  Pool.async([&] () {
     if (random()) Pool.async(doSomething);
  });
  Pool.wait();
}

Synchronizing on a future inside a task would have the same kind of issue as well.

[mehdi_amini]

lib/Support/ThreadPool.cpp

65

To answer the comment about why ActiveThreads was atomic: it is updated here guarded by CompletionLock, but was incremented a few lines above guarded by QueueLock.

[jlebar]

lib/Support/ThreadPool.cpp

79

The race is that we touch Tasks here while holding only CompletionLock, while another thread in asyncImpl may touch Tasks while holding only QueueLock.

[mehdi_amini]

lib/Support/ThreadPool.cpp

79

Mmmm the description of the revision was not about *enqueuing* like you suggest, but about poping.
I need another look then.

[mehdi_amini]

lib/Support/ThreadPool.cpp

79

The use case you're mentioning, i.e. waiting for the pool to finish in one thread, while another thread would enqueue to the pool is explicitly forbidden by the API: see the wait() documentation: It is an error to try to add new tasks while blocking on this call.

[jlebar]

One lock is simpler to understand, however it adds unnecessary contention, which is what I was trying to avoid originally.

My preference is that we write simple, correct code and worry about performance separately, when we have explicit use-cases, benchmarks, etc. Otherwise we're just prematurely optimizing. Does that sound OK to you?

include/llvm/Support/ThreadPool.h

129

This feels pretty unnecessary to me. We're already changing a bunch of things in this class; who cares if we rename one additional variable in this patch as opposed to in a separate one? All of these changes are related in that they're attempting to make it less likely that this class will have bugs in the future.

There's a cost to carrying small patches in your queue that are tightly intermingled with a base patch; every time you update the base patch, you have to rebase the dependent patch.

I like small patches too, but there is a line. :)

lib/Support/ThreadPool.cpp

39

Oh, indeed, you're right; it's &&, not ||. Carry on. :)

[jhen]

To elaborate on the race condition, TSAN warns about the following when running the AsyncBarrier unit test (irrelevant stack frames removed):

WARNING: ThreadSanitizer: data race (pid=690669)
  Read of size 8 at 0x7fff3627b2b8 by main thread (mutexes: write M1928):
    wait<(lambda at lib/Support/ThreadPool.cpp:79:28)>
  Previous write of size 8 at 0x7fff3627b2b8 by thread T5 (mutexes: write M1849):
    operator() lib/Support/ThreadPool.cpp:53

So basically the wait routine can be checking whether the task queue is empty at the same time a thread can be trying to remove an element from the task queue. The queue data structure is not thread safe, so this is a race.

Sorry for the misleading description. I hope this clears up the issue.

[mehdi_amini]

My preference is that we write simple, correct code and worry about performance separately, when we have explicit use-cases, benchmarks, etc. Otherwise we're just prematurely optimizing. Does that sound OK to you?

Well, one can use exactly the same sentence to argue against threading at all in the first place :)
This implementation is pretty dumb anyway and I don't expect it to scale, it is convenient for coarse grain tasks, so I am indeed not really worried about the contention part here.
There is always a tradeoff to be acknowledged, I wanted to make it clear here, because I felt it wasn't from the original description.
(i.e. threads will now take the lock and release it both when popping from the queue, and when the task is completed).

include/llvm/Support/ThreadPool.h

129

This does not seems unnecessary to me, and this is in line with the request I usually receive on my patches. git blame and git show are a lot more friendly to analyze why the code is the way this way. As we say written once, read many...
(also minimizing diff helps the review)

As of the cost you're mentioning, you don't have to carry small patches in your queue as you can commit the NFC/renaming change immediately without going through any review.
It seems totally "free" to me so I don't really see why not doing it?

[mehdi_amini]

jhen: can you try reversing the condition as I suggested and verify if TSAN still complains (I'm still trying to make sure I have a good understanding of the existing logic...)

[jhen]

To clarify further. The race can occur without submitting tasks while waiting, there just has to be a task in the queue at the time the wait is called. That task would have been enqueued before the call to wait.

[jlebar]

include/llvm/Support/ThreadPool.h

129

As we say written once, read many...

Surely that does not apply to blame anywhere close to the degree it applies to code itself.

Since code is read so much more often than blame, we should encourage and welcome cleanups, making them easy for the contributor. As we do, with post-commit review.

In this case, Jason already is carrying the rename in his tree. You're asking him to either

a) split it out and commit the rename after he's submitted this patch, in which case he has to carry the changes through review and rebase, as I described, or

b) commit the changes now, before this patch is ready, in which case they may not end up being necessary, depending on the direction this review takes. (Who knows, maybe we remove this variable.) Committing unnecessary changes adds noise to the mailing list and wastes everyone's time.

In either case it's not free.

[jhen]

joker.eph: You are right, the TSAN error goes away if the order of the conditions on line 79 is swapped. I'll update this patch to make only that change.

[jhen]

• Simplify the change to one line

[jhen]

OK, I've simplified the change to just the single line suggested by joker.eph and the race condition is now cleared up.

[mehdi_amini]

I'm fine with this revision, of course.

However the original work you did is still valuable I think, making the code much simpler.
If you'd like, I would not oppose to a separate clean-up "NFC" patch that mentions the tradeoff on the locking in favor of a more understandable and maintainable code.

[jhen]

Thanks for your help jlebar and joker.eph.

include/llvm/Support/ThreadPool.h

private:
 
   /// Threads in flight
   std::vector<llvm::thread> Threads;
 
   /// Tasks waiting for execution in the pool.
   std::queue<PackagedTaskTy> 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::mutex Mutex;
+  std::condition_variable SubmissionCondition;
   std::condition_variable CompletionCondition;
 
   /// Keep track of the number of thread actually busy
-  std::atomic<unsigned> ActiveThreads;
+  unsigned ActiveThreads;
 
 #if LLVM_ENABLE_THREADS // avoids warning for unused variable
   /// Signal for the destruction of the pool, asking thread to exit.
   bool EnableFlag;
 #endif
 };
 }
 
 #endif // LLVM_SUPPORT_THREAD_POOL_H

lib/Support/ThreadPool.cpp

 
 #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 loop forever, wait on QueueCondition
-  // for tasks to be queued or the Pool to be destroyed.
+  // Create ThreadCount threads that will loop forever, wait on
+  // SubmissionCondition 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;
         {
-          std::unique_lock<std::mutex> LockGuard(QueueLock);
+          std::unique_lock<std::mutex> LockGuard(Mutex);
           // Wait for tasks to be pushed in the queue
-          QueueCondition.wait(LockGuard,
-                              [&] { return !EnableFlag || !Tasks.empty(); });
+          SubmissionCondition.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);
+          std::unique_lock<std::mutex> LockGuard(Mutex);
           --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);
+  std::unique_lock<std::mutex> LockGuard(Mutex);
   CompletionCondition.wait(LockGuard,
                            [&] { return Tasks.empty() && !ActiveThreads; });
 }
 
 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();
   {
     // Lock the queue and push the new task
-    std::unique_lock<std::mutex> LockGuard(QueueLock);
+    std::unique_lock<std::mutex> LockGuard(Mutex);
 
     // Don't allow enqueueing after disabling the pool
     assert(EnableFlag && "Queuing a thread during ThreadPool destruction");
 
     Tasks.push(std::move(PackagedTask));
   }
-  QueueCondition.notify_one();
+  SubmissionCondition.notify_one();
   return Future.share();
 }
 
 // The destructor joins all threads, waiting for completion.
 ThreadPool::~ThreadPool() {
   {
-    std::unique_lock<std::mutex> LockGuard(QueueLock);
+    std::unique_lock<std::mutex> LockGuard(Mutex);
     EnableFlag = false;
   }
-  QueueCondition.notify_all();
+  SubmissionCondition.notify_all();
   for (auto &Worker : Threads)
     Worker.join();
 }
 
 #else // LLVM_ENABLE_THREADS Disabled
 
 ThreadPool::ThreadPool() : ThreadPool(0) {}
 
 // No threads are launched, issue a warning if ThreadCount is not 0
 ThreadPool::ThreadPool(unsigned ThreadCount)