source: trunk/yat/utility/Scheduler.h @ 3346

#ifndef theplu_yat_utility_scheduler
#define theplu_yat_utility_scheduler

// $Id: Scheduler.h 3346 2014-11-06 13:16:34Z peter $

/*
  Copyright (C) 2014 Peter Johansson

  This file is part of the yat library, http://dev.thep.lu.se/yat

  The yat library is free software; you can redistribute it and/or
  modify it under the terms of the GNU General Public License as
  published by the Free Software Foundation; either version 3 of the
  License, or (at your option) any later version.

  The yat library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  General Public License for more details.

  You should have received a copy of the GNU General Public License
  along with yat. If not, see <http://www.gnu.org/licenses/>.
*/

#include "Queue.h"

#include <boost/thread.hpp>
#include <boost/shared_ptr.hpp>

#include <set>
#include <deque>

namespace theplu {
namespace yat {
namespace utility {

  /**
     \brief Handle a number of jobs and send them to threads

     Scheduler starts a (user defined) number of threads and handles a
     series of Jobs. The Jobs can have dependencies, such that Job X
     must finish before Job Y can run, and the Scheduler takes care of
     these dependencies. It possible to create new Jobs within a Job
     (see Job::submit) and these will be processed when the current
     Job has completed. Here is a small code example in which two
     threads are used to process the four jobs, \c MyJob. The jobs
     have a dependency that job4 can not run until the three first
     jobs have completed.

     \code

     Scheduler scheduler(2);
     boost::shared_ptr<MyJob> job1(new MyJob("Hello"));
     boost::shared_ptr<MyJob> job1(new MyJob(" "));
     boost::shared_ptr<MyJob> job3(new MyJob("World"));
     boost::shared_ptr<MyJob> job4(new MyJob("\n"));
     job4.add_dependency(job1);
     job4.add_dependency(job2);
     job4.add_dependency(job3);
     scheduler.submit(job4);
     scheduler.launch();

     \endcode
   */
  class Scheduler
  {
  public:
    /**
       Base class that defines the interface for a Job
     */
    class Job
    {
    public:
      /**
         \brief constructor
       */
      Job(void);

      /**
         \brief destructor
       */
      virtual ~Job(void);

      /**
         \brief add a dependency

         This job will not be processed until \a job has finished.
       */
      void add_dependency(boost::shared_ptr<Job> job);

      /**
         This function defines the work done in thread.
       */
      virtual void operator()(void)=0;
    protected:
      /**
         \brief Add a Job

         Typically called within operator(). When this Job has
         completed, the Scheduler will collect jobs passed to submit
         and send them to the queue just like if they had been passed
         to Scheduler::submit.
       */
      void submit(boost::shared_ptr<Job> job);
    private:
      friend class Scheduler;
      // set of jobs that have to finish before this can run
      std::set<boost::shared_ptr<Job> > parents_;
      // jobs created within this job
      std::vector<boost::shared_ptr<Job> > sub_jobs_;
      enum status { pristine, prepared, running, completed};
      status status_;
    }; // end class Job

    /**
       \brief constructor

       \param threads number of threads that are used
     */
    Scheduler(unsigned int threads);

    /**
       Launch submitted jobs to threads and wait until all jobs have finished.
     */
    void launch(void);

    /**
       Add \a job to list of jobs that should be processed in launch(void)
     */
    void submit(boost::shared_ptr<Job> job);

  private:
    // \internal class that does the job
    //
    // It processes any job that is pushed to the \a queue until a
    // NULL Job shows up which signals that the work is done. When
    // NULL is observed the NULL Job is pushed to the queue so
    // co-workers are notified too.
    class Worker
    {
    public:
      Worker(Queue<boost::shared_ptr<Job> >& queue,
             Queue<boost::shared_ptr<Job> >& completed);
      void operator()(void);
    private:
      Queue<boost::shared_ptr<Job> >& queue_;
      Queue<boost::shared_ptr<Job> >& completed_;
    }; // end class Worker

    // function called when job has finished and returned from
    // worker. If there are any jobs that depends on \a job those jobs
    // are notified and if it makes them ready to be processed they
    // are sent to queue.
    void post_process(boost::shared_ptr<Job> job);

    // If \a job has parent jobs, which need to finish first, update
    // map children_ to reflect that. If all parents have finished,
    // send job to queue.
    void process(boost::shared_ptr<Job> job);

    // send job to queue
    void queue(boost::shared_ptr<Job> job);

    typedef boost::shared_ptr<Scheduler::Job> JobPtr;
    std::deque<JobPtr> jobs_;
    std::set<JobPtr> running_jobs_;

    // value lists all jobs that depend on key, i.e., key has to
    // finish before jobs in value can start
    std::map<JobPtr, std::vector<JobPtr> > children_;

    Queue<boost::shared_ptr<Job> > queue_;
    Queue<boost::shared_ptr<Job> > completed_;
    boost::thread_group workers_;

  }; // end class Scheduler

}}}
#endif