source: src/Fragmentation/Automation/FragmentController.cpp@ 685e28

/*
 * Project: MoleCuilder
 * Description: creates and alters molecular systems
 * Copyright (C)  2011 University of Bonn. All rights reserved.
 * Please see the LICENSE file or "Copyright notice" in builder.cpp for details.
 */

/*
 * FragmentController.cpp
 *
 *  Created on: Nov 27, 2011
 *      Author: heber
 */

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

// boost asio needs specific operator new
#include <boost/asio.hpp>

#include "CodePatterns/MemDebug.hpp"

#include "FragmentController.hpp"

#include "Connection.hpp" // Must come before boost/serialization headers.
#include "CodePatterns/Info.hpp"
#include "CodePatterns/Log.hpp"

#include "Operations/Controllers/CheckResultsOperation.hpp"
#include "Operations/Controllers/GetNextJobIdOperation.hpp"
#include "Operations/Controllers/ReceiveResultsOperation.hpp"
#include "Operations/Controllers/RemoveAllWorkerOperation.hpp"
#include "Operations/Controllers/SendJobsOperation.hpp"
#include "Operations/Controllers/ShutdownOperation.hpp"

#include "Jobs/MPQCCommandJob_MPQCData.hpp"
#include "JobId.hpp"

/** Constructor of class FragmentController.
 *
 * \param io_service io_service for the asynchronous operations
 * \param _host hostname of server that accepts jobs
 * \param _service of server
 */
FragmentController::FragmentController(
    boost::asio::io_service& io_service) :
    connection_(io_service),
    failed(boost::bind(&ExitflagContainer::setExitflag, this, ExitflagContainer::ErrorFlag))
{
  Info info(__FUNCTION__);

  // insert commands into registry
  Commands.registerInstance(new CheckResultsOperation(connection_, AsyncOperation::NoOpCallback, failed));
  Commands.registerInstance(new GetNextJobIdOperation(connection_, AsyncOperation::NoOpCallback, failed));
  Commands.registerInstance(new ReceiveResultsOperation(connection_, AsyncOperation::NoOpCallback, failed));
  Commands.registerInstance(new RemoveAllWorkerOperation(connection_));
  Commands.registerInstance(new SendJobsOperation(connection_, AsyncOperation::NoOpCallback, failed));
  Commands.registerInstance(new ShutdownOperation(connection_));
}

/** Destructor of class FragmentController.
 *
 */
FragmentController::~FragmentController()
{
  Commands.cleanup();
}

/** Requests an available id from server
 *
 * @param host address of server
 * @param service port/service of server
 * @param NumberIds number of desired ids to request with this connect
 */
void FragmentController::requestIds(
                const std::string &host,
                const std::string &service,
                const size_t NumberIds)
{
  GetNextJobIdOperation *getnextid = static_cast<GetNextJobIdOperation *>(
                Commands.getByName("getnextjobid"));
  (*getnextid).setDesiredIds(NumberIds);
  (*getnextid)(host,service);
}

/** Returns another available id from a finished GetNextJobIdOperation.
 *
 * @return next available id
 */
JobId_t FragmentController::getAvailableId()
{
  GetNextJobIdOperation *getnextid = static_cast<GetNextJobIdOperation *>(
                Commands.getByName("getnextjobid"));
  if( !getnextid->isNextIdAvailable()) {
    ELOG(1, "FragmentController::getAvailableId() - there are no more requested ids.");
    return JobId::IllegalJob;
  }
        const JobId_t nextid = getnextid->getNextId();
        LOG(1, "INFO: Next available id is " << nextid << ".");
        return nextid;
}

/** Adds a vector of jobs to the send operation.
 *
 * @param jobs jobs to add
 */
void FragmentController::addJobs(std::vector<FragmentJob::ptr> &jobs)
{
  SendJobsOperation *sendjobs = static_cast<SendJobsOperation *>(
                Commands.getByName("sendjobs"));
  sendjobs->addJobs(jobs);
  const size_t presentJobs = sendjobs->getPresentJobs();
        LOG(1, "INFO: #" << presentJobs << " jobs are now waiting to be transmitted.");
}

/** Sends contained jobs in operation to server
 *
 * @param host address of server
 * @param service port/service of server
 */
void FragmentController::sendJobs(
                const std::string &host,
                const std::string &service)
{
  SendJobsOperation *sendjobs = static_cast<SendJobsOperation *>(
                Commands.getByName("sendjobs"));
  const size_t presentJobs = sendjobs->getPresentJobs();
        LOG(1, "INFO: #" << presentJobs << " jobs are being sent to the server.");
  (*sendjobs)(host, service);
}

/** Obtains scheduled and done jobs from server
 *
 * @param host address of server
 * @param service port/service of server
 */
void FragmentController::checkResults(
                const std::string &host,
                const std::string &service)
{
  CheckResultsOperation *checkres = static_cast<CheckResultsOperation *>(
                Commands.getByName("checkresults"));
  (*checkres)(host, service);
}

/** Return scheduled and done jobs.
 *
 * @return pair of number of still scheduled and already done jobs
 */
std::pair<size_t, size_t> FragmentController::getJobStatus() const
{
  const CheckResultsOperation * const checkres = static_cast<const CheckResultsOperation *>(
                Commands.getByName("checkresults"));
  const size_t doneJobs = checkres->getDoneJobs();
  const size_t presentJobs = checkres->getPresentJobs();
  return make_pair(presentJobs, doneJobs);
}

/** Requests removal of all idle workers from server.
 *
 * @param host address of server
 * @param service port/service of server
 */
void FragmentController::removeall(
    const std::string &host,
    const std::string &service)
{
  RemoveAllWorkerOperation *removeall = static_cast<RemoveAllWorkerOperation *>(
      Commands.getByName("removeallworker"));
  (*removeall)(host, service);
}

/** Obtains results from done jobs from server.
 *
 * @param host address of server
 * @param service port/service of server
 */
void FragmentController::receiveResults(
                const std::string &host,
                const std::string &service)
{
  ReceiveResultsOperation *receiveres = static_cast<ReceiveResultsOperation *>(
                Commands.getByName("receiveresults"));
  (*receiveres)(host, service);
}

/** Getter for received results.
 *
 * @return vector with all received results
 */
std::vector<FragmentResult::ptr> FragmentController::getReceivedResults()
{
  ReceiveResultsOperation *receiveres = static_cast<ReceiveResultsOperation *>(
                Commands.getByName("receiveresults"));
  return receiveres->getResults();
}

/** Sends shutdown signal to server
 *
 * @param host address of server
 * @param service port/service of server
 */
void FragmentController::shutdown(
                const std::string &host,
                const std::string &service)
{
  ShutdownOperation *shutdown = static_cast<ShutdownOperation *>(
                Commands.getByName("shutdown"));
  (*shutdown)(host, service);
}