---

PsnSvm.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
 *
 */
#include <unistd.h>
#include <signal.h>
#include <PsnSvm.h>
#include <PsnSvmLink.h>
#include "PsSynchronisationMessage.h"
#include <PsnPvmMessage.h>
#include <PsnPvmMulticastMessage.h>
#include <PsnPvmUnicastMessage.h>
#include <PsnProcess.h>
#include <PsPvmController.h>
#include "PsPvmNameServer.h"
#include "PsPvmCentralNameServer.h"
#include "stdio.h"
#include "pvm3.h"
#include "PsnPvmSvm.h"

string PsnSvm::_slaveGroupName = "spawnedControllers";
string PsnSvm::_masterSiteName = "OpenMaskMasterDistributedSite" ;
//-----------------------------------------------------------------------

PsnSvm::PsnSvm (PsNameToPointerMap<PsnProcess> * tab, const PsDate & latence) : 
   _processTable ( tab ),
   _synchronisationLatency ( latence ),
   _siteId ( -1 ), //negative values indicates an error
   _numberOfSlaves ( 0 )
{
}


PsnSvm::~PsnSvm () 
{
}


const PsDate & PsnSvm::getSynchronisationLatency() 
{
  return _synchronisationLatency;
}



void PsnSvm::init (const PsDate & initialSimulationDate) 
{
   if ( getParentSiteId () == 0 )  // wasn't created by a spawn
     {
       createDistributedSimulation ( initialSimulationDate ) ;
     }
   else
     {
       connectToDistributedSimulation () ;
     }
}


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



// aspects fonctionnement secondaire
//****************************************************
// RK 1302 : Suppression du nom du pss

void PsnSvm::createDistributedSimulation(const PsDate & initialSimulationDate )
{
#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 () ;

}



void PsnSvm::serveNameRequestsUntilEnd (void) 
{
  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;
        }
    }
}



void PsnSvm::connectToDistributedSimulation () 
{
#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::disconnectFromDistributedSimulation (const PsDate & endOfSimulationDate ) 
{
#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 ) ;
      }
}


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


void PsnSvm::broadcast (PsnPvmMessage::MessageTag tag, PsnPvmMulticastMessage * mess)
{
  //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 ) ;

}



void PsnSvm::synchronizeOn ( PsPvmController & parsingController, PsnPvmMessage::MessageTag tag )
{
  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::syncDistributedSites () 
{
   groupBarrier ( _slaveGroupName, _numberOfSlaves ) ;
}

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

PsnSvmLink * PsnSvm::getLinkToProcessNamed (const PsName & nomPss) 
{
   // 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 ;
   }
}

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

PsnProcess * PsnSvm::getProcessDescriptorNamed (const PsName & nomPss) 
{
   // 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 ;
      }
}


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

void PsnSvm::processReceivedMessages (PsPvmController & parsingController, const PsnPvmMessage::MessageTag tag) 
{
#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) ;
                  }
            }
      }
}

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

void PsnSvm::waitAndProcessMessages (PsPvmController & parsingController, 
                                     const PsnPvmMessage::MessageTag tag) 
{
#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) ;
      }
   }
}

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

void PsnSvm::waitForMessage (PsPvmController & parsingController, const PsnPvmMessage::MessageTag tag) 
{
   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::synchroniseReceiveAndProcessMessages (PsPvmController & parsingController, const PsnPvmMessage::MessageTag tag) 
{
   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::sendCurrentBuffersWithTag(const PsnPvmMessage::MessageTag tag) 
{
   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::waitForMessageFrom (PsPvmController & parsingController,
                                 const PsName & nomPss,
                                 const PsnPvmMessage::MessageTag tag) 
{
   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;
}


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

void PsnSvm::timestampCurrentSendBuffers (const PsDate & date) 
{
   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 ) ;
           }
     }
}


const PsName & PsnSvm::getSiteName() const
{
   return _siteName;
}

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

PsnProcess * PsnSvm::waitForAnswerToBlockingRequest (PsPvmController & parsingController, PsnPvmMessage::MessageTag tag)
{
   // 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 ;
}

---