---

PsnPvmSvm.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 _PVM
#include <unistd.h>
#include <stdio.h>
#include <PsnPvmSvm.h>

#include <Psn.h>
#include <pvm3.h>
#include <PsnProcess.h>
#include <PsnPvmSvmLink.h>
#include <PsnPvmMessage.h>
#include <PsController.h>

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

PsnPvmSvm::PsnPvmSvm (PsNameToPointerMap<PsnProcess> * tab, const PsDate & latence, int argc, char * argv [])
   : PsnSvm (tab, latence),
     _argc ( argc),
     _argv ( argv )
{
  _siteId = pvm_mytid () ;
}

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

PsnPvmSvm::~PsnPvmSvm () 
{
   // 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 PsnPvmSvm::joinSvmGroup ( string & groupName ) 
{
  int resul = pvm_joingroup ( const_cast<char *> (groupName.c_str()) ) ;
  if ( resul < 0 )
    {
      cerr<<"PsnPvmSvm::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;
        }
      if (PsController::warningLevel >=  PsController::FatalWarnings) 
        {             
          exit (2) ;
        }
      cerr<<endl;
    }
}


void PsnPvmSvm::groupBarrier (string groupName, int numberToJoin)
{
   //   cerr<<"PsnPvmSvm::groupBarrier: Waiting for "<<_numberOfSlaves<<" in pvm group named _"<<groupName.c_str()<<"_"<<endl;
  int resul = pvm_barrier ( const_cast<char *> (groupName.c_str()), numberToJoin) ;
  //cerr<<"PsnPvmSvm::groupBarrier: Waiting done"<<endl;
  if ( resul < 0 )
    {
      cerr<<"PsnPvmSvm::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;
        }
      if (PsController::warningLevel >=  PsController::FatalWarnings) 
        {             
          exit (2) ;
        }
      cerr<<endl;
    }
}



void PsnPvmSvm::broadcastToGroup (string groupName, PsnPvmMessage::MessageTag tag)
{
  int resul = pvm_bcast ( const_cast<char *> (groupName.c_str()), tag ) ;
  if ( resul < 0 )
    {
      cerr<<"PsnPvmSvm::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;
        }
      if (PsController::warningLevel >=  PsController::FatalWarnings) 
        {             
          exit (2) ;
        }
      cerr<<endl;
    }
}

void PsnPvmSvm::initBeforeMessagePacking ()
{
  int resul = pvm_initsend ( PsnPvmSvm::pvmDataEncoding ) ;
  if ( resul < 0 )
    {
      cerr<<"PsnPvmSvm::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;
        }
      if (PsController::warningLevel >=  PsController::FatalWarnings) 
        {             
          exit (2) ;
        }
      cerr<<endl;
    }
}

int PsnPvmSvm::nonblockingReceive (PsnPvmMessage::MessageTag tag) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PsnPvmSvm::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<<"PsnPvmSvm::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;
        }
      if (PsController::warningLevel >=  PsController::FatalWarnings) 
        {             
          exit (2) ;
        }
      cerr<<endl;
   }
  return res ;
}



pair<PsnPvmMessage::MessageTag, int> 
PsnPvmSvm::waitForAnyRequests (PsnPvmIncomingMessage & receiveBuffer) 
{
  pair<PsnPvmMessage::MessageTag, int> result ;
  
  int bufid = pvm_recv ( -1, -1 ) ; //receive anything

  if ( bufid < 0 ) 
    {
      cerr<<"PsnSvm::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";
          if (PsController::warningLevel >=  PsController::FatalWarnings) 
            {         
              exit (2) ;
            }
        }
      cerr<<endl;
    }

  int bytes ;
  int senderSiteId ;
  int msgtag ;
  int info = pvm_bufinfo( bufid, &bytes , &msgtag , &result.second  );
  result.first = static_cast<PsnPvmMessage::MessageTag>(msgtag) ;
  if ( info < 0 ) 
    {
      cerr<<"PsnSvm::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";
          if (PsController::warningLevel >=  PsController::FatalWarnings) 
            {         
              exit (2) ;
            }
        }
      cerr<<endl;
    }

  receiveBuffer.initialise ( bufid ) ;
  
  return result ;
}



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

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

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

void PsnPvmSvm::addNewWorkstation (const PsName & 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 ;

   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;
                 PsController::error( "PsnPvmSvm::addNewWorkstation: pb pour ajouter une machine");
               } 
             else 
               {
                 sleep (5) ;
               }
           } 
         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 PsnPvmSvm::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 PsnPvmSvm::spawnProcess (PsnProcess * 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 ;
   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 ;
         PsController::error ("PsnPvmSvm::spawnProcess: spawn problem") ;
   }
   return (int)tidCtrlLoc ;
}


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

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

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

PsnSvmLink * PsnPvmSvm::createSvmLink (const int & d) 
{
   PsnPvmSvmLink * canal = new PsnPvmSvmLink (d) ;
   return (PsnSvmLink*)canal;       
}

//--------------------------------------------------------------------
#endif

---