]> SALOME platform Git repositories - tools/libbatch.git/blob - src/Local/Batch_BatchManager_Local.cxx
Salome HOME
A little fix for compilation on old gcc (i.e. 3.3.5)
[tools/libbatch.git] / src / Local / Batch_BatchManager_Local.cxx
1 //  Copyright (C) 2007-2008  CEA/DEN, EDF R&D, OPEN CASCADE
2 //
3 //  Copyright (C) 2003-2007  OPEN CASCADE, EADS/CCR, LIP6, CEA/DEN,
4 //  CEDRAT, EDF R&D, LEG, PRINCIPIA R&D, BUREAU VERITAS
5 //
6 //  This library is free software; you can redistribute it and/or
7 //  modify it under the terms of the GNU Lesser General Public
8 //  License as published by the Free Software Foundation; either
9 //  version 2.1 of the License.
10 //
11 //  This library is distributed in the hope that it will be useful,
12 //  but WITHOUT ANY WARRANTY; without even the implied warranty of
13 //  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14 //  Lesser General Public License for more details.
15 //
16 //  You should have received a copy of the GNU Lesser General Public
17 //  License along with this library; if not, write to the Free Software
18 //  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
19 //
20 //  See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
21 //
22 /*
23 * BatchManager_Local.cxx :
24 *
25 * Auteur : Ivan DUTKA-MALEN - EDF R&D
26 * Mail   : mailto:ivan.dutka-malen@der.edf.fr
27 * Date   : Thu Nov  6 10:17:22 2003
28 * Projet : Salome 2
29 *
30 * Refactored by Renaud Barate (EDF R&D) in September 2009 to use
31 * CommunicationProtocol classes and merge Local_SH, Local_RSH and Local_SSH batch
32 * managers.
33 */
34
35 #include <iostream>
36 #include <fstream>
37 #include <sstream>
38 #include <cstdlib>
39 #include <limits.h>
40
41 #include <sys/types.h>
42 #ifdef WIN32
43 #include <direct.h>
44 #include "Batch_RunTimeException.hxx"
45 #else
46 #include <sys/wait.h>
47 #include <unistd.h>
48 #endif
49 #include <ctime>
50 #include <pthread.h>
51 #include <signal.h>
52 #include <errno.h>
53 #include <string.h>
54 #include "Batch_IOMutex.hxx"
55 #include "Batch_BatchManager_Local.hxx"
56
57 using namespace std;
58
59 namespace Batch {
60
61
62   // Constructeur
63   BatchManager_Local::BatchManager_Local(const FactBatchManager * parent, const char * host,
64                                          CommunicationProtocolType protocolType)
65     : BatchManager(parent, host), _connect(0), _threads_mutex(), _threads(),
66       _protocol(CommunicationProtocol::getInstance(protocolType)),
67       _thread_id_id_association_mutex(), _thread_id_id_association_cond(), _thread_id_id_association()
68   {
69     pthread_mutex_init(&_threads_mutex, NULL);
70     pthread_mutex_init(&_thread_id_id_association_mutex, NULL);
71     pthread_cond_init(&_thread_id_id_association_cond, NULL);
72   }
73
74   // Destructeur
75   BatchManager_Local::~BatchManager_Local()
76   {
77     pthread_mutex_destroy(&_threads_mutex);
78     pthread_mutex_destroy(&_thread_id_id_association_mutex);
79     pthread_cond_destroy(&_thread_id_id_association_cond);
80   }
81
82   const CommunicationProtocol & BatchManager_Local::getProtocol() const
83   {
84     return _protocol;
85   }
86
87   // Methode pour le controle des jobs : soumet un job au gestionnaire
88   const JobId BatchManager_Local::submitJob(const Job & job)
89   {
90     Job_Local jobLocal = job;
91
92     pthread_t thread_id = submit(jobLocal);
93
94     ostringstream oss;
95     oss << getIdByThread_id(thread_id);
96
97     JobId id(this, oss.str());
98
99     return id;
100   }
101
102   // Methode pour le controle des jobs : retire un job du gestionnaire
103   void BatchManager_Local::deleteJob(const JobId & jobid)
104   {
105     Id id;
106
107     istringstream iss(jobid.getReference());
108     iss >> id;
109
110     // On retrouve le thread_id du thread
111     pthread_t thread_id;
112
113     // @@@ --------> SECTION CRITIQUE <-------- @@@
114     pthread_mutex_lock(&_threads_mutex);
115     if (_threads.find(id) != _threads.end())
116       thread_id = _threads[id].thread_id;
117     pthread_mutex_unlock(&_threads_mutex);
118     // @@@ --------> SECTION CRITIQUE <-------- @@@
119
120     cancel(thread_id);
121   }
122
123   // Methode pour le controle des jobs : suspend un job en file d'attente
124   void BatchManager_Local::holdJob(const JobId & jobid)
125   {
126     Id id;
127     istringstream iss(jobid.getReference());
128     iss >> id;
129
130     UNDER_LOCK( cout << "BatchManager is sending HOLD command to the thread " << id << endl );
131
132     // On introduit une commande dans la queue du thread
133     // @@@ --------> SECTION CRITIQUE <-------- @@@
134     pthread_mutex_lock(&_threads_mutex);
135     if (_threads.find(id) != _threads.end())
136       _threads[id].command_queue.push(HOLD);
137     pthread_mutex_unlock(&_threads_mutex);
138     // @@@ --------> SECTION CRITIQUE <-------- @@@
139   }
140
141   // Methode pour le controle des jobs : relache un job suspendu
142   void BatchManager_Local::releaseJob(const JobId & jobid)
143   {
144     Id id;
145     istringstream iss(jobid.getReference());
146     iss >> id;
147
148     UNDER_LOCK( cout << "BatchManager is sending RELEASE command to the thread " << id << endl );
149
150     // On introduit une commande dans la queue du thread
151     // @@@ --------> SECTION CRITIQUE <-------- @@@
152     pthread_mutex_lock(&_threads_mutex);
153     if (_threads.find(id) != _threads.end())
154       _threads[id].command_queue.push(RELEASE);
155     pthread_mutex_unlock(&_threads_mutex);
156     // @@@ --------> SECTION CRITIQUE <-------- @@@
157   }
158
159
160   // Methode pour le controle des jobs : modifie un job en file d'attente
161   void BatchManager_Local::alterJob(const JobId & jobid, const Parametre & param, const Environnement & env)
162   {
163   }
164
165   // Methode pour le controle des jobs : modifie un job en file d'attente
166   void BatchManager_Local::alterJob(const JobId & jobid, const Parametre & param)
167   {
168     alterJob(jobid, param, Environnement());
169   }
170
171   // Methode pour le controle des jobs : modifie un job en file d'attente
172   void BatchManager_Local::alterJob(const JobId & jobid, const Environnement & env)
173   {
174     alterJob(jobid, Parametre(), env);
175   }
176
177
178
179   // Methode pour le controle des jobs : renvoie l'etat du job
180   JobInfo BatchManager_Local::queryJob(const JobId & jobid)
181   {
182     Id id;
183     istringstream iss(jobid.getReference());
184     iss >> id;
185
186     Parametre param;
187     Environnement env;
188
189     //UNDER_LOCK( cout << "JobInfo BatchManager_Local::queryJob(const JobId & jobid) : AVANT section critique" << endl );
190     // @@@ --------> SECTION CRITIQUE <-------- @@@
191     pthread_mutex_lock(&_threads_mutex);
192     param = _threads[id].param;
193     env   = _threads[id].env;
194     pthread_mutex_unlock(&_threads_mutex);
195     // @@@ --------> SECTION CRITIQUE <-------- @@@
196     //UNDER_LOCK( cout << "JobInfo BatchManager_Local::queryJob(const JobId & jobid) : APRES section critique" << endl );
197
198     JobInfo_Local ji(param, env);
199     return ji;
200   }
201
202
203
204   // Methode pour le controle des jobs : teste si un job est present en machine
205   bool BatchManager_Local::isRunning(const JobId & jobid)
206   {
207     Id id;
208     istringstream iss(jobid.getReference());
209     iss >> id;
210
211     Status status;
212
213     //UNDER_LOCK( cout << "JobInfo BatchManager_Local::queryJob(const JobId & jobid) : AVANT section critique" << endl );
214     // @@@ --------> SECTION CRITIQUE <-------- @@@
215     pthread_mutex_lock(&_threads_mutex);
216     status = _threads[id].status;
217     pthread_mutex_unlock(&_threads_mutex);
218     // @@@ --------> SECTION CRITIQUE <-------- @@@
219     //UNDER_LOCK( cout << "JobInfo BatchManager_Local::queryJob(const JobId & jobid) : APRES section critique" << endl );
220
221     return (status == RUNNING);
222   }
223
224
225   // Methode d'execution d'un job
226   pthread_t BatchManager_Local::submit(const Job_Local & job)
227   {
228     // L'id du thread a creer
229     pthread_t thread_id =
230 #ifdef WIN32
231     {0,0};
232 #else
233       0;
234 #endif
235
236     // Les attributs du thread a sa creation
237     pthread_attr_t thread_attr;
238     pthread_attr_init(&thread_attr);
239     pthread_attr_setdetachstate(&thread_attr, PTHREAD_CREATE_DETACHED);
240
241     ThreadAdapter * p_ta = new ThreadAdapter(*this, job);
242
243     // Creation du thread qui va executer la commande systeme qu'on lui passe
244     int rc = pthread_create(&thread_id,
245       &thread_attr,
246       &ThreadAdapter::run,
247       static_cast<void *>(p_ta));
248     if (rc) {
249     }
250
251     // Liberation des zones memoire maintenant inutiles occupees par les attributs du thread
252     pthread_attr_destroy(&thread_attr);
253
254     return thread_id;
255   }
256
257
258   // Methode de destruction d'un job
259   void BatchManager_Local::cancel(pthread_t thread_id)
260   {
261     pthread_cancel(thread_id);
262   }
263
264
265   vector<string> BatchManager_Local::exec_command(const Parametre & param) const
266   {
267     ostringstream exec_sub_cmd;
268
269 #ifdef WIN32
270     char drive[_MAX_DRIVE];
271     _splitpath(string(param[WORKDIR]).c_str(), drive, NULL, NULL, NULL);
272     if (strlen(drive) > 0) exec_sub_cmd << drive << " && ";
273 #endif
274
275     exec_sub_cmd << "cd " << param[WORKDIR] << " && " << param[EXECUTABLE];
276
277     if (param.find(ARGUMENTS) != param.end()) {
278       Versatile V = param[ARGUMENTS];
279       for(Versatile::const_iterator it=V.begin(); it!=V.end(); it++) {
280         StringType argt = * static_cast<StringType *>(*it);
281         string     arg  = argt;
282         exec_sub_cmd << " " << arg;
283       }
284     }
285
286     string user;
287     Parametre::const_iterator it = param.find(USER);
288     if (it != param.end()) {
289       user = string(it->second);
290     }
291
292     return _protocol.getExecCommandArgs(exec_sub_cmd.str(), param[EXECUTIONHOST], user);
293   }
294
295
296   // Fabrique un identifiant unique pour les threads puisque le thread_id n'est pas unique
297   // au cours du temps (il peut etre reutilise lorsqu'un thread se termine)
298   // ATTENTION : cette methode est uniquement protegee par la section critique de l'association
299   // Thread_id / Id (_thread_id_id_association_mutex)
300   BatchManager_Local::Id BatchManager_Local::nextId()
301   {
302     static Id id = 0;
303     Id nextId = id++;
304     //UNDER_LOCK( cout << "BatchManager_Local::Id BatchManager_Local::nextId() : Id = " << nextId << endl );
305     return nextId;
306   }
307
308
309   // Retourne l'Id enregistre dans l'association Thread_id / Id et le detruit immediatement
310   BatchManager_Local::Id BatchManager_Local::getIdByThread_id(pthread_t thread_id)
311   {
312     // @@@ --------> SECTION CRITIQUE <-------- @@@
313     pthread_mutex_lock(&_thread_id_id_association_mutex);
314     bool threadIdFound = false;
315     std::list<struct ThreadIdIdAssociation>::iterator it;
316     while (!threadIdFound) {
317       for (it = _thread_id_id_association.begin() ;
318            it != _thread_id_id_association.end() && !pthread_equal(it->threadId, thread_id) ;
319            it++);
320       if (it == _thread_id_id_association.end())
321         pthread_cond_wait(&_thread_id_id_association_cond, &_thread_id_id_association_mutex);
322       else
323         threadIdFound = true;
324     }
325
326     Id id = it->id;
327     _thread_id_id_association.erase(it);
328
329     pthread_mutex_unlock(&_thread_id_id_association_mutex);
330     // @@@ --------> SECTION CRITIQUE <-------- @@@
331
332     //UNDER_LOCK( cout << "BatchManager_Local::Id BatchManager_Local::getIdByThread_id(pthread_t thread_id) : Id = " << id << " - thread_id = " << thread_id << endl );
333     return id;
334   }
335
336
337   // Associe un Thread_id a un Id nouvellement cree
338   BatchManager_Local::Id BatchManager_Local::registerThread_id(pthread_t thread_id)
339   {
340     Id id = -1;
341
342     // @@@ --------> SECTION CRITIQUE <-------- @@@
343     pthread_mutex_lock(&_thread_id_id_association_mutex);
344     std::list<struct ThreadIdIdAssociation>::iterator it;
345     for (it = _thread_id_id_association.begin() ;
346          it != _thread_id_id_association.end() && !pthread_equal(it->threadId, thread_id) ;
347          it++);
348     if (it == _thread_id_id_association.end()) {
349       struct ThreadIdIdAssociation newAssociation;
350       id = newAssociation.id = nextId();
351       newAssociation.threadId = thread_id;
352       _thread_id_id_association.push_back(newAssociation);
353       pthread_cond_signal(&_thread_id_id_association_cond);
354
355     } else {
356       UNDER_LOCK( cerr << "ERROR : Pthread Inconstency. Two threads own the same thread_id." << endl );
357     }
358     pthread_mutex_unlock(&_thread_id_id_association_mutex);
359     // @@@ --------> SECTION CRITIQUE <-------- @@@
360
361     //UNDER_LOCK( cout << "BatchManager_Local::Id BatchManager_Local::registerThread_id(pthread_t thread_id) : Id = " << id << " - thread_id = " << thread_id << endl );
362     return id;
363   }
364
365
366   // Constructeur de la classe ThreadAdapter
367   BatchManager_Local::ThreadAdapter::ThreadAdapter(BatchManager_Local & bm, const Job_Local & job) :
368   _bm(bm), _job(job)
369   {
370     // Nothing to do
371   }
372
373
374
375   // Methode d'execution du thread
376   void * BatchManager_Local::ThreadAdapter::run(void * arg)
377   {
378 #ifndef WIN32
379     // On bloque tous les signaux pour ce thread
380     sigset_t setmask;
381     sigfillset(&setmask);
382     pthread_sigmask(SIG_BLOCK, &setmask, NULL);
383 #endif
384
385     // On autorise la terminaison differee du thread
386     // (ces valeurs sont les valeurs par defaut mais on les force par precaution)
387     pthread_setcancelstate(PTHREAD_CANCEL_ENABLE,  NULL);
388     pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL);
389
390     // On enregistre la fonction de suppression du fils en cas d'arret du thread
391     // Cette fontion sera automatiquement appelee lorsqu'une demande d'annulation
392     // sera prise en compte par pthread_testcancel()
393     Process child;
394     pthread_cleanup_push(BatchManager_Local::kill_child_on_exit, static_cast<void *> (&child));
395     pthread_cleanup_push(BatchManager_Local::delete_on_exit, arg);
396
397     ThreadAdapter * p_ta = static_cast<ThreadAdapter *>(arg);
398
399
400
401
402     // Le code retour cumule (ORed) de tous les appels
403     // Nul en cas de reussite de l'ensemble des operations
404     int rc = 0;
405
406     // Cette table contient la liste des fichiers a detruire a la fin du processus
407     std::vector<string> files_to_delete;
408
409
410
411     // On copie les fichiers d'entree pour le fils
412     const Parametre param   = p_ta->_job.getParametre();
413     Parametre::const_iterator it;
414
415     // On initialise la variable workdir a la valeur du Current Working Directory
416     char * cwd =
417 #ifdef WIN32
418       _getcwd(NULL, 0);
419 #else
420       new char [PATH_MAX];
421     getcwd(cwd, PATH_MAX);
422 #endif
423     string workdir = cwd;
424     delete [] cwd;
425
426     if ( (it = param.find(WORKDIR)) != param.end() ) {
427       workdir = static_cast<string>( (*it).second );
428     }
429
430     string executionhost = string(param[EXECUTIONHOST]);
431     string user;
432     if ( (it = param.find(USER)) != param.end() ) {
433       user = string(it->second);
434     }
435
436     if ( (it = param.find(INFILE)) != param.end() ) {
437       Versatile V = (*it).second;
438       Versatile::iterator Vit;
439
440       for(Vit=V.begin(); Vit!=V.end(); Vit++) {
441         CoupleType cpt  = *static_cast< CoupleType * >(*Vit);
442         Couple cp       = cpt;
443         string local    = cp.getLocal();
444         string remote   = cp.getRemote();
445
446         std::cerr << workdir << std::endl;
447         std::cerr << remote << std::endl;
448
449         int status = p_ta->getBatchManager().getProtocol().copyFile(local, "", "",
450                                                                     workdir + "/" + remote,
451                                                                     executionhost, user);
452         if (status) {
453           // Echec de la copie
454           rc |= 1;
455         } else {
456           // On enregistre le fichier comme etant a detruire
457           files_to_delete.push_back(workdir + "/" + remote);
458         }
459
460       }
461     }
462
463
464
465
466     // On forke/exec un nouveau process pour pouvoir controler le fils
467     // (plus finement qu'avec un appel system)
468     // int rc = system(commande.c_str());
469     //char *const parmList[] = {"/usr/bin/ssh", "localhost", "-l", "aribes", "sleep 10 && echo end", NULL};
470     //execv("/usr/bin/ssh", parmList);
471 #ifdef WIN32
472     child = p_ta->launchWin32ChildProcess();
473     p_ta->pere(child);
474 #else
475     child = fork();
476     if (child < 0) { // erreur
477       UNDER_LOCK( cerr << "Fork impossible (rc=" << child << ")" << endl );
478
479     } else if (child > 0) { // pere
480       p_ta->pere(child);
481
482     } else { // fils
483       p_ta->fils();
484     }
485 #endif
486
487
488     // On copie les fichiers de sortie du fils
489     if ( (it = param.find(OUTFILE)) != param.end() ) {
490       Versatile V = (*it).second;
491       Versatile::iterator Vit;
492
493       for(Vit=V.begin(); Vit!=V.end(); Vit++) {
494         CoupleType cpt  = *static_cast< CoupleType * >(*Vit);
495         Couple cp       = cpt;
496         string local    = cp.getLocal();
497         string remote   = cp.getRemote();
498
499         int status = p_ta->getBatchManager().getProtocol().copyFile(workdir + "/" + remote,
500                                                                     executionhost, user,
501                                                                     local, "", "");
502         if (status) {
503           // Echec de la copie
504           rc |= 1;
505         } else {
506           // On enregistre le fichier comme etant a detruire
507           files_to_delete.push_back(workdir + "/" + remote);
508         }
509
510       }
511     }
512
513     // On efface les fichiers d'entree et de sortie du fils si les copies precedentes ont reussi
514     // ou si la creation du fils n'a pu avoir lieu
515     if ( (rc == 0) || (child < 0) ) {
516       std::vector<string>::const_iterator it;
517       for(it=files_to_delete.begin(); it!=files_to_delete.end(); it++) {
518         p_ta->getBatchManager().getProtocol().removeFile(*it, executionhost, user);
519 /*        string remove_cmd = p_ta->getBatchManager().remove_command(user, executionhost, *it);
520         UNDER_LOCK( cout << "Removing : " << remove_cmd << endl );
521 #ifdef WIN32
522         remove_cmd = string("\"") + remove_cmd + string("\"");
523 #endif
524         system(remove_cmd.c_str());*/
525       }
526     }
527
528
529
530     // On retire la fonction de nettoyage de la memoire
531     pthread_cleanup_pop(0);
532
533     // On retire la fonction de suppression du fils
534     pthread_cleanup_pop(0);
535
536
537
538     // On invoque la fonction de nettoyage de la memoire
539     delete_on_exit(arg);
540
541     UNDER_LOCK( cout << "Father is leaving" << endl );
542     pthread_exit(NULL);
543     return NULL;
544   }
545
546
547
548
549   void BatchManager_Local::ThreadAdapter::pere(Process child)
550   {
551     time_t child_starttime = time(NULL);
552
553     // On enregistre le fils dans la table des threads
554     pthread_t thread_id = pthread_self();
555     Id id = _bm.registerThread_id(thread_id);
556
557     Parametre param   = _job.getParametre();
558     Environnement env = _job.getEnvironnement();
559
560     ostringstream thread_id_sst;
561     thread_id_sst << id;
562     param[ID]         = thread_id_sst.str();
563     param[STATE]      = "Running";
564 #ifndef WIN32
565     param[PID]        = child;
566 #endif
567
568     // @@@ --------> SECTION CRITIQUE <-------- @@@
569     pthread_mutex_lock(&_bm._threads_mutex);
570     _bm._threads[id].thread_id = thread_id;
571 #ifndef WIN32
572     _bm._threads[id].pid       = child;
573 #endif
574     _bm._threads[id].status    = RUNNING;
575     _bm._threads[id].param     = param;
576     _bm._threads[id].env       = env;
577     _bm._threads[id].command_queue.push(NOP);
578     pthread_mutex_unlock(&_bm._threads_mutex);
579     // @@@ --------> SECTION CRITIQUE <-------- @@@
580
581
582
583
584     // on boucle en attendant que le fils ait termine
585     while (1) {
586 #ifdef WIN32
587       DWORD exitCode;
588       BOOL res = GetExitCodeProcess(child, &exitCode);
589       if (exitCode != STILL_ACTIVE) {
590         pthread_mutex_lock(&_bm._threads_mutex);
591         _bm._threads[id].status       = DONE;
592         _bm._threads[id].param[STATE] = "Done";
593         pthread_mutex_unlock(&_bm._threads_mutex);
594         // @@@ --------> SECTION CRITIQUE <-------- @@@
595         UNDER_LOCK( cout << "Father sees his child is DONE: exit code = " << exitCode << endl );
596         break;
597       }
598 #else
599       int child_rc = 0;
600       pid_t child_wait_rc = waitpid(child, &child_rc, WNOHANG /* | WUNTRACED */);
601       if (child_wait_rc > 0) {
602          UNDER_LOCK( cout << "Status is: " << WIFEXITED( child_rc) << endl);
603          UNDER_LOCK( cout << "Status is: " << WEXITSTATUS( child_rc) << endl);
604          UNDER_LOCK( cout << "Status is: " << WIFSIGNALED( child_rc) << endl);
605          UNDER_LOCK( cout << "Status is: " << WTERMSIG( child_rc) << endl);
606          UNDER_LOCK( cout << "Status is: " << WCOREDUMP( child_rc) << endl);
607          UNDER_LOCK( cout << "Status is: " << WIFSTOPPED( child_rc) << endl);
608          UNDER_LOCK( cout << "Status is: " << WSTOPSIG( child_rc) << endl);
609 #ifdef WIFCONTINUED
610          UNDER_LOCK( cout << "Status is: " << WIFCONTINUED( child_rc) << endl); // not compilable on sarge
611 #endif
612         if (WIFSTOPPED(child_rc)) {
613           // NOTA : pour rentrer dans cette section, il faut que le flag WUNTRACED
614           // soit positionne dans l'appel a waitpid ci-dessus. Ce flag est couramment
615           // desactive car s'il est possible de detecter l'arret d'un process, il est
616           // plus difficile de detecter sa reprise.
617
618           // Le fils est simplement stoppe
619           // @@@ --------> SECTION CRITIQUE <-------- @@@
620           pthread_mutex_lock(&_bm._threads_mutex);
621           _bm._threads[id].status       = STOPPED;
622           _bm._threads[id].param[STATE] = "Stopped";
623           pthread_mutex_unlock(&_bm._threads_mutex);
624           // @@@ --------> SECTION CRITIQUE <-------- @@@
625           UNDER_LOCK( cout << "Father sees his child is STOPPED : " << child_wait_rc << endl );
626
627         }
628         else {
629           // Le fils est termine, on sort de la boucle et du if englobant
630           // @@@ --------> SECTION CRITIQUE <-------- @@@
631           pthread_mutex_lock(&_bm._threads_mutex);
632           _bm._threads[id].status       = DONE;
633           _bm._threads[id].param[STATE] = "Done";
634           pthread_mutex_unlock(&_bm._threads_mutex);
635           // @@@ --------> SECTION CRITIQUE <-------- @@@
636           UNDER_LOCK( cout << "Father sees his child is DONE : " << child_wait_rc << " (child_rc=" << (WIFEXITED(child_rc) ? WEXITSTATUS(child_rc) : -1) << ")" << endl );
637           break;
638         }
639       }
640       else if (child_wait_rc == -1) {
641         // Le fils a disparu ...
642         // @@@ --------> SECTION CRITIQUE <-------- @@@
643         pthread_mutex_lock(&_bm._threads_mutex);
644         _bm._threads[id].status       = DEAD;
645         _bm._threads[id].param[STATE] = "Dead";
646         pthread_mutex_unlock(&_bm._threads_mutex);
647         // @@@ --------> SECTION CRITIQUE <-------- @@@
648         UNDER_LOCK( cout << "Father sees his child is DEAD : " << child_wait_rc << " (Reason : " << strerror(errno) << ")" << endl );
649         break;
650       }
651 #endif
652
653       // On teste si le thread doit etre detruit
654       pthread_testcancel();
655
656
657
658       // On regarde si le fils n'a pas depasse son temps (wallclock time)
659       time_t child_currenttime = time(NULL);
660       time_t child_elapsedtime = child_currenttime - child_starttime;
661       if (param.find(MAXWALLTIME) != param.end()) {
662         int maxwalltime = param[MAXWALLTIME];
663         //        cout << "child_starttime          = " << child_starttime        << endl
664         //             << "child_currenttime        = " << child_currenttime      << endl
665         //             << "child_elapsedtime        = " << child_elapsedtime      << endl
666         //             << "maxwalltime              = " << maxwalltime            << endl
667         //             << "int(maxwalltime * 1.1)   = " << int(maxwalltime * 1.1) << endl;
668         if (child_elapsedtime > int(maxwalltime * 1.1) ) { // On se donne 10% de marge avant le KILL
669           UNDER_LOCK( cout << "Father is sending KILL command to the thread " << id << endl );
670           // On introduit une commande dans la queue du thread
671           // @@@ --------> SECTION CRITIQUE <-------- @@@
672           pthread_mutex_lock(&_bm._threads_mutex);
673           if (_bm._threads.find(id) != _bm._threads.end())
674             _bm._threads[id].command_queue.push(KILL);
675           pthread_mutex_unlock(&_bm._threads_mutex);
676           // @@@ --------> SECTION CRITIQUE <-------- @@@
677
678
679         } else if (child_elapsedtime > maxwalltime ) {
680           UNDER_LOCK( cout << "Father is sending TERM command to the thread " << id << endl );
681           // On introduit une commande dans la queue du thread
682           // @@@ --------> SECTION CRITIQUE <-------- @@@
683           pthread_mutex_lock(&_bm._threads_mutex);
684           if (_bm._threads.find(id) != _bm._threads.end())
685             _bm._threads[id].command_queue.push(TERM);
686           pthread_mutex_unlock(&_bm._threads_mutex);
687           // @@@ --------> SECTION CRITIQUE <-------- @@@
688         }
689       }
690
691
692
693       // On regarde s'il y a quelque chose a faire dans la queue de commande
694       // @@@ --------> SECTION CRITIQUE <-------- @@@
695       pthread_mutex_lock(&_bm._threads_mutex);
696       if (_bm._threads.find(id) != _bm._threads.end()) {
697         while (_bm._threads[id].command_queue.size() > 0) {
698           Commande cmd = _bm._threads[id].command_queue.front();
699           _bm._threads[id].command_queue.pop();
700
701           switch (cmd) {
702     case NOP:
703       UNDER_LOCK( cout << "Father does nothing to his child" << endl );
704       break;
705 #ifndef WIN32
706     case HOLD:
707       UNDER_LOCK( cout << "Father is sending SIGSTOP signal to his child" << endl );
708       kill(child, SIGSTOP);
709       break;
710
711     case RELEASE:
712       UNDER_LOCK( cout << "Father is sending SIGCONT signal to his child" << endl );
713       kill(child, SIGCONT);
714       break;
715
716     case TERM:
717       UNDER_LOCK( cout << "Father is sending SIGTERM signal to his child" << endl );
718       kill(child, SIGTERM);
719       break;
720
721     case KILL:
722       UNDER_LOCK( cout << "Father is sending SIGKILL signal to his child" << endl );
723       kill(child, SIGKILL);
724       break;
725 #endif
726     case ALTER:
727       break;
728
729     default:
730       break;
731           }
732         }
733
734       }
735       pthread_mutex_unlock(&_bm._threads_mutex);
736       // @@@ --------> SECTION CRITIQUE <-------- @@@
737
738       // On fait une petite pause pour ne pas surcharger inutilement le processeur
739 #ifdef WIN32
740       Sleep(1000);
741 #else
742       sleep(1);
743 #endif
744
745     }
746
747   }
748
749
750
751 #ifndef WIN32
752
753   void BatchManager_Local::ThreadAdapter::fils()
754   {
755     Parametre param = _job.getParametre();
756     Parametre::iterator it;
757
758       //char *const parmList[] = {"/usr/bin/ssh", "localhost", "-l", "aribes", "sleep 1 && echo end", NULL};
759       //int result = execv("/usr/bin/ssh", parmList);
760       //UNDER_LOCK( cout << "*** debug_command = " << result << endl );
761       //UNDER_LOCK( cout << "*** debug_command = " << strerror(errno) << endl );
762     try {
763
764       // EXECUTABLE is MANDATORY, if missing, we exit with failure notification
765       vector<string> command;
766       if (param.find(EXECUTABLE) != param.end()) {
767         command = _bm.exec_command(param);
768       } else exit(1);
769
770       // Build the argument array argv from the command
771       char ** argv = new char * [command.size() + 1];
772       string comstr;
773       for (string::size_type i=0 ; i<command.size() ; i++) {
774         argv[i] = new char[command[i].size() + 1];
775         strncpy(argv[i], command[i].c_str(), command[i].size() + 1);
776         if (i>0) comstr += " # ";
777         comstr += command[i];
778       }
779
780       argv[command.size()] = NULL;
781
782       UNDER_LOCK( cout << "*** debug_command = " << comstr << endl );
783       UNDER_LOCK( cout << "*** debug_command = " << argv[0] << endl );
784
785       // Create the environment for the new process. Note (RB): Here we change the environment for
786       // the process launched in local. It would seem more logical to set the environment for the
787       // remote process.
788       Environnement env = _job.getEnvironnement();
789
790       char ** envp = NULL;
791       if(env.size() > 0) {
792         envp = new char * [env.size() + 1]; // 1 pour le NULL terminal
793         int i = 0;
794         for(Environnement::const_iterator it=env.begin(); it!=env.end(); it++, i++) {
795           const string  & key   = (*it).first;
796           const string  & value = (*it).second;
797           ostringstream oss;
798           oss << key << "=" << value;
799           envp[i]         = new char [oss.str().size() + 1];
800           strncpy(envp[i], oss.str().c_str(), oss.str().size() + 1);
801         }
802
803         // assert (i == env.size())
804         envp[i] = NULL;
805       }
806
807       //char *const parmList[] = {"/usr/bin/ssh", "localhost", "-l", "aribes", "sleep 1 && echo end", NULL};
808       //int result = execv("/usr/bin/ssh", parmList);
809       //UNDER_LOCK( cout << "*** debug_command = " << result << endl );
810       //UNDER_LOCK( cout << "*** debug_command = " << strerror(errno) << endl );
811
812
813
814       // On positionne les limites systeme imposees au fils
815       if (param.find(MAXCPUTIME) != param.end()) {
816         int maxcputime = param[MAXCPUTIME];
817         struct rlimit limit;
818         limit.rlim_cur = maxcputime;
819         limit.rlim_max = int(maxcputime * 1.1);
820         setrlimit(RLIMIT_CPU, &limit);
821       }
822
823       if (param.find(MAXDISKSIZE) != param.end()) {
824         int maxdisksize = param[MAXDISKSIZE];
825         struct rlimit limit;
826         limit.rlim_cur = maxdisksize * 1024;
827         limit.rlim_max = int(maxdisksize * 1.1) * 1024;
828         setrlimit(RLIMIT_FSIZE, &limit);
829       }
830
831       if (param.find(MAXRAMSIZE) != param.end()) {
832         int maxramsize = param[MAXRAMSIZE];
833         struct rlimit limit;
834         limit.rlim_cur = maxramsize * 1024;
835         limit.rlim_max = int(maxramsize * 1.1) * 1024;
836         setrlimit(RLIMIT_AS, &limit);
837       }
838
839
840       //char *const parmList[] = {"/usr/bin/ssh", "localhost", "-l", "aribes", "sleep 1 && echo end", NULL};
841       //int result = execv("/usr/bin/ssh", parmList);
842       //UNDER_LOCK( cout << "*** debug_command = " << result << endl );
843       //UNDER_LOCK( cout << "*** debug_command = " << strerror(errno) << endl );
844
845       // On cree une session pour le fils de facon a ce qu'il ne soit pas
846       // detruit lorsque le shell se termine (le shell ouvre une session et
847       // tue tous les process appartenant a la session en quittant)
848       setsid();
849
850
851       // On ferme les descripteurs de fichiers standards
852       //close(STDIN_FILENO);
853       //close(STDOUT_FILENO);
854       //close(STDERR_FILENO);
855
856
857       // On execute la commande du fils
858       int result = execve(argv[0], argv, envp);
859       UNDER_LOCK( cout << "*** debug_command = " << strerror(errno) << endl );
860       // No need to deallocate since nothing happens after a successful exec
861
862       // Normalement on ne devrait jamais arriver ici
863       ofstream file_err("error.log");
864       UNDER_LOCK( file_err << "Echec de l'appel a execve" << endl );
865
866     } catch (GenericException & e) {
867
868       std::cerr << "Caught exception : " << e.type << " : " << e.message << std::endl;
869     }
870
871     exit(99);
872   }
873
874 #else
875
876   BatchManager_Local::Process BatchManager_Local::ThreadAdapter::launchWin32ChildProcess()
877   {
878     Parametre param = _job.getParametre();
879     Parametre::iterator it;
880     PROCESS_INFORMATION pi;
881
882     try {
883
884       // EXECUTABLE is MANDATORY, if missing, we throw an exception
885       vector<string> exec_command;
886       if (param.find(EXECUTABLE) != param.end()) {
887         exec_command = _bm.exec_command(param);
888       } else {
889         throw RunTimeException("Parameter \"EXECUTABLE\" is mandatory for local batch submission");
890       }
891
892       // Build the command string from the command argument vector
893       string comstr;
894       for (unsigned int i=0 ; i<exec_command.size() ; i++) {
895         if (i>0) comstr += " ";
896         comstr += exec_command[i];
897       }
898
899       UNDER_LOCK( cout << "*** debug_command = " << comstr << endl );
900
901       // Create the environment for the new process. Note (RB): Here we change the environment for
902       // the process launched in local. It would seem more logical to set the environment for the
903       // remote process.
904       // Note that if no environment is specified, we reuse the current environment.
905       Environnement env = _job.getEnvironnement();
906       char * chNewEnv = NULL;
907
908       if(env.size() > 0) {
909         chNewEnv = new char[4096];
910         LPTSTR lpszCurrentVariable = chNewEnv;
911         for(Environnement::const_iterator it=env.begin() ; it!=env.end() ; it++) {
912           const string  & key   = (*it).first;
913           const string  & value = (*it).second;
914           string envvar = key + "=" + value;
915           envvar.copy(lpszCurrentVariable, envvar.size());
916           lpszCurrentVariable[envvar.size()] = '\0';
917           lpszCurrentVariable += lstrlen(lpszCurrentVariable) + 1;
918         }
919         // Terminate the block with a NULL byte.
920         *lpszCurrentVariable = '\0';
921       }
922
923
924       STARTUPINFO si;
925       ZeroMemory( &si, sizeof(si) );
926       si.cb = sizeof(si);
927       ZeroMemory( &pi, sizeof(pi) );
928
929       // Copy the command to a non-const buffer
930       char * buffer = strdup(comstr.c_str());
931
932       // launch the new process
933       BOOL res = CreateProcess(NULL, buffer, NULL, NULL, FALSE,
934                                CREATE_NO_WINDOW, chNewEnv, NULL, &si, &pi);
935
936       if (buffer) free(buffer);
937       if (!res) throw RunTimeException("Error while creating new process");
938
939       CloseHandle(pi.hThread);
940
941     } catch (GenericException & e) {
942
943       std::cerr << "Caught exception : " << e.type << " : " << e.message << std::endl;
944     }
945
946     return pi.hProcess;
947   }
948
949 #endif
950
951
952   void BatchManager_Local::kill_child_on_exit(void * p_pid)
953   {
954 #ifndef WIN32
955     //TODO: porting of following functionality
956     pid_t child = * static_cast<pid_t *>(p_pid);
957
958     // On tue le fils
959     kill(child, SIGTERM);
960
961     // Nota : on pourrait aussi faire a la suite un kill(child, SIGKILL)
962     // mais cette option n'est pas implementee pour le moment, car il est
963     // preferable de laisser le process fils se terminer normalement et seul.
964 #endif
965   }
966
967   void BatchManager_Local::delete_on_exit(void * arg)
968   {
969     ThreadAdapter * p_ta = static_cast<ThreadAdapter *>(arg);
970     delete p_ta;
971   }
972
973 }