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

// $Id: Scheduler.cc 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 <config.h>

#include "Scheduler.h"

#include <cassert>

namespace theplu {
namespace yat {
namespace utility {

  Scheduler::Scheduler(unsigned int threads)
  {
    assert(threads);
    for (size_t i=0; i<threads; ++i)
      workers_.create_thread(Worker(queue_, completed_));
  }


  void Scheduler::launch(void)
  {
    while (!jobs_.empty() || !running_jobs_.empty()) {
      while (!jobs_.empty()) {
        process(jobs_.front());
        jobs_.pop_front();
      }
      JobPtr job;
      completed_.pop(job);
      post_process(job);
    }

    // kill workers
    boost::shared_ptr<Job> end;
    queue_.push(end);

    // wait for workers to finish
    workers_.join_all();
  }


  void Scheduler::post_process(JobPtr job)
  {
    running_jobs_.erase(job);
    job->status_ = Job::completed;
    // Take care of sub-jobs
    for (size_t i=0; i<job->sub_jobs_.size(); ++i) {
      process(job->sub_jobs_[i]);
    }
    // check if there are jobs waiting for job
    std::map<JobPtr, std::vector<JobPtr> >::iterator it = children_.find(job);
    if (it == children_.end())
      return;
    // for convenience
    const std::vector<JobPtr>& vec = it->second;
    for (size_t i=0; i<vec.size(); ++i) {
      vec[i]->parents_.erase(job);
      if (vec[i]->parents_.empty())
        queue(vec[i]);
    }

    // entry no longer needed
    children_.erase(it);
  }


  void Scheduler::process(JobPtr job)
  {
    if (job->status_!=Job::pristine)
      return;
    job->status_ = Job::prepared;

    // If we have parents that need to be run first, well process them

    typedef std::set<JobPtr>::iterator iterator;
    for (iterator j = job->parents_.begin(); j!=job->parents_.end(); ) {
      if ((*j)->status_ == Job::completed)
        job->parents_.erase(j++);
      else {
        process(*j);
        // record that job is a child of j, so we can notify job when
        // j has completed
        children_[*j].push_back(job);
        ++j;
      }
    }

    // If job has no parents (or all parents are completed) send job toi queue
    if (job->parents_.empty()) {
      queue(job);
    }
  }


  void Scheduler::queue(JobPtr job)
  {
    job->status_ = Job::running;
    running_jobs_.insert(job);
    queue_.push(job);
  }


  void Scheduler::submit(JobPtr job)
  {
    jobs_.push_back(job);
  }


  // Scheduler::Job

  Scheduler::Job::Job(void)
    : status_(pristine) {}


  Scheduler::Job::~Job(void)
  {}


  void Scheduler::Job::add_dependency(boost::shared_ptr<Job> job)
  {
    parents_.insert(job);
  }


  void Scheduler::Job::submit(boost::shared_ptr<Job> job)
  {
    sub_jobs_.push_back(job);
  }


  // Scheduler::Worker

  Scheduler::Worker::Worker(Queue<boost::shared_ptr<Job> >& q,
                            Queue<boost::shared_ptr<Job> >& c)
    : queue_(q), completed_(c)
  {
  }


  void Scheduler::Worker::operator()(void)
  {
    while (true) {
      boost::shared_ptr<Job> job;
      // get next job
      queue_.pop(job);
      // NULL job indicates poison pill
      if (job.get()==NULL) {
        // kill other workers too
        queue_.push(job);
        break;
      }
      // action
      (*job)();
      // return job to scheduler
      completed_.push(job);

      // Make sure we can be interrupted
      boost::this_thread::interruption_point();
    }
  }

}}}