Scheduler.h

#ifndef theplu_yat_utility_scheduler
#define theplu_yat_utility_scheduler

// $Id: Scheduler.h 3855 2020-01-02 01:11:34Z peter $

/*
  Copyright (C) 2014, 2015, 2016, 2017, 2019, 2020 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 <list>
#include <set>
#include <vector>

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;
      friend class JobHandler;
      void add_prerequisite(const boost::shared_ptr<Job>&,
                            const boost::unique_lock<boost::mutex>& lock);
      // \brief remove job from list of prerequisite
      // \return number of prerequisite (after removal)
      size_t remove_prerequisite(const boost::shared_ptr<Job>&,
                                 const boost::unique_lock<boost::mutex>& lock);
      // set of jobs that have to finish before this can run
      std::set<boost::shared_ptr<Job> > prerequisite_;
      // avoid accessing prerequisite_ directly but rather via
      // functions below passing Dependency::Lock
      std::set<boost::shared_ptr<Job> >&
      prerequisite(const boost::unique_lock<boost::mutex>& lock);
      const std::set<boost::shared_ptr<Job> >&
      prerequisite(const boost::unique_lock<boost::mutex>& lock) const;
      // - 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_type { pristine, prepared, running, completed};
      status_type status_;
      status_type status(const boost::unique_lock<boost::mutex>& lock) const;
      void status(const boost::unique_lock<boost::mutex>& lock, status_type s);
      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);

    unsigned int threads(void) const;

    void threads(unsigned int n);

    void wait(void);

  private:
    // forward declaration
    class JobHandler;

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

    class Messenger
    {
    public:
      virtual void operator()(JobHandler& handler)=0;
    };

    typedef boost::shared_ptr<Messenger> MessengerPtr;
    typedef Queue<MessengerPtr> MessengerQueue;

    class JobMessenger : public Messenger
    {
    public:
      JobMessenger(const JobPtr&);
      void operator()(JobHandler& handler);
    private:
      JobPtr job_;
    };


    class SchedulerIsWaiting : public Messenger
    {
    public:
      void operator()(JobHandler& handler);
    };


    class InterruptWorkers : public Messenger
    {
    public:
      void operator()(JobHandler& handler);
    };


    // Class used to change number of workers
    class WorkForceSizer : public Messenger
    {
    public:
      void operator()(JobHandler& handler);
    };


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


    class Dependency
    {
    public:
      typedef boost::unique_lock<boost::mutex> Lock;

      void add(const JobPtr& child, const JobPtr& parent, const Lock& lock);
      void remove(const JobPtr& parent, const Lock& lock);
      std::vector<JobPtr>& children(const JobPtr& key, const Lock& lock);
      boost::mutex& mutex(void);
    private:
      boost::mutex mutex_;

      // job in key has to finish before jobs in vector
      std::map<JobPtr, std::vector<JobPtr> > children_;

      std::vector<JobPtr> empty_vector_;
    };


    class JobHandlerData
    {
    public:
      class Count
      {
      public:
        explicit Count(int x=0);
        void decrement(void);
        int get(void) const;
        void increment(void);
        void set(int x);
      private:
        mutable boost::mutex mutex_;
        int x_;
      };

      JobHandlerData(void);
      Queue<boost::exception_ptr>& error(void) const;

      Dependency& dependency(void);
      const Dependency& dependency(void) const;

      const MessengerQueue& messengers(void) const;
      MessengerQueue& messengers(void); //
      const JobQueue& queue(void) const; //
      JobQueue& queue(void);

      const Count& running_jobs(void) const;
      Count& running_jobs(void);

      const Count& n_threads(void) const;
      Count& n_threads(void);
    private:
      mutable Queue<boost::exception_ptr> error_;
      MessengerQueue messengers_;
      // This is the queue workers are consuming from.
      JobQueue queue_;

      boost::mutex mutex_;

      Count running_jobs_;
      Count threads_;

      Dependency dependency_;
    };


    // \internal Class that handles job
    class JobHandler
    {
    public:
      JobHandler(JobHandlerData& data);

      void operator()(void);
      void create_workers(unsigned int n);
      JobHandlerData& data(void);
      void interrupt_workers(void);
      void kill_workers(void);
      void kill_workers(unsigned int n);
      // 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(JobPtr& job);
      void remove_joined_workers(void);
      void scheduler_is_waiting(bool value);
      unsigned int n_target_workers(void) const;
      void wait_workers(void) const;
    private:

      // 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, const Dependency::Lock& lock);
      // handle jobs returned from 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(JobPtr job, const Dependency::Lock& lock);

      void send2queue(JobPtr& job,
                      const boost::unique_lock<boost::mutex>& lock);

      JobHandlerData* data_;
      bool scheduler_is_waiting_;
      unsigned int job_count_;
      void join_workers(void);

      typedef std::list<boost::shared_ptr<boost::thread> > WorkerList;
      // FIXME is this needed here or can it be in
      // JobHandler::operator() scope?  The reason we keep it here is
      // so we can access it from other functions and don't need to
      // pass it around.
      WorkerList workers_;
      // Number of Workers minus number of poison pills sent for them
      unsigned int n_target_workers_;
    }; // end class JobHandler

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

    mutable boost::mutex mutex_;
    JobHandlerData data_;
    boost::thread job_handler_;
  }; // end class Scheduler

}}}
#endif