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::setWCKey(const std::string & wckey)
 330 {
 331   _wckey = wckey;
 332 }
 333
 334 void
 335 Launcher::Job::setReference(const std::string & reference)
 336 {
 337   _reference = reference;
 338 }
 339
 340 std::string
 341 Launcher::Job::getWorkDirectory() const
 342 {
 343   return _work_directory;
 344 }
 345
 346 std::string
 347 Launcher::Job::getLocalDirectory() const
 348 {
 349   return _local_directory;
 350 }
 351
 352 std::string
 353 Launcher::Job::getResultDirectory() const
 354 {
 355   return _result_directory;
 356 }
 357
 358 const std::list<std::string> &
 359 Launcher::Job::get_in_files() const
 360 {
 361   return _in_files;
 362 }
 363
 364 const std::list<std::string> &
 365 Launcher::Job::get_out_files() const
 366 {
 367   return _out_files;
 368 }
 369
 370 std::string
 371 Launcher::Job::getMaximumDuration() const
 372 {
 373   return _maximum_duration;
 374 }
 375
 376 // For COORM
 377 std::string
 378 Launcher::Job::getLauncherFile() const
 379 {
 380         return _launcher_file;
 381 }
 382 std::string
 383 Launcher::Job::getLauncherArgs() const
 384 {
 385         return _launcher_args;
 386 }
 387
 388 resourceParams
 389 Launcher::Job::getResourceRequiredParams() const
 390 {
 391   return _resource_required_params;
 392 }
 393
 394 std::string
 395 Launcher::Job::getQueue() const
 396 {
 397   return _queue;
 398 }
 399
 400 bool
 401 Launcher::Job::getExclusive() const
 402 {
 403   return _exclusive;
 404 }
 405
 406 std::string
 407 Launcher::Job::getExclusiveStr() const
 408 {
 409   return _exclusive ? "true" : "false";
 410 }
 411
 412 unsigned long
 413 Launcher::Job::getMemPerCpu() const
 414 {
 415   return _mem_per_cpu;
 416 }
 417
 418 std::string
 419 Launcher::Job::getWCKey() const
 420 {
 421   return _wckey;
 422 }
 423
 424 std::string
 425 Launcher::Job::getReference() const
 426 {
 427   return _reference;
 428 }
 429
 430 void
 431 Launcher::Job::checkMaximumDuration(const std::string & maximum_duration)
 432 {
 433   std::string result("");
 434   std::string edt_value = maximum_duration;
 435   std::size_t pos = edt_value.find(":");
 436
 437   if (edt_value != "") {
 438     if (pos == edt_value.npos) {
 439       throw LauncherException("[Launcher::Job::checkMaximumDuration] Error on definition: " + edt_value);
 440     }
 441     std::string begin_edt_value = edt_value.substr(0, pos);
 442     std::string mid_edt_value = edt_value.substr(pos, 1);
 443     std::string end_edt_value = edt_value.substr(pos + 1, edt_value.npos);
 444
 445     long value;
 446     std::istringstream iss(begin_edt_value);
 447     if (!(iss >> value)) {
 448       result = "[Launcher::Job::checkExpectedDuration] Error on definition ! : " + edt_value;
 449     }
 450     else if (value < 0) {
 451       result = "[Launcher::Job::checkExpectedDuration] Error on definition time is negative ! : " + value;
 452     }
 453     std::istringstream iss_2(end_edt_value);
 454     if (!(iss_2 >> value)) {
 455       result = "[Launcher::Job::checkExpectedDuration] Error on definition ! : " + edt_value;
 456     }
 457     else if (value < 0) {
 458       result = "[Launcher::Job::checkExpectedDuration] Error on definition time is negative ! : " + value;
 459     }
 460     if (mid_edt_value != ":") {
 461       result = "[Launcher::Job::checkExpectedDuration] Error on definition ! :" + edt_value;
 462     }
 463   }
 464   if (result != "")
 465     throw LauncherException(result);
 466 }
 467
 468 void
 469 Launcher::Job::checkResourceRequiredParams(const resourceParams & resource_required_params)
 470 {
 471   // nb_proc has be to > 0
 472   if (resource_required_params.nb_proc <= 0)
 473   {
 474     std::string message("[Launcher::Job::checkResourceRequiredParams] proc number is not > 0 ! ");
 475     throw LauncherException(message);
 476   }
 477 }
 478
 479 long
 480 Launcher::Job::convertMaximumDuration(const std::string & edt)
 481 {
 482   long hh, mm, ret;
 483
 484   if( edt.size() == 0 )
 485     return -1;
 486
 487   std::string::size_type pos = edt.find(":");
 488   std::string h = edt.substr(0,pos);
 489   std::string m = edt.substr(pos+1,edt.size()-pos+1);
 490   std::istringstream issh(h);
 491   issh >> hh;
 492   std::istringstream issm(m);
 493   issm >> mm;
 494   ret = hh*60 + mm;
 495   ret = ret * 60;
 496
 497   return ret;
 498 }
 499
 500 std::string
 501 Launcher::Job::getLaunchDate() const
 502 {
 503   time_t rawtime;
 504   time(&rawtime);
 505   std::string launch_date = ctime(&rawtime);
 506   int i = 0 ;
 507   for (;i < launch_date.size(); i++)
 508     if (launch_date[i] == '/' ||
 509         launch_date[i] == '-' ||
 510         launch_date[i] == ':' ||
 511         launch_date[i] == ' ')
 512       launch_date[i] = '_';
 513   launch_date.erase(--launch_date.end()); // Last caracter is a \n
 514
 515   return launch_date;
 516 }
 517
 518 std::string
 519 Launcher::Job::updateJobState()
 520 {
 521
 522   if (_state != "FINISHED" &&
 523       _state != "ERROR"    &&
 524       _state != "FAILED")
 525   {
 526 #ifdef WITH_LIBBATCH
 527     if (_batch_job_id.getReference() != "undefined")
 528     {
 529       // A batch manager has been affected to the job
 530       Batch::JobInfo job_info = _batch_job_id.queryJob();
 531       Batch::Parametre par = job_info.getParametre();
 532       _state = par[Batch::STATE].str();
 533       _assigned_hostnames = (par.find(Batch::ASSIGNEDHOSTNAMES) == par.end())?
 534                             "" : par[Batch::ASSIGNEDHOSTNAMES].str();
 535       LAUNCHER_MESSAGE("State received is: " << par[Batch::STATE].str());
 536     }
 537 #endif
 538   }
 539   return _state;
 540 }
 541
 542 #ifdef WITH_LIBBATCH
 543 Batch::Job *
 544 Launcher::Job::getBatchJob()
 545 {
 546   update_job();
 547   return _batch_job;
 548 }
 549
 550 Batch::Parametre
 551 Launcher::Job::common_job_params()
 552 {
 553   Batch::Parametre params;
 554
 555   params[Batch::NAME] = getJobName();
 556   params[Batch::NBPROC] = _resource_required_params.nb_proc;
 557   params[Batch::NBPROCPERNODE] = _resource_required_params.nb_proc_per_node;
 558
 559   // Memory in megabytes
 560   if (_resource_required_params.mem_mb > 0)
 561   {
 562     params[Batch::MAXRAMSIZE] = _resource_required_params.mem_mb;
 563   }
 564   else if (_mem_per_cpu > 0)
 565   {
 566     params[Batch::MEMPERCPU] = (long)_mem_per_cpu;
 567   }
 568
 569   // We define a default directory based on user time
 570   if (_work_directory == "")
 571   {
 572     const size_t BUFSIZE = 32;
 573     char date[BUFSIZE];
 574     time_t curtime = time(NULL);
 575     strftime(date, BUFSIZE, "%Y_%m_%d__%H_%M_%S", localtime(&curtime));
 576     _work_directory = std::string("$HOME/Batch/workdir_");
 577     _work_directory += date;
 578   }
 579   params[Batch::WORKDIR] = _work_directory;
 580
 581   // Parameters for COORM
 582   params[Batch::LAUNCHER_FILE] = _launcher_file;
 583   params[Batch::LAUNCHER_ARGS] = _launcher_args;
 584
 585   // If result_directory is not defined, we use HOME environnement
 586   if (_result_directory == "")
 587     _result_directory = getenv("HOME");
 588
 589   // _in_files
 590   std::list<std::string> in_files(_in_files);
 591   in_files.push_back(_job_file);
 592   if (_env_file != "")
 593           in_files.push_back(_env_file);
 594   for(std::list<std::string>::iterator it = in_files.begin(); it != in_files.end(); it++)
 595   {
 596     std::string file = *it;
 597
 598     // local file -> If file is not an absolute path, we apply _local_directory
 599     std::string local_file;
 600     if (file.substr(0, 1) == std::string("/"))
 601       local_file = file;
 602     else
 603 #ifndef WIN32
 604       local_file = _local_directory + "/" + file;
 605 #else
 606           local_file = file;
 607 #endif
 608
 609     // remote file -> get only file name from in_files
 610     size_t found = file.find_last_of("/");
 611     std::string remote_file = _work_directory + "/" + file.substr(found+1);
 612
 613     params[Batch::INFILE] += Batch::Couple(local_file, remote_file);
 614   }
 615
 616   // _out_files
 617   for(std::list<std::string>::iterator it = _out_files.begin(); it != _out_files.end(); it++)
 618   {
 619     std::string file = *it;
 620
 621     // local file
 622     size_t found = file.find_last_of("/");
 623     std::string local_file = _result_directory +  "/" + file.substr(found+1);
 624
 625     // remote file -> If file is not an absolute path, we apply _work_directory
 626     std::string remote_file;
 627     if (file.substr(0, 1) == std::string("/"))
 628       remote_file = file;
 629     else
 630       remote_file = _work_directory + "/" + file;
 631
 632     params[Batch::OUTFILE] += Batch::Couple(local_file, remote_file);
 633   }
 634
 635   // Time
 636   if (_maximum_duration_in_second != -1)
 637     params[Batch::MAXWALLTIME] = _maximum_duration_in_second / 60;
 638
 639   // Queue
 640   if (_queue != "")
 641     params[Batch::QUEUE] = _queue;
 642
 643   // Exclusive
 644   if (getExclusive())
 645     params[Batch::EXCLUSIVE] = true;
 646
 647   // WC Key
 648   if (_wckey != "")
 649     params[Batch::WCKEY] = _wckey;
 650
 651   // Specific parameters
 652   std::map<std::string, std::string>::iterator it = _specific_parameters.find("LoalLevelerJobType");
 653   if (it != _specific_parameters.end())
 654     params["LL_JOBTYPE"] = it->second;
 655   return params;
 656 }
 657
 658 void
 659 Launcher::Job::setBatchManagerJobId(Batch::JobId batch_manager_job_id)
 660 {
 661   _batch_job_id = batch_manager_job_id;
 662 }
 663
 664 Batch::JobId
 665 Launcher::Job::getBatchManagerJobId() const
 666 {
 667   return _batch_job_id;
 668 }
 669 #endif
 670
 671 void
 672 Launcher::Job::addSpecificParameter(const std::string & name,
 673                                       const std::string & value)
 674 {
 675   _specific_parameters[name] = value;
 676 }
 677
 678 const std::map<std::string, std::string> &
 679 Launcher::Job::getSpecificParameters() const
 680 {
 681   return _specific_parameters;
 682 }
 683
 684 void
 685 Launcher::Job::checkSpecificParameters()
 686 {
 687 }