/home/mhx/git/github/libmoost/include/moost/thread/worker_group.hpp

Go to the documentation of this file.

/* vim:set ts=3 sw=3 sts=3 et: */
#ifndef MOOST_THREAD_WORKER_GROUP_HPP
#define MOOST_THREAD_WORKER_GROUP_HPP

#include <queue>
#include <csignal>

#include <boost/function.hpp>
#include <boost/thread.hpp>
#include <boost/bind.hpp>

namespace moost { namespace thread {

class worker_group : public boost::noncopyable
{
public:
   typedef boost::function0<void> job_t;

   explicit worker_group(size_t num_workers = 1)
      : m_running(1)
   {
      if (num_workers < 1)
      {
         throw std::runtime_error("invalid number of worker threads");
      }

      for (size_t i = 0; i < num_workers; ++i)
      {
         m_workers.create_thread(boost::bind(&worker_group::work, this));
      }
   }

   ~worker_group()
   {
      try
      {
         stop();
      }
      catch (...)
      {
      }
   }

   void stop()
   {
      if (m_running)
      {
         {
            boost::mutex::scoped_lock lock(m_mx);
            m_running = 0;
         }
         m_cond.notify_all();
         m_workers.join_all();
      }
   }

   bool running() const
   {
      return m_running;
   }

   bool add_job(job_t job)
   {
      bool added = m_running != 0;

      if (added)
      {
         {
            boost::mutex::scoped_lock lock(m_mx);
            m_jobs.push(job);
         }

         m_cond.notify_one();
      }

      return added;
   }

   size_t size() const
   {
      return m_workers.size();
   }

   size_t queued_jobs() const
   {
      boost::mutex::scoped_lock lock(m_mx);
      return m_jobs.size();
   }

private:
   typedef std::queue<job_t> jobs_t;

   void work()
   {
      for (;;)
      {
         job_t job;

         {
            boost::mutex::scoped_lock lock(m_mx);

            while (m_jobs.empty() && m_running)
            {
               m_cond.wait(lock);
            }

            if (m_jobs.empty())
            {
               if (m_running)
                  continue;
               else
                  break;
            }

            job = m_jobs.front();
            m_jobs.pop();
         }

         job();
      }
   }

   boost::thread_group m_workers;
   jobs_t m_jobs;
   boost::condition_variable m_cond;
   mutable boost::mutex m_mx;
   volatile sig_atomic_t m_running;
};

}}

#endif