source: src/JobMarket/Controller/controller_SystemCommandJob.cpp@ 404d2b

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

/*
 * controller_SystemCommandJob.cpp
 *
 *  Created on: 01.06.2012
 *      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 "JobMarket/Controller/controller_SystemCommandJob.hpp"

#include <boost/assign.hpp>
#include <boost/bind.hpp>
#include "CodePatterns/Log.hpp"
#include <vector>

#include "JobMarket/Controller/ControllerCommand.hpp"
#include "JobMarket/Controller/ControllerCommandRegistry.hpp"
#include "JobMarket/Controller/ControllerOptions_SystemCommandJob.hpp"
#include "JobMarket/Controller/FragmentController.hpp"
#include "JobMarket/Jobs/NoOpJob.hpp"
#include "JobMarket/Jobs/SystemCommandJob.hpp"
#include "JobMarket/Results/FragmentResult.hpp"
#include "JobMarket/types.hpp"


/** Print received results.
 *
 * @param results received results to print
 */
void printReceivedResults(const std::vector<FragmentResult::ptr> &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));
}

/** Creates a SystemCommandJob out of give \a command with \a argument.
 *
 * @param controller reference to controller to add jobs
 * @param ControllerInfo information on the job
 */
void createJobs(FragmentController &controller, const ControllerOptions_SystemCommandJob &ControllerInfo)
{
  if ((!ControllerInfo.jobcommand.empty()) && (!ControllerInfo.executable.empty())) {
    LOG(2, "DEBUG: Creating " << ControllerInfo.times << " new SystemCommandJob(s) with '"
        +toString(ControllerInfo.executable)+"' and argument '"
        +toString(ControllerInfo.jobcommand)+"'.");
    std::vector<FragmentJob::ptr> jobs;
    for (size_t times=ControllerInfo.times; times--;) {
      const JobId_t next_id = controller.getAvailableId();
      FragmentJob::ptr testJob(
          new SystemCommandJob(ControllerInfo.executable, ControllerInfo.jobcommand, next_id) );
      jobs.push_back(testJob);
    }
    controller.addJobs(jobs);
    controller.sendJobs(ControllerInfo.server, ControllerInfo.serverport);
    LOG(1, "INFO: Added SystemCommandJobs.");
  } else {
    LOG(2, "DEBUG: Creating " << ControllerInfo.times << " new NoOpJob(s).");
    std::vector<FragmentJob::ptr> jobs;
    for (size_t times=ControllerInfo.times; times--;) {
      const JobId_t next_id = controller.getAvailableId();
      FragmentJob::ptr testJob( new NoOpJob(next_id) );
      jobs.push_back(testJob);
    }
    controller.addJobs(jobs);
    controller.sendJobs(ControllerInfo.server, ControllerInfo.serverport);
    LOG(1, "INFO: Added NoOpJobs.");
  }
}

inline std::vector<std::string> getListOfCommands(const ControllerCommandRegistry &ControllerCommands)
{
  std::vector<std::string> Commands;
  for (ControllerCommandRegistry::const_iterator iter = ControllerCommands.getBeginIter();
      iter != ControllerCommands.getEndIter(); ++iter)
    Commands.push_back(iter->first);

  return Commands;
}

ControllerOptions *controller_SystemCommandJob::allocateControllerInfo()
{
  return new ControllerOptions_SystemCommandJob();
}

void controller_SystemCommandJob::addSpecificCommands(
    boost::function<void (ControllerCommand *)> &registrator,
    FragmentController &controller,
    ControllerOptions &ControllerInfo)
{
  ControllerOptions_SystemCommandJob &CI =
      reinterpret_cast<ControllerOptions_SystemCommandJob &>(ControllerInfo);
  registrator(new ControllerCommand("createjobs",
      boost::assign::list_of< ControllerCommand::commands_t >
        (boost::bind(&FragmentController::requestIds,
            boost::ref(controller),
            boost::cref(ControllerInfo.server), boost::cref(ControllerInfo.serverport), boost::cref(CI.times)))
        (boost::bind(&createJobs, boost::ref(controller), boost::cref(CI)))
      ));
  registrator(new ControllerCommand("receiveresults",
      boost::assign::list_of< ControllerCommand::commands_t >
        (boost::bind(&FragmentController_JobIdProxy::setJobids,
            boost::ref(controller), boost::cref(CI.ids)))
        (boost::bind(&FragmentController::receiveResults,
            boost::ref(controller), boost::cref(ControllerInfo.server), boost::cref(ControllerInfo.serverport)))
        (boost::bind(&printReceivedResults,
            boost::bind(&FragmentController::getReceivedResults, boost::ref(controller))))
      ));
}

void controller_SystemCommandJob::addSpecificOptions(
    boost::program_options::options_description_easy_init option)
{
  option
    ("jobcommand", boost::program_options::value< std::string >(), "command argument for executable for 'createjobs'")
    ("executable", boost::program_options::value< std::string >(), "executable for commands 'createjobs'")
    ("ids", boost::program_options::value< std::vector<JobId_t> >()->multitoken(), "ids to receive for 'receiveresults'")
    ("times", boost::program_options::value< size_t >(), "sets the number of times the 'createjobs' job is added")
    ;
}

int controller_SystemCommandJob::addOtherParsings(
    ControllerOptions &ControllerInfo,
    boost::program_options::variables_map &vm)
{
  ControllerOptions_SystemCommandJob &CI =
      reinterpret_cast<ControllerOptions_SystemCommandJob &>(ControllerInfo);
  int status = 0;
  status = CI.parseExecutable(vm);
  if (status) return status;
  status = CI.parseJobCommand(vm);
  if (status) return status;
  status = CI.parseTimes(vm);
  if (status) return status;
  status = CI.parseIds(vm);
  return status;
}