[Support] Fix races during thread pool teardown

[davezarzycki]

Summary

This fixes various races and cleans up the thread pool logic. In particular:

1. The ThreadPoolExecutor destructor now waits for the pool to fully exit before letting potentially in-use instance variables be implicitly destroyed.
2. Explicitly joining threads ensures that thread-specific data destruction is complete.

[davezarzycki]

A related question, why is this code reinventing thread joining? Would anybody mind if I clean this up?

[davezarzycki]

Rather than a targeted fix, let's just clean up the code and fix other bugs along the way.

[davezarzycki]

This should in theory be fixed by D70447.

lib/Support/Parallel.cpp

 //===- llvm/Support/Parallel.cpp - Parallel algorithms --------------------===//
 //
 // Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
 // See https://llvm.org/LICENSE.txt for license information.
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 //
 //===----------------------------------------------------------------------===//
 
 #include "llvm/Support/Parallel.h"
 #include "llvm/Config/llvm-config.h"
 
 #if LLVM_ENABLE_THREADS
 
-#include "llvm/Support/Threading.h"
-
 #include <atomic>
 #include <stack>
 #include <thread>
+#include <vector>
 
 namespace llvm {
 namespace parallel {
 namespace detail {
 
 namespace {
 
 /// An abstract class that takes closures and runs them asynchronously.
 class Executor {
 public:
   virtual ~Executor() = default;
   virtual void add(std::function<void()> func) = 0;
 
   static Executor *getDefaultExecutor();
 };
 
 /// An implementation of an Executor that runs closures on a thread pool
 ///   in filo order.
 class ThreadPoolExecutor : public Executor {
 public:
-  explicit ThreadPoolExecutor(unsigned ThreadCount = hardware_concurrency())
-      : Done(ThreadCount) {
-    // Spawn all but one of the threads in another thread as spawning threads
-    // can take a while.
-    std::thread([&, ThreadCount] {
-      for (size_t i = 1; i < ThreadCount; ++i) {
-        std::thread([=] { work(); }).detach();
-      }
-      work();
-    }).detach();
+  explicit ThreadPoolExecutor(
+      unsigned ThreadCount = std::thread::hardware_concurrency()) {
+    Pool.reserve(ThreadCount);
+    for (unsigned i = 0; i < ThreadCount; ++i) {
+      Pool.push_back(std::thread([=] { work(); }));
   }
 
   ~ThreadPoolExecutor() override {
     std::unique_lock<std::mutex> Lock(Mutex);
     Stop = true;
     Lock.unlock();
     Cond.notify_all();
-    // Wait for ~Latch.
+    for (auto &Thread : Pool) {
+      Thread.join();
+    }
   }
 
   void add(std::function<void()> F) override {
     std::unique_lock<std::mutex> Lock(Mutex);
     WorkStack.push(F);
     Lock.unlock();
     Cond.notify_one();
   }
 
 private:
   void work() {
     while (true) {
       std::unique_lock<std::mutex> Lock(Mutex);
       Cond.wait(Lock, [&] { return Stop || !WorkStack.empty(); });
       if (Stop)
         break;
       auto Task = WorkStack.top();
       WorkStack.pop();
       Lock.unlock();
       Task();
     }
-    Done.dec();
   }
 
   std::atomic<bool> Stop{false};
   std::stack<std::function<void()>> WorkStack;
   std::mutex Mutex;
   std::condition_variable Cond;
-  parallel::detail::Latch Done;
+  std::vector<std::thread> Pool;
 };
 
 Executor *Executor::getDefaultExecutor() {
   static ThreadPoolExecutor exec;
   return &exec;
 }
 } // namespace