src/Launcher/Launcher_Job.cxx

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