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