src/Launcher/Launcher_Job.cxx

   1 //  Copyright (C) 2009-2010  CEA/DEN, EDF R&D
   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 // Author: André RIBES - EDF R&D
  21 //
  22 #include "Launcher_Job.hxx"
  23 #include "Launcher.hxx"
  24
  25 Launcher::Job::Job()
  26 {
  27   _number = -1;
  28   _state = "CREATED";
  29   _launch_date = getLaunchDate();
  30
  31   _env_file = "";
  32   _job_file = "";
  33   _job_file_name = "";
  34   _job_file_name_complete = "";
  35   _work_directory = "";
  36   _local_directory = "";
  37   _result_directory = "";
  38   _maximum_duration = "";
  39   _maximum_duration_in_second = -1;
  40   _resource_required_params.name = "";
  41   _resource_required_params.hostname = "";
  42   _resource_required_params.OS = "";
  43   _resource_required_params.nb_proc = -1;
  44   _resource_required_params.nb_node = -1;
  45   _resource_required_params.nb_proc_per_node = -1;
  46   _resource_required_params.cpu_clock = -1;
  47   _resource_required_params.mem_mb = -1;
  48   _queue = "";
  49
  50 #ifdef WITH_LIBBATCH
  51   _batch_job = new Batch::Job();
  52 #endif
  53 }
  54
  55 Launcher::Job::~Job()
  56 {
  57   LAUNCHER_MESSAGE("Deleting job number: " << _number);
  58 #ifdef WITH_LIBBATCH
  59   if (_batch_job_id.getReference() != "undefined")
  60   {
  61     try
  62     {
  63       _batch_job_id.deleteJob();
  64     }
  65     catch (const Batch::EmulationException &ex)
  66     {
  67       LAUNCHER_INFOS("WARNING: exception when deleting the job: " << ex.message);
  68     }
  69   }
  70   if (_batch_job)
  71     delete _batch_job;
  72 #endif
  73 }
  74
  75 void
  76 Launcher::Job::setState(const std::string & state)
  77 {
  78   // State of a Job: CREATED, QUEUED, RUNNING, FINISHED, FAILED
  79   if (state != "CREATED" &&
  80       state != "QUEUED" &&
  81       state != "RUNNING" &&
  82       state != "FINISHED" &&
  83       state != "FAILED")
  84   {
  85     throw LauncherException("Bad state, this state does not exist: " + state);
  86   }
  87   _state = state;
  88 }
  89
  90 std::string
  91 Launcher::Job::getState()
  92 {
  93   return _state;
  94 }
  95
  96 void
  97 Launcher::Job::setNumber(const int & number)
  98 {
  99   if (_number != -1)
 100     std::cerr << "Launcher::Job::setNumber -- Job number was already defined, before: " << _number << " now: " << number << std::endl;
 101   _number = number;
 102 }
 103
 104 int
 105 Launcher::Job::getNumber()
 106 {
 107   return _number;
 108 }
 109
 110 void
 111 Launcher::Job::setResourceDefinition(const ParserResourcesType & resource_definition)
 112 {
 113   // Check machine_definition
 114   std::string user_name = "";
 115   if (resource_definition.UserName == "")
 116   {
 117     user_name = getenv("USER");
 118     if (user_name == "")
 119     {
 120       std::string mess = "You must define a user name: into your resource description or with env variable USER";
 121       throw LauncherException(mess);
 122     }
 123   }
 124   else
 125     user_name = resource_definition.UserName;
 126
 127   _resource_definition = resource_definition;
 128   _resource_definition.UserName = user_name;
 129 }
 130
 131 ParserResourcesType
 132 Launcher::Job::getResourceDefinition()
 133 {
 134   return _resource_definition;
 135 }
 136
 137 void
 138 Launcher::Job::setJobFile(const std::string & job_file)
 139 {
 140   // Check job file
 141   if (job_file == "")
 142   {
 143     std::string mess = "Empty Job File is forbidden !";
 144     throw LauncherException(mess);
 145   }
 146
 147   _job_file = job_file;
 148   std::string::size_type p1 = _job_file.find_last_of("/");
 149   std::string::size_type p2 = _job_file.find_last_of(".");
 150   _job_file_name_complete = _job_file.substr(p1+1);
 151   _job_file_name = _job_file.substr(p1+1,p2-p1-1);
 152
 153   if (_job_file != "")
 154     add_in_file(_job_file);
 155 }
 156
 157 std::string
 158 Launcher::Job::getJobFile()
 159 {
 160   return _job_file;
 161 }
 162 void
 163 Launcher::Job::setEnvFile(const std::string & env_file)
 164 {
 165   _env_file = env_file;
 166   if (_env_file != "")
 167     add_in_file(_env_file);
 168 }
 169
 170 std::string
 171 Launcher::Job::getEnvFile()
 172 {
 173   return _env_file;
 174 }
 175
 176 void
 177 Launcher::Job::setWorkDirectory(const std::string & work_directory)
 178 {
 179   _work_directory = work_directory;
 180 }
 181
 182 void
 183 Launcher::Job::setLocalDirectory(const std::string & local_directory)
 184 {
 185   _local_directory = local_directory;
 186 }
 187
 188 void
 189 Launcher::Job::setResultDirectory(const std::string & result_directory)
 190 {
 191   _result_directory = result_directory;
 192 }
 193
 194 void
 195 Launcher::Job::add_in_file(const std::string & file)
 196 {
 197   std::list<std::string>::iterator it = std::find(_in_files.begin(), _in_files.end(), file);
 198   if (it == _in_files.end())
 199     _in_files.push_back(file);
 200   else
 201     std::cerr << "Launcher::Job::add_in_file -- Warning file was already entered in in_files: " << file << std::endl;
 202 }
 203
 204 void
 205 Launcher::Job::add_out_file(const std::string & file)
 206 {
 207   std::list<std::string>::iterator it = std::find(_out_files.begin(), _out_files.end(), file);
 208   if (it == _out_files.end())
 209     _out_files.push_back(file);
 210   else
 211     std::cerr << "Launcher::Job::add_out_file -- Warning file was already entered in out_files: " << file << std::endl;
 212 }
 213
 214 void
 215 Launcher::Job::setMaximumDuration(const std::string & maximum_duration)
 216 {
 217   checkMaximumDuration(maximum_duration);
 218   _maximum_duration_in_second = convertMaximumDuration(maximum_duration);
 219   _maximum_duration = maximum_duration;
 220 }
 221
 222 void
 223 Launcher::Job::setResourceRequiredParams(const resourceParams & resource_required_params)
 224 {
 225   checkResourceRequiredParams(resource_required_params);
 226   _resource_required_params = resource_required_params;
 227 }
 228
 229 void
 230 Launcher::Job::setQueue(const std::string & queue)
 231 {
 232   _queue = queue;
 233 }
 234
 235 std::string
 236 Launcher::Job::getWorkDirectory()
 237 {
 238   return _work_directory;
 239 }
 240
 241 std::string
 242 Launcher::Job::getLocalDirectory()
 243 {
 244   return _local_directory;
 245 }
 246
 247 std::string
 248 Launcher::Job::getResultDirectory()
 249 {
 250   return _result_directory;
 251 }
 252
 253 const std::list<std::string> &
 254 Launcher::Job::get_in_files()
 255 {
 256   return _in_files;
 257 }
 258
 259 const std::list<std::string> &
 260 Launcher::Job::get_out_files()
 261 {
 262   return _out_files;
 263 }
 264
 265 std::string
 266 Launcher::Job::getMaximumDuration()
 267 {
 268   return _maximum_duration;
 269 }
 270
 271 resourceParams
 272 Launcher::Job::getResourceRequiredParams()
 273 {
 274   return _resource_required_params;
 275 }
 276
 277 std::string
 278 Launcher::Job::getQueue()
 279 {
 280   return _queue;
 281 }
 282
 283 void
 284 Launcher::Job::checkMaximumDuration(const std::string & maximum_duration)
 285 {
 286   std::string result("");
 287   std::string edt_value = maximum_duration;
 288   std::size_t pos = edt_value.find(":");
 289
 290   if (edt_value != "") {
 291     std::string begin_edt_value = edt_value.substr(0, pos);
 292     std::string mid_edt_value = edt_value.substr(pos, 1);
 293     std::string end_edt_value = edt_value.substr(pos + 1, edt_value.npos);
 294
 295     long value;
 296     std::istringstream iss(begin_edt_value);
 297     if (!(iss >> value)) {
 298       result = "[Launcher::Job::checkExpectedDuration] Error on definition ! : " + edt_value;
 299     }
 300     else if (value < 0) {
 301       result = "[Launcher::Job::checkExpectedDuration] Error on definition time is negative ! : " + value;
 302     }
 303     std::istringstream iss_2(end_edt_value);
 304     if (!(iss_2 >> value)) {
 305       result = "[Launcher::Job::checkExpectedDuration] Error on definition ! : " + edt_value;
 306     }
 307     else if (value < 0) {
 308       result = "[Launcher::Job::checkExpectedDuration] Error on definition time is negative ! : " + value;
 309     }
 310     if (mid_edt_value != ":") {
 311       result = "[Launcher::Job::checkExpectedDuration] Error on definition ! :" + edt_value;
 312     }
 313   }
 314   if (result != "")
 315     throw LauncherException(result);
 316 }
 317
 318 void
 319 Launcher::Job::checkResourceRequiredParams(const resourceParams & resource_required_params)
 320 {
 321   // nb_proc has be to > 0
 322   if (resource_required_params.nb_proc <= 0)
 323   {
 324     std::string message("[Launcher::Job::checkResourceRequiredParams] proc number is not > 0 ! ");
 325     throw LauncherException(message);
 326   }
 327 }
 328
 329 long
 330 Launcher::Job::convertMaximumDuration(const std::string & edt)
 331 {
 332   long hh, mm, ret;
 333
 334   if( edt.size() == 0 )
 335     return -1;
 336
 337   std::string::size_type pos = edt.find(":");
 338   std::string h = edt.substr(0,pos);
 339   std::string m = edt.substr(pos+1,edt.size()-pos+1);
 340   std::istringstream issh(h);
 341   issh >> hh;
 342   std::istringstream issm(m);
 343   issm >> mm;
 344   ret = hh*60 + mm;
 345   ret = ret * 60;
 346
 347   return ret;
 348 }
 349
 350 std::string
 351 Launcher::Job::getLaunchDate()
 352 {
 353   time_t rawtime;
 354   time(&rawtime);
 355   std::string launch_date = ctime(&rawtime);
 356   int i = 0 ;
 357   for (;i < launch_date.size(); i++)
 358     if (launch_date[i] == '/' ||
 359         launch_date[i] == '-' ||
 360         launch_date[i] == ':' ||
 361         launch_date[i] == ' ')
 362       launch_date[i] = '_';
 363   launch_date.erase(--launch_date.end()); // Last caracter is a \n
 364
 365   return launch_date;
 366 }
 367
 368 std::string
 369 Launcher::Job::updateJobState()
 370 {
 371 #ifdef WITH_LIBBATCH
 372   if (_batch_job_id.getReference() != "undefined")
 373   {
 374     // A batch manager has been affected to the job
 375     Batch::JobInfo job_info = _batch_job_id.queryJob();
 376     Batch::Parametre par = job_info.getParametre();
 377
 378     LAUNCHER_MESSAGE("State received is: " << par[Batch::STATE].str());
 379
 380     // TODO: Remove this if all tests pass with the new libBatch, otherwise fix the codes in libBatch
 381     // Patch until new LIBBATCH version
 382     // eSSH Client and ePBS Client and eSGE
 383 /*    if (par[STATE].str() == "Running" or par[STATE].str() == "E" or par[STATE].str() == "R" or par[STATE].str() == "r" or par[STATE].str() == "RUN")
 384       _state = "RUNNING";
 385     else if (par[STATE].str() == "Stopped")
 386       _state = "PAUSED";
 387     else if (par[STATE].str() == "Done" or par[STATE].str() == "U" or par[STATE].str() == "e" or par[STATE].str() == "DONE" or par[STATE].str() == "EXIT")
 388       _state = "FINISHED";
 389     else if (par[STATE].str() == "Dead" or par[STATE].str() == "Eqw")
 390       _state = "ERROR";
 391     else if (par[STATE].str() == "Q" or par[STATE].str() == "qw" or par[STATE].str() == "PEN")
 392       _state = "QUEUED";*/
 393     _state = par[Batch::STATE].str();
 394   }
 395 #endif
 396   return _state;
 397 }
 398
 399 #ifdef WITH_LIBBATCH
 400 Batch::Job *
 401 Launcher::Job::getBatchJob()
 402 {
 403   update_job();
 404   return _batch_job;
 405 }
 406
 407 Batch::Parametre
 408 Launcher::Job::common_job_params()
 409 {
 410   Batch::Parametre params;
 411
 412   params[Batch::USER] = _resource_definition.UserName;
 413   params[Batch::NBPROC] = _resource_required_params.nb_proc;
 414
 415   // Memory
 416   if (_resource_required_params.mem_mb > 0)
 417   {
 418     // Memory is in kilobytes
 419     params[Batch::MAXRAMSIZE] = _resource_required_params.mem_mb * 1024;
 420   }
 421
 422   // We define a default directory based on user time
 423   if (_work_directory == "")
 424   {
 425     std::string thedate;
 426     Batch::Date date = Batch::Date(time(0));
 427     thedate = date.str();
 428     int lend = thedate.size() ;
 429     int i = 0 ;
 430     while ( i < lend ) {
 431       if ( thedate[i] == '/' || thedate[i] == '-' || thedate[i] == ':' ) {
 432         thedate[i] = '_' ;
 433       }
 434       i++ ;
 435     }
 436     _work_directory = std::string("$HOME/Batch/");
 437     _work_directory += thedate;
 438   }
 439   params[Batch::WORKDIR] = _work_directory;
 440   params[Batch::TMPDIR] = _work_directory; // To Compatibility -- remove ??? TODO
 441
 442   // If result_directory is not defined, we use HOME environnement
 443   if (_result_directory == "")
 444     _result_directory = getenv("HOME");
 445
 446   // _in_files
 447   for(std::list<std::string>::iterator it = _in_files.begin(); it != _in_files.end(); it++)
 448   {
 449     std::string file = *it;
 450
 451     // local file -> If file is not an absolute path, we apply _local_directory
 452     std::string local_file;
 453     if (file.substr(0, 1) == std::string("/"))
 454       local_file = file;
 455     else
 456       local_file = _local_directory + "/" + file;
 457
 458     // remote file -> get only file name from _in_files
 459     size_t found = file.find_last_of("/");
 460     std::string remote_file = _work_directory + "/" + file.substr(found+1);
 461
 462     params[Batch::INFILE] += Batch::Couple(local_file, remote_file);
 463   }
 464
 465   // _out_files
 466   for(std::list<std::string>::iterator it = _out_files.begin(); it != _out_files.end(); it++)
 467   {
 468     std::string file = *it;
 469
 470     // local file
 471     size_t found = file.find_last_of("/");
 472     std::string local_file = _result_directory +  "/" + file.substr(found+1);
 473
 474     // remote file -> If file is not an absolute path, we apply _work_directory
 475     std::string remote_file;
 476     if (file.substr(0, 1) == std::string("/"))
 477       remote_file = file;
 478     else
 479       remote_file = _work_directory + "/" + file;
 480
 481     params[Batch::OUTFILE] += Batch::Couple(local_file, remote_file);
 482   }
 483
 484   // Time
 485   if (_maximum_duration_in_second != -1)
 486     params[Batch::MAXWALLTIME] = _maximum_duration_in_second;
 487
 488   // Queue
 489   if (_queue != "")
 490     params[Batch::QUEUE] = _queue;
 491
 492   return params;
 493 }
 494
 495 void
 496 Launcher::Job::setBatchManagerJobId(Batch::JobId batch_manager_job_id)
 497 {
 498   _batch_job_id = batch_manager_job_id;
 499 }
 500
 501 Batch::JobId
 502 Launcher::Job::getBatchManagerJobId()
 503 {
 504   return _batch_job_id;
 505 }
 506 #endif