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