---

PsnSvm Class Reference

Defines an abstract message passing virtual machine. More...

#include <PsnSvm.h>

Inheritance diagram for PsnSvm:

[legend]Collaboration diagram for PsnSvm:

[legend]List of all members.

Public Methods
	PsnSvm (PsNameToPointerMap< PsnProcess > *tab, const PsDate &latence)
	constructor
virtual	~PsnSvm ()
	Destructor.
virtual void	init (const PsDate &initialSimulationDate)
	initialise the distributed virtual machine
virtual void	syncDistributedSites ()
	sync all the distributed sites.
virtual void	connectToDistributedSimulation ()
	connect the distributed virtual machine to its peers
virtual void	createDistributedSimulation (const PsDate &initialSimulationDate)
	create a distributed simulation
virtual void	broadcast (PsnPvmMessage::MessageTag, PsnPvmMulticastMessage *message=NULL)
	broadcast a message
virtual void	synchronizeOn (PsPvmController &, PsnPvmMessage::MessageTag tag)
	enter a synchronisation barrier, between all distributed sites except the central site this is a blocking call, but the virtual machine still answer requests.
virtual void	disconnectFromDistributedSimulation (const PsDate &deconnexionDate)
	Disconnect the site from the distributed simulation.
virtual const PsName &	getSiteName () const
	get the site name in the simulation
virtual PsnSvmLink *	getLinkToProcessNamed (const PsName &processName)
	get acces to the link to a given process
virtual PsnProcess *	getProcessDescriptorNamed (const PsName &processName)
	get acces to a given process descriptor
void	processReceivedMessages (PsPvmController &parsingController, const PsnPvmMessage::MessageTag tag)
	Attempt to receive and process messages from all sites, in a non blocking fashion.
void	waitAndProcessMessages (PsPvmController &parsingController, const PsnPvmMessage::MessageTag tag)
	receive and process messages from all sites, and wait for the messages to arrive if necessary
void	waitForMessage (PsPvmController &parsingController, const PsnPvmMessage::MessageTag tag)
	wait until a message has arrived from one site
void	waitForMessageFrom (PsPvmController &parsingController, const PsName &processName, const PsnPvmMessage::MessageTag tag)
	receive a message from a particular process
void	synchroniseReceiveAndProcessMessages (PsPvmController &parsingController, const PsnPvmMessage::MessageTag tag)
	receive and process messages from all sites (except the central site), and wait until at least a recent (as defined by latency) message has arrived from each of these sites
PsnProcess *	waitForAnswerToBlockingRequest (PsPvmController &parsingController, PsnPvmMessage::MessageTag tag)
	receive and process messages from all sites (except the central site), and wait until at least on message of tag tag has arrived
void	sendCurrentBuffersWithTag (const PsnPvmMessage::MessageTag tag)
	send the messages contained in the send buffers of all links
void	timestampCurrentSendBuffers (const PsDate &date)
	timestamp the messages contained in the send buffers of all links
virtual const PsDate &	getSynchronisationLatency ()
	get the authorized latency for the relaxed receive
Protected Methods
virtual void	addNewWorkstation (const PsName &m)=0
	add a new machine to the virtual machine
virtual void	removeWorkstation (const PsName &m)=0
	remove a work station from the virtual machine
virtual int	spawnProcess (PsnProcess *p)=0
	spawn a copy of this proc'ess on another site
virtual PsnSvmLink *	createSvmLink (const int &d=0)=0
	create a link to a distant site
virtual int	getParentSiteId ()=0
	get the site id having spanned this process.
virtual int	getSiteId ()=0
	get the id in the svm of this site
virtual void	groupBarrier (string groupName, int numberToJoin)=0
	block the calling process until numberToJoin processes of group groupName have called groupBarrier
virtual void	broadcastToGroup (string groupName, PsnPvmMessage::MessageTag)=0
	broadcast an empty tagged message to all members of a broadcast group
virtual void	initBeforeMessagePacking ()=0
	do any usefull initialisations before messages are packed
virtual int	nonblockingReceive (PsnPvmMessage::MessageTag tag)=0
	try and receive message in a non-blocking fashion the received messages are lost
virtual pair< PsnPvmMessage::MessageTag, int >	waitForAnyRequests (PsnPvmIncomingMessage &receiveBuffer)=0
	wait for any message sent to this site.
virtual void	joinSvmGroup (string &groupName)=0
	join a synchronisation group join a group of processes in the siames virtual machine
virtual void	serveNameRequestsUntilEnd ()
	for the central site: go into server mode, receiving and processing any reveived request until all sites have disconnected
Protected Attributes
PsNameToPointerMap< PsnProcess > *	_processTable
	map containing all information about the different processes
map< int, PsnProcess * >	_idToProcessTable
	map storing correspondance between processId and PsnProcess
int	_siteId
	workstation identifier in the svm
PsnSvmLink *	_linkToCentralSite
	the link to the central site
PsName	_siteName
	Name of the process associated to this site.
int	_numberOfSlaves
	number of slaves ( PvmControllers ) in the simulation
const PsDate	_synchronisationLatency
	in millisecond, the authorized latency when synchronising all the sites
Static Protected Attributes
string	_slaveGroupName = "spawnedControllers"
	name of the group all the slave processes join
string	_masterSiteName = "OpenMaskMasterDistributedSite"
	reserved name of the master distributed site

---

Detailed Description

Defines an abstract message passing virtual machine.

Author:

Siames

Version:

2.1

• 1.0 -> 2.0 by David Margery (complete overhaul)
• 2.0 -> 2.1 (by David Margery 2002/09/04): added support for switch to server mode while waiting for a blocking request

Definition at line 40 of file PsnSvm.h.

---

Constructor & Destructor Documentation

PsnSvm::PsnSvm (  PsNameToPointerMap< PsnProcess > *    tab, const PsDate &    latence )

constructor Parameters: tab  a table containing the a map between a processName and the data structure definig a process latence  the authorized synchronisation latency Definition at line 38 of file PsnSvm.cxx. References PsDate. : _processTable ( tab ), _synchronisationLatency ( latence ), _siteId ( -1 ), //negative values indicates an error _numberOfSlaves ( 0 ) { }

PsnSvm::~PsnSvm (    )  [virtual]

Destructor. Definition at line 47 of file PsnSvm.cxx. { }

---

Member Function Documentation

virtual void PsnSvm::addNewWorkstation (  const PsName &    m )  [protected, pure virtual]

add a new machine to the virtual machine Implemented in PsnPvmSvm. Referenced by createDistributedSimulation().

void PsnSvm::broadcast (  PsnPvmMessage::MessageTag    tag, PsnPvmMulticastMessage *    mess = NULL )  [virtual]

broadcast a message reusing mess Definition at line 337 of file PsnSvm.cxx. References _processTable, _siteName, map< PsName, PsnProcess * >::begin(), map< PsName, PsnProcess * >::end(), PsnPvmOutgoingMessage::insertTimeStamp(), PsnPvmMessage::MessageTag, and vector< T, Alloc >::push_back(). Referenced by synchronizeOn(). { //the first implementation, commented, dies in pvm with an unexpected error //I haven't had time to trace the problem, so the second implementation is used /* first implementation initBeforeMessagePacking () ; if ( mess != NULL ) { mess->packMessage() ; } else { mess = new PsnPvmOutgoingMessage () ; mess->insertTimeStamp(0) ; *mess << "dummy"; mess->packMessage() ; } broadcastToGroup ( slaveGroup, tag ) ; */ #ifdef _DEBUGPVMMESS cerr<<"PsnSvm::broadcast ("<<tag<<","<<mess<<") ; "<<endl; #endif PsNameToPointerMap<PsnProcess>::iterator processIter ; PsnSvmLink * canal ; // Canal vers le precedent processus if (mess == NULL ) { vector<int> distantRecepients ; for (processIter = _processTable->begin () ; processIter != _processTable->end () ; ++processIter) { if (_siteName != (*processIter).second->getProcessName ()) { // On connait alors le canal a observer distantRecepients.push_back ( (*processIter).second->getSvmLink ()->getTID () ) ; } } mess = new PsnPvmMulticastMessage ( distantRecepients ) ; mess->insertTimeStamp(0) ; } mess -> send ( tag ) ; }

virtual void PsnSvm::broadcastToGroup (  string    groupName, PsnPvmMessage::MessageTag    )  [protected, pure virtual]

broadcast an empty tagged message to all members of a broadcast group Implemented in PsnPvmSvm.

void PsnSvm::connectToDistributedSimulation (    )  [virtual]

connect the distributed virtual machine to its peers Definition at line 242 of file PsnSvm.cxx. References _idToProcessTable, _linkToCentralSite, _numberOfSlaves, _processTable, _siteName, _slaveGroupName, map< PsName, PsnProcess * >::begin(), createSvmLink(), map< PsName, PsnProcess * >::end(), PsName::getNameServer(), getParentSiteId(), PsnProcess::getSvmLink(), PsnSvmLink::getTID(), map< int, PsnProcess * >::insert(), joinSvmGroup(), make_pair(), PsnPvmMessage::ProcessTable, PsDate, PsName::setNameServer(), PsnProcess::setSvmLink(), PsnPvmMessage::SiteName, syncDistributedSites(), and PsnSvmLink::waitForMessage(). Referenced by init(). { #ifdef _DEBUGPVMEXEC cerr<<"PsnSvm::connectToDistributedSimulation (): connecting to the distributed simulation"<<endl; #endif PsnPvmIncomingMessage * np = NULL ; // connect to the global controller _linkToCentralSite = createSvmLink ( getParentSiteId () ) ; // join the group of slave processes joinSvmGroup (_slaveGroupName) ; // create a distributed version of the nameServer PsPvmNameServer * pvmNameServer = new PsPvmNameServer ( getParentSiteId() , *PsName::getNameServer() ) ; PsName::setNameServer ( pvmNameServer ) ; // find out the name of this site PsDate initialDate ; np = & _linkToCentralSite->waitForMessage (PsnPvmMessage::SiteName) ; *np >> _siteName ; // find out the adresses of the other sites // - extract the timestamp of the communication _linkToCentralSite->waitForMessage (PsnPvmMessage::ProcessTable) ; // - extract, for each process, it's adress in the virtual machine and create a link PsNameToPointerMap<PsnProcess>::iterator i ; for (i = _processTable->begin () ; i != _processTable->end () ; i ++) { ++_numberOfSlaves ; //we add ourselves, but as _numberOfSlaves is initialised to 0, it's ok // Remarque : creer un canal sur soi-meme n'est pas genant, // on met juste a jour les tids dans la table des pss PsnProcess * p = (*i).second ; // creation d'un canal sans destinataire p->setSvmLink (createSvmLink ()) ; // maj du destinataire *np >> *p ; _idToProcessTable.insert ( make_pair ( p->getSvmLink()->getTID(),p ) ) ; } //wait for all other sites to be connected to the distributed simulation. //avoids sending messages to sites still connected or worse, not yet connected syncDistributedSites () ; }

void PsnSvm::createDistributedSimulation (  const PsDate &    initialSimulationDate )  [virtual]

create a distributed simulation Parameters: initialSimulationDate:  the initial simualtion date needed to start timestamping mexchanged messages Definition at line 80 of file PsnSvm.cxx. References _idToProcessTable, _masterSiteName, _processTable, _siteName, addNewWorkstation(), map< PsName, PsnProcess * >::begin(), createSvmLink(), map< PsName, PsnProcess * >::end(), PsnSvmLink::getOutgoingBuffer(), PsnProcess::getSvmLink(), PsnSvmLink::getTID(), map< int, PsnProcess * >::insert(), PsnPvmOutgoingMessage::insertTimeStamp(), make_pair(), PsnPvmMessage::ProcessTable, PsDate, vector< T, Alloc >::push_back(), PsnPvmMulticastMessage::send(), PsnPvmOutgoingMessage::send(), serveNameRequestsUntilEnd(), PsnProcess::setSvmLink(), PsnPvmMessage::SiteName, and spawnProcess(). Referenced by init(). { #ifdef _DEBUGPVMEXEC cerr << "PsnSvm::createDistributedSimulation : creating the distributed simulation" << endl ; #endif _siteName = _masterSiteName ; PsNameToPointerMap<PsnProcess>::iterator i ; for (i = _processTable->begin () ; i != _processTable->end () ; ++i ) { addNewWorkstation ( (*i).second->getHostMachineName () ) ; } // creation des processus locaux #ifdef _DEBUGPVMEXEC cerr << "Creation des processus locaux..................." << endl ; #endif PsnProcess * processInfo ; int adrLoc ; vector<int> processSiteIds ; for (i = _processTable->begin () ; i != _processTable->end () ; i ++) { processInfo = (*i).second ; // spawn a new process on the specified workstation adrLoc = spawnProcess (processInfo) ; // create a SvmLink, add it to processInfo processInfo->setSvmLink (createSvmLink (adrLoc)) ; _idToProcessTable.insert ( make_pair ( adrLoc,processInfo ) ) ; processInfo->getSvmLink ()->getOutgoingBuffer().insertTimeStamp ( initialSimulationDate ) ; // send the name of the process (in the Svm) to the created process processInfo->getSvmLink ()->getOutgoingBuffer()<<(*i).first ; processInfo->getSvmLink ()->getOutgoingBuffer().send (PsnPvmMessage::SiteName) ; //prepare processSiteIds for the next step processSiteIds.push_back ( processInfo->getSvmLink ()->getTID() ) ; #ifdef _DEBUGPVMMESS cerr<<typeid((*i).first).name()<<endl; cerr<<(*i).first<<" of id "<<processInfo->getSvmLink ()->getTID()<<endl; #endif } // Encodage des attributs Svm de chaque processus PsnPvmMulticastMessage nameToIdMessage ( processSiteIds ) ; nameToIdMessage.insertTimeStamp ( initialSimulationDate ) ; for (i = _processTable->begin () ; i != _processTable->end () ; i ++) { nameToIdMessage << *((*i).second) ; } // Transfert des adresses vers chaque controleur local nameToIdMessage.send ( PsnPvmMessage::ProcessTable ) ; serveNameRequestsUntilEnd () ; }

virtual PsnSvmLink* PsnSvm::createSvmLink (  const int &    d = 0 )  [protected, pure virtual]

create a link to a distant site Parameters: d  the site id of the distant site Implemented in PsnPvmSvm. Referenced by connectToDistributedSimulation(), and createDistributedSimulation().

void PsnSvm::disconnectFromDistributedSimulation (  const PsDate &    deconnexionDate )  [virtual]

Disconnect the site from the distributed simulation. Parameters: deconnexionDate.  date at wich the disconnection is happening (to timestamp the disconnection message) This will trigger diconnection on a sites. Definition at line 295 of file PsnSvm.cxx. References _processTable, map< PsName, PsnProcess * >::begin(), PsnPvmMessage::EndOfSimulation, getParentSiteId(), getSiteName(), PsnPvmOutgoingMessage::insertTimeStamp(), PsDate, vector< T, Alloc >::push_back(), PsnPvmMulticastMessage::send(), and PsnPvmUnicastMessage::send(). Referenced by serveNameRequestsUntilEnd(). { #ifdef _DEBUGPVMEXEC cout <<"PsnSvm::disconnectFromDistributedSimulation"<< endl ; #endif int recepient = getParentSiteId () ; if ( recepient != 0 ) // a local PvmController has finished. Notify the central controller { PsnPvmUnicastMessage endMessage (recepient) ; endMessage.insertTimeStamp ( endOfSimulationDate ) ; endMessage << getSiteName () ; endMessage.send ( PsnPvmMessage::EndOfSimulation ) ; } else //the central controller is deconnecting. Notify all local controllers { vector<int> processSiteIds ; for (PsNameToPointerMap<PsnProcess>::iterator i = _processTable->begin () ; i != _processTable-> end () ; i ++) { processSiteIds.push_back ( (i->second)->getSvmLink ()->getTID() ) ; } PsnPvmMulticastMessage endOfSimulationMessage ( processSiteIds ) ; endOfSimulationMessage.insertTimeStamp ( endOfSimulationDate ) ; endOfSimulationMessage << getSiteName () ; endOfSimulationMessage.send ( PsnPvmMessage::EndOfSimulation ) ; } }

PsnSvmLink * PsnSvm::getLinkToProcessNamed (  const PsName &    processName )  [virtual]

get acces to the link to a given process Definition at line 424 of file PsnSvm.cxx. References _processTable, map< PsName, PsnProcess * >::end(), map< PsName, PsnProcess * >::find(), PsNameToPointerMap< PsnProcess >::getObjectOfIndex(), PsnProcess::getSvmLink(), PsController::SomeWarnings, and PsController::warning(). Referenced by waitForMessageFrom(). { // On recupere le pss associe a l'objet if (_processTable->find (nomPss) != _processTable->end ()) { // On a bien trouve le processus PsnProcess * pss = _processTable->getObjectOfIndex (nomPss) ; return (pss->getSvmLink ()) ; } else { // On n'a pas le processus demande #ifdef _USESSTREAM ostringstream o ; o << "PsnSvm::getLinkToProcessNamed : Impossible de trouver le processus " << nomPss << " afin de recuperer son canal !" ; PsController::warning (o.str(), PsController::SomeWarnings ) ; #else ostrstream o ; o << "PsnSvm::getLinkToProcessNamed : Impossible de trouver le processus " << nomPss << " afin de recuperer son canal !" ; o.put ('\0') ; PsController::warning (o.str(), PsController::SomeWarnings ) ; delete o.str() ; #endif return NULL ; } }

virtual int PsnSvm::getParentSiteId (    )  [protected, pure virtual]

get the site id having spanned this process. Returns: the site id, 0 meaning the process wasn't spawned by the virtual machine Implemented in PsnPvmSvm. Referenced by connectToDistributedSimulation(), disconnectFromDistributedSimulation(), and init().

PsnProcess * PsnSvm::getProcessDescriptorNamed (  const PsName &    processName )  [virtual]

get acces to a given process descriptor Definition at line 454 of file PsnSvm.cxx. References _processTable, map< PsName, PsnProcess * >::end(), map< PsName, PsnProcess * >::find(), PsNameToPointerMap< PsnProcess >::getObjectOfIndex(), PsController::SomeWarnings, and PsController::warning(). { // On recupere le pss associe a l'objet if (_processTable->find (nomPss) != _processTable->end ()) { // On a bien trouve le processus PsnProcess * pss = _processTable->getObjectOfIndex (nomPss) ; return (pss) ; } else { // On n'a pas le processus demande #ifdef _USESSTREAM ostringstream o ; o << "PsnSvm::getProcessDescriptorNamed: process " << nomPss << " unknown !" ; PsController::warning (o.str(), PsController::SomeWarnings ) ; #else ostrstream o ; o << "PsnSvm::getProcessDescriptorNamed: process " << nomPss << " unknown !" ; o.put('\0') ; PsController::warning (o.str(), PsController::SomeWarnings ) ; delete o.str() ; #endif return NULL ; } }

virtual int PsnSvm::getSiteId (    )  [protected, pure virtual]

get the id in the svm of this site Implemented in PsnPvmSvm.

const PsName & PsnSvm::getSiteName (    )  const [virtual]

get the site name in the simulation Definition at line 805 of file PsnSvm.cxx. References _siteName. Referenced by disconnectFromDistributedSimulation(). { return _siteName; }

const PsDate & PsnSvm::getSynchronisationLatency (    )  [virtual]

get the authorized latency for the relaxed receive Definition at line 52 of file PsnSvm.cxx. References _synchronisationLatency, and PsDate. { return _synchronisationLatency; }

virtual void PsnSvm::groupBarrier (  string    groupName, int    numberToJoin )  [protected, pure virtual]

block the calling process until numberToJoin processes of group groupName have called groupBarrier Parameters: numberToJoin  number of processes to wait for groupName  the name of the process group Implemented in PsnPvmSvm. Referenced by syncDistributedSites().

void PsnSvm::init (  const PsDate &    initialSimulationDate )  [virtual]

initialise the distributed virtual machine Definition at line 59 of file PsnSvm.cxx. References connectToDistributedSimulation(), createDistributedSimulation(), getParentSiteId(), and PsDate. { if ( getParentSiteId () == 0 ) // wasn't created by a spawn { createDistributedSimulation ( initialSimulationDate ) ; } else { connectToDistributedSimulation () ; } }

virtual void PsnSvm::initBeforeMessagePacking (    )  [protected, pure virtual]

do any usefull initialisations before messages are packed Implemented in PsnPvmSvm.

virtual void PsnSvm::joinSvmGroup (  string &    groupName )  [protected, pure virtual]

join a synchronisation group 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 Implemented in PsnPvmSvm. Referenced by connectToDistributedSimulation().

virtual int PsnSvm::nonblockingReceive (  PsnPvmMessage::MessageTag    tag )  [protected, pure 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 Implemented in PsnPvmSvm.

void PsnSvm::processReceivedMessages (  PsPvmController &    parsingController, const PsnPvmMessage::MessageTag    tag )

Attempt to receive and process messages from all sites, in a non blocking fashion. Parameters: parsingController  the controller wich will parse the received messages tag  the tag of the messages to receive Definition at line 484 of file PsnSvm.cxx. References _processTable, _siteName, map< PsName, PsnProcess * >::begin(), map< PsName, PsnProcess * >::end(), PsnPvmMessage::getSize(), PsnPvmIncomingMessage::hasMessage(), PsnPvmMessage::MessageTag, PsPvmController::parseSynchronisationMessage(), and PsnSvmLink::testForMessage(). { #ifdef _DEBUGPVMMESS cerr << "PsnSvm::processReceivedMessages"<<endl; #endif PsNameToPointerMap<PsnProcess>::iterator pPss ; // Iterateur pour parcours tableau PsnSvmLink * canal ; // Canal vers le precedent processus PsnPvmIncomingMessage * messageRecu = NULL ; for (pPss = _processTable->begin () ; pPss != _processTable->end () ; pPss ++) { if (_siteName != (*pPss).second->getProcessName ()) { // On connait alors le canal a observer canal = (*pPss).second->getSvmLink () ; #ifdef _DEBUGPVMMESS cout << "PsnSvm::processReceivedMessages from " << (*pPss).second->getProcessName () << " tag : " << tag << endl ; //canal->affiche () ; #endif // On fait une reception non bloquante messageRecu = canal->testForMessage (tag) ; while ( messageRecu->hasMessage () ) { #ifdef _DEBUGPVMMESS cerr << "PsnSvm::processReceivedMessages : message received of size " <<messageRecu->getSize()<< endl; #endif #ifdef _DEBUGPVMMESS cerr << "PsnSvm::processReceivedMessages: extracted emission date: " <<messageREcu->getMessageDate()<< endl; #endif //cerr << "PsnSvm::processReceivedMessages : date " << dateEmission << endl ; // On envoie le message et l'objet au controleur local parsingController.parseSynchronisationMessage (messageRecu) ; messageRecu = canal->testForMessage (tag) ; } } } }

virtual void PsnSvm::removeWorkstation (  const PsName &    m )  [protected, pure virtual]

remove a work station from the virtual machine Implemented in PsnPvmSvm.

void PsnSvm::sendCurrentBuffersWithTag (  const PsnPvmMessage::MessageTag    tag )

send the messages contained in the send buffers of all links Parameters: tag  the tag to associate to the message Definition at line 715 of file PsnSvm.cxx. References _processTable, _siteName, map< PsName, PsnProcess * >::begin(), map< PsName, PsnProcess * >::end(), PsSynchronisationMessage::endOfSynchronisationFragment, PsnPvmMessage::MessageTag, PsnSvmLink::sendOutgoingBuffer(), and PsnPvmMessage::SynchronisationMessage. { PsNameToPointerMap<PsnProcess>::iterator pPss ; // Iterateur pour parcours tableau PsnSvmLink * canal ; // Canal vers le precedent processus #ifdef _DEBUGPVMMESS cerr<<"PsnSvm::sendCurrentBufferWithTag tag "<<tag<<endl<<"Processus : "<<endl; PsNameToPointerMap<PsnProcess>::iterator pPs ; for (pPs = _processTable->begin () ; pPs != _processTable->end () ; pPs ++) { cerr<<(*pPs).second->getProcessName (); } cerr<<endl; #endif for (pPss = _processTable->begin () ; pPss != _processTable->end () ; pPss ++) { #ifdef _DEBUGPVMMESS cerr<<"Traitement des emissions vers "<<(*pPss).second->getProcessName ()<<endl; #endif if (_siteName != (*pPss).second->getProcessName ()) { // On connait alors le canal a observer canal = (*pPss).second->getSvmLink () ; if ( tag == PsnPvmMessage::SynchronisationMessage ) { (*pPss).second->getSvmLink ()->getOutgoingBuffer()<<PsSynchronisationMessage::endOfSynchronisationFragment ; } // On transmet le message du canal canal->sendOutgoingBuffer ( tag ) ; #ifdef _DEBUGPVMMESS cerr << "PsmSvm::transmission avec tag : " << tag << endl ; #endif } #ifdef _DEBUGPVMMESS else { cerr << "Rien à faire "<<endl ; //canal->affiche () ; } #endif } }

void PsnSvm::serveNameRequestsUntilEnd (    )  [protected, virtual]

for the central site: go into server mode, receiving and processing any reveived request until all sites have disconnected Definition at line 153 of file PsnSvm.cxx. References _processTable, PsController::AllWarnings, disconnectFromDistributedSimulation(), PsnPvmMessage::EndOfSimulation, map< PsName, PsnProcess * >::erase(), PsController::FatalWarnings, pair< T1, T2 >::first, PsString::getCString(), PsNameServerT< STL_ALLOC >::getIdentifier(), PsnPvmIncomingMessage::getMessageDate(), PsName::getNameServer(), PsNameServerT< STL_ALLOC >::getStringAssociatedTo(), PsName::idType, PsnPvmMessage::NameServiceGetId, PsnPvmMessage::NameServiceGetString, PsnPvmMessage::NameServiceReturnId, PsnPvmMessage::NameServiceReturnString, PsnPvmMessage::NameServiceVerifyLocalNameServer, PsDate, PsnPvmSvm::pvmDataEncoding, pair< T1, T2 >::second, PsName::setNameServer(), map< PsName, PsnProcess * >::size(), PsPvmCentralNameServer::verifyCompatibilityWithLocalNameServer(), and waitForAnyRequests(). Referenced by createDistributedSimulation(). { bool serving = true ; //convert the actual name Server to a PvmCentralNameServer PsPvmCentralNameServer * nameServer = new PsPvmCentralNameServer ( *PsName::getNameServer() ) ; PsName::setNameServer (nameServer ) ; PsnPvmIncomingMessage receiveBuffer ; pair<PsnPvmMessage::MessageTag, int> receivedRequest ; while ( serving ) { receivedRequest = waitForAnyRequests ( receiveBuffer ) ; switch (receivedRequest.first) { case PsnPvmMessage::EndOfSimulation : { const PsDate & endDate (receiveBuffer.getMessageDate() ); PsName endedProcess ; receiveBuffer>>endedProcess ; cerr<<endedProcess<<" finished at "<<endDate<<endl; _processTable->erase ( endedProcess ) ; if ( _processTable->size() == 0 ) { disconnectFromDistributedSimulation ( endDate ) ; serving = false ; } } break; case PsnPvmMessage::NameServiceGetId : { int stringLength ; receiveBuffer>>stringLength ; char * stringId = new char[stringLength] ; receiveBuffer>>stringId ; assert ( stringId [stringLength - 1] == 0 ) ; cerr<<"Requested id for string "<<stringId<<" of length "<<stringLength <<" from "<< hex << receivedRequest.second << dec; PsName::idType id = nameServer->getIdentifier( PsString(stringId) ) ; cerr<<" sending "<<id<<endl; delete [] stringId ; pvm_initsend( PsnPvmSvm::pvmDataEncoding ) ; pvm_pklong (&id, 1, 1); pvm_send (receivedRequest.second,PsnPvmMessage::NameServiceReturnId) ; } break; case PsnPvmMessage::NameServiceGetString: { PsName::idType id ; receiveBuffer>>id ; cerr<<"Requested String for id "<<id <<" from "<< hex << receivedRequest.second<< dec ; const PsString & resultingString ( nameServer->getStringAssociatedTo (id) ) ; cerr<<" sending "<<resultingString<<endl; pvm_initsend( PsnPvmSvm::pvmDataEncoding ) ; pvm_pkstr (const_cast<char *> (resultingString.getCString() ) ) ; pvm_send (receivedRequest.second,PsnPvmMessage::NameServiceReturnString) ; } break; case PsnPvmMessage::NameServiceVerifyLocalNameServer: { nameServer->verifyCompatibilityWithLocalNameServer (receivedRequest.second, receiveBuffer) ; } break; default: //an unexpected request is received switch (PsController::warningLevel) { case PsController::AllWarnings : cerr<<"PsnSvm::serveNameRequestsUntilEnd() : unexpected request "<<receivedRequest.first<<endl; break ; case PsController::FatalWarnings : cerr<<"PsnSvm::serveNameRequestsUntilEnd() : unexpected request"<<receivedRequest.first<<endl; exit (2) ; break; default://nothing to do break; } break; } } }

virtual int PsnSvm::spawnProcess (  PsnProcess *    p )  [protected, pure virtual]

spawn a copy of this proc'ess on another site Implemented in PsnPvmSvm. Referenced by createDistributedSimulation().

void PsnSvm::syncDistributedSites (    )  [virtual]

sync all the distributed sites. blocks the calling process until all sites except the central site have called this member function Definition at line 417 of file PsnSvm.cxx. References _numberOfSlaves, _slaveGroupName, and groupBarrier(). Referenced by connectToDistributedSimulation(). { groupBarrier ( _slaveGroupName, _numberOfSlaves ) ; }

void PsnSvm::synchroniseReceiveAndProcessMessages (  PsPvmController &    parsingController, const PsnPvmMessage::MessageTag    tag )

receive and process messages from all sites (except the central site), and wait until at least a recent (as defined by latency) message has arrived from each of these sites Parameters: parsingController  the controller wich will parse the received messages tag  the tag of the messages to receive Definition at line 616 of file PsnSvm.cxx. References _processTable, _siteName, _synchronisationLatency, PsNameToPointerMap< typeObjet >::addObjectWithIndex(), PsController::AllWarnings, map< PsName, PsnProcess * >::begin(), map< PsName, typeObjet * >::empty(), map< PsName, PsnProcess * >::end(), PsnPvmMessage::EndOfSimulation, PsnPvmMessage::EnterBarrier, map< PsName, typeObjet * >::erase(), PsnPvmMessage::ExitBarrier, PsnPvmIncomingMessage::getMessageDate(), PsnProcess::getProcessName(), PsController::getSimulatedDate(), PsnProcess::getSvmLink(), PsnPvmIncomingMessage::hasMessage(), PsnPvmMessage::InitialValuesForMirror, PsnPvmMessage::LocalInitSuccessfull, PsnPvmMessage::MessageTag, PsnPvmMessage::MirrorNeedsInitialValues, PsnPvmMessage::NameServiceGetId, PsnPvmMessage::NameServiceGetString, PsnPvmMessage::NameServiceReturnId, PsnPvmMessage::NameServiceReturnString, PsnPvmMessage::NameServiceVerifyLocalNameServer, PsnPvmMessage::NameServiceVerifyResult, PsPvmController::parseSynchronisationMessage(), PsnPvmMessage::ProcessTable, PsPvmController::sendInitialValuesToMirror(), PsnProcess::setDateOfLastMessage(), PsnPvmMessage::SiteName, PsnPvmMessage::SynchronisationMessage, PsnSvmLink::testForAnyMessage(), and PsController::warning(). { PsNameToPointerMap<PsnProcess> aRecevoir ; //First make a list of processes that we need to receive values from for ( PsNameToPointerMap<PsnProcess>::iterator pPssE = _processTable->begin () ; pPssE != _processTable->end () ; pPssE ++) { if ( (pPssE->first != _siteName) && ( pPssE->second->getDateOfLastMessage () + pPssE->second->getPeriod () + _synchronisationLatency < parsingController.getSimulatedDate () ) ) { aRecevoir.addObjectWithIndex (pPssE->first, pPssE->second) ; #ifdef _DEBUGPVMMESS cerr << "PsnSvm::synchroniseReceiveAndProcessMessages: trying reception from: " << pPssE->first << endl ; #endif } } PsnPvmIncomingMessage receiveBuffer ; PsnPvmMessage::MessageTag receivedRequest ; PsnProcess * receivingProcess ; while (! aRecevoir.empty () ) { for ( PsNameToPointerMap<PsnProcess>::iterator processIterator = _processTable->begin () ; processIterator != _processTable->end () ; ++processIterator ) { //this new implementation doesn't do blocking receives, as some processes could send urgent requests receivingProcess = processIterator->second ; receivedRequest = receivingProcess->getSvmLink ()->testForAnyMessage ( receiveBuffer ) ; if ( receiveBuffer.hasMessage() ) { switch (receivedRequest) { case PsnPvmMessage::SynchronisationMessage : { //have the controller process the message parsingController.parseSynchronisationMessage ( &receiveBuffer ) ; receivingProcess->setDateOfLastMessage ( receiveBuffer.getMessageDate () ) ; //an additionnal test could be performed here to make sure correct date of last message has happened aRecevoir.erase ( receivingProcess->getProcessName() ) ; } break ; case PsnPvmMessage::MirrorNeedsInitialValues : { parsingController.sendInitialValuesToMirror ( receiveBuffer ) ; } break ; case PsnPvmMessage::InitialValuesForMirror : { //have the controller process the message parsingController.parseSynchronisationMessage ( &receiveBuffer ) ; } break ; case PsnPvmMessage::LocalInitSuccessfull : case PsnPvmMessage::ProcessTable : case PsnPvmMessage::EndOfSimulation : case PsnPvmMessage::EnterBarrier : case PsnPvmMessage::SiteName : case PsnPvmMessage::ExitBarrier : case PsnPvmMessage::NameServiceGetId : case PsnPvmMessage::NameServiceGetString : case PsnPvmMessage::NameServiceReturnId : case PsnPvmMessage::NameServiceReturnString : case PsnPvmMessage::NameServiceVerifyLocalNameServer : case PsnPvmMessage::NameServiceVerifyResult : default: { PsController::warning ("PsnSvm::waitForAnswerToBlockingRequest(): received unexpected message", PsController::AllWarnings ) ; } } } } } #ifdef _DEBUGPVMMESS cerr << "PsnSvm::synchroniseReceiveAndProcessMessages : on n'est pas bloque la " << endl ; #endif }

void PsnSvm::synchronizeOn (  PsPvmController &   , PsnPvmMessage::MessageTag    tag )  [virtual]

enter a synchronisation barrier, between all distributed sites except the central site this is a blocking call, but the virtual machine still answer requests. Definition at line 390 of file PsnSvm.cxx. References _numberOfSlaves, broadcast(), PsnPvmMessage::MessageTag, and waitForAnswerToBlockingRequest(). { broadcast ( tag ) ; #ifdef _DEBUGPVMSYNCHRO cerr<< "PsnSvm::waitForAllProcesses("<<tag<<") "<<_numberOfSlaves <<endl; usleep(1000000) ; #endif int numberOfReadyProcesses = 1 ; //this process has entered the barrier while ( numberOfReadyProcesses != _numberOfSlaves ) { // here, we ignore the results, because we suppose one process will only send tag once during synchronization waitForAnswerToBlockingRequest ( parsingController, tag ) ; numberOfReadyProcesses++ ; #ifdef _DEBUGPVMMESS cerr<< "PsnSvm::exitSoftBarrier numberToEnterBarrier: "<<numberOfReadyProcesses<<endl; usleep(1000000) ; #endif } }

void PsnSvm::timestampCurrentSendBuffers (  const PsDate &    date )

timestamp the messages contained in the send buffers of all links Definition at line 791 of file PsnSvm.cxx. References _processTable, _siteName, map< PsName, PsnProcess * >::begin(), map< PsName, PsnProcess * >::end(), and PsDate. { PsnPvmOutgoingMessage * message ; PsNameToPointerMap<PsnProcess>::iterator i ; for (i = _processTable->begin () ; i != _processTable->end () ; i ++) { if (_siteName != (*i).second->getProcessName ()) { i->second->getSvmLink ()->getOutgoingBuffer ().insertTimeStamp ( date ) ; } } }

void PsnSvm::waitAndProcessMessages (  PsPvmController &    parsingController, const PsnPvmMessage::MessageTag    tag )

receive and process messages from all sites, and wait for the messages to arrive if necessary Parameters: parsingController  the controller wich will parse the received messages tag  the tag of the messages to receive Definition at line 527 of file PsnSvm.cxx. References _processTable, _siteName, map< PsName, PsnProcess * >::begin(), map< PsName, PsnProcess * >::end(), PsnPvmMessage::MessageTag, PsPvmController::parseSynchronisationMessage(), and PsnSvmLink::waitForMessage(). { #ifdef _DEBUGPVMMESS cerr<<"PsnSvm::waitAndProcessMessages tag "<<tag<<endl<<"Processus : "<<endl; PsNameToPointerMap<PsnProcess>::iterator pPs ; for (pPs = _processTable->begin () ; pPs != _processTable->end () ; pPs ++) { cerr<<(*pPs).second->getProcessName (); } cerr<<endl; #endif PsNameToPointerMap<PsnProcess>::iterator pPss ; // Iterateur pour parcours tableau PsnSvmLink * canal ; // Canal vers le precedent processus PsnPvmIncomingMessage * messageRecu ; #ifdef _DEBUGPVMMESS cerr<<"PsnSvm::waitAndProcessMessages écoute de tous les processus"<<endl; #endif for (pPss = _processTable->begin () ; pPss!= _processTable->end () ; pPss ++) { #ifdef _DEBUGPVMMESS cerr<<"PsnSvm::waitAndProcessMessages écoute d'un de plus : "<<(*pPss).second->getProcessName ()<<endl; #endif if (_siteName != (*pPss).second->getProcessName ()) { // On connait alors le canal a observer canal = (*pPss).second->getSvmLink () ; #ifdef _DEBUGPVMMESS cout << "PsmSvm::waitAndProcessMessages sur " << (*pPss).second->getProcessName () << " tag : " << tag << endl ; canal->affiche () ; #endif // On recoit le message en bloquant messageRecu = & canal->waitForMessage (tag) ; #ifdef _DEBUGPVMMESS cerr << "PsnSvm::waitAndProcessMessages : message received"<< endl ; #endif // On envoie le message et l'objet au controleur local parsingController.parseSynchronisationMessage (messageRecu) ; } } }

PsnProcess * PsnSvm::waitForAnswerToBlockingRequest (  PsPvmController &    parsingController, PsnPvmMessage::MessageTag    tag )

receive and process messages from all sites (except the central site), and wait until at least on message of tag tag has arrived Parameters: parsingController  the controller wich will parse the received messages tag  the tag of the messages to receive Returns: a pointer to the process having sent the unblocking message Definition at line 812 of file PsnSvm.cxx. References _idToProcessTable, PsController::AllWarnings, map< int, PsnProcess * >::end(), PsnPvmMessage::EndOfSimulation, PsnPvmMessage::EnterBarrier, PsnPvmMessage::ExitBarrier, map< int, PsnProcess * >::find(), pair< T1, T2 >::first, PsnPvmIncomingMessage::getMessageDate(), PsnPvmMessage::InitialValuesForMirror, PsnPvmMessage::LocalInitSuccessfull, PsnPvmMessage::MessageTag, PsnPvmMessage::MirrorNeedsInitialValues, PsnPvmMessage::NameServiceGetId, PsnPvmMessage::NameServiceGetString, PsnPvmMessage::NameServiceReturnId, PsnPvmMessage::NameServiceReturnString, PsnPvmMessage::NameServiceVerifyLocalNameServer, PsnPvmMessage::NameServiceVerifyResult, PsPvmController::parseSynchronisationMessage(), PsnPvmMessage::ProcessTable, pair< T1, T2 >::second, PsPvmController::sendInitialValuesToMirror(), PsnProcess::setDateOfLastMessage(), PsnPvmMessage::SiteName, PsnPvmMessage::SynchronisationMessage, waitForAnyRequests(), and PsController::warning(). Referenced by synchronizeOn(). { // this implementation assumes that only oine thread at a time will enter this function // numberOfThread is an unsafe method to test this static int numberOfThreads = 0 ; ++numberOfThreads ; assert ( numberOfThreads == 1) ; PsnProcess * result ; PsnPvmIncomingMessage receiveBuffer ; pair<PsnPvmMessage::MessageTag, int> receivedRequest ; bool serving = true ; while ( serving ) { receivedRequest = waitForAnyRequests ( receiveBuffer ) ; //find the sender process (result of this member function) map<int,PsnProcess *>::iterator i = _idToProcessTable.find ( receivedRequest.second ) ; // the message should come from a known process assert ( i != _idToProcessTable.end() ) ; result = i->second ; switch (receivedRequest.first) { case PsnPvmMessage::SynchronisationMessage : { //have the controller process the message parsingController.parseSynchronisationMessage ( &receiveBuffer ) ; assert ( result != NULL ) ; result->setDateOfLastMessage ( receiveBuffer.getMessageDate () ) ; } break ; case PsnPvmMessage::MirrorNeedsInitialValues : { parsingController.sendInitialValuesToMirror ( receiveBuffer ) ; } break ; case PsnPvmMessage::InitialValuesForMirror : { //have the controller process the message parsingController.parseSynchronisationMessage ( &receiveBuffer ) ; } break ; case PsnPvmMessage::LocalInitSuccessfull : case PsnPvmMessage::ProcessTable : case PsnPvmMessage::EndOfSimulation : case PsnPvmMessage::EnterBarrier : case PsnPvmMessage::SiteName : case PsnPvmMessage::ExitBarrier : case PsnPvmMessage::NameServiceGetId : case PsnPvmMessage::NameServiceGetString : case PsnPvmMessage::NameServiceReturnId : case PsnPvmMessage::NameServiceReturnString : case PsnPvmMessage::NameServiceVerifyLocalNameServer : case PsnPvmMessage::NameServiceVerifyResult : default: { PsController::warning ("PsnSvm::waitForAnswerToBlockingRequest(): received unexpected message", PsController::AllWarnings ) ; } } serving = ( receivedRequest.first != tag ) ; } --numberOfThreads ; return result ; }

virtual pair<PsnPvmMessage::MessageTag, int> PsnSvm::waitForAnyRequests (  PsnPvmIncomingMessage &    receiveBuffer )  [protected, pure 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 Implemented in PsnPvmSvm. Referenced by serveNameRequestsUntilEnd(), and waitForAnswerToBlockingRequest().

void PsnSvm::waitForMessage (  PsPvmController &    parsingController, const PsnPvmMessage::MessageTag    tag )

wait until a message has arrived from one site Parameters: parsingController  the controller wich will parse the received messages tag  the tag of the messages to receive Definition at line 570 of file PsnSvm.cxx. References _linkToCentralSite, _processTable, _siteName, map< PsName, PsnProcess * >::begin(), map< PsName, PsnProcess * >::end(), PsnPvmIncomingMessage::hasMessage(), PsnPvmMessage::MessageTag, PsPvmController::parseSynchronisationMessage(), and PsnSvmLink::testForMessage(). { PsNameToPointerMap<PsnProcess>::iterator pPss ; // Iterateur pour parcours tableau PsnSvmLink * canal ; // Canal vers le precedent processus PsnPvmIncomingMessage * messageRecu = NULL ; bool rienRecu=true; while (rienRecu) {//on ne teste pas la nullite du message, //car en cas de message ne contenant que la date, on le vide ! for (pPss = _processTable->begin () ; pPss!= _processTable->end () ; pPss ++) { if (_siteName != (*pPss).second->getProcessName ()) { // On connait alors le canal a observer canal = (*pPss).second->getSvmLink () ; #ifdef _DEBUGPVMMESS cout << "PsmSvm::waitForMessage sur " << (*pPss).second->getProcessName () << " tag : " << tag << endl ; canal->affiche () ; #endif // On recoit le message en bloquant messageRecu = canal->testForMessage (tag) ; // On envoie le message et l'objet au controleur local if (messageRecu->hasMessage() ) { #ifdef _DEBUGPVMMESS cerr << "PsnSvm::waitForMessage : message received" << endl ; #endif parsingController.parseSynchronisationMessage (messageRecu) ; rienRecu=false; } } } messageRecu = _linkToCentralSite->testForMessage (tag) ; //cerr << "PsnSvm::waitForMessage : message recu : " << *messageRecu << endl ; if (messageRecu->hasMessage() ) { #ifdef _DEBUGPVMMESS cerr << "PsnSvm::waitForMessage : message received" << endl ; #endif parsingController.parseSynchronisationMessage (messageRecu) ; rienRecu=false; } } }

void PsnSvm::waitForMessageFrom (  PsPvmController &    parsingController, const PsName &    processName, const PsnPvmMessage::MessageTag    tag )

receive a message from a particular process Parameters: parsingController  the controller wich will parse the received messages tag  the tag of the messages to receive processName  the name of the process to receive a message from Definition at line 764 of file PsnSvm.cxx. References getLinkToProcessNamed(), PsnPvmMessage::MessageTag, PsPvmController::parseSynchronisationMessage(), and PsnSvmLink::waitForMessage(). { PsnSvmLink * canal ; // Canal vers le precedent processus PsnPvmIncomingMessage * messageRecu = NULL ; // On recupere le canal a observer canal = getLinkToProcessNamed (nomPss) ; #ifdef _DEBUGPVMMESS cout << "PsmSvm::waitForMessageFrom sur " << nomPss << " tag : " << tag << endl ; canal->affiche () ; #endif // On recoit le message en bloquant messageRecu = & canal->waitForMessage (tag) ; // On envoie le message et l'objet au controleur local parsingController.parseSynchronisationMessage (messageRecu) ; // On libere le message ??? //delete messageRecu; }

---

Member Data Documentation

map<int,PsnProcess *> PsnSvm::_idToProcessTable [protected]

map storing correspondance between processId and PsnProcess Definition at line 240 of file PsnSvm.h. Referenced by connectToDistributedSimulation(), createDistributedSimulation(), and waitForAnswerToBlockingRequest().

PsnSvmLink* PsnSvm::_linkToCentralSite [protected]

the link to the central site Definition at line 248 of file PsnSvm.h. Referenced by connectToDistributedSimulation(), and waitForMessage().

string PsnSvm::_masterSiteName = "OpenMaskMasterDistributedSite" [static, protected]

reserved name of the master distributed site Definition at line 35 of file PsnSvm.cxx. Referenced by createDistributedSimulation().

int PsnSvm::_numberOfSlaves [protected]

number of slaves ( PvmControllers ) in the simulation Definition at line 257 of file PsnSvm.h. Referenced by connectToDistributedSimulation(), syncDistributedSites(), and synchronizeOn().

PsNameToPointerMap<PsnProcess>* PsnSvm::_processTable [protected]

map containing all information about the different processes Definition at line 237 of file PsnSvm.h. Referenced by broadcast(), connectToDistributedSimulation(), createDistributedSimulation(), disconnectFromDistributedSimulation(), getLinkToProcessNamed(), getProcessDescriptorNamed(), processReceivedMessages(), sendCurrentBuffersWithTag(), serveNameRequestsUntilEnd(), synchroniseReceiveAndProcessMessages(), timestampCurrentSendBuffers(), waitAndProcessMessages(), and waitForMessage().

int PsnSvm::_siteId [protected]

workstation identifier in the svm Definition at line 244 of file PsnSvm.h.

PsName PsnSvm::_siteName [protected]

Name of the process associated to this site. Definition at line 252 of file PsnSvm.h. Referenced by broadcast(), connectToDistributedSimulation(), createDistributedSimulation(), getSiteName(), processReceivedMessages(), sendCurrentBuffersWithTag(), synchroniseReceiveAndProcessMessages(), timestampCurrentSendBuffers(), waitAndProcessMessages(), and waitForMessage().

string PsnSvm::_slaveGroupName = "spawnedControllers" [static, protected]

name of the group all the slave processes join Definition at line 34 of file PsnSvm.cxx. Referenced by connectToDistributedSimulation(), and syncDistributedSites().

const PsDate PsnSvm::_synchronisationLatency [protected]

in millisecond, the authorized latency when synchronising all the sites Definition at line 262 of file PsnSvm.h. Referenced by getSynchronisationLatency(), and synchroniseReceiveAndProcessMessages().

---

The documentation for this class was generated from the following files:

• PsnSvm.h
• PsnSvm.cxx

---