]> SALOME platform Git repositories - modules/kernel.git/blob - src/Launcher/Launcher_Job.cxx
Salome HOME
New access protocol rsync.
[modules/kernel.git] / src / Launcher / Launcher_Job.cxx
1 // Copyright (C) 2009-2017  CEA/DEN, EDF R&D, OPEN CASCADE
2 //
3 // This library is free software; you can redistribute it and/or
4 // modify it under the terms of the GNU Lesser General Public
5 // License as published by the Free Software Foundation; either
6 // version 2.1 of the License, or (at your option) any later version.
7 //
8 // This library is distributed in the hope that it will be useful,
9 // but WITHOUT ANY WARRANTY; without even the implied warranty of
10 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
11 // Lesser General Public License for more details.
12 //
13 // You should have received a copy of the GNU Lesser General Public
14 // License along with this library; if not, write to the Free Software
15 // Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
16 //
17 // See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
18 //
19
20 // Author: André RIBES - EDF R&D
21 //
22 //#define _DEBUG_
23 #include "Launcher_Job.hxx"
24 #include "Launcher.hxx"
25 #include <boost/filesystem.hpp>
26
27 #ifdef WITH_LIBBATCH
28 #include <libbatch/Constants.hxx>
29 #endif
30
31 #include <sstream>
32
33 Launcher::Job::Job()
34 {
35   _number = -1;
36   _state = "CREATED";
37   _launch_date = getLaunchDate();
38
39   _env_file = "";
40   _job_name = "";
41   _job_file = "";
42   _job_file_name = "";
43   _job_file_name_complete = "";
44   _pre_command = "";
45   _work_directory = "";
46   _local_directory = "";
47   _result_directory = "";
48   _maximum_duration = "";
49   _maximum_duration_in_second = -1;
50   _queue = "";
51   _partition = "";
52   _job_type = "";
53   _exclusive = false;
54   _mem_per_cpu = 0;
55
56   // Parameters for COORM
57   _launcher_file = "";
58   _launcher_args = "";
59
60 #ifdef WITH_LIBBATCH
61   _batch_job = new Batch::Job();
62 #endif
63 }
64
65 Launcher::Job::~Job()
66 {
67   LAUNCHER_MESSAGE("Deleting job number: " << _number);
68 #ifdef WITH_LIBBATCH
69   if (_batch_job)
70     delete _batch_job;
71 #endif
72 }
73
74 void
75 Launcher::Job::stopJob()
76 {
77   LAUNCHER_MESSAGE("Stop resquested for job number: " << _number);
78   setState("FAILED");
79 #ifdef WITH_LIBBATCH
80   if (_batch_job_id.getReference() != "undefined")
81   {
82     try
83     {
84       _batch_job_id.deleteJob();
85     }
86     catch (const Batch::GenericException &ex)
87     {
88       LAUNCHER_INFOS("WARNING: exception when stopping the job: " << ex.message);
89     }
90   }
91 #endif
92 }
93
94
95 void
96 Launcher::Job::removeJob()
97 {
98   LAUNCHER_MESSAGE("Removing job number: " << _number);
99 #ifdef WITH_LIBBATCH
100   if (_batch_job_id.getReference() != "undefined")
101   {
102     try
103     {
104       _batch_job_id.deleteJob();
105     }
106     catch (const Batch::GenericException &ex)
107     {
108       LAUNCHER_INFOS("WARNING: exception when removing the job: " << ex.message);
109     }
110   }
111 #endif
112 }
113
114 std::string
115 Launcher::Job::getJobType() const
116 {
117   return _job_type;
118 }
119
120 void
121 Launcher::Job::setJobName(const std::string & job_name)
122 {
123   _job_name = job_name;
124 }
125
126 std::string
127 Launcher::Job::getJobName() const
128 {
129   return _job_name;
130 }
131
132 void
133 Launcher::Job::setState(const std::string & state)
134 {
135   // State of a Job: CREATED, QUEUED, RUNNING, FINISHED, FAILED
136   if (state != "CREATED" &&
137       state != "IN_PROCESS" &&
138       state != "QUEUED" &&
139       state != "RUNNING" &&
140       state != "PAUSED" &&
141       state != "FINISHED" &&
142       state != "FAILED" &&
143       state != "ERROR")
144   {
145     throw LauncherException("Bad state, this state does not exist: " + state);
146   }
147   _state = state;
148 }
149
150 std::string
151 Launcher::Job::getState() const
152 {
153   return _state;
154 }
155
156 // Get names or ids of hosts assigned to the job
157 std::string
158 Launcher::Job::getAssignedHostnames()
159 {
160   return _assigned_hostnames;
161 }
162
163 void
164 Launcher::Job::setNumber(const int & number)
165 {
166   if (_number != -1)
167     std::cerr << "Launcher::Job::setNumber -- Job number was already defined, before: " << _number << " now: " << number << std::endl;
168   _number = number;
169 }
170
171 int
172 Launcher::Job::getNumber()
173 {
174   return _number;
175 }
176
177 void
178 Launcher::Job::setResourceDefinition(const ParserResourcesType & resource_definition)
179 {
180   // Check machine_definition
181   std::string user_name = "";
182   if (resource_definition.UserName == "")
183   {
184     user_name = getenv("USER");
185     if (user_name == "")
186       user_name = getenv("LOGNAME");
187     if (user_name == "")
188     {
189       std::string mess = "You must define a user name: into your resource description or with one of env variables USER/LOGNAME";
190       throw LauncherException(mess);
191     }
192   }
193   else
194     user_name = resource_definition.UserName;
195
196   _resource_definition = resource_definition;
197   _resource_definition.UserName = user_name;
198 }
199
200 ParserResourcesType
201 Launcher::Job::getResourceDefinition() const
202 {
203   return _resource_definition;
204 }
205
206 void
207 Launcher::Job::setJobFile(const std::string & job_file)
208 {
209   // Check job file
210   if (job_file == "")
211   {
212     std::string mess = "Empty Job File is forbidden !";
213     throw LauncherException(mess);
214   }
215
216   _job_file = job_file;
217   std::string::size_type p1 = _job_file.find_last_of("/");
218   std::string::size_type p2 = _job_file.find_last_of(".");
219   _job_file_name_complete = _job_file.substr(p1+1);
220   _job_file_name = _job_file.substr(p1+1,p2-p1-1);
221 }
222
223 std::string
224 Launcher::Job::getJobFile() const
225 {
226   return _job_file;
227 }
228 void
229 Launcher::Job::setEnvFile(const std::string & env_file)
230 {
231   _env_file = env_file;
232 }
233
234 std::string
235 Launcher::Job::getEnvFile() const
236 {
237   return _env_file;
238 }
239
240 void
241 Launcher::Job::setWorkDirectory(const std::string & work_directory)
242 {
243   _work_directory = work_directory;
244 }
245
246 void
247 Launcher::Job::setLocalDirectory(const std::string & local_directory)
248 {
249   _local_directory = local_directory;
250 }
251
252 void
253 Launcher::Job::setResultDirectory(const std::string & result_directory)
254 {
255   _result_directory = result_directory;
256 }
257
258 void
259 Launcher::Job::add_in_file(const std::string & file)
260 {
261   std::list<std::string>::iterator it = std::find(_in_files.begin(), _in_files.end(), file);
262   if (it == _in_files.end())
263     _in_files.push_back(file);
264   else
265     std::cerr << "Launcher::Job::add_in_file -- Warning file was already entered in in_files: " << file << std::endl;
266 }
267
268 void
269 Launcher::Job::add_out_file(const std::string & file)
270 {
271   std::list<std::string>::iterator it = std::find(_out_files.begin(), _out_files.end(), file);
272   if (it == _out_files.end())
273     _out_files.push_back(file);
274   else
275     std::cerr << "Launcher::Job::add_out_file -- Warning file was already entered in out_files: " << file << std::endl;
276 }
277
278 void
279 Launcher::Job::setMaximumDuration(const std::string & maximum_duration)
280 {
281   checkMaximumDuration(maximum_duration);
282   _maximum_duration_in_second = convertMaximumDuration(maximum_duration);
283   _maximum_duration = maximum_duration;
284 }
285
286 // For COORM
287 void
288 Launcher::Job::setLauncherFile(const std::string & launcher_file)
289 {
290         _launcher_file = launcher_file;
291 }
292 void
293 Launcher::Job::setLauncherArgs(const std::string & launcher_args)
294 {
295         _launcher_args = launcher_args;
296 }
297
298 void
299 Launcher::Job::setResourceRequiredParams(const resourceParams & resource_required_params)
300 {
301   checkResourceRequiredParams(resource_required_params);
302   _resource_required_params = resource_required_params;
303 }
304
305 void
306 Launcher::Job::setQueue(const std::string & queue)
307 {
308   _queue = queue;
309 }
310
311 void
312 Launcher::Job::setPartition(const std::string & partition)
313 {
314   _partition = partition;
315 }
316
317 void
318 Launcher::Job::setExclusive(bool exclusive)
319 {
320   _exclusive = exclusive;
321 }
322
323 void
324 Launcher::Job::setExclusiveStr(const std::string & exclusiveStr)
325 {
326   if (exclusiveStr == "true")
327     _exclusive = true;
328   else if (exclusiveStr == "false")
329     _exclusive = false;
330   else
331     throw LauncherException(std::string("Invalid boolean value for exclusive: ") + exclusiveStr);
332 }
333
334 void
335 Launcher::Job::setMemPerCpu(unsigned long mem_per_cpu)
336 {
337   _mem_per_cpu = mem_per_cpu;
338 }
339
340 void
341 Launcher::Job::setWCKey(const std::string & wckey)
342 {
343   _wckey = wckey;
344 }
345
346 void
347 Launcher::Job::setExtraParams(const std::string & extra_params)
348 {
349   _extra_params = extra_params;
350 }
351
352 void
353 Launcher::Job::setReference(const std::string & reference)
354 {
355   _reference = reference;
356 }
357
358 std::string
359 Launcher::Job::getWorkDirectory() const
360 {
361   return _work_directory;
362 }
363
364 std::string
365 Launcher::Job::getLocalDirectory() const
366 {
367   return _local_directory;
368 }
369
370 std::string
371 Launcher::Job::getResultDirectory() const
372 {
373   return _result_directory;
374 }
375
376 const std::list<std::string> &
377 Launcher::Job::get_in_files() const
378 {
379   return _in_files;
380 }
381
382 const std::list<std::string> &
383 Launcher::Job::get_out_files() const
384 {
385   return _out_files;
386 }
387
388 std::string
389 Launcher::Job::getMaximumDuration() const
390 {
391   return _maximum_duration;
392 }
393
394 // For COORM
395 std::string
396 Launcher::Job::getLauncherFile() const
397 {
398         return _launcher_file;
399 }
400 std::string
401 Launcher::Job::getLauncherArgs() const
402 {
403         return _launcher_args;
404 }
405
406 resourceParams
407 Launcher::Job::getResourceRequiredParams() const
408 {
409   return _resource_required_params;
410 }
411
412 std::string
413 Launcher::Job::getQueue() const
414 {
415   return _queue;
416 }
417
418 std::string
419 Launcher::Job::getPartition() const
420 {
421   return _partition;
422 }
423
424 bool
425 Launcher::Job::getExclusive() const
426 {
427   return _exclusive;
428 }
429
430 std::string
431 Launcher::Job::getExclusiveStr() const
432 {
433   return _exclusive ? "true" : "false";
434 }
435
436 unsigned long
437 Launcher::Job::getMemPerCpu() const
438 {
439   return _mem_per_cpu;
440 }
441
442 std::string
443 Launcher::Job::getWCKey() const
444 {
445   return _wckey;
446 }
447
448 std::string
449 Launcher::Job::getExtraParams() const
450 {
451   return _extra_params;
452 }
453
454 std::string
455 Launcher::Job::getReference() const
456 {
457   return _reference;
458 }
459
460 void
461 Launcher::Job::setPreCommand(const std::string & preCommand)
462 {
463   _pre_command = preCommand;
464 }
465
466 std::string
467 Launcher::Job::getPreCommand() const
468 {
469   return _pre_command;
470 }
471
472 void
473 Launcher::Job::checkMaximumDuration(const std::string & maximum_duration)
474 {
475   std::string result("");
476   std::string edt_value = maximum_duration;
477   std::size_t pos = edt_value.find(":");
478
479   if (edt_value != "") {
480     if (pos == edt_value.npos) {
481       throw LauncherException("[Launcher::Job::checkMaximumDuration] Error on definition: " + edt_value);
482     }
483     std::string begin_edt_value = edt_value.substr(0, pos);
484     std::string mid_edt_value = edt_value.substr(pos, 1);
485     std::string end_edt_value = edt_value.substr(pos + 1, edt_value.npos);
486
487     long value;
488     std::istringstream iss(begin_edt_value);
489     if (!(iss >> value)) {
490       result = "[Launcher::Job::checkExpectedDuration] Error on definition ! : " + edt_value;
491     }
492     else if (value < 0) {
493       result = "[Launcher::Job::checkExpectedDuration] Error on definition time is negative ! : " + value;
494     }
495     std::istringstream iss_2(end_edt_value);
496     if (!(iss_2 >> value)) {
497       result = "[Launcher::Job::checkExpectedDuration] Error on definition ! : " + edt_value;
498     }
499     else if (value < 0) {
500       result = "[Launcher::Job::checkExpectedDuration] Error on definition time is negative ! : " + value;
501     }
502     if (mid_edt_value != ":") {
503       result = "[Launcher::Job::checkExpectedDuration] Error on definition ! :" + edt_value;
504     }
505   }
506   if (result != "")
507     throw LauncherException(result);
508 }
509
510 void
511 Launcher::Job::checkResourceRequiredParams(const resourceParams & resource_required_params)
512 {
513   // nb_proc has be to > 0
514   if (resource_required_params.nb_proc <= 0)
515   {
516     std::string message("[Launcher::Job::checkResourceRequiredParams] proc number is not > 0 ! ");
517     throw LauncherException(message);
518   }
519 }
520
521 long
522 Launcher::Job::convertMaximumDuration(const std::string & edt)
523 {
524   long hh, mm, ret;
525
526   if( edt.size() == 0 )
527     return -1;
528
529   std::string::size_type pos = edt.find(":");
530   std::string h = edt.substr(0,pos);
531   std::string m = edt.substr(pos+1,edt.size()-pos+1);
532   std::istringstream issh(h);
533   issh >> hh;
534   std::istringstream issm(m);
535   issm >> mm;
536   ret = hh*60 + mm;
537   ret = ret * 60;
538
539   return ret;
540 }
541
542 std::string
543 Launcher::Job::getLaunchDate() const
544 {
545   time_t rawtime;
546   time(&rawtime);
547   std::string launch_date = ctime(&rawtime);
548   size_t i = 0 ;
549   for (;i < launch_date.size(); i++)
550     if (launch_date[i] == '/' ||
551         launch_date[i] == '-' ||
552         launch_date[i] == ':' ||
553         launch_date[i] == ' ')
554       launch_date[i] = '_';
555   launch_date.erase(--launch_date.end()); // Last character is a \n
556
557   return launch_date;
558 }
559
560 std::string
561 Launcher::Job::updateJobState()
562 {
563
564   if (_state != "FINISHED" &&
565       _state != "ERROR"    &&
566       _state != "FAILED")
567   {
568 #ifdef WITH_LIBBATCH
569     if (_batch_job_id.getReference() != "undefined")
570     {
571       // A batch manager has been affected to the job
572       Batch::JobInfo job_info = _batch_job_id.queryJob();
573       Batch::Parametre par = job_info.getParametre();
574       _state = par[Batch::STATE].str();
575       _assigned_hostnames = (par.find(Batch::ASSIGNEDHOSTNAMES) == par.end())?
576                             "" : par[Batch::ASSIGNEDHOSTNAMES].str();
577       LAUNCHER_MESSAGE("State received is: " << par[Batch::STATE].str());
578     }
579 #endif
580   }
581   return _state;
582 }
583
584 #ifdef WITH_LIBBATCH
585 Batch::Job *
586 Launcher::Job::getBatchJob()
587 {
588   update_job();
589   return _batch_job;
590 }
591
592 Batch::Parametre
593 Launcher::Job::common_job_params()
594 {
595   Batch::Parametre params;
596
597   params[Batch::NAME] = getJobName();
598   params[Batch::NBPROC] = _resource_required_params.nb_proc;
599   params[Batch::NBPROCPERNODE] = _resource_required_params.nb_proc_per_node;
600
601   if(_resource_required_params.nb_node > 0)
602     params[Batch::NBNODE] = _resource_required_params.nb_node;
603
604   // Memory in megabytes
605   if (_resource_required_params.mem_mb > 0)
606   {
607     params[Batch::MAXRAMSIZE] = _resource_required_params.mem_mb;
608   }
609   else if (_mem_per_cpu > 0)
610   {
611     params[Batch::MEMPERCPU] = (long)_mem_per_cpu;
612   }
613
614   // We define a default directory
615   if (_work_directory == "")
616   {
617     const size_t BUFSIZE = 32;
618     char date[BUFSIZE];
619     time_t curtime = time(NULL);
620     strftime(date, BUFSIZE, "%Y_%m_%d__%H_%M_%S", localtime(&curtime));
621     if(!_resource_definition.working_directory.empty())
622     {
623       std::string date_dir = std::string("/job_") + date;
624       std::ostringstream str_pid;
625       str_pid << ::getpid();
626       std::string job_dir = date_dir + "-" + str_pid.str();
627
628       _work_directory = _resource_definition.working_directory + job_dir;
629     }
630     else
631     {
632       _work_directory = std::string("/$HOME/Batch/workdir_");
633       _work_directory += date;
634     }
635   }
636   params[Batch::WORKDIR] = _work_directory;
637   std::string libbatch_pre_command("");
638   if(!_pre_command.empty())
639   {
640     boost::filesystem::path pre_command_path(_pre_command);
641     libbatch_pre_command += "./" + pre_command_path.filename().string();
642   }
643   params[Batch::PREPROCESS] = libbatch_pre_command;
644
645   // Parameters for COORM
646   params[Batch::LAUNCHER_FILE] = _launcher_file;
647   params[Batch::LAUNCHER_ARGS] = _launcher_args;
648
649   // If result_directory is not defined, we use HOME environnement
650   if (_result_directory == "")
651     _result_directory = getenv("HOME");
652
653   // _in_files
654   std::list<std::string> in_files(_in_files);
655   in_files.push_back(_job_file);
656   if (_env_file != "")
657           in_files.push_back(_env_file);
658   if(!_pre_command.empty())
659      in_files.push_back(_pre_command);
660   for(std::list<std::string>::iterator it = in_files.begin(); it != in_files.end(); it++)
661   {
662     std::string file = *it;
663
664     // local file -> If file is not an absolute path, we apply _local_directory
665     std::string local_file;
666     if (file.substr(0, 1) == std::string("/"))
667       local_file = file;
668     else if (file.substr(0, 1) == std::string("-")) // using rsync options
669       local_file = file;
670     else
671 #ifndef WIN32
672       // '/./' is used by rsync to find the root of the relative path
673       // /a/b/./c/f -> _working_directory/c/f
674       local_file = _local_directory + "/./" + file;
675 #else
676       local_file = file;
677 #endif
678
679     // remote file -> get only file name from in_files
680     std::string remote_file = _work_directory + "/";
681
682     params[Batch::INFILE] += Batch::Couple(local_file, remote_file);
683   }
684
685   // _out_files
686   for(std::list<std::string>::iterator it = _out_files.begin(); it != _out_files.end(); it++)
687   {
688     std::string file = *it;
689     // remote file -> If file is not an absolute path, we apply _work_directory
690     std::string remote_file;
691     std::string local_file;
692     if (file.substr(0, 1) == std::string("/"))
693     {
694       remote_file = file;
695       size_t found = file.find_last_of("/");
696       local_file = file.substr(found+1);
697     }
698     else if (file.substr(0, 1) == std::string("-")) // using rsync options
699     {
700       remote_file = file;
701       local_file = "";
702     }
703     else
704     {
705       // '/./' is used by rsync to find the root of the relative path
706       remote_file = _work_directory + "/./" + file;
707       local_file = "";
708     }
709
710     params[Batch::OUTFILE] += Batch::Couple(local_file, remote_file);
711   }
712
713   // Time
714   if (_maximum_duration_in_second != -1)
715     params[Batch::MAXWALLTIME] = _maximum_duration_in_second / 60;
716
717   // Queue
718   if (_queue != "")
719     params[Batch::QUEUE] = _queue;
720
721   // Partition
722   if (_partition != "")
723     params[Batch::PARTITION] = _partition;
724
725   // Exclusive
726   if (getExclusive())
727     params[Batch::EXCLUSIVE] = true;
728
729   // WC Key
730   if (_wckey != "")
731     params[Batch::WCKEY] = _wckey;
732
733   // Extra params
734   if (_extra_params != "")
735     params[Batch::EXTRAPARAMS] = _extra_params;
736
737   // Specific parameters
738   std::map<std::string, std::string>::iterator it = _specific_parameters.find("LoalLevelerJobType");
739   if (it != _specific_parameters.end())
740     params["LL_JOBTYPE"] = it->second;
741   return params;
742 }
743
744 void
745 Launcher::Job::setBatchManagerJobId(Batch::JobId batch_manager_job_id)
746 {
747   _batch_job_id = batch_manager_job_id;
748 }
749
750 Batch::JobId
751 Launcher::Job::getBatchManagerJobId() const
752 {
753   return _batch_job_id;
754 }
755 #endif
756
757 void
758 Launcher::Job::addSpecificParameter(const std::string & name,
759                                       const std::string & value)
760 {
761   _specific_parameters[name] = value;
762 }
763
764 const std::map<std::string, std::string> &
765 Launcher::Job::getSpecificParameters() const
766 {
767   return _specific_parameters;
768 }
769
770 void
771 Launcher::Job::checkSpecificParameters()
772 {
773 }