src/Launcher/Launcher_Job.cxx

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