source: src/Fragmentation/Automation/SpecificFragmentController_ResultContainer_impl.hpp@ d67e6c

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

#ifndef SPECIFICFRAGMENTCONTROLLER_IMPL_HPP_
#define SPECIFICFRAGMENTCONTROLLER_IMPL_HPP_


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

#include "SpecificFragmentController.hpp"

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


template <typename T>
size_t SpecificFragmentController::ResultContainer<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;
  ConvertFragmentResultTo(fragmentresults, fragmentData);

  // insert into map
  std::vector<FragmentResult::ptr>::const_iterator resultiter = fragmentresults.begin();
  typename std::vector<T>::const_iterator dataiter = fragmentData.begin();
  for(;resultiter != fragmentresults.end(); ++resultiter, ++dataiter) {
    const JobId_t _id = (*resultiter)->getId();
#ifndef NDEBUG
    std::pair<typename std::map<JobId_t, T>::iterator, bool> inserter =
#endif
    IdData.insert( std::make_pair( _id, *dataiter) );
    ASSERT( inserter.second,
        "SpecificFragmentController::ResultContainer::receiveResults() - result id "
        +toString(inserter.first->first)+" already present.");
  }
  ASSERT( dataiter == fragmentData.end(),
      "SpecificFragmentController::ResultContainer::receiveResults() - fragmentresults and fragmentdata differ in size.");
  return fragmentData.size();
}

template <typename T>
void SpecificFragmentController::ResultContainer<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;
  size_t oldNoCalculatedResults = 0;
  while (NoReceivedResults != NoExpectedResults) {
    // 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");

    const std::pair<size_t, size_t> JobStatus = callback.getJobStatus();
    const size_t NoCalculatedResults = JobStatus.second;
    if (NoCalculatedResults != oldNoCalculatedResults) {
      oldNoCalculatedResults = NoCalculatedResults;
      LOG(1, "INFO: #" << JobStatus.first << " are waiting in the queue and #" << JobStatus.second << " jobs are calculated so far.");
    }

    // if some are done, get them
    if (NoCalculatedResults != 0) {
      NoReceivedResults += receiveResults(callback);
    }
  }
}

template <typename T>
void SpecificFragmentController::ResultContainer<T>::ConvertFragmentResultTo(
    const std::vector<FragmentResult::ptr> &results,
    std::vector<T> &fragmentData)
{
  // extract results
  fragmentData.clear();
  fragmentData.reserve(results.size());

  LOG(2, "DEBUG: Parsing now through " << results.size() << " results.");
  for (std::vector<FragmentResult::ptr>::const_iterator iter = results.begin();
      iter != results.end(); ++iter) {
    //LOG(1, "RESULT: job #"+toString((*iter)->getId())+": "+toString((*iter)->result));
    T extractedData;
    std::stringstream inputstream((*iter)->result);
    LOG(2, "DEBUG: First 50 characters FragmentResult's string: "+(*iter)->result.substr(0, 50));
    boost::archive::text_iarchive ia(inputstream);
    ia >> extractedData;
    LOG(1, "INFO: extracted data is " << extractedData << ".");
    fragmentData.push_back(extractedData);
  }

  ASSERT( results.size() == fragmentData.size(),
      "ConvertFragmentResultTo() - the number of extracted data differs from the number of results.");
}

#endif /* SPECIFICFRAGMENTCONTROLLER_IMPL_HPP_ */