llvm/Support/Parallel.h

0001 //===- llvm/Support/Parallel.h - Parallel algorithms ----------------------===//
0002 //
0003 // Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
0004 // See https://llvm.org/LICENSE.txt for license information.
0005 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
0006 //
0007 //===----------------------------------------------------------------------===//
0008
0009 #ifndef LLVM_SUPPORT_PARALLEL_H
0010 #define LLVM_SUPPORT_PARALLEL_H
0011
0012 #include "llvm/ADT/STLExtras.h"
0013 #include "llvm/Config/llvm-config.h"
0014 #include "llvm/Support/Error.h"
0015 #include "llvm/Support/MathExtras.h"
0016 #include "llvm/Support/Threading.h"
0017
0018 #include <algorithm>
0019 #include <condition_variable>
0020 #include <functional>
0021 #include <mutex>
0022
0023 namespace llvm {
0024
0025 namespace parallel {
0026
0027 // Strategy for the default executor used by the parallel routines provided by
0028 // this file. It defaults to using all hardware threads and should be
0029 // initialized before the first use of parallel routines.
0030 extern ThreadPoolStrategy strategy;
0031
0032 #if LLVM_ENABLE_THREADS
0033 #define GET_THREAD_INDEX_IMPL                                                  \
0034   if (parallel::strategy.ThreadsRequested == 1)                                \
0035     return 0;                                                                  \
0036   assert((threadIndex != UINT_MAX) &&                                          \
0037          "getThreadIndex() must be called from a thread created by "           \
0038          "ThreadPoolExecutor");                                                \
0039   return threadIndex;
0040
0041 #ifdef _WIN32
0042 // Direct access to thread_local variables from a different DLL isn't
0043 // possible with Windows Native TLS.
0044 unsigned getThreadIndex();
0045 #else
0046 // Don't access this directly, use the getThreadIndex wrapper.
0047 extern thread_local unsigned threadIndex;
0048
0049 inline unsigned getThreadIndex() { GET_THREAD_INDEX_IMPL; }
0050 #endif
0051
0052 size_t getThreadCount();
0053 #else
0054 inline unsigned getThreadIndex() { return 0; }
0055 inline size_t getThreadCount() { return 1; }
0056 #endif
0057
0058 namespace detail {
0059 class Latch {
0060   uint32_t Count;
0061   mutable std::mutex Mutex;
0062   mutable std::condition_variable Cond;
0063
0064 public:
0065   explicit Latch(uint32_t Count = 0) : Count(Count) {}
0066   ~Latch() {
0067     // Ensure at least that sync() was called.
0068     assert(Count == 0);
0069   }
0070
0071   void inc() {
0072     std::lock_guard<std::mutex> lock(Mutex);
0073     ++Count;
0074   }
0075
0076   void dec() {
0077     std::lock_guard<std::mutex> lock(Mutex);
0078     if (--Count == 0)
0079       Cond.notify_all();
0080   }
0081
0082   void sync() const {
0083     std::unique_lock<std::mutex> lock(Mutex);
0084     Cond.wait(lock, [&] { return Count == 0; });
0085   }
0086 };
0087 } // namespace detail
0088
0089 class TaskGroup {
0090   detail::Latch L;
0091   bool Parallel;
0092
0093 public:
0094   TaskGroup();
0095   ~TaskGroup();
0096
0097   // Spawn a task, but does not wait for it to finish.
0098   // Tasks marked with \p Sequential will be executed
0099   // exactly in the order which they were spawned.
0100   void spawn(std::function<void()> f);
0101
0102   void sync() const { L.sync(); }
0103
0104   bool isParallel() const { return Parallel; }
0105 };
0106
0107 namespace detail {
0108
0109 #if LLVM_ENABLE_THREADS
0110 const ptrdiff_t MinParallelSize = 1024;
0111
0112 /// Inclusive median.
0113 template <class RandomAccessIterator, class Comparator>
0114 RandomAccessIterator medianOf3(RandomAccessIterator Start,
0115                                RandomAccessIterator End,
0116                                const Comparator &Comp) {
0117   RandomAccessIterator Mid = Start + (std::distance(Start, End) / 2);
0118   return Comp(*Start, *(End - 1))
0119              ? (Comp(*Mid, *(End - 1)) ? (Comp(*Start, *Mid) ? Mid : Start)
0120                                        : End - 1)
0121              : (Comp(*Mid, *Start) ? (Comp(*(End - 1), *Mid) ? Mid : End - 1)
0122                                    : Start);
0123 }
0124
0125 template <class RandomAccessIterator, class Comparator>
0126 void parallel_quick_sort(RandomAccessIterator Start, RandomAccessIterator End,
0127                          const Comparator &Comp, TaskGroup &TG, size_t Depth) {
0128   // Do a sequential sort for small inputs.
0129   if (std::distance(Start, End) < detail::MinParallelSize || Depth == 0) {
0130     llvm::sort(Start, End, Comp);
0131     return;
0132   }
0133
0134   // Partition.
0135   auto Pivot = medianOf3(Start, End, Comp);
0136   // Move Pivot to End.
0137   std::swap(*(End - 1), *Pivot);
0138   Pivot = std::partition(Start, End - 1, [&Comp, End](decltype(*Start) V) {
0139     return Comp(V, *(End - 1));
0140   });
0141   // Move Pivot to middle of partition.
0142   std::swap(*Pivot, *(End - 1));
0143
0144   // Recurse.
0145   TG.spawn([=, &Comp, &TG] {
0146     parallel_quick_sort(Start, Pivot, Comp, TG, Depth - 1);
0147   });
0148   parallel_quick_sort(Pivot + 1, End, Comp, TG, Depth - 1);
0149 }
0150
0151 template <class RandomAccessIterator, class Comparator>
0152 void parallel_sort(RandomAccessIterator Start, RandomAccessIterator End,
0153                    const Comparator &Comp) {
0154   TaskGroup TG;
0155   parallel_quick_sort(Start, End, Comp, TG,
0156                       llvm::Log2_64(std::distance(Start, End)) + 1);
0157 }
0158
0159 // TaskGroup has a relatively high overhead, so we want to reduce
0160 // the number of spawn() calls. We'll create up to 1024 tasks here.
0161 // (Note that 1024 is an arbitrary number. This code probably needs
0162 // improving to take the number of available cores into account.)
0163 enum { MaxTasksPerGroup = 1024 };
0164
0165 template <class IterTy, class ResultTy, class ReduceFuncTy,
0166           class TransformFuncTy>
0167 ResultTy parallel_transform_reduce(IterTy Begin, IterTy End, ResultTy Init,
0168                                    ReduceFuncTy Reduce,
0169                                    TransformFuncTy Transform) {
0170   // Limit the number of tasks to MaxTasksPerGroup to limit job scheduling
0171   // overhead on large inputs.
0172   size_t NumInputs = std::distance(Begin, End);
0173   if (NumInputs == 0)
0174     return std::move(Init);
0175   size_t NumTasks = std::min(static_cast<size_t>(MaxTasksPerGroup), NumInputs);
0176   std::vector<ResultTy> Results(NumTasks, Init);
0177   {
0178     // Each task processes either TaskSize or TaskSize+1 inputs. Any inputs
0179     // remaining after dividing them equally amongst tasks are distributed as
0180     // one extra input over the first tasks.
0181     TaskGroup TG;
0182     size_t TaskSize = NumInputs / NumTasks;
0183     size_t RemainingInputs = NumInputs % NumTasks;
0184     IterTy TBegin = Begin;
0185     for (size_t TaskId = 0; TaskId < NumTasks; ++TaskId) {
0186       IterTy TEnd = TBegin + TaskSize + (TaskId < RemainingInputs ? 1 : 0);
0187       TG.spawn([=, &Transform, &Reduce, &Results] {
0188         // Reduce the result of transformation eagerly within each task.
0189         ResultTy R = Init;
0190         for (IterTy It = TBegin; It != TEnd; ++It)
0191           R = Reduce(R, Transform(*It));
0192         Results[TaskId] = R;
0193       });
0194       TBegin = TEnd;
0195     }
0196     assert(TBegin == End);
0197   }
0198
0199   // Do a final reduction. There are at most 1024 tasks, so this only adds
0200   // constant single-threaded overhead for large inputs. Hopefully most
0201   // reductions are cheaper than the transformation.
0202   ResultTy FinalResult = std::move(Results.front());
0203   for (ResultTy &PartialResult :
0204        MutableArrayRef(Results.data() + 1, Results.size() - 1))
0205     FinalResult = Reduce(FinalResult, std::move(PartialResult));
0206   return std::move(FinalResult);
0207 }
0208
0209 #endif
0210
0211 } // namespace detail
0212 } // namespace parallel
0213
0214 template <class RandomAccessIterator,
0215           class Comparator = std::less<
0216               typename std::iterator_traits<RandomAccessIterator>::value_type>>
0217 void parallelSort(RandomAccessIterator Start, RandomAccessIterator End,
0218                   const Comparator &Comp = Comparator()) {
0219 #if LLVM_ENABLE_THREADS
0220   if (parallel::strategy.ThreadsRequested != 1) {
0221     parallel::detail::parallel_sort(Start, End, Comp);
0222     return;
0223   }
0224 #endif
0225   llvm::sort(Start, End, Comp);
0226 }
0227
0228 void parallelFor(size_t Begin, size_t End, function_ref<void(size_t)> Fn);
0229
0230 template <class IterTy, class FuncTy>
0231 void parallelForEach(IterTy Begin, IterTy End, FuncTy Fn) {
0232   parallelFor(0, End - Begin, [&](size_t I) { Fn(Begin[I]); });
0233 }
0234
0235 template <class IterTy, class ResultTy, class ReduceFuncTy,
0236           class TransformFuncTy>
0237 ResultTy parallelTransformReduce(IterTy Begin, IterTy End, ResultTy Init,
0238                                  ReduceFuncTy Reduce,
0239                                  TransformFuncTy Transform) {
0240 #if LLVM_ENABLE_THREADS
0241   if (parallel::strategy.ThreadsRequested != 1) {
0242     return parallel::detail::parallel_transform_reduce(Begin, End, Init, Reduce,
0243                                                        Transform);
0244   }
0245 #endif
0246   for (IterTy I = Begin; I != End; ++I)
0247     Init = Reduce(std::move(Init), Transform(*I));
0248   return std::move(Init);
0249 }
0250
0251 // Range wrappers.
0252 template <class RangeTy,
0253           class Comparator = std::less<decltype(*std::begin(RangeTy()))>>
0254 void parallelSort(RangeTy &&R, const Comparator &Comp = Comparator()) {
0255   parallelSort(std::begin(R), std::end(R), Comp);
0256 }
0257
0258 template <class RangeTy, class FuncTy>
0259 void parallelForEach(RangeTy &&R, FuncTy Fn) {
0260   parallelForEach(std::begin(R), std::end(R), Fn);
0261 }
0262
0263 template <class RangeTy, class ResultTy, class ReduceFuncTy,
0264           class TransformFuncTy>
0265 ResultTy parallelTransformReduce(RangeTy &&R, ResultTy Init,
0266                                  ReduceFuncTy Reduce,
0267                                  TransformFuncTy Transform) {
0268   return parallelTransformReduce(std::begin(R), std::end(R), Init, Reduce,
0269                                  Transform);
0270 }
0271
0272 // Parallel for-each, but with error handling.
0273 template <class RangeTy, class FuncTy>
0274 Error parallelForEachError(RangeTy &&R, FuncTy Fn) {
0275   // The transform_reduce algorithm requires that the initial value be copyable.
0276   // Error objects are uncopyable. We only need to copy initial success values,
0277   // so work around this mismatch via the C API. The C API represents success
0278   // values with a null pointer. The joinErrors discards null values and joins
0279   // multiple errors into an ErrorList.
0280   return unwrap(parallelTransformReduce(
0281       std::begin(R), std::end(R), wrap(Error::success()),
0282       [](LLVMErrorRef Lhs, LLVMErrorRef Rhs) {
0283         return wrap(joinErrors(unwrap(Lhs), unwrap(Rhs)));
0284       },
0285       [&Fn](auto &&V) { return wrap(Fn(V)); }));
0286 }
0287
0288 } // namespace llvm
0289
0290 #endif // LLVM_SUPPORT_PARALLEL_H