OMKPvmNameServer.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 "OMKPvmNameServer.h"

#include <cassert>
#include <pvm3.h>
#include "OMKSvmLink.h"
#include "OMKController.h"
#include "OMKPvmSvm.h"
using namespace std ;
using namespace OMK ;

//#define _DEBUGDISTRIBUTEDNAMESERVER

PvmNameServer::PvmNameServer ( int mainNameServerSiteId, NameServer & nameServer ) :
NameServer ( nameServer.getCanonicalRepresentation() ),
_mainNameServerSiteId ( mainNameServerSiteId )
{
#ifdef _DEBUGDISTRIBUTEDNAMESERVER
  cerr<<"PvmNameServer::PvmNameServer"<<endl;
#endif

  correspondenceContainerType::const_iterator i ;
  for ( i = _stringToIdCorrespondenceContainer.begin() ; 
        i != _stringToIdCorrespondenceContainer.end();
        ++i ) 
  {
    _idToNamesUsingIdContainer.insert( placeOfNamesContainerType::value_type((*i).second,
                                       new list<Name *>(nameServer.getNamesOfId ((*i).second) ) ) ) ;

    assert ( (*_idToNamesUsingIdContainer.find((*i).second)).second->size() == _tableNbInstances[(*i).second] ) ;
  }

  //make sure the constructed PvmName server is compatible with the central name server
  int oldbufid = pvm_setsbuf ( 0 ) ; //save the current send buffer 

  if ( oldbufid < 0 )
  {
    cerr<<"PvmNameServer::PvmNameServer: pvm_setsbuf error "<<oldbufid<<endl;
    OMASSERT( false ) ;
  }

  int bufid = pvm_initsend( PvmSvm::pvmDataEncoding ) ; 

  if ( bufid < 0 )
  {
    cerr<<"PvmNameServer::PvmNameServer: pvm_initsend error "<<bufid<<endl;
    OMASSERT( false ) ;
  }

  //pack a timeStamp
  long toto = 0 ;
  int infot = pvm_pklong ( &toto, 1, 1 ) ;

  //pack the current name server

  long numberOfPairs = _stringToIdCorrespondenceContainer.size() ;
  int info = pvm_pklong ( &numberOfPairs, 1, 1 ) ;

  long verifySize = 0 ;
  int stringLength ;
  for ( correspondenceContainerType::const_iterator i = _stringToIdCorrespondenceContainer.begin() ;
    i != _stringToIdCorrespondenceContainer.end() ;
    ++i )
  {
    ++verifySize ;
    stringLength = (*i).first.length() + 1 ;
    pvm_pkint ( const_cast<int *>( &stringLength ), 1, 1 ) ;
    pvm_pkstr ( const_cast<char *>( (*i).first.c_str() ) ) ;
    pvm_pklong ( const_cast<long *>( &( (*i).second ) ) , 1, 1 ) ;
  }

  assert (verifySize == numberOfPairs ) ;

  if (info < 0 )
  {
    cerr<<"PvmNameServer::PvmNameServer: pvm_pkstr error "<<info<<endl;
    OMASSERT( false ) ;
  }

  info = pvm_send (_mainNameServerSiteId, PvmMessage::NameServiceVerifyLocalNameServer ) ;

  if (info < 0 )
  {
    cerr<<"PvmNameServer::PvmNameServer: pvm_send error "<<info<<endl;
    OMASSERT( false ) ;
  }

  info = pvm_setsbuf ( oldbufid ) ; //replace the current send buffer

  if (info < 0 )
  {
    cerr<<"PvmNameServer::PvmNameServer: pvm_setsbuf error "<<info<<endl;
    OMASSERT( false ) ;
  }

  //save the current receive buffer
  oldbufid = pvm_setrbuf ( 0 ) ;

  //wait for the answer
  bufid = pvm_recv (_mainNameServerSiteId, PvmMessage::NameServiceVerifyResult ) ;

  if ( bufid < 0 )
  {
    cerr<<"PvmNameServer::PvmNameServer: pvm_recv error "<<bufid<<endl;
    OMASSERT( false ) ;
  }

  int numberOfIncompatibilities ;
  info = pvm_upkint ( & numberOfIncompatibilities, 1, 1 ) ;


  if (info < 0 )
  {
    cerr<<"PvmNameServer::PvmNameServer: pvm_upkint error "<<info<<endl;
    OMASSERT( false ) ;
  }

  Name::idType oldId ;
  Name::idType newId ;
  std::map<Name::idType, Name::idType> listOfNeededConversions ;
  for ( int i = 0 ; i < numberOfIncompatibilities ; ++ i )
  {
    info = pvm_upklong ( & oldId, 1, 1 ) ;

    if (info < 0 )
    {
      cerr<<"PvmNameServer::PvmNameServer: pvm_upklong error "<<info<<endl;
      OMASSERT( false ) ;
    }

    info = pvm_upklong ( & newId, 1, 1 ) ;      
    if (info < 0 )
    {
      cerr<<"PvmNameServer::PvmNameServer: pvm_upklong error "<<info<<endl;
      OMASSERT( false ) ;
    }

    listOfNeededConversions[oldId] = newId ;

  }


  std::map<Name::idType, Name::idType>::iterator  j = listOfNeededConversions.begin() ;
  while ( j != listOfNeededConversions.end() )
  {
    oldId = (*j).first ;
    newId = (*j).second ;

    cerr<<"PvmNameServer::PvmNameServer: converting "<<oldId<<" to "<<newId<<endl;

    assert ( oldId < _nextIdentifier ) ;

    if ( newId >= _nextIdentifier )
    {
      cerr<<"PvmNameServer::PvmNameServer: updating nextIdentifier from "<<_nextIdentifier<<" to "<< newId + 1 <<endl;
      _nextIdentifier = newId + 1 ;
    }

    //replace the id in the name server warning : newId could be known or unknown from the name server
    bool newIdPresent = ( _tableNbInstances.find(newId) != _tableNbInstances.end() ) ;

    // - first take care of the number of references to the string corresponding to each Id
    referenceCountingContainerType::iterator oldRefCount = _tableNbInstances.find(oldId) ;
    referenceCountingContainerType::iterator newRefCount = _tableNbInstances.find(newId) ;

    int referencesToOldId = (*oldRefCount).second ;
    _tableNbInstances.erase ( oldRefCount ) ;


    if ( newIdPresent )
    {
      assert ( newRefCount != _tableNbInstances.end() ) ;
      int referencesToNewId = (*newRefCount).second ;
      _tableNbInstances.erase ( newRefCount ) ;
      _tableNbInstances.insert(referenceCountingContainerType::value_type(oldId, referencesToNewId ) ) ;
    }

    //    - only insert once the eventual copy has been archived
    _tableNbInstances.insert(referenceCountingContainerType::value_type(newId, referencesToOldId ) ) ;


    // - then take care of the list of Name instances of each Id              
    placeOfNamesContainerType::iterator oldNameList = _idToNamesUsingIdContainer.find(oldId) ;
    placeOfNamesContainerType::iterator newNameList = _idToNamesUsingIdContainer.find(newId) ;

    list<Name *> * referencesToOldList = (*oldNameList).second ;
    _idToNamesUsingIdContainer.erase ( oldNameList ) ;

    if ( newIdPresent )
    {
      assert ( newNameList != _idToNamesUsingIdContainer.end() ) ;
      list<Name *> * referencesToNewList = (*newNameList).second ;
      _idToNamesUsingIdContainer.erase ( newNameList ) ;
      _idToNamesUsingIdContainer.insert(placeOfNamesContainerType::value_type(oldId, referencesToNewList) ) ;
    }

    //    - only insert once the eventual copy has been archived
    _idToNamesUsingIdContainer.insert(placeOfNamesContainerType::value_type(newId, referencesToOldList) ) ;


    // - then take care of the std::string to Id references 
    std::string newStringOfNewId ( (*(*_fastStringAccessTable.find(oldId)).second ).first ) ;
    _fastStringAccessTable.erase ( _fastStringAccessTable.find ( oldId ) ) ;
    _stringToIdCorrespondenceContainer.erase ( _stringToIdCorrespondenceContainer.find (newStringOfNewId) );

    if ( newIdPresent )
    {
      std::string newStringOfOldId ( (*(*_fastStringAccessTable.find(newId)).second ).first ) ;
      _fastStringAccessTable.erase ( _fastStringAccessTable.find ( newId ) ) ;
      _stringToIdCorrespondenceContainer.erase ( _stringToIdCorrespondenceContainer.find (newStringOfOldId) );
      pair<correspondenceContainerType::iterator,bool> insertRes = _stringToIdCorrespondenceContainer.insert(correspondenceContainerType::value_type(newStringOfOldId, oldId)) ;
      _fastStringAccessTable[oldId] = insertRes.first ;
    }

    //    - only insert once the eventual copy has been archived
    pair<correspondenceContainerType::iterator,bool> res = _stringToIdCorrespondenceContainer.insert(correspondenceContainerType::value_type(newStringOfNewId, newId)) ;
    _fastStringAccessTable[newId] = res.first ;


    //here, newId and oldId have been exchanged in tha data structures, but not in the OMKNames nor in the listOfNeededConversions. Therefore, we will replace the Name containing oldId with OMKNames containing newId, and change the list Ofneeded conversions so that the conversion from newId to anotherId is replaced by a conversion of oldId to anotherId

    // - replace the id in the OMKNames
    newNameList = _idToNamesUsingIdContainer.find ( newId ) ;

    assert ( newNameList != _idToNamesUsingIdContainer.end() ) ;
    assert ( (*newNameList).second != NULL ) ;
    assert ( (*newNameList).second->size() == _tableNbInstances[ newId ] ) ;

    for ( list<Name *>::iterator listIter = (*newNameList).second->begin() ;
      listIter != (*newNameList).second->end() ;
      ++listIter )
    {
      changeNamesId (*listIter, oldId, newId) ;
    }

    listOfNeededConversions.erase (j) ;

    if ( newIdPresent )
    {
      // - update the conversion from newId to anotherId by a conversion of oldId to anotherId
      std::map<Name::idType, Name::idType>::iterator k = listOfNeededConversions.find ( newId ) ;
      assert ( k != listOfNeededConversions.end() ) ;

      Name::idType anotherId = (*k).second ;

      listOfNeededConversions.erase ( k ) ;

      if ( anotherId != oldId )
      {
        listOfNeededConversions[oldId] = anotherId ;
      }

      // - update the Name so that assertions won't fail and for final permutations
      newNameList = _idToNamesUsingIdContainer.find ( oldId ) ;
      for ( list<Name *>::iterator listIter = (*newNameList).second->begin() ;
        listIter != (*newNameList).second->end() ;
        ++listIter )
      {
        changeNamesId (*listIter, newId, oldId) ;
      }

      cerr<<"PvmNameServer::PvmNameServer: "<<oldId<<" will now be converted in "<<anotherId<<endl; 
    }

    j = listOfNeededConversions.begin() ;
  }

  printToStream ( cout, "\n" ) ;

  cerr<<"Done "<<endl;


  //restore old receive buffer
  info = pvm_setrbuf ( oldbufid );
  assert (info != PvmBadParam ) ;
  assert (info != PvmNoSuchBuf ) ;

#ifdef _DEBUGDISTRIBUTEDNAMESERVER
  cerr<<"PvmNameServer::PvmNameServer done"<<endl;
#endif

}


PvmNameServer::~PvmNameServer () 
{

} 

Name::idType
PvmNameServer::getIdentifierAsFrom(const std::string & name, Name::idType * nextId)
{
#ifdef _DEBUGDISTRIBUTEDNAMESERVER
  cerr<<"PvmNameServer::getIdentifierAsFrom"<<endl;
#endif    

  correspondenceContainerType::iterator i;
  i = _stringToIdCorrespondenceContainer.find( name );

  if (i == _stringToIdCorrespondenceContainer.end()) 
  {
    //we are going to change the values contained in the NameServer
    //lock() ; 

    i = _stringToIdCorrespondenceContainer.find( name ) ; //name might have been added by an other thread

    if (i == _stringToIdCorrespondenceContainer.end() ) //we have to add nom
    { 
#ifdef _DEBUGDISTRIBUTEDNAMESERVER
      cerr<<"   PvmNameServer::getIdentifierAsFrom ("<<name<<") on central name server"<<endl;
#endif    
      //ask the central name server for the id corresponding to nom

      int oldbufid = pvm_setsbuf ( 0 ) ; //save the current send buffer 

      if ( oldbufid < 0 )
      {
        cerr<<"PvmNameServer::getIdentifierAsFrom: pvm_setsbuf error "<<oldbufid<<endl;
        OMASSERT( false ) ;
      }

      int bufid = pvm_initsend( PvmSvm::pvmDataEncoding ) ; 

      //pack a timeStamp
      long toto = 0 ;
      int infot = pvm_pklong ( &toto, 1, 1 ) ;

      if ( bufid < 0 )
      {
        cerr<<"PvmNameServer::getIdentifierAsFrom: pvm_initsend error "<<bufid<<endl;
        OMASSERT( false ) ;
      }
      int numberOfCaracters = name.length() + 1 ;
      int info = pvm_pkint ( &numberOfCaracters , 1, 1 ) ;

      info = pvm_pkstr ( const_cast<char *> (name.c_str()) ) ;

      if (info < 0 )
      {
        cerr<<"PvmNameServer::getIdentifierAsFrom: pvm_pkstr error "<<info<<endl;
        OMASSERT( false ) ;
      }

      info = pvm_send (_mainNameServerSiteId, PvmMessage::NameServiceGetId ) ;

      if (info < 0 )
      {
        cerr<<"PvmNameServer::getIdentifierAsFrom: pvm_send error "<<info<<endl;
        OMASSERT( false ) ;
      }

      info = pvm_setsbuf ( oldbufid ) ; //replace the current send buffer

      if (info < 0 )
      {
        cerr<<"PvmNameServer::getIdentifierAsFrom: pvm_setsbuf error "<<info<<endl;
        OMASSERT( false ) ;
      }

      //save the current receive buffer
      oldbufid = pvm_setrbuf ( 0 ) ;

      //wait for the answer
      bufid = pvm_recv (_mainNameServerSiteId, PvmMessage::NameServiceReturnId ) ;

      if ( bufid < 0 )
      {
        cerr<<"PvmNameServer::getIdentifierAsFrom: pvm_recv error "<<bufid<<endl;
        OMASSERT( false ) ;
      }

      Name::idType requestedId ;
      info = pvm_upklong ( & requestedId, 1, 1 ) ;

#ifdef _DEBUGDISTRIBUTEDNAMESERVER
      cerr<<"   got "<<requestedId<<" from the central name server"<<endl;
#endif    
      if (info < 0 )
      {
        cerr<<"PvmNameServer::getIdentifierAsFrom: pvm_upklong error "<<info<<endl;
        OMASSERT( false ) ;
      }

      //restore old receive buffer
      info = pvm_setrbuf ( oldbufid );
      assert (info != PvmBadParam ) ;
      assert (info != PvmNoSuchBuf ) ;

      pair<correspondenceContainerType::iterator,bool> res = _stringToIdCorrespondenceContainer.insert(correspondenceContainerType::value_type(name,requestedId)) ; 
      _fastStringAccessTable[requestedId] = res.first ;

      *nextId = requestedId + 1 ;

      i = res.first ;
    }

    //unlock() ;
  }
#ifdef _DEBUGDISTRIBUTEDNAMESERVER
  cerr<<"PvmNameServer::getIdentifierAsFrom returning "<<(*i).second<<endl;
#endif    
  return (*i).second;
}

const std::string& PvmNameServer::getStringAssociatedTo (Name::idType id) const 
{
  if ( _fastStringAccessTable.find(id) == _fastStringAccessTable.end() )
  {
    return const_cast<PvmNameServer *>(this)->getStringAssociatedToOnCentralServer( id ) ;
  }
  else 
  {
    return  (*(*_fastStringAccessTable.find(id)).second).first ;
  }
}

const std::string& PvmNameServer::getStringAssociatedToOnCentralServer (Name::idType id)
{
#ifdef _DEBUGDISTRIBUTEDNAMESERVER
  cerr<<"PvmNameServer::getStringAssociatedTo ("<<id<<")"<<endl;
#endif    
  if ( _fastStringAccessTable.find(id) == _fastStringAccessTable.end() )
  {
    //lock () ;

    if ( _fastStringAccessTable.find(id) == _fastStringAccessTable.end() ) 
      //the request might have allready been fullfilled
    {
      //ask the central name server for the string

      int oldbufid = pvm_setsbuf ( 0 ) ; //save the current send buffer 

      if ( oldbufid < 0 )
      {
        cerr<<"PvmNameServer::getStringAssociatedTo: pvm_setsbuf error "<<oldbufid<<endl;
        OMASSERT( false ) ;
      }

      int bufid = pvm_initsend( PvmSvm::pvmDataEncoding ) ; 

      //pack a timeStamp
      long toto = 0 ;
      int infot = pvm_pklong ( &toto, 1, 1 ) ;

      if ( bufid < 0 )
      {
        cerr<<"PvmNameServer::getStringAssociatedTo: pvm_initsend error "<<bufid<<endl;
        OMASSERT( false ) ;
      }

      int info = pvm_pklong ( &id, 1, 1 ) ;;

      if (info < 0 )
      {
        cerr<<"PvmNameServer::getStringAssociatedTo: pvm_pklong error "<<info<<endl;
        OMASSERT( false ) ;
      }

      info = pvm_send (_mainNameServerSiteId, PvmMessage::NameServiceGetString ) ;

      if (info < 0 )
      {
        cerr<<"PvmNameServer::getStringAssociatedTo: pvm_send error "<<info<<endl;
        OMASSERT( false ) ;
      }

      info = pvm_setsbuf ( oldbufid ) ; //replace the current send buffer

      if (info < 0 )
      {
        cerr<<"PvmNameServer::getStringAssociatedTo: pvm_setsbuf error "<<info<<endl;
        OMASSERT( false ) ;
      }

      //save the current receive buffer
      oldbufid = pvm_setrbuf ( 0 ) ;

      //wait for the answer
      bufid = pvm_recv (_mainNameServerSiteId, PvmMessage::NameServiceReturnString ) ;

      if ( bufid < 0 )
      {
        cerr<<"PvmNameServer::getStringAssociatedTo: pvm_recv error "<<bufid<<endl;
        OMASSERT( false ) ;
      }

      int bytes, msgtag, tid ;

      info = pvm_bufinfo ( bufid, &bytes, &msgtag, & tid ) ;

      if (info < 0 )
      {
        cerr<<"PvmNameServer::getStringAssociatedTo: pvm_bufinfo error "<<info<<endl;
        OMASSERT( false ) ;
      }

      char * resultingString = new char [bytes] ;

      info = pvm_upkstr ( resultingString ) ;

      if (info < 0 )
      {
        cerr<<"PvmNameServer::getStringAssociatedTo: pvm_upkstr error "<<info<<endl;
        OMASSERT( false ) ;
      }

      //restore old receive buffer
      info = pvm_setrbuf ( oldbufid );
      assert (info != PvmBadParam ) ;
      assert (info != PvmNoSuchBuf ) ;

      //insert the new pair in the data structures
      pair<correspondenceContainerType::iterator,bool> res = _stringToIdCorrespondenceContainer.insert(correspondenceContainerType::value_type( std::string( resultingString ), id)) ; 
      _fastStringAccessTable[id] = res.first ;

      if ( id >= _nextIdentifier ) 
      {
        _nextIdentifier = id + 1 ;
      }
    }

    //unlock() ;
  }
#ifdef _DEBUGDISTRIBUTEDNAMESERVER
  cerr<<"PvmNameServer::getStringAssociatedTo ("<<id<<") ="<<(*(*_fastStringAccessTable.find(id)).second).first<<endl;
#endif    
  return  (*(*_fastStringAccessTable.find(id)).second).first ;
}

bool PvmNameServer::includes (const NameServer & otherNameServer ) 
{
  return true ;
}

void PvmNameServer::deleted(Name::idType id, Name * name)
{
}

void PvmNameServer::created(Name::idType id, Name * )
{
}

---