itkThreadPool.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.
 *
 *=========================================================================*/
#ifndef itkThreadPool_h
#define itkThreadPool_h

#include "itkConfigure.h"
#include "itkIntTypes.h"

#include "itkThreadSupport.h"

#if defined(ITK_USE_PTHREADS)
#include <pthread.h>
#include <semaphore.h>
#include <unistd.h> // for sleep
#elif defined(ITK_USE_WIN32_THREADS)
#include <windows.h>
#endif

#if defined __APPLE__
#include <mach/mach_init.h>
#include <mach/mach_error.h>
#include <mach/semaphore.h>
#include <mach/task.h>
#include <mach/task_info.h>
#endif

#include <map>
#include <set>

#include "itkThreadJob.h"
#include "itkObject.h"
#include "itkObjectFactory.h"
#include "itkSimpleFastMutexLock.h"
#include "itkMutexLockHolder.h"

namespace itk
{

class ITKCommon_EXPORT ThreadPool : public Object
{
public:

  typedef ThreadPool               Self;
  typedef Object                   Superclass;
  typedef SmartPointer< Self >     Pointer;
  typedef SmartPointer<const Self> ConstPointer;

  typedef int                  ThreadTimeType;
  typedef unsigned int         ThreadCountType;
  typedef ThreadJob::JobIdType ThreadJobIdType;

  itkTypeMacro(ThreadPool, Object);

  static Pointer New();

  static Pointer GetInstance();

  ThreadProcessIdType AssignWork(ThreadJob worker);

  void InitializeThreads(ThreadCountType maxThreads);

  bool WaitForJobOnThreadHandle(ThreadProcessIdType handle);

protected:
  class ThreadSemaphorePair
  {
  public:
    ThreadSemaphorePair(const ThreadProcessIdType & tph);
    int SemaphoreWait();
    int SemaphorePost();
#if defined(__APPLE__)
    semaphore_t           m_Semaphore;
#elif defined(_WIN32) || defined(_WIN64)
    HANDLE               m_Semaphore;
#elif defined(ITK_USE_PTHREADS)
    sem_t                m_Semaphore;
#endif

    ThreadProcessIdType  m_ThreadProcessHandle;

  private:
    static int m_SemaphoreCount;
    ThreadSemaphorePair(); //purposefully not implemented.
  };

  ThreadPool();  // Protected so that only the GetThreadPool can create a thread
                 // pool
  virtual ~ThreadPool();

private:
  ThreadPool(ThreadPool const &) ITK_DELETE_FUNCTION;

  ThreadPool & operator=(ThreadPool const &) ITK_DELETE_FUNCTION;

  bool m_ScheduleForDestruction;

  ThreadCountType m_ThreadCount;

  static SimpleFastMutexLock m_NumberOfPendingJobsToBeRunMutex;

  unsigned int m_IdCounter;

  bool m_ExceptionOccurred;

  typedef std::map<ThreadJobIdType,ThreadJob> ThreadJobContainerType;

  typedef std::pair<ThreadJobIdType,ThreadJob>          ThreadJobContainerPairType;

  typedef std::set<ThreadJobIdType>      ThreadJobIdsContainerType;

  typedef std::set<ThreadProcessIdType>  ThreadProcessIdContainerType;

  ThreadJobContainerType m_WorkerQueue;

  static SimpleFastMutexLock m_WorkerQueueMutex;

  ThreadProcessIdContainerType m_ThreadHandles;

  enum {
    JOB_THREADHANDLE_JUST_ADDED=-3, //means thread just added
    JOB_THREADHANDLE_IS_FREE=-2,    // means this particular threadhandle
                                    // (thread) is free
    JOB_THREADHANDLE_IS_DONE=-1     // means the thread finished with the
                                    // assigned job and is waiting until
                                    // WaitForJobOnThreadHandle method is called
                                    // otherwise threadhandle is actively
                                    // running a job.
    };

  class ThreadProcessIdentifiers
  {
public:
#if defined(ITK_USE_WIN32_THREADS)
    typedef DWORD WinThreadIdType;
#else
    typedef unsigned long WinThreadIdType;
#endif
    ThreadProcessIdentifiers(const int tnid,
                            const ThreadProcessIdType tph,
                            const WinThreadIdType winThreadId) :
      m_ThreadNumericId(tnid),
      m_ThreadProcessHandle(tph),
      m_WinThreadId(winThreadId)
    {
    }

    int                 m_ThreadNumericId;
    ThreadProcessIdType m_ThreadProcessHandle;
    WinThreadIdType     m_WinThreadId;
private:
    ThreadProcessIdentifiers(); //purposefully not implemented.
  };

  // the ThreadProcessIdentifiersVector ThreadSemHandlePairingQueue and
  // ThreadSemHandlePairingForWaitQueue really want to be STL
  // containers with more efficient searchability than O(N) But the
  // pthread_t type isn't guaranteed to be an integral type so it
  // can't be used as a Key on a keyed STL container.
  typedef std::vector<ThreadProcessIdentifiers> ThreadProcessIdentifiersVecType;
  ThreadProcessIdentifiersVecType m_ThreadProcessIdentifiersVector;

  //ThreadSemaphorePair *m_ThreadSemHandlePairingQueue;
  typedef std::vector<ThreadSemaphorePair *> ThreadSemHandlePairingQueueType;

  ThreadSemHandlePairingQueueType m_ThreadSemHandlePairingQueue;
  ThreadSemHandlePairingQueueType m_ThreadSemHandlePairingForWaitQueue;

  static SimpleFastMutexLock m_ThreadProcessIdentifiersVectorMutex;

  const ThreadJob &FetchWork(ThreadProcessIdType t);

  void RemoveActiveId(ThreadJobIdType id);

  void AddThread();

  ThreadProcessIdentifiers *FindThreadToRun();

  static Pointer m_ThreadPoolInstance;
  static SimpleFastMutexLock m_ThreadPoolInstanceMutex;

  ThreadSemaphorePair* GetSemaphore(ThreadSemHandlePairingQueueType &q,ThreadProcessIdType threadHandle);

  ThreadSemaphorePair* GetSemaphoreForThread(ThreadProcessIdType threadHandle);

  ThreadSemaphorePair* GetSemaphoreForThreadWait(ThreadProcessIdType threadHandle);

  void DeallocateThreadSemSet(ThreadSemHandlePairingQueueType &q);

  static void * ThreadExecute(void *param);

  static bool CompareThreadHandles(ThreadProcessIdType t1, ThreadProcessIdType t2);

  ThreadProcessIdType GetThreadHandleForThreadId(ThreadIdType id);


};

}
#endif