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

Go to the documentation of this file.

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

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

#include "../utils/foreach.hpp"

#include "worker_group.hpp"

namespace moost { namespace thread {

class async_batch_processor
{
private:
   class batch_state
   {
   public:
      batch_state(size_t todo)
         : m_todo(todo)
      {
      }

      void one_done()
      {
         {
            boost::lock_guard<boost::mutex> lock(m_mx);
            --m_todo;
         }

         m_cond.notify_one();
      }

      void wait_for_all_done()
      {
         boost::unique_lock<boost::mutex> lock(m_mx);

         while(m_todo > 0) // when m_todo == 0 all jobs are done
         {
            // wait for workers to signal when they're done
            m_cond.wait(lock);
         }
      }

   private:
      volatile size_t m_todo;
      boost::mutex m_mx;
      boost::condition_variable m_cond;
   };

public:
   typedef boost::function0<void> job_t;
   typedef std::vector<job_t> jobs_t;

   async_batch_processor(size_t num_threads)
      : m_wg(num_threads)
   {
   }

   void dispatch(jobs_t & jobs)
   {
      if (!jobs.empty())
      {
         boost::shared_ptr<batch_state> state(new batch_state(jobs.size()));

         // dispatch jobs
         foreach (job_t& job, jobs)
         {
            m_wg.add_job(boost::bind(&async_batch_processor::handler, boost::ref(job), state));
         }

         // and wait until all are done
         state->wait_for_all_done();
      }
   }

private:
   static void handler(job_t& job, boost::shared_ptr<batch_state> state)
   {
      // do job
      job();

      // and tell the state object we're done
      state->one_done();
   }

   worker_group m_wg;
};

}}

#endif // MOOST_THREAD_ASYNC_BATCH_PROCESSOR