[Support][Parallel] Add sequential mode to TaskGroup::spawn().

[avl]

Summary

This patch allows to specify that some part of tasks should be
done in sequential order. It makes it possible to not use
condition operator for separating sequential tasks:

TaskGroup tg;
for () {
  if(condition)      ==>   tg.spawn([]{fn();}, condition)
    fn();
  else
    tg.spawn(, []{fn();});
}

It also prevents execution on main thread. Which allows adding
checks for getThreadIndex() function discussed in D142318.

The patch also replaces std::stack with std::queue in the
ThreadPoolExecutor to have natural execution order in case
(parallel::strategy.ThreadsRequested == 1).

[MaskRay]

llvm/lib/Support/Parallel.cpp

145

After Lock.unclok(), another thread may grab the lock and run a task concurrently.

llvm/unittests/Support/ParallelTest.cpp

100

llvm style prefers pre-increments ++Count

[andrewng]

llvm/lib/Support/Parallel.cpp

16–17

Unfortunately, at least in my testing on Windows, this change causes a drop in performance of around 3-4%. I believe that's why std::stack is used.

104–109

Would this be nicer?

114

Make const?

118

Make const?

145

Yes, this can happen but I believe the key point is that it can only be a thread with a different "index". However, one subtlety is that "sequential" tasks will run on different threads and thus use a different "index".

148

I think that moving this up after the other std::atomic would feel better.

llvm/unittests/Support/ParallelTest.cpp

95–119

I'm not entirely sure what this test case adds in terms of coverage that isn't covered by the test case that follows.

[andrewng]

llvm/lib/Support/Parallel.cpp

140–143

Not entirely sure this is better but does remove the code duplication.

[avl]

llvm/lib/Support/Parallel.cpp

16–17

Unfortunately, at least in my testing on Windows, this change causes a drop in performance of around 3-4%. I believe that's why std::stack is used.

will check it.

145

After Lock.unclok(), another thread may grab the lock and run a task concurrently.

145

After Lock.unclok(), another thread may grab the lock and run a task concurrently.

After Lock.unlock(); the SequentialQueueIsLocked is true. Thus another thread could not start new sequential task until SequentialQueueIsLocked becomes false. SequentialQueueIsLocked set to false when sequential task is finished. Thus there could not be two sequential tasks executing in parallel.

145

Yes, this can happen but I believe the key point is that it can only be a thread with a different "index". However, one subtlety is that "sequential" tasks will run on different threads and thus use a different "index".

Yes, "sequential" tasks may be run on different threads and use a different "index". I supposed that this is not a problem.

llvm/unittests/Support/ParallelTest.cpp

95–119

I'm not entirely sure what this test case adds in terms of coverage that isn't covered by the test case that follows.

The idea was to test a case when previous task is already executed before new .spawn() is called.
Will remove if it seems redundant.

[avl]

llvm/lib/Support/Parallel.cpp

16–17

Unfortunately, at least in my testing on Windows, this change causes a drop in performance of around 3-4%. I believe that's why std::stack is used.

yes, there is small performance degradation in case std::queue is used instead of std::stack.
But, to have it noticeable, there should be created very large amount of tasks.
For the 100000000 tasks the std::queue works 8sec slower:

llvm::parallel::TaskGroup tg;
for (size_t Idx = 0; Idx < 100000000; Idx++) {
  tg.spawn([](){});
}

                        Time(sec)  Memory(kB)

queue TaskGroup          121.28    441568

stack TaskGroup          113.61    471820

Generally, it is a bad idea to create such amount of tasks.
parallelFor has optimization to not create a lot of tasks.
There is no any performance difference if parallelFor is used:

parallelFor(0,100000000,[](size_t){});

                        Time(sec)  Memory(kB)

queue parallelFor          0.03    9128

stack parallelFor          0.03    9128

So, probably using std::queue is OK.

Or, Dou you have a test case when using queue still bad idea?

[avl]

addressed comments(except using std::queue and extra test)

[andrewng]

llvm/lib/Support/Parallel.cpp

16–17

So, probably using std::queue is OK.

Or, Dou you have a test case when using queue still bad idea?

I used my test case from D147493 which is a UE4 based link with a LLVM 16 libc++ self-hosted release build with the rpmalloc allocator on Windows 10, AMD Ryzen 3900X 12C/24T, Samsung 970 EVO NVMe SSD, 64GB RAM. This link shows the 3-4% drop in performance which I think is significant enough to justify consideration.

One solution would be to use std::deque and add "work" to the front or back dependent on Sequential, e.g.

void add(std::function<void()> F, bool Sequential = false) override {
  {
    std::lock_guard<std::mutex> Lock(Mutex);
    if (Sequential)
      WorkQueueSequential.emplace_front(std::move(F));
    else
      WorkQueue.emplace_back(std::move(F));
  }
  Cond.notify_one();
}

void work(ThreadPoolStrategy S, unsigned ThreadID) {
  ...
    auto &Queue = Sequential ? WorkQueueSequential : WorkQueue;
    auto Task = std::move(Queue.back());
    Queue.pop_back();
  ...
}

This change seems to restore the performance of my test link. What do you think?

TBH, I think we need to be a little bit careful with any further development of these parallel methods. IIRC, they were added to support basic and simple parallelism in LLD. I don't think the intention was ever for them to be general parallel algorithms for all general use cases, e.g. nested parallel for. I think my main concern is that making them more general will likely add threading overhead which in turn may impact the effectiveness and performance in LLD.

145

Yes, "sequential" tasks may be run on different threads and use a different "index". I supposed that this is not a problem.

I don't think it is a problem as far as correctness, but might be somewhat unexpected.

llvm/unittests/Support/ParallelTest.cpp

95–119

The idea was to test a case when previous task is already executed before new .spawn() is called.
Will remove if it seems redundant.

OK, I can kind of see the intent, although I don't think the test actually tests or demonstrates anything extra. Personally, I think it can be removed. If you do remove, you can remove the two includes too.

[andrewng]

I think it would be worthwhile adding a test that shows that sequential tasks are executed in the order of submission.

[avl]

In D148728#4292033, @andrewng wrote:

I think it would be worthwhile adding a test that shows that sequential tasks are executed in the order of submission.

It looks like TaskGroupSequentialFor already do this?

llvm/lib/Support/Parallel.cpp

16–17

One solution would be to use std::deque and add "work" to the front or back dependent on Sequential, e.g.

void add(std::function<void()> F, bool Sequential = false) override {
  {
    std::lock_guard<std::mutex> Lock(Mutex);
    if (Sequential)
      WorkQueueSequential.emplace_front(std::move(F));
    else
      WorkQueue.emplace_back(std::move(F));
  }
  Cond.notify_one();
}

void work(ThreadPoolStrategy S, unsigned ThreadID) {
  ...
    auto &Queue = Sequential ? WorkQueueSequential : WorkQueue;
    auto Task = std::move(Queue.back());
    Queue.pop_back();
  ...
}

This change seems to restore the performance of my test link. What do you think?

this is OK for me. will implement it.

By the way, as LLD is sensitive to changing stack to the queue, it might be beneficial to replace for loops with parallelFor some when.

for (InputSectionBase *s : f->getSections()) {
....
for (Partition &part : partitions)

TBH, I think we need to be a little bit careful with any further development of these parallel methods. IIRC, they were added to support basic and simple parallelism in LLD. I don't think the intention was ever for them to be general parallel algorithms for all general use cases, e.g. nested parallel for. I think my main concern is that making them more general will likely add threading overhead which in turn may impact the effectiveness and performance in LLD.

yep. My intention is not to support nested parallel for by parallel methods. But allow to use these methods with other solutions. f.e. nested parallel for might be done using existing ThreadPool and basic parallel methods

std::function<void()> Fn = [&]() {
  parallelFor(Passes) {
  }
};

ThreadPool Pool;

for (Files) {
  Pool.async(Fn);

Pool.wait();

In that case parallel methods just work as simple parallel methods.

145

will document that behavior.

[andrewng]

I think it would be worthwhile adding a test that shows that sequential tasks are executed in the order of submission.

It looks like TaskGroupSequentialFor already do this?

Yes, sorry, of course it does!

[avl]

addressed comments(deleted extra test, changed queue to deque,
put tasks in front of deque in case sequential or parallel::strategy.ThreadsRequested == 1)

[andrewng]

Apart from the comment nits, LGTM. Thanks!

llvm/include/llvm/Support/Parallel.h

88

how -> which.

89

done -> executed or run. Drop the the.

[avl]

Thank you for the review!

lld/ELF/OutputSections.cpp

void OutputSection::writeTo(uint8_t *buf, parallel::TaskGroup &tg) {
   // time with other output sections. Note, if a linker script specifies
   // overlapping output sections (needs --noinhibit-exec or --no-check-sections
   // to supress the error), the output may be non-deterministic.
   const size_t taskSizeLimit = 4 << 20;
   for (size_t begin = 0, i = 0, taskSize = 0;;) {
     taskSize += sections[i]->getSize();
     bool done = ++i == numSections;
     if (done || taskSize >= taskSizeLimit) {
-      tg.execute([=] { fn(begin, i); });
+      tg.spawn([=] { fn(begin, i); });
       if (done)
         break;
       begin = i;
       taskSize = 0;
     }
   }
 }
 

lld/ELF/Relocations.cpp

auto fn = [f]() {
       RelocationScanner scanner;
       for (InputSectionBase *s : f->getSections()) {
         if (s && s->kind() == SectionBase::Regular && s->isLive() &&
             (s->flags & SHF_ALLOC) &&
             !(s->type == SHT_ARM_EXIDX && config->emachine == EM_ARM))
           scanner.template scanSection<ELFT>(*s);
       }
     };
-    if (serial)
-      fn();
-    else
-      tg.execute(fn);
+    tg.spawn(fn, serial);
   }
 
   // Both the main thread and thread pool index 0 use getThreadIndex()==0. Be
   // careful that they don't concurrently run scanSections. When serial is
   // true, fn() has finished at this point, so running execute is safe.
-  tg.execute([] {
+  tg.spawn([] {
     RelocationScanner scanner;
     for (Partition &part : partitions) {
       for (EhInputSection *sec : part.ehFrame->sections)
         scanner.template scanSection<ELFT>(*sec);
       if (part.armExidx && part.armExidx->isLive())
         for (InputSection *sec : part.armExidx->exidxSections)
           scanner.template scanSection<ELFT>(*sec);
     }

llvm/include/llvm/Support/Parallel.h

class TaskGroup {
   detail::Latch L;
   bool Parallel;
 
 public:
   TaskGroup();
   ~TaskGroup();
 
   // Spawn a task, but does not wait for it to finish.
-  void spawn(std::function<void()> f);
-
-  // Similar to spawn, but execute the task immediately when ThreadsRequested ==
-  // 1. The difference is to give the following pattern a more intuitive order
-  // when single threading is requested.
-  //
-  // for (size_t begin = 0, i = 0, taskSize = 0;;) {
-  //   taskSize += ...
-  //   bool done = ++i == end;
-  //   if (done || taskSize >= taskSizeLimit) {
-  //     tg.execute([=] { fn(begin, i); });
-  //     if (done)
-  //       break;
-  //     begin = i;
-  //     taskSize = 0;
-  //   }
-  // }
-  void execute(std::function<void()> f);
+  // Tasks marked with \p Sequential will be executed
+  // exactly in the order which they were spawned.
+  // Note: Sequential tasks may be executed on different
+  // threads, but strictly in sequential order.
+  void spawn(std::function<void()> f, bool Sequential = false);
 
   void sync() const { L.sync(); }
 };
 
 namespace detail {
 
 #if LLVM_ENABLE_THREADS
 const ptrdiff_t MinParallelSize = 1024;

llvm/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"
 #include "llvm/Support/ManagedStatic.h"
 #include "llvm/Support/Threading.h"
 
 #include <atomic>
+#include <deque>
 #include <future>
-#include <stack>
 #include <thread>
 #include <vector>
 
 llvm::ThreadPoolStrategy llvm::parallel::strategy;
 
 namespace llvm {
 namespace parallel {
 #if LLVM_ENABLE_THREADS
 
 #ifdef _WIN32
 static thread_local unsigned threadIndex;
 
 unsigned getThreadIndex() { return threadIndex; }
 #else
 thread_local unsigned threadIndex;
 #endif
 
 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;
+  virtual void add(std::function<void()> func, bool Sequential = false) = 0;
 
   static Executor *getDefaultExecutor();
 };
 
 /// An implementation of an Executor that runs closures on a thread pool
 ///   in filo order.
 class ThreadPoolExecutor : public Executor {
 public:
public:
 
   struct Creator {
     static void *call() { return new ThreadPoolExecutor(strategy); }
   };
   struct Deleter {
     static void call(void *Ptr) { ((ThreadPoolExecutor *)Ptr)->stop(); }
   };
 
-  void add(std::function<void()> F) override {
+  void add(std::function<void()> F, bool Sequential = false) override {
     {
+      bool UseSequentialQueue =
+          Sequential || parallel::strategy.ThreadsRequested == 1;
       std::lock_guard<std::mutex> Lock(Mutex);
-      WorkStack.push(std::move(F));
+      if (UseSequentialQueue)
+        WorkQueueSequential.emplace_front(std::move(F));
+      else
+        WorkQueue.emplace_back(std::move(F));
     }
     Cond.notify_one();
   }
 
 private:
+  bool hasSequentialTasks() const {
+    return !WorkQueueSequential.empty() && !SequentialQueueIsLocked;
+  }
+
+  bool hasGeneralTasks() const { return !WorkQueue.empty(); }
+
   void work(ThreadPoolStrategy S, unsigned ThreadID) {
     threadIndex = ThreadID;
     S.apply_thread_strategy(ThreadID);
     while (true) {
       std::unique_lock<std::mutex> Lock(Mutex);
-      Cond.wait(Lock, [&] { return Stop || !WorkStack.empty(); });
+      Cond.wait(Lock, [&] {
+        return Stop || hasGeneralTasks() || hasSequentialTasks();
+      });
       if (Stop)
         break;
-      auto Task = std::move(WorkStack.top());
-      WorkStack.pop();
+      bool Sequential = hasSequentialTasks();
+      if (Sequential)
+        SequentialQueueIsLocked = true;
+      else
+        assert(hasGeneralTasks());
+
+      auto &Queue = Sequential ? WorkQueueSequential : WorkQueue;
+      auto Task = std::move(Queue.back());
+      Queue.pop_back();
       Lock.unlock();
       Task();
+      if (Sequential)
+        SequentialQueueIsLocked = false;
     }
   }
 
   std::atomic<bool> Stop{false};
-  std::stack<std::function<void()>> WorkStack;
+  std::atomic<bool> SequentialQueueIsLocked{false};
+  std::deque<std::function<void()>> WorkQueue;
+  std::deque<std::function<void()>> WorkQueueSequential;
   std::mutex Mutex;
   std::condition_variable Cond;
   std::promise<void> ThreadsCreated;
   std::vector<std::thread> Threads;
 };
 
 Executor *Executor::getDefaultExecutor() {
   // The ManagedStatic enables the ThreadPoolExecutor to be stopped via

 TaskGroup::TaskGroup() : Parallel(TaskGroupInstances++ == 0) {}
 TaskGroup::~TaskGroup() {
   // We must ensure that all the workloads have finished before decrementing the
   // instances count.
   L.sync();
   --TaskGroupInstances;
 }
 
-void TaskGroup::spawn(std::function<void()> F) {
+void TaskGroup::spawn(std::function<void()> F, bool Sequential) {
 #if LLVM_ENABLE_THREADS
   if (Parallel) {
     L.inc();
-    detail::Executor::getDefaultExecutor()->add([&, F = std::move(F)] {
+    detail::Executor::getDefaultExecutor()->add(
+        [&, F = std::move(F)] {
           F();
           L.dec();
-    });
+        },
+        Sequential);
     return;
   }
 #endif
   F();
 }
 
-void TaskGroup::execute(std::function<void()> F) {
-  if (parallel::strategy.ThreadsRequested == 1)
-    F();
-  else
-    spawn(F);
-}
 } // namespace parallel
 } // namespace llvm
 
 void llvm::parallelFor(size_t Begin, size_t End,
                        llvm::function_ref<void(size_t)> Fn) {
   // If we have zero or one items, then do not incur the overhead of spinning up
   // a task group.  They are surprisingly expensive, and because they do not
   // support nested parallelism, a single entry task group can block parallel

llvm/unittests/Support/ParallelTest.cpp

if ((v & 1) == 0)
       return createStringError(std::errc::invalid_argument, "asdf");
     return Error::success();
   });
   EXPECT_TRUE(e.isA<ErrorList>());
   std::string errText = toString(std::move(e));
   EXPECT_EQ(errText, std::string("asdf\nasdf\nasdf"));
 }
 
+TEST(Parallel, TaskGroupSequentialFor) {
+  size_t Count = 0;
+  {
+    parallel::TaskGroup tg;
+    for (size_t Idx = 0; Idx < 500; Idx++)
+      tg.spawn([&Count, Idx]() { EXPECT_EQ(Count++, Idx); }, true);
+  }
+  EXPECT_EQ(Count, 500ul);
+}
+
 #endif