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

Go to the documentation of this file.

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

#include <cassert>
#include <queue>
#include <string>
#include <vector>

#include <boost/noncopyable.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/bind.hpp>

#include "job_batch.hpp"
#include "worker_group.hpp"

namespace moost { namespace thread {

class threaded_job_batch : public job_batch
                         , public boost::noncopyable
                         , public boost::enable_shared_from_this<threaded_job_batch>
{
public:
   threaded_job_batch()
      : m_todo(0)
      , m_count(0)
   {
   }

   virtual void add(job_t job)
   {
      {
         boost::mutex::scoped_lock lock(m_mx);
         m_jobs.push(job);
         ++m_todo;
      }

      m_cond.notify_one();
   }

   void run(worker_group& wg)
   {
      while (!done())
      {
         do_one(wg);
      }
   }

   size_t count() const
   {
      boost::mutex::scoped_lock lock(m_mx);
      return m_count;
   }

   const std::vector<std::string>& errors() const
   {
      boost::mutex::scoped_lock lock(m_mx);

      if (m_todo > 0)
      {
         throw std::runtime_error("cannot call errors() when there are unfinished jobs");
      }

      return m_errors;
   }

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

   void do_one(worker_group& wg)
   {
      job_t job;

      {
         boost::mutex::scoped_lock lock(m_mx);
         assert(!m_jobs.empty());
         job = m_jobs.front();
         m_jobs.pop();
      }

      wg.add_job(boost::bind(&threaded_job_batch::run_one, shared_from_this(), job));
   }

   bool done() const
   {
      boost::mutex::scoped_lock lock(m_mx);

      while (m_todo > 0 && m_jobs.empty())
      {
         m_cond.wait(lock);
      }

      return m_todo == 0;
   }

   void run_one(job_t& job)
   {
      try
      {
         job();  // JFDI! :)
      }
      catch (const std::exception& e)
      {
         boost::mutex::scoped_lock lock(m_mx_errors);
         m_errors.push_back(e.what());
      }
      catch (...)
      {
         boost::mutex::scoped_lock lock(m_mx_errors);
         m_errors.push_back("unknown exception caught");
      }

      {
         boost::mutex::scoped_lock lock(m_mx);
         --m_todo;
         ++m_count;
      }

      m_cond.notify_one();
   }

   jobs_t m_jobs;
   mutable boost::condition_variable m_cond;
   mutable boost::mutex m_mx;
   volatile size_t m_todo;
   volatile size_t m_count;
   std::vector<std::string> m_errors;
   mutable boost::mutex m_mx_errors;
};

class threaded_job_scheduler : public boost::noncopyable
{
public:
   typedef threaded_job_batch job_batch_type;

   explicit threaded_job_scheduler(size_t num_workers = 1)
      : m_workers(num_workers)
   {
   }

   void dispatch(boost::shared_ptr<job_batch_type> batch)
   {
      batch->run(m_workers);
   }

private:
   worker_group m_workers;
};

}}

#endif