src/COORM/BatchManager_COORM.cxx

   1 //  Copyright (C) 2012-2013  INRIA
   2 //
   3 //  This library is free software; you can redistribute it and/or
   4 //  modify it under the terms of the GNU Lesser General Public
   5 //  License as published by the Free Software Foundation; either
   6 //  version 2.1 of the License.
   7 //
   8 //  This library is distributed in the hope that it will be useful,
   9 //  but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 //  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  11 //  Lesser General Public License for more details.
  12 //
  13 //  You should have received a copy of the GNU Lesser General Public
  14 //  License along with this library; if not, write to the Free Software
  15 //  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  16 //
  17 //  See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
  18 //
  19
  20 #include <cstdlib>
  21 #include <fstream>
  22
  23 #include <Constants.hxx>
  24 #include <Utils.hxx>
  25 #include <Log.hxx>
  26
  27 #include "BatchManager_COORM.hxx"
  28 #include "JobInfo_COORM.hxx"
  29
  30 using namespace std;
  31
  32 namespace Batch
  33 {
  34         BatchManager_COORM::BatchManager_COORM(const FactBatchManager * parent, const char * host,
  35                         const char * username,
  36                         CommunicationProtocolType protocolType, const char * mpiImpl)
  37                 : BatchManager(parent, host, username, protocolType, mpiImpl)
  38         {
  39         }
  40
  41         BatchManager_COORM::~BatchManager_COORM()
  42         {
  43         }
  44
  45         // Soumet un job au gestionnaire
  46         const JobId BatchManager_COORM::submitJob(const Job & job)
  47         {
  48                 Parametre params = job.getParametre();
  49                 const string workDir = params[WORKDIR];
  50                 const string fileToExecute = params[EXECUTABLE];
  51
  52                 // For CooRM
  53                 const string launcherArgs = params[LAUNCHER_ARGS];
  54                 const string launcherFile  = params[LAUNCHER_FILE];
  55
  56                 const string jobName = params[NAME];
  57
  58                 string::size_type p1 = fileToExecute.find_last_of("/");
  59                 string::size_type p2 = fileToExecute.find_last_of(".");
  60                 std::string fileNameToExecute = fileToExecute.substr(p1+1,p2-p1-1);
  61
  62                 // For CooRM
  63                 p1 = launcherFile.find_last_of("/");
  64                 std::string fileNameToLaunch = launcherFile.substr(p1+1);
  65
  66
  67                 // export input files on cluster
  68                 exportInputFiles(job);
  69
  70                 // build batch script for job
  71                 string scriptFile = buildBatchScript(job);
  72
  73                 // Get REMOTE_COORM_PATH environment variable
  74                 const char * remote_coorm_path = getenv("REMOTE_COORM_PATH");
  75                 if (remote_coorm_path == NULL)
  76                 {
  77                         throw RunTimeException("Unable to get REMOTE_COORM_PATH environment variable");
  78                 }
  79
  80                 // We need omniORB to execute launcher.py
  81                 const string set_env_cmd = "source " + string(remote_coorm_path) + "/coorm_prerequis.sh;";
  82
  83
  84                 // define command to submit batch
  85                 string subCommand = set_env_cmd + "python " + workDir + "/" + fileNameToLaunch + " --name="+ jobName +
  86                                     " --workdir=" + workDir + " --outputs=" + workDir + "/logs/outputs.log" +
  87                                     " --errors=" + workDir + "/logs/errors.log" +
  88                                     " --executable=" + scriptFile + " " + launcherArgs;
  89     string command = _protocol.getExecCommand(subCommand, _hostname, _username);
  90     command += " 2>&1";
  91     LOG(command);
  92
  93     // submit job
  94     string output;
  95     int status = Utils::getCommandOutput(command, output);
  96     LOG(output);
  97     if (status != 0) throw RunTimeException("Can't submit job, error was: " + output);
  98
  99     // read id of submitted job in output
 100     istringstream logfile(output);
 101     string sline, idline, id;
 102
 103                 if (logfile)
 104                 {
 105                         while (getline(logfile, sline) && sline != "")
 106                         {
 107                                 idline = sline;
 108                         }
 109
 110                         vector<string> tokens;
 111                         JobInfo::Tokenize(idline, tokens, "=");
 112                         id = tokens[1] ;
 113                 }
 114                 else
 115                 {
 116                         throw RunTimeException("Error in the submission of the job on the remote host");
 117                 }
 118
 119                 JobId jobid(this, (string) id);
 120                 return jobid;
 121         }
 122
 123         // retire un job du gestionnaire
 124         void BatchManager_COORM::deleteJob(const JobId & jobid)
 125         {
 126     // Get REMOTE_COORM_PATH environment variable
 127     const char * remote_coorm_path = getenv("REMOTE_COORM_PATH");
 128     if (remote_coorm_path == NULL)
 129     {
 130       throw RunTimeException("Unable to get REMOTE_COORM_PATH environment variable");
 131     }
 132
 133     // We need omniORB to execute launcher.py
 134     const string set_env_cmd = "source " + string(remote_coorm_path) + "/coorm_prerequis.sh;";
 135
 136     // define command to delete job
 137     string subCommand = set_env_cmd + "python " + string(remote_coorm_path) + "/coormdel.py --jobID=" + jobid.getReference();
 138     string command = _protocol.getExecCommand(subCommand, _hostname, _username);
 139     LOG(command);
 140
 141     int status = system(command.c_str());
 142     if (status)
 143       throw RunTimeException("Can't delete job " + jobid.getReference());
 144
 145     LOG("job " << jobid.getReference() << " killed");
 146         }
 147
 148         // Renvoie l'etat du job
 149         JobInfo BatchManager_COORM::queryJob(const JobId & jobid)
 150         {
 151                 // Get REMOTE_COORM_PATH environment variable
 152                 const char * remote_coorm_path = getenv("REMOTE_COORM_PATH");
 153                 if (remote_coorm_path == NULL)
 154                 {
 155                         throw RunTimeException("Unable to get REMOTE_COORM_PATH environment variable");
 156                 }
 157
 158                 // We need omniORB to execute launcher.py
 159                 const string set_env_cmd = "source " + string(remote_coorm_path) + "/coorm_prerequis.sh;";
 160
 161                 // define command to query batch
 162                 string subCommand = set_env_cmd + "python " + string(remote_coorm_path) + "/coormstat.py --jobID=" + jobid.getReference();
 163     string command = _protocol.getExecCommand(subCommand, _hostname, _username);
 164     LOG(command);
 165     string output;
 166     int status = Utils::getCommandOutput(command, output);
 167     if(status && status != 153 && status != 256*153)
 168       throw RunTimeException("Error of connection on remote host");
 169
 170     JobInfo_COORM jobinfo = JobInfo_COORM(jobid.getReference(), output);
 171     return jobinfo;
 172         }
 173
 174         string BatchManager_COORM::buildBatchScript(const Job & job)
 175         {
 176                 Parametre params = job.getParametre();
 177
 178                 // Job Parameters
 179                 string workDir       = "";
 180                 string fileToExecute = "";
 181                 string tmpDir = "";
 182                 int nbproc               = 0;
 183                 int edt          = 0;
 184                 int mem              = 0;
 185                 string queue         = "";
 186
 187     // Mandatory parameters
 188     if (params.find(WORKDIR) != params.end())
 189       workDir = params[WORKDIR].str();
 190     else
 191       throw RunTimeException("params[WORKDIR] is not defined. Please define it, cannot submit this job.");
 192     if (params.find(EXECUTABLE) != params.end())
 193       fileToExecute = params[EXECUTABLE].str();
 194     else
 195       throw RunTimeException("params[EXECUTABLE] is not defined. Please define it, cannot submit this job.");
 196
 197                 string::size_type p1 = fileToExecute.find_last_of("/");
 198                 string::size_type p2 = fileToExecute.find_last_of(".");
 199                 string rootNameToExecute = fileToExecute.substr(p1+1,p2-p1-1);
 200                 string fileNameToExecute = fileToExecute.substr(p1+1);
 201
 202     // Create batch submit file
 203     ofstream tempOutputFile;
 204     string tmpFileName = Utils::createAndOpenTemporaryFile("COORM-script", tempOutputFile);
 205
 206                 tempOutputFile << "#!/bin/sh -f" << endl;
 207                 tempOutputFile << "export LIBBATCH_NODEFILE=$COORM_NODEFILE" << endl;
 208                 // Launch the executable
 209                 tempOutputFile << "cd " << tmpDir << endl;
 210                 tempOutputFile << "./" + fileNameToExecute << endl;
 211                 tempOutputFile.flush();
 212                 tempOutputFile.close();
 213
 214                 Utils::chmod(tmpFileName.c_str(), 0x1ED);
 215                 LOG("Batch script file generated is: " << tmpFileName);
 216
 217                 string remoteFileName = rootNameToExecute + "_Batch.sh";
 218     int status = _protocol.copyFile(tmpFileName, "", "",
 219                                     workDir + "/" + remoteFileName,
 220                                     _hostname, _username);
 221     if (status)
 222       throw RunTimeException("Cannot copy batch submission file on host " + _hostname);
 223
 224     return remoteFileName;
 225         }
 226
 227         void BatchManager_COORM::exportInputFiles(const Job & job)
 228         {
 229                 BatchManager::exportInputFiles(job);
 230
 231                 int status;
 232                 Parametre params = job.getParametre();
 233
 234                 string launcherFile = params[LAUNCHER_FILE];
 235
 236                 if (launcherFile.size() != 0)
 237                 {
 238                         // Copy launcherFile into batch working directory
 239       status = _protocol.copyFile(launcherFile, "", "",
 240                                   params[WORKDIR], _hostname, _username);
 241       if (status) {
 242         std::ostringstream oss;
 243         oss << "Cannot copy file " << launcherFile << " on host " << _hostname;
 244         oss << ". Return status is " << status;
 245         throw RunTimeException(oss.str());
 246       }
 247
 248       string remoteLauncher = launcherFile;
 249       remoteLauncher = remoteLauncher.substr(remoteLauncher.rfind("/") + 1, remoteLauncher.length());
 250       remoteLauncher = string(params[WORKDIR]) + "/" + remoteLauncher;
 251
 252       string subCommand = string("chmod u+x ") + remoteLauncher;
 253       string command = _protocol.getExecCommand(subCommand, _hostname, _username);
 254       LOG(command);
 255       status = system(command.c_str());
 256       if (status) {
 257         std::ostringstream oss;
 258         oss << "Cannot change permissions of file " << remoteLauncher << " on host " << _hostname;
 259         oss << ". Return status is " << status;
 260         throw RunTimeException(oss.str());
 261       }
 262                 }
 263         }
 264 }