src/Launcher/Launcher_Job.cxx

   1 // Copyright (C) 2009-2015  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 using namespace std;
  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     {
 184       std::string mess = "You must define a user name: into your resource description or with env variable USER";
 185       throw LauncherException(mess);
 186     }
 187   }
 188   else
 189     user_name = resource_definition.UserName;
 190
 191   _resource_definition = resource_definition;
 192   _resource_definition.UserName = user_name;
 193 }
 194
 195 ParserResourcesType
 196 Launcher::Job::getResourceDefinition() const
 197 {
 198   return _resource_definition;
 199 }
 200
 201 void
 202 Launcher::Job::setJobFile(const std::string & job_file)
 203 {
 204   // Check job file
 205   if (job_file == "")
 206   {
 207     std::string mess = "Empty Job File is forbidden !";
 208     throw LauncherException(mess);
 209   }
 210
 211   _job_file = job_file;
 212   std::string::size_type p1 = _job_file.find_last_of("/");
 213   std::string::size_type p2 = _job_file.find_last_of(".");
 214   _job_file_name_complete = _job_file.substr(p1+1);
 215   _job_file_name = _job_file.substr(p1+1,p2-p1-1);
 216 }
 217
 218 std::string
 219 Launcher::Job::getJobFile() const
 220 {
 221   return _job_file;
 222 }
 223 void
 224 Launcher::Job::setEnvFile(const std::string & env_file)
 225 {
 226   _env_file = env_file;
 227 }
 228
 229 std::string
 230 Launcher::Job::getEnvFile() const
 231 {
 232   return _env_file;
 233 }
 234
 235 void
 236 Launcher::Job::setWorkDirectory(const std::string & work_directory)
 237 {
 238   _work_directory = work_directory;
 239 }
 240
 241 void
 242 Launcher::Job::setLocalDirectory(const std::string & local_directory)
 243 {
 244   _local_directory = local_directory;
 245 }
 246
 247 void
 248 Launcher::Job::setResultDirectory(const std::string & result_directory)
 249 {
 250   _result_directory = result_directory;
 251 }
 252
 253 void
 254 Launcher::Job::add_in_file(const std::string & file)
 255 {
 256   std::list<std::string>::iterator it = std::find(_in_files.begin(), _in_files.end(), file);
 257   if (it == _in_files.end())
 258     _in_files.push_back(file);
 259   else
 260     std::cerr << "Launcher::Job::add_in_file -- Warning file was already entered in in_files: " << file << std::endl;
 261 }
 262
 263 void
 264 Launcher::Job::add_out_file(const std::string & file)
 265 {
 266   std::list<std::string>::iterator it = std::find(_out_files.begin(), _out_files.end(), file);
 267   if (it == _out_files.end())
 268     _out_files.push_back(file);
 269   else
 270     std::cerr << "Launcher::Job::add_out_file -- Warning file was already entered in out_files: " << file << std::endl;
 271 }
 272
 273 void
 274 Launcher::Job::setMaximumDuration(const std::string & maximum_duration)
 275 {
 276   checkMaximumDuration(maximum_duration);
 277   _maximum_duration_in_second = convertMaximumDuration(maximum_duration);
 278   _maximum_duration = maximum_duration;
 279 }
 280
 281 // For COORM
 282 void
 283 Launcher::Job::setLauncherFile(const std::string & launcher_file)
 284 {
 285         _launcher_file = launcher_file;
 286 }
 287 void
 288 Launcher::Job::setLauncherArgs(const std::string & launcher_args)
 289 {
 290         _launcher_args = launcher_args;
 291 }
 292
 293 void
 294 Launcher::Job::setResourceRequiredParams(const resourceParams & resource_required_params)
 295 {
 296   checkResourceRequiredParams(resource_required_params);
 297   _resource_required_params = resource_required_params;
 298 }
 299
 300 void
 301 Launcher::Job::setQueue(const std::string & queue)
 302 {
 303   _queue = queue;
 304 }
 305
 306 void
 307 Launcher::Job::setExclusive(bool exclusive)
 308 {
 309   _exclusive = exclusive;
 310 }
 311
 312 void
 313 Launcher::Job::setExclusiveStr(const std::string & exclusiveStr)
 314 {
 315   if (exclusiveStr == "true")
 316     _exclusive = true;
 317   else if (exclusiveStr == "false")
 318     _exclusive = false;
 319   else
 320     throw LauncherException(std::string("Invalid boolean value for exclusive: ") + exclusiveStr);
 321 }
 322
 323 void
 324 Launcher::Job::setMemPerCpu(unsigned long mem_per_cpu)
 325 {
 326   _mem_per_cpu = mem_per_cpu;
 327 }
 328
 329 void
 330 Launcher::Job::setWCKey(const std::string & wckey)
 331 {
 332   _wckey = wckey;
 333 }
 334
 335 void
 336 Launcher::Job::setExtraParams(const std::string & extra_params)
 337 {
 338   _extra_params = extra_params;
 339 }
 340
 341 void
 342 Launcher::Job::setReference(const std::string & reference)
 343 {
 344   _reference = reference;
 345 }
 346
 347 std::string
 348 Launcher::Job::getWorkDirectory() const
 349 {
 350   return _work_directory;
 351 }
 352
 353 std::string
 354 Launcher::Job::getLocalDirectory() const
 355 {
 356   return _local_directory;
 357 }
 358
 359 std::string
 360 Launcher::Job::getResultDirectory() const
 361 {
 362   return _result_directory;
 363 }
 364
 365 const std::list<std::string> &
 366 Launcher::Job::get_in_files() const
 367 {
 368   return _in_files;
 369 }
 370
 371 const std::list<std::string> &
 372 Launcher::Job::get_out_files() const
 373 {
 374   return _out_files;
 375 }
 376
 377 std::string
 378 Launcher::Job::getMaximumDuration() const
 379 {
 380   return _maximum_duration;
 381 }
 382
 383 // For COORM
 384 std::string
 385 Launcher::Job::getLauncherFile() const
 386 {
 387         return _launcher_file;
 388 }
 389 std::string
 390 Launcher::Job::getLauncherArgs() const
 391 {
 392         return _launcher_args;
 393 }
 394
 395 resourceParams
 396 Launcher::Job::getResourceRequiredParams() const
 397 {
 398   return _resource_required_params;
 399 }
 400
 401 std::string
 402 Launcher::Job::getQueue() const
 403 {
 404   return _queue;
 405 }
 406
 407 bool
 408 Launcher::Job::getExclusive() const
 409 {
 410   return _exclusive;
 411 }
 412
 413 std::string
 414 Launcher::Job::getExclusiveStr() const
 415 {
 416   return _exclusive ? "true" : "false";
 417 }
 418
 419 unsigned long
 420 Launcher::Job::getMemPerCpu() const
 421 {
 422   return _mem_per_cpu;
 423 }
 424
 425 std::string
 426 Launcher::Job::getWCKey() const
 427 {
 428   return _wckey;
 429 }
 430
 431 std::string
 432 Launcher::Job::getExtraParams() const
 433 {
 434   return _extra_params;
 435 }
 436
 437 std::string
 438 Launcher::Job::getReference() const
 439 {
 440   return _reference;
 441 }
 442
 443 void
 444 Launcher::Job::checkMaximumDuration(const std::string & maximum_duration)
 445 {
 446   std::string result("");
 447   std::string edt_value = maximum_duration;
 448   std::size_t pos = edt_value.find(":");
 449
 450   if (edt_value != "") {
 451     if (pos == edt_value.npos) {
 452       throw LauncherException("[Launcher::Job::checkMaximumDuration] Error on definition: " + edt_value);
 453     }
 454     std::string begin_edt_value = edt_value.substr(0, pos);
 455     std::string mid_edt_value = edt_value.substr(pos, 1);
 456     std::string end_edt_value = edt_value.substr(pos + 1, edt_value.npos);
 457
 458     long value;
 459     std::istringstream iss(begin_edt_value);
 460     if (!(iss >> value)) {
 461       result = "[Launcher::Job::checkExpectedDuration] Error on definition ! : " + edt_value;
 462     }
 463     else if (value < 0) {
 464       result = "[Launcher::Job::checkExpectedDuration] Error on definition time is negative ! : " + value;
 465     }
 466     std::istringstream iss_2(end_edt_value);
 467     if (!(iss_2 >> value)) {
 468       result = "[Launcher::Job::checkExpectedDuration] Error on definition ! : " + edt_value;
 469     }
 470     else if (value < 0) {
 471       result = "[Launcher::Job::checkExpectedDuration] Error on definition time is negative ! : " + value;
 472     }
 473     if (mid_edt_value != ":") {
 474       result = "[Launcher::Job::checkExpectedDuration] Error on definition ! :" + edt_value;
 475     }
 476   }
 477   if (result != "")
 478     throw LauncherException(result);
 479 }
 480
 481 void
 482 Launcher::Job::checkResourceRequiredParams(const resourceParams & resource_required_params)
 483 {
 484   // nb_proc has be to > 0
 485   if (resource_required_params.nb_proc <= 0)
 486   {
 487     std::string message("[Launcher::Job::checkResourceRequiredParams] proc number is not > 0 ! ");
 488     throw LauncherException(message);
 489   }
 490 }
 491
 492 long
 493 Launcher::Job::convertMaximumDuration(const std::string & edt)
 494 {
 495   long hh, mm, ret;
 496
 497   if( edt.size() == 0 )
 498     return -1;
 499
 500   std::string::size_type pos = edt.find(":");
 501   std::string h = edt.substr(0,pos);
 502   std::string m = edt.substr(pos+1,edt.size()-pos+1);
 503   std::istringstream issh(h);
 504   issh >> hh;
 505   std::istringstream issm(m);
 506   issm >> mm;
 507   ret = hh*60 + mm;
 508   ret = ret * 60;
 509
 510   return ret;
 511 }
 512
 513 std::string
 514 Launcher::Job::getLaunchDate() const
 515 {
 516   time_t rawtime;
 517   time(&rawtime);
 518   std::string launch_date = ctime(&rawtime);
 519   int i = 0 ;
 520   for (;i < launch_date.size(); i++)
 521     if (launch_date[i] == '/' ||
 522         launch_date[i] == '-' ||
 523         launch_date[i] == ':' ||
 524         launch_date[i] == ' ')
 525       launch_date[i] = '_';
 526   launch_date.erase(--launch_date.end()); // Last caracter is a \n
 527
 528   return launch_date;
 529 }
 530
 531 std::string
 532 Launcher::Job::updateJobState()
 533 {
 534
 535   if (_state != "FINISHED" &&
 536       _state != "ERROR"    &&
 537       _state != "FAILED")
 538   {
 539 #ifdef WITH_LIBBATCH
 540     if (_batch_job_id.getReference() != "undefined")
 541     {
 542       // A batch manager has been affected to the job
 543       Batch::JobInfo job_info = _batch_job_id.queryJob();
 544       Batch::Parametre par = job_info.getParametre();
 545       _state = par[Batch::STATE].str();
 546       _assigned_hostnames = (par.find(Batch::ASSIGNEDHOSTNAMES) == par.end())?
 547                             "" : par[Batch::ASSIGNEDHOSTNAMES].str();
 548       LAUNCHER_MESSAGE("State received is: " << par[Batch::STATE].str());
 549     }
 550 #endif
 551   }
 552   return _state;
 553 }
 554
 555 #ifdef WITH_LIBBATCH
 556 Batch::Job *
 557 Launcher::Job::getBatchJob()
 558 {
 559   update_job();
 560   return _batch_job;
 561 }
 562
 563 Batch::Parametre
 564 Launcher::Job::common_job_params()
 565 {
 566   Batch::Parametre params;
 567
 568   params[Batch::NAME] = getJobName();
 569   params[Batch::NBPROC] = _resource_required_params.nb_proc;
 570   params[Batch::NBPROCPERNODE] = _resource_required_params.nb_proc_per_node;
 571
 572   // Memory in megabytes
 573   if (_resource_required_params.mem_mb > 0)
 574   {
 575     params[Batch::MAXRAMSIZE] = _resource_required_params.mem_mb;
 576   }
 577   else if (_mem_per_cpu > 0)
 578   {
 579     params[Batch::MEMPERCPU] = (long)_mem_per_cpu;
 580   }
 581
 582   // We define a default directory
 583   if (_work_directory == "")
 584   {
 585     const size_t BUFSIZE = 32;
 586     char date[BUFSIZE];
 587     time_t curtime = time(NULL);
 588     strftime(date, BUFSIZE, "%Y_%m_%d__%H_%M_%S", localtime(&curtime));
 589     if(!_resource_definition.working_directory.empty())
 590     {
 591       std::string job_dir = std::string("/job_") + date;
 592       _work_directory = _resource_definition.working_directory + job_dir;
 593     }
 594     else
 595     {
 596       _work_directory = std::string("/$HOME/Batch/workdir_");
 597       _work_directory += date;
 598     }
 599   }
 600   params[Batch::WORKDIR] = _work_directory;
 601
 602   // Parameters for COORM
 603   params[Batch::LAUNCHER_FILE] = _launcher_file;
 604   params[Batch::LAUNCHER_ARGS] = _launcher_args;
 605
 606   // If result_directory is not defined, we use HOME environnement
 607   if (_result_directory == "")
 608     _result_directory = getenv("HOME");
 609
 610   // _in_files
 611   std::list<std::string> in_files(_in_files);
 612   in_files.push_back(_job_file);
 613   if (_env_file != "")
 614           in_files.push_back(_env_file);
 615   for(std::list<std::string>::iterator it = in_files.begin(); it != in_files.end(); it++)
 616   {
 617     std::string file = *it;
 618
 619     // local file -> If file is not an absolute path, we apply _local_directory
 620     std::string local_file;
 621     if (file.substr(0, 1) == std::string("/"))
 622       local_file = file;
 623     else
 624 #ifndef WIN32
 625       local_file = _local_directory + "/" + file;
 626 #else
 627       local_file = file;
 628 #endif
 629
 630     // remote file -> get only file name from in_files
 631     size_t found = file.find_last_of("/");
 632     std::string remote_file = _work_directory + "/" + file.substr(found+1);
 633
 634     params[Batch::INFILE] += Batch::Couple(local_file, remote_file);
 635   }
 636
 637   // _out_files
 638   for(std::list<std::string>::iterator it = _out_files.begin(); it != _out_files.end(); it++)
 639   {
 640     std::string file = *it;
 641     // remote file -> If file is not an absolute path, we apply _work_directory
 642     std::string remote_file;
 643     std::string local_file;
 644     if (file.substr(0, 1) == std::string("/"))
 645     {
 646       remote_file = file;
 647       size_t found = file.find_last_of("/");
 648       local_file = file.substr(found+1);
 649     }
 650     else
 651     {
 652       remote_file = _work_directory + "/" + file;
 653       local_file = file;
 654     }
 655
 656     params[Batch::OUTFILE] += Batch::Couple(local_file, remote_file);
 657   }
 658
 659   // Time
 660   if (_maximum_duration_in_second != -1)
 661     params[Batch::MAXWALLTIME] = _maximum_duration_in_second / 60;
 662
 663   // Queue
 664   if (_queue != "")
 665     params[Batch::QUEUE] = _queue;
 666
 667   // Exclusive
 668   if (getExclusive())
 669     params[Batch::EXCLUSIVE] = true;
 670
 671   // WC Key
 672   if (_wckey != "")
 673     params[Batch::WCKEY] = _wckey;
 674
 675   // Extra params
 676   if (_extra_params != "")
 677     params[Batch::EXTRAPARAMS] = _extra_params;
 678
 679   // Specific parameters
 680   std::map<std::string, std::string>::iterator it = _specific_parameters.find("LoalLevelerJobType");
 681   if (it != _specific_parameters.end())
 682     params["LL_JOBTYPE"] = it->second;
 683   return params;
 684 }
 685
 686 void
 687 Launcher::Job::setBatchManagerJobId(Batch::JobId batch_manager_job_id)
 688 {
 689   _batch_job_id = batch_manager_job_id;
 690 }
 691
 692 Batch::JobId
 693 Launcher::Job::getBatchManagerJobId() const
 694 {
 695   return _batch_job_id;
 696 }
 697 #endif
 698
 699 void
 700 Launcher::Job::addSpecificParameter(const std::string & name,
 701                                       const std::string & value)
 702 {
 703   _specific_parameters[name] = value;
 704 }
 705
 706 const std::map<std::string, std::string> &
 707 Launcher::Job::getSpecificParameters() const
 708 {
 709   return _specific_parameters;
 710 }
 711
 712 void
 713 Launcher::Job::checkSpecificParameters()
 714 {
 715 }