OMKPvmSvmLink.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 <OMKPvmSvmLink.h>

#include <pvm3.h>

#include <OMKPvmMessage.h>
#include <OMKController.h>
#include "OMKPvmException.h"
using namespace std ;
using namespace OMK ;

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

PvmSvmLink::PvmSvmLink (int td) : 
  SvmLink(),
  _outgoingBuffer ( td ),
  _distantSiteId ( td ) 
{
}

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

PvmSvmLink::~PvmSvmLink() 
{
}



void PvmSvmLink::send (PvmOutgoingMessage & m, PvmMessage::MessageTag tag) 
{
  m.send ( tag ) ;
}



void PvmSvmLink::sendOutgoingBuffer(PvmMessage::MessageTag tag) 
{
#ifdef _DEBUGPVMMESS
  cerr << "PvmSvmLink::sendOutgoingBuffer (" << tag <<") "<<endl;
#endif
  _outgoingBuffer.send ( tag ) ;
}

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

PvmIncomingMessage & PvmSvmLink::waitForMessage(PvmMessage::MessageTag tag) 
{
  // attente en reception
  int res = pvm_recv (_distantSiteId, tag) ;
   
  _incomingBuffer.initialise ( res ) ;
#ifdef _DEBUGPVMMESS
  cerr << "waitForMessage : On recoit un message  tag: " << tag << " de " << hex << _distantSiteId << dec << endl;
#endif
  // On renvoie le message
  return _incomingBuffer ;
}

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

PvmIncomingMessage * PvmSvmLink::testForMessage (PvmMessage::MessageTag tag) 
{
#ifdef _DEBUGPVMMESS
  cerr << "PvmSvmLink::testForMessage: tag  " <<tag<<endl;
#endif        

  int res = pvm_nrecv (_distantSiteId, tag) ;

  if (res >= 0) 
    { 
      _incomingBuffer.initialise ( res );
      return &_incomingBuffer ;
    } 
  else 
    {
      return NULL ;
    }
}

PvmIncomingMessage*  PvmSvmLink::testIfRequestUpdateProcessTable (PvmMessage::MessageTag tag) 
{
#ifdef _DEBUGPVMMESS
  cerr << "PvmSvmLink::testForMessage: tag  " <<tag<<endl;
#endif        
 
  int res = pvm_nrecv (_distantSiteId, PvmMessage::updateProcessTable) ;
 
  if (res > 0) 
    { 
      _incomingBuffer.initialise ( res );
      return &_incomingBuffer ;
    } 
  else 
    {
      return NULL ;
    }
}
PvmMessage::MessageTag 
PvmSvmLink::testForAnyMessage (PvmIncomingMessage & receiveBuffer) 
{
  PvmMessage::MessageTag result ;
  
  int bufid = pvm_nrecv  ( _distantSiteId, -1 ) ; //test if any message has arrived from distantDiteId

  if ( bufid < 0 ) 
    {
      std::cerr<<"Svm::waitForAnyRequests ERROR ";
      switch (bufid)
        {
        case PvmBadParam:
          std::cerr<<"giving an invalid tid value, or msgtag < -1";
          break;
        case PvmSysErr:
          std::cerr<<"pvmd not responding";
          break;
        default:
          std::cerr<<"unexpected error";
                OMASSERT( false ) ;
        }
      std::cerr<<std::endl;
    }

  if ( bufid != 0 )
    {
      int bytes ;
      int senderSiteId ;
      int msgtag ;
      int info = pvm_bufinfo( bufid, &bytes , &msgtag , &senderSiteId  ) ;
        
      result = static_cast<PvmMessage::MessageTag>(msgtag) ;
        
      if ( info < 0 ) 
        {
          cerr<<"Svm::waitForAnyRequests ERROR in pvm_bufinfo ";
          switch (info)
            {
            case PvmBadParam:
              cerr<<"invalid argument";
              break;
            case PvmNoSuchBuf:
              cerr<<"specified buffer does not exist";
              break;
            default:
              cerr<<"unexpected error";
                OMASSERT( false ) ;
            }
          cerr<<endl;
        }
    }

  receiveBuffer.initialise ( bufid ) ;
  
  return result ;
}

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

void PvmSvmLink::printDebuggingInformation () 
{
  cerr << "PvmSvmLink:"<<this<<":printDebuggingInformation:_distantSiteIdant : " << hex << _distantSiteId << dec << endl ;
}

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

int PvmSvmLink::getTID () 
{
  return _distantSiteId ;
}

void PvmSvmLink::pack (OutgoingSynchronisationMessage & out) const 
{
  out << _distantSiteId ;
}


void PvmSvmLink::unpack (IncomingSynchronisationMessage & in) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmSvmLink::unpack: ";
#endif
  in >> _distantSiteId ;

  _outgoingBuffer.setDistantSiteId ( _distantSiteId ) ;
#ifdef _DEBUGPVMMESS
  cerr<<"distant site Id: "<<_distantSiteId<<endl;
#endif
}



PvmOutgoingMessage & PvmSvmLink::getOutgoingBuffer () 
{
  return _outgoingBuffer ;
}

---