Scheduler.h

#ifndef theplu_yat_utility_scheduler
#define theplu_yat_utility_scheduler

// $Id: Scheduler.h 3550 2017-01-03 05:41:02Z peter $

/*
  Copyright (C) 2014, 2015, 2016 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 "PriorityQueue.h"
#include "Queue.h"

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

#include <set>
#include <deque>

namespace theplu {
namespace yat {
namespace utility {

  class Scheduler
  {
  public:
    class Job
    {
    public:
      explicit Job(unsigned int prio=0);

      virtual ~Job(void);

      unsigned int priority(void) const;

      virtual void operator()(void)=0;
    private:
      friend class Scheduler;
      // set of jobs that have to finish before this can run
      std::set<boost::shared_ptr<Job> > prerequisite_;
      // jobs that have *this as prerequisite
      std::vector<boost::shared_ptr<Job> > observers_;
      enum status { pristine, prepared, running, completed};
      status status_;
      unsigned priority_;
      unsigned id_;
      boost::exception_ptr error_;
    }; // end class Job

    Scheduler(unsigned int threads);

    void add_dependency(boost::shared_ptr<Job> job,
                        boost::shared_ptr<Job> prerequisite);
    void interrupt(void);

    void submit(boost::shared_ptr<Job> job);

    void wait(void);

  private:
    typedef boost::shared_ptr<Scheduler::Job> JobPtr;

    struct LowerPriority
    {
      bool operator()(const JobPtr& lhs, const JobPtr& rhs) const;
    };

    // some typedefs for convenience
    typedef PriorityQueue<JobPtr, LowerPriority> JobQueue;

    // \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(JobQueue& queue, Queue<JobPtr>& completed);
      void operator()(void);
    private:
      JobQueue& queue_;
      Queue<JobPtr>& completed_;
    }; // end class Worker

    // function called when job has finished and returned from
    // worker. If there are any jobs that depend 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);

    int running_jobs_;

    JobQueue queue_;
    Queue<JobPtr> completed_;
    boost::thread_group workers_;
    unsigned long int job_counter_;
  }; // end class Scheduler

}}}
#endif