src/Launcher/Launcher_Job.cxx

   1 // Copyright (C) 2009-2011  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.
   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 <Batch/Batch_Constants.hxx>
  27 #endif
  28
  29 Launcher::Job::Job()
  30 {
  31   _number = -1;
  32   _state = "CREATED";
  33   _launch_date = getLaunchDate();
  34
  35   _env_file = "";
  36   _job_name = "";
  37   _job_file = "";
  38   _job_file_name = "";
  39   _job_file_name_complete = "";
  40   _work_directory = "";
  41   _local_directory = "";
  42   _result_directory = "";
  43   _maximum_duration = "";
  44   _maximum_duration_in_second = -1;
  45   _resource_required_params.name = "";
  46   _resource_required_params.hostname = "";
  47   _resource_required_params.OS = "";
  48   _resource_required_params.nb_proc = -1;
  49   _resource_required_params.nb_node = -1;
  50   _resource_required_params.nb_proc_per_node = -1;
  51   _resource_required_params.cpu_clock = -1;
  52   _resource_required_params.mem_mb = -1;
  53   _queue = "";
  54   _job_type = "";
  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::EmulationException &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::EmulationException &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()
 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()
 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()
 148 {
 149   return _state;
 150 }
 151
 152 void
 153 Launcher::Job::setNumber(const int & number)
 154 {
 155   if (_number != -1)
 156     std::cerr << "Launcher::Job::setNumber -- Job number was already defined, before: " << _number << " now: " << number << std::endl;
 157   _number = number;
 158 }
 159
 160 int
 161 Launcher::Job::getNumber()
 162 {
 163   return _number;
 164 }
 165
 166 void
 167 Launcher::Job::setResourceDefinition(const ParserResourcesType & resource_definition)
 168 {
 169   // Check machine_definition
 170   std::string user_name = "";
 171   if (resource_definition.UserName == "")
 172   {
 173     user_name = getenv("USER");
 174     if (user_name == "")
 175     {
 176       std::string mess = "You must define a user name: into your resource description or with env variable USER";
 177       throw LauncherException(mess);
 178     }
 179   }
 180   else
 181     user_name = resource_definition.UserName;
 182
 183   _resource_definition = resource_definition;
 184   _resource_definition.UserName = user_name;
 185 }
 186
 187 ParserResourcesType
 188 Launcher::Job::getResourceDefinition()
 189 {
 190   return _resource_definition;
 191 }
 192
 193 void
 194 Launcher::Job::setJobFile(const std::string & job_file)
 195 {
 196   // Check job file
 197   if (job_file == "")
 198   {
 199     std::string mess = "Empty Job File is forbidden !";
 200     throw LauncherException(mess);
 201   }
 202
 203   _job_file = job_file;
 204   std::string::size_type p1 = _job_file.find_last_of("/");
 205   std::string::size_type p2 = _job_file.find_last_of(".");
 206   _job_file_name_complete = _job_file.substr(p1+1);
 207   _job_file_name = _job_file.substr(p1+1,p2-p1-1);
 208 }
 209
 210 std::string
 211 Launcher::Job::getJobFile()
 212 {
 213   return _job_file;
 214 }
 215 void
 216 Launcher::Job::setEnvFile(const std::string & env_file)
 217 {
 218   _env_file = env_file;
 219 }
 220
 221 std::string
 222 Launcher::Job::getEnvFile()
 223 {
 224   return _env_file;
 225 }
 226
 227 void
 228 Launcher::Job::setWorkDirectory(const std::string & work_directory)
 229 {
 230   _work_directory = work_directory;
 231 }
 232
 233 void
 234 Launcher::Job::setLocalDirectory(const std::string & local_directory)
 235 {
 236   _local_directory = local_directory;
 237 }
 238
 239 void
 240 Launcher::Job::setResultDirectory(const std::string & result_directory)
 241 {
 242   _result_directory = result_directory;
 243 }
 244
 245 void
 246 Launcher::Job::add_in_file(const std::string & file)
 247 {
 248   std::list<std::string>::iterator it = std::find(_in_files.begin(), _in_files.end(), file);
 249   if (it == _in_files.end())
 250     _in_files.push_back(file);
 251   else
 252     std::cerr << "Launcher::Job::add_in_file -- Warning file was already entered in in_files: " << file << std::endl;
 253 }
 254
 255 void
 256 Launcher::Job::add_out_file(const std::string & file)
 257 {
 258   std::list<std::string>::iterator it = std::find(_out_files.begin(), _out_files.end(), file);
 259   if (it == _out_files.end())
 260     _out_files.push_back(file);
 261   else
 262     std::cerr << "Launcher::Job::add_out_file -- Warning file was already entered in out_files: " << file << std::endl;
 263 }
 264
 265 void
 266 Launcher::Job::setMaximumDuration(const std::string & maximum_duration)
 267 {
 268   checkMaximumDuration(maximum_duration);
 269   _maximum_duration_in_second = convertMaximumDuration(maximum_duration);
 270   _maximum_duration = maximum_duration;
 271 }
 272
 273 void
 274 Launcher::Job::setResourceRequiredParams(const resourceParams & resource_required_params)
 275 {
 276   checkResourceRequiredParams(resource_required_params);
 277   _resource_required_params = resource_required_params;
 278 }
 279
 280 void
 281 Launcher::Job::setQueue(const std::string & queue)
 282 {
 283   _queue = queue;
 284 }
 285
 286 std::string
 287 Launcher::Job::getWorkDirectory()
 288 {
 289   return _work_directory;
 290 }
 291
 292 std::string
 293 Launcher::Job::getLocalDirectory()
 294 {
 295   return _local_directory;
 296 }
 297
 298 std::string
 299 Launcher::Job::getResultDirectory()
 300 {
 301   return _result_directory;
 302 }
 303
 304 const std::list<std::string> &
 305 Launcher::Job::get_in_files()
 306 {
 307   return _in_files;
 308 }
 309
 310 const std::list<std::string> &
 311 Launcher::Job::get_out_files()
 312 {
 313   return _out_files;
 314 }
 315
 316 std::string
 317 Launcher::Job::getMaximumDuration()
 318 {
 319   return _maximum_duration;
 320 }
 321
 322 resourceParams
 323 Launcher::Job::getResourceRequiredParams()
 324 {
 325   return _resource_required_params;
 326 }
 327
 328 std::string
 329 Launcher::Job::getQueue()
 330 {
 331   return _queue;
 332 }
 333
 334 void
 335 Launcher::Job::checkMaximumDuration(const std::string & maximum_duration)
 336 {
 337   std::string result("");
 338   std::string edt_value = maximum_duration;
 339   std::size_t pos = edt_value.find(":");
 340
 341   if (edt_value != "") {
 342     std::string begin_edt_value = edt_value.substr(0, pos);
 343     std::string mid_edt_value = edt_value.substr(pos, 1);
 344     std::string end_edt_value = edt_value.substr(pos + 1, edt_value.npos);
 345
 346     long value;
 347     std::istringstream iss(begin_edt_value);
 348     if (!(iss >> value)) {
 349       result = "[Launcher::Job::checkExpectedDuration] Error on definition ! : " + edt_value;
 350     }
 351     else if (value < 0) {
 352       result = "[Launcher::Job::checkExpectedDuration] Error on definition time is negative ! : " + value;
 353     }
 354     std::istringstream iss_2(end_edt_value);
 355     if (!(iss_2 >> value)) {
 356       result = "[Launcher::Job::checkExpectedDuration] Error on definition ! : " + edt_value;
 357     }
 358     else if (value < 0) {
 359       result = "[Launcher::Job::checkExpectedDuration] Error on definition time is negative ! : " + value;
 360     }
 361     if (mid_edt_value != ":") {
 362       result = "[Launcher::Job::checkExpectedDuration] Error on definition ! :" + edt_value;
 363     }
 364   }
 365   if (result != "")
 366     throw LauncherException(result);
 367 }
 368
 369 void
 370 Launcher::Job::checkResourceRequiredParams(const resourceParams & resource_required_params)
 371 {
 372   // nb_proc has be to > 0
 373   if (resource_required_params.nb_proc <= 0)
 374   {
 375     std::string message("[Launcher::Job::checkResourceRequiredParams] proc number is not > 0 ! ");
 376     throw LauncherException(message);
 377   }
 378 }
 379
 380 long
 381 Launcher::Job::convertMaximumDuration(const std::string & edt)
 382 {
 383   long hh, mm, ret;
 384
 385   if( edt.size() == 0 )
 386     return -1;
 387
 388   std::string::size_type pos = edt.find(":");
 389   std::string h = edt.substr(0,pos);
 390   std::string m = edt.substr(pos+1,edt.size()-pos+1);
 391   std::istringstream issh(h);
 392   issh >> hh;
 393   std::istringstream issm(m);
 394   issm >> mm;
 395   ret = hh*60 + mm;
 396   ret = ret * 60;
 397
 398   return ret;
 399 }
 400
 401 std::string
 402 Launcher::Job::getLaunchDate()
 403 {
 404   time_t rawtime;
 405   time(&rawtime);
 406   std::string launch_date = ctime(&rawtime);
 407   int i = 0 ;
 408   for (;i < launch_date.size(); i++)
 409     if (launch_date[i] == '/' ||
 410         launch_date[i] == '-' ||
 411         launch_date[i] == ':' ||
 412         launch_date[i] == ' ')
 413       launch_date[i] = '_';
 414   launch_date.erase(--launch_date.end()); // Last caracter is a \n
 415
 416   return launch_date;
 417 }
 418
 419 std::string
 420 Launcher::Job::updateJobState()
 421 {
 422
 423   if (_state != "FINISHED" &&
 424       _state != "ERROR"    &&
 425       _state != "FAILED")
 426   {
 427 #ifdef WITH_LIBBATCH
 428     if (_batch_job_id.getReference() != "undefined")
 429     {
 430       // A batch manager has been affected to the job
 431       Batch::JobInfo job_info = _batch_job_id.queryJob();
 432       Batch::Parametre par = job_info.getParametre();
 433       _state = par[Batch::STATE].str();
 434       LAUNCHER_MESSAGE("State received is: " << par[Batch::STATE].str());
 435     }
 436 #endif
 437   }
 438   return _state;
 439 }
 440
 441 #ifdef WITH_LIBBATCH
 442 Batch::Job *
 443 Launcher::Job::getBatchJob()
 444 {
 445   update_job();
 446   return _batch_job;
 447 }
 448
 449 Batch::Parametre
 450 Launcher::Job::common_job_params()
 451 {
 452   Batch::Parametre params;
 453
 454   params[Batch::NAME] = getJobName();
 455   params[Batch::USER] = _resource_definition.UserName;
 456   params[Batch::NBPROC] = _resource_required_params.nb_proc;
 457
 458   // Memory in megabytes
 459   if (_resource_required_params.mem_mb > 0)
 460   {
 461     params[Batch::MAXRAMSIZE] = _resource_required_params.mem_mb;
 462   }
 463
 464   // We define a default directory based on user time
 465   if (_work_directory == "")
 466   {
 467     std::string thedate;
 468     Batch::Date date = Batch::Date(time(0));
 469     thedate = date.str();
 470     int lend = thedate.size() ;
 471     int i = 0 ;
 472     while ( i < lend ) {
 473       if ( thedate[i] == '/' || thedate[i] == '-' || thedate[i] == ':' ) {
 474         thedate[i] = '_' ;
 475       }
 476       i++ ;
 477     }
 478     _work_directory = std::string("$HOME/Batch/");
 479     _work_directory += thedate;
 480   }
 481   params[Batch::WORKDIR] = _work_directory;
 482   params[Batch::TMPDIR] = _work_directory; // To Compatibility -- remove ??? TODO
 483
 484   // If result_directory is not defined, we use HOME environnement
 485   if (_result_directory == "")
 486     _result_directory = getenv("HOME");
 487
 488   // _in_files
 489   std::list<std::string> in_files(_in_files);
 490   in_files.push_back(_job_file);
 491   if (_env_file != "")
 492           in_files.push_back(_env_file);
 493   for(std::list<std::string>::iterator it = in_files.begin(); it != in_files.end(); it++)
 494   {
 495     std::string file = *it;
 496
 497     // local file -> If file is not an absolute path, we apply _local_directory
 498     std::string local_file;
 499     if (file.substr(0, 1) == std::string("/"))
 500       local_file = file;
 501     else
 502       local_file = _local_directory + "/" + file;
 503
 504     // remote file -> get only file name from in_files
 505     size_t found = file.find_last_of("/");
 506     std::string remote_file = _work_directory + "/" + file.substr(found+1);
 507
 508     params[Batch::INFILE] += Batch::Couple(local_file, remote_file);
 509   }
 510
 511   // _out_files
 512   for(std::list<std::string>::iterator it = _out_files.begin(); it != _out_files.end(); it++)
 513   {
 514     std::string file = *it;
 515
 516     // local file
 517     size_t found = file.find_last_of("/");
 518     std::string local_file = _result_directory +  "/" + file.substr(found+1);
 519
 520     // remote file -> If file is not an absolute path, we apply _work_directory
 521     std::string remote_file;
 522     if (file.substr(0, 1) == std::string("/"))
 523       remote_file = file;
 524     else
 525       remote_file = _work_directory + "/" + file;
 526
 527     params[Batch::OUTFILE] += Batch::Couple(local_file, remote_file);
 528   }
 529
 530   // Time
 531   if (_maximum_duration_in_second != -1)
 532     params[Batch::MAXWALLTIME] = _maximum_duration_in_second / 60;
 533
 534   // Queue
 535   if (_queue != "")
 536     params[Batch::QUEUE] = _queue;
 537
 538   // Specific parameters
 539   std::map<std::string, std::string>::iterator it = _specific_parameters.find("LoalLevelerJobType");
 540   if (it != _specific_parameters.end())
 541     params["LL_JOBTYPE"] = it->second;
 542   return params;
 543 }
 544
 545 void
 546 Launcher::Job::setBatchManagerJobId(Batch::JobId batch_manager_job_id)
 547 {
 548   _batch_job_id = batch_manager_job_id;
 549 }
 550
 551 Batch::JobId
 552 Launcher::Job::getBatchManagerJobId()
 553 {
 554   return _batch_job_id;
 555 }
 556 #endif
 557
 558 void
 559 Launcher::Job::addToXmlDocument(xmlNodePtr root_node)
 560 {
 561   // Begin job
 562   xmlNodePtr job_node = xmlNewChild(root_node, NULL, xmlCharStrdup("job"), NULL);
 563   xmlNewProp(job_node, xmlCharStrdup("type"), xmlCharStrdup(getJobType().c_str()));
 564   xmlNewProp(job_node, xmlCharStrdup("name"), xmlCharStrdup(getJobName().c_str()));
 565
 566   // Add user part
 567   xmlNodePtr node = xmlNewChild(job_node, NULL, xmlCharStrdup("user_part"), NULL);
 568
 569   xmlNewChild(node, NULL, xmlCharStrdup("job_file"),         xmlCharStrdup(getJobFile().c_str()));
 570   xmlNewChild(node, NULL, xmlCharStrdup("env_file"),         xmlCharStrdup(getEnvFile().c_str()));
 571   xmlNewChild(node, NULL, xmlCharStrdup("work_directory"),   xmlCharStrdup(getWorkDirectory().c_str()));
 572   xmlNewChild(node, NULL, xmlCharStrdup("local_directory"),  xmlCharStrdup(getLocalDirectory().c_str()));
 573   xmlNewChild(node, NULL, xmlCharStrdup("result_directory"), xmlCharStrdup(getResultDirectory().c_str()));
 574
 575   // Files
 576   xmlNodePtr files_node = xmlNewChild(node, NULL, xmlCharStrdup("files"), NULL);
 577   std::list<std::string> in_files  = get_in_files();
 578   std::list<std::string> out_files = get_out_files();
 579   for(std::list<std::string>::iterator it = in_files.begin(); it != in_files.end(); it++)
 580     xmlNewChild(files_node, NULL, xmlCharStrdup("in_file"), xmlCharStrdup((*it).c_str()));
 581   for(std::list<std::string>::iterator it = out_files.begin(); it != out_files.end(); it++)
 582     xmlNewChild(files_node, NULL, xmlCharStrdup("out_file"), xmlCharStrdup((*it).c_str()));
 583
 584   // Resource part
 585   resourceParams resource_params = getResourceRequiredParams();
 586   xmlNodePtr res_node = xmlNewChild(node, NULL, xmlCharStrdup("resource_params"), NULL);
 587   xmlNewChild(res_node, NULL, xmlCharStrdup("name"),   xmlCharStrdup(resource_params.name.c_str()));
 588   xmlNewChild(res_node, NULL, xmlCharStrdup("hostname"),   xmlCharStrdup(resource_params.hostname.c_str()));
 589   xmlNewChild(res_node, NULL, xmlCharStrdup("OS"),   xmlCharStrdup(resource_params.OS.c_str()));
 590   std::ostringstream nb_proc_stream;
 591   std::ostringstream nb_node_stream;
 592   std::ostringstream nb_proc_per_node_stream;
 593   std::ostringstream cpu_clock_stream;
 594   std::ostringstream mem_mb_stream;
 595   nb_proc_stream << resource_params.nb_proc;
 596   nb_node_stream << resource_params.nb_node;
 597   nb_proc_per_node_stream << resource_params.nb_proc_per_node;
 598   cpu_clock_stream << resource_params.cpu_clock;
 599   mem_mb_stream << resource_params.mem_mb;
 600   xmlNewChild(res_node, NULL, xmlCharStrdup("nb_proc"),            xmlCharStrdup(nb_proc_stream.str().c_str()));
 601   xmlNewChild(res_node, NULL, xmlCharStrdup("nb_node"),            xmlCharStrdup(nb_node_stream.str().c_str()));
 602   xmlNewChild(res_node, NULL, xmlCharStrdup("nb_proc_per_node"),   xmlCharStrdup(nb_proc_per_node_stream.str().c_str()));
 603   xmlNewChild(res_node, NULL, xmlCharStrdup("cpu_clock"),          xmlCharStrdup(cpu_clock_stream.str().c_str()));
 604   xmlNewChild(res_node, NULL, xmlCharStrdup("mem_mb"),             xmlCharStrdup(mem_mb_stream.str().c_str()));
 605
 606   xmlNewChild(node, NULL, xmlCharStrdup("maximum_duration"), xmlCharStrdup(getMaximumDuration().c_str()));
 607   xmlNewChild(node, NULL, xmlCharStrdup("queue"),            xmlCharStrdup(getQueue().c_str()));
 608
 609   // Specific parameters part
 610   xmlNodePtr specific_parameters_node = xmlNewChild(node, NULL, xmlCharStrdup("specific_parameters"), NULL);
 611   std::map<std::string, std::string> specific_parameters = getSpecificParameters();
 612   for(std::map<std::string, std::string>::iterator it = specific_parameters.begin(); it != specific_parameters.end(); it++)
 613   {
 614     xmlNodePtr specific_parameter_node = xmlNewChild(specific_parameters_node, NULL, xmlCharStrdup("specific_parameter"), NULL);
 615     xmlNewChild(specific_parameter_node, NULL, xmlCharStrdup("name"), xmlCharStrdup((it->first).c_str()));
 616     xmlNewChild(specific_parameter_node, NULL, xmlCharStrdup("value"), xmlCharStrdup((it->second).c_str()));
 617   }
 618
 619   // Run part
 620   xmlNodePtr run_node = xmlNewChild(job_node, NULL, xmlCharStrdup("run_part"), NULL);
 621   xmlNewChild(run_node, NULL, xmlCharStrdup("job_state"), xmlCharStrdup(getState().c_str()));
 622   ParserResourcesType resource_definition = getResourceDefinition();
 623   xmlNewChild(run_node, NULL, xmlCharStrdup("resource_choosed_name"), xmlCharStrdup(resource_definition.Name.c_str()));
 624
 625 #ifdef WITH_LIBBATCH
 626   Batch::JobId job_id = getBatchManagerJobId();
 627   xmlNewChild(run_node, NULL, xmlCharStrdup("job_reference"), xmlCharStrdup(job_id.getReference().c_str()));
 628 #endif
 629 }
 630
 631 void
 632 Launcher::Job::addSpecificParameter(const std::string & name,
 633                                       const std::string & value)
 634 {
 635   _specific_parameters[name] = value;
 636 }
 637
 638 const std::map<std::string, std::string> &
 639 Launcher::Job::getSpecificParameters()
 640 {
 641   return _specific_parameters;
 642 }
 643
 644 void
 645 Launcher::Job::checkSpecificParameters()
 646 {
 647 }