OMK::PvmController Class Reference

Class defining a local controller for a pvm session. More...

#include <OMKPvmController.h>

Inheritance diagram for OMK::PvmController:

[legend]Collaboration diagram for OMK::PvmController:

[legend]List of all members.

Public Member Functions
	PvmController (ObjectDescriptor &arbre, const Date &initialSimulationDate, int argc, char *argv[])
	Constructor The constructor initializes the distributed virtual machine.
virtual	~PvmController ()
	Destructor.
virtual void	run ()
	redefine run so that it does nothing unless the controller belongs to a spawned process
virtual void	advanceSimulatedDate ()
	redefine advanceSimulatedDate to include datation of synchronisation messages any try to send data between a call to computeNextSimualtionStep and advanceSimulatedDate will cause an assertion to fail
virtual void	computeNextSimulationStep ()
	redefine computeNextSimulationStep to include the synchronisation primitives
virtual void	finish ()
	redefine to avoid going into blocking mode when finishing the simulation
virtual const Name &	getProcessName () const
	get the name of the process this controller is controlling
virtual Svm *	getDistributedVirtualMachine ()
	get access to the siames distributed virtual machine
virtual void	makeSynchronisationMessage (NameToPointerMap< ReferenceObjectHandle > &referenceObjects)
	make the synchronisation message to be sent to all processes
virtual void	parseSynchronisationMessage (PvmIncomingMessage *message)
	parse the received synchronisation messages.
virtual void	waitForAnswerToBlockingRequest (PvmMessage::MessageTag tag)
	waitForAnswerToBlockingRequest.
virtual void	sendInitialValuesToMirror (PvmIncomingMessage &message)
	answer to a MirrorNeedsInitialValues message received by the distributed subsystem
virtual void	init ()
	redefine for distributed initialisation
virtual std::vector< Name >	getIsolatedMirrorList ()
virtual std::vector< Name >	getNonIsolatedMirrorList ()
virtual bool	testIfTroubleOccured ()
virtual void	createNewSite (const Date &)
virtual std::list< Name >	getDisconnectedProcessusList ()
virtual std::map< Name, int >	getDisconnectedProcessusMap ()
Public Attributes
std::vector< Name >	_publicIsolatedMirrorObjectsVector
std::vector< Name >	_publicNonIsolatedMirrorObjectsVector
Protected Member Functions
virtual Date	getPurgeDate ()
	redefine to take latency into account
virtual MirrorObjectHandle *	createMirrorObject (ObjectDescriptor *objectDescription)
	Create a miror object.
virtual DuplicatedObjectHandle *	newOMKDuplicatedObjectHandle (SimulatedObject &object)
	create an object handle adapted to handling of duplicated objects
virtual LocalObjectHandle *	newOMKLocalObjectHandle (SimulatedObject &object)
	CHADI.
virtual MirrorObjectHandle *	newOMKMirrorObjectHandle (SimulatedObject &obj)
	create an object handle for objects mirrored by this controller
virtual ReferenceObjectHandle *	newOMKReferenceObjectHandle (SimulatedObject &object, Controller &controller, SignalDispatcher *signalDispatcher)
	create a reference object handler appropriate for this controller
virtual int	getOutputHistorySize (void)
	get the size of the history fifo needed when running a distributed simulation.
virtual ObjectHandle *	removeObjectFromDataStructures (const Name &nom)
	gestion de la destruction distribuée : destruction des structures locales
void	synchronizedReceiveAndProcessMessages (void)
void	relaxedReceiveAndProcessMessages (void)
Protected Attributes
Svm *	_distributedVirtualMachine
	the distributed virtual machine used by the controller
bool	_finishing
	boolean indicating the controller is going into shutdown mode.
void(PvmController::*	receiveAndProcessMessages )(void)
	to manage the choice of the synchronization

---

Detailed Description

Class defining a local controller for a pvm session.

the local controller is in charge of

• creating local referenceObjects
• execution synchronisation
• data synchronisation

  Author:

  Siames

  Version:

  2.5

  • 2.2 -> 2.3 (by David Margery 2002/02/07 ) : added creation of a proper PvmReferenceObject
  • 2.3 -> 2.4 (by David Margery 2002/02/15 ) : added creation of a proper PvmDuplicatedObjectHandle
  • 2.4 -> 2.5 (by David Margery 2002/09/04 ) : added support for urgent requests for initial values

Definition at line 47 of file OMKPvmController.h.

---

Constructor & Destructor Documentation

PvmController::PvmController	(	ObjectDescriptor &	arbre,
		const Date &	initialSimulationDate,
		int	argc,
		char *	argv[]
	)

Constructor The constructor initializes the distributed virtual machine.

The created distributed virtual machine acts as a server if the process wasn't spawned. Other wise, the controller acts a a local controller synchronized throught the distributed virtual machine to the other controllers

Definition at line 52 of file OMKPvmController.cxx.

References OMK::Controller::_computeStatePointer, _distributedVirtualMachine, OMK::SimulatedObject::_objectHandle, OMK::DistributedController::_processName, OMK::Controller::_simulationTree, OMK::NameToPointerMap< ObjectType >::addObjectWithIndex(), OMK::DuplicatedObjectHandle::addProcessOfDuplicate(), OMK::ParametersAccessor::get(), OMK::ConfigurationParameterDescriptor::getAssociatedString(), OMK::ObjectDescriptor::getDescendants(), OMK::Process::getFrequency(), OMK::Process::getMaxFrequencyOfHostedObjects(), OMK::SimulatedObject::getName(), OMK::MultipleConfigurationParameter::getNameOfSubDescriptor(), OMK::MultipleConfigurationParameter::getNumberOfSubItems(), OMK::NameToPointerMap< ObjectType >::getObjectOfIndex(), OMK::Controller::getSchedulingParametersOfObject(), OMK::Svm::getSiteName(), OMK::ConfigurationParameterDescriptor::getSubDescriptorByName(), OMK::MultipleConfigurationParameter::getSubDescriptorByName(), OMK::MultipleConfigurationParameter::getSubDescriptorByPosition(), OMK::Svm::init(), OMK::ReferenceObjectHandle::initial, OMK::Controller::lcm(), OMK::SystemEventIdentifier::MaskObjectDestroyed, newOMKDuplicatedObjectHandle(), receiveAndProcessMessages, OMK::SimulatedObject::registerForSignalBy(), relaxedReceiveAndProcessMessages(), OMK::Process::setFrequency(), OMK::Process::setMaxHostedFrequency(), OMK::Controller::setProcessOfDescriptor(), and synchronizedReceiveAndProcessMessages().

                                    : DistributedController (initialObjects, initialDate),
                                    _finishing ( false ) 
{

  Date latence = 20 ;
  Date timeOut = 10 ;
  std::string synchro ("strong") ;
  int deadReckoning = 70 ;
  bool yieldNeeded = false ;
  NameToPointerMap<Process> * processTable = new NameToPointerMap<Process>() ;
  receiveAndProcessMessages = &PvmController::synchronizedReceiveAndProcessMessages ;

  MultipleConfigurationParameter * schedulingParameters = getSchedulingParametersOfObject( initialObjects ) ;
  if ( schedulingParameters != NULL)
  {
    ConfigurationParameterDescriptor * param = schedulingParameters->getSubDescriptorByName("Latency") ;
    if (param != NULL)
    {
      latence = atoi (param->getAssociatedString().c_str()) ;
    }
    std::cerr << "Latency : " << latence << std::endl ;

    synchro = "strong" ;
    if( ParametersAccessor::get( schedulingParameters, "Synchronization", synchro ) 
      && ( synchro == "relaxed" ) )
    {
      receiveAndProcessMessages = &PvmController::relaxedReceiveAndProcessMessages ;
    }
    std::cerr << "Synchronization : " << synchro << std::endl ;

    ParametersAccessor::get( schedulingParameters, "YieldNeeded", yieldNeeded ) ;
    std::cerr << "YieldNeeded : " << yieldNeeded << std::endl ;

    const ConfigurationParameterDescriptor * timeOutParamDescriptor = 
      schedulingParameters->getSubDescriptorByName ("TimeOut") ;
    if (timeOutParamDescriptor != NULL) {
      timeOut = atoi (timeOutParamDescriptor->getAssociatedString().c_str()) ;
    }
    std::cerr << "TimeOut : " << timeOut << std::endl ;

    const ConfigurationParameterDescriptor * deadReckoningParamDescriptor = 
      schedulingParameters->getSubDescriptorByName ("DeadReckoningStepInterval") ;
    if (deadReckoningParamDescriptor != NULL) {
      deadReckoning = atoi (deadReckoningParamDescriptor->getAssociatedString().c_str()) ;
    }
    std::cerr << "DeadReckoningStepInterval : " << deadReckoning << std::endl ;

    param = schedulingParameters->getSubDescriptorByName("Machines") ;
    MultipleConfigurationParameter * describedMachines = dynamic_cast<MultipleConfigurationParameter *>(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 Process ( describedMachines->getNameOfSubDescriptor (machineNumber),
          param->getAssociatedString() )
          ) ;
      }
    }
  }

  // virtual member functions aren't initiallised in constructor
  // therefore, recreate the correct objectHandle
  delete _objectHandle ;

  _objectHandle = NULL ;

  DuplicatedObjectHandle * localObjectHandle = newOMKDuplicatedObjectHandle( *this ) ;
  _objectHandle = localObjectHandle ;
  // TDTD ajout pour fixer bug mémoire : comme le _objectHandle a été effacé, ça ne marche plus...
  _computeStatePointer = const_cast<ReferenceObjectHandle::SimulatedObjectComputingState *>(&(dynamic_cast<DuplicatedObjectHandle *>(_objectHandle)->getComputingState () )) ;

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


  _distributedVirtualMachine = new PvmSvm (processTable, latence, timeOut, deadReckoning, yieldNeeded, argc, argv) ;

  _distributedVirtualMachine->init (initialDate ) ;

  _processName = _distributedVirtualMachine->getSiteName () ;

  setProcessOfDescriptor ( initialObjects, _processName ) ;

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

  // initialisation de l'état du controleur
  // TDTD suite ajout pour fixer bug mémoire : comme le _objectHandle a été effacé, ça ne marche plus...
  assert ( _computeStatePointer != NULL ) ;
  *_computeStatePointer = ReferenceObjectHandle::initial ;


  //rewritten 2001/12/5 by D. Margery
  list < ObjectDescriptor *> * descriptorList = _simulationTree.getDescendants() ;
  for (list < ObjectDescriptor *>::iterator descriptorIterator = descriptorList->begin() ;
    descriptorIterator != descriptorList->end() ;
    ++ descriptorIterator )
  {
    Frequency currentFreq = (*descriptorIterator)->getFrequency() ;
    Name currentProcess = (*descriptorIterator)->getProcess() ;
    Process * 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 ( Controller::lcm(currentFreq, currentProcessDescriptor->getFrequency() ) ) ;
    }
  }
  delete descriptorList ;

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

}

PvmController::~PvmController

(

)

[virtual]

Destructor.

Definition at line 185 of file OMKPvmController.cxx.

References _distributedVirtualMachine, OMK::Svm::disconnectFromDistributedSimulation(), and OMK::Controller::getSimulatedDate().

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

---

Member Function Documentation

void PvmController::run

(

)

[virtual]

redefine run so that it does nothing unless the controller belongs to a spawned process

Reimplemented from OMK::Controller.

Definition at line 198 of file OMKPvmController.cxx.

References OMK::Controller::_date, _distributedVirtualMachine, OMK::SimulatedObject::_objectHandle, OMK::Controller::_referenceObjectsMap, OMK::Controller::_stepPeriod, advanceSimulatedDate(), OMK::Svm::getSynchronisationLatency(), makeSynchronisationMessage(), OMK::ReferenceObjectHandle::processEvents(), OMK::Svm::processReceivedMessages(), OMK::Controller::run(), OMK::Svm::sendCurrentBuffersWithTag(), and OMK::PvmMessage::SynchronisationMessage.

{
  if ( pvm_parent() != PvmNoParent )
  {
    DistributedController::run() ;

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

    while (_date < lastDate )
    {
#ifdef _DEBUGPVMMESS
      cerr<<"PvmController::run(): trying a clean exit"<<endl;
#endif
      ReferenceObjectHandle * objectHandle = dynamic_cast<ReferenceObjectHandle *> ( _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 (PvmMessage::SynchronisationMessage) ; 

      //wait a little, to try and get clean exit
#ifdef _MSC_VER
      Sleep( 1000 ) ;
#else
      sleep(1) ;
#endif

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

      advanceSimulatedDate() ;
    } 
  }

}

void PvmController::advanceSimulatedDate

(

)

[virtual]

redefine advanceSimulatedDate to include datation of synchronisation messages any try to send data between a call to computeNextSimualtionStep and advanceSimulatedDate will cause an assertion to fail

Reimplemented from OMK::Controller.

Definition at line 262 of file OMKPvmController.cxx.

References OMK::Controller::_date, _distributedVirtualMachine, OMK::Controller::advanceSimulatedDate(), OMK::Controller::initialSimulationDate, OMK::Svm::sendCurrentBuffersWithTag(), OMK::PvmMessage::SynchronisationMessage, and OMK::Svm::timestampCurrentSendBuffers().

Referenced by run().

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

  DistributedController::advanceSimulatedDate () ;

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

void PvmController::computeNextSimulationStep

(

)

[virtual]

redefine computeNextSimulationStep to include the synchronisation primitives

Reimplemented from OMK::Controller.

Definition at line 304 of file OMKPvmController.cxx.

References _distributedVirtualMachine, OMK::Controller::_referenceObjectsMap, OMK::Controller::computeNextSimulationStep(), makeSynchronisationMessage(), receiveAndProcessMessages, OMK::Svm::sendCurrentBuffersWithTag(), and OMK::PvmMessage::SynchronisationMessage.

                                               {

#ifdef _DEBUGEXECPVM
  cerr << "PvmController::computeNextSimulation : computing yet another simulation step for " 
    <<notreProcessus()<< endl;
#endif

  // compute the simulation step
  DistributedController::computeNextSimulationStep () ;

  // compute the synchronisation messages
  makeSynchronisationMessage ( _referenceObjectsMap ) ;

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

  // TDTD : pour un fonctionnement en synchro ou relaxed
  (this->*receiveAndProcessMessages) () ;
  //synchronizedReceiveAndProcessMessages () ;
  // TDTD : en cas de desynchro, il faut régler le pb du flush des messages inutiles... ::: à peu près réalisé...

  //added by chadi to test if a new process has been added dynamically
  // TDTD : à remettre plus tard
  //_distributedVirtualMachine->testIfNewProcessAdded () ;
  /*
  if (testIfTroubleOccured())
  {
  bool tico = _distributedVirtualMachine->reconnexionEstablished() ;
  }
  */

  //    {
  //     cout<<"bien bien bien"<<endl;
  // makeSynchronisationMessage ( _referenceObjectsMap ) ;       
  // _distributedVirtualMachine->sendCurrentBuffersWithTag (PvmMessage::SynchronisationMessage) ;  
  //   }

  //  if (testIfTroubleOccured())
  //        {
  //     _distributedVirtualMachine->sendPingMessage() ;
  //        }



  //  if ( testIfTroubleOccured() )
  //        {
  //     NameToPointerMap<ReferenceObjectHandle>::iterator pTabRef ;
  //     for (pTabRef = _referenceObjectsMap.begin () ;
  //          pTabRef != _referenceObjectsMap.end () ;
  //          pTabRef ++) 
  //       {

  //         PvmReferenceObjectHandle * pvmObjectHandle = dynamic_cast<PvmReferenceObjectHandle *> ( (*pTabRef).second ) ; 
  //         if ( pvmObjectHandle != NULL )
  //           {
  //             pvmObjectHandle->flushDisconnectedSitesOutgoingBuffer() ;
  //           }

  //       }

  //        }

}

void PvmController::finish

(

)

[virtual]

redefine to avoid going into blocking mode when finishing the simulation

Reimplemented from OMK::Controller.

Definition at line 793 of file OMKPvmController.cxx.

References _finishing, and OMK::Controller::finish().

{
  _finishing = true ;
  DistributedController::finish() ;
}

const Name & PvmController::getProcessName

(

)

const [virtual]

get the name of the process this controller is controlling

Definition at line 256 of file OMKPvmController.cxx.

References _distributedVirtualMachine, and OMK::Svm::getSiteName().

Referenced by OMK::PvmMirrorObjectHandle::registerToReferenceObject().

{
  return (_distributedVirtualMachine->getSiteName ()) ;
}

Svm * PvmController::getDistributedVirtualMachine

(

)

[virtual]

get access to the siames distributed virtual machine

Definition at line 710 of file OMKPvmController.cxx.

References _distributedVirtualMachine.

Referenced by sendInitialValuesToMirror().

{
  return _distributedVirtualMachine;
}

void PvmController::makeSynchronisationMessage

(

NameToPointerMap< ReferenceObjectHandle > &

referenceObjects

)

[virtual]

make the synchronisation message to be sent to all processes

Definition at line 382 of file OMKPvmController.cxx.

References OMK::Controller::_date, and OMK::PvmReferenceObjectHandle::makeSynchronisationMessage().

Referenced by computeNextSimulationStep(), and run().

{
  static Name endMessage ("_OpenMASKEndOfSynchronisationMessage") ;

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

  NameToPointerMap<ReferenceObjectHandle>::iterator pTabRef ;
  // ask all referentials to contribute to the synchronisation message
  for (pTabRef = referenceObjects.begin () ;
    pTabRef != referenceObjects.end () ;
    pTabRef ++) 
  {
#ifdef _DEBUGPVMMESS
    cerr<<"PvmController::makeSynchronisationMessage for object " <<(*pTabRef).second->getSimulatedObject().getName()<<endl;
#endif

    PvmReferenceObjectHandle * pvmObjectHandle = dynamic_cast<PvmReferenceObjectHandle *> ( (*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<<"PvmController::makeSynchronisationMessage of " <<(*pTabRef).second->getSimulatedObject().getName()<<" done"<<endl;
#endif
  } 
#ifdef _DEBUGPVMMESS
  cerr<<"PvmController::makeSynchronisationMessage done"<<endl;
#endif
}

void PvmController::parseSynchronisationMessage

(

PvmIncomingMessage *

message

)

[virtual]

parse the received synchronisation messages.

this member function is called by the siames virtual machine when it receives synchronisation messages from distant workstations

Parameters:

message

the message to parse

Definition at line 423 of file OMKPvmController.cxx.

References OMK::DistributedController::_mirrorObjectsMap, OMK::Controller::_referenceObjectsMap, OMK::EventCreator::createEvent(), OMK::SynchronisationMessage::endOfSynchronisationFragment, OMK::Controller::error(), OMK::EventReceived, OMK::Controller::getSimulatedDate(), OMK::Registration, OMK::Event::unpack(), OMK::SimulatedObject::unpack(), OMK::Flowable::unpack(), OMK::MirrorObjectHandle::unpack(), and OMK::Name::unpack().

Referenced by OMK::Svm::processReceivedMessages(), OMK::Svm::relaxedSynchroniseReceiveAndProcessMessages(), OMK::Svm::synchroniseReceiveAndProcessMessages(), OMK::Svm::waitAndProcessMessages(), OMK::Svm::waitForAnswerToBlockingRequest(), OMK::Svm::waitForMessage(), and OMK::Svm::waitForMessageFrom().

{

  ReferenceObjectHandle * refer ; // Referentiel de l'objet destinataire
  MirrorObjectHandle * miroir ; // Miroir de l'objet destinataire
  Name nomObj ; // Nom du miroir destinataire
  bool stillMessagesToParse = true ;

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


  while ( stillMessagesToParse ) 
  {
    // get the message recepient
    nomObj.unpack ( *message ) ;


#ifdef _DEBUGPVMMESS
    cerr << "PvmController::parseSynchronisationMessage: message received for " << nomObj << endl;
#endif
    if ( nomObj == SynchronisationMessage::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 << "PvmController::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
        std::cout << "PvmController::parseSynchronisationMessage : for referential : " << nomObj <<std::endl ;
        //std::cout << "message : " << *message << std::endl ;
#endif
        refer = dynamic_cast<PvmReferenceObjectHandle *>(_referenceObjectsMap.getObjectOfIndex (nomObj) ) ;
        if ( refer != NULL )
        {
          refer->unpack (*message)  ;
        }
        else
        {
          dynamic_cast<PvmDuplicatedObjectHandle *>(_referenceObjectsMap.getObjectOfIndex (nomObj) )->unpack ( *message ) ;
        }
      } 
      else if (nomObj == "") 
      {
#ifdef _DEBUGPVMMESS
        cout << "PvmController::parseSynchronisationMessage :CL : message vide d'abonnement ... pour synchro" << endl ;
#endif
      } 
      else 
      {
#ifdef _DEBUGPVMMESS
        cerr << "PvmController::parseSynchronisationMessage: "<< endl ;
        cerr << "for unknown object: <" << nomObj << "> (was probably destroyed)" << endl ;
        error ("PvmController::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 )
        {
          Name processName ;
          *message >> processName ;     
        }
        else if ( typeMess == EventReceived )
        {
          Event * event ;
          Name classToCreate ;
          EventIdentifier eventId;
          Date eventDate;
          Name sender, receiver ;
          *message >> classToCreate >> eventId>> eventDate>>sender>>receiver ;
          event = EventCreator::createEvent (classToCreate, eventId, eventDate, sender, receiver ) ;
          event->unpack ( *message ) ;
          //cerr << "PvmController::parseSynchronisationMessage event "<<*event<<" undelivered"<<endl;
          delete event ;
        }
      }
    }
  }
#ifdef _DEBUGPVMMESS
  cerr << "PvmController::parseSynchronisationMessage : message parsed"<< endl ;
#endif
}

void PvmController::waitForAnswerToBlockingRequest

(

PvmMessage::MessageTag

tag

)

[virtual]

waitForAnswerToBlockingRequest.

make the controller fall bake in a server mode until message tag arrives and is succesfully processed

Definition at line 784 of file OMKPvmController.cxx.

References _distributedVirtualMachine, _finishing, and OMK::Svm::waitForAnswerToBlockingRequest().

Referenced by OMK::PvmMirrorObjectHandle::registerToReferenceObject().

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

void PvmController::sendInitialValuesToMirror

(

PvmIncomingMessage &

message

)

[virtual]

answer to a MirrorNeedsInitialValues message received by the distributed subsystem

Definition at line 718 of file OMKPvmController.cxx.

References OMK::Controller::_date, OMK::Controller::_deletedObjectHandles, _finishing, OMK::Controller::_referenceObjectsMap, OMK::SynchronisationMessage::endOfSynchronisationFragment, getDistributedVirtualMachine(), OMK::Svm::getLinkToProcessNamed(), OMK::SvmLink::getTID(), OMK::PvmMessage::InitialValuesForMirror, OMK::PvmOutgoingMessage::insertTimeStamp(), OMK::Name::pack(), OMK::PvmReferenceObjectHandle::packInitialValues(), OMK::PvmUnicastMessage::send(), and OMK::Name::unpack().

Referenced by OMK::Svm::relaxedSynchroniseReceiveAndProcessMessages(), OMK::Svm::synchroniseReceiveAndProcessMessages(), and OMK::Svm::waitForAnswerToBlockingRequest().

{
  if ( ! _finishing )
  {
    // first find out the name of the objects whose initial values are needed
    Name referentialName ;
    referentialName.unpack ( message ) ;


#ifdef _DEBUGEXECPVM
    cerr<<"Looking for initial values of "<<referentialName<<endl;
#endif

    //then find the corresponding referential
    NameToPointerMap<ReferenceObjectHandle>::iterator i = _referenceObjectsMap.find ( referentialName ) ;

    PvmReferenceObjectHandle * referential = NULL ;
    if ( i != _referenceObjectsMap.end() ) 
    {
      referential = dynamic_cast<PvmReferenceObjectHandle *> (i->second) ;
    }
    else
    {
      //the referential might have been destroyed: look for it in the list of deleted object handles
      list <pair <Date, ObjectHandle *> >::iterator i = _deletedObjectHandles.begin() ;
      while ( i != _deletedObjectHandles.end() )
      {
        if ( i->second->getSimulatedObject().getName() == referentialName )
        {
          referential = dynamic_cast<PvmReferenceObjectHandle *> ( i->second );
          i = _deletedObjectHandles.end() ;
        }
        else
        {
          ++ i ;
        }
      }
    }

    assert ( referential != NULL ) ;


    //find the name of the requesting process
    Name mirrorProcessName ;
    mirrorProcessName.unpack( message ) ;

    //then find the link to the corresponding process
    SvmLink * process = getDistributedVirtualMachine()->getLinkToProcessNamed ( mirrorProcessName ) ;

    assert ( process != NULL ) ;

    //build the answer
    PvmUnicastMessage * urgentAnswer = new PvmUnicastMessage ( process->getTID() ) ;

    urgentAnswer->insertTimeStamp( _date ) ;

    referential->packInitialValues ( * urgentAnswer ) ;

    SynchronisationMessage::endOfSynchronisationFragment.pack ( * urgentAnswer ) ;  

    // and send it
    urgentAnswer->send ( PvmMessage::InitialValuesForMirror ) ;
  }
}

void PvmController::init

(

)

[virtual]

redefine for distributed initialisation

Reimplemented from OMK::Controller.

Definition at line 529 of file OMKPvmController.cxx.

References OMK::Controller::_date, _distributedVirtualMachine, OMK::Controller::init(), OMK::PvmMessage::LocalInitSuccessfull, OMK::Svm::sendCurrentBuffersWithTag(), OMK::PvmMessage::SynchronisationMessage, OMK::Svm::synchronizeOn(), OMK::DistributedController::tableDesNonInitialises, and OMK::Svm::timestampCurrentSendBuffers().

{

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

    // try a classic initialisation
    Controller::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<<"PvmController::init : classic initialisation tried"<<endl;
#endif
    // send all registrations
    _distributedVirtualMachine->sendCurrentBuffersWithTag (PvmMessage::SynchronisationMessage) ;

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


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

}

vector< Name > PvmController::getIsolatedMirrorList

(

)

[virtual]

Reimplemented from OMK::DistributedController.

Definition at line 631 of file OMKPvmController.cxx.

References _distributedVirtualMachine, OMK::DistributedController::_mirrorObjectsMap, _publicIsolatedMirrorObjectsVector, and OMK::Svm::disconnectedProcess().

{

  for ( MirrorObjectsContainerType::iterator i = _mirrorObjectsMap.begin(); i!= _mirrorObjectsMap.end(); i++ )
  {
    Name referentialProcess = i->second->getSimulatedObject().getObjectDescriptor().getProcess();
    if ( _distributedVirtualMachine->disconnectedProcess( referentialProcess ) )
    {
      _publicIsolatedMirrorObjectsVector.push_back( i->first) ;
    }
  }

  return  _publicIsolatedMirrorObjectsVector;
}

vector< Name > PvmController::getNonIsolatedMirrorList

(

)

[virtual]

Reimplemented from OMK::DistributedController.

Definition at line 646 of file OMKPvmController.cxx.

References _distributedVirtualMachine, OMK::DistributedController::_mirrorObjectsMap, _publicIsolatedMirrorObjectsVector, _publicNonIsolatedMirrorObjectsVector, OMK::Svm::disconnectedProcess(), and OMK::Svm::disconnectedTableIsEmpty().

{

  if (_distributedVirtualMachine->disconnectedTableIsEmpty())
  {
    _publicIsolatedMirrorObjectsVector.clear();
    return _publicIsolatedMirrorObjectsVector;
  }
  else
  {

    for ( MirrorObjectsContainerType::iterator i = _mirrorObjectsMap.begin(); i!= _mirrorObjectsMap.end(); i++ )
    {
      Name referentialProcess = i->second->getSimulatedObject().getObjectDescriptor().getProcess();
      if ( _distributedVirtualMachine->disconnectedProcess( referentialProcess ) )
      {

      }
      else
      {
        _publicNonIsolatedMirrorObjectsVector.push_back( i->first) ;
      }
    }

    return  _publicNonIsolatedMirrorObjectsVector;
  }
}

bool PvmController::testIfTroubleOccured

(

)

[virtual]

Reimplemented from OMK::DistributedController.

Definition at line 675 of file OMKPvmController.cxx.

References _distributedVirtualMachine, and OMK::Svm::disconnectedTableIsEmpty().

{
  //  if (_distributedVirtualMachine->getTroubleFlag() )
  if (_distributedVirtualMachine->disconnectedTableIsEmpty() )
  {
    return false;
  }
  else
    return true ;
  // dans le cas ou il y a plusieurs sites > 2 , si 2 sites sont deconnectes puis un des 2 s'est retabli alors
  // le troubleFlag est Faux par contre on a tjrs un site deconnecte donc le trouble existe encore ....
  /*  else if (_distributedVirtualMachine->disconnectedTableIsEmpty() )
  {
  return false;
  }
  return true;
  */
}

void PvmController::createNewSite

(

const Date &

)

[virtual]

Reimplemented from OMK::DistributedController.

Definition at line 694 of file OMKPvmController.cxx.

References _distributedVirtualMachine, and OMK::Svm::addNewSiteRequest().

{
  _distributedVirtualMachine->addNewSiteRequest (date);
}

std::list< Name > PvmController::getDisconnectedProcessusList

(

)

[virtual]

Reimplemented from OMK::Controller.

Definition at line 808 of file OMKPvmController.cxx.

References _distributedVirtualMachine, and OMK::Svm::getDisconnectedProcessusList().

                                                           {
  return _distributedVirtualMachine->getDisconnectedProcessusList () ;
}

std::map< Name, int > PvmController::getDisconnectedProcessusMap

(

)

[virtual]

Reimplemented from OMK::Controller.

Definition at line 812 of file OMKPvmController.cxx.

References _distributedVirtualMachine, and OMK::Svm::getDisconnectedProcessusMap().

                                                              {
  return _distributedVirtualMachine->getDisconnectedProcessusMap () ;
}

Date PvmController::getPurgeDate

(

)

[protected, virtual]

redefine to take latency into account

Reimplemented from OMK::Controller.

Definition at line 800 of file OMKPvmController.cxx.

References _distributedVirtualMachine, OMK::Controller::getPurgeDate(), and OMK::Svm::getSynchronisationLatency().

{
  //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 DistributedController::getPurgeDate() - 2 * _distributedVirtualMachine->getSynchronisationLatency() ;
}

MirrorObjectHandle * PvmController::createMirrorObject

(

ObjectDescriptor *

objectDescription

)

[protected, virtual]

Create a miror object.

Reimplemented from OMK::DistributedController.

Definition at line 243 of file OMKPvmController.cxx.

References OMK::DistributedController::createMirrorObject(), and OMK::MirrorObjectHandle::registerToReferenceObject().

{
  MirrorObjectHandle * mirror =  DistributedController::createMirrorObject( objectDescription ) ;

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

  return mirror ;
}

DuplicatedObjectHandle * PvmController::newOMKDuplicatedObjectHandle

(

SimulatedObject &

object

)

[protected, virtual]

create an object handle adapted to handling of duplicated objects

Parameters:

object

: the object the object handle will have to handle : the adapted duplicated object handle

Implements OMK::DistributedController.

Definition at line 618 of file OMKPvmController.cxx.

Referenced by PvmController().

{
  return new PvmDuplicatedObjectHandle( obj, *this );
}

LocalObjectHandle * PvmController::newOMKLocalObjectHandle

(

SimulatedObject &

object

)

[protected, virtual]

CHADI.

Implements OMK::DistributedController.

Definition at line 625 of file OMKPvmController.cxx.

{
  return new PvmLocalObjectHandle( obj, *this );
}

MirrorObjectHandle * PvmController::newOMKMirrorObjectHandle

(

SimulatedObject &

obj

)

[protected, virtual]

create an object handle for objects mirrored by this controller

Parameters:

object

: the object the object handle will have to handle : the mirror object handle

Implements OMK::DistributedController.

Definition at line 612 of file OMKPvmController.cxx.

{
  return new PvmMirrorObjectHandle(obj);
}

ReferenceObjectHandle * PvmController::newOMKReferenceObjectHandle	(	SimulatedObject &	object,
		Controller &	controller,
		SignalDispatcher *	signalDispatcher
	)			[protected, virtual]

create a reference object handler appropriate for this controller

Parameters:

obj

the object to handle the reference object created

Reimplemented from OMK::Controller.

Definition at line 703 of file OMKPvmController.cxx.

{
  return new PvmReferenceObjectHandle(object, controller, signalDispatcher );
}

int PvmController::getOutputHistorySize

(

void

)

[protected, virtual]

get the size of the history fifo needed when running a distributed simulation.

The result depends on the latency used

Reimplemented from OMK::Controller.

Definition at line 566 of file OMKPvmController.cxx.

References _distributedVirtualMachine, OMK::Controller::_stepPeriod, OMK::Controller::getOutputHistorySize(), and OMK::Svm::getSynchronisationLatency().

                                            {
  // 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 DistributedController::getOutputHistorySize()+nbrValeurPendantLatence;
}  

ObjectHandle * PvmController::removeObjectFromDataStructures

(

const Name &

nom

)

[protected, virtual]

gestion de la destruction distribuée : destruction des structures locales

Reimplemented from OMK::Controller.

Definition at line 580 of file OMKPvmController.cxx.

References OMK::DistributedController::_duplicatedObjectsMap, OMK::DistributedController::_mirrorObjectsMap, OMK::Controller::_referenceObjectsMap, and OMK::Controller::removeObjectFromDataStructures().

{

  ObjectHandle * result = NULL ;

  if (_referenceObjectsMap.find (nom) != _referenceObjectsMap.end ()) 
  {
    result = Controller::removeObjectFromDataStructures(nom);
  } 
  else 

  {
    MirrorObjectsContainerType::iterator i = _mirrorObjectsMap.find (nom) ;
    if (i != _mirrorObjectsMap.end ()) 
    {
      MirrorObjectHandle * 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 ; 
}

void PvmController::synchronizedReceiveAndProcessMessages

(

void

)

[protected]

Definition at line 370 of file OMKPvmController.cxx.

References _distributedVirtualMachine, OMK::PvmMessage::SynchronisationMessage, and OMK::Svm::synchroniseReceiveAndProcessMessages().

Referenced by PvmController().

                                                               {
  _distributedVirtualMachine->synchroniseReceiveAndProcessMessages (*this, PvmMessage::SynchronisationMessage) ;
}

void PvmController::relaxedReceiveAndProcessMessages

(

void

)

[protected]

Definition at line 376 of file OMKPvmController.cxx.

References _distributedVirtualMachine, OMK::Svm::relaxedSynchroniseReceiveAndProcessMessages(), and OMK::PvmMessage::SynchronisationMessage.

Referenced by PvmController().

                                                          {
  _distributedVirtualMachine->relaxedSynchroniseReceiveAndProcessMessages (*this, PvmMessage::SynchronisationMessage) ;
}

---

Member Data Documentation

std::vector<Name> OMK::PvmController::_publicIsolatedMirrorObjectsVector

Definition at line 120 of file OMKPvmController.h.

Referenced by getIsolatedMirrorList(), and getNonIsolatedMirrorList().

std::vector<Name> OMK::PvmController::_publicNonIsolatedMirrorObjectsVector

Definition at line 122 of file OMKPvmController.h.

Referenced by getNonIsolatedMirrorList().

Svm* OMK::PvmController::_distributedVirtualMachine [protected]

the distributed virtual machine used by the controller

Definition at line 184 of file OMKPvmController.h.

Referenced by advanceSimulatedDate(), computeNextSimulationStep(), createNewSite(), getDisconnectedProcessusList(), getDisconnectedProcessusMap(), getDistributedVirtualMachine(), getIsolatedMirrorList(), getNonIsolatedMirrorList(), getOutputHistorySize(), getProcessName(), getPurgeDate(), init(), PvmController(), relaxedReceiveAndProcessMessages(), run(), synchronizedReceiveAndProcessMessages(), testIfTroubleOccured(), waitForAnswerToBlockingRequest(), and ~PvmController().

bool OMK::PvmController::_finishing [protected]

boolean indicating the controller is going into shutdown mode.

No blocking calls should be made

Definition at line 193 of file OMKPvmController.h.

Referenced by finish(), sendInitialValuesToMirror(), and waitForAnswerToBlockingRequest().

void(PvmController::* OMK::PvmController::receiveAndProcessMessages)(void) [protected]

to manage the choice of the synchronization

Referenced by computeNextSimulationStep(), and PvmController().

---