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