itkMultiThreaderBase.h

Go to the documentation of this file.

/*=========================================================================
 *
 *  Copyright NumFOCUS
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *         https://www.apache.org/licenses/LICENSE-2.0.txt
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 *
 *=========================================================================*/
/*=========================================================================
 *
 *  Portions of this file are subject to the VTK Toolkit Version 3 copyright.
 *
 *  Copyright (c) Ken Martin, Will Schroeder, Bill Lorensen
 *
 *  For complete copyright, license and disclaimer of warranty information
 *  please refer to the NOTICE file at the top of the ITK source tree.
 *
 *=========================================================================*/
#ifndef itkMultiThreaderBase_h
#define itkMultiThreaderBase_h
 
#include "itkObject.h"
#include "itkThreadSupport.h"
#include "itkObjectFactory.h"
#include "itkIntTypes.h"
#include "itkImageRegion.h"
#include "itkImageIORegion.h"
#include "itkSingletonMacro.h"
#include <atomic>
#include <functional>
#include <thread>
#include "itkProgressReporter.h"
 
 
namespace itk
{
 
class MultiThreaderBaseEnums
{
public:
  enum class Threader : int8_t
  {
    Platform = 0,
    First = Platform,
    Pool,
    TBB,
    Last = TBB,
    Unknown = -1
  };
 
  enum class ThreadExitCode : uint8_t
  {
    SUCCESS,
    ITK_EXCEPTION,
    ITK_PROCESS_ABORTED_EXCEPTION,
    STD_EXCEPTION,
    UNKNOWN
  };
};
// Define how to print enumeration
extern ITKCommon_EXPORT std::ostream &
                        operator<<(std::ostream & out, const MultiThreaderBaseEnums::Threader value);
extern ITKCommon_EXPORT std::ostream &
                        operator<<(std::ostream & out, const MultiThreaderBaseEnums::ThreadExitCode value);
 
struct MultiThreaderBaseGlobals;
class ProcessObject;
 
class ITKCommon_EXPORT MultiThreaderBase : public Object
{
public:
  ITK_DISALLOW_COPY_AND_MOVE(MultiThreaderBase);
 
  using Self = MultiThreaderBase;
  using Superclass = Object;
  using Pointer = SmartPointer<Self>;
  using ConstPointer = SmartPointer<const Self>;
 
  static Pointer
  New();
 
  itkTypeMacro(MultiThreaderBase, Object);
 
  virtual void
  SetMaximumNumberOfThreads(ThreadIdType numberOfThreads);
  itkGetConstMacro(MaximumNumberOfThreads, ThreadIdType);
  virtual void
  SetNumberOfWorkUnits(ThreadIdType numberOfWorkUnits);
  itkGetConstMacro(NumberOfWorkUnits, ThreadIdType);
  virtual void
  SetUpdateProgress(bool updates);
  itkGetConstMacro(UpdateProgress, bool);
 
  static void
  SetGlobalMaximumNumberOfThreads(ThreadIdType val);
  static ThreadIdType
  GetGlobalMaximumNumberOfThreads();
  itkLegacyMacro(static void SetGlobalDefaultUseThreadPool(const bool GlobalDefaultUseThreadPool));
  itkLegacyMacro(static bool GetGlobalDefaultUseThreadPool());
  using ThreaderEnum = MultiThreaderBaseEnums::Threader;
#if !defined(ITK_LEGACY_REMOVE)
  using ThreaderType = ThreaderEnum;
  static constexpr ThreaderEnum Platform = ThreaderEnum::Platform;
  static constexpr ThreaderEnum First = ThreaderEnum::First;
  static constexpr ThreaderEnum Pool = ThreaderEnum::Pool;
  static constexpr ThreaderEnum TBB = ThreaderEnum::TBB;
  static constexpr ThreaderEnum Last = ThreaderEnum::Last;
  static constexpr ThreaderEnum Unknown = ThreaderEnum::Unknown;
#endif
 
  static ThreaderEnum
  ThreaderTypeFromString(std::string threaderString);
 
  static std::string
  ThreaderTypeToString(ThreaderEnum threader)
  {
    switch (threader)
    {
      case ThreaderEnum::Platform:
        return "Platform";
        break;
      case ThreaderEnum::Pool:
        return "Pool";
        break;
      case ThreaderEnum::TBB:
        return "TBB";
        break;
      case ThreaderEnum::Unknown:
      default:
        return "Unknown";
        break;
    }
  }
  static void
  SetGlobalDefaultThreader(ThreaderEnum threaderType);
  static ThreaderEnum
  GetGlobalDefaultThreader();
  static void
  SetGlobalDefaultNumberOfThreads(ThreadIdType val);
  static ThreadIdType
  GetGlobalDefaultNumberOfThreads();
#if !defined(ITK_LEGACY_REMOVE)
 
  itkLegacyMacro(virtual void SetNumberOfThreads(ThreadIdType numberOfThreads))
  {
    this->SetMaximumNumberOfThreads(numberOfThreads);
    this->SetNumberOfWorkUnits(this->GetMaximumNumberOfThreads()); // Might be clamped
  }
  itkLegacyMacro(virtual ThreadIdType GetNumberOfThreads()) { return this->GetNumberOfWorkUnits(); }
  // clang-format off
ITK_GCC_PRAGMA_DIAG_PUSH()
ITK_GCC_PRAGMA_DIAG(ignored "-Wattributes")
INTEL_PRAGMA_WARN_PUSH
INTEL_SUPPRESS_warning_1292
CLANG_PRAGMA_PUSH
CLANG_SUPPRESS_Wcpp14_extensions
  // clang-format on
#  ifdef ITK_LEGACY_SILENT
    struct ThreadInfoStruct
#  else
    struct [[deprecated("Use WorkUnitInfo, ThreadInfoStruct is deprecated since ITK 5.0")]] ThreadInfoStruct
#  endif
      // clang-format off
CLANG_PRAGMA_POP
INTEL_PRAGMA_WARN_POP
  // clang-format on
  {
    ThreadIdType       ThreadID;
    ThreadIdType       NumberOfThreads;
    void *             UserData;
    ThreadFunctionType ThreadFunction;
    enum
    {
      SUCCESS,
      ITK_EXCEPTION,
      ITK_PROCESS_ABORTED_EXCEPTION,
      STD_EXCEPTION,
      UNKNOWN
    } ThreadExitCode;
  };
  // clang-format off
ITK_GCC_PRAGMA_DIAG_POP()
  // clang-format on
#endif // ITK_LEGACY_REMOVE
 
  struct WorkUnitInfo
  {
    ThreadIdType       WorkUnitID;
    ThreadIdType       NumberOfWorkUnits;
    void *             UserData;
    ThreadFunctionType ThreadFunction;
    using ThreadExitCodeEnum = MultiThreaderBaseEnums::ThreadExitCode;
    ThreadExitCodeEnum ThreadExitCode;
#if !defined(ITK_LEGACY_REMOVE)
    using ThreadExitCodeType = ThreadExitCodeEnum;
    static constexpr ThreadExitCodeEnum SUCCESS = ThreadExitCodeEnum::SUCCESS;
    static constexpr ThreadExitCodeEnum ITK_EXCEPTION = ThreadExitCodeEnum::ITK_EXCEPTION;
    static constexpr ThreadExitCodeEnum ITK_PROCESS_ABORTED_EXCEPTION =
      ThreadExitCodeEnum::ITK_PROCESS_ABORTED_EXCEPTION;
    static constexpr ThreadExitCodeEnum STD_EXCEPTION = ThreadExitCodeEnum::STD_EXCEPTION;
    static constexpr ThreadExitCodeEnum UNKNOWN = ThreadExitCodeEnum::UNKNOWN;
#endif
  };
  virtual void
  SingleMethodExecute() = 0;
 
  virtual void
  SetSingleMethod(ThreadFunctionType, void * data) = 0;
 
  template <unsigned int VDimension>
  using TemplatedThreadingFunctorType = std::function<void(const ImageRegion<VDimension> &)>;
  using ThreadingFunctorType = std::function<void(const IndexValueType index[], const SizeValueType size[])>;
  using ArrayThreadingFunctorType = std::function<void(SizeValueType)>;
 
  virtual void
  ParallelizeArray(SizeValueType             firstIndex,
                   SizeValueType             lastIndexPlus1,
                   ArrayThreadingFunctorType aFunc,
                   ProcessObject *           filter);
 
  template <unsigned int VDimension>
  ITK_TEMPLATE_EXPORT void
  ParallelizeImageRegion(const ImageRegion<VDimension> &           requestedRegion,
                         TemplatedThreadingFunctorType<VDimension> funcP,
                         ProcessObject *                           filter)
  {
    this->ParallelizeImageRegion(
      VDimension,
      requestedRegion.GetIndex().m_InternalArray,
      requestedRegion.GetSize().m_InternalArray,
      [funcP](const IndexValueType index[], const SizeValueType size[]) {
        ImageRegion<VDimension> region;
        for (unsigned int d = 0; d < VDimension; ++d)
        {
          region.SetIndex(d, index[d]);
          region.SetSize(d, size[d]);
        }
        funcP(region);
      },
      filter);
  }
  template <unsigned int VDimension>
  ITK_TEMPLATE_EXPORT void
  ParallelizeImageRegionRestrictDirection(unsigned int                              restrictedDirection,
                                          const ImageRegion<VDimension> &           requestedRegion,
                                          TemplatedThreadingFunctorType<VDimension> funcP,
                                          ProcessObject *                           filter)
  {
    if (VDimension <= 1) // Cannot split, no parallelization
    {
 
      ProgressReporter progress(filter, 0, requestedRegion.GetNumberOfPixels());
      funcP(requestedRegion);
    }
    else // Can split, parallelize!
    {
      constexpr unsigned int SplitDimension = (VDimension - 1) ? VDimension - 1 : VDimension;
      using SplitRegionType = ImageRegion<SplitDimension>;
 
      SplitRegionType splitRegion;
      for (unsigned int splitDimension = 0, dimension = 0; dimension < VDimension; ++dimension)
      {
        if (dimension == restrictedDirection)
        {
          continue;
        }
        splitRegion.SetIndex(splitDimension, requestedRegion.GetIndex(dimension));
        splitRegion.SetSize(splitDimension, requestedRegion.GetSize(dimension));
        ++splitDimension;
      }
 
      this->ParallelizeImageRegion(
        SplitDimension,
        splitRegion.GetIndex().m_InternalArray,
        splitRegion.GetSize().m_InternalArray,
        [&](const IndexValueType index[], const SizeValueType size[]) {
          ImageRegion<VDimension> restrictedRequestedRegion;
          restrictedRequestedRegion.SetIndex(restrictedDirection, requestedRegion.GetIndex(restrictedDirection));
          restrictedRequestedRegion.SetSize(restrictedDirection, requestedRegion.GetSize(restrictedDirection));
          for (unsigned int splitDimension = 0, dimension = 0; dimension < VDimension; ++dimension)
          {
            if (dimension == restrictedDirection)
            {
              continue;
            }
            restrictedRequestedRegion.SetIndex(dimension, index[splitDimension]);
            restrictedRequestedRegion.SetSize(dimension, size[splitDimension]);
            ++splitDimension;
          }
          funcP(restrictedRequestedRegion);
        },
        filter);
    }
  }
 
  virtual void
  ParallelizeImageRegion(unsigned int         dimension,
                         const IndexValueType index[],
                         const SizeValueType  size[],
                         ThreadingFunctorType funcP,
                         ProcessObject *      filter);
 
protected:
  MultiThreaderBase();
  ~MultiThreaderBase() override;
  void
  PrintSelf(std::ostream & os, Indent indent) const override;
 
  struct ArrayCallback
  {
    ArrayThreadingFunctorType functor;
    const SizeValueType       firstIndex;
    const SizeValueType       lastIndexPlus1;
    ProcessObject *           filter;
  };
 
  static ITK_THREAD_RETURN_FUNCTION_CALL_CONVENTION
  ParallelizeArrayHelper(void * arg);
 
  struct RegionAndCallback
  {
    ThreadingFunctorType   functor;
    unsigned int           dimension;
    const IndexValueType * index;
    const SizeValueType *  size;
    ProcessObject *        filter;
  };
 
  static ITK_THREAD_RETURN_FUNCTION_CALL_CONVENTION
  ParallelizeImageRegionHelper(void * arg);
 
  ThreadIdType m_NumberOfWorkUnits;
 
  ThreadIdType m_MaximumNumberOfThreads;
 
  static ITK_THREAD_RETURN_FUNCTION_CALL_CONVENTION
  SingleMethodProxy(void * arg);
 
  ThreadFunctionType m_SingleMethod{ nullptr };
 
  void * m_SingleData{ nullptr };
 
private:
  static void
  SetGlobalDefaultThreaderPrivate(ThreaderEnum threaderType);
  static ThreaderEnum
  GetGlobalDefaultThreaderPrivate();
 
  itkGetGlobalDeclarationMacro(MultiThreaderBaseGlobals, PimplGlobals);
 
  std::atomic<bool> m_UpdateProgress{ true };
 
  static MultiThreaderBaseGlobals * m_PimplGlobals;
  friend class ProcessObject;
 
  static ThreadIdType
  GetGlobalDefaultNumberOfThreadsByPlatform();
};
 
 
} // namespace itk
 
#endif