Scheduler.h

#ifndef theplu_yat_utility_scheduler
#define theplu_yat_utility_scheduler

// $Id: Scheduler.h 3694 2017-09-23 11:35:31Z peter $

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

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

#ifdef YAT_HAVE_ATOMIC
#include <atomic>
#endif
#include <set>
#include <deque>

namespace theplu {
namespace yat {
namespace utility {

  class Scheduler
  {
#ifdef YAT_HAVE_ATOMIC
    typedef std::atomic<int> running_jobs_type;
#else
    typedef int running_jobs_type;
#endif
  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;
      friend class JobHandler;
      mutable boost::mutex mutex_;
      void add_prerequisite(const boost::shared_ptr<Job>&);
      // \brief remove job from list of prerequisite
      // \return number of prerequisite (after removal)
      size_t remove_prerequisite(const boost::shared_ptr<Job>&);
      void add_observer(const boost::shared_ptr<Job>&);

      // 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_;
      // - pristine is what it says
      // - prepared - job has been either submitted directly a job that
      // depends on job has been submitted et.c.
      // - running - job has been sent to queue for workers to chew on
      // - completed - job has returned
      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);

    size_t jobs(void) const;

    void submit(const 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

    // \internal Class that handles job
    class JobHandler
    {
    public:
      JobHandler(unsigned int threads, JobQueue& queue, Queue<JobPtr>& jobs,
                 running_jobs_type& running_jobs,
                 Queue<boost::exception_ptr>& error);

      void operator()(void);
    private:
      void process(JobPtr& job);

      // If job is ready to be submitted i.e. all prerequisite have
      // completed, then submit job to queue for workers to chew on.
      void prepare(JobPtr job);
      // handle jobs returned from worker
      void post_process(JobPtr job);

      void send2queue(JobPtr& job);
      unsigned int threads_;
      JobQueue& queue_;
      Queue<JobPtr>& jobs_;
      running_jobs_type& running_jobs_;
      Queue<boost::exception_ptr>& error_;
      int job_counter_;
    };


    // 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);

    // If an error has been detected (and stored in error_), rethrow it.
    void throw_if_error(void) const;

    JobQueue queue_;
    Queue<JobPtr> jobs_;
    running_jobs_type running_jobs_;
    mutable Queue<boost::exception_ptr> error_;
    boost::thread job_handler_;
  }; // end class Scheduler

}}}
#endif