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