threadpool.h

Go to the documentation of this file.

/*
 * threadpool.h
 *
 * Generalised Thread Pooling functions
 *
 * Portable Tools Library
 *
 * Copyright (C) 2009 Post Increment
 *
 * The contents of this file are subject to the Mozilla Public License
 * Version 1.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.mozilla.org/MPL/
 *
 * Software distributed under the License is distributed on an "AS IS"
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
 * the License for the specific language governing rights and limitations
 * under the License.
 *
 * The Original Code is Portable Windows Library.
 *
 * The Initial Developer of the Original Code is Post Increment
 *
 * Portions of this code were written with the financial assistance of 
 * Metreos Corporation (http://www.metros.com).
 *
 * Contributor(s): ______________________________________.
 *
 * $Revision: 32255 $
 * $Author: rjongbloed $
 * $Date: 2014-06-29 17:49:31 +1000 (Sun, 29 Jun 2014) $
 */


#ifndef PTLIB_THREADPOOL_H
#define PTLIB_THREADPOOL_H

#ifdef P_USE_PRAGMA
#pragma interface
#endif


#include <ptlib/thread.h>
#include <ptlib/safecoll.h>
#include <map>
#include <queue>


class PThreadPoolBase : public PObject
{
  public:
    class WorkerThreadBase : public PThread
    {
      protected:
        WorkerThreadBase(Priority priority, const char * threadName)
          : PThread(100, NoAutoDeleteThread, priority, threadName)
          , m_shutdown(false)
        { }

      public:
        virtual void Shutdown() = 0;
        virtual unsigned GetWorkSize() const = 0;

        bool   m_shutdown;
        PMutex m_workerMutex;
    };

    class InternalWorkBase
    {
      public:
        InternalWorkBase(const char * group)
        { 
          if (group != NULL)
            m_group = group;
        }
        std::string m_group;
    };

    ~PThreadPoolBase();

    virtual WorkerThreadBase * CreateWorkerThread() = 0;
    virtual WorkerThreadBase * AllocateWorker();
    virtual WorkerThreadBase * NewWorker();

    unsigned GetMaxWorkers() const { return m_maxWorkerCount; }

    void SetMaxWorkers(
      unsigned count
    ) { m_maxWorkerCount = count; }

    unsigned GetMaxUnits() const { return m_maxWorkUnitCount; }

    void SetMaxUnits(
      unsigned count
    ) { m_maxWorkUnitCount = count; }

  protected:
    PThreadPoolBase(
      unsigned maxWorkerCount,
      unsigned maxWorkUnitCount,
      const char * threadName,
      PThread::Priority priority
    );

    virtual bool CheckWorker(WorkerThreadBase * worker);
    void StopWorker(WorkerThreadBase * worker);
    PMutex m_listMutex;

    typedef std::vector<WorkerThreadBase *> WorkerList_t;
    WorkerList_t m_workers;

    unsigned          m_maxWorkerCount;
    unsigned          m_maxWorkUnitCount;
    unsigned          m_highWaterMark; // For logging
    PString           m_threadName;
    PThread::Priority m_priority;
};


template <class Work_T>
class PThreadPool : public PThreadPoolBase
{
  PCLASSINFO(PThreadPool, PThreadPoolBase);
  public:
    //
    //  constructor
    //
    PThreadPool(
      unsigned maxWorkers = 10,
      unsigned maxWorkUnits = 0,
      const char * threadName = NULL,
      PThread::Priority priority = PThread::NormalPriority
    ) : PThreadPoolBase(maxWorkers, maxWorkUnits, threadName, priority)
    { }

    //
    // define the ancestor of the worker thread
    //
    class WorkerThread : public WorkerThreadBase
    {
      protected:
        WorkerThread(PThreadPool & pool, Priority priority = NormalPriority, const char * threadName = NULL)
          : WorkerThreadBase(priority, threadName)
          , m_pool(pool)
        {
        }

      public:
        virtual void AddWork(Work_T * work) = 0;
        virtual void RemoveWork(Work_T * work) = 0;
        virtual void Main() = 0;
  
      protected:
        PThreadPool & m_pool;
    };

    //
    // define internal worker wrapper class
    //
    class InternalWork : public InternalWorkBase
    {
      public:
        InternalWork(WorkerThread * worker, Work_T * work, const char * group)
          : InternalWorkBase(group)
          , m_worker(worker)
          , m_work(work)
        { 
        }

        WorkerThread * m_worker;
        Work_T * m_work;
    };

    //
    // define map for external work units to internal work
    //
    typedef std::map<Work_T *, InternalWork> ExternalToInternalWorkMap_T;
    ExternalToInternalWorkMap_T m_externalToInternalWorkMap;


    //
    // define class for storing group informationm
    //
    struct GroupInfo {
      unsigned m_count;
      WorkerThread * m_worker;
    };


    //
    //  define map for group ID to group information
    //
    typedef std::map<std::string, GroupInfo> GroupInfoMap_t;
    GroupInfoMap_t m_groupInfoMap;


    //
    //  add a new unit of work to a worker thread
    //
    bool AddWork(Work_T * work, const char * group = NULL)
    {
      PWaitAndSignal m(m_listMutex);

      // allocate by group if specified
      // else allocate to least busy
      WorkerThread * worker;
      if ((group == NULL) || (strlen(group) == 0)) {
        worker = (WorkerThread *)AllocateWorker();
      }
      else {

        // find the worker thread with the matching group ID
        // if no matching Id, then create a new thread
        typename GroupInfoMap_t::iterator g = m_groupInfoMap.find(group);
        if (g == m_groupInfoMap.end()) 
          worker = (WorkerThread *)AllocateWorker();
        else {
          worker = g->second.m_worker;
          PTRACE(4, "ThreadPool\tAllocated worker thread by group Id " << group);
        }
      }

      // if cannot allocate worker, return
      if (worker == NULL) 
        return false;

      // create internal work structure
      InternalWork internalWork(worker, work, group);

      // add work to external to internal map
      m_externalToInternalWorkMap.insert(typename ExternalToInternalWorkMap_T::value_type(work, internalWork));

      // add group ID to map
      if (!internalWork.m_group.empty()) {
        typename GroupInfoMap_t::iterator r = m_groupInfoMap.find(internalWork.m_group);
        if (r != m_groupInfoMap.end())
          ++r->second.m_count;
        else {
          GroupInfo info;
          info.m_count  = 1;
          info.m_worker = worker;
          m_groupInfoMap.insert(typename GroupInfoMap_t::value_type(internalWork.m_group, info));
        }
      }
      
      // give the work to the worker
      worker->AddWork(work);
    
      return true;
    }

    //
    //  remove a unit of work from a worker thread
    //
    bool RemoveWork(Work_T * work, bool removeFromWorker = true)
    {
      PWaitAndSignal m(m_listMutex);

      // find worker with work unit to remove
      typename ExternalToInternalWorkMap_T::iterator iterWork = m_externalToInternalWorkMap.find(work);
      if (iterWork == m_externalToInternalWorkMap.end())
        return false;

      InternalWork & internalWork = iterWork->second;

      // tell worker to stop processing work
      if (removeFromWorker)
        internalWork.m_worker->RemoveWork(work);

      // update group information
      if (!internalWork.m_group.empty()) {
        typename GroupInfoMap_t::iterator iterGroup = m_groupInfoMap.find(internalWork.m_group);
        PAssert(iterGroup != m_groupInfoMap.end(), "Attempt to find thread from unknown work group");
        if (iterGroup != m_groupInfoMap.end()) {
          if (--iterGroup->second.m_count == 0)
            m_groupInfoMap.erase(iterGroup);
        }
      }

      // see if workers need reorganising
      CheckWorker(internalWork.m_worker);

      // remove element from work unit map
      m_externalToInternalWorkMap.erase(iterWork);

      return true;
    }
};


template <class Work_T>
class PQueuedThreadPool : public PThreadPool<Work_T>
{
  public:
    //
    //  constructor
    //
    PQueuedThreadPool(
      unsigned maxWorkers = 10,
      unsigned maxWorkUnits = 0,
      const char * threadName = NULL,
      PThread::Priority priority = PThread::NormalPriority
    ) : PThreadPool<Work_T>(maxWorkers, maxWorkUnits, threadName, priority)
    { }

    class QueuedWorkerThread : public PThreadPool<Work_T>::WorkerThread
    {
      public:
        QueuedWorkerThread(PThreadPool<Work_T> & pool,
                           PThread::Priority priority = PThread::NormalPriority,
                           const char * threadName = NULL)
          : PThreadPool<Work_T>::WorkerThread(pool, priority, threadName)
          , m_available(0, INT_MAX)
        {
        }

        ~QueuedWorkerThread()
        { }

        void AddWork(Work_T * work)
        {
          m_mutex.Wait();
          m_queue.push(work);
          m_available.Signal();
          m_mutex.Signal();
        }

        void RemoveWork(Work_T * )
        {
          m_mutex.Wait();
          Work_T * work = m_queue.front();
          m_queue.pop();
          m_mutex.Signal();
          delete work;
        }

        unsigned GetWorkSize() const
        {
          return (unsigned)m_queue.size();
        }

        void Main()
        {
          for (;;) {
            m_available.Wait();
            if (PThreadPool<Work_T>::WorkerThread::m_shutdown)
              break;

            m_mutex.Wait();
            Work_T * work = m_queue.empty() ? NULL : m_queue.front();
            m_mutex.Signal();

            if (work != NULL) {
              work->Work();
              PThreadPool<Work_T>::WorkerThread::m_pool.RemoveWork(work);
            }
          }
        }

        void Shutdown()
        {
          PThreadPool<Work_T>::WorkerThread::m_shutdown = true;
          m_available.Signal();
        }

      protected:
        typedef std::queue<Work_T *> Queue;
        Queue      m_queue;
        PMutex     m_mutex;
        PSemaphore m_available;
    };


    virtual PThreadPoolBase::WorkerThreadBase * CreateWorkerThread()
    {
      return new QueuedWorkerThread(*this, this->m_priority, this->m_threadName);
    }
};


class PSafeWork : public PSafePtrBase {
  public:
    PSafeWork(
      PSafeObject * ptr
    ) : PSafePtrBase(ptr) { }

    virtual void Work()
    {
      PSafeObject * ptr = this->GetObject();
      if (ptr != NULL) {
        PTRACE_CONTEXT_ID_PUSH_THREAD(ptr);
        CallFunction(*ptr);
      }
    }

    virtual void CallFunction(PSafeObject & obj) = 0;
};


typedef PQueuedThreadPool<PSafeWork> PSafeThreadPool;


template <class PtrClass, typename FuncRet = void>
class PSafeWorkNoArg : public PSafeWork {
  public:
    typedef FuncRet (PtrClass::*Function)();

  protected:
    P_ALIGN_FIELD(Function,m_function,16);

  public:
    PSafeWorkNoArg(
      PtrClass * ptr,
      Function function
    ) : PSafeWork(ptr)
      , m_function(function)
    { }

    virtual void CallFunction(PSafeObject & obj)
    {
      (dynamic_cast<PtrClass&>(obj).*(this->m_function))();
    }
};


template <
  class PtrClass,
  typename Arg1Type,
  typename FuncRet = void
>
class PSafeWorkArg1 : public PSafeWork {
  public:
    typedef FuncRet (PtrClass::*Function)(Arg1Type arg1);

  protected:
    P_ALIGN_FIELD(Function,m_function,16);
    Arg1Type m_arg1;

  public:
    PSafeWorkArg1(
      PtrClass * ptr,
      Arg1Type arg1,
      Function function
    ) : PSafeWork(ptr)
      , m_function(function)
      , m_arg1(arg1)
    { }

    virtual void CallFunction(PSafeObject & obj)
    {
      (dynamic_cast<PtrClass&>(obj).*(this->m_function))(m_arg1);
    }
};


template <
  class PtrClass,
  typename Arg1Type,
  typename Arg2Type,
  typename FuncRet = void
>
class PSafeWorkArg2 : public PSafeWork {
  public:
    typedef FuncRet (PtrClass::*Function)(Arg1Type arg1, Arg2Type arg2);

  protected:
    P_ALIGN_FIELD(Function,m_function,16);
    Arg1Type m_arg1;
    Arg2Type m_arg2;

  public:
    PSafeWorkArg2(
      PtrClass * ptr,
      Arg1Type arg1,
      Arg2Type arg2,
      Function function
    ) : PSafeWork(ptr)
      , m_function(function)
      , m_arg1(arg1)
      , m_arg2(arg2)
    { }

    virtual void CallFunction(PSafeObject & obj)
    {
      (dynamic_cast<PtrClass&>(obj).*(this->m_function))(m_arg1, m_arg2);
    }
};


template <
  class PtrClass,
  typename Arg1Type,
  typename Arg2Type,
  typename Arg3Type,
  typename FuncRet = void
>
class PSafeWorkArg3 : public PSafeWork {
  public:
    typedef FuncRet (PtrClass::*Function)(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3);

  protected:
    P_ALIGN_FIELD(Function,m_function,16);
    Arg1Type m_arg1;
    Arg2Type m_arg2;
    Arg3Type m_arg3;

  public:
    PSafeWorkArg3(
      PtrClass * ptr,
      Arg1Type arg1,
      Arg2Type arg2,
      Arg2Type arg3,
      Function function
    ) : PSafeWork(ptr)
      , m_function(function)
      , m_arg1(arg1)
      , m_arg2(arg2)
      , m_arg3(arg3)
    { }

    virtual void CallFunction(PSafeObject & obj)
    {
      (dynamic_cast<PtrClass&>(obj).*(this->m_function))(m_arg1, m_arg2, m_arg3);
    }
};


#endif // PTLIB_THREADPOOL_H


// End Of File ///////////////////////////////////////////////////////////////