source: src/JobMarket/FragmentQueue.cpp@ 404d2b

/*
 * Project: JobMarket
 * Description: asynchronous Server/Controller/Client-approach to parallel computing, based on boost::asio
 * Copyright (C)  2011 Frederik Heber. All rights reserved.
 * 
 */

/*
 * FragmentQueue.cpp
 *
 *  Created on: Oct 19, 2011
 *      Author: heber
 */

// include config.h
#ifdef HAVE_CONFIG_H
#include <config.h>
#endif

#include "CodePatterns/MemDebug.hpp"

#include "JobMarket/FragmentQueue.hpp"

#include "CodePatterns/Assert.hpp"
#include "CodePatterns/Observer/Channels.hpp"
#include "CodePatterns/Log.hpp"

// functions wrapping some static instances
const FragmentResult::ptr FragmentQueue::getNoResult() const {
  static const FragmentResult::ptr NoResult( new FragmentResult(-1) );
  return NoResult;
}

const FragmentResult::ptr FragmentQueue::getNoResultQueued() const {
  static const FragmentResult::ptr NoResultQueued( new FragmentResult(-2) );
  return NoResultQueued;
}

const FragmentResult::ptr FragmentQueue::getResultDelivered() const {
  static const FragmentResult::ptr ResultDelivered( new FragmentResult(-3) );
  return ResultDelivered;
}

size_t FragmentQueue::Max_Attempts = (size_t)3;

/** Constructor for class FragmentQueue.
 *
 */
FragmentQueue::FragmentQueue() :
    Observable("FragmentQueue"),
    doneJobs(0)
{
  // observable stuff
  Channels *OurChannel = new Channels;
  Observable::insertNotificationChannel( std::make_pair(static_cast<Observable *>(this), OurChannel) );
  // add instance for each notification type
  for (size_t type = 0; type < NotificationType_MAX; ++type)
    OurChannel->addChannel(type);
}

/** Destructor for class FragmentQueue.
 *
 */
FragmentQueue::~FragmentQueue()
{
  while (!jobs.empty())
    jobs.pop();
  backup.clear();
  results.clear();
  attempts.clear();
}

/** Checks whether there are jobs in the queue at all.
 * \return true - jobs present, false - queue is empty
 */
bool FragmentQueue::isJobPresent() const
{
  return !jobs.empty();
}

/** Pushes a FragmentJob into the internal queue for delivery to server.
 *
 * \note we throw assertion when jobid has already been used.
 *
 * \param job job to enter into queue
 */
void FragmentQueue::pushJob(FragmentJob::ptr job)
{
  if ((job->getId() != JobId::IllegalJob) && (!results.count(job->getId()))) {
    OBSERVE;
    NOTIFY(JobAdded);
    results.insert( std::make_pair(job->getId(), getNoResult()));
    jobs.push(job);
  } else {
    ELOG(1, "job to push has IllegalJob id or id "+toString(job->getId())+" has already been used.");
    ASSERT(false,
        "job to push has IllegalJob id or id "+toString(job->getId())+" has already been used.");
  }
}

/** Pushes a bunch of FragmentJob's into the internal queue for delivery to server.
 *
 * \note we throw assertion when jobid has already been used.
 *
 * \sa pushJob()
 *
 * \param _jobs jobs to enter into queue
 * \return true - all job ids ok, false - at least one id is taken or illegal
 */
bool FragmentQueue::pushJobs(std::vector<FragmentJob::ptr> &_jobs)
{
  // check all presented jobs
  bool status = true;
  for (std::vector<FragmentJob::ptr>::const_iterator iter = _jobs.begin();
      iter != _jobs.end(); ++iter) {
    const FragmentJob::ptr &job = *iter;
    if ((job->getId() == JobId::IllegalJob) || (results.count(job->getId()))) {
      ELOG(1, "job to push has IllegalJob id or id "+toString(job->getId())+" has already been used.");
      ASSERT(false,
          "job to push has IllegalJob id or id "+toString(job->getId())+" has already been used.");
      status = false;
    }
  }
  // if ok, add jobs to queue
  if (status) {
    for (std::vector<FragmentJob::ptr>::iterator iter = _jobs.begin();
        iter != _jobs.end(); ++iter) {
      FragmentJob::ptr &job = *iter;
      results.insert( std::make_pair(job->getId(), getNoResult()));
      jobs.push(job);
    }
    // then give signal for each job (we don't know how many workers there are)
    for (std::vector<FragmentJob::ptr>::const_iterator iter = _jobs.begin();
        iter != _jobs.end(); ++iter) {
      OBSERVE;
      NOTIFY(JobAdded);
    }
  }
  return status;
}

/** Pops top-most FragmentJob from internal queue.
 *
 * From here on, we expect a result in FragmentQueue::results.
 *
 * \return job topmost job from queue
 */
FragmentJob::ptr FragmentQueue::popJob()
{
  ASSERT(jobs.size(),
      "FragmentQueue::popJob() - there are no jobs on the queue.");
  FragmentJob::ptr job = jobs.top();
#ifndef NDEBUG
  std::pair< BackupMap::iterator, bool> inserter =
#endif
  backup.insert( std::make_pair( job->getId(), job ));
  ASSERT (inserter.second,
      "FragmentQueue::popJob() - job "+toString(job->getId())+
      " is already in the backup.");
  ResultMap::iterator iter = results.find(job->getId());
  ASSERT(iter != results.end(),
      "FragmentQueue::popJob() - for job "+toString(job->getId())+" no result place has been stored.");
  iter->second = getNoResultQueued();
  jobs.pop();
  return job;
}

/** Internal function to check whether result is not one of static entities.
 *
 * @param result result to check against
 * @return true - result is a present, valid result, false - result is one of the statics
 */
bool FragmentQueue::isPresentResult(const FragmentResult::ptr result) const
{
  return (*result != *getNoResult())
      && (*result != *getNoResultQueued())
      && (*result != *getResultDelivered());
}

/** Queries whether a job has already been finished and the result is present.
 *
 * \param jobid id of job to query
 * \return true - result is present, false - result is not present
 */
bool FragmentQueue::isResultPresent(JobId_t jobid) const
{
  ResultMap::const_iterator iter = results.find(jobid);
  return ((iter != results.end())
      && isPresentResult(iter->second));
}

/** Counts the number of jobs for a given set of ids for which we have a calculated result present.
 *
 * \param ids vector of job ids to check
 * \return number of calculated results
 */
size_t FragmentQueue::getDoneJobs(const std::set<JobId_t> &ids) const
{
  size_t doneJobs = 0;
  for (std::set<JobId_t>::const_iterator iter = ids.begin();
      iter != ids.end(); ++iter) {
    ResultMap::const_iterator resiter = results.find(*iter);
    if (resiter != results.end())
      if (isPresentResult(resiter->second))
        ++doneJobs;
  }
  return doneJobs;
}

/** Counts the number of jobs for which still have to be calculated.
 *
 * \return number of jobs to be calculated
 */
size_t FragmentQueue::getPresentJobs() const
{
  const size_t presentJobs = jobs.size();
  return presentJobs;
}

/** Delivers result for a finished job.
 *
 * \note we throw assertion if not present
 *
 * \param jobid id of job
 * \return result for job of given \a jobid
 */
FragmentResult::ptr FragmentQueue::getResult(JobId_t jobid)
{
  ResultMap::iterator iter = results.find(jobid);
  ASSERT(iter != results.end(),
      "FragmentQueue::pushResult() - job "+toString(jobid)+" is not known to us.");
  ASSERT(*iter->second != *getNoResult(),
      "FragmentQueue::pushResult() - job "+toString(jobid)+" has not been request for calculation yet.");
  ASSERT(*iter->second != *getNoResultQueued(),
      "FragmentQueue::pushResult() - job "+toString(jobid)+"'s calculation is underway but not result has arrived yet.");
  ASSERT(*iter->second != *getResultDelivered(),
      "FragmentQueue::pushResult() - job "+toString(jobid)+"'s result has already been delivered.");
  /// store result
  FragmentResult::ptr _result = iter->second;
  /// mark as delivered in map
  if (iter->second != getResultDelivered()) {
    iter->second = getResultDelivered();
    --doneJobs;
  }
  /// and return result
  return _result;
}

std::vector<FragmentResult::ptr> FragmentQueue::getAllResults()
{
  std::vector<FragmentResult::ptr> returnresults;
  for (ResultMap::iterator iter = results.begin();
      iter != results.end(); ++iter) {
    if (isPresentResult(iter->second)) {
      returnresults.push_back(getResult(iter->first));
      iter = results.begin();
    }
  }

  return returnresults;
}

std::vector<FragmentResult::ptr> FragmentQueue::getAllResults(const std::set<JobId_t> &ids)
{
  std::vector<FragmentResult::ptr> returnresults;
  for (std::set<JobId_t>::const_iterator iter = ids.begin();
      iter != ids.end(); ++iter) {
    ResultMap::const_iterator resiter = results.find(*iter);
    if (resiter != results.end())
      if (isPresentResult(resiter->second))
        returnresults.push_back(getResult(*iter));
  }

  return returnresults;
}


/** Pushes a result for a finished job.
 *
 * \note we throw assertion if job already has result or is not known.
 *
 * \param result result of job to store
 */
void FragmentQueue::pushResult(FragmentResult::ptr &_result)
{
  const JobId_t id = _result->getId();
  /// check for presence
  ResultMap::iterator iter = results.find(id);
  ASSERT(iter != results.end(),
      "FragmentQueue::pushResult() - job "+toString(id)+" is not known to us.");
  ASSERT(*iter->second == *getNoResultQueued(),
      "FragmentQueue::pushResult() - is not waiting for the result of job "+toString(id)+".");
  // check whether this is a resubmitted job
  AttemptsMap::iterator attemptiter = attempts.find(id);
  // check whether succeeded or (finally) failed
  if ((_result->exitflag == 0) 
       || ((attemptiter != attempts.end()) && (attemptiter->second >= Max_Attempts))
       || (Max_Attempts == 1)) {
    // give notice if it is resubmitted job
    if (attemptiter != attempts.end()) {
      if (attemptiter->second >= Max_Attempts)
        ELOG(1, "Job #" << id << " failed on " << Max_Attempts << "th attempt for the last time.");
      else
        LOG(1, "INFO: Job #" << id << " succeeded on " << attemptiter->second << "th attempt.");
    }
    // remove in attempts
    if (attemptiter != attempts.end())
      attempts.erase(attemptiter);
    // remove in backup map
    BackupMap::iterator backupiter = backup.find(id);
    ASSERT( backupiter != backup.end(),
        "FragmentQueue::pushResult() - cannot find job "+toString(id)
        +" in backup.");
    backup.erase(backupiter);
    /// and overwrite NoResult in found entry
    iter->second = _result;
    // increase done jobs
    ++doneJobs;
  } else {
    LOG(1, "Job " << id << " failed, resubmitting.");
    // increase attempts
    if (attemptiter != attempts.end())
      ++(attemptiter->second);
    else
      attempts.insert( std::make_pair(id, (size_t)1) );
    // resubmit job
    resubmitJob(id);
  }
}

/** Resubmit a job which a worker failed to calculate.
 *
 * @param jobid id of the failed job
 */
void FragmentQueue::resubmitJob(const JobId_t jobid)
{
  BackupMap::iterator iter = backup.find(jobid);
  ASSERT( iter != backup.end(),
      "FragmentQueue::resubmitJob() - job id "+toString(jobid)
      +" not stored in backup.");
  if (iter != backup.end()) {
    // remove present result
    ResultMap::iterator resiter = results.find(jobid);
    ASSERT( resiter != results.end(),
        "FragmentQueue::resubmitJob() - job "+toString(jobid)
        +" to resubmit has no result present.");
    results.erase(resiter);
    const FragmentJob::ptr job = iter->second;
    backup.erase(iter);
    pushJob(job);
  }
}

/** Removes all done results currently in the queue.
 *
 */
void FragmentQueue::removeWaitingResults()
{
  for(ResultMap::iterator iter = results.begin();
      iter != results.end();) {
    if (isPresentResult(iter->second)) {
      results.erase(iter++);
      --doneJobs;
    } else
      iter++;
  }
}

/** Removes all waiting jobs currently in the queue.
 *
 */
void FragmentQueue::removeWaitingJobs()
{
  // remove placeholder results
  for(ResultMap::iterator iter = results.begin();
      iter != results.end();) {
    if (*(iter->second) == *getNoResult()) {
      results.erase(iter++);
    } else
      iter++;
  }
  // remove all jobs in the queue
  while (!jobs.empty())
    jobs.pop();
}