[ThreadPool] Do not return shared futures.

[fhahn]

Summary

The only users of returned futures from ThreadPool is llvm-reduce after
D113857.

There should be no cases where multiple threads wait on the same future,
so there should be no need to return std::shared_future<>. Instead return
plain std::future<>.

If users need to share a future between multiple threads, they can share
the futures themselves.

[mehdi_amini]

I don't remember why this API was written with shared_future in the first place, this was part of the original implementation. The change makes sense though.

[Meinersbur]

LGTM as well. Indeed, users that need shared_future can call .share() themselves. Nothing else than the caller to async can retrieve the future anyway.

[fhahn]

Rebase so this can be landed independent of D113857.

It also updates the only place I could find that was storing futures in a container (in mlir/include/mlir/IR/Threading.h). The futures are only used by the main thread for waiting, so they shouldnt need to be shared.

[fhahn]

Funnily enough this is causing a build failure when threads are disabled, because Tasks.push([Future]() { Future.get(); }); won't work. I reverted the patch for now.

Any ideas on how to best deal with that? :)

[mehdi_amini]

Ah, that may be why I used shared_future in the first place! :)

[fhahn]

In D114363#3152115, @mehdi_amini wrote:

Ah, that may be why I used shared_future in the first place! :)

Yeah, it seems a shame though if we have to use std::shared_future to support the single threaded uses cases :(

[dblaikie]

In D114363#3152052, @fhahn wrote:

Funnily enough this is causing a build failure when threads are disabled, because Tasks.push([Future]() { Future.get(); }); won't work. I reverted the patch for now.

Any ideas on how to best deal with that? :)

Sounds like it might be due to lambda capture by value/copy (but if you could copy/paste the error it might help diagnose) - maybe capturing by move is all that's required?

[Future = std::move(Future)]() { Future.get(); }

[DanielMcIntosh-IBM]

In D114363#3152433, @dblaikie wrote:

In D114363#3152052, @fhahn wrote:

Funnily enough this is causing a build failure when threads are disabled, because Tasks.push([Future]() { Future.get(); }); won't work. I reverted the patch for now.

Any ideas on how to best deal with that? :)

Sounds like it might be due to lambda capture by value/copy (but if you could copy/paste the error it might help diagnose) - maybe capturing by move is all that's required?

[Future = std::move(Future)]() { Future.get(); }

I put the error message in a comment on this review: D114183, since that is the one that introduced the failure.

[fhahn]

In D114363#3154232, @DanielMcIntosh-IBM wrote:

In D114363#3152433, @dblaikie wrote:

In D114363#3152052, @fhahn wrote:

Funnily enough this is causing a build failure when threads are disabled, because Tasks.push([Future]() { Future.get(); }); won't work. I reverted the patch for now.

Any ideas on how to best deal with that? :)

Sounds like it might be due to lambda capture by value/copy (but if you could copy/paste the error it might help diagnose) - maybe capturing by move is all that's required?

[Future = std::move(Future)]() { Future.get(); }

I put the error message in a comment on this review: D114183, since that is the one that introduced the failure.

Are you sure? IIUC the problem was the change to not using shared_future. It built fine with LLVM_ENABLE_THREADS=Off with *this* patch.

[DanielMcIntosh-IBM]

Are you sure? IIUC the problem was the change to not using shared_future. It built fine with LLVM_ENABLE_THREADS=Off with *this* patch.

No, I was mistaken. I assumed that since it was still broken after rGfb46e64a013a2c7457db3312da07db1b25ad5219, this also needed to be reverted. Looks like the problem was actually solved by bc41653a1f289a63a1472b7364437515c9659f77 though.

llvm/include/llvm/Support/ThreadPool.h

public:
   inline auto async(Function &&F, Args &&...ArgList) {
     auto Task =
         std::bind(std::forward<Function>(F), std::forward<Args>(ArgList)...);
     return async(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.
-  template <typename Func>
-  auto async(Func &&F) -> std::shared_future<decltype(F())> {
+  template <typename Func> auto async(Func &&F) -> std::future<decltype(F())> {
     return asyncImpl(std::function<decltype(F())()>(std::forward<Func>(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();
 
   unsigned getThreadCount() const { return ThreadCount; }
return {[Promise = std::move(Promise), Task]() {
             std::move(F)};
   }
 
   bool workCompletedUnlocked() { return !ActiveThreads && Tasks.empty(); }
 
   /// 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 ResTy>
-  std::shared_future<ResTy> asyncImpl(std::function<ResTy()> Task) {
+  std::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);
 
     {
       // 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));
     }
     QueueCondition.notify_one();
-    return R.second.share();
+    return std::move(R.second);
 
 #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();
+    auto Future = std::async(std::launch::deferred, std::move(Task));
     // 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
   }
 
   /// Threads in flight

llvm/tools/llvm-reduce/deltas/Delta.cpp

do {
     // When running with more than one thread, serialize the original bitcode
     // to OriginalBC.
     SmallString<0> OriginalBC;
     if (NumJobs > 1) {
       raw_svector_ostream BCOS(OriginalBC);
       WriteBitcodeToFile(*Test.getProgram().M, BCOS);
     }
 
-    std::deque<std::shared_future<SmallString<0>>> TaskQueue;
+    std::deque<std::future<SmallString<0>>> TaskQueue;
     for (auto I = ChunksStillConsideredInteresting.rbegin(),
               E = ChunksStillConsideredInteresting.rend();
          I != E; ++I) {
       std::unique_ptr<ReducerWorkItem> Result = nullptr;
       unsigned WorkLeft = std::distance(I, E);
 
       // Run in parallel mode, if the user requested more than one thread and
       // there are at least a few chunks to process.