src/Batch/Batch_BatchManager_Local.cxx

   1 // Copyright (C) 2005  OPEN CASCADE, CEA, EDF R&D, LEG
   2 //           PRINCIPIA R&D, EADS CCR, Lip6, BV, CEDRAT
   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.
   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 * BatchManager_Local.cxx :
  21 *
  22 * Auteur : Ivan DUTKA-MALEN - EDF R&D
  23 * Mail   : mailto:ivan.dutka-malen@der.edf.fr
  24 * Date   : Thu Nov  6 10:17:22 2003
  25 * Projet : Salome 2
  26 *
  27 */
  28
  29 #include <iostream>
  30 #include <fstream>
  31 #include <sstream>
  32 #include <cstdlib>
  33 #include <sys/types.h>
  34 #ifdef WIN32
  35 # include <direct.h>
  36 #else
  37 # include <sys/wait.h>
  38 # include <unistd.h>
  39 #endif
  40 #include <ctime>
  41 #include <pthread.h>
  42 #include <signal.h>
  43 #include <errno.h>
  44 #include <string.h>
  45 #include <limits.h>
  46 #include "Batch_IOMutex.hxx"
  47 #include "Batch_BatchManager_Local.hxx"
  48
  49 using namespace std;
  50
  51 namespace Batch {
  52
  53
  54   // Constructeur
  55   BatchManager_Local::BatchManager_Local(const FactBatchManager * parent, const char * host) throw(InvalidArgumentException,ConnexionFailureException) : BatchManager(parent, host), _connect(0), _threads_mutex(), _threads(), _thread_id_id_association_mutex(), _thread_id_id_association_cond()
  56 #ifndef WIN32 //TODO: porting of following functionality
  57     ,_thread_id_id_association()
  58 #endif
  59   {
  60     pthread_mutex_init(&_threads_mutex, NULL);
  61     pthread_mutex_init(&_thread_id_id_association_mutex, NULL);
  62     pthread_cond_init(&_thread_id_id_association_cond, NULL);
  63   }
  64
  65   // Destructeur
  66   BatchManager_Local::~BatchManager_Local()
  67   {
  68     pthread_mutex_destroy(&_threads_mutex);
  69     pthread_mutex_destroy(&_thread_id_id_association_mutex);
  70     pthread_cond_destroy(&_thread_id_id_association_cond);
  71   }
  72
  73   // Methode pour le controle des jobs : soumet un job au gestionnaire
  74   const JobId BatchManager_Local::submitJob(const Job & job)
  75   {
  76     Job_Local jobLocal = job;
  77
  78     pthread_t thread_id = submit(jobLocal);
  79
  80     ostringstream oss;
  81     oss << getIdByThread_id(thread_id);
  82
  83     JobId id(this, oss.str());
  84
  85     return id;
  86   }
  87
  88   // Methode pour le controle des jobs : retire un job du gestionnaire
  89   void BatchManager_Local::deleteJob(const JobId & jobid)
  90   {
  91     Id id;
  92
  93     istringstream iss(jobid.getReference());
  94     iss >> id;
  95
  96     // On retrouve le thread_id du thread
  97     pthread_t thread_id;
  98
  99     // @@@ --------> SECTION CRITIQUE <-------- @@@
 100     pthread_mutex_lock(&_threads_mutex);
 101     if (_threads.find(id) != _threads.end())
 102       thread_id = _threads[id].thread_id;
 103     pthread_mutex_unlock(&_threads_mutex);
 104     // @@@ --------> SECTION CRITIQUE <-------- @@@
 105
 106     cancel(thread_id);
 107   }
 108
 109   // Methode pour le controle des jobs : suspend un job en file d'attente
 110   void BatchManager_Local::holdJob(const JobId & jobid)
 111   {
 112     Id id;
 113     istringstream iss(jobid.getReference());
 114     iss >> id;
 115
 116     UNDER_LOCK( cout << "BatchManager is sending HOLD command to the thread " << id << endl );
 117
 118     // On introduit une commande dans la queue du thread
 119     // @@@ --------> SECTION CRITIQUE <-------- @@@
 120     pthread_mutex_lock(&_threads_mutex);
 121     if (_threads.find(id) != _threads.end())
 122       _threads[id].command_queue.push(HOLD);
 123     pthread_mutex_unlock(&_threads_mutex);
 124     // @@@ --------> SECTION CRITIQUE <-------- @@@
 125   }
 126
 127   // Methode pour le controle des jobs : relache un job suspendu
 128   void BatchManager_Local::releaseJob(const JobId & jobid)
 129   {
 130     Id id;
 131     istringstream iss(jobid.getReference());
 132     iss >> id;
 133
 134     UNDER_LOCK( cout << "BatchManager is sending RELEASE command to the thread " << id << endl );
 135
 136     // On introduit une commande dans la queue du thread
 137     // @@@ --------> SECTION CRITIQUE <-------- @@@
 138     pthread_mutex_lock(&_threads_mutex);
 139     if (_threads.find(id) != _threads.end())
 140       _threads[id].command_queue.push(RELEASE);
 141     pthread_mutex_unlock(&_threads_mutex);
 142     // @@@ --------> SECTION CRITIQUE <-------- @@@
 143   }
 144
 145
 146   // Methode pour le controle des jobs : modifie un job en file d'attente
 147   void BatchManager_Local::alterJob(const JobId & jobid, const Parametre & param, const Environnement & env)
 148   {
 149   }
 150
 151   // Methode pour le controle des jobs : modifie un job en file d'attente
 152   void BatchManager_Local::alterJob(const JobId & jobid, const Parametre & param)
 153   {
 154     alterJob(jobid, param, Environnement());
 155   }
 156
 157   // Methode pour le controle des jobs : modifie un job en file d'attente
 158   void BatchManager_Local::alterJob(const JobId & jobid, const Environnement & env)
 159   {
 160     alterJob(jobid, Parametre(), env);
 161   }
 162
 163
 164
 165   // Methode pour le controle des jobs : renvoie l'etat du job
 166   JobInfo BatchManager_Local::queryJob(const JobId & jobid)
 167   {
 168     Id id;
 169     istringstream iss(jobid.getReference());
 170     iss >> id;
 171
 172     Parametre param;
 173     Environnement env;
 174
 175     //UNDER_LOCK( cout << "JobInfo BatchManager_Local::queryJob(const JobId & jobid) : AVANT section critique" << endl );
 176     // @@@ --------> SECTION CRITIQUE <-------- @@@
 177     pthread_mutex_lock(&_threads_mutex);
 178     param = _threads[id].param;
 179     env   = _threads[id].env;
 180     pthread_mutex_unlock(&_threads_mutex);
 181     // @@@ --------> SECTION CRITIQUE <-------- @@@
 182     //UNDER_LOCK( cout << "JobInfo BatchManager_Local::queryJob(const JobId & jobid) : APRES section critique" << endl );
 183
 184     JobInfo_Local ji(param, env);
 185     return ji;
 186   }
 187
 188
 189
 190   // Methode pour le controle des jobs : teste si un job est present en machine
 191   bool BatchManager_Local::isRunning(const JobId & jobid)
 192   {
 193     Id id;
 194     istringstream iss(jobid.getReference());
 195     iss >> id;
 196
 197     Status status;
 198
 199     //UNDER_LOCK( cout << "JobInfo BatchManager_Local::queryJob(const JobId & jobid) : AVANT section critique" << endl );
 200     // @@@ --------> SECTION CRITIQUE <-------- @@@
 201     pthread_mutex_lock(&_threads_mutex);
 202     status = _threads[id].status;
 203     pthread_mutex_unlock(&_threads_mutex);
 204     // @@@ --------> SECTION CRITIQUE <-------- @@@
 205     //UNDER_LOCK( cout << "JobInfo BatchManager_Local::queryJob(const JobId & jobid) : APRES section critique" << endl );
 206
 207     return (status == RUNNING);
 208   }
 209
 210
 211   // Methode d'execution d'un job
 212   pthread_t BatchManager_Local::submit(const Job_Local & job)
 213   {
 214     // L'id du thread a creer
 215     pthread_t thread_id =
 216 #ifdef WIN32
 217     {0,0};
 218 #else
 219       0;
 220 #endif
 221
 222     // Les attributs du thread a sa creation
 223     pthread_attr_t thread_attr;
 224     pthread_attr_init(&thread_attr);
 225     pthread_attr_setdetachstate(&thread_attr, PTHREAD_CREATE_DETACHED);
 226
 227     ThreadAdapter * p_ta = new ThreadAdapter(*this, job);
 228
 229     // Creation du thread qui va executer la commande systeme qu'on lui passe
 230     int rc = pthread_create(&thread_id,
 231       &thread_attr,
 232       &ThreadAdapter::run,
 233       static_cast<void *>(p_ta));
 234     if (rc) {
 235     }
 236
 237     // Liberation des zones memoire maintenant inutiles occupees par les attributs du thread
 238     pthread_attr_destroy(&thread_attr);
 239
 240     return thread_id;
 241   }
 242
 243
 244   // Methode de destruction d'un job
 245   void BatchManager_Local::cancel(pthread_t thread_id)
 246   {
 247     pthread_cancel(thread_id);
 248   }
 249
 250
 251   // Fabrique un identifiant unique pour les threads puisque le thread_id n'est pas unique
 252   // au cours du temps (il peut etre reutilise lorsqu'un thread se termine)
 253   // ATTENTION : cette methode est uniquement protegee par la section critique de l'association
 254   // Thread_id / Id (_thread_id_id_association_mutex)
 255   BatchManager_Local::Id BatchManager_Local::nextId()
 256   {
 257     static Id id = 0;
 258     Id nextId = id++;
 259     //UNDER_LOCK( cout << "BatchManager_Local::Id BatchManager_Local::nextId() : Id = " << nextId << endl );
 260     return nextId;
 261   }
 262
 263
 264   // Retourne l'Id enregistre dans l'association Thread_id / Id et le detruit immediatement
 265   BatchManager_Local::Id BatchManager_Local::getIdByThread_id(pthread_t thread_id)
 266   {
 267     Id id = -1;
 268
 269     // @@@ --------> SECTION CRITIQUE <-------- @@@
 270     pthread_mutex_lock(&_thread_id_id_association_mutex);
 271 #ifndef WIN32 //TODO: porting of following functionality
 272     while (_thread_id_id_association.find(thread_id) == _thread_id_id_association.end())
 273       pthread_cond_wait(&_thread_id_id_association_cond, &_thread_id_id_association_mutex);
 274
 275     id = _thread_id_id_association[thread_id];
 276     _thread_id_id_association.erase(thread_id);
 277 #endif
 278
 279     pthread_mutex_unlock(&_thread_id_id_association_mutex);
 280     // @@@ --------> SECTION CRITIQUE <-------- @@@
 281
 282     //UNDER_LOCK( cout << "BatchManager_Local::Id BatchManager_Local::getIdByThread_id(pthread_t thread_id) : Id = " << id << " - thread_id = " << thread_id << endl );
 283     return id;
 284   }
 285
 286
 287   // Associe un Thread_id a un Id nouvellement cree
 288   BatchManager_Local::Id BatchManager_Local::registerThread_id(pthread_t thread_id)
 289   {
 290     Id id = -1;
 291
 292     // @@@ --------> SECTION CRITIQUE <-------- @@@
 293     pthread_mutex_lock(&_thread_id_id_association_mutex);
 294 #ifndef WIN32 //TODO: porting of following functionality
 295     if (_thread_id_id_association.find(thread_id) == _thread_id_id_association.end()) {
 296       id = _thread_id_id_association[thread_id] = nextId();
 297       pthread_cond_signal(&_thread_id_id_association_cond);
 298
 299     } else {
 300       UNDER_LOCK( cerr << "ERROR : Pthread Inconstency. Two threads own the same thread_id." << endl );
 301     }
 302 #endif
 303     pthread_mutex_unlock(&_thread_id_id_association_mutex);
 304     // @@@ --------> SECTION CRITIQUE <-------- @@@
 305
 306     //UNDER_LOCK( cout << "BatchManager_Local::Id BatchManager_Local::registerThread_id(pthread_t thread_id) : Id = " << id << " - thread_id = " << thread_id << endl );
 307     return id;
 308   }
 309
 310
 311   // Constructeur de la classe ThreadAdapter
 312   BatchManager_Local::ThreadAdapter::ThreadAdapter(BatchManager_Local & bm, const Job_Local & job) :
 313   _bm(bm), _job(job)
 314   {
 315     // Nothing to do
 316   }
 317
 318
 319
 320   // Methode d'execution du thread
 321   void * BatchManager_Local::ThreadAdapter::run(void * arg)
 322   {
 323 #ifndef WIN32 //TODO: porting of following functionality
 324     // On bloque tous les signaux pour ce thread
 325     sigset_t setmask;
 326     sigfillset(&setmask);
 327     pthread_sigmask(SIG_BLOCK, &setmask, NULL);
 328
 329     // On autorise la terminaison differee du thread
 330     // (ces valeurs sont les valeurs par defaut mais on les force par precaution)
 331     pthread_setcancelstate(PTHREAD_CANCEL_ENABLE,  NULL);
 332     pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL);
 333
 334     // On enregistre la fonction de suppression du fils en cas d'arret du thread
 335     // Cette fontion sera automatiquement appelee lorsqu'une demande d'annulation
 336     // sera prise en compte par pthread_testcancel()
 337     pid_t child;
 338     pthread_cleanup_push(BatchManager_Local::kill_child_on_exit, static_cast<void *> (&child));
 339     pthread_cleanup_push(BatchManager_Local::delete_on_exit, arg);
 340
 341     ThreadAdapter * p_ta = static_cast<ThreadAdapter *>(arg);
 342
 343
 344
 345
 346     // Le code retour cumule (ORed) de tous les appels
 347     // Nul en cas de reussite de l'ensemble des operations
 348     int rc = 0;
 349
 350     // Cette table contient la liste des fichiers a detruire a la fin du processus
 351     std::vector<string> files_to_delete;
 352
 353
 354
 355     // On copie les fichiers d'entree pour le fils
 356     const Parametre param   = p_ta->_job.getParametre();
 357     Parametre::const_iterator it;
 358
 359     // On initialise la variable workdir a la valeur du Current Working Directory
 360     char * cwd =
 361 #ifdef WIN32
 362       _getcwd(NULL, 0);
 363 #else
 364       new char [PATH_MAX];
 365     getcwd(cwd, PATH_MAX);
 366 #endif
 367     string workdir = cwd;
 368     delete [] cwd;
 369
 370     if ( (it = param.find(WORKDIR)) != param.end() ) {
 371       workdir = static_cast<string>( (*it).second );
 372     }
 373
 374     string executionhost = string(param[EXECUTIONHOST]);
 375
 376     if ( (it = param.find(INFILE)) != param.end() ) {
 377       Versatile V = (*it).second;
 378       Versatile::iterator Vit;
 379
 380       for(Vit=V.begin(); Vit!=V.end(); Vit++) {
 381         CoupleType cpt  = *static_cast< CoupleType * >(*Vit);
 382         Couple cp       = cpt;
 383         string local    = cp.getLocal();
 384         string remote   = cp.getRemote();
 385
 386         string copy_cmd = p_ta->getBatchManager().copy_command("", local, executionhost, workdir + "/" + remote);
 387         UNDER_LOCK( cout << "Copying : " << copy_cmd << endl );
 388
 389         if (system(copy_cmd.c_str()) ) {
 390           // Echec de la copie
 391           rc |= 1;
 392         } else {
 393           // On enregistre le fichier comme etant a detruire
 394           files_to_delete.push_back(workdir + "/" + remote);
 395         }
 396
 397       }
 398     }
 399
 400
 401
 402
 403 #ifdef WIN32
 404     //TODO
 405     //Using CreateThread instead fork() POSIX function
 406 #else
 407     // On forke/exec un nouveau process pour pouvoir controler le fils
 408     // (plus finement qu'avec un appel system)
 409     // int rc = system(commande.c_str());
 410     child = fork();
 411     if (child < 0) { // erreur
 412       UNDER_LOCK( cerr << "Fork impossible (rc=" << child << ")" << endl );
 413
 414     } else if (child > 0) { // pere
 415       p_ta->pere(child);
 416
 417     } else { // fils
 418       p_ta->fils();
 419     }
 420 #endif
 421
 422
 423
 424     // On copie les fichiers de sortie du fils
 425     if ( (it = param.find(OUTFILE)) != param.end() ) {
 426       Versatile V = (*it).second;
 427       Versatile::iterator Vit;
 428
 429       for(Vit=V.begin(); Vit!=V.end(); Vit++) {
 430         CoupleType cpt  = *static_cast< CoupleType * >(*Vit);
 431         Couple cp       = cpt;
 432         string local    = cp.getLocal();
 433         string remote   = cp.getRemote();
 434
 435         string copy_cmd = p_ta->getBatchManager().copy_command(executionhost, workdir + "/" + remote, "", local);
 436         UNDER_LOCK( cout << "Copying : " << copy_cmd << endl );
 437
 438         if (system(copy_cmd.c_str()) ) {
 439           // Echec de la copie
 440           rc |= 1;
 441         } else {
 442           // On enregistre le fichier comme etant a detruire
 443           files_to_delete.push_back(workdir + "/" + remote);
 444         }
 445
 446       }
 447     }
 448
 449
 450
 451
 452     // On efface les fichiers d'entree et de sortie du fils si les copies precedentes ont reussi
 453     // ou si la creation du fils n'a pu avoir lieu
 454     if ( (rc == 0) || (child < 0) ) {
 455       std::vector<string>::const_iterator it;
 456       for(it=files_to_delete.begin(); it!=files_to_delete.end(); it++) {
 457         string remove_cmd = p_ta->getBatchManager().remove_command(executionhost, *it);
 458         UNDER_LOCK( cout << "Removing : " << remove_cmd << endl );
 459         system(remove_cmd.c_str());
 460       }
 461     }
 462
 463
 464
 465     // On retire la fonction de nettoyage de la memoire
 466     pthread_cleanup_pop(0);
 467
 468     // On retire la fonction de suppression du fils
 469     pthread_cleanup_pop(0);
 470
 471
 472
 473     // On invoque la fonction de nettoyage de la memoire
 474     delete_on_exit(arg);
 475
 476     UNDER_LOCK( cout << "Father is leaving" << endl );
 477     pthread_exit(NULL);
 478 #endif
 479     return NULL;
 480   }
 481
 482
 483
 484
 485   void BatchManager_Local::ThreadAdapter::pere(pid_t child)
 486   {
 487 #ifndef WIN32 //TODO: porting of following functionality
 488     time_t child_starttime = time(NULL);
 489
 490     // On enregistre le fils dans la table des threads
 491     pthread_t thread_id = pthread_self();
 492     Id id = _bm.registerThread_id(thread_id);
 493
 494     Parametre param   = _job.getParametre();
 495     Environnement env = _job.getEnvironnement();
 496
 497     ostringstream thread_id_sst;
 498     thread_id_sst << id;
 499     param[ID]         = thread_id_sst.str();
 500     param[STATE]      = "Running";
 501     param[PID]        = child;
 502
 503     // @@@ --------> SECTION CRITIQUE <-------- @@@
 504     pthread_mutex_lock(&_bm._threads_mutex);
 505     _bm._threads[id].thread_id = thread_id;
 506     _bm._threads[id].pid       = child;
 507     _bm._threads[id].status    = RUNNING;
 508     _bm._threads[id].param     = param;
 509     _bm._threads[id].env       = env;
 510     _bm._threads[id].command_queue.push(NOP);
 511     pthread_mutex_unlock(&_bm._threads_mutex);
 512     // @@@ --------> SECTION CRITIQUE <-------- @@@
 513
 514
 515
 516
 517
 518     // on boucle en attendant que le fils ait termine
 519     while (1) {
 520       int child_rc = 0;
 521       pid_t child_wait_rc = waitpid(child, &child_rc, WNOHANG /* | WUNTRACED */);
 522       if (child_wait_rc > 0) {
 523         if (WIFSTOPPED(child_rc)) {
 524           // NOTA : pour rentrer dans cette section, il faut que le flag WUNTRACED
 525           // soit positionne dans l'appel a waitpid ci-dessus. Ce flag est couramment
 526           // desactive car s'il est possible de detecter l'arret d'un process, il est
 527           // plus difficile de detecter sa reprise.
 528
 529           // Le fils est simplement stoppe
 530           // @@@ --------> SECTION CRITIQUE <-------- @@@
 531           pthread_mutex_lock(&_bm._threads_mutex);
 532           _bm._threads[id].status       = STOPPED;
 533           _bm._threads[id].param[STATE] = "Stopped";
 534           pthread_mutex_unlock(&_bm._threads_mutex);
 535           // @@@ --------> SECTION CRITIQUE <-------- @@@
 536           UNDER_LOCK( cout << "Father sees his child is STOPPED : " << child_wait_rc << endl );
 537
 538         }
 539         else {
 540           // Le fils est termine, on sort de la boucle et du if englobant
 541           // @@@ --------> SECTION CRITIQUE <-------- @@@
 542           pthread_mutex_lock(&_bm._threads_mutex);
 543           _bm._threads[id].status       = DONE;
 544           _bm._threads[id].param[STATE] = "Done";
 545           pthread_mutex_unlock(&_bm._threads_mutex);
 546           // @@@ --------> SECTION CRITIQUE <-------- @@@
 547           UNDER_LOCK( cout << "Father sees his child is DONE : " << child_wait_rc << " (child_rc=" << (WIFEXITED(child_rc) ? WEXITSTATUS(child_rc) : -1) << ")" << endl );
 548           break;
 549         }
 550       }
 551       else if (child_wait_rc == -1) {
 552         // Le fils a disparu ...
 553         // @@@ --------> SECTION CRITIQUE <-------- @@@
 554         pthread_mutex_lock(&_bm._threads_mutex);
 555         _bm._threads[id].status       = DEAD;
 556         _bm._threads[id].param[STATE] = "Dead";
 557         pthread_mutex_unlock(&_bm._threads_mutex);
 558         // @@@ --------> SECTION CRITIQUE <-------- @@@
 559         UNDER_LOCK( cout << "Father sees his child is DEAD : " << child_wait_rc << " (Reason : " << strerror(errno) << ")" << endl );
 560         break;
 561       }
 562
 563
 564
 565       // On teste si le thread doit etre detruit
 566       pthread_testcancel();
 567
 568
 569
 570       // On regarde si le fils n'a pas depasse son temps (wallclock time)
 571       time_t child_currenttime = time(NULL);
 572       time_t child_elapsedtime = child_currenttime - child_starttime;
 573       if (param.find(MAXWALLTIME) != param.end()) {
 574         int maxwalltime = param[MAXWALLTIME];
 575         //        cout << "child_starttime          = " << child_starttime        << endl
 576         //             << "child_currenttime        = " << child_currenttime      << endl
 577         //             << "child_elapsedtime        = " << child_elapsedtime      << endl
 578         //             << "maxwalltime              = " << maxwalltime            << endl
 579         //             << "int(maxwalltime * 1.1)   = " << int(maxwalltime * 1.1) << endl;
 580         if (child_elapsedtime > int(maxwalltime * 1.1) ) { // On se donne 10% de marge avant le KILL
 581           UNDER_LOCK( cout << "Father is sending KILL command to the thread " << id << endl );
 582           // On introduit une commande dans la queue du thread
 583           // @@@ --------> SECTION CRITIQUE <-------- @@@
 584           pthread_mutex_lock(&_bm._threads_mutex);
 585           if (_bm._threads.find(id) != _bm._threads.end())
 586             _bm._threads[id].command_queue.push(KILL);
 587           pthread_mutex_unlock(&_bm._threads_mutex);
 588           // @@@ --------> SECTION CRITIQUE <-------- @@@
 589
 590
 591         } else if (child_elapsedtime > maxwalltime ) {
 592           UNDER_LOCK( cout << "Father is sending TERM command to the thread " << id << endl );
 593           // On introduit une commande dans la queue du thread
 594           // @@@ --------> SECTION CRITIQUE <-------- @@@
 595           pthread_mutex_lock(&_bm._threads_mutex);
 596           if (_bm._threads.find(id) != _bm._threads.end())
 597             _bm._threads[id].command_queue.push(TERM);
 598           pthread_mutex_unlock(&_bm._threads_mutex);
 599           // @@@ --------> SECTION CRITIQUE <-------- @@@
 600         }
 601       }
 602
 603
 604
 605       // On regarde s'il y a quelque chose a faire dans la queue de commande
 606       // @@@ --------> SECTION CRITIQUE <-------- @@@
 607       pthread_mutex_lock(&_bm._threads_mutex);
 608       if (_bm._threads.find(id) != _bm._threads.end()) {
 609         while (_bm._threads[id].command_queue.size() > 0) {
 610           Commande cmd = _bm._threads[id].command_queue.front();
 611           _bm._threads[id].command_queue.pop();
 612
 613           switch (cmd) {
 614     case NOP:
 615       UNDER_LOCK( cout << "Father does nothing to his child" << endl );
 616       break;
 617
 618     case HOLD:
 619       UNDER_LOCK( cout << "Father is sending SIGSTOP signal to his child" << endl );
 620       kill(child, SIGSTOP);
 621       break;
 622
 623     case RELEASE:
 624       UNDER_LOCK( cout << "Father is sending SIGCONT signal to his child" << endl );
 625       kill(child, SIGCONT);
 626       break;
 627
 628     case TERM:
 629       UNDER_LOCK( cout << "Father is sending SIGTERM signal to his child" << endl );
 630       kill(child, SIGTERM);
 631       break;
 632
 633     case KILL:
 634       UNDER_LOCK( cout << "Father is sending SIGKILL signal to his child" << endl );
 635       kill(child, SIGKILL);
 636       break;
 637
 638     case ALTER:
 639       break;
 640
 641     default:
 642       break;
 643           }
 644         }
 645
 646       }
 647       pthread_mutex_unlock(&_bm._threads_mutex);
 648       // @@@ --------> SECTION CRITIQUE <-------- @@@
 649
 650       // On fait une petite pause pour ne pas surcharger inutilement le processeur
 651       sleep(1);
 652
 653     }
 654 #endif
 655
 656
 657   }
 658
 659
 660
 661
 662   void BatchManager_Local::ThreadAdapter::fils()
 663   {
 664 #ifndef WIN32 //TODO: porting of following functionality
 665     Parametre param = _job.getParametre();
 666     Parametre::iterator it;
 667
 668     try {
 669
 670       // On se place dans le repertoire de travail
 671       if ( (it = param.find(WORKDIR)) != param.end() ) {
 672         string workdir = static_cast<string>( (*it).second );
 673         chdir(workdir.c_str());
 674       }
 675
 676
 677
 678
 679       // EXECUTABLE is MANDATORY, if missing, we exit with failure notification
 680       char * execpath = NULL;
 681       if (param.find(EXECUTABLE) != param.end()) {
 682         string executable = _bm.exec_command(param);
 683         execpath          = new char [executable.size() + 1];
 684         strncpy(execpath, executable.c_str(), executable.size() + 1);
 685       } else exit(1);
 686
 687       string debug_command = execpath;
 688
 689       string name = (param.find(NAME) != param.end()) ? param[NAME] : param[EXECUTABLE];
 690
 691       char **  argv = NULL;
 692       if (param.find(ARGUMENTS) != param.end()) {
 693         Versatile V = param[ARGUMENTS];
 694
 695         argv = new char * [V.size() + 2]; // 1 pour name et 1 pour le NULL terminal
 696
 697         argv[0] = new char [name.size() + 1];
 698         strncpy(argv[0], name.c_str(), name.size() + 1);
 699
 700         debug_command  += string(" # ") + argv[0];
 701
 702         int i = 1;
 703         for(Versatile::const_iterator it=V.begin(); it!=V.end(); it++, i++) {
 704           StringType argt = * static_cast<StringType *>(*it);
 705           string     arg  = argt;
 706           argv[i]         = new char [arg.size() + 1];
 707           strncpy(argv[i], arg.c_str(), arg.size() + 1);
 708           debug_command  += string(" # ") + argv[i];
 709         }
 710
 711         // assert (i == V.size() + 1)
 712         argv[i] = NULL;
 713       }
 714
 715
 716       UNDER_LOCK( cout << "*** debug_command = " << debug_command << endl );
 717
 718
 719
 720       Environnement env = _job.getEnvironnement();
 721
 722
 723       char ** envp = NULL;
 724       if(env.size() > 0) {
 725         envp = new char * [env.size() + 1]; // 1 pour le NULL terminal
 726         int i = 0;
 727         for(Environnement::const_iterator it=env.begin(); it!=env.end(); it++, i++) {
 728           const string  & key   = (*it).first;
 729           const string  & value = (*it).second;
 730           ostringstream oss;
 731           oss << key << "=" << value;
 732           envp[i]         = new char [oss.str().size() + 1];
 733           strncpy(envp[i], oss.str().c_str(), oss.str().size() + 1);
 734         }
 735
 736         // assert (i == env.size())
 737         envp[i] = NULL;
 738       }
 739
 740
 741
 742
 743       // On positionne les limites systeme imposees au fils
 744       if (param.find(MAXCPUTIME) != param.end()) {
 745         int maxcputime = param[MAXCPUTIME];
 746         struct rlimit limit;
 747         limit.rlim_cur = maxcputime;
 748         limit.rlim_max = int(maxcputime * 1.1);
 749         setrlimit(RLIMIT_CPU, &limit);
 750       }
 751
 752       if (param.find(MAXDISKSIZE) != param.end()) {
 753         int maxdisksize = param[MAXDISKSIZE];
 754         struct rlimit limit;
 755         limit.rlim_cur = maxdisksize * 1024;
 756         limit.rlim_max = int(maxdisksize * 1.1) * 1024;
 757         setrlimit(RLIMIT_FSIZE, &limit);
 758       }
 759
 760       if (param.find(MAXRAMSIZE) != param.end()) {
 761         int maxramsize = param[MAXRAMSIZE];
 762         struct rlimit limit;
 763         limit.rlim_cur = maxramsize * 1024;
 764         limit.rlim_max = int(maxramsize * 1.1) * 1024;
 765         setrlimit(RLIMIT_AS, &limit);
 766       }
 767
 768
 769
 770       // On cree une session pour le fils de facon a ce qu'il ne soit pas
 771       // detruit lorsque le shell se termine (le shell ouvre une session et
 772       // tue tous les process appartenant a la session en quittant)
 773       setsid();
 774
 775
 776       // On ferme les descripteurs de fichiers standards
 777       //close(STDIN_FILENO);
 778       //close(STDOUT_FILENO);
 779       //close(STDERR_FILENO);
 780
 781
 782       // On execute la commande du fils
 783       execve(execpath, argv, envp);
 784
 785       // No need to deallocate since nothing happens after a successful exec
 786
 787       // Normalement on ne devrait jamais arriver ici
 788       ofstream file_err("error.log");
 789       UNDER_LOCK( file_err << "Echec de l'appel a execve" << endl );
 790
 791     } catch (GenericException & e) {
 792
 793       std::cerr << "Caught exception : " << e.type << " : " << e.message << std::endl;
 794     }
 795
 796     exit(99);
 797 #endif
 798   }
 799
 800
 801
 802
 803   void BatchManager_Local::kill_child_on_exit(void * p_pid)
 804   {
 805 #ifndef WIN32
 806     //TODO: porting of following functionality
 807     pid_t child = * static_cast<pid_t *>(p_pid);
 808
 809     // On tue le fils
 810     kill(child, SIGTERM);
 811
 812     // Nota : on pourrait aussi faire a la suite un kill(child, SIGKILL)
 813     // mais cette option n'est pas implementee pour le moment, car il est
 814     // preferable de laisser le process fils se terminer normalement et seul.
 815 #endif
 816   }
 817
 818   void BatchManager_Local::delete_on_exit(void * arg)
 819   {
 820     ThreadAdapter * p_ta = static_cast<ThreadAdapter *>(arg);
 821     delete p_ta;
 822   }
 823
 824 }