[Parallel] Add support for parallel execution policies to match C++ Parallelism TS

[zturner]

Summary

This is the next step towards moving LLD's parallel algorithms over to LLVM.

With this patch, I introduce parallel_execution_policy, sequential_execution_policy, and is_execution_policy to as described in N4507.

It does not introduce the type erasing dynamic execution policy.

All callers of existing parallel algorithms are updated to use the new function syntax, and parallel_for is renamed to for_each_n to match the specification.

After this change, I am hoping this library can be moved to LLVM wholesale with no further changes.

[ruiu]

lld/include/lld/Core/Parallel.h

31–32

nit: do you need {}?

242

I think you can remove detail::sequential_sort and use std::sort instead.

256

This function is called once. You can remove the function and inline.

[ruiu]

I think I prefer inlining them as sequential_sort and sequential_for_each_n are in the end just sort and for_each.

[ruiu]

LGTM with inlining.

[zturner]

Inlined sequential functions. I know you already LGTM'ed it, but I'm re-uploading because I want to get feedback from chandlerc@ before submitting. I'd like to be sure this is close enough to what is needed from the Parallelism TS that no more major changes are required before trying to move this down to LLVM again. I will, of course, still submit another review to LLVM for the final move, but I just want to make sure this is close enough to try again.

[zturner]

I went ahead and added the dynamic execution policy.

[chandlerc]

Other than a YAGNI argument against the type-erased stuff, seems fine to me.

lld/include/lld/Core/Parallel.h

54

This seems like a *lot* of machinery for one use case in one part of LLD.... I'd honestly just not implement this until someone really wants it because they need it....

lld/COFF/ICF.cpp

void ICF::forEachClass(std::function<void(size_t, size_t)> Fn) {
   if (Chunks.size() < 1024) {
     forEachClassRange(0, Chunks.size(), Fn);
     return;
   }
 
   // Split sections into 256 shards and call Fn in parallel.
   size_t NumShards = 256;
   size_t Step = Chunks.size() / NumShards;
-  parallel_for(size_t(0), NumShards, [&](size_t I) {
+  for_each_n(parallel::par, size_t(0), NumShards, [&](size_t I) {
     forEachClassRange(I * Step, (I + 1) * Step, Fn);
   });
   forEachClassRange(Step * NumShards, Chunks.size(), Fn);
 }
 
 // Merge identical COMDAT sections.
 // Two sections are considered the same if their section headers,
 // contents and relocations are all the same.

lld/COFF/MapFile.cpp

static SymbolMapTy getSectionSyms(ArrayRef<DefinedRegular *> Syms) {
   }
   return Ret;
 }
 
 // Construct a map from symbols to their stringified representations.
 static DenseMap<DefinedRegular *, std::string>
 getSymbolStrings(ArrayRef<DefinedRegular *> Syms) {
   std::vector<std::string> Str(Syms.size());
-  parallel_for((size_t)0, Syms.size(), [&](size_t I) {
+  for_each_n(parallel::par, (size_t)0, Syms.size(), [&](size_t I) {
     raw_string_ostream OS(Str[I]);
     writeHeader(OS, Syms[I]->getRVA(), 0, 0);
     OS << indent(2) << toString(*Syms[I]);
   });
 
   DenseMap<DefinedRegular *, std::string> Ret;
   for (size_t I = 0, E = Syms.size(); I < E; ++I)
     Ret[Syms[I]] = std::move(Str[I]);

lld/COFF/Writer.cpp

void Writer::writeSections() {
   uint8_t *Buf = Buffer->getBufferStart();
   for (OutputSection *Sec : OutputSections) {
     uint8_t *SecBuf = Buf + Sec->getFileOff();
     // Fill gaps between functions in .text with INT3 instructions
     // instead of leaving as NUL bytes (which can be interpreted as
     // ADD instructions).
     if (Sec->getPermissions() & IMAGE_SCN_CNT_CODE)
       memset(SecBuf, 0xCC, Sec->getRawSize());
-    parallel_for_each(Sec->getChunks().begin(), Sec->getChunks().end(),
-                      [&](Chunk *C) { C->writeTo(SecBuf); });
+    for_each(parallel::par, Sec->getChunks().begin(), Sec->getChunks().end(),
   }
 }
 
 // Sort .pdata section contents according to PE/COFF spec 5.5.
 void Writer::sortExceptionTable() {
   OutputSection *Sec = findSection(".pdata");
   if (!Sec)
     return;
   // We assume .pdata contains function table entries only.
   uint8_t *Begin = Buffer->getBufferStart() + Sec->getFileOff();
   uint8_t *End = Begin + Sec->getVirtualSize();
   if (Config->Machine == AMD64) {
     struct Entry { ulittle32_t Begin, End, Unwind; };
-    parallel_sort(
-        (Entry *)Begin, (Entry *)End,
+    sort(parallel::par, (Entry *)Begin, (Entry *)End,
          [](const Entry &A, const Entry &B) { return A.Begin < B.Begin; });
     return;
   }
   if (Config->Machine == ARMNT) {
     struct Entry { ulittle32_t Begin, Unwind; };
-    parallel_sort(
-        (Entry *)Begin, (Entry *)End,
+    sort(parallel::par, (Entry *)Begin, (Entry *)End,
          [](const Entry &A, const Entry &B) { return A.Begin < B.Begin; });
     return;
   }
   errs() << "warning: don't know how to handle .pdata.\n";
 }
 
 // Backfill the CVSignature in a PDB70 Debug Record.  This backfilling allows us
 // to get reproducible builds.
 void Writer::writeBuildId() {

lld/ELF/Threads.h

 #include <algorithm>
 #include <functional>
 
 namespace lld {
 namespace elf {
 
 template <class IterTy, class FuncTy>
 void parallelForEach(IterTy Begin, IterTy End, FuncTy Fn) {
+  parallel::execution_policy pol = parallel::seq;
   if (Config->Threads)
-    parallel_for_each(Begin, End, Fn);
-  else
-    std::for_each(Begin, End, Fn);
+    pol = parallel::par;
+  parallel::for_each(pol, Begin, End, Fn);
 }
 
 inline void parallelFor(size_t Begin, size_t End,
                         std::function<void(size_t)> Fn) {
-  if (Config->Threads) {
-    parallel_for(Begin, End, Fn);
-  } else {
-    for (size_t I = Begin; I < End; ++I)
-      Fn(I);
-  }
+  parallel::execution_policy pol = parallel::seq;
+  if (Config->Threads)
+    pol = parallel::par;
+  parallel::for_each_n(pol, Begin, End, Fn);
 }
 }
 }
 
 #endif

lld/include/lld/Core/Parallel.h

 //===- lld/Core/Parallel.h - Parallel utilities ---------------------------===//
 //
 //                             The LLVM Linker
 //
 // This file is distributed under the University of Illinois Open Source
 // License. See LICENSE.TXT for details.
 //
 //===----------------------------------------------------------------------===//
 
 #ifndef LLD_CORE_PARALLEL_H
 #define LLD_CORE_PARALLEL_H
 
 #include "lld/Core/LLVM.h"
 #include "lld/Core/TaskGroup.h"
-#include "llvm/Support/MathExtras.h"
+#include "llvm/ADT/STLExtras.h"
 #include "llvm/Config/llvm-config.h"
+#include "llvm/Support/MathExtras.h"
 
 #include <algorithm>
 
 #if defined(_MSC_VER) && LLVM_ENABLE_THREADS
 #include <concrt.h>
 #include <ppl.h>
 #endif
 
 namespace lld {
 
-#if !LLVM_ENABLE_THREADS
-template <class RandomAccessIterator, class Comparator>
-void parallel_sort(
-    RandomAccessIterator Start, RandomAccessIterator End,
-    const Comparator &Comp = std::less<
-        typename std::iterator_traits<RandomAccessIterator>::value_type>()) {
-  std::sort(Start, End, Comp);
+namespace parallel {
+struct sequential_execution_policy {};
+struct parallel_execution_policy {};
+struct execution_policy;
+
+template <typename T>
+struct is_execution_policy
+    : public std::integral_constant<
+          bool,
+          llvm::is_one_of<T, sequential_execution_policy,
+                          parallel_execution_policy, execution_policy>::value> {
+};
+
+namespace detail {
+template <typename Policy> struct policy_traits;
+template <> struct policy_traits<sequential_execution_policy> {
+  static constexpr unsigned id = 0;
+};
+template <> struct policy_traits<parallel_execution_policy> {
+  static constexpr unsigned id = 1;
+};
+template <> struct policy_traits<execution_policy> {
+  static constexpr unsigned id = 2;
+};
+}
+
+// Dynamic execution policy
+struct execution_policy {
+private:
+  struct impl_base {
+    virtual ~impl_base() {}
+    virtual impl_base *clone() const = 0;
+    virtual unsigned id() const = 0;
+  };
+
+  template <typename Policy> struct impl : public impl_base {
+    explicit impl(const Policy &P) : P(P) {}
+
+    static_assert(is_execution_policy<Policy>::value,
+                  "Not an execution policy!");
+
+    static bool classof(const impl_base *Other) {
+      return Other->id() == detail::policy_traits<Policy>::id;
+    }
+
+    unsigned id() const override { return detail::policy_traits<Policy>::id; }
+
+    impl_base *clone() const override { return new impl<Policy>(P); }
+
+    const Policy &P;
+  };
+
+public:
+  template <class Policy>
+  execution_policy(const Policy &exec)
+      : Impl(llvm::make_unique<impl<Policy>>(exec)) {
+    static_assert(is_execution_policy<Policy>::value,
+                  "Not an execution policy!");
+  }
+
+  execution_policy(const execution_policy &other) : Impl(other.Impl->clone()) {}
+
+  template <class Policy> execution_policy &operator=(const Policy &exec) {
+    static_assert(is_execution_policy<Policy>::value,
+                  "Not an execution policy!");
+    Impl = llvm::make_unique<impl<Policy>>(exec);
+    return *this;
+  }
+
+  execution_policy &operator=(const execution_policy &other) {
+    Impl.reset(other.Impl->clone());
+    return *this;
+  }
+
+  template <class Policy> Policy *get() {
+    static_assert(is_execution_policy<Policy>::value,
+                  "Not an execution policy!");
+    impl<Policy> *Ptr = llvm::dyn_cast<impl<Policy>>(Impl.get());
+    return Ptr ? &Ptr->P : nullptr;
+  }
+  template <class Policy> const Policy *get() const {
+    static_assert(is_execution_policy<Policy>::value,
+                  "Not an execution policy!");
+    impl<Policy> *Ptr = llvm::dyn_cast<impl<Policy>>(Impl.get());
+    return Ptr ? &Ptr->P : nullptr;
   }
-#elif defined(_MSC_VER)
-// Use ppl parallel_sort on Windows.
+
+private:
+  std::unique_ptr<impl_base> Impl;
+};
+
+constexpr sequential_execution_policy seq{};
+constexpr parallel_execution_policy par{};
+
+#if LLVM_ENABLE_THREADS
+
+namespace detail {
+
+#if defined(_MSC_VER)
 template <class RandomAccessIterator, class Comparator>
-void parallel_sort(
-    RandomAccessIterator Start, RandomAccessIterator End,
-    const Comparator &Comp = std::less<
-        typename std::iterator_traits<RandomAccessIterator>::value_type>()) {
+void parallel_sort(RandomAccessIterator Start, RandomAccessIterator End,
+                   const Comparator &Comp) {
   concurrency::parallel_sort(Start, End, Comp);
 }
+template <class IterTy, class FuncTy>
+void parallel_for_each(IterTy Begin, IterTy End, FuncTy Fn) {
+  concurrency::parallel_for_each(Begin, End, Fn);
+}
+
+template <class IndexTy, class FuncTy>
+void parallel_for_each_n(IndexTy Begin, IndexTy End, FuncTy Fn) {
+  concurrency::parallel_for(Begin, End, Fn);
+}
+
 #else
-namespace detail {
 const ptrdiff_t MinParallelSize = 1024;
 
 /// \brief Inclusive median.
 template <class RandomAccessIterator, class Comparator>
 RandomAccessIterator medianOf3(RandomAccessIterator Start,
                                RandomAccessIterator End,
                                const Comparator &Comp) {
   RandomAccessIterator Mid = Start + (std::distance(Start, End) / 2);
void parallel_quick_sort(RandomAccessIterator Start, RandomAccessIterator End,
   std::swap(*Pivot, *(End - 1));
 
   // Recurse.
   TG.spawn([=, &Comp, &TG] {
     parallel_quick_sort(Start, Pivot, Comp, TG, Depth - 1);
   });
   parallel_quick_sort(Pivot + 1, End, Comp, TG, Depth - 1);
 }
-}
 
 template <class RandomAccessIterator, class Comparator>
-void parallel_sort(
-    RandomAccessIterator Start, RandomAccessIterator End,
-    const Comparator &Comp = std::less<
-        typename std::iterator_traits<RandomAccessIterator>::value_type>()) {
+void parallel_sort(RandomAccessIterator Start, RandomAccessIterator End,
+                   const Comparator &Comp) {
   TaskGroup TG;
-  detail::parallel_quick_sort(Start, End, Comp, TG,
-                              llvm::Log2_64(std::distance(Start, End)) + 1);
+  parallel_quick_sort(Start, End, Comp, TG,
 }
-#endif
-
-template <class T> void parallel_sort(T *Start, T *End) {
-  parallel_sort(Start, End, std::less<T>());
-}
-
-#if !LLVM_ENABLE_THREADS
-template <class IterTy, class FuncTy>
-void parallel_for_each(IterTy Begin, IterTy End, FuncTy Fn) {
-  std::for_each(Begin, End, Fn);
-}
-
-template <class IndexTy, class FuncTy>
-void parallel_for(IndexTy Begin, IndexTy End, FuncTy Fn) {
-  for (IndexTy I = Begin; I != End; ++I)
-    Fn(I);
-}
-#elif defined(_MSC_VER)
-// Use ppl parallel_for_each on Windows.
-template <class IterTy, class FuncTy>
-void parallel_for_each(IterTy Begin, IterTy End, FuncTy Fn) {
-  concurrency::parallel_for_each(Begin, End, Fn);
-}
 
-template <class IndexTy, class FuncTy>
-void parallel_for(IndexTy Begin, IndexTy End, FuncTy Fn) {
-  concurrency::parallel_for(Begin, End, Fn);
-}
-#else
 template <class IterTy, class FuncTy>
 void parallel_for_each(IterTy Begin, IterTy End, FuncTy Fn) {
   // TaskGroup has a relatively high overhead, so we want to reduce
   // the number of spawn() calls. We'll create up to 1024 tasks here.
   // (Note that 1024 is an arbitrary number. This code probably needs
   // improving to take the number of available cores into account.)
   ptrdiff_t TaskSize = std::distance(Begin, End) / 1024;
   if (TaskSize == 0)
     TaskSize = 1;
 
   TaskGroup TG;
   while (TaskSize <= std::distance(Begin, End)) {
     TG.spawn([=, &Fn] { std::for_each(Begin, Begin + TaskSize, Fn); });
     Begin += TaskSize;
   }
   TG.spawn([=, &Fn] { std::for_each(Begin, End, Fn); });
 }
 
 template <class IndexTy, class FuncTy>
-void parallel_for(IndexTy Begin, IndexTy End, FuncTy Fn) {
+void parallel_for_each_n(IndexTy Begin, IndexTy End, FuncTy Fn) {
   ptrdiff_t TaskSize = (End - Begin) / 1024;
   if (TaskSize == 0)
     TaskSize = 1;
 
   TaskGroup TG;
   IndexTy I = Begin;
   for (; I + TaskSize < End; I += TaskSize) {
     TG.spawn([=, &Fn] {
       for (IndexTy J = I, E = I + TaskSize; J != E; ++J)
         Fn(J);
     });
   }
   TG.spawn([=, &Fn] {
     for (IndexTy J = I; J < End; ++J)
       Fn(J);
   });
 }
+
+#endif
+
+template <typename Iter>
+using DefComparator =
+    std::less<typename std::iterator_traits<Iter>::value_type>;
+
+} // namespace detail
+#endif
+
+// sequential algorithm implementations.
+template <class Policy, class RandomAccessIterator,
+          class Comparator = detail::DefComparator<RandomAccessIterator>>
+void sort(Policy policy, RandomAccessIterator Start, RandomAccessIterator End,
+          const Comparator &Comp = Comparator()) {
+  static_assert(is_execution_policy<Policy>::value,
+                "Invalid execution policy!");
+  std::sort(Start, End, Comp);
+}
+
+template <class Policy, class IterTy, class FuncTy>
+void for_each(Policy policy, IterTy Begin, IterTy End, FuncTy Fn) {
+  static_assert(is_execution_policy<Policy>::value,
+                "Invalid execution policy!");
+  std::for_each(Begin, End, Fn);
+}
+
+template <class Policy, class IndexTy, class FuncTy>
+void for_each_n(Policy policy, IndexTy Begin, IndexTy End, FuncTy Fn) {
+  static_assert(is_execution_policy<Policy>::value,
+                "Invalid execution policy!");
+  for (IndexTy I = Begin; I != End; ++I)
+    Fn(I);
+}
+
+// Parallel algorithm implementations, only available when LLVM_ENABLE_THREADS
+// is true.
+#if defined(LLVM_ENABLE_THREADS)
+template <class RandomAccessIterator,
+          class Comparator = detail::DefComparator<RandomAccessIterator>>
+void sort(parallel_execution_policy policy, RandomAccessIterator Start,
+          RandomAccessIterator End, const Comparator &Comp = Comparator()) {
+  detail::parallel_sort(Start, End, Comp);
+}
+
+template <class IterTy, class FuncTy>
+void for_each(parallel_execution_policy policy, IterTy Begin, IterTy End,
+              FuncTy Fn) {
+  detail::parallel_for_each(Begin, End, Fn);
+}
+
+template <class IndexTy, class FuncTy>
+void for_each_n(parallel_execution_policy policy, IndexTy Begin, IndexTy End,
+                FuncTy Fn) {
+  detail::parallel_for_each_n(Begin, End, Fn);
+}
 #endif
+
+// Dynamic execution policy algorithms.
+template <class RandomAccessIterator,
+          class Comparator = detail::DefComparator<RandomAccessIterator>>
+void sort(const execution_policy &Policy, RandomAccessIterator Start,
+          RandomAccessIterator End, const Comparator &Comp = Comparator()) {
+  if (auto P = Policy.get<sequential_execution_policy>())
+    sort(*P, Start, End, Comp);
+  else if (auto P = Policy.get<parallel_execution_policy>())
+    sort(*P, Start, End, Comp);
+  else
+    llvm_unreachable("Unknown execution policy!");
+}
+
+template <class IterTy, class FuncTy>
+void for_each(const execution_policy &Policy, IterTy Begin, IterTy End,
+              FuncTy Fn) {
+  if (auto P = Policy.get<sequential_execution_policy>())
+    for_each(*P, Begin, End, Fn);
+  else if (auto P = Policy.get<parallel_execution_policy>())
+    for_each(*P, Begin, End, Fn);
+  else
+    llvm_unreachable("Unknown execution policy!");
+}
+
+template <class IndexTy, class FuncTy>
+void for_each_n(const execution_policy &Policy, IndexTy Begin, IndexTy End,
+                FuncTy Fn) {
+  if (auto P = Policy.get<sequential_execution_policy>())
+    for_each_n(*P, Begin, End, Fn);
+  else if (auto P = Policy.get<parallel_execution_policy>())
+    for_each_n(*P, Begin, End, Fn);
+  else
+    llvm_unreachable("Unknown execution policy!");
+}
+
+} // namespace parallel
 } // End namespace lld
 
 #endif // LLD_CORE_PARALLEL_H

lld/lib/ReaderWriter/MachO/LayoutPass.cpp

llvm::Error LayoutPass::perform(SimpleFile &mergedFile) {
   buildOrdinalOverrideMap(atomRange);
 
   DEBUG({
     llvm::dbgs() << "unsorted atoms:\n";
     printDefinedAtoms(atomRange);
   });
 
   std::vector<LayoutPass::SortKey> vec = decorate(atomRange);
-  parallel_sort(vec.begin(), vec.end(),
+  sort(parallel::par, vec.begin(), vec.end(),
        [&](const LayoutPass::SortKey &l, const LayoutPass::SortKey &r) -> bool {
          return compareAtoms(l, r, _customSorter);
        });
   DEBUG(checkTransitivity(vec, _customSorter));
   undecorate(atomRange, vec);
 
   DEBUG({
     llvm::dbgs() << "sorted atoms:\n";
     printDefinedAtoms(atomRange);
   });
 

lld/unittests/CoreTests/ParallelTest.cpp

 
 TEST(Parallel, sort) {
   std::mt19937 randEngine;
   std::uniform_int_distribution<uint32_t> dist;
 
   for (auto &i : array)
     i = dist(randEngine);
 
-  lld::parallel_sort(std::begin(array), std::end(array));
+  sort(lld::parallel::par, std::begin(array), std::end(array));
   ASSERT_TRUE(std::is_sorted(std::begin(array), std::end(array)));
 }
 
 TEST(Parallel, parallel_for) {
   // We need to test the case with a TaskSize > 1. We are white-box testing
   // here. The TaskSize is calculated as (End - Begin) / 1024 at the time of
   // writing.
   uint32_t range[2050];
   std::fill(range, range + 2050, 1);
-  lld::parallel_for(0, 2049, [&range](size_t I) { ++range[I]; });
+  for_each_n(lld::parallel::par, 0, 2049, [&range](size_t I) { ++range[I]; });
 
   uint32_t expected[2049];
   std::fill(expected, expected + 2049, 2);
   ASSERT_TRUE(std::equal(range, range + 2049, expected));
   // Check that we don't write past the end of the requested range.
   ASSERT_EQ(range[2049], 1u);
 }
+
+TEST(Paralle, parallel_dynamic_sort) {
+  std::mt19937 randEngine;
+  std::uniform_int_distribution<uint32_t> dist;
+
+  for (auto &i : array)
+    i = dist(randEngine);
+
+  lld::parallel::execution_policy policy = lld::parallel::par;
+  sort(policy, std::begin(array), std::end(array));
+  ASSERT_TRUE(std::is_sorted(std::begin(array), std::end(array)));
+}
+
+TEST(Paralle, sequential_dynamic_sort) {
+  std::mt19937 randEngine;
+  std::uniform_int_distribution<uint32_t> dist;
+
+  for (auto &i : array)
+    i = dist(randEngine);
+
+  lld::parallel::execution_policy policy = lld::parallel::seq;
+  sort(policy, std::begin(array), std::end(array));
+  ASSERT_TRUE(std::is_sorted(std::begin(array), std::end(array)));
+}
 No newline at end of file