OMKPvmReferenceObjectHandle.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 "OMKTracer.h"
#include <cassert>
#include <OMKPvmReferenceObjectHandle.h>
#include <OMKSimulatedObject.h>
#include <OMKPvmMessage.h>
#include <OMKObjectDescriptor.h>
#include <OMKSvmLink.h>
#include <OMKProcess.h>
#include <OMKController.h>
//#include <OMKNameServer.h>

#include <OMKPvmController.h>
#include "OMKSvm.h"
#include "OMKEventListener.h"
#include "OMKIncomingSynchronisationMessage.h"

using namespace std ;
using namespace OMK ;

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

PvmReferenceObjectHandle::PvmReferenceObjectHandle (SimulatedObject & object, 
                                                          Controller & controller)
  : ReferenceObjectHandle ( object, controller),
    morphosisPhase ( false )
{
}

PvmReferenceObjectHandle::PvmReferenceObjectHandle (SimulatedObject & object, 
                                                          Controller & controller, 
                                                          SignalDispatcher * signalDispatcher)
  : ReferenceObjectHandle ( object, controller, signalDispatcher),
    morphosisPhase ( false )
{
}

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

PvmReferenceObjectHandle::~PvmReferenceObjectHandle () 
{
}


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

void PvmReferenceObjectHandle::emettreValAttributs (PvmOutgoingMessage *messFD,
                                                       PvmOutgoingMessage *messEV) 
{

}


void PvmReferenceObjectHandle::insertInStream (std::ostream & out) const 
{
  Controller::error ("On utilise l'insert du referentiel") ;
}

//-----------------------------------------------------------------------------
void PvmReferenceObjectHandle::unpack (IncomingSynchronisationMessage & in) 
{
#if defined (_DEBUGPVMMESS)
  cerr << "PvmReferenceObjectHandle::unpack"<<endl;
#endif
  // Frequency max des refs du pss distant
  Frequency freqOMKsDist ;
  // Frequency de ce pss
  Frequency freqLocale ;
  // Frequency de communication
  Frequency  freqComm ;

  RequestType typeMess ;
  int messageType ;

  in >> messageType ;
  typeMess = static_cast <RequestType> (messageType) ;

#if defined (_DEBUGPVMMESS)
  cerr << "PvmReferenceObjectHandle::unpack: received a message of type " 
       <<typeMess<<" "<<Registration
       <<" at date "<<getSimulatedObject().getSimulatedDate()
       <<endl;
#endif
 
  if (typeMess == Registration) 
    {
      // name of the process sending the message
      Name processName ;
         
      processName.unpack ( in ) ;

#ifdef _DEBUGPVMMESS
      cout << "PvmReferenceObjectHandle::unpack " << processName << " registering a mirror "<< endl ;
#endif
      SvmLink * canal = dynamic_cast<PvmController &>(_controller).getDistributedVirtualMachine()->getLinkToProcessNamed (processName) ;

      Process * processDescriptor = dynamic_cast<PvmController &>(_controller).getDistributedVirtualMachine()->getProcessDescriptorNamed (processName) ;
         
      //the registering process needs a complete version of the outputs of the object in the next synchronisation message
      _needAllOutputValues.insert ( make_pair(processName, canal) ) ;

      //         // On recupere la frequency max des refs du pss distant
      //         freqOMKsDist = processDescriptor->getMaxFrequencyOfHostedObjects () ;
      //         // On recupere la frequency locale
      //         freqLocale = _myObject. getObjectDescriptor ().getFrequency () ;
      //         // On calcule la frequency de communication
      //  #ifdef _DEBUGPVMMESS
      //         cout << "freqLocale : " << freqLocale << endl ;
      //         cout << "freqOMKsDist : " << freqOMKsDist << endl ;
      //  #endif
      //         if (freqLocale > freqOMKsDist) 
      //            {
      //               freqComm = plusPetitSSMultSup(freqLocale,
      //                                              freqOMKsDist) ;
      //  #ifdef _DEBUGPVMMESS
      //               cout << "freqComm : " << freqComm << endl ;
      //  #endif
      //            } 
      //         else 
      //            {
      //               freqComm = freqLocale ;
      //  #ifdef _DEBUGPVMMESS
      //               cout << "freqComm  : " << freqComm << endl ;
      //  #endif
      //            }
      //         // On la stocke
      //         _processFrequencies.insert (ProcessFrequenciesType::value_type(processName, freqComm)) ;
    } 
  else if (typeMess == CancelRegistration) 
    {
     OMTRACEID("DEBUGPVM", "Cancel : " << "avant unpack"); 
      // name of the process sending the message
      Name processName ;
      processName.unpack ( in ) ;
      OMTRACEID("DEBUGPVM", "Cancel : " << "avant ProcessName is" << processName); 

      _needUpdateOutputValues.erase (processName) ; 
      //_processFrequencies.erase ( processName ) ;
      _needAllOutputValues.erase ( processName ) ;
    } 
  else if (typeMess == EventReceived) 
    {
#if ( defined (_DEBUGPVMMESS) || defined (_DEBUGPVMEVENTS ) )
      cerr << "PvmReferenceObjectHandle::unpack: received an event relayed from an mirror" << endl ;
#endif
      Event * event ;
      Name classToCreate ;
      EventIdentifier eventId;
      Date eventDate;
      Name sender, receiver ;
      in >> classToCreate >> eventId>> eventDate>>sender>>receiver ;
      event = EventCreator::createEvent (classToCreate, eventId, eventDate, sender, receiver ) ;
      event->unpack ( in ) ;
      ReferenceObjectHandle::receiveEvent ( event ) ;
    } 
  else if (typeMess == RegisterForSignal) 
    {
#ifdef _DEBUGPVMMESS
      cout << "PvmReferenceObjectHandle::unpack: RegisterForSignal" << endl ;
#endif
      Name sigId ;
      Name registrant ;
      Name eventId ;
      in >> sigId >> registrant >> eventId ;
      receiveRegistrationForSignal (sigId, registrant, eventId );
    } 
  else if (typeMess == CancelRegistrationForSignal) 
    {
#ifdef _DEBUGPVMMESS
      cout << "PvmReferenceObjectHandle::unpack: CancelRegistrationForSignal" << endl ;
#endif
      Name sigId ;
      Name registrant ;
      in >> sigId >> registrant ;
      cancelRegistrationForSignal (sigId, registrant) ;
    }
  else if (typeMess == SynchronisationMessage) // ajout TDTD pour essayer de voir ce qui ne va pas...
    {
       //if (morphosisPhase) { // TDTD il me semble pourtant que ce serait bien...
       //cerr<<"PvmReferenceObjectHandle::unpack : TDTD : SynchronisationMessage for" << getSimulatedObject ().getName () <<endl; ;
          _myObject.unpack ( in ) ;
          // semble poser des pbs pour la suppression d'objets locaux...
          //_myObject.unpackAllValues ( in ) ;
          //}
    }
  else 
    {
      cerr<<"PvmReferenceObjectHandle::unpack : unknown request"<<typeMess<<endl; ;
      cerr << "messageType : " << messageType << endl ;
          assert( false && "You should not be there..." ) ;
    }

#if defined (_DEBUGPVMMESS)
  cerr << "PvmReferenceObjectHandle::unpack finished"<<typeMess <<endl;
#endif
}
   


void PvmReferenceObjectHandle::extract( istream & in ) 
{
#if defined (_DEBUGPVMMESS)
  cerr << "PvmReferenceObjectHandle::extract:"<<endl;
#endif
  // Type de message
  std::string typeMess ;
  // Nom du processus ou se trouve le miroir (cas de l'EV)
  Name processName ;

  in >> typeMess ;
  //cerr << "PvmReferenceObjectHandle::extract: received a message of type " <<typeMess<<endl;
  if( typeMess == "EV" ) 
  {
    int typeEv ;
    in >> typeEv ;
    cerr << "PvmReferenceObjectHandle::extract: received an event message of type " <<typeEv<<endl;
    if (typeEv == Registration) 
    {
      // Frequency max des refs du pss distant
      Frequency freqOMKsDist ;
      // Frequency de ce pss
      Frequency freqLocale ;
      // Frequency de communication
      Frequency  freqComm ;

      // On recupere le nom du processus associe
      in >> processName ;
#ifdef _DEBUGPVMMESS
      cout << "PvmReferenceObjectHandle::extract "<<processName << " registred a mirror "<< endl ;
#endif
      // On recupere le processus associe
      SvmLink * canal = dynamic_cast<PvmController &>(_controller).getDistributedVirtualMachine()->getLinkToProcessNamed (processName) ;

      Process * processDescriptor = dynamic_cast<PvmController &>(_controller).getDistributedVirtualMachine()->getProcessDescriptorNamed (processName) ;

      //the registering process needs a complete version of the outputs of the object in the next synchronisation message
      _needAllOutputValues.insert ( make_pair (processName,canal) ) ;

      //            // On recupere la frequency max des refs du pss distant
      //            freqOMKsDist = processDescriptor->getMaxFrequencyOfHostedObjects () ;
      //            // On recupere la frequency locale
      //            freqLocale = _myObject. getObjectDescriptor ().getFrequency () ;
      //            // On calcule la frequency de communication
      //  #ifdef _DEBUGPVMMESS
      //            cout << "freqLocale : " << freqLocale << endl ;
      //            cout << "freqOMKsDist : " << freqOMKsDist << endl ;
      //  #endif
      //            if (freqLocale > freqOMKsDist) {
      //               freqComm = plusPetitSSMultSup(freqLocale,
      //                                              freqOMKsDist) ;
      //  #ifdef _DEBUGPVMMESS
      //               cout << "freqComm : " << freqComm << endl ;
      //  #endif
      //            } else {
      //            freqComm = freqLocale ;
      //  #ifdef _DEBUGPVMMESS
      //            cout << "freqComm  : " << freqComm << endl ;
      //  #endif
      //            }
      //         // On la stocke
      //            _processFrequencies.insert(ProcessFrequenciesType::value_type(processName, freqComm)) ;
    } 
    else if (typeEv == CancelRegistration) {
      //cout << "On se desabonne" << endl ;
    } 
    else if (typeEv == EventReceived) {
      //#ifdef DEBUG
      cerr << "ALERTE PvmReferenceObjectHandle::extract : On recoit un evenement en provenance d'un miroir" << endl ;
      //#endif
      assert ( false ) ;
      //Event evt ( in ) ;
      //in >> evt ;
      //receiveReallyEvent (evt) ;
    } 
    else if (typeEv == RegisterForSignal) 
    {
#ifdef DEBUG
      cout << "On recoit un signal en provenance d'un miroir" << endl ;
#endif
      Name sigId ;
      Name abonne ;
      Name eventId ;
      in >> sigId >> abonne >> eventId ;
      receiveRegistrationForSignal (sigId, abonne, eventId );
    } 
    else {
      cerr<<"PvmReferenceObjectHandle::extract : Type d'evenement incorrect"<<endl ;
    }
  }
  else {
    cerr<<"PvmReferenceObjectHandle::extract : Type d'evenement incorrect" <<endl ;
  }
}


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

void PvmReferenceObjectHandle::makeSynchronisationMessage( const Date & date ) 
{
  if (!morphosisPhase)
  {
#if defined (_DEBUGEXECPVM) || defined (_DEBUGPVMMESS) 
    cerr<<"PvmReferenceObjectHandle::makeSynchronisationMessage ( const Date & "<<date<< ")"<<endl;
#endif
    PvmOutgoingMessage * message ;

    if ( _dateOfLastActivation == date )
      // only send a synchronisation message if the handled object has been activated
    {
      NameToPointerMap<SvmLink>::iterator pCanal ; // iterateur
      //         ProcessFrequenciesType::iterator pFreq ; // iterateur
      //         Date dtComm ;
      //         Frequency freqComm ;
#ifdef _DEBUGPVMMESS
      cout<<"makeSynchronisationMessages of "<<getSimulatedObject().getName()<<endl;
#endif
      // On initialise l'iterateur sur la frequency
      //pFreq = _processFrequencies.begin () ;
      //Pour tous les abonnes
      for (pCanal = _needUpdateOutputValues.begin () ;
        pCanal != _needUpdateOutputValues.end ();
        pCanal ++) 
      {
        // On calcule le pas de temps de la communication
        //             freqComm = (*pFreq).second ;
        //             dtComm = 1000 / freqComm ;
#ifdef _DEBUGPVMMESS
        //             cout << "dtComm : " << dtComm << endl ;
#endif
        // On verifie qu'il faut reellement emettre
        // devrait maintenant etre inutile a cause des Frames majeures et mineures
        //             if (_controller.getSimulatedDate () % dtComm == 0) 
        //                {
        // On recupere le message ou stocker les donnees
        message = &(*pCanal).second->getOutgoingBuffer () ;  
#ifdef _DEBUGPVMMESS
        cout <<"PvmReferenceObjectHandle::makeSynchronisationMessage writing to the buffer: " <<message<< endl ;
#endif
        // On va construire le message
        // On ajoute le destinataire, puis le type du message puis les outputs
        *message << _myObject.getName ()  
          << static_cast<int> (SynchronisationMessage) 
          <<  _myObject ;

#ifdef _DEBUGDISTRIBUTEDINIT
        std::cerr<<"PvmReferenceObjectHandle::makeSynchronisationMessage : sending a message to ";
        (*pCanal).second->printDebuggingInformation();
#endif
      }
      // On passe a la frequency suivante
      //        pFreq ++ ;
      //        }
    }
    else
    {
#ifdef _DEBUGPVMMESS
      cerr <<"PvmReferenceObjectHandle::makeSynchronisationMessage : No synchronisation needed at " 
        << date 
        <<", last activation : "<<_dateOfLastActivation<<endl ;
#endif
    }
    // send complete values to the processes needing a complete list of the values with their associated production date.
    // this is done after updating, because processes in that list are then automatically inserted in the list of process needing only updates
    for (std::map<Name, SvmLink *>::iterator i = _needAllOutputValues.begin() ;
      i != _needAllOutputValues.end() ;
      ++i)
    {
      packInitialValues ( (i->second)->getOutgoingBuffer () ) ; 

      _needUpdateOutputValues.insert ( *i ) ;
    }
    _needAllOutputValues.clear() ;
#ifdef _DEBUGPVMMESS
    cerr <<"PvmReferenceObjectHandle::makeSynchronisationMessage end" <<endl ;
#endif 

  }
}

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

int PvmReferenceObjectHandle::plusPetitSSMultSup (const int A, const int B) {
  // Fournit le plus petit sous-multiple de A plus grand que B
  int i = B ; // Valeur testee
  // Valeur max au dela de laquelle pas ss-multiple
  int valMax = div (A, 2).quot ; 
  bool ssMult = false ; // i ss-multiple de A ???

#ifdef DEBUG
  cout << "plusPetitSSMultSup***************************************" << endl ;
  cout << "A : " << A << ", B : " << B << endl ;
  cout << "valMax : " << valMax << endl ;
#endif

  // On verifie si i ss-multiple de A
  ssMult = (div (A, i).rem == 0) ;
  while ((! ssMult) && (i <= valMax)) {
    i++ ;
    // On verifie si i ss-multiple de A
    ssMult = (div (A, i).rem == 0) ;
  }
  if (i > valMax) {
#ifdef DEBUG
    cout << "res : " << A << endl ;
    cout << "Fin de plusPetitSSMultSup***************************************" << endl;
#endif  
    return A ;
  } else {
#ifdef DEBUG
    cout << "res : " << i << endl ;
    cout << "Fin de plusPetitSSMultSup***************************************" << endl;
#endif  
    return i ;
  }
}


void PvmReferenceObjectHandle::packInitialValues ( PvmOutgoingMessage & message ) 
{
  _myObject.getName ().pack ( message ) ; 
  message << static_cast<int> (InitialValuesMessage) ;
  _myObject.packAllValues ( message ) ;
}

void PvmReferenceObjectHandle::flushDisconnectedSitesOutgoingBuffer (  ) 
{

  NameToPointerMap<SvmLink>::iterator pCanal ;
  for (pCanal = _needUpdateOutputValues.begin () ;
    pCanal != _needUpdateOutputValues.end ();
    pCanal ++) 
  {
    (*pCanal).second->getOutgoingBuffer().flushCurrentBuffer() ;  
  }
}

void PvmReferenceObjectHandle::setMorphosisPhaseTrue ()
{
  morphosisPhase = true ;
}

---