src/Launcher/Launcher_Job.cxx

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