itkMultiThreaderBase.h

Go to the documentation of this file.

/*=========================================================================
 *
 *  Copyright Insight Software Consortium
 *
 *  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
 *
 *         http://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 <functional>
#include <thread>


namespace itk
{
struct MultiThreaderBaseGlobals;
class ProcessObject;

class ITKCommon_EXPORT MultiThreaderBase : public Object
{
public:
  ITK_DISALLOW_COPY_AND_ASSIGN(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 );

  static void SetGlobalMaximumNumberOfThreads(ThreadIdType val);
  static ThreadIdType GetGlobalMaximumNumberOfThreads();

  itkLegacyMacro(static void SetGlobalDefaultUseThreadPool( const bool GlobalDefaultUseThreadPool ));
  itkLegacyMacro(static bool GetGlobalDefaultUseThreadPool( ));

  enum ThreaderType { Platform = 0, First = Platform, Pool, TBB, Last = TBB, Unknown = -1 };

  static ThreaderType ThreaderTypeFromString(std::string threaderString);

  static std::string ThreaderTypeToString(ThreaderType threader)
  {
    switch (threader)
      {
      case ThreaderType::Platform:
        return "Platform";
        break;
      case ThreaderType::Pool:
        return "Pool";
        break;
      case ThreaderType::TBB:
        return "TBB";
        break;
      case ThreaderType::Unknown:
      default:
        return "Unknown";
        break;
      }
  }

  static void SetGlobalDefaultThreader(ThreaderType threaderType);
  static ThreaderType 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();
  }

#ifdef ITK_HAS_GCC_PRAGMA_DIAG_PUSHPOP
  ITK_GCC_PRAGMA_DIAG_PUSH()
#endif
ITK_GCC_PRAGMA_DIAG(ignored "-Wattributes")
INTEL_PRAGMA_WARN_PUSH
INTEL_SUPPRESS_warning_1292
CLANG_PRAGMA_PUSH
CLANG_SUPPRESS_Wc__14_extensions
  struct [[deprecated( "Use WorkUnitInfo, ThreadInfoStruct is deprecated since ITK 5.0" )]] ThreadInfoStruct
CLANG_PRAGMA_POP
INTEL_PRAGMA_WARN_POP
  {
    ThreadIdType ThreadID;
    ThreadIdType NumberOfThreads;
    void* UserData;
    ThreadFunctionType ThreadFunction;
    enum { SUCCESS, ITK_EXCEPTION, ITK_PROCESS_ABORTED_EXCEPTION, STD_EXCEPTION, UNKNOWN } ThreadExitCode;
    };
#ifdef ITK_HAS_GCC_PRAGMA_DIAG_PUSHPOP
  ITK_GCC_PRAGMA_DIAG_POP()
#else
  ITK_GCC_PRAGMA_DIAG(warning "-Wattributes")
#endif
#endif //ITK_LEGACY_REMOVE


  struct WorkUnitInfo
  {
    ThreadIdType       WorkUnitID;
    ThreadIdType       NumberOfWorkUnits;
    void*              UserData;
    ThreadFunctionType ThreadFunction;
    enum
    {
      SUCCESS,
      ITK_EXCEPTION,
      ITK_PROCESS_ABORTED_EXCEPTION,
      STD_EXCEPTION,
      UNKNOWN
    } ThreadExitCode;
  };

  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
      {
      MultiThreaderBase::HandleFilterProgress( filter, 0.0f );
      funcP(requestedRegion);
      MultiThreaderBase::HandleFilterProgress( filter, 1.0f );
      }
    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);

  static void HandleFilterProgress(ProcessObject *filter, float progress = -1.0f);

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;
    std::thread::id callingThread;
    std::atomic<SizeValueType> progress;
  };

  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;
    std::thread::id callingThread;
    SizeValueType pixelCount;
    std::atomic<SizeValueType> pixelProgress;
  };

  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;

  void *m_SingleData;

private:

  itkGetGlobalDeclarationMacro(MultiThreaderBaseGlobals, PimplGlobals);

  static MultiThreaderBaseGlobals * m_PimplGlobals;
  friend class ProcessObject;

  static ThreadIdType GetGlobalDefaultNumberOfThreadsByPlatform();

};

ITKCommon_EXPORT std::ostream& operator << (std::ostream& os,
    const MultiThreaderBase::ThreaderType& threader);

} // end namespace itk
#endif