src/Batch/Batch_BatchManager_Local.cxx

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