Revised on 4 Mar 2004

BoundedThreadPool.hpp

/*
 * Copyright (C) 2003-2004
 * Slawomir Lisznianski <slisznianski@asyncnet.com>
 *
 * Permission to use, copy, modify, distribute and sell this software
 * and its documentation for any purpose is hereby granted without fee,
 * provided that the above copyright notice appear in all copies and
 * that both that copyright notice and this permission notice appear
 * in supporting documentation. Slawomir Lisznianski makes no
 * representations about the suitability of this software for any
 * purpose. It is provided "as is" without express or implied warranty.
 *
 */

#ifndef RHA_BOUNDEDTHREADPOOL_HPP
#define RHA_BOUNDEDTHREADPOOL_HPP

#include <map> // For thread group.
#include <vector> // For expired threads collection.

// Threading and mutexing
// NOTE: compile with BOOST_HAS_THREADS
//
#include <boost/config.hpp>
#ifndef BOOST_HAS_THREADS
#   error   Thread support is unavailable!
#endif

// Threading and mutexing.
//
#include <boost/thread/condition.hpp>
#include <boost/thread/thread.hpp>

// Smart pointers
//
#include <boost/scoped_ptr.hpp>
#include <boost/shared_ptr.hpp>

namespace Rha
{
  template<class Q>
  class BoundedThreadPool
  {
  public:
    typedef BoundedThreadPool<Q> OfType;
    typedef typename Q::Task     Task;
    typedef Q                    Queue;

    class Configurator
    {
    public:
      // Default values.
      // VC++ 6 doesn't support in-place initialization of static const
      // per C2258 and C225.
      //
      enum {defaultMinThreads = 0,
            defaultMaxThreads = 5,
            defaultGracefulTermination = 1};

      Configurator()
      {
        init__();
      }

      unsigned int minThreads() const
      {
        return M_minThreads;
      }

      Configurator& minThreads(unsigned int minThreads)
      {
        M_minThreads = minThreads;
        validate__();
        return (*this);
      }

      unsigned int maxThreads() const
      {
        return M_maxThreads;
      }

      Configurator& maxThreads(unsigned int maxThreads)
      {
        M_maxThreads = maxThreads;
        validate__();
        return (*this);
      }

      bool gracefulTermination() const
      {
        return M_gracefulTermination;
      }

      Configurator& gracefulTermination(bool gracefulTermination)
      {
        M_gracefulTermination = gracefulTermination;
        validate__();
        return (*this);
      }

    private:
      void init__()
      {
        M_minThreads = defaultMinThreads;
        M_maxThreads = defaultMaxThreads;
        M_gracefulTermination = (defaultGracefulTermination != 0);
      }

      void validate__()
      {
        // Are parameters sane?
        //
        if (M_maxThreads < M_minThreads)
          throw std::invalid_argument("upper thread boundry less than low thread boundry");
        if (M_maxThreads < 1)
          throw std::invalid_argument("upper thread boundry less than one");
      }

      unsigned int M_minThreads;
      unsigned int M_maxThreads;
      bool         M_gracefulTermination;
    };

    BoundedThreadPool();

    BoundedThreadPool(const OfType& exec);

    BoundedThreadPool(const Configurator& config);

    ~BoundedThreadPool();

    void setUp(Queue* queuePtr);

    void operator()(const Task& task);

  private:
    // Helper struct used to encapsulate thread object and its
    // executor data. Its destructor is responsible for joining
    // with the thread it wraps prior to returning.
    //
    struct ThreadHandler :
      boost::noncopyable
    {
      ThreadHandler(BoundedThreadPool& executor, const unsigned int threadId, boost::thread* threadPtr) :
        M_threadImplPtr(threadPtr),
        M_executor(executor),
        M_threadId(threadId)
      {
        ++executor.M_createdThreads;
      }

      ~ThreadHandler()
      {
        M_threadImplPtr->join();
      }

      boost::scoped_ptr<boost::thread> M_threadImplPtr;
      BoundedThreadPool& M_executor;
      const unsigned int M_threadId;
    };
    friend struct ThreadHandler;
    // Helper struct used to encapsulate user's job. It has various
    // hooks to help in keeping track of job execution status.
    //
    class ExecutableTask
    {
    public:
      ExecutableTask(const unsigned int threadId, BoundedThreadPool& executor) :
        M_executor(executor),
        M_threadId(threadId)
      {}

      void operator ()()
      {
        try
        {
          // This aux flag helps in reducing number of mutexing
          // required when updating number of working threads.
          //
          for(bool firstIteration__(true);; firstIteration__ = false)
          {
            boost::scoped_ptr<Task> taskPtr__;
            {
              // Synchronize on the queue.
              //
              QueueScopedLock lock__(M_executor.MQueuePtrAccessToken);
              // If this is not our first iteration
              // we have to decrement working thread count.
              //
              if (!firstIteration__)
                --M_executor.M_workingThreads;

              // Wait until task queue is not empty or
              // we are exiting.
              //
              while (!M_executor.M_stopped && M_executor.MQueuePtr->empty())
              {
                // Stick around only if number of created
                // threads is below or equal to the minimum threads required.
                //
                if (M_executor.M_createdThreads > M_executor.M_configurator.minThreads())
                {
                  threadCleanUp__();
                  return;
                }
                M_executor.M_taskPushed.wait(lock__);
              }

              // There are two scenarios in which we will exit...
              // 1) there are no more jobs queued and we are told to stop running or
              // 2) there are outstanting jobs, but GracefulShutdown flag is
              //    false in which case outstanding jobs are ignored.
              //
              if (M_executor.M_stopped && (!M_executor.M_configurator.gracefulTermination() ||
                                              M_executor.MQueuePtr->empty()))
              {
                threadCleanUp__();
                return;
              }

              // Make a copy of the task. It may throw...
              //
              taskPtr__.reset(new Task(M_executor.MQueuePtr->front()));

              // Remove it from the job queue.
              //
              M_executor.MQueuePtr->pop();

              // Signal the task was popped.
              //
              M_executor.M_taskPoped.notify_one();

              // Increment the working threads count
              //
              ++M_executor.M_workingThreads;

              // Since there could be outstanding tasks in the queue
              // notify fellow threads if they are waiting.
              //
              if (!M_executor.MQueuePtr->empty())
              {
                // We may need more threads too.
                //
                M_executor.createThreadIfNeeded__();
                M_executor.M_taskPushed.notify_one();
              }
            }
            (*taskPtr__)();
          }
        }
        catch (...)
        {
          // Synchronize on the queue.
          //
          QueueScopedLock lock__(M_executor.MQueuePtrAccessToken);
          onException__();
        }
      }
    private:
      // Cleans up thread resources.
      //
      void threadCleanUp__()
      {
        // Pass the exiting thread to thread collector.
        //
        M_executor.M_expiredThreads.push_back(M_threadId);
        // Decrement number of created threads to prevent other threads from expiring
        // unnceseraly. New threads will not be created until existing 'dead' threads
        // are joined, so we should never leak threads.
        // This operation was intentionally not put in an auto-variable (RAII idiom)
        // as we want to reuse existing synchronized section.
        //
        --M_executor.M_createdThreads;
      }

      // Handles exception within thread.
      //
      void onException__()
      {
        // Since tasks are not allowed to leak exceptions,
        // we consider this condition unexpected.
        //
        std::unexpected();
      }

      BoundedThreadPool& M_executor;
      const unsigned int M_threadId;
    };
    friend class ExecutableTask;

    // Unconditionally creates a new thread.
    // This method must be invoked within a synchronized section.
    //
    void createThread__();

    // Conditionally creates a new thread.
    // This method must be invoked within a synchronized section.
    //
    void createThreadIfNeeded__();

    typedef boost::condition                                           Condition;
    typedef std::vector<unsigned int>                                  ThreadIdSeq;
    typedef std::map<unsigned int,  boost::shared_ptr<ThreadHandler> > ThreadGroup;
    typedef boost::mutex                                               QueueAccessToken;
    typedef boost::mutex::scoped_lock                                  QueueScopedLock;

    // Executor configurator
    Configurator M_configurator;
    // Thread accounting
    unsigned int     M_workingThreads;
    unsigned int     M_createdThreads;
    ThreadIdSeq      M_expiredThreads;
    ThreadGroup      M_threadGroup;
    unsigned int     M_threadId;
    // State signals, access control and state holders.
    QueueAccessToken MQueuePtrAccessToken;
    Condition        M_taskPushed;
    Condition        M_taskPoped;
    bool             M_stopped;
    Queue*           MQueuePtr;
  };
}

using namespace Rha;

template<class Q>
BoundedThreadPool<Q>::BoundedThreadPool()
{ }

template<class Q>
BoundedThreadPool<Q>::BoundedThreadPool(const Configurator& config) :
  M_configurator(config)
{ }

template<class Q>
BoundedThreadPool<Q>::BoundedThreadPool(const OfType& exec) :
  M_configurator(exec.M_configurator)
{ }

template<class Q>
void
BoundedThreadPool<Q>::setUp(Queue* queuePtr)
{
  // Reset stats.
  //
  M_workingThreads = 0;
  M_createdThreads = 0;
  M_threadId = 0;
  M_stopped = false;

  // Set pointer to queue.
  //
  MQueuePtr = queuePtr;

  // Optionally start worker threads.
  //
  for (unsigned int i = 0; i < M_configurator.minThreads(); ++i)
    createThread__();
}

template<class Q>
BoundedThreadPool<Q>::~BoundedThreadPool()
{
  {
    // Synchronize on the queue.
    //
    QueueScopedLock lock__(MQueuePtrAccessToken);

    // Set the flag to indicate we are exiting...
    //
    M_stopped = true;

    // Emit signal to force worker threads
    // check their 'stopped' flag.
    //
    M_taskPushed.notify_all();

    // Emit signal to force incoming call threads
    // check their 'stopped' flag.
    //
    M_taskPoped.notify_all();
  } // Store mutex released.
  //
  // Clear thread group. It will wait for
  // each worker thread to exit.
  //
  M_threadGroup.clear();
}

template<class Q>
void
BoundedThreadPool<Q>::operator()(const Task& task)
{
  // Synchronize queue access.
  //
  QueueScopedLock lock__(MQueuePtrAccessToken);

  // Check if there are any expired threads
  // that need to be cleaned up first.
  //
  if (!M_expiredThreads.empty())
  {
    //std::for_each(M_expiredThreads.begin(), M_expiredThreads.end(), boost::bind(&ThreadGroup::erase, M_threadGroup, _1));
    for (ThreadIdSeq::const_iterator threadIdIt__ = M_expiredThreads.begin();
         threadIdIt__ != M_expiredThreads.end();
         ++threadIdIt__)
    {
      M_threadGroup.erase(*threadIdIt__);
    }
  }
  // We may have to create a thread
  // if user specified 0 (zero) for minimum.
  //
  createThreadIfNeeded__();

  // If queue reports full, we have to wait.
  //
  while (MQueuePtr->full())
    M_taskPoped.wait(lock__);

  // Push job for pickup by worker thread.
  //
  MQueuePtr->push(task);

  // Send out the signal that the job is ready
  // for processing.
  //
  M_taskPushed.notify_one();
}

template<class Q>
void
BoundedThreadPool<Q>::createThread__()
{
  ++M_threadId;
  std::auto_ptr<boost::thread> threadPtr__(new boost::thread(ExecutableTask(M_threadId, *this)));
  boost::shared_ptr<ThreadHandler> tHandlerPtr__(new ThreadHandler(*this, M_threadId, threadPtr__.release()));
  M_threadGroup.insert(std::make_pair<unsigned int, boost::shared_ptr<ThreadHandler> >(M_threadId, tHandlerPtr__));
}

template<class Q>
void
BoundedThreadPool<Q>::createThreadIfNeeded__()
{
  // Check if there is enough worker threads
  // handling the load.
  // We will create a new thread if one of the
  // following conditions exists:
  // - There is currently more working threads than
  //   user specified miniumum threads and number of created
  //   threads is less than maximum allowed threads,
  // - Or, number of created threads is less than minimum
  //   threads. This is an abnormal condition unless exception
  //   is allowed to leak from a job and thread exits,
  // - Or, number of user specified minimum threads equals
  //   zero and number of created threads is zero
  //   (boostrap).
  //
  if ((M_workingThreads >= M_configurator.minThreads() && M_createdThreads < M_configurator.maxThreads()) ||
       M_createdThreads < M_configurator.minThreads() ||
      M_configurator.minThreads() == M_createdThreads && M_configurator.minThreads() == 0)
  {
    createThread__();
  }
}

#endif // RHA_BOUNDEDTHREADPOOL_HPP

---

© Copyright Slawomir Lisznianski 2003-2004. All Rights Reserved.

Permission to use, copy, modify, distribute and sell this software and its documentation for any purpose is hereby granted without fee, provided that the above copyright notice appear in all copies and that both that copyright notice and this permission notice appear in supporting documentation. Slawomir Lisznianski makes no representations about the suitability of this software for any purpose. It is provided "as is" without express or implied warranty.