source: src/JobMarket/Jobs/SystemCommandJob.cpp@ 404d2b

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

/*
 * SystemCommandJob.cpp
 *
 *  Created on: Feb 5, 2012
 *      Author: heber
 */

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

#include <boost/iostreams/device/file_descriptor.hpp>
#include <boost/iostreams/stream.hpp>

#include "CodePatterns/MemDebug.hpp"

// include headers that implement a archive in simple text format
// otherwise BOOST_CLASS_EXPORT_IMPLEMENT has no effect
#include <boost/archive/text_oarchive.hpp>
#include <boost/archive/text_iarchive.hpp>

#include "JobMarket/Jobs/SystemCommandJob.hpp"

#include <cstdio>
#include <cstdlib>
#include <fcntl.h>
#include <fstream>
#include <iostream>
#include <string>
#include <streambuf>
#include <boost/filesystem.hpp>

#include "CodePatterns/Chronos.hpp"
#include "CodePatterns/Info.hpp"
#include "CodePatterns/Log.hpp"
#include "CodePatterns/toString.hpp"

/** Constructor for class SystemCommandJob.
 *
 */
SystemCommandJob::SystemCommandJob() :
  FragmentJob(JobId::IllegalJob)
{}

/** Constructor for class SystemCommandJob.
 *
 * \param _command command to execute
 * \param _outputfile configuration file for solver
 * \param _JobId unique id of this job
 * \param _suffix possible suffix to temporary file name, may be left empty
 */
SystemCommandJob::SystemCommandJob(
    const std::string &_command, 
    const std::string &_outputfile, 
    const JobId_t _JobId,
    const std::string &_suffix) :
  FragmentJob(_JobId),
  command(_command),
  suffix(_suffix),
  outputfile(_outputfile)
{}

/** Destructor for class SystemCommandJob.
 *
 */
SystemCommandJob::~SystemCommandJob()
{}

/** Work routine of this SystemCommandJob.
 *
 * This function encapsulates all the work that has to be done to generate
 * a FragmentResult. Hence, the FragmentWorker does not need to know anything
 * about the operation: it just receives it and executes this function.
 *
 * We obtain FragmentResult::exitflag from std::system's return value and
 * FragmentResult::result from the contents of the piped output file.
 *
 * \return result of this job
 */
FragmentResult::ptr SystemCommandJob::Work()
{
  Info info((std::string(__FUNCTION__)+std::string(", id #")+toString(getId())).c_str());

  // the following is taken from http://stackoverflow.com/questions/2746168/how-to-construct-a-c-fstream-from-a-posix-file-descriptor
  char *tmpTemplate = NULL;
  const std::string idstring(toString(getId()));
  const size_t idlength = idstring.length();
  {
    std::string Template("/tmp/XXXXXXX_");
    ASSERT(idlength <= 8,
      "SystemCommandJob::Work() - the id contains more than 8 digits.");
    Template += idstring;
    if (!suffix.empty())
      Template += suffix;
    LOG(2, "DEBUG: Temporary template is " << Template << ".");
    tmpTemplate = new char[Template.length()+1];
    strncpy(tmpTemplate, Template.c_str(), Template.length()+1);
  }
  const int fd = mkstemps(tmpTemplate, idlength + suffix.length()+1);

  // write outputfile to temporary file
  LOG(2, "DEBUG: Temporary file is " << tmpTemplate << ".");
  std::ofstream output(tmpTemplate);
  ASSERT(output.is_open(),
    "SystemCommandJob::Work() - the temporary file could not be opened.");
  output << outputfile << std::endl;
  output.close();

  // fork into subprocess and launch command
  const std::string WorkName = std::string("WORK")+idstring;
  Chronos::getInstance().startTiming(WorkName);
  FragmentResult::ptr s;
  {
    // open process
    std::string command_args = command+std::string(" ")+tmpTemplate;
    LOG(1, "INFO: Executing '" << command_args << "'.");
    FILE *stdoutstream = NULL;
    while (stdoutstream == NULL) {
      stdoutstream = popen(command_args.c_str(), "r");
      if (stdoutstream == NULL) {
        ELOG(2, "File descriptors are full, waiting for 1 sec...");
        sleep(1);
      }
    }

    int exitflag;
    // read stdout from process
    const int fd = fileno(stdoutstream);
    fcntl(fd, F_SETFD, fcntl(fd, F_GETFD) | FD_CLOEXEC);
    boost::iostreams::stream<boost::iostreams::file_descriptor_source> stdoutfile(fd, boost::iostreams::never_close_handle);
    stdoutfile.set_auto_close(false); // https://svn.boost.org/trac/boost/ticket/3517
    std::istreambuf_iterator<char> beginiter = (std::istreambuf_iterator<char>(stdoutfile));
    std::string resultstring( beginiter, std::istreambuf_iterator<char>());

    // construct result
    LOG(2, "DEBUG: First 50 characters of output: " << resultstring.substr(0,50));
    s = extractResult(resultstring);

    // end process
    if ((exitflag = pclose(stdoutstream)) == -1)
      ELOG(0, "pclose error");

    if (exitflag != 0)
      ELOG(1, "Job " << getId() << " failed on executing: " << command_args);
    s->exitflag = exitflag;
  }
  Chronos::getInstance().endTiming(WorkName);

  // remove temporary file
  boost::filesystem::path tmpPath(tmpTemplate);
  if (boost::filesystem::exists(boost::filesystem::status(tmpPath))) {
    LOG(2, "DEBUG: Removing " << tmpPath.string());
    boost::filesystem::remove(tmpPath);
  }
  delete[] tmpTemplate;
  // close temporary file!
  close(fd);

  // obtain timing and place in FragmentResult
  s->time_Work = Chronos::getInstance().getTime(WorkName);
  LOG(1, "INFO: Work() required " << s->time_Work << " seconds to complete.");

  // return result
  return s;
}

/** Default function for result extraction is just copy.
 *
 * @param resultstring output of system command
 * @return copy of \a resultstring
 */
FragmentResult::ptr SystemCommandJob::extractResult(const std::string &resultstring)
{
  return FragmentResult::ptr (new FragmentResult(getId(), resultstring) );
}


/** Comparator for class SystemCommandJob.
 * \param other instance to compare to
 * \return every member variable is the same, else - is not
 */
bool SystemCommandJob::operator==(const SystemCommandJob &other) const
{
  if (command != other.command) {
    LOG(1, "INFO: command's of two SystemCommandJobs differ: " << command << " != " << other.command << ".");
    return false;
  }
  if (outputfile != other.outputfile) {
    LOG(1, "INFO: outputfile's of two SystemCommandJobs differ: " << outputfile << " != " << other.outputfile << ".");
    return false;
  }
  return (dynamic_cast<const FragmentJob &>(*this) == dynamic_cast<const FragmentJob &>(other));
}

// we need to explicitly instantiate the serialization functions as
// its is only serialized through its base class FragmentJob
BOOST_CLASS_EXPORT_IMPLEMENT(SystemCommandJob)