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