src/Launcher/Launcher_Job.cxx

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