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