OMK::PvmSvm Class Reference

class encapsulating the PVM virtual machine More...

#include <OMKPvmSvm.h>

Inheritance diagram for OMK::PvmSvm:

[legend]Collaboration diagram for OMK::PvmSvm:

[legend]List of all members.

Public Member Functions
	PvmSvm (NameToPointerMap< Process > *tab, const Date &latence, const Date &timeOut, const int deadReckoningInterval, const bool yieldNeeded, int argc, char *argv[])
	constructor
virtual	~PvmSvm ()
	destructor
Static Public Attributes
static int	pvmDataEncoding
	define the encoding used when sending and receiveing messages
static int	pvmSpawnFlags
	define the flags used when spawning a new task using PVM
Protected Member Functions
virtual void	joinSvmGroup (std::string &groupName)
	joinSvmGroup.
virtual void	groupBarrier (std::string groupName, int numberToJoin)
	block the calling process until numberToJoin processes of group groupName have called syncDistributedSites
virtual void	broadcastToGroup (std::string groupName, PvmMessage::MessageTag tag)
	broadcast an empty tagged message to all members of a broadcast group
virtual int	nonblockingReceive (PvmMessage::MessageTag tag)
	try and receive message in a non-blocking fashion the received messages are lost
virtual void	initBeforeMessagePacking ()
	do any usefull initialisations before messages are packed
virtual std::pair< PvmMessage::MessageTag, int >	waitForAnyRequests (PvmIncomingMessage &receiveBuffer)
	wait for any message sent to this site.
virtual int	getSiteId ()
	get the id in the svm of this site
virtual int	getParentSiteId ()
	get the site id having spanned this process.
virtual void	addNewWorkstation (const Name &m)
	add a new machine to the virtual machine
virtual void	removeWorkstation (const Name &m)
	remove a work station from the virtual machine
virtual int	spawnProcess (Process *p)
	spawn a copy of this proc'ess on another site
virtual SvmLink *	createSvmLink (const int &d)
	create a link to a distant site
virtual bool	testIfSiteRecovered (Process *p)
	CHADI.
Protected Attributes
int	_argc
	remember the arguments to use when spawning: their number
char **	_argv
	remember the arguments to use when spawning: their contents

---

Detailed Description

class encapsulating the PVM virtual machine

Author:

Siames

Version:

2.0

• 1.0 -> 2.0 by David Margery: major overhaul of the API and code

Definition at line 32 of file OMKPvmSvm.h.

---

Constructor & Destructor Documentation

PvmSvm::PvmSvm	(	NameToPointerMap< Process > *	tab,
		const Date &	latence,
		const Date &	timeOut,
		const int	deadReckoningInterval,
		const bool	yieldNeeded,
		int	argc,
		char *	argv[]
	)

constructor

Parameters:

	tab	a table containing the a map between a processName and the data structure definig a process
	latence	the authorized synchronisation latency
	argc	the number of arguments given at the command line
	argv	the arguments given at the command line

Definition at line 40 of file OMKPvmSvm.cxx.

References _argc, _argv, and OMK::Svm::_siteId.

   : 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

(

)

[virtual]

destructor

Definition at line 62 of file OMKPvmSvm.cxx.

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

}

---

Member Function Documentation

void PvmSvm::joinSvmGroup

(

std::string &

groupName

)

[protected, virtual]

joinSvmGroup.

join a group of processes in the siames virtual machine

Parameters:

groupName

the name of the group of processes to join. This parameter is not const, because of the underlying C compatibility layer

Implements OMK::Svm.

Definition at line 72 of file OMKPvmSvm.cxx.

References OMASSERT.

{
  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
	)			[protected, virtual]

block the calling process until numberToJoin processes of group groupName have called syncDistributedSites

Parameters:

	numberToJoin	number of processes to wait for
	groupName	the name of the process group

Implements OMK::Svm.

Definition at line 98 of file OMKPvmSvm.cxx.

References OMASSERT.

{
   //   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
	)			[protected, virtual]

broadcast an empty tagged message to all members of a broadcast group

Implements OMK::Svm.

Definition at line 130 of file OMKPvmSvm.cxx.

References OMASSERT.

{
  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;
    }
}

int PvmSvm::nonblockingReceive

(

PvmMessage::MessageTag

tag

)

[protected, virtual]

try and receive message in a non-blocking fashion the received messages are lost

Parameters:

tag

the tag of the messages to receive

Returns:

number of received messages with that tag

Implements OMK::Svm.

Definition at line 177 of file OMKPvmSvm.cxx.

References OMASSERT.

{
#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 ;
}

void PvmSvm::initBeforeMessagePacking

(

)

[protected, virtual]

do any usefull initialisations before messages are packed

Implements OMK::Svm.

Definition at line 155 of file OMKPvmSvm.cxx.

References OMASSERT, and pvmDataEncoding.

{
  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;
    }
}

std::pair< PvmMessage::MessageTag, int > PvmSvm::waitForAnyRequests

(

PvmIncomingMessage &

receiveBuffer

)

[protected, virtual]

wait for any message sent to this site.

This call is blocking

Parameters:

receiveBuffer

: the user buffer to use to receive any message a pair giving the tag of the received message and the siteId of the sender

Implements OMK::Svm.

Definition at line 212 of file OMKPvmSvm.cxx.

References OMK::PvmIncomingMessage::initialise(), and OMASSERT.

{
  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

(

)

[protected, virtual]

get the id in the svm of this site

Implements OMK::Svm.

Definition at line 268 of file OMKPvmSvm.cxx.

References OMK::Svm::_siteId.

                       {
  assert ( _siteId >= 0 ) ;
  return _siteId ; 
}

int PvmSvm::getParentSiteId

(

)

[protected, virtual]

get the site id having spanned this process.

Returns:

the site id, 0 meaning the process wasn't spawned by the virtual machine

Implements OMK::Svm.

Definition at line 336 of file OMKPvmSvm.cxx.

References pvmDataEncoding.

                             {
   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 ;
     }
}

void PvmSvm::addNewWorkstation

(

const Name &

m

)

[protected, virtual]

add a new machine to the virtual machine

Implements OMK::Svm.

Definition at line 275 of file OMKPvmSvm.cxx.

References OMK::Svm::_numberOfSlaves, OMK::Controller::error(), and OMK::Name::getCString().

                                              {
   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;
}

void PvmSvm::removeWorkstation

(

const Name &

m

)

[protected, virtual]

remove a work station from the virtual machine

Implements OMK::Svm.

Definition at line 415 of file OMKPvmSvm.cxx.

References OMK::Name::getCString().

{
   int res ;
   char * mac [2] ;
   mac [0]= (char *)(m.getCString ()) ;
   mac [1] = (char *)NULL ;
   
   pvm_delhosts (mac, 1, &res) ;
}

int PvmSvm::spawnProcess

(

Process *

p

)

[protected, virtual]

spawn a copy of this proc'ess on another site

Implements OMK::Svm.

Definition at line 359 of file OMKPvmSvm.cxx.

References _argc, _argv, OMK::Controller::error(), OMK::Name::getCString(), OMK::Process::getHostMachineName(), and pvmSpawnFlags.

{
   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 ;
}

SvmLink * PvmSvm::createSvmLink

(

const int &

d

)

[protected, virtual]

create a link to a distant site

Parameters:

d

the site id of the distant site

Implements OMK::Svm.

Definition at line 427 of file OMKPvmSvm.cxx.

{
   PvmSvmLink * canal = new PvmSvmLink (d) ;
   return (SvmLink*)canal;       
}

bool PvmSvm::testIfSiteRecovered

(

Process *

p

)

[protected, virtual]

CHADI.

Implements OMK::Svm.

Definition at line 436 of file OMKPvmSvm.cxx.

References OMK::Process::getHostMachineName(), OMK::Process::getSvmLink(), and OMK::SvmLink::getTID().

{
   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;
 }

---

Member Data Documentation

int PvmSvm::pvmDataEncoding [static]

define the encoding used when sending and receiveing messages

Definition at line 117 of file OMKPvmSvm.h.

Referenced by OMK::PvmNameServer::getIdentifierAsFrom(), getParentSiteId(), OMK::PvmNameServer::getStringAssociatedToOnCentralServer(), initBeforeMessagePacking(), OMK::PvmNameServer::PvmNameServer(), OMK::PvmOutgoingMessage::PvmOutgoingMessage(), OMK::PvmOutgoingMessage::reinitAndRevertPvmContext(), OMK::Svm::serveNameRequestsUntilEnd(), and OMK::PvmCentralNameServer::verifyCompatibilityWithLocalNameServer().

int PvmSvm::pvmSpawnFlags [static]

define the flags used when spawning a new task using PVM

Definition at line 121 of file OMKPvmSvm.h.

Referenced by spawnProcess().

int OMK::PvmSvm::_argc [protected]

remember the arguments to use when spawning: their number

Definition at line 126 of file OMKPvmSvm.h.

Referenced by PvmSvm(), and spawnProcess().

char** OMK::PvmSvm::_argv [protected]

remember the arguments to use when spawning: their contents

Definition at line 130 of file OMKPvmSvm.h.

Referenced by PvmSvm(), and spawnProcess().

---