---

PsPvmController.cxx

Go to the documentation of this file.

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

#include <PsObjectDescriptor.h>
#include <PsSimulatedObject.h>
#include <PsDuplicatedObject.h>
#include <PsnProcess.h>
#include <PsnMirrorObjectHandle.h>
#include <PsnPvmReferenceObjectHandle.h>
#include <PsnPvmDuplicatedObjectHandle.h>
#include <unistd.h>
#include <PsnDoubleListElement.h>
#include <PsnDoubleList.h>
#include "PsSystemEventIdentifier.h"
#include "PsMultipleConfigurationParameter.h"

#include "PsPvmException.h"
#include "stdio.h"
#include "pvm3.h"

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

PsPvmController::PsPvmController (PsObjectDescriptor & initialObjects,
                                  const PsDate & initialDate,
                                  int argc,
                                  char * argv [])
   : PsDistributedController (initialObjects, initialDate),
     _finishing ( false ) 
{
   PsDate latence = 0 ;
   PsNameToPointerMap<PsnProcess> * processTable = new PsNameToPointerMap<PsnProcess>() ;

   PsMultipleConfigurationParameter * schedulingParameters = getSchedulingParametersOfObject( initialObjects ) ;
   if ( schedulingParameters != NULL)
      {
         PsConfigurationParameterDescriptor * param = schedulingParameters->getSubDescriptorByName("Latency") ;
         if (param != NULL)
            {
               latence = atoi (param->getAssociatedString().c_str()) ;
            }
         param = schedulingParameters->getSubDescriptorByName("Machines") ;
         PsMultipleConfigurationParameter * describedMachines = dynamic_cast<PsMultipleConfigurationParameter *>(param);
         if (describedMachines != NULL)
            {
               int numberOfMachines = describedMachines->getNumberOfSubItems () ;
               for (int machineNumber = 0 ;
                    machineNumber != numberOfMachines ;
                    ++machineNumber )
                  {
                     param = describedMachines->getSubDescriptorByPosition ( machineNumber ) ;
                     processTable->addObjectWithIndex ( describedMachines->getNameOfSubDescriptor (machineNumber),
                                                        new PsnProcess ( describedMachines->getNameOfSubDescriptor (machineNumber),
                                                                         param->getAssociatedString() )
                                                        ) ;
                  }
            }
      }

   // virtual member functions aren't initiallised in constructor
   // therefore, recreate the correct objectHandle
   delete _objectHandle ;
   
   _objectHandle = NULL ;
   
   PsnDuplicatedObjectHandle * localObjectHandle = newPsnDuplicatedObjectHandle( *this ) ;
   _objectHandle = localObjectHandle ;

  for (PsNameToPointerMap<PsnProcess>::const_iterator i = processTable->begin() ;
       i != processTable->end () ;
       ++i) 
     {
        localObjectHandle->addProcessOfDuplicate ( (*i).second ) ;
     }


   _distributedVirtualMachine = new PsnPvmSvm (processTable, latence, argc, argv) ;

   _distributedVirtualMachine->init (initialDate ) ;

   _processName = _distributedVirtualMachine->getSiteName () ;

   setProcessOfDescriptor ( initialObjects, _processName ) ;

   cout << "PsPvmController::PsPvmController (...)" 
        << "of process: " << _distributedVirtualMachine->getSiteName () << endl ;


   //rewritten 2001/12/5 by D. Margery
   list < PsObjectDescriptor *> * descriptorList = _simulationTree.getDescendants() ;
   for (list < PsObjectDescriptor *>::iterator descriptorIterator = descriptorList->begin() ;
        descriptorIterator != descriptorList->end() ;
        ++ descriptorIterator )
      {
         PsFrequency currentFreq = (*descriptorIterator)->getFrequency() ;
         PsName currentProcess = (*descriptorIterator)->getProcess() ;
         PsnProcess * currentProcessDescriptor = processTable->getObjectOfIndex( currentProcess ) ;
         if ( currentProcessDescriptor != NULL )
            // the process could have been removed by the svm
            {
               if ( currentFreq > currentProcessDescriptor->getMaxFrequencyOfHostedObjects () ) 
                  {
                     currentProcessDescriptor->setMaxHostedFrequency (currentFreq) ;
                  }
               currentProcessDescriptor->setFrequency ( PsController::lcm(currentFreq, currentProcessDescriptor->getFrequency() ) ) ;
            }
      }
   delete descriptorList ;

   //subscribe to system signals to receive notification on what is happening on other controllers
   registerForSignalBy(PsSystemEventIdentifier::MaskObjectDestroyed, getName()) ;

}

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

PsPvmController::~PsPvmController () 
{
   // deconnexion de la machine virtuelle
#ifdef _DEBUGEXECPVM
   cout << "PsPvmController::~PsPvmController : " << endl;
#endif
   _distributedVirtualMachine->disconnectFromDistributedSimulation ( getSimulatedDate () ) ;
   delete _distributedVirtualMachine ;  
}


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

void PsPvmController::run()
{
   if ( pvm_parent() != PvmNoParent )
      {
         PsDistributedController::run() ;

         // the simulation is locally finished, so notify all the controllers before exiting 
         PsDate lastDate = _date + 2 * _stepPeriod + _distributedVirtualMachine->getSynchronisationLatency() ;

         while (_date < lastDate )
            {
#ifdef _DEBUGPVMMESS
               cerr<<"PsPvmController::run(): trying a clean exit"<<endl;
#endif
               PsnReferenceObjectHandle * objectHandle = dynamic_cast<PsnReferenceObjectHandle *> ( _objectHandle ) ;

               assert ( objectHandle != NULL ) ;
               
               //the controller has to process residual system events adressed to itself
               objectHandle->processEvents() ;
               
               //send enought empty signals to enable other controllers to finish their job
               makeSynchronisationMessage( _referenceObjectsMap ) ;
               
               // send the synchronisation messages
               _distributedVirtualMachine->sendCurrentBuffersWithTag (PsnPvmMessage::SynchronisationMessage) ; 

               //wait a little, to try and get clean exit
               sleep(1) ;

               // for cleanup, read any arrived messages
               _distributedVirtualMachine->processReceivedMessages (*this, PsnPvmMessage::SynchronisationMessage) ;

               advanceSimulatedDate() ;
            } 
      }
   
}



PsnMirrorObjectHandle * PsPvmController::createMirrorObject (PsObjectDescriptor * objectDescription) 
{
   PsnMirrorObjectHandle * mirror =  PsDistributedController::createMirrorObject( objectDescription ) ;

   // Register the mirror to the reference object for regular updates of outputs
   mirror->registerToReferenceObject () ;
   
   return mirror ;
}


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

const PsName & PsPvmController::getProcessName () const 
{
   return (_distributedVirtualMachine->getSiteName ()) ;
}

//------------------------------------------------------------------------------
void PsPvmController::advanceSimulatedDate () 
{
#ifdef _DEBUGEXECPVM
   cerr << "PsPvmController::advanceSimulatedDate () from " <<_date<<endl;
#endif
   //send any messages collected during init
   if (_date < initialSimulationDate )
      {
         _distributedVirtualMachine->sendCurrentBuffersWithTag (PsnPvmMessage::SynchronisationMessage) ;  
      }

   PsDistributedController::advanceSimulatedDate () ;

   // prepare buffers to receive any message generate during this simulation step
   _distributedVirtualMachine->timestampCurrentSendBuffers (_date) ;   
}


void PsPvmController::computeNextSimulationStep () {
#ifdef _DEBUGEXECPVM
   cerr << "PsPvmController::computeNextSimulation : computing yet another simulation step for " 
        <<notreProcessus()<< endl;
#endif
  
   // compute the simulation step
   PsDistributedController::computeNextSimulationStep () ;

   // compute the synchronisation messages
   makeSynchronisationMessage ( _referenceObjectsMap ) ;

   // send the synchronisation messages
   _distributedVirtualMachine->sendCurrentBuffersWithTag (PsnPvmMessage::SynchronisationMessage) ;  

   // proceed with a relaxed synchronisation
   _distributedVirtualMachine->synchroniseReceiveAndProcessMessages (*this, PsnPvmMessage::SynchronisationMessage) ; 
}

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

void PsPvmController::makeSynchronisationMessage (PsNameToPointerMap<PsnReferenceObjectHandle> & referenceObjects ) 
{
   static PsName endMessage ("_OpenMASKEndOfSynchronisationMessage") ;

#ifdef _DEBUGPVMMESS
   cerr<<"PsPvmController::makeSynchronisationMessage for "<<referenceObjects.size() <<" objects "<<endl;
#endif

  PsNameToPointerMap<PsnReferenceObjectHandle>::iterator pTabRef ;
  PsDate dt ;
  // ask all referentials to contribute to the synchronisation message
  for (pTabRef = referenceObjects.begin () ;
       pTabRef != referenceObjects.end () ;
       pTabRef ++) 
    {
#ifdef _DEBUGPVMMESS
      cerr<<"PsPvmController::makeSynchronisationMessage for object " <<(*pTabRef).second->getSimulatedObject().getName()<<endl;
#endif
      
      PsnPvmReferenceObjectHandle * pvmObjectHandle = dynamic_cast<PsnPvmReferenceObjectHandle *> ( (*pTabRef).second ) ; 
      if ( pvmObjectHandle != NULL )
         {
            pvmObjectHandle->makeSynchronisationMessage ( _date ) ;
         }
      else
         {
#ifdef _DEBUGPVMMESS
            cerr<<"No synchronisation message for "<<(*pTabRef).second->getSimulatedObject().getName()<<endl;
#endif      
         }
#ifdef _DEBUGPVMMESS
      cerr<<"PsPvmController::makeSynchronisationMessage of " <<(*pTabRef).second->getSimulatedObject().getName()<<" done"<<endl;
#endif
    } 
#ifdef _DEBUGPVMMESS
      cerr<<"PsPvmController::makeSynchronisationMessage of done"<<endl;
#endif
}


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

void PsPvmController::parseSynchronisationMessage (PsnPvmIncomingMessage * message) 
{

  PsnReferenceObjectHandle * refer ; // Referentiel de l'objet destinataire
  PsnMirrorObjectHandle * miroir ; // Miroir de l'objet destinataire
  PsName nomObj ; // Nom du miroir destinataire
  bool stillMessagesToParse = true ;

#ifdef _DEBUGPVMMESS
  cerr << "PsPvmController::parseSynchronisationMessage at "<<getSimulatedDate()<<":"<< endl ;
#endif
    

   while ( stillMessagesToParse ) 
      {
         // get the message recepient
         nomObj.unpack ( *message ) ;
         
         
#ifdef _DEBUGPVMMESS
         cerr << "PsPvmController::parseSynchronisationMessage: message received for " << nomObj << endl;
#endif
         if ( nomObj == PsSynchronisationMessage::endOfSynchronisationFragment )
            // the synchronisation message has been completely parsed
            {
               stillMessagesToParse = false ;
            }
         else 
            {
               MirrorObjectsContainerType::iterator i = _mirrorObjectsMap.find ( nomObj ) ;
               if (i != _mirrorObjectsMap.end ()) 
                  {
                     // On lui transmet le message
#ifdef _DEBUGPVMMESS
                     cout << "PsPvmController::parseSynchronisationMessage: for miror" << endl ;
#endif
                     miroir = i->second ;
                     miroir->unpack ( *message ) ;
                  } 
               else if (_referenceObjectsMap.find (nomObj) != _referenceObjectsMap.end ()) 
                  {
                     // On lui transmet le message
#ifdef _DEBUGPVMMESS
                     cout << "PsPvmController::parseSynchronisationMessage : for referential : " << nomObj <<endl ;
#endif
                     refer = dynamic_cast<PsnPvmReferenceObjectHandle *>(_referenceObjectsMap.getObjectOfIndex (nomObj) ) ;
                     if ( refer != NULL )
                        {
                           refer->unpack (*message)  ;
                        }
                     else
                        {
                           dynamic_cast<PsnPvmDuplicatedObjectHandle *>(_referenceObjectsMap.getObjectOfIndex (nomObj) )->unpack ( *message ) ;
                        }
                  } 
               else if (nomObj == "") 
                  {
#ifdef _DEBUGPVMMESS
                     cout << "PsPvmController::parseSynchronisationMessage :CL : message vide d'abonnement ... pour synchro" << endl ;
#endif
                  } 
               else 
                  {
#ifdef _DEBUGPVMMESS
                     cerr << "PsPvmController::parseSynchronisationMessage: "<< endl ;
                     cerr << "for unknown object: <" << nomObj << "> (was probably destroyed)" << endl ;
                     error ("PsPvmController::parseSynchronisationMessage : Objet ni miroir ni referentiel") ;
#endif
                     //probably a request for an object that was destroyed
                     int typeMess ;
                     *message >> typeMess ;
                     // hope it is a registration for a just destroyed referential, and interprete it
                     if ( typeMess == Registration )
                        {
                           PsName processName ;
                           *message >> processName ;    
                        }
                     else if ( typeMess == EventReceived )
                        {
                           PsEvent * event ;
                           PsName classToCreate ;
                           PsEventIdentifier eventId;
                           PsDate eventDate;
                           PsName sender, receiver ;
                           *message >> classToCreate >> eventId>> eventDate>>sender>>receiver ;
                           event = PsEventCreator::createEvent (classToCreate, eventId, eventDate, sender, receiver ) ;
                           event->unpack ( *message ) ;
                           //cerr << "PsPvmController::parseSynchronisationMessage event "<<*event<<" undelivered"<<endl;
                           delete event ;
                        }
                  }
            }
      }
#ifdef _DEBUGPVMMESS
   cerr << "PsPvmController::parseSynchronisationMessage : message parsed"<< endl ;
#endif
}






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

void PsPvmController::init () 
{

   if ( pvm_parent() != PvmNoParent )
      {
         //on tente l'initialisation normale : pour chaque init, cette initialisation appelle initObjetReferentiel 
#ifdef _DEBUGDISTRIBUTEDINIT
         cerr<<"PsPvmController::init preparing the outgoing message buffers "<<endl;
#endif   
         _distributedVirtualMachine->timestampCurrentSendBuffers ( _date ) ;

         // try a classic initialisation
         PsController::init () ;

         //initialisation shouldn't fail
         assert ( tableDesNonInitialises.empty () ) ;

         //here, as advanceSimulatedDate has been called to restore the initial simulation date, no need to timestamp messages
         
#ifdef _DEBUGDISTRIBUTEDINIT
         cerr<<"PsPvmController::init : classic initialisation tried"<<endl;
#endif
         // send all registrations
         _distributedVirtualMachine->sendCurrentBuffersWithTag (PsnPvmMessage::SynchronisationMessage) ;

         // broadcast a empty message signalling our initialisation is finished, and wait for all processes
         _distributedVirtualMachine->synchronizeOn ( *this, PsnPvmMessage::LocalInitSuccessfull ) ;


         //restore state to the supposed state resulting of a call to the ancestor init
         _distributedVirtualMachine->timestampCurrentSendBuffers (_date) ;
      }
   
}

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

int PsPvmController::getOutputHistorySize(void) {
  // Taille d'une file d'une output pour un controleur local
  // nbMinor * 4 --> pour guarantir la recuperation d'au plus 4 valeurs dans la file
  // 1 ?? surete ??
   int nbrValeurPendantLatence = 1 ;//la latence minimal est un pas de simulation au sens controleur
   nbrValeurPendantLatence+=_distributedVirtualMachine->getSynchronisationLatency()/_stepPeriod;
   return PsDistributedController::getOutputHistorySize()+nbrValeurPendantLatence;
}  



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

PsnObjectHandle * 
PsPvmController::removeObjectFromDataStructures(const PsName & nom) 
{
   PsnObjectHandle * result = NULL ;
   if (_referenceObjectsMap.find (nom) != _referenceObjectsMap.end ()) 
      {
         result = PsController::removeObjectFromDataStructures(nom);
      } 
   else 
      {
         MirrorObjectsContainerType::iterator i = _mirrorObjectsMap.find (nom) ;
         if (i != _mirrorObjectsMap.end ()) 
            {
               PsnMirrorObjectHandle * miroir = i->second ;
               _mirrorObjectsMap.erase( i );
               result = miroir;
            }
         else if (_duplicatedObjectsMap.find (nom) != _duplicatedObjectsMap.end()) 
            {
               //a duplicated object is a referential present in the duplicated objects table
               _duplicatedObjectsMap.erase(nom);
            }  
      } 
   return result ; 
}


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

PsnMirrorObjectHandle * PsPvmController::newPsnMirrorObjectHandle (PsSimulatedObject & obj) 
{
  return new PsnMirrorObjectHandle(obj);
}


PsnDuplicatedObjectHandle * PsPvmController::newPsnDuplicatedObjectHandle (PsSimulatedObject & obj) 
{
  return new PsnPvmDuplicatedObjectHandle( obj, *this );
}


PsnReferenceObjectHandle * PsPvmController::newPsnReferenceObjectHandle(PsSimulatedObject & object, 
                                                                        PsController & controller,
                                                                        PsnSignalDispatcher * signalDispatcher)
{
  return new PsnPvmReferenceObjectHandle(object, controller, signalDispatcher );
}

PsnSvm * PsPvmController::getDistributedVirtualMachine ()
{
   return _distributedVirtualMachine;
}

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


void PsPvmController::sendInitialValuesToMirror(PsnPvmIncomingMessage & message)
{
   if ( ! _finishing )
      {
         // first find out the name of the objects whose initial values are needed
         PsName referentialName ;
         referentialName.unpack ( message ) ;
         
         cerr<<"Looking for initial values of "<<referentialName<<endl;

         //then find the corresponding referential
         PsNameToPointerMap<PsnReferenceObjectHandle>::iterator i = _referenceObjectsMap.find ( referentialName ) ;
         PsnPvmReferenceObjectHandle * referential = NULL ;
         if ( i != _referenceObjectsMap.end() ) 
            {
               referential = dynamic_cast<PsnPvmReferenceObjectHandle *> (i->second) ;
            }
         else
            {
               //the referential might have been destroyed: look for it in the list of deleted object handles
               list <pair <PsDate, PsnObjectHandle *> >::iterator i = _deletedObjectHandles.begin() ;
               while ( i != _deletedObjectHandles.end() )
                  {
                     if ( i->second->getSimulatedObject().getName() == referentialName )
                        {
                           referential = dynamic_cast<PsnPvmReferenceObjectHandle *> ( i->second );
                           i = _deletedObjectHandles.end() ;
                        }
                     else
                        {
                           ++ i ;
                        }
                  }
            }
         assert ( referential != NULL ) ;
         
         //find the name of the requesting process
         PsName mirrorProcessName ;
         mirrorProcessName.unpack( message ) ;
         
         //then find the link to the corresponding process
         PsnSvmLink * process = getDistributedVirtualMachine()->getLinkToProcessNamed ( mirrorProcessName ) ;
         
         assert ( process != NULL ) ;
         
         //build the answer
         PsnPvmUnicastMessage * urgentAnswer = new PsnPvmUnicastMessage ( process->getTID() ) ;
         
         urgentAnswer->insertTimeStamp( _date ) ;
         
         referential->packInitialValues ( * urgentAnswer ) ;
         
         PsSynchronisationMessage::endOfSynchronisationFragment.pack ( * urgentAnswer ) ;  
         
         // and send it
         urgentAnswer->send ( PsnPvmMessage::InitialValuesForMirror ) ;
      }
}


void PsPvmController::waitForAnswerToBlockingRequest (PsnPvmMessage::MessageTag tag)
{
 if ( ! _finishing )
    // avoid blocking the contyroller while terminating, because other controllers could have exited
    {
       _distributedVirtualMachine->waitForAnswerToBlockingRequest ( *this, tag ) ;      
    }
}

void PsPvmController::finish()
{
   _finishing = true ;
   PsDistributedController::finish() ;
}

PsDate PsPvmController::getPurgeDate () 
{
   //in the worst case we could reveive message from an object whose simulation date is current date - _latency -stepPeriod ;
   // make sure by doubling latency
   return PsDistributedController::getPurgeDate() - 2 * _distributedVirtualMachine->getSynchronisationLatency() ;
}

#endif

---