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: 24047 $
 * $Author: rjongbloed $
 * $Date: 2010-02-10 17:15:13 -0600 (Wed, 10 Feb 2010) $
 */


#ifndef PTLIB_THREADPOOL_H
#define PTLIB_THREADPOOL_H

#ifdef P_USE_PRAGMA
#pragma interface
#endif

#include <map>
#include <queue>


class PThreadPoolBase : public PObject
{
  public:
    class WorkerThreadBase : public PThread
    {
      public:
        WorkerThreadBase()
          : PThread(100, NoAutoDeleteThread, NormalPriority, "Pool")
          , m_shutdown(false)
        { }

        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(unsigned maxWorkerCount = 10, unsigned maxWorkUnitCount = 0);
    ~PThreadPoolBase();

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

  protected:
    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;
};


template <class Work_T>
class PThreadPool : public PThreadPoolBase
{
  PCLASSINFO(PThreadPool, PThreadPoolBase);
  public:
    //
    // define the ancestor of the worker thread
    //
    class WorkerThread : public WorkerThreadBase
    {
      public:
        WorkerThread(PThreadPool & pool_)
          : m_pool(pool_)
        {
        }

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

    class QueuedWorkerThread : public WorkerThread
    {
      public:
        QueuedWorkerThread(PThreadPool & pool)
          : WorkerThread(pool)
          , m_available(0, INT_MAX)
        {
        }

        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();
          delete m_queue.front();
          m_queue.pop();
          m_mutex.Signal();
        }

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

        void Main()
        {
          for (;;) {
            m_available.Wait();
            if (WorkerThread::m_shutdown)
              break;

            m_mutex.Wait();

            if (!m_queue.empty()) {
              Work_T * work = m_queue.front();
              if (work != NULL) {
                work->Work();
                WorkerThread::m_pool.RemoveWork(work);
              }
            }

            m_mutex.Signal();
          }
        }

        void Shutdown()
        {
          WorkerThread::m_shutdown = true;
          m_available.Signal();
        }

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

    //
    // 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;


    //
    //  constructor
    //
    PThreadPool(unsigned maxWorkers = 10, unsigned maxWorkUnits = 0)
      : PThreadPoolBase(maxWorkers, maxWorkUnits) 
    { }


    //
    //  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;
    }
};


#endif // PTLIB_THREADPOOL_H


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

---