source: src/Fragmentation/Automation/SpecificFragmentController_ReceiveResultContainer_impl.hpp@ ab2e834

/*
 * SpecificFragmentController_ReceiveResultContainer_impl.hpp
 *
 *  Created on: Aug 27, 2012
 *      Author: heber
 */

#ifndef SPECIFICFRAGMENTCONTROLLER_RECEIVERESULTCONTAINER_IMPL_HPP_
#define SPECIFICFRAGMENTCONTROLLER_RECEIVERESULTCONTAINER_IMPL_HPP_


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

#include "SpecificFragmentController.hpp"

#include "CodePatterns/Assert.hpp"
#include "CodePatterns/toString.hpp"

#include <numeric>

template <typename T>
size_t SpecificFragmentController::ReceiveResultContainer<T>::receiveResults(
    SpecificFragmentController &callback)
{
  // receive (and remove the respective ids)
  callback.receiveResults(callback.host, callback.port);
  callback.RunService("Requesting results");
  std::vector<FragmentResult::ptr> fragmentresults = callback.getReceivedResults();

  // convert
  std::vector<T> fragmentData;
  SpecificFragmentController::ReceiveResultContainer<T>::ConvertFragmentResultTo(
    fragmentresults, 
    fragmentData);

  // insert into map
  SpecificFragmentController::ReceiveResultContainer<T>::insertResults(
    fragmentresults, 
    fragmentData);

  return fragmentData.size();
}

template <typename T>
void SpecificFragmentController::ReceiveResultContainer<T>::waitforResults(
    const size_t NoExpectedResults,
    boost::asio::io_service &io_service,
    SpecificFragmentController &callback)
{
  // wait but receive all results that are already done
  size_t NoReceivedResults = 0;
  while ((NoReceivedResults != NoExpectedResults)
      && (callback.getExitflag() == 0)) {
    // wait a bit
    boost::asio::deadline_timer timer(io_service);
    timer.expires_from_now(boost::posix_time::milliseconds(500));
    timer.wait();
    // then request status
    callback.checkResults(callback.host, callback.port);
    callback.RunService("Checking on results");
    if (callback.getExitflag() != 0)
      break;

    const std::pair<size_t, size_t> JobStatus = callback.getJobStatus();
    const size_t NoCalculatedResults = JobStatus.second;
    // if some are done, get them
    if (NoCalculatedResults != 0) {
      NoReceivedResults += receiveResults(callback);
      callback.handler(NoReceivedResults, NoExpectedResults);
      LOG(1, "INFO: #" << JobStatus.first << " are waiting in the queue and #" << NoReceivedResults << " of " << NoExpectedResults << " jobs are calculated so far.");
    }

    // if not all jobs are done yet, check number of present workers
    if (NoReceivedResults != NoExpectedResults) {
      callback.checkEnrolledWorkers(callback.host, callback.port);
      callback.RunService("Checking on number of workers");
      if (callback.getExitflag() != 0)
        break;
      const std::vector<size_t>& NumberWorkers = callback.getNumberOfWorkers();
      const int TotalNumberWorkers = std::accumulate(
          NumberWorkers.begin(), NumberWorkers.end(), 0);
      if (TotalNumberWorkers == 0) {
        ELOG(0, "Not all jobs are finished, but no workers are present working on it?!");
        callback.setExitflag(ExitflagContainer::ErrorFlag);
        break;
      } else {
        LOG(4, "There are " << NumberWorkers << " workers enrolled.");
      }
    }
}
}

#endif /* SPECIFICFRAGMENTCONTROLLER_RECEIVERESULTCONTAINER_IMPL_HPP_ */