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: 25362 $
* $Author: rjongbloed $
* $Date: 2011-03-20 18:27:31 -0500 (Sun, 20 Mar 2011) $
*/
#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(Priority priority = NormalPriority)
: PThread(100, NoAutoDeleteThread, priority, "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();
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 = 10, unsigned maxWorkUnitCount = 0);
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:
//
// constructor
//
PThreadPool(unsigned maxWorkers = 10, unsigned maxWorkUnits = 0)
: PThreadPoolBase(maxWorkers, maxWorkUnits)
{ }
//
// define the ancestor of the worker thread
//
class WorkerThread : public WorkerThreadBase
{
public:
WorkerThread(PThreadPool & pool, Priority priority = NormalPriority)
: WorkerThreadBase(priority)
, 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;
};
//
// 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)
: PThreadPool<Work_T>(maxWorkers, maxWorkUnits)
{ }
class QueuedWorkerThread : public PThreadPool<Work_T>::WorkerThread
{
public:
QueuedWorkerThread(PThreadPool<Work_T> & pool, PThread::Priority priority = PThread::NormalPriority)
: PThreadPool<Work_T>::WorkerThread(pool, priority)
, 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();
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);
}
};
#endif // PTLIB_THREADPOOL_H
// End Of File ///////////////////////////////////////////////////////////////
PThreadPoolBase::CheckWorker
virtual bool CheckWorker(WorkerThreadBase *worker)
PWaitAndSignal
This class waits for the semaphore on construction and automatically signals the semaphore on destruc...
Definition: psync.h:86
PThreadPoolBase::SetMaxUnits
void SetMaxUnits(unsigned count)
Definition: threadpool.h:193
PSemaphore
This class defines a thread synchronisation object.
Definition: semaphor.h:78
PThreadPool::InternalWork
Definition: threadpool.h:249
PThreadPoolBase::~PThreadPoolBase
~PThreadPoolBase()
PThreadPoolBase::GetMaxUnits
unsigned GetMaxUnits() const
Definition: threadpool.h:191
PQueuedThreadPool::QueuedWorkerThread::m_mutex
PMutex m_mutex
Definition: threadpool.h:451
PThreadPool::InternalWork::m_work
Work_T * m_work
Definition: threadpool.h:260
PThreadPool::PThreadPool
PThreadPool(unsigned maxWorkers=10, unsigned maxWorkUnits=0)
Definition: threadpool.h:222
PThreadPoolBase::WorkerList_t
std::vector< WorkerThreadBase * > WorkerList_t
Definition: threadpool.h:204
PThreadPoolBase::SetMaxWorkers
void SetMaxWorkers(unsigned count)
Definition: threadpool.h:187
PQueuedThreadPool::QueuedWorkerThread
Definition: threadpool.h:393
PThreadPoolBase::m_workers
WorkerList_t m_workers
Definition: threadpool.h:205
PThreadPoolBase::InternalWorkBase
Definition: threadpool.h:168
PThreadPool::WorkerThread::Main
virtual void Main()=0
PThreadPool::GroupInfo::m_count
unsigned m_count
Definition: threadpool.h:274
PThreadPoolBase::StopWorker
void StopWorker(WorkerThreadBase *worker)
PThreadPoolBase::PThreadPoolBase
PThreadPoolBase(unsigned maxWorkerCount=10, unsigned maxWorkUnitCount=0)
PThreadPoolBase::WorkerThreadBase::WorkerThreadBase
WorkerThreadBase(Priority priority=NormalPriority)
Definition: threadpool.h:156
PThreadPool::RemoveWork
bool RemoveWork(Work_T *work, bool removeFromWorker=true)
Definition: threadpool.h:344
PThreadPool::GroupInfo::m_worker
WorkerThread * m_worker
Definition: threadpool.h:275
PThreadPool::m_externalToInternalWorkMap
ExternalToInternalWorkMap_T m_externalToInternalWorkMap
Definition: threadpool.h:267
PThread::Priority
Priority
Codes for thread priorities.
Definition: thread.h:74
PTRACE
#define PTRACE(level, args)
Output trace.
Definition: object.h:530
PThreadPool::m_groupInfoMap
GroupInfoMap_t m_groupInfoMap
Definition: threadpool.h:283
PSemaphore::Signal
virtual void Signal()
If there are waiting (blocked) threads then unblock the first one that was blocked.
PThreadPool::ExternalToInternalWorkMap_T
std::map< Work_T *, InternalWork > ExternalToInternalWorkMap_T
Definition: threadpool.h:266
PSemaphore::Wait
virtual void Wait()
If the semaphore count is > 0, decrement the semaphore and return.
PThreadPoolBase::WorkerThreadBase::m_workerMutex
PMutex m_workerMutex
Definition: threadpool.h:165
PThreadPool::WorkerThread::RemoveWork
virtual void RemoveWork(Work_T *work)=0
PThreadPoolBase::m_maxWorkUnitCount
unsigned m_maxWorkUnitCount
Definition: threadpool.h:208
PQueuedThreadPool
High Level (queued work item) thread pool.
Definition: threadpool.h:383
PQueuedThreadPool::QueuedWorkerThread::GetWorkSize
unsigned GetWorkSize() const
Definition: threadpool.h:419
PThreadPoolBase
These classes and templates implement a generic thread pooling mechanism.
Definition: threadpool.h:150
PQueuedThreadPool::QueuedWorkerThread::AddWork
void AddWork(Work_T *work)
Definition: threadpool.h:402
PQueuedThreadPool::CreateWorkerThread
virtual PThreadPoolBase::WorkerThreadBase * CreateWorkerThread()
Definition: threadpool.h:456
PThreadPoolBase::WorkerThreadBase::m_shutdown
bool m_shutdown
Definition: threadpool.h:164
PThreadPool::InternalWork::m_worker
WorkerThread * m_worker
Definition: threadpool.h:259
PQueuedThreadPool::QueuedWorkerThread::Shutdown
void Shutdown()
Definition: threadpool.h:442
PThreadPoolBase::CreateWorkerThread
virtual WorkerThreadBase * CreateWorkerThread()=0
PQueuedThreadPool::PQueuedThreadPool
PQueuedThreadPool(unsigned maxWorkers=10, unsigned maxWorkUnits=0)
Definition: threadpool.h:389
PQueuedThreadPool::QueuedWorkerThread::m_queue
Queue m_queue
Definition: threadpool.h:450
PThread::NoAutoDeleteThread
Don't delete thread as it may not be on heap.
Definition: thread.h:94
PThreadPool::GroupInfoMap_t
std::map< std::string, GroupInfo > GroupInfoMap_t
Definition: threadpool.h:282
PThreadPoolBase::WorkerThreadBase
Definition: threadpool.h:153
PThreadPool::GroupInfo
Definition: threadpool.h:273
PThreadPool::WorkerThread::m_pool
PThreadPool & m_pool
Definition: threadpool.h:243
PThread
This class defines a thread of execution in the system.
Definition: thread.h:66
PQueuedThreadPool::QueuedWorkerThread::Main
void Main()
Definition: threadpool.h:424
PQueuedThreadPool::QueuedWorkerThread::QueuedWorkerThread
QueuedWorkerThread(PThreadPool< Work_T > &pool, PThread::Priority priority=PThread::NormalPriority)
Definition: threadpool.h:396
PThreadPoolBase::AllocateWorker
virtual WorkerThreadBase * AllocateWorker()
PThreadPoolBase::WorkerThreadBase::Shutdown
virtual void Shutdown()=0
PThreadPool::AddWork
bool AddWork(Work_T *work, const char *group=NULL)
Definition: threadpool.h:289
PThreadPoolBase::m_listMutex
PMutex m_listMutex
Definition: threadpool.h:202
PThreadPoolBase::m_maxWorkerCount
unsigned m_maxWorkerCount
Definition: threadpool.h:207
PThreadPoolBase::NewWorker
virtual WorkerThreadBase * NewWorker()
PThreadPool::WorkerThread::WorkerThread
WorkerThread(PThreadPool &pool, Priority priority=NormalPriority)
Definition: threadpool.h:232
PMutex
Synonym for PTimedMutex.
PAssert
#define PAssert(b, msg)
This macro is used to assert that a condition must be true.
Definition: object.h:192
PQueuedThreadPool::QueuedWorkerThread::RemoveWork
void RemoveWork(Work_T *)
Definition: threadpool.h:410
PThreadPoolBase::InternalWorkBase::InternalWorkBase
InternalWorkBase(const char *group)
Definition: threadpool.h:171
PThreadPoolBase::InternalWorkBase::m_group
std::string m_group
Definition: threadpool.h:176
PThreadPool::InternalWork::InternalWork
InternalWork(WorkerThread *worker, Work_T *work, const char *group)
Definition: threadpool.h:252
PThreadPool::WorkerThread::AddWork
virtual void AddWork(Work_T *work)=0
PThreadPool::WorkerThread
Definition: threadpool.h:229
PQueuedThreadPool::QueuedWorkerThread::m_available
PSemaphore m_available
Definition: threadpool.h:452
PObject
Ultimate parent class for all objects in the class library.
Definition: object.h:1118
PThreadPoolBase::GetMaxWorkers
unsigned GetMaxWorkers() const
Definition: threadpool.h:185
PThread::NormalPriority
Normal priority for a thread.
Definition: thread.h:79
PQueuedThreadPool::QueuedWorkerThread::Queue
std::queue< Work_T * > Queue
Definition: threadpool.h:449
PThreadPool
Low Level thread pool.
Definition: threadpool.h:215
PThreadPoolBase::WorkerThreadBase::GetWorkSize
virtual unsigned GetWorkSize() const =0