src/Launcher/Launcher_Job.cxx

   1 // Copyright (C) 2009-2014  CEA/DEN, EDF R&D, OPEN CASCADE
   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, or (at your option) any later version.
   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 #ifdef WITH_LIBBATCH
  26 #include <libbatch/Constants.hxx>
  27 #endif
  28
  29 using namespace std;
  30
  31 Launcher::Job::Job()
  32 {
  33   _number = -1;
  34   _state = "CREATED";
  35   _launch_date = getLaunchDate();
  36
  37   _env_file = "";
  38   _job_name = "";
  39   _job_file = "";
  40   _job_file_name = "";
  41   _job_file_name_complete = "";
  42   _work_directory = "";
  43   _local_directory = "";
  44   _result_directory = "";
  45   _maximum_duration = "";
  46   _maximum_duration_in_second = -1;
  47   _queue = "";
  48   _job_type = "";
  49   _exclusive = false;
  50   _mem_per_cpu = 0;
  51
  52   // Parameters for COORM
  53   _launcher_file = "";
  54   _launcher_args = "";
  55
  56 #ifdef WITH_LIBBATCH
  57   _batch_job = new Batch::Job();
  58 #endif
  59 }
  60
  61 Launcher::Job::~Job()
  62 {
  63   LAUNCHER_MESSAGE("Deleting job number: " << _number);
  64 #ifdef WITH_LIBBATCH
  65   if (_batch_job)
  66     delete _batch_job;
  67 #endif
  68 }
  69
  70 void
  71 Launcher::Job::stopJob()
  72 {
  73   LAUNCHER_MESSAGE("Stop resquested for job number: " << _number);
  74   setState("FAILED");
  75 #ifdef WITH_LIBBATCH
  76   if (_batch_job_id.getReference() != "undefined")
  77   {
  78     try
  79     {
  80       _batch_job_id.deleteJob();
  81     }
  82     catch (const Batch::GenericException &ex)
  83     {
  84       LAUNCHER_INFOS("WARNING: exception when stopping the job: " << ex.message);
  85     }
  86   }
  87 #endif
  88 }
  89
  90
  91 void
  92 Launcher::Job::removeJob()
  93 {
  94   LAUNCHER_MESSAGE("Removing job number: " << _number);
  95 #ifdef WITH_LIBBATCH
  96   if (_batch_job_id.getReference() != "undefined")
  97   {
  98     try
  99     {
 100       _batch_job_id.deleteJob();
 101     }
 102     catch (const Batch::GenericException &ex)
 103     {
 104       LAUNCHER_INFOS("WARNING: exception when removing the job: " << ex.message);
 105     }
 106   }
 107 #endif
 108 }
 109
 110 std::string
 111 Launcher::Job::getJobType() const
 112 {
 113   return _job_type;
 114 }
 115
 116 void
 117 Launcher::Job::setJobName(const std::string & job_name)
 118 {
 119   _job_name = job_name;
 120 }
 121
 122 std::string
 123 Launcher::Job::getJobName() const
 124 {
 125   return _job_name;
 126 }
 127
 128 void
 129 Launcher::Job::setState(const std::string & state)
 130 {
 131   // State of a Job: CREATED, QUEUED, RUNNING, FINISHED, FAILED
 132   if (state != "CREATED" &&
 133       state != "IN_PROCESS" &&
 134       state != "QUEUED" &&
 135       state != "RUNNING" &&
 136       state != "PAUSED" &&
 137       state != "FINISHED" &&
 138       state != "FAILED" &&
 139       state != "ERROR")
 140   {
 141     throw LauncherException("Bad state, this state does not exist: " + state);
 142   }
 143   _state = state;
 144 }
 145
 146 std::string
 147 Launcher::Job::getState() const
 148 {
 149   return _state;
 150 }
 151
 152 // Get names or ids of hosts assigned to the job
 153 std::string
 154 Launcher::Job::getAssignedHostnames()
 155 {
 156   return _assigned_hostnames;
 157 }
 158
 159 void
 160 Launcher::Job::setNumber(const int & number)
 161 {
 162   if (_number != -1)
 163     std::cerr << "Launcher::Job::setNumber -- Job number was already defined, before: " << _number << " now: " << number << std::endl;
 164   _number = number;
 165 }
 166
 167 int
 168 Launcher::Job::getNumber()
 169 {
 170   return _number;
 171 }
 172
 173 void
 174 Launcher::Job::setResourceDefinition(const ParserResourcesType & resource_definition)
 175 {
 176   // Check machine_definition
 177   std::string user_name = "";
 178   if (resource_definition.UserName == "")
 179   {
 180     user_name = getenv("USER");
 181     if (user_name == "")
 182     {
 183       std::string mess = "You must define a user name: into your resource description or with env variable USER";
 184       throw LauncherException(mess);
 185     }
 186   }
 187   else
 188     user_name = resource_definition.UserName;
 189
 190   _resource_definition = resource_definition;
 191   _resource_definition.UserName = user_name;
 192 }
 193
 194 ParserResourcesType
 195 Launcher::Job::getResourceDefinition() const
 196 {
 197   return _resource_definition;
 198 }
 199
 200 void
 201 Launcher::Job::setJobFile(const std::string & job_file)
 202 {
 203   // Check job file
 204   if (job_file == "")
 205   {
 206     std::string mess = "Empty Job File is forbidden !";
 207     throw LauncherException(mess);
 208   }
 209
 210   _job_file = job_file;
 211   std::string::size_type p1 = _job_file.find_last_of("/");
 212   std::string::size_type p2 = _job_file.find_last_of(".");
 213   _job_file_name_complete = _job_file.substr(p1+1);
 214   _job_file_name = _job_file.substr(p1+1,p2-p1-1);
 215 }
 216
 217 std::string
 218 Launcher::Job::getJobFile() const
 219 {
 220   return _job_file;
 221 }
 222 void
 223 Launcher::Job::setEnvFile(const std::string & env_file)
 224 {
 225   _env_file = env_file;
 226 }
 227
 228 std::string
 229 Launcher::Job::getEnvFile() const
 230 {
 231   return _env_file;
 232 }
 233
 234 void
 235 Launcher::Job::setWorkDirectory(const std::string & work_directory)
 236 {
 237   _work_directory = work_directory;
 238 }
 239
 240 void
 241 Launcher::Job::setLocalDirectory(const std::string & local_directory)
 242 {
 243   _local_directory = local_directory;
 244 }
 245
 246 void
 247 Launcher::Job::setResultDirectory(const std::string & result_directory)
 248 {
 249   _result_directory = result_directory;
 250 }
 251
 252 void
 253 Launcher::Job::add_in_file(const std::string & file)
 254 {
 255   std::list<std::string>::iterator it = std::find(_in_files.begin(), _in_files.end(), file);
 256   if (it == _in_files.end())
 257     _in_files.push_back(file);
 258   else
 259     std::cerr << "Launcher::Job::add_in_file -- Warning file was already entered in in_files: " << file << std::endl;
 260 }
 261
 262 void
 263 Launcher::Job::add_out_file(const std::string & file)
 264 {
 265   std::list<std::string>::iterator it = std::find(_out_files.begin(), _out_files.end(), file);
 266   if (it == _out_files.end())
 267     _out_files.push_back(file);
 268   else
 269     std::cerr << "Launcher::Job::add_out_file -- Warning file was already entered in out_files: " << file << std::endl;
 270 }
 271
 272 void
 273 Launcher::Job::setMaximumDuration(const std::string & maximum_duration)
 274 {
 275   checkMaximumDuration(maximum_duration);
 276   _maximum_duration_in_second = convertMaximumDuration(maximum_duration);
 277   _maximum_duration = maximum_duration;
 278 }
 279
 280 // For COORM
 281 void
 282 Launcher::Job::setLauncherFile(const std::string & launcher_file)
 283 {
 284         _launcher_file = launcher_file;
 285 }
 286 void
 287 Launcher::Job::setLauncherArgs(const std::string & launcher_args)
 288 {
 289         _launcher_args = launcher_args;
 290 }
 291
 292 void
 293 Launcher::Job::setResourceRequiredParams(const resourceParams & resource_required_params)
 294 {
 295   checkResourceRequiredParams(resource_required_params);
 296   _resource_required_params = resource_required_params;
 297 }
 298
 299 void
 300 Launcher::Job::setQueue(const std::string & queue)
 301 {
 302   _queue = queue;
 303 }
 304
 305 void
 306 Launcher::Job::setExclusive(bool exclusive)
 307 {
 308   _exclusive = exclusive;
 309 }
 310
 311 void
 312 Launcher::Job::setExclusiveStr(const std::string & exclusiveStr)
 313 {
 314   if (exclusiveStr == "true")
 315     _exclusive = true;
 316   else if (exclusiveStr == "false")
 317     _exclusive = false;
 318   else
 319     throw LauncherException(std::string("Invalid boolean value for exclusive: ") + exclusiveStr);
 320 }
 321
 322 void
 323 Launcher::Job::setMemPerCpu(unsigned long mem_per_cpu)
 324 {
 325   _mem_per_cpu = mem_per_cpu;
 326 }
 327
 328 void
 329 Launcher::Job::setReference(const std::string & reference)
 330 {
 331   _reference = reference;
 332 }
 333
 334 std::string
 335 Launcher::Job::getWorkDirectory() const
 336 {
 337   return _work_directory;
 338 }
 339
 340 std::string
 341 Launcher::Job::getLocalDirectory() const
 342 {
 343   return _local_directory;
 344 }
 345
 346 std::string
 347 Launcher::Job::getResultDirectory() const
 348 {
 349   return _result_directory;
 350 }
 351
 352 const std::list<std::string> &
 353 Launcher::Job::get_in_files() const
 354 {
 355   return _in_files;
 356 }
 357
 358 const std::list<std::string> &
 359 Launcher::Job::get_out_files() const
 360 {
 361   return _out_files;
 362 }
 363
 364 std::string
 365 Launcher::Job::getMaximumDuration() const
 366 {
 367   return _maximum_duration;
 368 }
 369
 370 // For COORM
 371 std::string
 372 Launcher::Job::getLauncherFile() const
 373 {
 374         return _launcher_file;
 375 }
 376 std::string
 377 Launcher::Job::getLauncherArgs() const
 378 {
 379         return _launcher_args;
 380 }
 381
 382 resourceParams
 383 Launcher::Job::getResourceRequiredParams() const
 384 {
 385   return _resource_required_params;
 386 }
 387
 388 std::string
 389 Launcher::Job::getQueue() const
 390 {
 391   return _queue;
 392 }
 393
 394 bool
 395 Launcher::Job::getExclusive() const
 396 {
 397   return _exclusive;
 398 }
 399
 400 std::string
 401 Launcher::Job::getExclusiveStr() const
 402 {
 403   return _exclusive ? "true" : "false";
 404 }
 405
 406 unsigned long
 407 Launcher::Job::getMemPerCpu() const
 408 {
 409   return _mem_per_cpu;
 410 }
 411
 412 std::string
 413 Launcher::Job::getReference() const
 414 {
 415   return _reference;
 416 }
 417
 418 void
 419 Launcher::Job::checkMaximumDuration(const std::string & maximum_duration)
 420 {
 421   std::string result("");
 422   std::string edt_value = maximum_duration;
 423   std::size_t pos = edt_value.find(":");
 424
 425   if (edt_value != "") {
 426     if (pos == edt_value.npos) {
 427       throw LauncherException("[Launcher::Job::checkMaximumDuration] Error on definition: " + edt_value);
 428     }
 429     std::string begin_edt_value = edt_value.substr(0, pos);
 430     std::string mid_edt_value = edt_value.substr(pos, 1);
 431     std::string end_edt_value = edt_value.substr(pos + 1, edt_value.npos);
 432
 433     long value;
 434     std::istringstream iss(begin_edt_value);
 435     if (!(iss >> value)) {
 436       result = "[Launcher::Job::checkExpectedDuration] Error on definition ! : " + edt_value;
 437     }
 438     else if (value < 0) {
 439       result = "[Launcher::Job::checkExpectedDuration] Error on definition time is negative ! : " + value;
 440     }
 441     std::istringstream iss_2(end_edt_value);
 442     if (!(iss_2 >> value)) {
 443       result = "[Launcher::Job::checkExpectedDuration] Error on definition ! : " + edt_value;
 444     }
 445     else if (value < 0) {
 446       result = "[Launcher::Job::checkExpectedDuration] Error on definition time is negative ! : " + value;
 447     }
 448     if (mid_edt_value != ":") {
 449       result = "[Launcher::Job::checkExpectedDuration] Error on definition ! :" + edt_value;
 450     }
 451   }
 452   if (result != "")
 453     throw LauncherException(result);
 454 }
 455
 456 void
 457 Launcher::Job::checkResourceRequiredParams(const resourceParams & resource_required_params)
 458 {
 459   // nb_proc has be to > 0
 460   if (resource_required_params.nb_proc <= 0)
 461   {
 462     std::string message("[Launcher::Job::checkResourceRequiredParams] proc number is not > 0 ! ");
 463     throw LauncherException(message);
 464   }
 465 }
 466
 467 long
 468 Launcher::Job::convertMaximumDuration(const std::string & edt)
 469 {
 470   long hh, mm, ret;
 471
 472   if( edt.size() == 0 )
 473     return -1;
 474
 475   std::string::size_type pos = edt.find(":");
 476   std::string h = edt.substr(0,pos);
 477   std::string m = edt.substr(pos+1,edt.size()-pos+1);
 478   std::istringstream issh(h);
 479   issh >> hh;
 480   std::istringstream issm(m);
 481   issm >> mm;
 482   ret = hh*60 + mm;
 483   ret = ret * 60;
 484
 485   return ret;
 486 }
 487
 488 std::string
 489 Launcher::Job::getLaunchDate() const
 490 {
 491   time_t rawtime;
 492   time(&rawtime);
 493   std::string launch_date = ctime(&rawtime);
 494   int i = 0 ;
 495   for (;i < launch_date.size(); i++)
 496     if (launch_date[i] == '/' ||
 497         launch_date[i] == '-' ||
 498         launch_date[i] == ':' ||
 499         launch_date[i] == ' ')
 500       launch_date[i] = '_';
 501   launch_date.erase(--launch_date.end()); // Last caracter is a \n
 502
 503   return launch_date;
 504 }
 505
 506 std::string
 507 Launcher::Job::updateJobState()
 508 {
 509
 510   if (_state != "FINISHED" &&
 511       _state != "ERROR"    &&
 512       _state != "FAILED")
 513   {
 514 #ifdef WITH_LIBBATCH
 515     if (_batch_job_id.getReference() != "undefined")
 516     {
 517       // A batch manager has been affected to the job
 518       Batch::JobInfo job_info = _batch_job_id.queryJob();
 519       Batch::Parametre par = job_info.getParametre();
 520       _state = par[Batch::STATE].str();
 521       _assigned_hostnames = (par.find(Batch::ASSIGNEDHOSTNAMES) == par.end())?
 522                             "" : par[Batch::ASSIGNEDHOSTNAMES].str();
 523       LAUNCHER_MESSAGE("State received is: " << par[Batch::STATE].str());
 524     }
 525 #endif
 526   }
 527   return _state;
 528 }
 529
 530 #ifdef WITH_LIBBATCH
 531 Batch::Job *
 532 Launcher::Job::getBatchJob()
 533 {
 534   update_job();
 535   return _batch_job;
 536 }
 537
 538 Batch::Parametre
 539 Launcher::Job::common_job_params()
 540 {
 541   Batch::Parametre params;
 542
 543   params[Batch::NAME] = getJobName();
 544   params[Batch::NBPROC] = _resource_required_params.nb_proc;
 545   params[Batch::NBPROCPERNODE] = _resource_required_params.nb_proc_per_node;
 546
 547   // Memory in megabytes
 548   if (_resource_required_params.mem_mb > 0)
 549   {
 550     params[Batch::MAXRAMSIZE] = _resource_required_params.mem_mb;
 551   }
 552   else if (_mem_per_cpu > 0)
 553   {
 554     params[Batch::MEMPERCPU] = (long)_mem_per_cpu;
 555   }
 556
 557   // We define a default directory based on user time
 558   if (_work_directory == "")
 559   {
 560     const size_t BUFSIZE = 32;
 561     char date[BUFSIZE];
 562     time_t curtime = time(NULL);
 563     strftime(date, BUFSIZE, "%Y_%m_%d__%H_%M_%S", localtime(&curtime));
 564     _work_directory = std::string("$HOME/Batch/workdir_");
 565     _work_directory += date;
 566   }
 567   params[Batch::WORKDIR] = _work_directory;
 568
 569   // Parameters for COORM
 570   params[Batch::LAUNCHER_FILE] = _launcher_file;
 571   params[Batch::LAUNCHER_ARGS] = _launcher_args;
 572
 573   // If result_directory is not defined, we use HOME environnement
 574   if (_result_directory == "")
 575     _result_directory = getenv("HOME");
 576
 577   // _in_files
 578   std::list<std::string> in_files(_in_files);
 579   in_files.push_back(_job_file);
 580   if (_env_file != "")
 581           in_files.push_back(_env_file);
 582   for(std::list<std::string>::iterator it = in_files.begin(); it != in_files.end(); it++)
 583   {
 584     std::string file = *it;
 585
 586     // local file -> If file is not an absolute path, we apply _local_directory
 587     std::string local_file;
 588     if (file.substr(0, 1) == std::string("/"))
 589       local_file = file;
 590     else
 591 #ifndef WIN32
 592       local_file = _local_directory + "/" + file;
 593 #else
 594           local_file = file;
 595 #endif
 596
 597     // remote file -> get only file name from in_files
 598     size_t found = file.find_last_of("/");
 599     std::string remote_file = _work_directory + "/" + file.substr(found+1);
 600
 601     params[Batch::INFILE] += Batch::Couple(local_file, remote_file);
 602   }
 603
 604   // _out_files
 605   for(std::list<std::string>::iterator it = _out_files.begin(); it != _out_files.end(); it++)
 606   {
 607     std::string file = *it;
 608
 609     // local file
 610     size_t found = file.find_last_of("/");
 611     std::string local_file = _result_directory +  "/" + file.substr(found+1);
 612
 613     // remote file -> If file is not an absolute path, we apply _work_directory
 614     std::string remote_file;
 615     if (file.substr(0, 1) == std::string("/"))
 616       remote_file = file;
 617     else
 618       remote_file = _work_directory + "/" + file;
 619
 620     params[Batch::OUTFILE] += Batch::Couple(local_file, remote_file);
 621   }
 622
 623   // Time
 624   if (_maximum_duration_in_second != -1)
 625     params[Batch::MAXWALLTIME] = _maximum_duration_in_second / 60;
 626
 627   // Queue
 628   if (_queue != "")
 629     params[Batch::QUEUE] = _queue;
 630
 631   // Exclusive
 632   if (getExclusive())
 633     params[Batch::EXCLUSIVE] = true;
 634
 635   // Specific parameters
 636   std::map<std::string, std::string>::iterator it = _specific_parameters.find("LoalLevelerJobType");
 637   if (it != _specific_parameters.end())
 638     params["LL_JOBTYPE"] = it->second;
 639   return params;
 640 }
 641
 642 void
 643 Launcher::Job::setBatchManagerJobId(Batch::JobId batch_manager_job_id)
 644 {
 645   _batch_job_id = batch_manager_job_id;
 646 }
 647
 648 Batch::JobId
 649 Launcher::Job::getBatchManagerJobId() const
 650 {
 651   return _batch_job_id;
 652 }
 653 #endif
 654
 655 void
 656 Launcher::Job::addSpecificParameter(const std::string & name,
 657                                       const std::string & value)
 658 {
 659   _specific_parameters[name] = value;
 660 }
 661
 662 const std::map<std::string, std::string> &
 663 Launcher::Job::getSpecificParameters() const
 664 {
 665   return _specific_parameters;
 666 }
 667
 668 void
 669 Launcher::Job::checkSpecificParameters()
 670 {
 671 }