How to use the same thread pool batch by batch

Question

I found a good implementation of boost based thread pool which is an improvement over this and this . it is very easy to understand and test. It looks like this:

#include <boost/thread/thread.hpp>
#include <boost/asio.hpp>
// the actual thread pool
struct ThreadPool {
   ThreadPool(std::size_t);
   template<class F>
   void enqueue(F f);
   ~ThreadPool();    

   // the io_service we are wrapping
   boost::asio::io_service io_service;
   // dont let io_service stop
   boost::shared_ptr<boost::asio::io_service::work> work;
   //the threads
   boost::thread_group threads;
};

// the constructor just launches some amount of workers
ThreadPool::ThreadPool(size_t nThreads)
   :io_service()
   ,work(new boost::asio::io_service::work(io_service))
{
   for ( std::size_t i = 0; i < nThreads; ++i ) {
    threads.create_thread(boost::bind(&boost::asio::io_service::run, &io_service));
   }
}

// add new work item to the pool
template<class F>
void ThreadPool::enqueue(F f) {
   io_service.post(f);
}

// the destructor joins all threads
ThreadPool::~ThreadPool() {
work.reset();
io_service.run();
}

//tester: 
void f(int i)
{
    std::cout << "hello " << i << std::endl;
    boost::this_thread::sleep(boost::posix_time::milliseconds(300));
    std::cout << "world " << i << std::endl;
}

//it can be tested via:

int main() {
   // create a thread pool of 4 worker threads
   ThreadPool pool(4);

   // queue a bunch of "work items"
   for( int i = 0; i < 8; ++i ) {
      std::cout << "task " << i << " created" << std::endl;
      pool.enqueue(boost::bind(&f,i));
   }
}

g++ ThreadPool-4.cpp -lboost_system -lboost_thread

Now the question: I need to know how I can modify the implementation to be able to use this thread pool batch by batch- only when the first set of my work is fully completed by the thread pool, I need to supply the second set and so on. I tried to play with .run() and .reset() (found in the destructor) between the batch jobs but no luck:

//adding methods to the tread pool :
//reset the asio work and thread
void ThreadPool::reset(size_t nThreads){

work.reset(new boost::asio::io_service::work(io_service));
   for ( std::size_t i = 0; i < nThreads; ++i ) {
    threads.create_thread(boost::bind(&boost::asio::io_service::run, &io_service));
   }
    std::cout << "group size : " << threads.size() << std::endl;
}

//join, and even , interrupt
void ThreadPool::joinAll(){   
  threads.join_all();
  threads.interrupt_all();
}

//tester
int main() {
   // create a thread pool of 4 worker threads
   ThreadPool pool(4);

   // queue a bunch of "work items"
   for( int i = 0; i < 20; ++i ) {
      std::cout << "task " << i << " created" << std::endl;
          pool.enqueue(boost::bind(&f,i));
   }
   //here i play with the asio work , io_service and and the thread group
   pool.work.reset();
   pool.io_service.run();
   std::cout << "after run" << std::endl; 
   pool.joinAll();
   std::cout << "after join all" << std::endl; 
   pool.reset(4);
   std::cout << "new thread group size: " << pool.threads.size() << std::endl;///btw: new threa group size is 8. I expected 4! 
    // second batch... never completes
   for( int i = 20; i < 30; ++i ) {
          pool.enqueue(boost::bind(&f,i));
   }
}

The second batch doesn't complete. I will appreciate if you help me fix this. thank you

UPDATE- Solution:

based on a solution by Nik, I developed a solution using condition variable. Just add the following code to the original class:

// add new work item to the pool
template<class F>
void ThreadPool::enqueue(F f) {
    {
        boost::unique_lock<boost::mutex> lock(mutex_);
        nTasks ++;
    }
    //forwarding the job to wrapper()
    void (ThreadPool::*ff)(boost::tuple<F>) = &ThreadPool::wrapper<F>;
    io_service.post(boost::bind(ff, this, boost::make_tuple(f))); //using a tuple seems to be the only practical way. it is mentioned in boost examples.
}
//run+notfiy
template<class F>
void ThreadPool::wrapper(boost::tuple<F> f) {
    boost::get<0>(f)();//this is the task (function and its argument) that has to be executed by a thread
    {
        boost::unique_lock<boost::mutex> lock(mutex_);
        nTasks --;
        cond.notify_one();
    }
}

void ThreadPool::wait(){
    boost::unique_lock<boost::mutex> lock(mutex_);
    while(nTasks){
        cond.wait(lock);
    }
}

Now you may call wait() method between batches of work. one problem however: Even after the last batch, I have to call pool.wait() because the thread pool's scope will end after that and thread pool's destructor will be invoked. During destruction, some of the jobs are done and it will be the time to call the .notify(). As the Threadpool::mutex during destruction is invalidated, exceptions occur during locking. your suggestion will be appreciated.

Answer 1

A condition variable could be used to achieve desired result.

Implement a function responsible for calling enqueue the tasks and wait on a condition variable. Condition variable is notified when all tasks assigned to the pool are complete.

Every thread checks if the jobs are complete or not. Once all the jobs are complete condition variable is notified.

//An example of what you could try, this just an hint for what could be explored.

     void jobScheduler()
    {
      int jobs = numberOfJobs; //this could vary and can be made shared memory

       // queue a bunch of "work items"
       for( int i = 0; i < jobs; ++i ) 
       {
          std::cout << "task " << i << " created" << std::endl;
          pool.enqueue(boost::bind(&f,i));
       }
       //wait on a condition variable
      boost::mutex::scoped_lock lock(the_mutex);
      conditionVariable.wait(lock); //Have this varibale notified from any thread which realizes that all jobs are complete.
     }

---

Solution 2

I have a new working solution, with some assumption about syntax of functions being called back, but that could be changed as per requirement.

Continuing on the lines of above I use condition variable for managing my tasks but with a difference.

1. Create a queue of jobs.
2. A Manager which waits for new JOBS in the queue.
3. Once a job is received a notification is sent to waiting manager about the same.
4. Worker maintains a handle to Manager. When all the tasks assigned are complete Manger is informed.
5. Manager on getting a call for end, stops waiting for new JOBS in queue and exits.

---

#include <iostream>
#include <queue>
#include <boost/thread/thread.hpp>
#include <boost/asio.hpp>
#include <boost/tuple/tuple.hpp> 
#include <boost/tuple/tuple_io.hpp> 
#include <boost/function.hpp> 

///JOB Queue hold all jobs required to be executed
template<typename Job>
class JobQueue
{
  private:

    std::queue<Job> _queue;
    mutable boost::mutex _mutex;
    boost::condition_variable _conditionVariable;

  public:
    void push(Job const& job)
    {
      boost::mutex::scoped_lock lock(_mutex);
      _queue.push(job);
      lock.unlock();
      _conditionVariable.notify_one();
    }

    bool empty() const
    {
      boost::mutex::scoped_lock lock(_mutex);
      return _queue.empty();
    }

    bool tryPop(Job& poppedValue)
    {
      boost::mutex::scoped_lock lock(_mutex);
      if(_queue.empty())
      {
        return false;
      }

      poppedValue = _queue.front();
      _queue.pop();
      return true;
    }

    void waitAndPop(Job& poppedValue)
    {
      boost::mutex::scoped_lock lock(_mutex);
      while(_queue.empty())
      {
        _conditionVariable.wait(lock);
      }

      poppedValue = _queue.front();
      _queue.pop();
    }

};

///Thread pool for posting jobs to io service
class ThreadPool
{
  public :
    ThreadPool( int noOfThreads = 1) ;
    ~ThreadPool() ;

    template< class func >
      void post( func f ) ;

    boost::asio::io_service &getIoService() ;

  private :
    boost::asio::io_service _ioService;
    boost::asio::io_service::work _work ;
    boost::thread_group _threads;
};

  inline ThreadPool::ThreadPool( int noOfThreads )
: _work( _ioService )
{
  for(int i = 0; i < noOfThreads ; ++i) // 4
    _threads.create_thread(boost::bind(&boost::asio::io_service::run, &_ioService));
}

inline ThreadPool::~ThreadPool()
{
  _ioService.stop() ;
  _threads.join_all() ;
}

inline boost::asio::io_service &ThreadPool::getIoService()
{
  return _ioService ;
}

  template< class func >
void ThreadPool::post( func f )
{
  _ioService.post( f ) ;
}


template<typename T>
class Manager;

///Worker doing some work.
template<typename T>
class Worker{

    T _data;
    int _taskList;
    boost::mutex _mutex;
    Manager<T>* _hndl;

  public:

    Worker(T data, int task, Manager<T>* hndle):
    _data(data),
    _taskList(task),
    _hndl(hndle)
    {
    }

    bool job()
    {
      boost::mutex::scoped_lock lock(_mutex);
      std::cout<<"...Men at work..."<<++_data<<std::endl;
      --_taskList;
      if(taskDone())
       _hndl->end();
    } 

    bool taskDone()
    {
      std::cout<<"Tasks  "<<_taskList<<std::endl<<std::endl;
      if(_taskList == 0)
      {
        std::cout<<"Tasks done "<<std::endl;
        return true;
      }
      else false;
    }

};

///Job handler waits for new jobs and
///execute them as when a new job is received using Thread Pool.
//Once all jobs are done hndler exits.
template<typename T>
class Manager{

 public:

   typedef boost::function< bool (Worker<T>*)> Func;

   Manager(int threadCount):
   _threadCount(threadCount),
   _isWorkCompleted(false)
   {
     _pool = new ThreadPool(_threadCount);

     boost::thread jobRunner(&Manager::execute, this);
   }

   void add(Func f, Worker<T>* instance)
   {
     Job job(instance, f);
     _jobQueue.push(job);
   }

   void end()
   {
     boost::mutex::scoped_lock lock(_mutex);
     _isWorkCompleted = true;
     //send a dummy job
     add( NULL, NULL);
   }

   void workComplete()
   {
     std::cout<<"Job well done."<<std::endl;
   }

   bool isWorkDone()
   {
     boost::mutex::scoped_lock lock(_mutex);
     if(_isWorkCompleted)
       return true;
     return false;
   }

   void execute()
   {
      Job job;

     while(!isWorkDone())
     {
       _jobQueue.waitAndPop(job);

        Func f  = boost::get<1>(job);
        Worker<T>* ptr = boost::get<0>(job);

        if(f)
        {
          _pool->post(boost::bind(f, ptr));
        }
        else
          break;
     }

     std::cout<<"Complete"<<std::endl;
   }


 private:

  ThreadPool *_pool;
  int _threadCount;
  typedef boost::tuple<Worker<T>*, Func > Job;
  JobQueue<Job> _jobQueue;
  bool _isWorkCompleted;
  boost::mutex _mutex;
};

typedef boost::function< bool (Worker<int>*)> IntFunc;
typedef boost::function< bool (Worker<char>*)> CharFunc;


int main()
{
  boost::asio::io_service ioService;

  Manager<int> jobHndl(2);
  Worker<int> wrk1(0,4, &jobHndl);

  IntFunc f= &Worker<int>::job;

  jobHndl.add(f, &wrk1);
  jobHndl.add(f, &wrk1);
  jobHndl.add(f, &wrk1);
  jobHndl.add(f, &wrk1);

  Manager<char> jobHndl2(2);
  Worker<char> wrk2(0,'a', &jobHndl2);

  CharFunc f2= &Worker<char>::job;

  jobHndl2.add(f2, &wrk2);
  jobHndl2.add(f2, &wrk2);
  jobHndl2.add(f2, &wrk2);
  jobHndl2.add(f2, &wrk2);

  ioService.run();
  while(1){}
  return 0;
}

Answer 2

The third solution is the best (easiest IMHO), the one from the asio father;

You have to understand that you will stay blocked on "Threads.join_all()" statement while there is still a thread alive. Then you can call again with other work to do.

May be an alternative is to use taskqueue "A task queue that uses a thread pool to complete tasks in parallel", you fill up the queue with your works, it ensures that there will be no more than 'x' tasks working in parallel. Sample is easy to understand.

May be you need to add that member function to TaskQueue class in order to solve your "pool.wait()" issue:

void WaitForEmpty(){
    while( NumPendingTasks() || threads_.size() ){
      boost::wait_for_any(futures_.begin(), futures_.end());
    }
}

Enjoy !