ThreadPool Class Reference

#include <threadpool.hh>

Collaboration diagram for ThreadPool:

Classes
struct	Job
class	PrivateData
class	Queue
class	WeakQueue

Public Types
using	Ptr = boost::shared_ptr< ThreadPool >
using	ThreadPtr = boost::shared_ptr< boost::thread >

Public Member Functions
	ThreadPool (bool completeAllJobsBeforeDestruction=false)
	ThreadPool. More...
	ThreadPool (int count, bool completeAllJobsBeforeDestruction=false)
	ThreadPool (const ThreadPool &)=delete
	ThreadPool (ThreadPool &&)=delete
ThreadPool	operator= (const ThreadPool &)=delete
ThreadPool	operator= (ThreadPool &&)=delete
	~ThreadPool ()
void	schedule (boost::function< void()> const &fn, int priority=defaultPriority, const Queue::Ptr &queue=defaultQueue())
void	schedule (boost::function< void()> const &fn, const Queue::Ptr &queue)
template<typename T >
void	schedule (boost::shared_ptr< T > fn, int priority=defaultPriority, const Queue::Ptr &queue=defaultQueue())
template<typename T >
void	schedule (boost::shared_ptr< T > fn, const Queue::Ptr &queue)
void	schedule (boost::function< void()> const &fn, int priority, const WeakQueue::Ptr &queue)
void	schedule (boost::function< void()> const &fn, const WeakQueue::Ptr &queue)
template<typename T >
void	schedule (boost::shared_ptr< T > fn, int priority, WeakQueue::Ptr queue)
template<typename T >
void	schedule (boost::shared_ptr< T > fn, WeakQueue::Ptr queue)
template<typename R >
boost::unique_future< R >	schedule (boost::function< R()> const &fn, int priority=defaultPriority, const Queue::Ptr &queue=defaultQueue())
template<typename R >
boost::unique_future< R >	schedule (boost::function< R()> const &fn, const Queue::Ptr &queue)
template<typename R , typename T >
boost::unique_future< R >	schedule (boost::shared_ptr< T > fn, int priority=defaultPriority, const Queue::Ptr &queue=defaultQueue())
template<typename R , typename T >
boost::unique_future< R >	schedule (boost::shared_ptr< T > fn, const Queue::Ptr &queue)
template<typename R >
boost::unique_future< R >	schedule (boost::function< R()> const &fn, int priority, WeakQueue::Ptr queue)
template<typename R >
boost::unique_future< R >	schedule (boost::function< R()> const &fn, WeakQueue::Ptr queue)
template<typename R , typename T >
boost::unique_future< R >	schedule (boost::shared_ptr< T > fn, int priority, WeakQueue::Ptr queue)
template<typename R , typename T >
boost::unique_future< R >	schedule (boost::shared_ptr< T > fn, WeakQueue::Ptr queue)
ThreadPtr	add ()
std::vector< ThreadPtr >	add (int count)

Static Public Member Functions
static ThreadPool::Ptr	create (bool completeAllJobsBeforeDestruction=false)
static ThreadPool::Ptr	create (int count, bool completeAllJobsBeforeDestruction=false)

Static Private Member Functions
static void	work (const PrivateData::Ptr &priv)
static void	do_one (const PrivateData::Ptr &priv)
static Queue::Ptr	defaultQueue ()

Private Attributes
std::list< ThreadPtr >	threads
PrivateData::Ptr	priv

Static Private Attributes
static const int	defaultPriority = PRIO_NORMAL

Detailed Description

Generic threadpool

A ThreadPool is basically a collection of threads you can schedule() work on.

Member Typedef Documentation

Ptr

using ThreadPool::Ptr = boost::shared_ptr<ThreadPool>

ThreadPtr

using ThreadPool::ThreadPtr = boost::shared_ptr<boost::thread>

Constructor & Destructor Documentation

ThreadPool() [1/4]

ThreadPool::ThreadPool ( bool  completeAllJobsBeforeDestruction = false )

explicit

ThreadPool.

Create a ThreadPool with one thread for each core in the system

  : priv(PrivateData::create(completeAllJobsBeforeDestruction))
{
  const int count = boost::thread::hardware_concurrency();
#ifndef MULTITHREADING
  if(count > 1)
    count = 1;
#endif
  add(count);
}

Referenced by CpuBound(), create(), and Sequentially().

Here is the call graph for this function:

Here is the caller graph for this function:

ThreadPool() [2/4]

ThreadPool::ThreadPool ( int  count, bool  completeAllJobsBeforeDestruction = false  )

explicit

Create a ThreadPool with the given number of threads

  : priv(PrivateData::create(completeAllJobsBeforeDestruction))
{
#ifndef MULTITHREADING
  if(count > 1)
    count = 1;
#endif
  add(count);
}

Here is the call graph for this function:

ThreadPool() [3/4]

ThreadPool::ThreadPool ( const ThreadPool &  )

delete

ThreadPool() [4/4]

ThreadPool::ThreadPool ( ThreadPool &&  )

delete

~ThreadPool()

ThreadPool::~ThreadPool

(

)

Destructor

• stop all threads
• throw away all jobs that haven't been executed

{
  // Destroying the threadpool used to be done by interrupting all
  // threads, but this doesn't work reliably, at least until boost
  // 1.45. Hence, we're back to using an "alive" boolean and a regular
  // condition variable.
  //
  // See also https://svn.boost.org/trac/boost/ticket/2330
 
  {
    boost::mutex::scoped_lock const lock(priv->mut);
    priv->alive = false;
    priv->cond.notify_all();
  }
}

Member Function Documentation

add() [1/2]

ThreadPool::ThreadPtr ThreadPool::add

(

)

Add an additional thread to the pool.

This is mostly used for testing

Returns

a reference to the newly added thread.

{
  ThreadPool::ThreadPtr t = ThreadPool::ThreadPtr(new boost::thread(boost::bind(&ThreadPool::work, priv)));
  threads.push_back(t);
  ThreadList::instance()->add(t);
  return t;
}

Referenced by add(), BOOST_AUTO_TEST_CASE(), and ThreadPool().

Here is the call graph for this function:

Here is the caller graph for this function:

add() [2/2]

std::vector< ThreadPool::ThreadPtr > ThreadPool::add

(

int

count

)

Add the given number of threads to the pool.

This is mostly used for testing

Returns

references to the newly added threads.

{
  std::vector<ThreadPool::ThreadPtr> result(count);
  for(int i = 0; i < count; i++)
  {
    result[i] = add();
  }
 
  return result;
}

Here is the call graph for this function:

create() [1/2]

ThreadPool::Ptr ThreadPool::create ( bool  completeAllJobsBeforeDestruction = false )

static

Create a ThreadPool with one thread for each core in the system

{
  return ThreadPool::Ptr(new ThreadPool(completeAllJobsBeforeDestruction));
}

Referenced by BOOST_AUTO_TEST_CASE().

Here is the call graph for this function:

Here is the caller graph for this function:

create() [2/2]

ThreadPool::Ptr ThreadPool::create ( int  count, bool  completeAllJobsBeforeDestruction = false  )

static

Create a ThreadPool with the given number of threads

{
  return ThreadPool::Ptr(new ThreadPool(count, completeAllJobsBeforeDestruction));
}

Here is the call graph for this function:

defaultQueue()

ThreadPool::Queue::Ptr ThreadPool::defaultQueue ( )

staticprivate

{
  static ThreadPool::Queue::Ptr const queue = ThreadPool::Queue::create();
  return queue;
}

Here is the call graph for this function:

do_one()

void ThreadPool::do_one ( const PrivateData::Ptr &  priv )

staticprivate

Execute one job.

This gets called from work(). It fetches and executes the highest-prio job from ThreadPool::jobs

{
  ThreadPool::Job job;
 
  {
    boost::mutex::scoped_lock const lock(priv->mut);
 
    while(!priv->jobs.empty() && priv->jobs.begin()->second.empty())
    {
      priv->jobs.erase(priv->jobs.begin());
    }
 
    if(!priv->jobs.empty() && !priv->jobs.begin()->second.empty())
    {
      job = priv->jobs.begin()->second.front();
      priv->jobs.begin()->second.pop();
    }
    else
    {
      defect_message("JobQueue empty while it shouldn't be");
    }
  }
 
  if(job.queue)
  {
    QueueLock const l(job.queue);
    if(l.queueExists())
    {
      boost::this_thread::disable_interruption const while_executing_jobs;
      job.fn();
    }
  }
}

Referenced by work().

Here is the call graph for this function:

Here is the caller graph for this function:

operator=() [1/2]

ThreadPool ThreadPool::operator= ( const ThreadPool &  )

delete

operator=() [2/2]

ThreadPool ThreadPool::operator= ( ThreadPool &&  )

delete

schedule() [1/16]

template<typename R >

boost::unique_future< R > ThreadPool::schedule	(	boost::function< R()> const &	fn,
		const Queue::Ptr &	queue
	)

{
  return schedule(fn, defaultPriority, queue);
}

Here is the call graph for this function:

schedule() [2/16]

template<typename R >

boost::unique_future< R > ThreadPool::schedule	(	boost::function< R()> const &	fn,
		int	priority,
		WeakQueue::Ptr	queue
	)

{
  // Todo: If boost::function supported move semantics, we could do without
  // the shared pointer.
 
  // Todo: Without the static cast, Boost 1.53 packaged_task stores a
  // reference to fn, which is a temporary, and hence results in
  // undefined behaviour. Move semantics seem to work OK there...
  //
  // See https://svn.boost.org/trac/boost/ticket/8596
  boost::shared_ptr<boost::packaged_task<R>> const t(new boost::packaged_task<R>(static_cast<boost::function<R()>>(fn)));
  boost::unique_future<R>                          f = t->get_future();
  schedule(boost::bind(Detail::threadPoolExecute<void, boost::packaged_task<R>>, t), priority, queue);
  return f;
}

Here is the call graph for this function:

schedule() [3/16]

template<typename R >

boost::unique_future< R > ThreadPool::schedule	(	boost::function< R()> const &	fn,
		int	priority = defaultPriority,
		const Queue::Ptr &	queue = defaultQueue()
	)

{
  return schedule(fn, priority, queue->getWeak());
}

Here is the call graph for this function:

schedule() [4/16]

template<typename R >

boost::unique_future< R > ThreadPool::schedule	(	boost::function< R()> const &	fn,
		WeakQueue::Ptr	queue
	)

{
  return schedule(fn, defaultPriority, queue);
}

Here is the call graph for this function:

schedule() [5/16]

void ThreadPool::schedule	(	boost::function< void()> const &	fn,
		const Queue::Ptr &	queue
	)

Schedule the given job at the given queue

{
  schedule(fn, defaultPriority, std::move(queue));
}

Here is the call graph for this function:

schedule() [6/16]

void ThreadPool::schedule	(	boost::function< void()> const &	fn,
		const WeakQueue::Ptr &	queue
	)

Schedule the given job at the given queue

{
  schedule(fn, defaultPriority, std::move(queue));
}

Here is the call graph for this function:

schedule() [7/16]

void ThreadPool::schedule	(	boost::function< void()> const &	fn,
		int	priority,
		const WeakQueue::Ptr &	queue
	)

Schedule the given job at the given priority

{
  boost::mutex::scoped_lock const lock(priv->mut);
  priv->jobs[priority].emplace(fn, queue);
  priv->jobcount++;
  priv->cond.notify_one();
}

schedule() [8/16]

void ThreadPool::schedule	(	boost::function< void()> const &	fn,
		int	priority = defaultPriority,
		const Queue::Ptr &	queue = defaultQueue()
	)

Schedule the given job at the given priority

{
  schedule(fn, priority, queue->getWeak());
}

Referenced by BOOST_AUTO_TEST_CASE(), has_at_least_n_threads(), and schedule().

Here is the caller graph for this function:

schedule() [9/16]

template<typename T >

void ThreadPool::schedule	(	boost::shared_ptr< T >	fn,
		const Queue::Ptr &	queue
	)

Schedule the given job at the given priority

Precondition

T::operator()() must be defined

{
  schedule(fn, defaultPriority, queue);
}

Here is the call graph for this function:

schedule() [10/16]

template<typename R , typename T >

boost::unique_future< R > ThreadPool::schedule	(	boost::shared_ptr< T >	fn,
		const Queue::Ptr &	queue
	)

{
  return schedule<R, T>(fn, defaultPriority, queue);
}

schedule() [11/16]

template<typename T >

void ThreadPool::schedule	(	boost::shared_ptr< T >	fn,
		int	priority,
		WeakQueue::Ptr	queue
	)

Schedule the given job at the given priority

Precondition

T::operator()() must be defined

{
  schedule(boost::bind(Detail::threadPoolExecute<void, T>, fn), priority, queue);
}

Here is the call graph for this function:

schedule() [12/16]

template<typename R , typename T >

boost::unique_future< R > ThreadPool::schedule	(	boost::shared_ptr< T >	fn,
		int	priority,
		WeakQueue::Ptr	queue
	)

{
  // Todo: If boost::function supported move semantics, we could do without
  // the shared pointer.
  boost::shared_ptr<boost::packaged_task<R>> const t(
    new boost::packaged_task<R>(boost::bind(Detail::threadPoolExecute<R, T>, fn)));
  boost::unique_future<R> f = t->get_future();
  schedule(boost::bind(Detail::threadPoolExecute<void, boost::packaged_task<R>>, t), priority, queue);
  return f;
}

Here is the call graph for this function:

schedule() [13/16]

template<typename T >

void ThreadPool::schedule	(	boost::shared_ptr< T >	fn,
		int	priority = defaultPriority,
		const Queue::Ptr &	queue = defaultQueue()
	)

Schedule the given job at the given priority

Precondition

T::operator()() must be defined

{
  schedule(fn, priority, queue->getWeak());
}

Here is the call graph for this function:

schedule() [14/16]

template<typename R , typename T >

boost::unique_future< R > ThreadPool::schedule	(	boost::shared_ptr< T >	fn,
		int	priority = defaultPriority,
		const Queue::Ptr &	queue = defaultQueue()
	)

{
  return schedule<R, T>(fn, priority, queue->getWeak());
}

schedule() [15/16]

template<typename T >

void ThreadPool::schedule	(	boost::shared_ptr< T >	fn,
		WeakQueue::Ptr	queue
	)

Schedule the given job at the given priority

Precondition

T::operator()() must be defined

{
  schedule(fn, defaultPriority, queue);
}

Here is the call graph for this function:

schedule() [16/16]

template<typename R , typename T >

boost::unique_future< R > ThreadPool::schedule	(	boost::shared_ptr< T >	fn,
		WeakQueue::Ptr	queue
	)

{
  return schedule(fn, defaultPriority, queue);
}

Here is the call graph for this function:

work()

void ThreadPool::work ( const PrivateData::Ptr &  priv )

staticprivate

This is executed by each of the threads in the ThreadPool

• Wait for a job to be scheduled
• Fetch the highest-prio job from ThreadPool::jobs
• Execute

Those last two tasks will be performed by do_one()

{
  boost::mutex::scoped_lock lock(priv->mut);
  while(priv->alive)
  {
    if(priv->jobcount > 0)
    {
      priv->jobcount--;
      lock.unlock();
      do_one(priv);
      lock.lock();
    }
    else
    {
      priv->cond.wait(lock);
    }
  }
 
  bool busy = priv->completeAllJobsBeforeDestruction;
  while(busy)
  {
    if(priv->jobcount > 0)
    {
      priv->jobcount--;
      lock.unlock();
      do_one(priv);
      lock.lock();
    }
    else
    {
      busy = false;
    }
  }
}

Referenced by add().

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

defaultPriority

const int ThreadPool::defaultPriority = PRIO_NORMAL

staticprivate

Referenced by schedule().

priv

PrivateData::Ptr ThreadPool::priv

private

Referenced by add(), do_one(), schedule(), work(), and ~ThreadPool().

threads

std::list<ThreadPtr> ThreadPool::threads

private

Threads in this ThreadPool

Referenced by add().

---

The documentation for this class was generated from the following files:

• inc/scroom/threadpool.hh
• inc/scroom/impl/threadpoolimpl.hh
• libs/threadpool/src/threadpoolimpl.cc