coloredrangeexecutor.hh

Go to the documentation of this file.

// SPDX-FileCopyrightText: Copyright © DUNE Project contributors, see file AUTHORS.md
// SPDX-License-Identifier: LicenseRef-GPL-2.0-only-with-DUNE-exception OR LGPL-3.0-or-later
 
#ifndef DUNE_ASSEMBLER_PARALLEL_COLOREDRANGEEXECUTOR_HH
#define DUNE_ASSEMBLER_PARALLEL_COLOREDRANGEEXECUTOR_HH
 
#include <algorithm>
#include <cstddef>
#include <vector>
#include <thread>
#include <atomic>
 
#include <dune/common/iteratorrange.hh>
 
#include <dune/assembler/parallel/barrier.hh>
 
namespace Dune::Assembler::Experimental {
 
 
 
template <class Range>
static auto rangeSegment (Range& range, std::size_t i, std::size_t n) {
  std::size_t length = range.size() / n;
  std::size_t remainder = range.size() % n;
  auto begin = length*i + std::min(remainder, i);
  auto end = length*(i+1) + std::min(remainder, i+1);
  return Dune::IteratorRange(range.begin()+begin, range.begin()+end);
}
 
 
 
template <class ColoredRange>
class ColoredRangeExecutor
{
 
  class ThreadContext
  {
    std::size_t threadCount_;
    std::size_t threadIndex_;
    ColoredRange& coloredRange_;
    Dune::Assembler::Experimental::Barrier& barrier_;
    std::atomic<std::size_t>& counter_;
    bool useWorkStealing_;
    std::size_t minChunkSize_ = 8;
 
  public:
 
    ThreadContext (std::size_t threadCount, std::size_t threadIndex, ColoredRange& coloredRange, Barrier& barrier, std::atomic<std::size_t>& counter, bool useWorkStealing)
      : threadCount_(threadCount)
      , threadIndex_(threadIndex)
      , coloredRange_(coloredRange)
      , barrier_(barrier)
      , counter_(counter)
      , useWorkStealing_(useWorkStealing)
    {}
 
    std::size_t threadIndex () const
    {
      return threadIndex_;
    }
 
    std::size_t threadCount () const
    {
      return threadCount_;
    }
 
    Barrier& barrier () const
    {
      return barrier_;
    }
 
    // For each loop over the stored colored range.
    // Notice that the additional range argument is ignored.
    // It is required to ensure a common interface that can
    // also be implemented by the default constructible
    // SequentialExecutor. Ideally, this method would check
    // if the passed and stored range coincide and dispatch
    // to a sequential loop otherwise.
    template <class Range, class CallBack>
    void forEach (Range&&, CallBack&& callBack)
    {
      if (threadCount()>1)
      {
        if (useWorkStealing_)
        {
          for (auto&& singleColorRange : coloredRange_)
          {
            if (threadIndex()==0)
              counter_ = 0;
            barrier_.arrive_and_wait();
            auto size = singleColorRange.size();
            auto chunkSize = std::size_t(size/(2*threadCount()));
            while (true)
            {
              auto chunkStart = counter_.fetch_add(chunkSize, std::memory_order_relaxed);
              if (chunkStart>=size)
                break;
              for (std::size_t i : Dune::range(chunkStart, std::min(chunkStart+chunkSize,size)))
                callBack(singleColorRange[i]);
              chunkSize = std::max(chunkSize/2, minChunkSize_);
            }
            barrier_.arrive_and_wait();
          }
        }
        else
        {
          for (auto&& singleColorRange : coloredRange_)
          {
            for (auto&& entry : rangeSegment(singleColorRange, threadIndex_, threadCount_))
              callBack(entry);
            barrier_.arrive_and_wait();
          }
        }
      }
      else
        for (auto&& singleColorRange : coloredRange_)
          for (auto&& entry : singleColorRange)
            callBack(entry);
    }
 
  };
 
public:
 
  static constexpr bool isParallel = true;
 
  ColoredRangeExecutor (ColoredRange& coloredRange, std::size_t threadCount, bool useWorkStealing = true)
    : coloredRange_(coloredRange)
    , threadCount_(threadCount)
    , useWorkStealing_(useWorkStealing)
  {}
 
  template <class ThreadImpl>
  void operator() (ThreadImpl&& threadImpl) const {
    auto barrier = Barrier(threadCount_);
    auto counter = std::atomic<std::size_t>(0);
    auto threads = std::vector<std::thread>();
    for (std::size_t i=0; i<threadCount_; ++i)
      threads.emplace_back(threadImpl, ThreadContext(threadCount_, i, coloredRange_, barrier, counter, useWorkStealing_));
    for (auto& thread : threads)
      thread.join();
  }
 
private:
  ColoredRange& coloredRange_;
  std::size_t threadCount_;
  bool useWorkStealing_;
};
 
 
 
 
} // end namespace Dune::Assembler::Experimental
 
#endif // DUNE_ASSEMBLER_PARALLEL_COLOREDRANGEEXECUTOR_HH