src/Launcher/Launcher.cxx

   1 // Copyright (C) 2007-2016  CEA/DEN, EDF R&D, OPEN CASCADE
   2 //
   3 // Copyright (C) 2003-2007  OPEN CASCADE, EADS/CCR, LIP6, CEA/DEN,
   4 // CEDRAT, EDF R&D, LEG, PRINCIPIA R&D, BUREAU VERITAS
   5 //
   6 // This library is free software; you can redistribute it and/or
   7 // modify it under the terms of the GNU Lesser General Public
   8 // License as published by the Free Software Foundation; either
   9 // version 2.1 of the License, or (at your option) any later version.
  10 //
  11 // This library is distributed in the hope that it will be useful,
  12 // but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14 // Lesser General Public License for more details.
  15 //
  16 // You should have received a copy of the GNU Lesser General Public
  17 // License along with this library; if not, write to the Free Software
  18 // Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  19 //
  20 // See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
  21 //
  22
  23 #include <list>
  24 #include <iostream>
  25 #include <sstream>
  26 #include <sys/stat.h>
  27 #include <time.h>
  28
  29 #ifdef WITH_LIBBATCH
  30 #include <libbatch/BatchManagerCatalog.hxx>
  31 #include <libbatch/FactBatchManager.hxx>
  32 #include <libbatch/BatchManager.hxx>
  33 #endif
  34
  35 #include "Basics_Utils.hxx"
  36 #include "Basics_DirUtils.hxx"
  37 #include "SALOME_Launcher_Handler.hxx"
  38 #include "Launcher.hxx"
  39 #include "Launcher_Job_Command.hxx"
  40 #include "Launcher_XML_Persistence.hxx"
  41
  42 using namespace std;
  43
  44 //=============================================================================
  45 /*!
  46  *  Constructor
  47  *  \param orb
  48  *  Define a CORBA single thread policy for the server, which avoid to deal
  49  *  with non thread-safe usage like Change_Directory in SALOME naming service
  50  */
  51 //=============================================================================
  52 Launcher_cpp::Launcher_cpp()
  53 {
  54   LAUNCHER_MESSAGE("Launcher_cpp constructor");
  55   _job_cpt = 0;
  56   _job_cpt_mutex = new pthread_mutex_t();
  57   pthread_mutex_init(_job_cpt_mutex, NULL);
  58 }
  59
  60 //=============================================================================
  61 /*!
  62  * destructor
  63  */
  64 //=============================================================================
  65 Launcher_cpp::~Launcher_cpp()
  66 {
  67   LAUNCHER_MESSAGE("Launcher_cpp destructor");
  68 #ifdef WITH_LIBBATCH
  69   std::map<int, Launcher::Job *>::const_iterator it_job;
  70   for(it_job = _launcher_job_map.begin(); it_job != _launcher_job_map.end(); it_job++)
  71     delete it_job->second;
  72   std::map <int, Batch::BatchManager * >::const_iterator it1;
  73   for(it1=_batchmap.begin();it1!=_batchmap.end();it1++)
  74     delete it1->second;
  75 #endif
  76
  77   pthread_mutex_destroy(_job_cpt_mutex);
  78   delete _job_cpt_mutex;
  79 }
  80
  81 #ifdef WITH_LIBBATCH
  82
  83 //=============================================================================
  84 /*!
  85  * Add a job into the launcher - check resource and choose one
  86  */
  87 //=============================================================================
  88 void
  89 Launcher_cpp::createJob(Launcher::Job * new_job)
  90 {
  91   LAUNCHER_MESSAGE("Creating a new job");
  92   // Add job to the jobs map
  93   pthread_mutex_lock(_job_cpt_mutex);
  94   new_job->setNumber(_job_cpt);
  95   _job_cpt++;
  96   pthread_mutex_unlock(_job_cpt_mutex);
  97   std::map<int, Launcher::Job *>::const_iterator it_job = _launcher_job_map.find(new_job->getNumber());
  98   if (it_job == _launcher_job_map.end())
  99   {
 100     _launcher_job_map[new_job->getNumber()] = new_job;
 101   }
 102   else
 103   {
 104     LAUNCHER_INFOS("A job as already the same id: " << new_job->getNumber());
 105     delete new_job;
 106     throw LauncherException("A job as already the same id - job is not created !");
 107   }
 108   LAUNCHER_MESSAGE("New Job created");
 109 }
 110
 111 //=============================================================================
 112 /*!
 113  * Launch a job
 114  */
 115 //=============================================================================
 116 void
 117 Launcher_cpp::launchJob(int job_id)
 118 {
 119   LAUNCHER_MESSAGE("Launch a job");
 120
 121   // Check if job exist
 122   Launcher::Job * job = findJob(job_id);
 123
 124   // Check job state (cannot launch a job already launched...)
 125   if (job->getState() != "CREATED")
 126   {
 127     LAUNCHER_INFOS("Bad state of the job: " << job->getState());
 128     throw LauncherException("Bad state of the job: " + job->getState());
 129   }
 130
 131   // Third step search batch manager for the job into the map -> instantiate one if does not exist
 132   std::map<int, Batch::BatchManager *>::const_iterator it = _batchmap.find(job_id);
 133   if(it == _batchmap.end())
 134   {
 135     createBatchManagerForJob(job);
 136   }
 137
 138   try {
 139     Batch::JobId batch_manager_job_id = _batchmap[job_id]->submitJob(*(job->getBatchJob()));
 140     job->setBatchManagerJobId(batch_manager_job_id);
 141     job->setState("QUEUED");
 142     job->setReference(batch_manager_job_id.getReference());
 143   }
 144   catch(const Batch::GenericException &ex)
 145   {
 146     LAUNCHER_INFOS("Job is not launched, exception in submitJob: " << ex.message);
 147     throw LauncherException(ex.message.c_str());
 148   }
 149   LAUNCHER_MESSAGE("Job launched");
 150 }
 151
 152 //=============================================================================
 153 /*!
 154  * Get job state
 155  */
 156 //=============================================================================
 157 const char *
 158 Launcher_cpp::getJobState(int job_id)
 159 {
 160   LAUNCHER_MESSAGE("Get job state");
 161
 162   // Check if job exist
 163   Launcher::Job * job = findJob(job_id);
 164
 165   std::string state;
 166   try
 167   {
 168     state = job->updateJobState();
 169   }
 170   catch(const Batch::GenericException &ex)
 171   {
 172     LAUNCHER_INFOS("getJobState failed, exception: " << ex.message);
 173     throw LauncherException(ex.message.c_str());
 174   }
 175
 176   return state.c_str();
 177 }
 178
 179 //=============================================================================
 180 /*!
 181  * Get job assigned hostnames
 182  */
 183 //=============================================================================
 184 const char *
 185 Launcher_cpp::getAssignedHostnames(int job_id)
 186 {
 187   LAUNCHER_MESSAGE("Get job assigned hostnames");
 188
 189   // Check if job exist
 190   Launcher::Job * job = findJob(job_id);
 191   std::string assigned_hostnames = job->getAssignedHostnames();
 192
 193   return assigned_hostnames.c_str();
 194 }
 195
 196 //=============================================================================
 197 /*!
 198  * Get Job result - the result directory could be changed
 199  */
 200 //=============================================================================
 201 void
 202 Launcher_cpp::getJobResults(int job_id, std::string directory)
 203 {
 204   LAUNCHER_MESSAGE("Get Job results");
 205
 206   Launcher::Job * job = findJob(job_id);
 207   std::string resource_name = job->getResourceDefinition().Name;
 208   try
 209   {
 210     if (directory != "")
 211       _batchmap[job_id]->importOutputFiles(*(job->getBatchJob()), directory);
 212     else
 213       _batchmap[job_id]->importOutputFiles(*(job->getBatchJob()), job->getResultDirectory());
 214   }
 215   catch(const Batch::GenericException &ex)
 216   {
 217     LAUNCHER_INFOS("getJobResult is maybe incomplete, exception: " << ex.message);
 218     throw LauncherException(ex.message.c_str());
 219   }
 220   LAUNCHER_MESSAGE("getJobResult ended");
 221 }
 222
 223 //=============================================================================
 224 /*!
 225  * Clear the remote working directory
 226  */
 227 //=============================================================================
 228 void
 229 Launcher_cpp::clearJobWorkingDir(int job_id)
 230 {
 231   LAUNCHER_MESSAGE("Clear the remote working directory");
 232
 233   Launcher::Job * job = findJob(job_id);
 234   try
 235   {
 236     _batchmap[job_id]->clearWorkingDir(*(job->getBatchJob()));
 237   }
 238   catch(const Batch::GenericException &ex)
 239   {
 240     LAUNCHER_INFOS("getJobResult is maybe incomplete, exception: " << ex.message);
 241     throw LauncherException(ex.message.c_str());
 242   }
 243   LAUNCHER_MESSAGE("getJobResult ended");
 244 }
 245
 246 //=============================================================================
 247 /*!
 248  * Get Job dump state - the result directory could be changed
 249  */
 250 //=============================================================================
 251 bool
 252 Launcher_cpp::getJobDumpState(int job_id, std::string directory)
 253 {
 254   bool rtn;
 255   LAUNCHER_MESSAGE("Get Job dump state");
 256
 257   Launcher::Job * job = findJob(job_id);
 258   std::string resource_name = job->getResourceDefinition().Name;
 259   try
 260   {
 261     if (directory != "")
 262       rtn = _batchmap[job_id]->importDumpStateFile(*(job->getBatchJob()), directory);
 263     else
 264       rtn = _batchmap[job_id]->importDumpStateFile(*(job->getBatchJob()), job->getResultDirectory());
 265   }
 266   catch(const Batch::GenericException &ex)
 267   {
 268     LAUNCHER_INFOS("getJobResult is maybe incomplete, exception: " << ex.message);
 269     throw LauncherException(ex.message.c_str());
 270   }
 271   LAUNCHER_MESSAGE("getJobResult ended");
 272   return rtn;
 273 }
 274
 275 //=============================================================================
 276 /*!
 277  * Get one file from the working directory - the result directory can be changed
 278  */
 279 //=============================================================================
 280 bool
 281 Launcher_cpp::getJobWorkFile(int job_id,
 282                              std::string work_file,
 283                              std::string directory)
 284 {
 285   bool rtn;
 286   LAUNCHER_MESSAGE("Get working file " << work_file);
 287
 288   Launcher::Job * job = findJob(job_id);
 289   std::string resource_name = job->getResourceDefinition().Name;
 290   try
 291   {
 292     if (directory != "")
 293       rtn = _batchmap[job_id]->importWorkFile(*(job->getBatchJob()), work_file, directory);
 294     else
 295       rtn = _batchmap[job_id]->importWorkFile(*(job->getBatchJob()), work_file, job->getResultDirectory());
 296   }
 297   catch(const Batch::GenericException &ex)
 298   {
 299     LAUNCHER_INFOS("getJobWorkFile is maybe incomplete, exception: " << ex.message);
 300     throw LauncherException(ex.message.c_str());
 301   }
 302   LAUNCHER_MESSAGE("getJobWorkFile ended");
 303   return rtn;
 304 }
 305
 306 //=============================================================================
 307 /*!
 308  * Remove the job - into the Launcher and its batch manager
 309  */
 310 //=============================================================================
 311 void
 312 Launcher_cpp::removeJob(int job_id)
 313 {
 314   LAUNCHER_MESSAGE("Remove Job");
 315
 316   // Check if job exist
 317   std::map<int, Launcher::Job *>::iterator it_job = _launcher_job_map.find(job_id);
 318   if (it_job == _launcher_job_map.end())
 319   {
 320     LAUNCHER_INFOS("Cannot find the job, is it created ? job number: " << job_id);
 321     throw LauncherException("Cannot find the job, is it created ?");
 322   }
 323
 324   it_job->second->removeJob();
 325   delete it_job->second;
 326   _launcher_job_map.erase(it_job);
 327 }
 328
 329 //=============================================================================
 330 /*!
 331  * stop the job
 332  */
 333 //=============================================================================
 334 void
 335 Launcher_cpp::stopJob(int job_id)
 336 {
 337   LAUNCHER_MESSAGE("Stop Job");
 338
 339   Launcher::Job * job = findJob(job_id);
 340   job->stopJob();
 341 }
 342
 343 std::string
 344 Launcher_cpp::dumpJob(int job_id)
 345 {
 346   LAUNCHER_MESSAGE("dump Job");
 347
 348   Launcher::Job * job = findJob(job_id);
 349   return Launcher::XML_Persistence::dumpJob(*job);
 350 }
 351
 352 int
 353 Launcher_cpp::restoreJob(const std::string& dumpedJob)
 354 {
 355   LAUNCHER_MESSAGE("restore Job");
 356   Launcher::Job * new_job=NULL;
 357   int jobId = -1;
 358   try
 359   {
 360     new_job = Launcher::XML_Persistence::createJobFromString(dumpedJob);
 361     if(new_job)
 362     {
 363       jobId = addJob(new_job);
 364       if(jobId < 0)
 365         delete new_job;
 366     }
 367   }
 368   catch(const LauncherException &ex)
 369   {
 370     LAUNCHER_INFOS("Cannot load the job. Exception: " << ex.msg.c_str());
 371     if(new_job)
 372       delete new_job;
 373   }
 374   return jobId;
 375 }
 376
 377 //=============================================================================
 378 /*!
 379  *  create a launcher job based on a file
 380  *  \param xmlExecuteFile     : to define the execution on the batch cluster
 381  */
 382 //=============================================================================
 383 long
 384 Launcher_cpp::createJobWithFile(const std::string xmlExecuteFile,
 385                                 const std::string clusterName)
 386 {
 387   LAUNCHER_MESSAGE("Begin of Launcher_cpp::createJobWithFile");
 388
 389   // Parsing xml file
 390   ParserLauncherType job_params = ParseXmlFile(xmlExecuteFile);
 391
 392   // Creating a new job
 393   Launcher::Job_Command * new_job = new Launcher::Job_Command();
 394
 395   std::string cmdFile = Kernel_Utils::GetTmpFileName();
 396 #ifndef WIN32
 397   cmdFile += ".sh";
 398 #else
 399   cmdFile += ".bat";
 400 #endif
 401   std::ofstream os;
 402   os.open(cmdFile.c_str(), std::ofstream::out );
 403   os << "#! /bin/sh" << std::endl;
 404   os << job_params.Command;
 405   os.close();
 406
 407   new_job->setJobFile(cmdFile);
 408   new_job->setLocalDirectory(job_params.RefDirectory);
 409   new_job->setWorkDirectory(job_params.MachinesList[clusterName].WorkDirectory);
 410   new_job->setEnvFile(job_params.MachinesList[clusterName].EnvFile);
 411
 412   for(int i=0; i < job_params.InputFile.size(); i++)
 413     new_job->add_in_file(job_params.InputFile[i]);
 414   for(int i=0; i < job_params.OutputFile.size();i++)
 415     new_job->add_out_file(job_params.OutputFile[i]);
 416
 417   resourceParams p;
 418   p.hostname = clusterName;
 419   p.name = "";
 420   p.OS = "";
 421   p.nb_proc = job_params.NbOfProcesses;
 422   p.nb_node = 0;
 423   p.nb_proc_per_node = 0;
 424   p.cpu_clock = 0;
 425   p.mem_mb = 0;
 426   new_job->setResourceRequiredParams(p);
 427
 428   createJob(new_job);
 429   return new_job->getNumber();
 430 }
 431
 432 //=============================================================================
 433 /*!
 434  *  Factory to instantiate the good batch manager for chosen cluster.
 435  */
 436 //=============================================================================
 437 Batch::BatchManager *
 438 Launcher_cpp::FactoryBatchManager(ParserResourcesType& params)
 439 {
 440   std::string mpi;
 441   Batch::CommunicationProtocolType protocol;
 442   Batch::FactBatchManager * fact;
 443
 444   std::string hostname = params.HostName;
 445
 446   switch(params.Protocol)
 447   {
 448     case sh:
 449       protocol = Batch::SH;
 450       break;
 451     case rsh:
 452       protocol = Batch::RSH;
 453       break;
 454     case ssh:
 455       protocol = Batch::SSH;
 456       break;
 457     case rsync:
 458       protocol = Batch::RSYNC;
 459       break;
 460     default:
 461       throw LauncherException("Unknown protocol for this resource");
 462       break;
 463   }
 464
 465   switch(params.mpi)
 466   {
 467     case lam:
 468       mpi = "lam";
 469       break;
 470     case mpich1:
 471       mpi = "mpich1";
 472       break;
 473     case mpich2:
 474       mpi = "mpich2";
 475       break;
 476     case openmpi:
 477       mpi = "openmpi";
 478       break;
 479     case ompi:
 480       mpi = "ompi";
 481       break;
 482     case slurmmpi:
 483       mpi = "slurmmpi";
 484       break;
 485     case prun:
 486       mpi = "prun";
 487       break;
 488     default:
 489       mpi = "nompi";
 490   }
 491
 492   const char * bmType;
 493   switch( params.Batch )
 494   {
 495     case pbs:
 496       bmType = "PBS";
 497       break;
 498     case lsf:
 499       bmType = "LSF";
 500       break;
 501     case sge:
 502       bmType = "SGE";
 503       break;
 504     case ccc:
 505       bmType = "CCC";
 506       break;
 507     case slurm:
 508       bmType = "SLURM";
 509       break;
 510     case none:
 511       bmType = "LOCAL";
 512       break;
 513     case ll:
 514       bmType = "LL";
 515       break;
 516     case vishnu:
 517       bmType = "VISHNU";
 518       break;
 519     case oar:
 520       bmType = "OAR";
 521       break;
 522     case coorm:
 523       bmType = "COORM";
 524       break;
 525     default:
 526       LAUNCHER_MESSAGE("Bad batch description of the resource: Batch = " << params.Batch);
 527       throw LauncherException("No batchmanager for that cluster - Bad batch description of the resource");
 528   }
 529   Batch::BatchManagerCatalog & cata = Batch::BatchManagerCatalog::getInstance();
 530   fact = dynamic_cast<Batch::FactBatchManager*>(cata(bmType));
 531   if (fact == NULL) {
 532     LAUNCHER_MESSAGE("Cannot find batch manager factory for " << bmType << ". Check your version of libBatch.");
 533     throw LauncherException("Cannot find batch manager factory");
 534   }
 535   LAUNCHER_MESSAGE("Instantiation of batch manager of type: " << bmType);
 536   Batch::BatchManager * batch_client = (*fact)(hostname.c_str(), params.UserName.c_str(),
 537                                                protocol, mpi.c_str());
 538   return batch_client;
 539 }
 540
 541 //----------------------------------------------------------
 542 // Without LIBBATCH - Launcher_cpp do nothing...
 543 //----------------------------------------------------------
 544 #else
 545
 546 void
 547 Launcher_cpp::createJob(Launcher::Job * new_job)
 548 {
 549   LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot create a job !!!");
 550   delete new_job;
 551   throw LauncherException("Method Launcher_cpp::createJob is not available "
 552                           "(libBatch was not present at compilation time)");
 553 }
 554
 555 void
 556 Launcher_cpp::launchJob(int job_id)
 557 {
 558   LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot launch a job !!!");
 559   throw LauncherException("Method Launcher_cpp::launchJob is not available "
 560                           "(libBatch was not present at compilation time)");
 561 }
 562
 563 const char *
 564 Launcher_cpp::getJobState(int job_id)
 565 {
 566   LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot get job state!!!");
 567   throw LauncherException("Method Launcher_cpp::getJobState is not available "
 568                           "(libBatch was not present at compilation time)");
 569 }
 570
 571 const char *
 572 Launcher_cpp::getAssignedHostnames(int job_id)
 573 {
 574   LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot get job assigned hostnames!!!");
 575   throw LauncherException("Method Launcher_cpp::getAssignedHostnames is not available "
 576                           "(libBatch was not present at compilation time)");
 577 }
 578
 579 void
 580 Launcher_cpp::getJobResults(int job_id, std::string directory)
 581 {
 582   LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot get job results!!!");
 583   throw LauncherException("Method Launcher_cpp::getJobResults is not available "
 584                           "(libBatch was not present at compilation time)");
 585 }
 586
 587 void
 588 Launcher_cpp::clearJobWorkingDir(int job_id)
 589 {
 590   LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot clear directory!!!");
 591   throw LauncherException("Method Launcher_cpp::clearJobWorkingDir is not available "
 592                           "(libBatch was not present at compilation time)");
 593 }
 594
 595 bool
 596 Launcher_cpp::getJobDumpState(int job_id, std::string directory)
 597 {
 598   LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot get job dump state!!!");
 599   throw LauncherException("Method Launcher_cpp::getJobDumpState is not available "
 600                           "(libBatch was not present at compilation time)");
 601 }
 602
 603 bool
 604 Launcher_cpp::getJobWorkFile(int job_id, std::string work_file, std::string directory)
 605 {
 606   LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot get job dump state!!!");
 607   throw LauncherException("Method Launcher_cpp::getJobWorkFile is not available "
 608                           "(libBatch was not present at compilation time)");
 609 }
 610
 611 void
 612 Launcher_cpp::removeJob(int job_id)
 613 {
 614   LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot remove job!!!");
 615   throw LauncherException("Method Launcher_cpp::removeJob is not available "
 616                           "(libBatch was not present at compilation time)");
 617 }
 618
 619 void
 620 Launcher_cpp::stopJob(int job_id)
 621 {
 622   throw LauncherException("Method Launcher_cpp::stopJob is not available "
 623                           "(libBatch was not present at compilation time)");
 624 }
 625
 626 std::string
 627 Launcher_cpp::dumpJob(int job_id)
 628 {
 629   LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot dump job!!!");
 630   throw LauncherException("Method Launcher_cpp::dumpJob is not available "
 631                           "(libBatch was not present at compilation time)");
 632   return "";
 633 }
 634
 635 int
 636 Launcher_cpp::restoreJob(const std::string& dumpedJob)
 637 {
 638   LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot restore job!!!");
 639   throw LauncherException("Method Launcher_cpp::restoreJob is not available "
 640                           "(libBatch was not present at compilation time)");
 641   return 0;
 642 }
 643
 644 long
 645 Launcher_cpp::createJobWithFile( const std::string xmlExecuteFile, std::string clusterName)
 646 {
 647   throw LauncherException("Method Launcher_cpp::createJobWithFile is not available "
 648                           "(libBatch was not present at compilation time)");
 649   return 0;
 650 }
 651
 652 #endif
 653
 654 ParserLauncherType
 655 Launcher_cpp::ParseXmlFile(std::string xmlExecuteFile)
 656 {
 657   ParserLauncherType job_params;
 658   SALOME_Launcher_Handler * handler = new SALOME_Launcher_Handler(job_params);
 659
 660   const char* aFilePath = xmlExecuteFile.c_str();
 661   FILE* aFile = fopen(aFilePath, "r");
 662   if (aFile != NULL)
 663   {
 664     xmlDocPtr aDoc = xmlReadFile(aFilePath, NULL, 0);
 665     if (aDoc != NULL)
 666       handler->ProcessXmlDocument(aDoc);
 667     else
 668     {
 669       std::string message = "ResourcesManager_cpp: could not parse file: " + xmlExecuteFile;
 670       LAUNCHER_MESSAGE(message);
 671       delete handler;
 672       throw LauncherException(message);
 673     }
 674     // Free the document
 675     xmlFreeDoc(aDoc);
 676     fclose(aFile);
 677   }
 678   else
 679   {
 680     std::string message = "ResourcesManager_cpp: file is not readable: " + xmlExecuteFile;
 681     LAUNCHER_MESSAGE(message);
 682     delete handler;
 683     throw LauncherException(message);
 684   }
 685
 686   // Return
 687   delete handler;
 688   return job_params;
 689 }
 690
 691 std::map<int, Launcher::Job *>
 692 Launcher_cpp::getJobs()
 693 {
 694   return _launcher_job_map;
 695 }
 696
 697 void
 698 Launcher_cpp::createBatchManagerForJob(Launcher::Job * job)
 699 {
 700   int job_id = job->getNumber();
 701
 702   // Select a resource for the job
 703   std::vector<std::string> ResourceList;
 704   resourceParams params = job->getResourceRequiredParams();
 705   // Consider only resources that can launch batch jobs
 706   params.can_launch_batch_jobs = true;
 707   try
 708   {
 709     ResourceList = _ResManager->GetFittingResources(params);
 710   }
 711   catch(const ResourcesException &ex)
 712   {
 713     throw LauncherException(ex.msg.c_str());
 714   }
 715   if (ResourceList.size() == 0)
 716   {
 717     LAUNCHER_INFOS("No adequate resource found for the job, number " << job->getNumber());
 718     job->setState("ERROR");
 719     throw LauncherException("No resource found the job");
 720   }
 721
 722   // Configure the job with the resource selected - the first of the list
 723   ParserResourcesType resource_definition = _ResManager->GetResourcesDescr(ResourceList[0]);
 724
 725   // Set resource definition to the job
 726   // The job will check if the definitions needed
 727   try
 728   {
 729     job->setResourceDefinition(resource_definition);
 730   }
 731   catch(const LauncherException &ex)
 732   {
 733     LAUNCHER_INFOS("Error in the definition of the resource, mess: " << ex.msg);
 734     job->setState("ERROR");
 735     throw ex;
 736   }
 737
 738   // Step 2: We can now add a Factory if the resource is correctly define
 739 #ifdef WITH_LIBBATCH
 740   std::map<int, Batch::BatchManager *>::const_iterator it = _batchmap.find(job_id);
 741   if(it == _batchmap.end())
 742   {
 743     try
 744     {
 745       // Warning cannot write on one line like this, because map object is constructed before
 746       // the method is called...
 747       //_batchmap[job_id] = FactoryBatchManager(resource_definition);
 748       Batch::BatchManager * batch_client = FactoryBatchManager(resource_definition);
 749       _batchmap[job_id] = batch_client;
 750     }
 751     catch(const LauncherException &ex)
 752     {
 753       LAUNCHER_INFOS("Error during creation of the batch manager of the job, mess: " << ex.msg);
 754       throw ex;
 755     }
 756     catch(const Batch::GenericException &ex)
 757     {
 758       LAUNCHER_INFOS("Error during creation of the batch manager of the job, mess: " << ex.message);
 759       throw LauncherException(ex.message);
 760     }
 761   }
 762 #endif
 763 }
 764
 765 void
 766 Launcher_cpp::addJobDirectlyToMap(Launcher::Job * new_job)
 767 {
 768   // Step 0: Calculated job_id
 769   pthread_mutex_lock(_job_cpt_mutex);
 770   int job_id = _job_cpt;
 771   _job_cpt++;
 772   new_job->setNumber(job_id);
 773   pthread_mutex_unlock(_job_cpt_mutex);
 774
 775   // Step 1: check if resource is already in the map
 776   createBatchManagerForJob(new_job);
 777
 778   // Step 2: add the job to the batch manager
 779 #ifdef WITH_LIBBATCH
 780   try
 781   {
 782     Batch::JobId batch_manager_job_id = _batchmap[job_id]->addJob(*(new_job->getBatchJob()),
 783                                                                   new_job->getReference());
 784     new_job->setBatchManagerJobId(batch_manager_job_id);
 785   }
 786   catch(const Batch::GenericException &ex)
 787   {
 788     LAUNCHER_INFOS("Job cannot be added, exception in addJob: " << ex.message);
 789     throw LauncherException(ex.message.c_str());
 790   }
 791
 792   // Step 3: add job to launcher map
 793   std::map<int, Launcher::Job *>::const_iterator it_job = _launcher_job_map.find(new_job->getNumber());
 794   if (it_job == _launcher_job_map.end())
 795   {
 796     _launcher_job_map[new_job->getNumber()] = new_job;
 797   }
 798   else
 799   {
 800     LAUNCHER_INFOS("A job as already the same id: " << new_job->getNumber());
 801     delete new_job;
 802     throw LauncherException("A job as already the same id - job is not created !");
 803   }
 804   LAUNCHER_MESSAGE("New job added");
 805 #endif
 806 }
 807
 808 int
 809 Launcher_cpp::addJob(Launcher::Job * new_job)
 810 {
 811   string job_state = new_job->getState();
 812   int jobId = -1;
 813   if (job_state == "CREATED")
 814   {
 815     // In this case, we ignore run_part information
 816     createJob(new_job);
 817     jobId = new_job->getNumber();
 818   }
 819   else if (job_state == "QUEUED"     ||
 820             job_state == "RUNNING"    ||
 821             job_state == "IN_PROCESS" ||
 822             job_state == "PAUSED")
 823   {
 824     addJobDirectlyToMap(new_job);
 825     jobId = new_job->getNumber();
 826
 827     // We check that the BatchManager could resume the job
 828 #ifdef WITH_LIBBATCH
 829     if (new_job->getBatchManagerJobId().getReference() != new_job->getReference())
 830     {
 831       LAUNCHER_INFOS("BatchManager type cannot resume a job - job state is set to ERROR");
 832       new_job->setState("ERROR");
 833     }
 834 #endif
 835   }
 836   else if (job_state == "FINISHED" ||
 837             job_state == "FAILED"   ||
 838             job_state == "ERROR")
 839   {
 840     // We add run_part information
 841     addJobDirectlyToMap(new_job);
 842     jobId = new_job->getNumber();
 843   }
 844   else
 845   {
 846     LAUNCHER_INFOS("A bad job is found, state unknown " << job_state);
 847     jobId = -1;
 848   }
 849   return jobId;
 850 }
 851
 852 list<int>
 853 Launcher_cpp::loadJobs(const char* jobs_file)
 854 {
 855   list<int> new_jobs_id_list;
 856
 857   // Load the jobs from XML file
 858   list<Launcher::Job *> jobs_list = Launcher::XML_Persistence::loadJobs(jobs_file);
 859
 860   // Create each job in the launcher
 861   list<Launcher::Job *>::const_iterator it_job;
 862   for (it_job = jobs_list.begin(); it_job != jobs_list.end(); it_job++)
 863   {
 864     Launcher::Job * new_job = *it_job;
 865     int jobId = -1;
 866     try
 867     {
 868       jobId = addJob(new_job);
 869       if(jobId >= 0)
 870         new_jobs_id_list.push_back(jobId);
 871       else
 872         delete new_job;
 873     }
 874     catch(const LauncherException &ex)
 875     {
 876       LAUNCHER_INFOS("Cannot load the job. Exception: " << ex.msg.c_str());
 877       delete new_job;
 878     }
 879   }
 880
 881   return new_jobs_id_list;
 882 }
 883
 884 void
 885 Launcher_cpp::saveJobs(const char* jobs_file)
 886 {
 887   // Create a sorted list from the internal job map
 888   list<const Launcher::Job *> jobs_list;
 889   for (int i=0; i<_job_cpt; i++)
 890   {
 891     map<int, Launcher::Job *>::const_iterator it_job = _launcher_job_map.find(i);
 892     if (it_job != _launcher_job_map.end())
 893       jobs_list.push_back(it_job->second);
 894   }
 895
 896   // Save the jobs in XML file
 897   Launcher::XML_Persistence::saveJobs(jobs_file, jobs_list);
 898 }
 899
 900 Launcher::Job *
 901 Launcher_cpp::findJob(int job_id)
 902 {
 903   std::map<int, Launcher::Job *>::const_iterator it_job = _launcher_job_map.find(job_id);
 904   if (it_job == _launcher_job_map.end())
 905   {
 906     LAUNCHER_INFOS("Cannot find the job, is it created ? job number: " << job_id);
 907     throw LauncherException("Cannot find the job, is it created ?");
 908   }
 909   Launcher::Job * job = it_job->second;
 910   return job;
 911 }