-// Copyright (C) 2007-2008 CEA/DEN, EDF R&D, OPEN CASCADE
+// Copyright (C) 2007-2021 CEA/DEN, EDF R&D, OPEN CASCADE
//
-// Copyright (C) 2003-2007 OPEN CASCADE, EADS/CCR, LIP6, CEA/DEN,
-// CEDRAT, EDF R&D, LEG, PRINCIPIA R&D, BUREAU VERITAS
+// Copyright (C) 2003-2007 OPEN CASCADE, EADS/CCR, LIP6, CEA/DEN,
+// CEDRAT, EDF R&D, LEG, PRINCIPIA R&D, BUREAU VERITAS
//
-// This library is free software; you can redistribute it and/or
-// modify it under the terms of the GNU Lesser General Public
-// License as published by the Free Software Foundation; either
-// version 2.1 of the License.
+// This library is free software; you can redistribute it and/or
+// modify it under the terms of the GNU Lesser General Public
+// License as published by the Free Software Foundation; either
+// version 2.1 of the License, or (at your option) any later version.
//
-// This library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
-// Lesser General Public License for more details.
+// This library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+// Lesser General Public License for more details.
//
-// You should have received a copy of the GNU Lesser General Public
-// License along with this library; if not, write to the Free Software
-// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+// You should have received a copy of the GNU Lesser General Public
+// License along with this library; if not, write to the Free Software
+// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
//
-// See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
+// See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
//
-#include "Launcher.hxx"
-#include "Batch_Date.hxx"
-#include "Batch_FactBatchManager_eLSF.hxx"
-#include "Batch_FactBatchManager_ePBS.hxx"
-#include "Batch_BatchManager_eClient.hxx"
-#include "Batch_FactBatchManager_eSGE.hxx"
-#include "SALOME_Launcher_Handler.hxx"
-#include "Launcher.hxx"
+#include <list>
#include <iostream>
#include <sstream>
#include <sys/stat.h>
#include <time.h>
+#include <memory>
+
+#ifdef WITH_LIBBATCH
+#include <libbatch/BatchManagerCatalog.hxx>
+#include <libbatch/FactBatchManager.hxx>
+#include <libbatch/BatchManager.hxx>
+#endif
+
+#include "Basics_Utils.hxx"
+#include "Basics_DirUtils.hxx"
+#include "SALOME_Launcher_Handler.hxx"
+#include "Launcher.hxx"
+#include "Launcher_Job_Command.hxx"
+#include "Launcher_Job_YACSFile.hxx"
+#include "Launcher_Job_PythonSALOME.hxx"
+#include "Launcher_Job_CommandSALOME.hxx"
+#include "Launcher_XML_Persistence.hxx"
using namespace std;
//=============================================================================
-/*!
+/*!
* Constructor
* \param orb
* Define a CORBA single thread policy for the server, which avoid to deal
* with non thread-safe usage like Change_Directory in SALOME naming service
*/
//=============================================================================
-
Launcher_cpp::Launcher_cpp()
{
-#if defined(_DEBUG_) || defined(_DEBUG)
- cerr << "Launcher_cpp constructor" << endl;
-#endif
+ LAUNCHER_MESSAGE("Launcher_cpp constructor");
+ _job_cpt = 0;
}
//=============================================================================
-/*!
+/*!
* destructor
*/
//=============================================================================
-
Launcher_cpp::~Launcher_cpp()
{
-#if defined(_DEBUG_) || defined(_DEBUG)
- cerr << "Launcher_cpp destructor" << endl;
-#endif
- std::map < string, Batch::BatchManager_eClient * >::const_iterator it1;
+ LAUNCHER_MESSAGE("Launcher_cpp destructor");
+#ifdef WITH_LIBBATCH
+ std::map<int, Launcher::Job *>::const_iterator it_job;
+ for(it_job = _launcher_job_map.begin(); it_job != _launcher_job_map.end(); it_job++)
+ delete it_job->second;
+ std::map <int, Batch::BatchManager * >::const_iterator it1;
for(it1=_batchmap.begin();it1!=_batchmap.end();it1++)
delete it1->second;
- std::map < std::pair<std::string,long> , Batch::Job* >::const_iterator it2;
- for(it2=_jobmap.begin();it2!=_jobmap.end();it2++)
- delete it2->second;
+#endif
}
+#ifdef WITH_LIBBATCH
+
//=============================================================================
-/*! CORBA Method:
- * Submit a batch job on a cluster and returns the JobId
- * \param xmlExecuteFile : to define the execution on the batch cluster
- * \param clusterName : name of the batch cluster
+/*!
+ * Add a job into the launcher - check resource and choose one
*/
//=============================================================================
-long Launcher_cpp::submitJob( const std::string xmlExecuteFile,
- const std::string clusterName) throw(LauncherException)
+void
+Launcher_cpp::createJob(Launcher::Job * new_job)
{
-#if defined(_DEBUG_) || defined(_DEBUG)
- cout << "BEGIN OF Launcher_cpp::submitJob" << endl;
-#endif
- long jobId;
- vector<string> aMachineList;
-
- if(!_ResManager)
- throw LauncherException("You must set Resources Manager to Launcher!!");
-
- // verify if cluster is in resources catalog
- machineParams params;
- params.hostname = clusterName;
- vector<string> aCompoList ;
- try{
- aMachineList = _ResManager->GetFittingResources(params, aCompoList);
+ LAUNCHER_MESSAGE("Creating a new job");
+ // Add job to the jobs map
+ new_job->setNumber(_job_cpt);
+ _job_cpt++;
+ std::map<int, Launcher::Job *>::const_iterator it_job = _launcher_job_map.find(new_job->getNumber());
+ if (it_job == _launcher_job_map.end())
+ {
+ _launcher_job_map[new_job->getNumber()] = new_job;
}
- catch(const ResourcesException &ex){
- throw LauncherException(ex.msg.c_str());
+ else
+ {
+ LAUNCHER_INFOS("A job has already the same id: " << new_job->getNumber());
+ throw LauncherException("A job has already the same id - job is not created !");
}
- if (aMachineList.size() == 0)
- throw LauncherException("This cluster is not in resources catalog");
+ LAUNCHER_MESSAGE("New Job created");
+}
- // Parsing xml file
- ParseXmlFile(xmlExecuteFile);
+//=============================================================================
+/*!
+ * Add a job into the launcher - check resource and choose one
+ */
+//=============================================================================
+int
+Launcher_cpp::createJob(const JobParameters_cpp& job_parameters)
+{
+ std::string job_type = job_parameters.job_type;
+ Launcher::Job * new_job; // It is Launcher_cpp that is going to destroy it
+
+ if (job_type == Launcher::Job_Command::TYPE_NAME)
+ new_job = new Launcher::Job_Command();
+ else if (job_type == Launcher::Job_CommandSALOME::TYPE_NAME)
+ new_job = new Launcher::Job_CommandSALOME();
+ else if (job_type == Launcher::Job_YACSFile::TYPE_NAME)
+ new_job = new Launcher::Job_YACSFile();
+ else if (job_type == Launcher::Job_PythonSALOME::TYPE_NAME)
+ new_job = new Launcher::Job_PythonSALOME();
+ else
+ {
+ std::string message("Launcher_cpp::createJob: bad job type: ");
+ message += job_type;
+ throw LauncherException(message.c_str());
+ }
- // verify if clustername is in xml file
- map<std::string,MachineParameters>::const_iterator it1 = _launch.MachinesList.find(clusterName);
- if(it1 == _launch.MachinesList.end())
- throw LauncherException("This cluster is not in xml file");
+ // Name
+ new_job->setJobName(job_parameters.job_name);
- ParserResourcesType p = _ResManager->GetResourcesList(aMachineList[0]);
- string cname(p.Alias);
-#if defined(_DEBUG_) || defined(_DEBUG)
- cout << "Choose cluster: " << cname << endl;
-#endif
+ // Directories
+ new_job->setWorkDirectory(job_parameters.work_directory);
+ new_job->setLocalDirectory(job_parameters.local_directory);
+ new_job->setResultDirectory(job_parameters.result_directory);
- // search batch manager for that cluster in map or instanciate one
- map < string, Batch::BatchManager_eClient * >::const_iterator it2 = _batchmap.find(cname);
- if(it2 == _batchmap.end())
- {
- _batchmap[cname] = FactoryBatchManager(p);
- // TODO: Add a test for the cluster !
- }
-
- try{
-
- // directory on cluster to put files to execute
- string remotedir = _launch.MachinesList[clusterName].WorkDirectory;
- // local directory to get files to execute and to put results
- string localdir = _launch.RefDirectory;
-
- int idx1 = xmlExecuteFile.find_last_of("/");
- if(idx1 == string::npos) idx1 = -1;
- int idx2 = xmlExecuteFile.find(".xml");
- string logfile = xmlExecuteFile.substr(idx1+1,idx2-idx1-1);
- string ologfile = logfile + ".output.log";
- string elogfile = logfile + ".error.log";
-
- // create and submit job on cluster
- Batch::Parametre param;
- param[USER] = p.UserName;
- param[EXECUTABLE] = "";
- for(int i=0; i<_launch.InputFile.size();i++)
- param[INFILE] += Batch::Couple( localdir + "/" + _launch.InputFile[i], remotedir + "/" + _launch.InputFile[i] );
- for(int i=0; i<_launch.OutputFile.size();i++)
- param[OUTFILE] += Batch::Couple( localdir + "/" + _launch.OutputFile[i], remotedir + "/" + _launch.OutputFile[i] );
- param[OUTFILE] += Batch::Couple( localdir + "/" + ologfile, remotedir + "/" + ologfile );
- param[OUTFILE] += Batch::Couple( localdir + "/" + elogfile, remotedir + "/" + elogfile );
- param[NBPROC] = _launch.NbOfProcesses;
- param[WORKDIR] = remotedir;
- param[TMPDIR] = remotedir;
- param[MAXWALLTIME] = getWallTime("");
- param[MAXRAMSIZE] = getRamSize("");
- param[HOMEDIR] = "";
-
- Batch::Environnement env;
- env["COMMAND"] = _launch.Command;
- env["SOURCEFILE"] = _launch.MachinesList[clusterName].EnvFile;
- env["LOGFILE"] = logfile;
-
- Batch::Job* job = new Batch::Job(param,env);
-
- // submit job on cluster
- Batch::JobId jid = _batchmap[cname]->submitJob(*job);
-
- // get job id in long
- istringstream iss(jid.getReference());
- iss >> jobId;
-
- _jobmap[ pair<string,long>(cname,jobId) ] = job;
- }
- catch(const Batch::EmulationException &ex){
- throw LauncherException(ex.msg.c_str());
- }
+ // Parameters for COORM
+ new_job->setLauncherFile(job_parameters.launcher_file);
+ new_job->setLauncherArgs(job_parameters.launcher_args);
- return jobId;
+ // Job File
+ new_job->setJobFile(job_parameters.job_file);
+ new_job->setPreCommand(job_parameters.pre_command);
+
+ // Files
+ new_job->setEnvFile(job_parameters.env_file);
+ for(const std::string param : job_parameters.in_files)
+ new_job->add_in_file(param);
+ for(const std::string param : job_parameters.out_files)
+ new_job->add_out_file(param);
+
+ new_job->setMaximumDuration(job_parameters.maximum_duration);
+ new_job->setQueue(job_parameters.queue);
+ new_job->setPartition(job_parameters.partition);
+ new_job->setExclusive(job_parameters.exclusive);
+ new_job->setMemPerCpu(job_parameters.mem_per_cpu);
+ new_job->setWCKey(job_parameters.wckey);
+ new_job->setExtraParams(job_parameters.extra_params);
+
+ // Resources requirements
+ new_job->setResourceRequiredParams(job_parameters.resource_required);
+
+ // Adding specific parameters to the job
+ for (const auto& it_specific : job_parameters.specific_parameters)
+ new_job->addSpecificParameter(it_specific.first, it_specific.second);
+
+ new_job->checkSpecificParameters();
+
+ createJob(new_job);
+
+ return new_job->getNumber();
}
//=============================================================================
-/*! CORBA Method:
- * Submit a batch job on a cluster and returns the JobId
- * \param fileToExecute : .py/.exe/.sh/... to execute on the batch cluster
- * \param filesToExport : to export on the batch cluster
- * \param NumberOfProcessors : Number of processors needed on the batch cluster
- * \param params : Constraints for the choice of the batch cluster
+/*!
+ * Launch a job
*/
//=============================================================================
-long Launcher_cpp::submitSalomeJob( const string fileToExecute ,
- const vector<string>& filesToExport ,
- const vector<string>& filesToImport ,
- const batchParams& batch_params,
- const machineParams& params) throw(LauncherException)
-{
-#if defined(_DEBUG_) || defined(_DEBUG)
- cerr << "BEGIN OF Launcher_cpp::submitSalomeJob" << endl;
-#endif
- long jobId;
- vector<string> aMachineList;
+void
+Launcher_cpp::launchJob(int job_id)
+{
+ LAUNCHER_MESSAGE("Launch a job");
- if(!_ResManager)
- throw LauncherException("You must set Resources Manager to Launcher!!");
+ // Check if job exists
+ Launcher::Job * job = findJob(job_id);
+
+ // Check job state (cannot launch a job already launched...)
+ if (job->getState() != "CREATED")
+ {
+ LAUNCHER_INFOS("Bad state of the job: " << job->getState());
+ throw LauncherException("Bad state of the job: " + job->getState());
+ }
- // check batch params
- if ( !check(batch_params) )
- throw LauncherException("Batch parameters are bad (see informations above)");
+ Batch::BatchManager * bm = getBatchManager(job);
- // find a cluster matching the structure params
- vector<string> aCompoList ;
- try{
- aMachineList = _ResManager->GetFittingResources(params, aCompoList);
+ try {
+ Batch::JobId batch_manager_job_id = bm->submitJob(*(job->getBatchJob()));
+ job->setBatchManagerJobId(batch_manager_job_id);
+ job->setState("QUEUED");
+ job->setReference(batch_manager_job_id.getReference());
}
- catch(const ResourcesException &ex){
- throw LauncherException(ex.msg.c_str());
+ catch(const Batch::GenericException &ex)
+ {
+ LAUNCHER_INFOS("Job is not launched, exception in submitJob: " << ex.message);
+ throw LauncherException(ex.message.c_str());
}
- if (aMachineList.size() == 0)
- throw LauncherException("No resources have been found with your parameters");
+ LAUNCHER_MESSAGE("Job launched");
+}
- ParserResourcesType p = _ResManager->GetResourcesList(aMachineList[0]);
- string clustername(p.Alias);
-#if defined(_DEBUG_) || defined(_DEBUG)
- cerr << "Choose cluster: " << clustername << endl;
-#endif
-
- // search batch manager for that cluster in map or instanciate one
- map < string, Batch::BatchManager_eClient * >::const_iterator it = _batchmap.find(clustername);
- if(it == _batchmap.end())
- {
- _batchmap[clustername] = FactoryBatchManager(p);
- // TODO: Add a test for the cluster !
- }
-
- try{
- // tmp directory on cluster to put files to execute
- string tmpdir = getTmpDirForBatchFiles();
-
- // create and submit job on cluster
- Batch::Parametre param;
- param[USER] = p.UserName;
- param[EXECUTABLE] = buildSalomeCouplingScript(fileToExecute,tmpdir,p);
- param[INFILE] = Batch::Couple( fileToExecute, getRemoteFile(tmpdir,fileToExecute) );
- for(int i=0;i<filesToExport.size();i++)
- param[INFILE] += Batch::Couple( filesToExport[i], getRemoteFile(tmpdir,filesToExport[i]) );
-
- ostringstream file_name_output;
- file_name_output << "~/" << tmpdir << "/" << "output.log*";
- ostringstream file_name_error;
- file_name_error << "~/" << tmpdir << "/" << "error.log*";
- ostringstream file_container_log;
- file_container_log << "~/" << tmpdir << "/" << "YACS_Server*";
- param[OUTFILE] = Batch::Couple( "", file_name_output.str());
- param[OUTFILE] += Batch::Couple( "", file_name_error.str());
- param[OUTFILE] += Batch::Couple( "", file_container_log.str());
-
- for(int i=0;i<filesToImport.size();i++)
- param[OUTFILE] += Batch::Couple( "", filesToImport[i] );
-
- param[NBPROC] = batch_params.nb_proc;
- param[WORKDIR] = batch_params.batch_directory;
- param[TMPDIR] = tmpdir;
- param[MAXWALLTIME] = getWallTime(batch_params.expected_during_time);
- param[MAXRAMSIZE] = getRamSize(batch_params.mem);
- param[HOMEDIR] = getHomeDir(p, tmpdir);
- param[QUEUE] = p.batchQueue;
-
- Batch::Environnement env;
-
- Batch::Job* job = new Batch::Job(param,env);
-
- // submit job on cluster
- Batch::JobId jid = _batchmap[clustername]->submitJob(*job);
-
- // get job id in long
- istringstream iss(jid.getReference());
- iss >> jobId;
-
- _jobmap[ pair<string,long>(clustername,jobId) ] = job;
- }
- catch(const Batch::EmulationException &ex){
- throw LauncherException(ex.msg.c_str());
+//=============================================================================
+/*!
+ * Get job state
+ */
+//=============================================================================
+std::string
+Launcher_cpp::getJobState(int job_id)
+{
+ LAUNCHER_MESSAGE("Get job state");
+
+ // Check if job exist
+ Launcher::Job * job = findJob(job_id);
+
+ std::string state;
+ try
+ {
+ state = job->updateJobState();
+ }
+ catch(const Batch::GenericException &ex)
+ {
+ LAUNCHER_INFOS("getJobState failed, exception: " << ex.message);
+ throw LauncherException(ex.message.c_str());
}
- return jobId;
+ return state;
}
//=============================================================================
-/*! CORBA Method:
- * Query a batch job on a cluster and returns the status of job
- * \param jobId : identification of Salome job
- * \param params : Constraints for the choice of the batch cluster
+/*!
+ * Get job assigned hostnames
*/
//=============================================================================
-string Launcher_cpp::queryJob( long id,
- const machineParams& params) throw(LauncherException)
+std::string
+Launcher_cpp::getAssignedHostnames(int job_id)
{
- if(!_ResManager)
- throw LauncherException("You must set Resources Manager to Launcher!!");
-
- // find a cluster matching params structure
- vector<string> aCompoList ;
- vector<string> aMachineList = _ResManager->GetFittingResources( params , aCompoList ) ;
- ParserResourcesType p = _ResManager->GetResourcesList(aMachineList[0]);
- string clustername(p.Alias);
-
- // search batch manager for that cluster in map
- std::map < string, Batch::BatchManager_eClient * >::const_iterator it = _batchmap.find(clustername);
- if(it == _batchmap.end())
- throw LauncherException("no batchmanager for that cluster");
-
- Batch::Parametre par;
- try{
- ostringstream oss;
- oss << id;
- Batch::JobId jobId( _batchmap[clustername], oss.str() );
-
- Batch::JobInfo jinfo = jobId.queryJob();
- par = jinfo.getParametre();
- }
- catch(const Batch::EmulationException &ex){
- throw LauncherException(ex.msg.c_str());
+ LAUNCHER_MESSAGE("Get job assigned hostnames");
+
+ // Check if job exist
+ Launcher::Job * job = findJob(job_id);
+ std::string assigned_hostnames = job->getAssignedHostnames();
+
+ return assigned_hostnames;
+}
+
+void
+Launcher_cpp::exportInputFiles(int job_id)
+{
+ LAUNCHER_MESSAGE("Copy in files");
+ // Check if job exists
+ Launcher::Job * job = findJob(job_id);
+
+ // Check job state (cannot launch a job already launched...)
+ if (job->getState() != "CREATED")
+ {
+ LAUNCHER_INFOS("Bad state of the job: " << job->getState());
+ throw LauncherException("Bad state of the job: " + job->getState());
}
- return par[STATE];
+ Batch::BatchManager * bm = getBatchManager(job);
+
+ try {
+ bm->exportInputFiles(*(job->getBatchJob()));
+ }
+ catch(const Batch::GenericException &ex)
+ {
+ LAUNCHER_INFOS("Failed to copy in files: " << ex.message);
+ throw LauncherException(ex.message.c_str());
+ }
+ LAUNCHER_MESSAGE("Files copied");
}
-string Launcher_cpp::queryJob( long id,
- const std::string clusterName)
+//=============================================================================
+/*!
+ * Get Job result - the result directory could be changed
+ */
+//=============================================================================
+void
+Launcher_cpp::getJobResults(int job_id, std::string directory)
{
- machineParams params;
- params.hostname = clusterName;
- return queryJob(id,params);
+ LAUNCHER_MESSAGE("Get Job results");
+
+ Launcher::Job * job = findJob(job_id);
+ std::string resource_name = job->getResourceDefinition().Name;
+ try
+ {
+ if (directory != "")
+ _batchmap[job_id]->importOutputFiles(*(job->getBatchJob()), directory);
+ else
+ _batchmap[job_id]->importOutputFiles(*(job->getBatchJob()), job->getResultDirectory());
+ }
+ catch(const Batch::GenericException &ex)
+ {
+ LAUNCHER_INFOS("getJobResult is maybe incomplete, exception: " << ex.message);
+ throw LauncherException(ex.message.c_str());
+ }
+ LAUNCHER_MESSAGE("getJobResult ended");
}
//=============================================================================
-/*! CORBA Method:
- * Delete a batch job on a cluster
- * \param jobId : identification of Salome job
- * \param params : Constraints for the choice of the batch cluster
+/*!
+ * Clear the remote working directory
*/
//=============================================================================
-void Launcher_cpp::deleteJob( const long id,
- const machineParams& params) throw(LauncherException)
+void
+Launcher_cpp::clearJobWorkingDir(int job_id)
{
- if(!_ResManager)
- throw LauncherException("You must set Resources Manager to Launcher!!");
-
- // find a cluster matching params structure
- vector<string> aCompoList ;
- vector<string> aMachineList = _ResManager->GetFittingResources( params , aCompoList ) ;
- ParserResourcesType p = _ResManager->GetResourcesList(aMachineList[0]);
- string clustername(p.Alias);
-
- // search batch manager for that cluster in map
- map < string, Batch::BatchManager_eClient * >::const_iterator it = _batchmap.find(clustername);
- if(it == _batchmap.end())
- throw LauncherException("no batchmanager for that cluster");
-
- ostringstream oss;
- oss << id;
- Batch::JobId jobId( _batchmap[clustername], oss.str() );
+ LAUNCHER_MESSAGE("Clear the remote working directory");
- jobId.deleteJob();
+ Launcher::Job * job = findJob(job_id);
+ try
+ {
+ _batchmap[job_id]->clearWorkingDir(*(job->getBatchJob()));
+ }
+ catch(const Batch::GenericException &ex)
+ {
+ LAUNCHER_INFOS("getJobResult is maybe incomplete, exception: " << ex.message);
+ throw LauncherException(ex.message.c_str());
+ }
+ LAUNCHER_MESSAGE("getJobResult ended");
}
-void Launcher_cpp::deleteJob( long id,
- const std::string clusterName)
+//=============================================================================
+/*!
+ * Get Job dump state - the result directory could be changed
+ */
+//=============================================================================
+bool
+Launcher_cpp::getJobDumpState(int job_id, std::string directory)
{
- machineParams params;
- params.hostname = clusterName;
- deleteJob(id,params);
+ bool rtn;
+ LAUNCHER_MESSAGE("Get Job dump state");
+
+ Launcher::Job * job = findJob(job_id);
+ std::string resource_name = job->getResourceDefinition().Name;
+ try
+ {
+ if (directory != "")
+ rtn = _batchmap[job_id]->importDumpStateFile(*(job->getBatchJob()), directory);
+ else
+ rtn = _batchmap[job_id]->importDumpStateFile(*(job->getBatchJob()), job->getResultDirectory());
+ }
+ catch(const Batch::GenericException &ex)
+ {
+ LAUNCHER_INFOS("getJobResult is maybe incomplete, exception: " << ex.message);
+ throw LauncherException(ex.message.c_str());
+ }
+ LAUNCHER_MESSAGE("getJobResult ended");
+ return rtn;
}
//=============================================================================
-/*! CORBA Method:
- * Get result files of job on a cluster
- * \param jobId : identification of Salome job
- * \param params : Constraints for the choice of the batch cluster
+/*!
+ * Get one file from the working directory - the result directory can be changed
*/
//=============================================================================
-void Launcher_cpp::getResultsJob( const string directory,
- const long id,
- const machineParams& params) throw(LauncherException)
+bool
+Launcher_cpp::getJobWorkFile(int job_id,
+ std::string work_file,
+ std::string directory)
{
- if(!_ResManager)
- throw LauncherException("You must set Resources Manager to Launcher!!");
-
- vector<string> aCompoList ;
- vector<string> aMachineList = _ResManager->GetFittingResources( params , aCompoList ) ;
- ParserResourcesType p = _ResManager->GetResourcesList(aMachineList[0]);
- string clustername(p.Alias);
-
- // search batch manager for that cluster in map
- map < string, Batch::BatchManager_eClient * >::const_iterator it = _batchmap.find(clustername);
- if(it == _batchmap.end())
- throw LauncherException("no batchmanager for that cluster");
-
- Batch::Job* job = _jobmap[ pair<string,long>(clustername,id) ];
-
- _batchmap[clustername]->importOutputFiles( *job, directory );
+ bool rtn;
+ LAUNCHER_MESSAGE("Get working file " << work_file);
+
+ Launcher::Job * job = findJob(job_id);
+ std::string resource_name = job->getResourceDefinition().Name;
+ try
+ {
+ if (directory != "")
+ rtn = _batchmap[job_id]->importWorkFile(*(job->getBatchJob()), work_file, directory);
+ else
+ rtn = _batchmap[job_id]->importWorkFile(*(job->getBatchJob()), work_file, job->getResultDirectory());
+ }
+ catch(const Batch::GenericException &ex)
+ {
+ LAUNCHER_INFOS("getJobWorkFile is maybe incomplete, exception: " << ex.message);
+ throw LauncherException(ex.message.c_str());
+ }
+ LAUNCHER_MESSAGE("getJobWorkFile ended");
+ return rtn;
}
-void Launcher_cpp::getResultsJob( const std::string directory,
- long id,
- const std::string clusterName)
+//=============================================================================
+/*!
+ * Remove the job - into the Launcher and its batch manager
+ */
+//=============================================================================
+void
+Launcher_cpp::removeJob(int job_id)
{
- machineParams params;
- params.hostname = clusterName;
- getResultsJob(directory,id,params);
+ LAUNCHER_MESSAGE("Remove Job");
+
+ // Check if job exist
+ std::map<int, Launcher::Job *>::iterator it_job = _launcher_job_map.find(job_id);
+ if (it_job == _launcher_job_map.end())
+ {
+ LAUNCHER_INFOS("Cannot find the job, is it created ? job number: " << job_id);
+ throw LauncherException("Cannot find the job, is it created ?");
+ }
+
+ it_job->second->removeJob();
+ delete it_job->second;
+ _launcher_job_map.erase(it_job);
}
//=============================================================================
/*!
- * Factory to instanciate the good batch manager for choosen cluster.
- */
+ * stop the job
+ */
//=============================================================================
+void
+Launcher_cpp::stopJob(int job_id)
+{
+ LAUNCHER_MESSAGE("Stop Job");
-Batch::BatchManager_eClient *Launcher_cpp::FactoryBatchManager( const ParserResourcesType& params ) throw(LauncherException)
-{
-
- std::string hostname, protocol, mpi;
- Batch::FactBatchManager_eClient* fact;
-
- hostname = params.Alias;
- switch(params.Protocol){
- case rsh:
- protocol = "rsh";
- break;
- case ssh:
- protocol = "ssh";
- break;
- default:
- throw LauncherException("unknown protocol");
- break;
- }
- switch(params.mpi){
- case lam:
- mpi = "lam";
- break;
- case mpich1:
- mpi = "mpich1";
- break;
- case mpich2:
- mpi = "mpich2";
- break;
- case openmpi:
- mpi = "openmpi";
- break;
- case slurm:
- mpi = "slurm";
- break;
- case prun:
- mpi = "prun";
- break;
- case nompi:
- throw LauncherException("you must specified an mpi implementation for batch manager");
- break;
- default:
- throw LauncherException("unknown mpi implementation");
- break;
- }
-#if defined(_DEBUG_) || defined(_DEBUG)
- cerr << "Instanciation of batch manager" << endl;
-#endif
- switch( params.Batch ){
- case pbs:
-#if defined(_DEBUG_) || defined(_DEBUG)
- cerr << "Instantiation of PBS batch manager" << endl;
-#endif
- fact = new Batch::FactBatchManager_ePBS;
- break;
- case lsf:
-#if defined(_DEBUG_) || defined(_DEBUG)
- cerr << "Instantiation of LSF batch manager" << endl;
-#endif
- fact = new Batch::FactBatchManager_eLSF;
- break;
- case sge:
-#if defined(_DEBUG_) || defined(_DEBUG)
- cout << "Instantiation of SGE batch manager" << endl;
-#endif
- fact = new Batch::FactBatchManager_eSGE;
- break;
- default:
-#if defined(_DEBUG_) || defined(_DEBUG)
- cerr << "BATCH = " << params.Batch << endl;
-#endif
- throw LauncherException("no batchmanager for that cluster");
- }
- return (*fact)(hostname.c_str(),protocol.c_str(),mpi.c_str());
-}
-
-string Launcher_cpp::buildSalomeCouplingScript(const string fileToExecute, const string dirForTmpFiles, const ParserResourcesType& params)
-{
-#ifndef WIN32 //TODO: need for porting on Windows
- int idx = dirForTmpFiles.find("Batch/");
- std::string filelogtemp = dirForTmpFiles.substr(idx+6, dirForTmpFiles.length());
-
- string::size_type p1 = fileToExecute.find_last_of("/");
- string::size_type p2 = fileToExecute.find_last_of(".");
- std::string fileNameToExecute = fileToExecute.substr(p1+1,p2-p1-1);
- std::string TmpFileName = "/tmp/runSalome_" + fileNameToExecute + ".sh";
-
- MpiImpl* mpiImpl = FactoryMpiImpl(params.mpi);
-
- ofstream tempOutputFile;
- tempOutputFile.open(TmpFileName.c_str(), ofstream::out );
-
- // Begin
- tempOutputFile << "#! /bin/sh -f" << endl ;
- tempOutputFile << "cd " ;
- tempOutputFile << params.AppliPath << endl ;
- tempOutputFile << "export SALOME_BATCH=1\n";
- tempOutputFile << "export PYTHONPATH=~/" ;
- tempOutputFile << dirForTmpFiles ;
- tempOutputFile << ":$PYTHONPATH" << endl ;
-
- // Adding user script
- std::string script = params.userCommands;
- if (script != "")
- tempOutputFile << script << endl;
- // Test node rank
- tempOutputFile << "if test \"" ;
- tempOutputFile << mpiImpl->rank() ;
- tempOutputFile << "\" = \"0\"; then" << endl ;
-
- // -----------------------------------------------
- // Code for rank 0 : launch runAppli and a container
- // RunAppli
- if(params.ModulesList.size()>0)
- tempOutputFile << " ./runAppli --terminal --modules=" ;
- else
- tempOutputFile << " ./runAppli --terminal ";
- for ( int i = 0 ; i < params.ModulesList.size() ; i++ ) {
- tempOutputFile << params.ModulesList[i] ;
- if ( i != params.ModulesList.size()-1 )
- tempOutputFile << "," ;
- }
- tempOutputFile << " --standalone=registry,study,moduleCatalog --ns-port-log="
- << filelogtemp
- << " &\n";
-
- // Wait NamingService
- tempOutputFile << " current=0\n"
- << " stop=20\n"
- << " while ! test -f " << filelogtemp << "\n"
- << " do\n"
- << " sleep 2\n"
- << " let current=current+1\n"
- << " if [ \"$current\" -eq \"$stop\" ] ; then\n"
- << " echo Error Naming Service failed ! >&2"
- << " exit\n"
- << " fi\n"
- << " done\n"
- << " port=`cat " << filelogtemp << "`\n";
-
- // Wait other containers
- tempOutputFile << " for ((ip=1; ip < ";
- tempOutputFile << mpiImpl->size();
- tempOutputFile << " ; ip++))" << endl;
- tempOutputFile << " do" << endl ;
- tempOutputFile << " arglist=\"$arglist YACS_Server_\"$ip" << endl ;
- tempOutputFile << " done" << endl ;
- tempOutputFile << " sleep 5" << endl ;
- tempOutputFile << " ./runSession waitContainers.py $arglist" << endl ;
-
- // Launch user script
- tempOutputFile << " ./runSession python ~/" << dirForTmpFiles << "/" << fileNameToExecute << ".py" << endl;
-
- // Stop application
- tempOutputFile << " rm " << filelogtemp << "\n"
- << " ./runSession shutdownSalome.py" << endl;
-
- // -------------------------------------
- // Other nodes launch a container
- tempOutputFile << "else" << endl ;
-
- // Wait NamingService
- tempOutputFile << " current=0\n"
- << " stop=20\n"
- << " while ! test -f " << filelogtemp << "\n"
- << " do\n"
- << " sleep 2\n"
- << " let current=current+1\n"
- << " if [ \"$current\" -eq \"$stop\" ] ; then\n"
- << " echo Error Naming Service failed ! >&2"
- << " exit\n"
- << " fi\n"
- << " done\n"
- << " port=`cat " << filelogtemp << "`\n";
-
- // Launching container
- tempOutputFile << " ./runSession SALOME_Container YACS_Server_";
- tempOutputFile << mpiImpl->rank()
- << " > ~/" << dirForTmpFiles << "/YACS_Server_"
- << mpiImpl->rank() << "_container_log." << filelogtemp
- << " 2>&1\n";
- tempOutputFile << "fi" << endl ;
- tempOutputFile.flush();
- tempOutputFile.close();
- chmod(TmpFileName.c_str(), 0x1ED);
-#if defined(_DEBUG_) || defined(_DEBUG)
- cerr << TmpFileName.c_str() << endl;
-#endif
+ Launcher::Job * job = findJob(job_id);
+ job->stopJob();
+}
- delete mpiImpl;
+std::string
+Launcher_cpp::dumpJob(int job_id)
+{
+ LAUNCHER_MESSAGE("dump Job");
- return TmpFileName;
-#else
- return "";
-#endif
-
-}
-
-MpiImpl *Launcher_cpp::FactoryMpiImpl(MpiImplType mpi) throw(LauncherException)
-{
- switch(mpi){
- case lam:
- return new MpiImpl_LAM();
- case mpich1:
- return new MpiImpl_MPICH1();
- case mpich2:
- return new MpiImpl_MPICH2();
- case openmpi:
- return new MpiImpl_OPENMPI();
- case slurm:
- return new MpiImpl_SLURM();
- case prun:
- return new MpiImpl_PRUN();
- case nompi:
- throw LauncherException("you must specify an mpi implementation for batch manager");
- default:
- ostringstream oss;
- oss << mpi << " : not yet implemented";
- throw LauncherException(oss.str().c_str());
- }
-
-}
-
-string Launcher_cpp::getTmpDirForBatchFiles()
-{
- string ret;
- string thedate;
-
- // Adding date to the directory name
- Batch::Date date = Batch::Date(time(0));
- thedate = date.str();
- int lend = thedate.size() ;
- int i = 0 ;
- while ( i < lend ) {
- if ( thedate[i] == '/' || thedate[i] == '-' || thedate[i] == ':' ) {
- thedate[i] = '_' ;
+ Launcher::Job * job = findJob(job_id);
+ return Launcher::XML_Persistence::dumpJob(*job);
+}
+
+int
+Launcher_cpp::restoreJob(const std::string& dumpedJob)
+{
+ LAUNCHER_MESSAGE("restore Job");
+ Launcher::Job* new_job(nullptr);
+ int jobId = -1;
+ try
+ {
+ {
+ new_job = Launcher::XML_Persistence::createJobFromString(dumpedJob);
}
- i++ ;
+ if(new_job)
+ {
+ jobId = addJob(new_job);
+ }
+ else
+ LAUNCHER_INFOS("Failed to create a new job.");
}
-
- ret = string("Batch/");
- ret += thedate;
- return ret;
+ catch(const LauncherException &ex)
+ {
+ LAUNCHER_INFOS("Cannot load the job. Exception: " << ex.msg.c_str());
+ }
+ return jobId;
}
-string Launcher_cpp::getRemoteFile( std::string remoteDir, std::string localFile )
+JobParameters_cpp
+Launcher_cpp::getJobParameters(int job_id)
{
- string::size_type pos = localFile.find_last_of("/") + 1;
- int ln = localFile.length() - pos;
- string remoteFile = remoteDir + "/" + localFile.substr(pos,ln);
- return remoteFile;
+ Launcher::Job * job = findJob(job_id);
+ JobParameters_cpp job_parameters;
+ job_parameters.job_name = job->getJobName();
+ job_parameters.job_type = job->getJobType();
+ job_parameters.job_file = job->getJobFile();
+ job_parameters.env_file = job->getEnvFile();
+ job_parameters.work_directory = job->getWorkDirectory();
+ job_parameters.local_directory = job->getLocalDirectory();
+ job_parameters.result_directory = job->getResultDirectory();
+ job_parameters.pre_command = job->getPreCommand();
+
+ // Parameters for COORM
+ job_parameters.launcher_file = job->getLauncherFile();
+ job_parameters.launcher_args = job->getLauncherArgs();
+
+ job_parameters.in_files = job->get_in_files();
+ job_parameters.out_files = job->get_out_files();
+
+ job_parameters.maximum_duration = job->getMaximumDuration();
+ job_parameters.queue = job->getQueue();
+ job_parameters.partition = job->getPartition();
+ job_parameters.exclusive = job->getExclusive();
+ job_parameters.mem_per_cpu = job->getMemPerCpu();
+ job_parameters.wckey = job->getWCKey();
+ job_parameters.extra_params = job->getExtraParams();
+
+ resourceParams resource_params = job->getResourceRequiredParams();
+ job_parameters.resource_required.name = resource_params.name;
+ job_parameters.resource_required.hostname = resource_params.hostname;
+ job_parameters.resource_required.OS = resource_params.OS;
+ job_parameters.resource_required.nb_proc = resource_params.nb_proc;
+ job_parameters.resource_required.nb_node = resource_params.nb_node;
+ job_parameters.resource_required.nb_proc_per_node = resource_params.nb_proc_per_node;
+ job_parameters.resource_required.cpu_clock = resource_params.cpu_clock;
+ job_parameters.resource_required.mem_mb = resource_params.mem_mb;
+
+ job_parameters.specific_parameters = job->getSpecificParameters();
+
+ return job_parameters;
}
-bool Launcher_cpp::check(const batchParams& batch_params)
+//=============================================================================
+/*!
+ * create a launcher job based on a file
+ * \param xmlExecuteFile : to define the execution on the batch cluster
+ */
+//=============================================================================
+long
+Launcher_cpp::createJobWithFile(const std::string xmlExecuteFile,
+ const std::string clusterName)
{
- bool rtn = true;
-#if defined(_DEBUG_) || defined(_DEBUG)
- cerr << "Job parameters are :" << endl;
- cerr << "Directory : $HOME/Batch/$date" << endl;
-#endif
+ LAUNCHER_MESSAGE("Begin of Launcher_cpp::createJobWithFile");
- // check expected_during_time (check the format)
- std::string edt_info = batch_params.expected_during_time;
- std::string edt_value = batch_params.expected_during_time;
- if (edt_value != "") {
- std::string begin_edt_value = edt_value.substr(0, 2);
- std::string mid_edt_value = edt_value.substr(2, 1);
- std::string end_edt_value = edt_value.substr(3);
-
- long value;
- std::istringstream iss(begin_edt_value);
- if (!(iss >> value)) {
- edt_info = "Error on definition ! : " + edt_value;
- rtn = false;
- }
- else if (value < 0) {
- edt_info = "Error on definition time is negative ! : " + value;
- rtn = false;
- }
- std::istringstream iss_2(end_edt_value);
- if (!(iss_2 >> value)) {
- edt_info = "Error on definition ! : " + edt_value;
- rtn = false;
- }
- else if (value < 0) {
- edt_info = "Error on definition time is negative ! : " + value;
- rtn = false;
- }
- if (mid_edt_value != ":") {
- edt_info = "Error on definition ! :" + edt_value;
- rtn = false;
- }
- }
- else {
- edt_info = "No value given";
- }
-#if defined(_DEBUG_) || defined(_DEBUG)
- cerr << "Expected during time : " << edt_info << endl;;
+ // Parsing xml file
+ ParserLauncherType job_params = ParseXmlFile(xmlExecuteFile);
+
+ // Creating a new job
+ std::unique_ptr<Launcher::Job_Command> new_job(new Launcher::Job_Command);
+
+ std::string cmdFile = Kernel_Utils::GetTmpFileName();
+#ifndef WIN32
+ cmdFile += ".sh";
+#else
+ cmdFile += ".bat";
#endif
+ std::ofstream os;
+ os.open(cmdFile.c_str(), std::ofstream::out );
+ os << "#! /bin/sh" << std::endl;
+ os << job_params.Command;
+ os.close();
+
+ new_job->setJobFile(cmdFile);
+ new_job->setLocalDirectory(job_params.RefDirectory);
+ new_job->setWorkDirectory(job_params.MachinesList[clusterName].WorkDirectory);
+ new_job->setEnvFile(job_params.MachinesList[clusterName].EnvFile);
+
+ for(size_t i=0; i < job_params.InputFile.size(); i++)
+ new_job->add_in_file(job_params.InputFile[i]);
+ for(size_t i=0; i < job_params.OutputFile.size();i++)
+ new_job->add_out_file(job_params.OutputFile[i]);
+
+ resourceParams p;
+ p.hostname = clusterName;
+ p.name = "";
+ p.OS = "";
+ p.nb_proc = job_params.NbOfProcesses;
+ p.nb_node = 0;
+ p.nb_proc_per_node = 0;
+ p.cpu_clock = 0;
+ p.mem_mb = 0;
+ new_job->setResourceRequiredParams(p);
+
+ createJob(new_job.get());
+ return new_job->getNumber();
+}
- // check memory (check the format)
- std::string mem_info;
- std::string mem_value = batch_params.mem;
- if (mem_value != "") {
- std::string begin_mem_value = mem_value.substr(0, mem_value.length()-2);
- long re_mem_value;
- std::istringstream iss(begin_mem_value);
- if (!(iss >> re_mem_value)) {
- mem_info = "Error on definition ! : " + mem_value;
- rtn = false;
- }
- else if (re_mem_value <= 0) {
- mem_info = "Error on definition memory is negative ! : " + mem_value;
- rtn = false;
- }
- std::string end_mem_value = mem_value.substr(mem_value.length()-2);
- if (end_mem_value != "gb" && end_mem_value != "mb") {
- mem_info = "Error on definition, type is bad ! " + mem_value;
- rtn = false;
- }
+//=============================================================================
+/*!
+ * Factory to instantiate the good batch manager for chosen cluster.
+ */
+//=============================================================================
+Batch::BatchManager *
+Launcher_cpp::FactoryBatchManager(ParserResourcesType& params)
+{
+ std::string mpi;
+ Batch::CommunicationProtocolType protocol;
+ Batch::FactBatchManager * fact;
+
+ std::string hostname = params.HostName;
+
+ switch(params.Protocol)
+ {
+ case sh:
+ protocol = Batch::SH;
+ break;
+ case rsh:
+ protocol = Batch::RSH;
+ break;
+ case ssh:
+ protocol = Batch::SSH;
+ break;
+ case rsync:
+ protocol = Batch::RSYNC;
+ break;
+ default:
+ throw LauncherException("Unknown protocol for this resource");
+ break;
}
- else {
- mem_info = "No value given";
+
+ switch(params.mpi)
+ {
+ case lam:
+ mpi = "lam";
+ break;
+ case mpich1:
+ mpi = "mpich1";
+ break;
+ case mpich2:
+ mpi = "mpich2";
+ break;
+ case openmpi:
+ mpi = "openmpi";
+ break;
+ case ompi:
+ mpi = "ompi";
+ break;
+ case slurmmpi:
+ mpi = "slurmmpi";
+ break;
+ case prun:
+ mpi = "prun";
+ break;
+ default:
+ mpi = "nompi";
}
-#if defined(_DEBUG_) || defined(_DEBUG)
- cerr << "Memory : " << mem_info << endl;
-#endif
- // check nb_proc
- std::string nb_proc_info;
- ostringstream nb_proc_value;
- nb_proc_value << batch_params.nb_proc;
- if(batch_params.nb_proc <= 0) {
- nb_proc_info = "Bad value ! nb_proc = ";
- nb_proc_info += nb_proc_value.str();
- rtn = false;
+ const char * bmType;
+ switch( params.Batch )
+ {
+ case pbs:
+ bmType = "PBS";
+ break;
+ case lsf:
+ bmType = "LSF";
+ break;
+ case sge:
+ bmType = "SGE";
+ break;
+ case ccc:
+ bmType = "CCC";
+ break;
+ case slurm:
+ bmType = "SLURM";
+ break;
+ case none:
+ bmType = "LOCAL";
+ break;
+ case ll:
+ bmType = "LL";
+ break;
+ case vishnu:
+ bmType = "VISHNU";
+ break;
+ case oar:
+ bmType = "OAR";
+ break;
+ case coorm:
+ bmType = "COORM";
+ break;
+ default:
+ LAUNCHER_MESSAGE("Bad batch description of the resource: Batch = " << params.Batch);
+ throw LauncherException("No batchmanager for that cluster - Bad batch description of the resource");
}
- else {
- nb_proc_info = nb_proc_value.str();
+ Batch::BatchManagerCatalog & cata = Batch::BatchManagerCatalog::getInstance();
+ fact = dynamic_cast<Batch::FactBatchManager*>(cata(bmType));
+ if (fact == NULL) {
+ LAUNCHER_MESSAGE("Cannot find batch manager factory for " << bmType << ". Check your version of libBatch.");
+ throw LauncherException("Cannot find batch manager factory");
}
-#if defined(_DEBUG_) || defined(_DEBUG)
- cerr << "Nb of processors : " << nb_proc_info << endl;
-#endif
+ LAUNCHER_MESSAGE("Instantiation of batch manager of type: " << bmType);
+ Batch::BatchManager * batch_client = (*fact)(hostname.c_str(), params.UserName.c_str(),
+ protocol, mpi.c_str());
+ return batch_client;
+}
- return rtn;
+//----------------------------------------------------------
+// Without LIBBATCH - Launcher_cpp do nothing...
+//----------------------------------------------------------
+#else
+
+void
+Launcher_cpp::createJob(Launcher::Job * new_job)
+{
+ LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot create a job !!!");
+ throw LauncherException("Method Launcher_cpp::createJob is not available "
+ "(libBatch was not present at compilation time)");
}
-long Launcher_cpp::getWallTime(std::string edt)
+int
+Launcher_cpp::createJob(const JobParameters_cpp& job_parameters)
{
- long hh, mm, ret;
+ LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot create a job !!!");
+ throw LauncherException("Method Launcher_cpp::createJob is not available "
+ "(libBatch was not present at compilation time)");
+}
- if( edt.size() == 0 )
- return 0;
+void
+Launcher_cpp::launchJob(int job_id)
+{
+ LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot launch a job !!!");
+ throw LauncherException("Method Launcher_cpp::launchJob is not available "
+ "(libBatch was not present at compilation time)");
+}
- string::size_type pos = edt.find(":");
- string h = edt.substr(0,pos);
- string m = edt.substr(pos+1,edt.size()-pos+1);
- istringstream issh(h);
- issh >> hh;
- istringstream issm(m);
- issm >> mm;
- ret = hh*60 + mm;
- return ret;
+std::string
+Launcher_cpp::getJobState(int job_id)
+{
+ LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot get job state!!!");
+ throw LauncherException("Method Launcher_cpp::getJobState is not available "
+ "(libBatch was not present at compilation time)");
}
-long Launcher_cpp::getRamSize(std::string mem)
+std::string
+Launcher_cpp::getAssignedHostnames(int job_id)
{
- long mv;
+ LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot get job assigned hostnames!!!");
+ throw LauncherException("Method Launcher_cpp::getAssignedHostnames is not available "
+ "(libBatch was not present at compilation time)");
+}
- if( mem.size() == 0 )
- return 0;
+void
+Launcher_cpp::exportInputFiles(int job_id)
+{
+ LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot copy the files!!!");
+ throw LauncherException("Method Launcher_cpp::exportInputFiles is not available "
+ "(libBatch was not present at compilation time)");
+}
- string ram = mem.substr(0,mem.size()-2);
- istringstream iss(ram);
- iss >> mv;
- string unity = mem.substr(mem.size()-2,2);
- if( (unity.find("gb") != string::npos) || (unity.find("GB") != string::npos) )
- return mv*1024;
- else if( (unity.find("mb") != string::npos) || (unity.find("MB") != string::npos) )
- return mv;
- else if( (unity.find("kb") != string::npos) || (unity.find("KB") != string::npos) )
- return mv/1024;
- else if( (unity.find("b") != string::npos) || (unity.find("B") != string::npos) )
- return mv/(1024*1024);
- else
- return 0;
+void
+Launcher_cpp::getJobResults(int job_id, std::string directory)
+{
+ LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot get job results!!!");
+ throw LauncherException("Method Launcher_cpp::getJobResults is not available "
+ "(libBatch was not present at compilation time)");
+}
+
+void
+Launcher_cpp::clearJobWorkingDir(int job_id)
+{
+ LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot clear directory!!!");
+ throw LauncherException("Method Launcher_cpp::clearJobWorkingDir is not available "
+ "(libBatch was not present at compilation time)");
+}
+
+bool
+Launcher_cpp::getJobDumpState(int job_id, std::string directory)
+{
+ LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot get job dump state!!!");
+ throw LauncherException("Method Launcher_cpp::getJobDumpState is not available "
+ "(libBatch was not present at compilation time)");
+}
+
+bool
+Launcher_cpp::getJobWorkFile(int job_id, std::string work_file, std::string directory)
+{
+ LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot get job dump state!!!");
+ throw LauncherException("Method Launcher_cpp::getJobWorkFile is not available "
+ "(libBatch was not present at compilation time)");
+}
+
+void
+Launcher_cpp::removeJob(int job_id)
+{
+ LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot remove job!!!");
+ throw LauncherException("Method Launcher_cpp::removeJob is not available "
+ "(libBatch was not present at compilation time)");
}
-void Launcher_cpp::ParseXmlFile(string xmlExecuteFile)
+void
+Launcher_cpp::stopJob(int job_id)
{
- SALOME_Launcher_Handler* handler = new SALOME_Launcher_Handler(_launch);
+ throw LauncherException("Method Launcher_cpp::stopJob is not available "
+ "(libBatch was not present at compilation time)");
+}
+
+std::string
+Launcher_cpp::dumpJob(int job_id)
+{
+ LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot dump job!!!");
+ throw LauncherException("Method Launcher_cpp::dumpJob is not available "
+ "(libBatch was not present at compilation time)");
+ return "";
+}
+
+int
+Launcher_cpp::restoreJob(const std::string& dumpedJob)
+{
+ LAUNCHER_INFOS("Launcher compiled without LIBBATCH - cannot restore job!!!");
+ throw LauncherException("Method Launcher_cpp::restoreJob is not available "
+ "(libBatch was not present at compilation time)");
+ return 0;
+}
+
+long
+Launcher_cpp::createJobWithFile( const std::string xmlExecuteFile, std::string clusterName)
+{
+ throw LauncherException("Method Launcher_cpp::createJobWithFile is not available "
+ "(libBatch was not present at compilation time)");
+ return 0;
+}
+
+JobParameters_cpp
+Launcher_cpp::getJobParameters(int job_id)
+{
+ throw LauncherException("Method Launcher_cpp::getJobParamaters is not available "
+ "(libBatch was not present at compilation time)");
+}
+
+#endif
+
+ParserLauncherType
+Launcher_cpp::ParseXmlFile(std::string xmlExecuteFile)
+{
+ ParserLauncherType job_params;
+ SALOME_Launcher_Handler * handler = new SALOME_Launcher_Handler(job_params);
const char* aFilePath = xmlExecuteFile.c_str();
FILE* aFile = fopen(aFilePath, "r");
-
if (aFile != NULL)
+ {
+ xmlDocPtr aDoc = xmlReadFile(aFilePath, NULL, 0);
+ if (aDoc != NULL)
+ handler->ProcessXmlDocument(aDoc);
+ else
{
- xmlDocPtr aDoc = xmlReadFile(aFilePath, NULL, 0);
-
- if (aDoc != NULL)
- handler->ProcessXmlDocument(aDoc);
- else{
-#if defined(_DEBUG_) || defined(_DEBUG)
- cout << "ResourcesManager_cpp: could not parse file "<< aFilePath << endl;
-#endif
- }
-
- // Free the document
- xmlFreeDoc(aDoc);
-
- fclose(aFile);
+ std::string message = "ResourcesManager_cpp: could not parse file: " + xmlExecuteFile;
+ LAUNCHER_MESSAGE(message);
+ delete handler;
+ throw LauncherException(message);
}
- else{
-#if defined(_DEBUG_) || defined(_DEBUG)
- cout << "Launcher_cpp: file "<<aFilePath<<" is not readable." << endl;
-#endif
+ // Free the document
+ xmlFreeDoc(aDoc);
+ fclose(aFile);
}
-
+ else
+ {
+ std::string message = "ResourcesManager_cpp: file is not readable: " + xmlExecuteFile;
+ LAUNCHER_MESSAGE(message);
+ delete handler;
+ throw LauncherException(message);
+ }
+
+ // Return
delete handler;
+ return job_params;
+}
+
+std::map<int, Launcher::Job *>
+Launcher_cpp::getJobs()
+{
+ return _launcher_job_map;
+}
+
+#ifdef WITH_LIBBATCH
+Batch::BatchManager*
+Launcher_cpp::getBatchManager(Launcher::Job * job)
+{
+ if(!_ResManager)
+ throw LauncherException("Resource manager is not set.");
+
+ Batch::BatchManager* result = nullptr;
+ int job_id = job->getNumber();
+
+ // Select a resource for the job
+ std::vector<std::string> ResourceList;
+ resourceParams params = job->getResourceRequiredParams();
+ // Consider only resources that can launch batch jobs
+ params.can_launch_batch_jobs = true;
+ try
+ {
+ ResourceList = _ResManager->GetFittingResources(params);
+ }
+ catch(const ResourcesException &ex)
+ {
+ throw LauncherException(ex.msg.c_str());
+ }
+ if (ResourceList.size() == 0)
+ {
+ LAUNCHER_INFOS("No adequate resource found for the job, number " << job->getNumber());
+ job->setState("ERROR");
+ throw LauncherException("No resource found the job");
+ }
+
+ // Configure the job with the resource selected - the first of the list
+ ParserResourcesType resource_definition = _ResManager->GetResourcesDescr(ResourceList[0]);
+
+ // Set resource definition to the job
+ // The job will check if the definitions needed
+ try
+ {
+ job->setResourceDefinition(resource_definition);
+ }
+ catch(const LauncherException &ex)
+ {
+ LAUNCHER_INFOS("Error in the definition of the resource, mess: " << ex.msg);
+ job->setState("ERROR");
+ throw ex;
+ }
+ // Step 2: We can now add a Factory if the resource is correctly define
+ std::map<int, Batch::BatchManager *>::const_iterator it = _batchmap.find(job_id);
+ if(it == _batchmap.end())
+ {
+ try
+ {
+ // Warning cannot write on one line like this, because map object is constructed before
+ // the method is called...
+ //_batchmap[job_id] = FactoryBatchManager(resource_definition);
+ result = FactoryBatchManager(resource_definition);
+ _batchmap[job_id] = result;
+ }
+ catch(const LauncherException &ex)
+ {
+ LAUNCHER_INFOS("Error during creation of the batch manager of the job, mess: " << ex.msg);
+ throw ex;
+ }
+ catch(const Batch::GenericException &ex)
+ {
+ LAUNCHER_INFOS("Error during creation of the batch manager of the job, mess: " << ex.message);
+ throw LauncherException(ex.message);
+ }
+ }
+ else
+ result = it->second;
+ return result;
}
+#endif
-std::string Launcher_cpp::getHomeDir(const ParserResourcesType& p, const std::string& tmpdir)
+void
+Launcher_cpp::addJobDirectlyToMap(Launcher::Job * new_job)
{
- std::string home;
- std::string command;
- int idx = tmpdir.find("Batch/");
- std::string filelogtemp = tmpdir.substr(idx+6, tmpdir.length());
- filelogtemp = "/tmp/logs" + filelogtemp + "_home";
-
- if( p.Protocol == rsh )
- command = "rsh ";
- else if( p.Protocol == ssh )
- command = "ssh ";
+ // Step 0: Calculated job_id
+ new_job->setNumber(_job_cpt);
+ _job_cpt++;
+
+#ifdef WITH_LIBBATCH
+ // Step 1: check if resource is already in the map
+ Batch::BatchManager * bm = getBatchManager(new_job);
+
+ // Step 2: add the job to the batch manager
+ try
+ {
+ Batch::JobId batch_manager_job_id = bm->addJob(*(new_job->getBatchJob()),
+ new_job->getReference());
+ new_job->setBatchManagerJobId(batch_manager_job_id);
+ }
+ catch(const Batch::GenericException &ex)
+ {
+ LAUNCHER_INFOS("Job cannot be added, exception in addJob: " << ex.message);
+ throw LauncherException(ex.message.c_str());
+ }
+
+ // Step 3: add job to launcher map
+ std::map<int, Launcher::Job *>::const_iterator it_job = _launcher_job_map.find(new_job->getNumber());
+ if (it_job == _launcher_job_map.end())
+ {
+ _launcher_job_map[new_job->getNumber()] = new_job;
+ }
else
- throw LauncherException("Unknown protocol");
- if (p.UserName != ""){
- command += p.UserName;
- command += "@";
- }
- command += p.Alias;
- command += " 'echo $HOME' > ";
- command += filelogtemp;
-#if defined(_DEBUG_) || defined(_DEBUG)
- std::cerr << command.c_str() << std::endl;
+ {
+ LAUNCHER_INFOS("A job as already the same id: " << new_job->getNumber());
+ throw LauncherException("A job as already the same id - job is not created !");
+ }
+ LAUNCHER_MESSAGE("New job added");
#endif
- int status = system(command.c_str());
- if(status)
- throw LauncherException("Error of launching home command on remote host");
-
- std::ifstream file_home(filelogtemp.c_str());
- std::getline(file_home, home);
- file_home.close();
- return home;
+}
+
+int
+Launcher_cpp::addJob(Launcher::Job * new_job)
+{
+ string job_state = new_job->getState();
+ int jobId = -1;
+ if (job_state == "CREATED")
+ {
+ // In this case, we ignore run_part information
+ createJob(new_job);
+ jobId = new_job->getNumber();
+ }
+ else if (job_state == "QUEUED" ||
+ job_state == "RUNNING" ||
+ job_state == "IN_PROCESS" ||
+ job_state == "PAUSED")
+ {
+ addJobDirectlyToMap(new_job);
+ jobId = new_job->getNumber();
+
+ // We check that the BatchManager could resume the job
+#ifdef WITH_LIBBATCH
+ if (new_job->getBatchManagerJobId().getReference() != new_job->getReference())
+ {
+ LAUNCHER_INFOS("BatchManager type cannot resume a job - job state is set to ERROR");
+ new_job->setState("ERROR");
+ }
+#endif
+ }
+ else if (job_state == "FINISHED" ||
+ job_state == "FAILED" ||
+ job_state == "ERROR")
+ {
+ // We add run_part information
+ addJobDirectlyToMap(new_job);
+ jobId = new_job->getNumber();
+ }
+ else
+ {
+ LAUNCHER_INFOS("A bad job is found, state unknown " << job_state);
+ jobId = -1;
+ }
+ return jobId;
+}
+
+list<int>
+Launcher_cpp::loadJobs(const char* jobs_file)
+{
+ list<int> new_jobs_id_list;
+
+ // Load the jobs from XML file
+ list<Launcher::Job *> jobs_list = Launcher::XML_Persistence::loadJobs(jobs_file);
+
+ // Create each job in the launcher
+ list<Launcher::Job *>::const_iterator it_job;
+ for (it_job = jobs_list.begin(); it_job != jobs_list.end(); it_job++)
+ {
+ Launcher::Job* new_job(*it_job);
+ int jobId = -1;
+ try
+ {
+ jobId = addJob(new_job);
+ if(jobId >= 0)
+ new_jobs_id_list.push_back(jobId);
+ }
+ catch(const LauncherException &ex)
+ {
+ LAUNCHER_INFOS("Cannot load the job. Exception: " << ex.msg.c_str());
+ }
+ }
+
+ return new_jobs_id_list;
+}
+
+void
+Launcher_cpp::saveJobs(const char* jobs_file)
+{
+ // Create a sorted list from the internal job map
+ list<const Launcher::Job *> jobs_list;
+ {
+ for (int i=0; i<_job_cpt; i++)
+ {
+ map<int, Launcher::Job *>::const_iterator it_job = _launcher_job_map.find(i);
+ if (it_job != _launcher_job_map.end())
+ jobs_list.push_back(it_job->second);
+ }
+ }
+ // Save the jobs in XML file
+ Launcher::XML_Persistence::saveJobs(jobs_file, jobs_list);
+}
+
+Launcher::Job *
+Launcher_cpp::findJob(int job_id)
+{
+ std::map<int, Launcher::Job *>::const_iterator it_job = _launcher_job_map.find(job_id);
+ if (it_job == _launcher_job_map.end())
+ {
+ LAUNCHER_INFOS("Cannot find the job, is it created ? job number: " << job_id);
+ throw LauncherException("Cannot find the job, is it created ?");
+ }
+ Launcher::Job * job = it_job->second;
+ return job;
}