}
// Methode pour le controle des jobs : soumet un job au gestionnaire
- const JobId BatchManager_CCC::submitJob(const Job & job)
+ const JobId BatchManager_CCC::runJob(const Job & job)
{
Parametre params = job.getParametre();
const std::string workDir = params[WORKDIR];
string::size_type p2 = fileToExecute.find_last_of(".");
std::string fileNameToExecute = fileToExecute.substr(p1+1,p2-p1-1);
- // export input files on cluster
- LOG("Export des fichiers en entree");
- exportInputFiles(job);
-
// build batch script for job
LOG("Construction du script de batch");
buildBatchScript(job);
// static string BatchManager_LSF::getDefaultServer();
// Methodes pour le controle des jobs
- virtual const JobId submitJob(const Job & job); // soumet un job au gestionnaire
+ virtual const JobId runJob(const Job & job); // soumet un job au gestionnaire
virtual void deleteJob(const JobId & jobid); // retire un job du gestionnaire
virtual JobInfo queryJob(const JobId & jobid); // renvoie l'etat du job
virtual bool isRunning(const JobId & jobid); // teste si un job est present en machine
namespace Batch
{
- BatchManager_COORM::BatchManager_COORM(const FactBatchManager * parent, const char * host,
- const char * username,
- CommunicationProtocolType protocolType, const char * mpiImpl)
- : BatchManager(parent, host, username, protocolType, mpiImpl)
- {
- }
-
- BatchManager_COORM::~BatchManager_COORM()
- {
- }
-
- // Soumet un job au gestionnaire
- const JobId BatchManager_COORM::submitJob(const Job & job)
- {
- Parametre params = job.getParametre();
- const string workDir = params[WORKDIR];
- const string fileToExecute = params[EXECUTABLE];
-
- // For CooRM
- const string launcherArgs = params[LAUNCHER_ARGS];
- const string launcherFile = params[LAUNCHER_FILE];
-
- const string jobName = params[NAME];
-
- 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);
-
- // For CooRM
- p1 = launcherFile.find_last_of("/");
- std::string fileNameToLaunch = launcherFile.substr(p1+1);
-
-
- // export input files on cluster
- exportInputFiles(job);
-
- // build batch script for job
- string scriptFile = buildBatchScript(job);
-
- // Get REMOTE_COORM_PATH environment variable
- const char * remote_coorm_path = getenv("REMOTE_COORM_PATH");
- if (remote_coorm_path == NULL)
- {
- throw RunTimeException("Unable to get REMOTE_COORM_PATH environment variable");
- }
-
- // We need omniORB to execute launcher.py
- const string set_env_cmd = "source " + string(remote_coorm_path) + "/coorm_prerequis.sh;";
-
-
- // define command to submit batch
- string subCommand = set_env_cmd + "python " + workDir + "/" + fileNameToLaunch + " --name="+ jobName +
- " --workdir=" + workDir + " --outputs=" + workDir + "/logs/outputs.log" +
- " --errors=" + workDir + "/logs/errors.log" +
- " --executable=" + scriptFile + " " + launcherArgs;
- string command = _protocol.getExecCommand(subCommand, _hostname, _username);
- command += " 2>&1";
- LOG(command);
-
- // submit job
- string output;
- int status = Utils::getCommandOutput(command, output);
- LOG(output);
- if (status != 0) throw RunTimeException("Can't submit job, error was: " + output);
-
- // read id of submitted job in output
- istringstream logfile(output);
- string sline, idline, id;
-
- if (logfile)
- {
- while (getline(logfile, sline) && sline != "")
- {
- idline = sline;
- }
-
- vector<string> tokens;
- JobInfo::Tokenize(idline, tokens, "=");
- id = tokens[1] ;
- }
- else
- {
- throw RunTimeException("Error in the submission of the job on the remote host");
- }
-
- JobId jobid(this, (string) id);
- return jobid;
- }
-
- // retire un job du gestionnaire
- void BatchManager_COORM::deleteJob(const JobId & jobid)
- {
- // Get REMOTE_COORM_PATH environment variable
- const char * remote_coorm_path = getenv("REMOTE_COORM_PATH");
- if (remote_coorm_path == NULL)
- {
- throw RunTimeException("Unable to get REMOTE_COORM_PATH environment variable");
- }
-
- // We need omniORB to execute launcher.py
- const string set_env_cmd = "source " + string(remote_coorm_path) + "/coorm_prerequis.sh;";
-
- // define command to delete job
- string subCommand = set_env_cmd + "python " + string(remote_coorm_path) + "/coormdel.py --jobID=" + jobid.getReference();
- string command = _protocol.getExecCommand(subCommand, _hostname, _username);
- LOG(command);
-
- int status = system(command.c_str());
- if (status)
- throw RunTimeException("Can't delete job " + jobid.getReference());
-
- LOG("job " << jobid.getReference() << " killed");
- }
-
- // Renvoie l'etat du job
- JobInfo BatchManager_COORM::queryJob(const JobId & jobid)
- {
- // Get REMOTE_COORM_PATH environment variable
- const char * remote_coorm_path = getenv("REMOTE_COORM_PATH");
- if (remote_coorm_path == NULL)
- {
- throw RunTimeException("Unable to get REMOTE_COORM_PATH environment variable");
- }
-
- // We need omniORB to execute launcher.py
- const string set_env_cmd = "source " + string(remote_coorm_path) + "/coorm_prerequis.sh;";
-
- // define command to query batch
- string subCommand = set_env_cmd + "python " + string(remote_coorm_path) + "/coormstat.py --jobID=" + jobid.getReference();
- string command = _protocol.getExecCommand(subCommand, _hostname, _username);
- LOG(command);
- string output;
- int status = Utils::getCommandOutput(command, output);
- if(status && status != 153 && status != 256*153)
- throw RunTimeException("Error of connection on remote host");
-
- JobInfo_COORM jobinfo = JobInfo_COORM(jobid.getReference(), output);
- return jobinfo;
- }
-
- string BatchManager_COORM::buildBatchScript(const Job & job)
- {
- Parametre params = job.getParametre();
-
- // Job Parameters
- string workDir = "";
- string fileToExecute = "";
- string tmpDir = "";
- int nbproc = 0;
- int edt = 0;
- int mem = 0;
- string queue = "";
-
- // Mandatory parameters
- if (params.find(WORKDIR) != params.end())
- workDir = params[WORKDIR].str();
- else
- throw RunTimeException("params[WORKDIR] is not defined. Please define it, cannot submit this job.");
- if (params.find(EXECUTABLE) != params.end())
- fileToExecute = params[EXECUTABLE].str();
- else
- throw RunTimeException("params[EXECUTABLE] is not defined. Please define it, cannot submit this job.");
-
- string::size_type p1 = fileToExecute.find_last_of("/");
- string::size_type p2 = fileToExecute.find_last_of(".");
- string rootNameToExecute = fileToExecute.substr(p1+1,p2-p1-1);
- string fileNameToExecute = fileToExecute.substr(p1+1);
-
- // Create batch submit file
- ofstream tempOutputFile;
- string tmpFileName = Utils::createAndOpenTemporaryFile("COORM-script", tempOutputFile);
-
- tempOutputFile << "#!/bin/sh -f" << endl;
- tempOutputFile << "export LIBBATCH_NODEFILE=$COORM_NODEFILE" << endl;
- // Launch the executable
- tempOutputFile << "cd " << tmpDir << endl;
- tempOutputFile << "./" + fileNameToExecute << endl;
- tempOutputFile.flush();
- tempOutputFile.close();
-
- Utils::chmod(tmpFileName.c_str(), 0x1ED);
- LOG("Batch script file generated is: " << tmpFileName);
-
- string remoteFileName = rootNameToExecute + "_Batch.sh";
- int status = _protocol.copyFile(tmpFileName, "", "",
- workDir + "/" + remoteFileName,
- _hostname, _username);
- if (status)
- throw RunTimeException("Cannot copy batch submission file on host " + _hostname);
-
- return remoteFileName;
- }
-
- void BatchManager_COORM::exportInputFiles(const Job & job)
- {
- BatchManager::exportInputFiles(job);
-
- int status;
- Parametre params = job.getParametre();
-
- string launcherFile = params[LAUNCHER_FILE];
-
- if (launcherFile.size() != 0)
- {
- // Copy launcherFile into batch working directory
- status = _protocol.copyFile(launcherFile, "", "",
- params[WORKDIR], _hostname, _username);
- if (status) {
- std::ostringstream oss;
- oss << "Cannot copy file " << launcherFile << " on host " << _hostname;
- oss << ". Return status is " << status;
- throw RunTimeException(oss.str());
+ BatchManager_COORM::BatchManager_COORM(const FactBatchManager * parent, const char * host,
+ const char * username,
+ CommunicationProtocolType protocolType, const char * mpiImpl)
+ : BatchManager(parent, host, username, protocolType, mpiImpl)
+ {
}
- string remoteLauncher = launcherFile;
- remoteLauncher = remoteLauncher.substr(remoteLauncher.rfind("/") + 1, remoteLauncher.length());
- remoteLauncher = string(params[WORKDIR]) + "/" + remoteLauncher;
-
- string subCommand = string("chmod u+x ") + remoteLauncher;
- string command = _protocol.getExecCommand(subCommand, _hostname, _username);
- LOG(command);
- status = system(command.c_str());
- if (status) {
- std::ostringstream oss;
- oss << "Cannot change permissions of file " << remoteLauncher << " on host " << _hostname;
- oss << ". Return status is " << status;
- throw RunTimeException(oss.str());
+ BatchManager_COORM::~BatchManager_COORM()
+ {
+ }
+
+ // Soumet un job au gestionnaire
+ const JobId BatchManager_COORM::runJob(const Job & job)
+ {
+ Parametre params = job.getParametre();
+ const string workDir = params[WORKDIR];
+ const string fileToExecute = params[EXECUTABLE];
+
+ // For CooRM
+ const string launcherArgs = params[LAUNCHER_ARGS];
+ const string launcherFile = params[LAUNCHER_FILE];
+
+ const string jobName = params[NAME];
+
+ 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);
+
+ // For CooRM
+ p1 = launcherFile.find_last_of("/");
+ std::string fileNameToLaunch = launcherFile.substr(p1+1);
+
+ // build batch script for job
+ string scriptFile = buildBatchScript(job);
+
+ // Get REMOTE_COORM_PATH environment variable
+ const char * remote_coorm_path = getenv("REMOTE_COORM_PATH");
+ if (remote_coorm_path == NULL)
+ {
+ throw RunTimeException("Unable to get REMOTE_COORM_PATH environment variable");
+ }
+
+ // We need omniORB to execute launcher.py
+ const string set_env_cmd = "source " + string(remote_coorm_path) + "/coorm_prerequis.sh;";
+
+
+ // define command to submit batch
+ string subCommand = set_env_cmd + "python " + workDir + "/" + fileNameToLaunch + " --name="+ jobName +
+ " --workdir=" + workDir + " --outputs=" + workDir + "/logs/outputs.log" +
+ " --errors=" + workDir + "/logs/errors.log" +
+ " --executable=" + scriptFile + " " + launcherArgs;
+ string command = _protocol.getExecCommand(subCommand, _hostname, _username);
+ command += " 2>&1";
+ LOG(command);
+
+ // submit job
+ string output;
+ int status = Utils::getCommandOutput(command, output);
+ LOG(output);
+ if (status != 0) throw RunTimeException("Can't submit job, error was: " + output);
+
+ // read id of submitted job in output
+ istringstream logfile(output);
+ string sline, idline, id;
+
+ if (logfile)
+ {
+ while (getline(logfile, sline) && sline != "")
+ {
+ idline = sline;
+ }
+
+ vector<string> tokens;
+ JobInfo::Tokenize(idline, tokens, "=");
+ id = tokens[1] ;
+ }
+ else
+ {
+ throw RunTimeException("Error in the submission of the job on the remote host");
+ }
+
+ JobId jobid(this, (string) id);
+ return jobid;
+ }
+
+ // retire un job du gestionnaire
+ void BatchManager_COORM::deleteJob(const JobId & jobid)
+ {
+ // Get REMOTE_COORM_PATH environment variable
+ const char * remote_coorm_path = getenv("REMOTE_COORM_PATH");
+ if (remote_coorm_path == NULL)
+ {
+ throw RunTimeException("Unable to get REMOTE_COORM_PATH environment variable");
+ }
+
+ // We need omniORB to execute launcher.py
+ const string set_env_cmd = "source " + string(remote_coorm_path) + "/coorm_prerequis.sh;";
+
+ // define command to delete job
+ string subCommand = set_env_cmd + "python " + string(remote_coorm_path) + "/coormdel.py --jobID=" + jobid.getReference();
+ string command = _protocol.getExecCommand(subCommand, _hostname, _username);
+ LOG(command);
+
+ int status = system(command.c_str());
+ if (status)
+ throw RunTimeException("Can't delete job " + jobid.getReference());
+
+ LOG("job " << jobid.getReference() << " killed");
+ }
+
+ // Renvoie l'etat du job
+ JobInfo BatchManager_COORM::queryJob(const JobId & jobid)
+ {
+ // Get REMOTE_COORM_PATH environment variable
+ const char * remote_coorm_path = getenv("REMOTE_COORM_PATH");
+ if (remote_coorm_path == NULL)
+ {
+ throw RunTimeException("Unable to get REMOTE_COORM_PATH environment variable");
+ }
+
+ // We need omniORB to execute launcher.py
+ const string set_env_cmd = "source " + string(remote_coorm_path) + "/coorm_prerequis.sh;";
+
+ // define command to query batch
+ string subCommand = set_env_cmd + "python " + string(remote_coorm_path) + "/coormstat.py --jobID=" + jobid.getReference();
+ string command = _protocol.getExecCommand(subCommand, _hostname, _username);
+ LOG(command);
+ string output;
+ int status = Utils::getCommandOutput(command, output);
+ if(status && status != 153 && status != 256*153)
+ throw RunTimeException("Error of connection on remote host");
+
+ JobInfo_COORM jobinfo = JobInfo_COORM(jobid.getReference(), output);
+ return jobinfo;
+ }
+
+ string BatchManager_COORM::buildBatchScript(const Job & job)
+ {
+ Parametre params = job.getParametre();
+
+ // Job Parameters
+ string workDir = "";
+ string fileToExecute = "";
+ string tmpDir = "";
+ int nbproc = 0;
+ int edt = 0;
+ int mem = 0;
+ string queue = "";
+
+ // Mandatory parameters
+ if (params.find(WORKDIR) != params.end())
+ workDir = params[WORKDIR].str();
+ else
+ throw RunTimeException("params[WORKDIR] is not defined. Please define it, cannot submit this job.");
+ if (params.find(EXECUTABLE) != params.end())
+ fileToExecute = params[EXECUTABLE].str();
+ else
+ throw RunTimeException("params[EXECUTABLE] is not defined. Please define it, cannot submit this job.");
+
+ string::size_type p1 = fileToExecute.find_last_of("/");
+ string::size_type p2 = fileToExecute.find_last_of(".");
+ string rootNameToExecute = fileToExecute.substr(p1+1,p2-p1-1);
+ string fileNameToExecute = fileToExecute.substr(p1+1);
+
+ // Create batch submit file
+ ofstream tempOutputFile;
+ string tmpFileName = Utils::createAndOpenTemporaryFile("COORM-script", tempOutputFile);
+
+ tempOutputFile << "#!/bin/sh -f" << endl;
+ tempOutputFile << "export LIBBATCH_NODEFILE=$COORM_NODEFILE" << endl;
+ // Launch the executable
+ tempOutputFile << "cd " << tmpDir << endl;
+ tempOutputFile << "./" + fileNameToExecute << endl;
+ tempOutputFile.flush();
+ tempOutputFile.close();
+
+ Utils::chmod(tmpFileName.c_str(), 0x1ED);
+ LOG("Batch script file generated is: " << tmpFileName);
+
+ string remoteFileName = rootNameToExecute + "_Batch.sh";
+ int status = _protocol.copyFile(tmpFileName, "", "",
+ workDir + "/" + remoteFileName,
+ _hostname, _username);
+ if (status)
+ throw RunTimeException("Cannot copy batch submission file on host " + _hostname);
+
+ return remoteFileName;
+ }
+
+ void BatchManager_COORM::exportInputFiles(const Job & job)
+ {
+ BatchManager::exportInputFiles(job);
+
+ int status;
+ Parametre params = job.getParametre();
+
+ string launcherFile = params[LAUNCHER_FILE];
+
+ if (launcherFile.size() != 0)
+ {
+ // Copy launcherFile into batch working directory
+ status = _protocol.copyFile(launcherFile, "", "",
+ params[WORKDIR], _hostname, _username);
+ if (status) {
+ std::ostringstream oss;
+ oss << "Cannot copy file " << launcherFile << " on host " << _hostname;
+ oss << ". Return status is " << status;
+ throw RunTimeException(oss.str());
+ }
+
+ string remoteLauncher = launcherFile;
+ remoteLauncher = remoteLauncher.substr(remoteLauncher.rfind("/") + 1, remoteLauncher.length());
+ remoteLauncher = string(params[WORKDIR]) + "/" + remoteLauncher;
+
+ string subCommand = string("chmod u+x ") + remoteLauncher;
+ string command = _protocol.getExecCommand(subCommand, _hostname, _username);
+ LOG(command);
+ status = system(command.c_str());
+ if (status) {
+ std::ostringstream oss;
+ oss << "Cannot change permissions of file " << remoteLauncher << " on host " << _hostname;
+ oss << ". Return status is " << status;
+ throw RunTimeException(oss.str());
+ }
+ }
}
- }
- }
}
namespace Batch
{
- class BATCH_EXPORT BatchManager_COORM : public BatchManager
- {
- public:
- // Constructeur
- BatchManager_COORM(const FactBatchManager * parent, const char * host="localhost",
- const char * username="",
- CommunicationProtocolType protocolType = SSH, const char * mpiImpl="nompi");
+ class BATCH_EXPORT BatchManager_COORM : public BatchManager
+ {
+ public:
+ // Constructeur
+ BatchManager_COORM(const FactBatchManager * parent, const char * host="localhost",
+ const char * username="",
+ CommunicationProtocolType protocolType = SSH, const char * mpiImpl="nompi");
- // Destructeur
- virtual ~BatchManager_COORM();
+ // Destructeur
+ virtual ~BatchManager_COORM();
- // Soumet un job
- virtual const JobId submitJob(const Job & job);
+ // Soumet un job
+ virtual const JobId runJob(const Job & job);
- // Supprime un job
- virtual void deleteJob(const JobId & jobid);
+ // Supprime un job
+ virtual void deleteJob(const JobId & jobid);
- // Donne l'etat du job
- virtual JobInfo queryJob(const JobId & jobid);
+ // Donne l'etat du job
+ virtual JobInfo queryJob(const JobId & jobid);
- // Modifie un job en file d'attente
- virtual void setParametre(const JobId & jobid, const Parametre & param) { return alterJob(jobid, param); }
+ // Modifie un job en file d'attente
+ virtual void setParametre(const JobId & jobid, const Parametre & param) { return alterJob(jobid, param); }
- // Modifie un job en file d'attente
- virtual void setEnvironnement(const JobId & jobid, const Environnement & env) { return alterJob(jobid, env); }
+ // Modifie un job en file d'attente
+ virtual void setEnvironnement(const JobId & jobid, const Environnement & env) { return alterJob(jobid, env); }
- protected:
- std::string buildBatchScript(const Job & job);
- void exportInputFiles(const Job & job);
+ protected:
+ std::string buildBatchScript(const Job & job);
+ void exportInputFiles(const Job & job);
#ifdef SWIG
- public:
- // Recupere le l'identifiant d'un job deja soumis au BatchManager
- virtual const JobId getJobIdByReference(const char * ref) { return BatchManager::getJobIdByReference(ref); }
+ public:
+ // Recupere le l'identifiant d'un job deja soumis au BatchManager
+ virtual const JobId getJobIdByReference(const char * ref) { return BatchManager::getJobIdByReference(ref); }
#endif
- };
+ };
}
#endif
// Methode pour le controle des jobs : soumet un job au gestionnaire
const JobId BatchManager::submitJob(const Job & job)
{
- throw NotYetImplementedException("Method submitJob not implemented by Batch Manager \"" + _type + "\"");
+ exportInputFiles(job);
+ preprocess(job);
+ return runJob(job);
}
// Methode pour le controle des jobs : retire un job du gestionnaire
return _protocol;
}
+ void BatchManager::preprocess(const Batch::Job & job)
+ {
+ std::string preproCommand;
+ std::string workDir;
+ Parametre params = job.getParametre();
+ if (params.find(PREPROCESS) != params.end())
+ preproCommand = params[PREPROCESS].str();
+ if (params.find(WORKDIR) != params.end())
+ workDir = params[WORKDIR].str();
+
+ if(!preproCommand.empty() && !workDir.empty())
+ {
+ std::string subCommand = "cd " + workDir + "; " + preproCommand;
+ std::string command = _protocol.getExecCommand(subCommand, _hostname, _username);
+ command += " 2>&1";
+ LOG(command);
+
+ // submit job
+ std::string output;
+ int status = Utils::getCommandOutput(command, output);
+ LOG(output);
+ if (status != 0)
+ throw RunTimeException("Error when executing: " + command +
+ "\nOutput:" + output);
+ }
+ }
+
+ const Batch::JobId BatchManager::runJob(const Batch::Job & job)
+ {
+ throw NotYetImplementedException("Method runJob not implemented by Batch Manager \"" + _type + "\"");
+ }
}
MpiImpl *_mpiImpl; // Mpi implementation to launch executable in batch script
MpiImpl* FactoryMpiImpl(std::string mpiImpl);
- void exportInputFiles(const Job & job);
+ virtual void exportInputFiles(const Job & job);
+
+ // Preprocessing done on the frontal using "PREPROCESS" parameter as a script.
+ // May throw exceptions in case of failure.
+ virtual void preprocess(const Batch::Job & job);
+
+ // Submit to batch manager, but we suppose input files are already copied
+ // and preprocess finished without error
+ virtual const Batch::JobId runJob(const Batch::Job & job);
private:
def_Constant(MEMPERCPU);
def_Constant(WCKEY);
def_Constant(EXTRAPARAMS);
+ def_Constant(PREPROCESS);
// Parameters for COORM
def_Constant(LAUNCHER_FILE);
decl_extern_Constant(MEMPERCPU);
decl_extern_Constant(WCKEY);
decl_extern_Constant(EXTRAPARAMS);
+ decl_extern_Constant(PREPROCESS);
// Parameters for COORM
decl_extern_Constant(LAUNCHER_FILE);
addParameter("MEMPERCPU", LONG, 1);
addParameter("WCKEY", STRING, 1);
addParameter("EXTRAPARAMS", STRING, 1);
+ addParameter("PREPROCESS", STRING, 1);
- // Parameters for COORM
+ // Parameters for COORM
addParameter("LAUNCHER_FILE", STRING, 1);
addParameter("LAUNCHER_ARGS", STRING, 1);
}
class BATCH_EXPORT RunTimeException : public GenericException
{
public:
- // Constructeur
+ // Constructeur
RunTimeException(std::string ch = "undefined") : GenericException("RunTimeException", ch) {}
};
}
// Methode pour le controle des jobs : soumet un job au gestionnaire
- const JobId BatchManager_LSF::submitJob(const Job & job)
+ const JobId BatchManager_LSF::runJob(const Job & job)
{
Parametre params = job.getParametre();
const std::string workDir = params[WORKDIR];
- // export input files on cluster
- LOG("Export des fichiers en entree");
- exportInputFiles(job);
-
// build batch script for job
LOG("Construction du script de batch");
string scriptFile = buildSubmissionScript(job);
// static string BatchManager_LSF::getDefaultServer();
// Methodes pour le controle des jobs
- virtual const JobId submitJob(const Job & job); // soumet un job au gestionnaire
+ virtual const JobId runJob(const Job & job); // soumet un job au gestionnaire
virtual void deleteJob(const JobId & jobid); // retire un job du gestionnaire
virtual JobInfo queryJob(const JobId & jobid); // renvoie l'etat du job
virtual bool isRunning(const JobId & jobid); // teste si un job est present en machine
}
// Method to submit a job to the batch manager
- const JobId BatchManager_LL::submitJob(const Job & job)
+ const JobId BatchManager_LL::runJob(const Job & job)
{
Parametre params = job.getParametre();
const string workDir = params[WORKDIR];
- // export input files on cluster
- exportInputFiles(job);
-
// build command file to submit the job and copy it on the server
string cmdFile = buildCommandFile(job);
virtual ~BatchManager_LL();
// Methods to control jobs
- virtual const JobId submitJob(const Job & job);
+ virtual const JobId runJob(const Job & job);
virtual void deleteJob(const JobId & jobid);
virtual JobInfo queryJob(const JobId & jobid);
}
// Methode pour le controle des jobs : soumet un job au gestionnaire
- const JobId BatchManager_Local::submitJob(const Job & job)
+ const JobId BatchManager_Local::runJob(const Job & job)
{
- // export input files in the working directory of the execution host
- exportInputFiles(job);
-
Job_Local jobLocal = job;
Id id = _idCounter++;
ThreadAdapter * p_ta = new ThreadAdapter(*this, job, id);
// static string BatchManager_Local::getDefaultServer();
// Methodes pour le controle des jobs
- virtual const JobId submitJob(const Job & job); // soumet un job au gestionnaire
+ virtual const JobId runJob(const Job & job); // soumet un job au gestionnaire
virtual void deleteJob(const JobId & jobid); // retire un job du gestionnaire
virtual void holdJob(const JobId & jobid); // suspend un job en file d'attente
virtual void releaseJob(const JobId & jobid); // relache un job suspendu
namespace Batch
{
- BatchManager_OAR::BatchManager_OAR(const FactBatchManager * parent, const char * host,
- const char * username,
- CommunicationProtocolType protocolType, const char * mpiImpl)
- : BatchManager(parent, host, username, protocolType, mpiImpl)
- {
- }
-
- BatchManager_OAR::~BatchManager_OAR()
- {
- }
-
- // Soumet un job au gestionnaire
- const JobId BatchManager_OAR::submitJob(const Job & job)
- {
- Parametre params = job.getParametre();
- const string workDir = params[WORKDIR];
- const string fileToExecute = params[EXECUTABLE];
- 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);
-
- // export input files on cluster
- exportInputFiles(job);
-
- // build batch script for job
- string scriptFile = buildBatchScript(job);
+ BatchManager_OAR::BatchManager_OAR(const FactBatchManager * parent, const char * host,
+ const char * username,
+ CommunicationProtocolType protocolType, const char * mpiImpl)
+ : BatchManager(parent, host, username, protocolType, mpiImpl)
+ {
+ }
+
+ BatchManager_OAR::~BatchManager_OAR()
+ {
+ }
+
+ // Soumet un job au gestionnaire
+ const JobId BatchManager_OAR::runJob(const Job & job)
+ {
+ Parametre params = job.getParametre();
+ const string workDir = params[WORKDIR];
+ const string fileToExecute = params[EXECUTABLE];
+ 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);
+
+ // build batch script for job
+ string scriptFile = buildBatchScript(job);
// define command to submit batch
- string subCommand = string("oarsub -t allow_classic_ssh -d ") + workDir + " -S " + workDir + "/" + scriptFile;
- string command = _protocol.getExecCommand(subCommand, _hostname, _username);
+ string subCommand = string("oarsub -t allow_classic_ssh -d ") + workDir + " -S " + workDir + "/" + scriptFile;
+ string command = _protocol.getExecCommand(subCommand, _hostname, _username);
command += " 2>&1";
LOG(command);
LOG(output);
if (status != 0) throw RunTimeException("Can't submit job, error was: " + output);
- // read id of submitted job in output
- istringstream logfile(output);
- string sline, idline, id;
-
- if (logfile)
- {
- while (getline(logfile, sline) && sline != "")
- {
- idline = sline;
- }
-
- vector<string> tokens;
- JobInfo::Tokenize(idline, tokens, "=");
- id = tokens[1];
- }
- else
- {
- throw RunTimeException("Error in the submission of the job on the remote host");
- }
-
- JobId jobid(this, id);
- return jobid;
- }
-
- // retire un job du gestionnaire
- void BatchManager_OAR::deleteJob(const JobId & jobid)
- {
+ // read id of submitted job in output
+ istringstream logfile(output);
+ string sline, idline, id;
+
+ if (logfile)
+ {
+ while (getline(logfile, sline) && sline != "")
+ {
+ idline = sline;
+ }
+
+ vector<string> tokens;
+ JobInfo::Tokenize(idline, tokens, "=");
+ id = tokens[1];
+ }
+ else
+ {
+ throw RunTimeException("Error in the submission of the job on the remote host");
+ }
+
+ JobId jobid(this, id);
+ return jobid;
+ }
+
+ // retire un job du gestionnaire
+ void BatchManager_OAR::deleteJob(const JobId & jobid)
+ {
// define command to delete job
string subCommand = "oardel " + jobid.getReference();
string command = _protocol.getExecCommand(subCommand, _hostname, _username);
throw RunTimeException("Can't delete job " + jobid.getReference());
LOG("job " << jobid.getReference() << " killed");
- }
+ }
- // Renvoie l'etat du job
- JobInfo BatchManager_OAR::queryJob(const JobId & jobid)
- {
+ // Renvoie l'etat du job
+ JobInfo BatchManager_OAR::queryJob(const JobId & jobid)
+ {
// define command to query batch
string subCommand = "oarstat -fj " + jobid.getReference();
string command = _protocol.getExecCommand(subCommand, _hostname, _username);
JobInfo_OAR jobinfo = JobInfo_OAR(jobid.getReference(), output);
return jobinfo;
- }
+ }
- string BatchManager_OAR::buildBatchScript(const Job & job)
- {
- Parametre params = job.getParametre();
+ string BatchManager_OAR::buildBatchScript(const Job & job)
+ {
+ Parametre params = job.getParametre();
- // Job Parameters
- string workDir = "";
- string fileToExecute = "";
- string tmpDir = "";
- int nbproc = 0;
- int mem = 0;
- string queue = "";
+ // Job Parameters
+ string workDir = "";
+ string fileToExecute = "";
+ string tmpDir = "";
+ int nbproc = 0;
+ int mem = 0;
+ string queue = "";
// Mandatory parameters
if (params.find(WORKDIR) != params.end())
else
throw RunTimeException("params[EXECUTABLE] is not defined. Please define it, cannot submit this job.");
- // Optional parameters
- if (params.find(NBPROC) != params.end())
- nbproc = params[NBPROC];
+ // Optional parameters
+ if (params.find(NBPROC) != params.end())
+ nbproc = params[NBPROC];
int nbprocpernode = 1;
if (params.find(NBPROCPERNODE) != params.end())
nbprocpernode = params[NBPROCPERNODE];
long walltimeSecs = 0;
- if (params.find(MAXWALLTIME) != params.end())
- walltimeSecs = (long)params[MAXWALLTIME] * 60;
- if (params.find(MAXRAMSIZE) != params.end())
- mem = params[MAXRAMSIZE];
- if (params.find(QUEUE) != params.end())
- queue = params[QUEUE].str();
-
- string::size_type p1 = fileToExecute.find_last_of("/");
- string::size_type p2 = fileToExecute.find_last_of(".");
- string rootNameToExecute = fileToExecute.substr(p1+1,p2-p1-1);
- string fileNameToExecute = fileToExecute.substr(p1+1);
-
- // Create batch submit file
+ if (params.find(MAXWALLTIME) != params.end())
+ walltimeSecs = (long)params[MAXWALLTIME] * 60;
+ if (params.find(MAXRAMSIZE) != params.end())
+ mem = params[MAXRAMSIZE];
+ if (params.find(QUEUE) != params.end())
+ queue = params[QUEUE].str();
+
+ string::size_type p1 = fileToExecute.find_last_of("/");
+ string::size_type p2 = fileToExecute.find_last_of(".");
+ string rootNameToExecute = fileToExecute.substr(p1+1,p2-p1-1);
+ string fileNameToExecute = fileToExecute.substr(p1+1);
+
+ // Create batch submit file
ofstream tempOutputFile;
string tmpFileName = Utils::createAndOpenTemporaryFile("OAR-script", tempOutputFile);
- tempOutputFile << "#!/bin/sh -f" << endl;
-
- int nb_full_nodes(0);
- int nb_proc_on_last_node(0);
-
- if (nbproc > 0)
- {
- nb_full_nodes = nbproc / nbprocpernode;
- nb_proc_on_last_node = nbproc % nbprocpernode;
-
- // In exclusive mode, we reserve all procs on the nodes
- if (params.find(EXCLUSIVE) != params.end() && params[EXCLUSIVE] && nb_proc_on_last_node > 0)
- {
- nb_full_nodes += 1;
- nb_proc_on_last_node = 0;
- }
- }
-
- if (nb_full_nodes > 0)
- {
- tempOutputFile << "#OAR -l nodes=" << nb_full_nodes;
- if (walltimeSecs > 0)
- {
- tempOutputFile << ",walltime=" << convertSecTo_H_M_S(walltimeSecs) << endl;
- }
- else
- {
- tempOutputFile << endl;
- }
- }
- else
- {
- if (walltimeSecs > 0)
- {
- tempOutputFile << "#OAR -l walltime=" << convertSecTo_H_M_S(walltimeSecs) << endl;
- }
- }
-
- if (queue != "")
- {
- tempOutputFile << "#OAR -q " << queue << endl;
- }
-
- tempOutputFile << "#OAR -O " << tmpDir << "/logs/output.log." << rootNameToExecute << endl;
- tempOutputFile << "#OAR -E " << tmpDir << "/logs/error.log." << rootNameToExecute << endl;
-
- tempOutputFile << "export LIBBATCH_NODEFILE=$OAR_NODEFILE" << endl;
-
- // Launch the executable
- tempOutputFile << "cd " << tmpDir << endl;
- tempOutputFile << "./" + fileNameToExecute << endl;
- tempOutputFile.flush();
- tempOutputFile.close();
-
- Utils::chmod(tmpFileName.c_str(), 0x1ED);
- LOG("Batch script file generated is: " << tmpFileName);
+ tempOutputFile << "#!/bin/sh -f" << endl;
+
+ int nb_full_nodes(0);
+ int nb_proc_on_last_node(0);
+
+ if (nbproc > 0)
+ {
+ nb_full_nodes = nbproc / nbprocpernode;
+ nb_proc_on_last_node = nbproc % nbprocpernode;
+
+ // In exclusive mode, we reserve all procs on the nodes
+ if (params.find(EXCLUSIVE) != params.end() && params[EXCLUSIVE] && nb_proc_on_last_node > 0)
+ {
+ nb_full_nodes += 1;
+ nb_proc_on_last_node = 0;
+ }
+ }
+
+ if (nb_full_nodes > 0)
+ {
+ tempOutputFile << "#OAR -l nodes=" << nb_full_nodes;
+ if (walltimeSecs > 0)
+ {
+ tempOutputFile << ",walltime=" << convertSecTo_H_M_S(walltimeSecs) << endl;
+ }
+ else
+ {
+ tempOutputFile << endl;
+ }
+ }
+ else
+ {
+ if (walltimeSecs > 0)
+ {
+ tempOutputFile << "#OAR -l walltime=" << convertSecTo_H_M_S(walltimeSecs) << endl;
+ }
+ }
+
+ if (queue != "")
+ {
+ tempOutputFile << "#OAR -q " << queue << endl;
+ }
+
+ tempOutputFile << "#OAR -O " << tmpDir << "/logs/output.log." << rootNameToExecute << endl;
+ tempOutputFile << "#OAR -E " << tmpDir << "/logs/error.log." << rootNameToExecute << endl;
+
+ tempOutputFile << "export LIBBATCH_NODEFILE=$OAR_NODEFILE" << endl;
+
+ // Launch the executable
+ tempOutputFile << "cd " << tmpDir << endl;
+ tempOutputFile << "./" + fileNameToExecute << endl;
+ tempOutputFile.flush();
+ tempOutputFile.close();
+
+ Utils::chmod(tmpFileName.c_str(), 0x1ED);
+ LOG("Batch script file generated is: " << tmpFileName);
string remoteFileName = rootNameToExecute + "_Batch.sh";
int status = _protocol.copyFile(tmpFileName, "", "",
throw RunTimeException("Cannot copy batch submission file on host " + _hostname);
return remoteFileName;
- }
-
- string BatchManager_OAR::convertSecTo_H_M_S(long seconds) const
- {
- int h(seconds / 3600);
- int m((seconds % 3600) / 60);
- int s((seconds % 3600) % 60);
-
- stringstream ss;
- ss << h << ":" << m << ":" << s;
-
- return ss.str();
- }
+ }
+
+ string BatchManager_OAR::convertSecTo_H_M_S(long seconds) const
+ {
+ int h(seconds / 3600);
+ int m((seconds % 3600) / 60);
+ int s((seconds % 3600) % 60);
+
+ stringstream ss;
+ ss << h << ":" << m << ":" << s;
+
+ return ss.str();
+ }
}
namespace Batch
{
- class BATCH_EXPORT BatchManager_OAR : public BatchManager
- {
- public:
- // Constructeur
- BatchManager_OAR(const FactBatchManager * parent, const char * host="localhost",
- const char * username="",
- CommunicationProtocolType protocolType = SSH, const char * mpiImpl="nompi");
+ class BATCH_EXPORT BatchManager_OAR : public BatchManager
+ {
+ public:
+ // Constructeur
+ BatchManager_OAR(const FactBatchManager * parent, const char * host="localhost",
+ const char * username="",
+ CommunicationProtocolType protocolType = SSH, const char * mpiImpl="nompi");
- // Destructeur
- virtual ~BatchManager_OAR();
+ // Destructeur
+ virtual ~BatchManager_OAR();
- // Soumet un job
- virtual const JobId submitJob(const Job & job);
+ // Soumet un job
+ virtual const JobId runJob(const Job & job);
- // Supprime un job
- virtual void deleteJob(const JobId & jobid);
+ // Supprime un job
+ virtual void deleteJob(const JobId & jobid);
- // Donne l'etat du job
- virtual JobInfo queryJob(const JobId & jobid);
+ // Donne l'etat du job
+ virtual JobInfo queryJob(const JobId & jobid);
- // Modifie un job en file d'attente
- virtual void setParametre(const JobId & jobid, const Parametre & param) { return alterJob(jobid, param); }
+ // Modifie un job en file d'attente
+ virtual void setParametre(const JobId & jobid, const Parametre & param) { return alterJob(jobid, param); }
- // Modifie un job en file d'attente
- virtual void setEnvironnement(const JobId & jobid, const Environnement & env) { return alterJob(jobid, env); }
+ // Modifie un job en file d'attente
+ virtual void setEnvironnement(const JobId & jobid, const Environnement & env) { return alterJob(jobid, env); }
- protected:
- std::string buildBatchScript(const Job & job);
- std::string convertSecTo_H_M_S(long seconds) const;
+ protected:
+ std::string buildBatchScript(const Job & job);
+ std::string convertSecTo_H_M_S(long seconds) const;
#ifdef SWIG
- public:
- // Recupere le l'identifiant d'un job deja soumis au BatchManager
- virtual const JobId getJobIdByReference(const char * ref) { return BatchManager::getJobIdByReference(ref); }
+ public:
+ // Recupere l'identifiant d'un job deja soumis au BatchManager
+ virtual const JobId getJobIdByReference(const char * ref) { return BatchManager::getJobIdByReference(ref); }
#endif
- };
+ };
}
#endif
}
// Methode pour le controle des jobs : soumet un job au gestionnaire
- const JobId BatchManager_PBS::submitJob(const Job & job)
+ const JobId BatchManager_PBS::runJob(const Job & job)
{
Parametre params = job.getParametre();
const std::string workDir = params[WORKDIR];
- // export input files on cluster
- exportInputFiles(job);
-
// build batch script for job
string scriptFile = buildSubmissionScript(job);
// static string BatchManager_LSF::getDefaultServer();
// Methodes pour le controle des jobs
- virtual const JobId submitJob(const Job & job); // soumet un job au gestionnaire
+ virtual const JobId runJob(const Job & job); // soumet un job au gestionnaire
virtual void deleteJob(const JobId & jobid); // retire un job du gestionnaire
virtual JobInfo queryJob(const JobId & jobid); // renvoie l'etat du job
virtual bool isRunning(const JobId & jobid); // teste si un job est present en machine
}
// Methode pour le controle des jobs : soumet un job au gestionnaire
- const JobId BatchManager_SGE::submitJob(const Job & job)
+ const JobId BatchManager_SGE::runJob(const Job & job)
{
Parametre params = job.getParametre();
const std::string workDir = params[WORKDIR];
string::size_type p2 = fileToExecute.find_last_of(".");
std::string fileNameToExecute = fileToExecute.substr(p1+1,p2-p1-1);
- // export input files on cluster
- exportInputFiles(job);
-
// build batch script for job
buildBatchScript(job);
// static string BatchManager_LSF::getDefaultServer();
// Methodes pour le controle des jobs
- virtual const JobId submitJob(const Job & job); // soumet un job au gestionnaire
+ virtual const JobId runJob(const Job & job); // soumet un job au gestionnaire
virtual void deleteJob(const JobId & jobid); // retire un job du gestionnaire
virtual JobInfo queryJob(const JobId & jobid); // renvoie l'etat du job
virtual bool isRunning(const JobId & jobid); // teste si un job est present en machine
}
// Method to submit a job to the batch manager
- const JobId BatchManager_Slurm::submitJob(const Job & job)
+ const JobId BatchManager_Slurm::runJob(const Job & job)
{
Parametre params = job.getParametre();
const string workDir = params[WORKDIR];
- // export input files on cluster
- exportInputFiles(job);
-
// build command file to submit the job and copy it on the server
string cmdFile = buildCommandFile(job);
virtual ~BatchManager_Slurm();
// Methods to control jobs
- virtual const JobId submitJob(const Job & job);
+ virtual const JobId runJob(const Job & job);
virtual void deleteJob(const JobId & jobid);
virtual JobInfo queryJob(const JobId & jobid);
}
// Method to submit a job to the batch manager
- const JobId BatchManager_Vishnu::submitJob(const Job & job)
+ const JobId BatchManager_Vishnu::runJob(const Job & job)
{
- // export input files on cluster
- exportInputFiles(job);
-
// build command file to submit the job
string cmdFile = buildCommandFile(job);
virtual ~BatchManager_Vishnu();
// Methods to control jobs
- virtual const JobId submitJob(const Job & job);
+ virtual const JobId runJob(const Job & job);
virtual void deleteJob(const JobId & jobid);
virtual JobInfo queryJob(const JobId & jobid);
virtual void importOutputFiles(const Job & job, const std::string directory);
