OMKPvmSvm.cxx

Go to the documentation of this file.

/*
 * This file is part of openMask © INRIA, CNRS, Universite de Rennes 1 1993-2002, thereinafter the Software
 * 
 * The Software has been developped within the Siames Project. 
 * INRIA, the University of Rennes 1 and CNRS jointly hold intellectual property rights
 * 
 * The Software has been registered with the Agence pour la Protection des
 * Programmes (APP) under registration number IDDN.FR.001.510008.00.S.P.2001.000.41200
 *  
 * This file may be distributed under the terms of the Q Public License
 * version 1.0 as defined by Trolltech AS of Norway and appearing in the file
 * LICENSE.QPL included in the packaging of this file.
 *
 * Licensees holding valid specific licenses issued by INRIA, CNRS or Université de Rennes 1 
 * for the software may use this file in accordance with that specific license
 *
 */

#ifdef _MSC_VER
#  include <windows.h>
#else
#  include <unistd.h>
#endif
#include <cstring>

#include <OMKPvmSvm.h>

#include <pvm3.h>
#include <OMKProcess.h>
#include <OMKPvmSvmLink.h>
#include <OMKPvmMessage.h>
#include <OMKController.h>
using namespace std ;
using namespace OMK ;

int PvmSvm::pvmDataEncoding = PvmDataDefault ;
int PvmSvm::pvmSpawnFlags = PvmTaskHost ;
//--------------------------------------------------------------------

PvmSvm::PvmSvm (NameToPointerMap<Process> * tab, const Date & latence, const Date & timeOut, const int deadReckoningInterval, const bool yieldNeeded, int argc, char * argv [])
   : Svm (tab, latence, timeOut, deadReckoningInterval, yieldNeeded),
     _argc ( argc)//, _argv ( argv ) // TDTD code indigent et scandaleux ?
{
// TDTD tentative d'initialisation du tableau de chaînes
   _argv = new char * [_argc + 1] ;
   for (int i = 0 ; i < _argc ; i ++) {
      _argv [i] = strdup (argv [i]) ;
   }
   _argv [_argc] = NULL ;

  _siteId = pvm_mytid () ;

  // added By Guillermo Andrade
  // this option allow to use direct connection between pvm tasks
  // using TCP instead of UDP packing message via pvmd-pvmd connection
  pvm_setopt(PvmRoute,PvmRouteDirect);

}

//--------------------------------------------------------------------

PvmSvm::~PvmSvm () 
{
   // do not wait for output of spawned tasks to be written
   pvm_catchout (0) ;
   
   // signal end of pvm to the pvm deamon
   pvm_exit () ;

}

void PvmSvm::joinSvmGroup ( std::string & groupName ) 
{
  int resul = pvm_joingroup ( const_cast<char *> (groupName.c_str()) ) ;
  if ( resul < 0 )
    {
      cerr<<"PvmSvm::joinSvmGroup ERROR: ";
      switch ( resul )
        {
        case PvmSysErr: 
          cerr<<"pvmd was not started or has crashed.";
          break;
        case PvmBadParam:
          cerr<<"giving a NULL group name.";
          break;
        case PvmDupGroup:
          cerr<<"trying to join a group you are allready in.";
          break;
        default:
          cerr<<"unexpected error: "<<resul;
        }
      OMASSERT( false ) ;
      cerr<<endl;
    }
}


void PvmSvm::groupBarrier (std::string groupName, int numberToJoin)
{
   //   cerr<<"PvmSvm::groupBarrier: Waiting for "<<_numberOfSlaves<<" in pvm group named _"<<groupName.c_str()<<"_"<<endl;
  int resul = pvm_barrier ( const_cast<char *> (groupName.c_str()), numberToJoin) ;
  //cerr<<"PvmSvm::groupBarrier: Waiting done"<<endl;
  if ( resul < 0 )
    {
      cerr<<"PvmSvm::groupBarrier ERROR: ";
      switch ( resul )
        {
        case PvmSysErr: 
          cerr<<"pvmd was not started or has crashed.";
          break;
        case PvmBadParam:
          cerr<<"giving a numberToJoin < 1.";
          break;
        case PvmNoGroup:
          cerr<<"giving a non-existent group name.";
          break;
        case PvmNotInGroup:
          cerr<<"calling process is not in specified group.";
          break;
        default:
          cerr<<"unexpected error: "<<resul;
        }
        OMASSERT( false ) ;
      cerr<<endl;
    }
}



void PvmSvm::broadcastToGroup (std::string groupName, PvmMessage::MessageTag tag)
{
  int resul = pvm_bcast ( const_cast<char *> (groupName.c_str()), tag ) ;
  if ( resul < 0 )
    {
      cerr<<"PvmSvm::broadcastToGroup ERROR: ";
      switch ( resul )
        {
        case PvmSysErr: 
          cerr<<"pvmd was not started or has crashed.";
          break;
        case PvmBadParam:
          cerr<<"giving a negative message tag.";
          break;
        case PvmNoGroup:
          cerr<<"giving a non-existent group name.";
          break;
        default:
          cerr<<"unexpected error: "<<resul;
        }
        OMASSERT( false ) ;
      cerr<<endl;
    }
}

void PvmSvm::initBeforeMessagePacking ()
{
  int resul = pvm_initsend ( PvmSvm::pvmDataEncoding ) ;
  if ( resul < 0 )
    {
      cerr<<"PvmSvm::initBeforeMessagePacking ERROR: ";
      switch ( resul )
        {
        case PvmBadParam:
          cerr<<"giving a invalid encoding value.";
          break;
        case PvmNoMem:
          cerr<<"Malloc has failed. There is not enough memory to create the buffer";
          break;
        default:
          cerr<<"unexpected error: "<<resul;
        }
        OMASSERT( false ) ;
      cerr<<endl;
    }
}

int PvmSvm::nonblockingReceive (PvmMessage::MessageTag tag) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmSvm::nonblockingReceive ("<<tag<<")" <<endl;
#endif
  int res = 0 ;
  int bufid = pvm_nrecv (-1, tag) ;
  while ( bufid > 0)
    {
      ++res ;
      bufid = pvm_nrecv (-1, tag) ;
    }
  if ( bufid < 0 )
    {
      cerr<<"PvmSvm::nonblockingReceive ERROR: ";
      switch ( bufid )
        {
        case PvmBadParam:
          cerr<<"giving a invalid encoding tid value or msgtag.";
          break;
        case PvmSysErr:
          cerr<<"pvmd not responding.";
          break;
        default:
          cerr<<"unexpected error: "<<bufid;
        }
        OMASSERT( false ) ;
      cerr<<endl;
   }
  return res ;
}



std::pair<PvmMessage::MessageTag, int> 
PvmSvm::waitForAnyRequests (PvmIncomingMessage & receiveBuffer) 
{
  std::pair<PvmMessage::MessageTag, int> result ;
  
  int bufid = pvm_recv ( -1, -1 ) ; //receive anything

  if ( bufid < 0 ) 
    {
      cerr<<"Svm::waitForAnyRequests ERROR ";
      switch (bufid)
        {
        case PvmBadParam:
          cerr<<"giving an invalid tid value, or msgtag < -1";
          break;
        case PvmSysErr:
          cerr<<"pvmd not responding";
          break;
        default:
          cerr<<"unexpected error";
        OMASSERT( false ) ;
        }
      cerr<<endl;
    }

  int bytes ;
  //  int senderSiteId ;
  int msgtag ;
  int info = pvm_bufinfo( bufid, &bytes , &msgtag , &result.second  );
  result.first = static_cast<PvmMessage::MessageTag>(msgtag) ;
  if ( info < 0 ) 
    {
      cerr<<"Svm::waitForAnyRequests ERROR in pvm_bufinfo ";
      switch (info)
        {
        case PvmBadParam:
          cerr<<"invalid argument";
          break;
        case PvmNoSuchBuf:
          cerr<<"specified buffer does not exist";
          break;
        default:
          cerr<<"unexpected error";
        OMASSERT( false ) ;
        }
      cerr<<endl;
    }

  receiveBuffer.initialise ( bufid ) ;
  
  return result ;
}



//--------------------------------------------------------------------

int PvmSvm::getSiteId () {
  assert ( _siteId >= 0 ) ;
  return _siteId ; 
}

//--------------------------------------------------------------------

void PvmSvm::addNewWorkstation (const Name & m) {
   char * mac [2] ; // Machine a ajouter
   int res ; // Resultat du add_host
   int infos [2] ; // Informations sur chaque host
   int tentative = 0 ; // Numero de la tentative
   int nbTentative = 40 ; // On essaie 40 fois

   // here, const casting because pvm uses a C interface without const
   mac [0] = const_cast<char *> (m.getCString ()) ;
   mac [1] = (char *)NULL ;

//TDTD ajout d'une temporisation le 24/05/2006 pour atténuer les pbs lors d'ajouts de plusieurs process sur la même machine
#ifdef _MSC_VER
                                Sleep(1);
#else
                                sleep (1) ;
#endif
   while (tentative < nbTentative) 
      {
         cout << "Tentative number " << tentative+1
              << " for machine " << mac [0] <<endl;
      res = pvm_addhosts (mac, 1, infos) ;
      if (res < 1) { // On a une erreur
         tentative++;

         if (infos[0] != PvmDupHost) 
           { 
             if (tentative >= nbTentative) 
               {
                 cerr << "Type d'erreur : " << res << endl;
                 cerr << "Infos[0] : " << infos[0] << endl;
                 cerr << "Machine : " << mac [0] << ", " << mac [1] << endl;
                 Controller::error( "PvmSvm::addNewWorkstation: pb pour ajouter une machine");
               } 
             else 
               {
#ifdef _MSC_VER
                                Sleep(5);
#else
                                sleep (5) ;
#endif
               }
           } 
         else 
           { // workstation wasn't added because it is allready present !
             // no need to try more
             tentative = nbTentative;
           }
      } 
      else 
        { // On n'a pas d'erreur, on arrete d'essayer
          tentative = nbTentative;
          ++_numberOfSlaves ;
        }
   }
   cout << "Machine " << mac [0] << " added" << endl ;           
   //cout << "status of machine: "<<pvm_mstat(mac [0])<<endl;
}

//--------------------------------------------------------------------

int PvmSvm::getParentSiteId () {
   int tidCtrlGlo = pvm_parent () ;
   // options PVM:
   //pvm_setopt (PvmSelfTraceTid, tidCtrlGlo) ; 
   //pvm_setopt (PvmSelfTraceCode, 666) ; 
   //pvm_setopt (PvmSelfOutputTid, tidCtrlGlo) ; 
   //pvm_setopt (PvmSelfOutputCode, 667) ; 
   //pvm_catchout (stdout) ;
   // Message au format XDR
   pvm_initsend (pvmDataEncoding) ;
   if (tidCtrlGlo == PvmNoParent ) 
     {
       return 0 ;
     } 
   else 
     {
       assert ( tidCtrlGlo != 0 ) ; //for coherance
       return tidCtrlGlo ;
     }
}

//--------------------------------------------------------------------

int PvmSvm::spawnProcess (Process * p) 
{
   assert ( _argv[_argc] == NULL ) ;
   //   char * argv [5] ;
   int tidCtrlLoc ;
//     istrstream is (fConfig.getCString (), strlen (fConfig.getCString ())) ;
//     argv [0] = new char [256] ;
//     argv [1] = new char [256] ;
//     argv [2] = new char [256] ;
//     argv [3] = new char [256] ;
//     is >> argv [0] >> argv [1] >> argv [2] >> argv [3];
//     argv [4] = (char *)NULL ;
   std::cerr << "PvmSvm::spawnProcess : " ;
       for ( int i = 0 ; i < _argc ; ++i) 
         {
            std::cerr << " <" << _argv[i] << ">" ;
         }
   std::cerr << std::endl ;

//    int res = pvm_spawn ("tutorialValgrind", 
//                      const_cast<char **>(&_argv[1]), 
//                      pvmSpawnFlags,
//                      (char *)(p->getHostMachineName ().getCString ()), 
//                      1, 
//                      &tidCtrlLoc) ;

   int res = pvm_spawn (const_cast<char *>(_argv[0]), 
                        const_cast<char **>(&_argv[1]), 
                        pvmSpawnFlags,
                        (char *)(p->getHostMachineName ().getCString ()), 
                        1, 
                        &tidCtrlLoc) ;
   if (res != 1) 
     {
       cout << "Executable : " << _argv[0] << endl ;
       //p->Executable ().getCString ()
       char ** argv = &_argv[1] ;
       for ( int i = 0 ; i < _argc-1 ; ++i) 
         {
           cout << "_argv ["<<i<<"] : " << argv[i] << endl ;
         }
       cout << "on machine: " 
            << p->getHostMachineName ().getCString () << endl ;
       cout << "status of machine: "
            <<pvm_mstat(const_cast<char *>(p->getHostMachineName ().
                                           getCString ()))<<endl;
       cout << "tidCtrlLoc: " << tidCtrlLoc << endl ;
       cout << "Number of tasks started : " << res << endl ;
       Controller::error ("PvmSvm::spawnProcess: spawn problem") ;
     }
   return (int)tidCtrlLoc ;
}


//--------------------------------------------------------------------

void PvmSvm::removeWorkstation (const Name & m) 
{
   int res ;
   char * mac [2] ;
   mac [0]= (char *)(m.getCString ()) ;
   mac [1] = (char *)NULL ;
   
   pvm_delhosts (mac, 1, &res) ;
}

//--------------------------------------------------------------------

SvmLink * PvmSvm::createSvmLink (const int & d) 
{
   PvmSvmLink * canal = new PvmSvmLink (d) ;
   return (SvmLink*)canal;       
}

//--------------------------------------------------------------------

bool PvmSvm::testIfSiteRecovered (Process * p)  //(const Name &nom)
{
   bool flag;
 //  int mstat = pvm_mstat( const_cast<char *> (nom.getCString()) );
 //int mstat = pvm_mstat( const_cast<char *> (p->getHostMachineName().getCString()));
 // cout << "status of machine: "<<pvm_mstat(const_cast<char *>(nom.getCString ()))<<endl;
  cout << "nom de la machine: "<<p->getHostMachineName()<<endl;
  cout<<"TID de truc "<<p->getSvmLink()->getTID()<<endl;
   int pstat = pvm_pstat ( p->getSvmLink()->getTID() ) ;
 
   switch (pstat)
     {
     case PvmOk:
       flag = true ;
       cout<<"koukou"<<endl;
       break;
     case PvmSysErr:
       cout<<"putain de merde"<<endl;
       flag = false;
       break;
     case PvmNoTask:
       flag = false;
       break;
     default:
       flag = false ;
       cout<<"kiki"<<endl;
     }
   return flag;
 }

---