OMKPvmOutgoingMessage.cxx

Go to the documentation of this file.

#include "OMKPvmOutgoingMessage.h"

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

const Date & PvmOutgoingMessage::getMessageDate () const
{
  return _timestamp ;
}

void PvmOutgoingMessage::pack (bool val) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack (bool val) "<<val << " Buffer " << _currentBuffer <<endl;
#endif
  //assert ( _timeStamped ) ;
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  char realVal = val ;
  int info = pvm_pkbyte (&realVal, 1, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}


void PvmOutgoingMessage::pack (char val) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack (char val) "<<val << " Buffer " << _currentBuffer <<endl;
#endif
  //assert ( _timeStamped ) ;
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkbyte (&val, 1, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}

void PvmOutgoingMessage::pack (long val) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack (long val) "<<val << " Buffer " << _currentBuffer <<endl;

#endif
  //assert ( _timeStamped ) ;
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pklong (&val, 1, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}


void PvmOutgoingMessage::pack (unsigned long val) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack (unsigned long val) "<<val<< " Buffer " << _currentBuffer << endl;
#endif
  //assert ( _timeStamped ) ;
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkulong (&val, 1, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}

void PvmOutgoingMessage::pack (int val) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack (int val) "<<val<< " Buffer " << _currentBuffer << endl;
#endif
  //assert ( _timeStamped ) ;
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkint (&val, 1, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}


void PvmOutgoingMessage::pack (unsigned int val) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack (unsigned int val) "<<val<< " Buffer " << _currentBuffer << endl;
#endif
  //assert ( _timeStamped ) ;
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkuint (&val, 1, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}

void PvmOutgoingMessage::pack (float val) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack (float val) "<<val<< " Buffer " << _currentBuffer << endl;
#endif
  //assert ( _timeStamped ) ;
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkfloat (&val, 1, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}


void PvmOutgoingMessage::pack (double val) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack (double val) "<<val<< " Buffer " << _currentBuffer << endl;
#endif
  //assert ( _timeStamped ) ;
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkdouble (&val, 1, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}



void PvmOutgoingMessage::pack (short val) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack (short val) "<<val<< " Buffer " << _currentBuffer << endl;
#endif
  //assert ( _timeStamped ) ;
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkshort (&val, 1, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}



void PvmOutgoingMessage::pack (unsigned short val) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack (unsigned short val) "<<val<< " Buffer " << _currentBuffer << endl;
#endif
  //assert ( _timeStamped ) ;
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkushort (&val, 1, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}



void PvmOutgoingMessage::pack (char * val) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack (char * val) "<<val<< " Buffer " << _currentBuffer << endl;
#endif
  //assert ( _timeStamped ) ;
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkstr ( val ) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}

void PvmOutgoingMessage::pack (const char * val) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack (char * val) "<<val<< " Buffer " << _currentBuffer << endl;
#endif
  //assert ( _timeStamped ) ;
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkstr ( const_cast<char *>( val ) ) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}

void PvmOutgoingMessage::pack (const std::string & val) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack (const std::string & val) "<<val<< " Buffer " << _currentBuffer << endl;
#endif
  //assert ( _timeStamped ) ;
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  //the string is packed, with the length needed for unpacking
  unsigned int strSize =   val.size() + 1 ;
  int info = pvm_pkuint (  &strSize, 1, 1 ) ;

  checkForPackErrors ( info ) ;

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

  pvm_setsbuf ( oldbuf ) ;
}


PvmOutgoingMessage::PvmOutgoingMessage ( ) : 
PvmMessage(0),
_timeStamped ( false )
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":PvmOutgoingMessage ()"<<endl;
#endif

  _currentBuffer = pvm_mkbuf ( PvmSvm::pvmDataEncoding ) ;

#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":PvmOutgoingMessage (): new send buffer"<<_currentBuffer<<endl;
#endif
}


PvmOutgoingMessage::~PvmOutgoingMessage ()
{
  pvm_freebuf ( _currentBuffer ) ;
}



void PvmOutgoingMessage::reinitAndRevertPvmContext(int oldbuf)
{
  assert (oldbuf != 0 ) ;
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":reinitAndRevertPvmContext () send buffer "<<_currentBuffer<<endl;
#endif

  // As send has been done, reinitialise everything
  OutgoingSynchronisationMessage::reinit() ;

  _timeStamped = false ;

  pvm_freebuf ( _currentBuffer ) ;

  _currentBuffer = pvm_mkbuf ( PvmSvm::pvmDataEncoding ) ;

  // remplaced by the prepeceeding lines, because init_send changes the current buffer
  // _currentBuffer = pvm_initsend ( PvmSvm::pvmDataEncoding ) ; changes _currentBuffer )

  int info = pvm_setsbuf ( oldbuf ) ;

  checkForChangedBuffer ( info, __LINE__, __FILE__ ) ;
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":new send buffer "<<_currentBuffer<<endl;
#endif
}

bool PvmOutgoingMessage::isTimeStamped() const 
{
  return _timeStamped ;
}


void PvmOutgoingMessage::insertTimeStamp ( const Date & date ) 
{

#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":insertTimeStamp ("<<date<<") "<<_currentBuffer
    <<" of size "<<getSize()
    <<endl;
#endif

  if ( !_timeStamped )
  {
    _timeStamped = true ;

    *this<<date;

    _timestamp = date ;

    assert ( getSize() == 4 ) ;

  }
  else
  {
    if ( date != _timestamp ) {
      //cerr<<"WARNING : timestamping at"<<date<<" message with timestamp "<<_timestamp<<endl;
      //assert ( date == _timestamp ) ;
    }
  }
}

void PvmOutgoingMessage::checkForChangedBuffer (int info, int lineNumber, char * file)
{
  if ( info < 0 )
  {
    cerr<<"PvmOutgoingMessage::checkForChangedBuffer line "<<lineNumber<<" of "<<file<<" ";
    switch ( info ) 
    {
    case PvmSysErr:
      cerr<<"pvmd not responding";
      break;
    case PvmBadParam:
      cerr<<"giving an invalid bufid.";
      break;
    case PvmNoSuchBuf:
      cerr<<"Switching to a nonexistent message buffer.";
      break;
    default:
      cerr<<"unexpected error: ";
    }
    cerr<<endl;   
    OMASSERT( false ) ;
  } 
}


void PvmOutgoingMessage::checkForSendErrors (int info)
{
  if ( info < 0 )
  {
    cerr<<"PvmOutgoingMessage::checkForSendErrors ERROR";
    switch ( info ) 
    {
    case PvmSysErr:
      cerr<<"pvmd not responding";
      break;
    case PvmBadParam:
      cerr<<"giving an invalid tid or a msgtag.";
      break;
    case PvmNoBuf:
      cerr<<"There is no active send buffer to pack into. Try calling pvm_initsend before sending.";
      break;
    default:
      cerr<<"unexpected error: ";
    }
    cerr<<endl;   
    OMASSERT( false ) ;
  } 
}

void PvmOutgoingMessage::checkForPackErrors (int info)
{
  if (info < 0 )
  {
    cerr<<"PvmOutgoingMessage::checkForPackErrors ERROR";
    switch ( info ) 
    {
    case PvmNoMem:
      cerr<<"Malloc has failed. Message buffer size has exceeded the available memory on this host.";
      break;
    case PvmNoBuf:
      cerr<<"There is no active send buffer to pack into. Try calling pvm_initsend before packing message.";
      break;
    default:
      cerr<<"unexpected error: ";
    }
    cerr<<endl;   
    OMASSERT( false ) ;
  }
}


void PvmOutgoingMessage::pack ( const long * val, int cnt )
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack ( const long * val, int cnt) "<<val<<endl;
#endif
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pklong ( const_cast<long *>(val), cnt, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}

void PvmOutgoingMessage::pack ( const unsigned long * val, int cnt )
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack ( const unsigned long * val, int cnt) "<<val<<endl;
#endif
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkulong ( const_cast<unsigned long *>(val), cnt, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}

void PvmOutgoingMessage::pack ( const int * val, int cnt )
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack ( const int * val, int cnt) "<<val<<endl;
#endif
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkint ( const_cast<int *>(val), cnt, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}

void PvmOutgoingMessage::pack ( const unsigned int * val, int cnt )
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack ( const unsigned int * val, int cnt) "<<val<<endl;
#endif
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkuint ( const_cast<unsigned int *>(val), cnt, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}

void PvmOutgoingMessage::pack ( const float * val, int cnt ) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack ( const float * val, int cnt) "<<val<<endl;
#endif
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkfloat ( const_cast<float *>(val), cnt, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}

void PvmOutgoingMessage::pack ( const double * val, int cnt )  
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack ( const double * val, int cnt) "<<val<<endl;
#endif
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkdouble ( const_cast<double *>(val), cnt, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}


void PvmOutgoingMessage::pack ( const short * val, int cnt )
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack ( const short * val, int cnt) "<<val<<endl;
#endif
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkshort ( const_cast< short *>(val), cnt, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}

void PvmOutgoingMessage::pack ( const unsigned short * val, int cnt ) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack (const unsigned short * val, int cnt) "<<val<<endl;
#endif
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkushort ( const_cast<unsigned short *>(val), cnt, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}



void PvmOutgoingMessage::pack ( long * val, int cnt )
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack ( long * val, int cnt) "<<val<<endl;
#endif
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pklong ( val, cnt, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}

void PvmOutgoingMessage::pack ( unsigned long * val, int cnt ) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack ( unsigned long * val, int cnt) "<<val<<endl;
#endif
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkulong ( val, cnt, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}

void PvmOutgoingMessage::pack ( int * val, int cnt )
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack ( int * val, int cnt) "<<val<<endl;
#endif
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkint ( val, cnt, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}

void PvmOutgoingMessage::pack ( unsigned int * val, int cnt )
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack ( unsigned int * val, int cnt) "<<val<<endl;
#endif
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkuint ( val, cnt, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}

void PvmOutgoingMessage::pack ( float * val, int cnt ) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack ( float * val, int cnt) "<<val<<endl;
#endif
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkfloat ( val, cnt, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}

void PvmOutgoingMessage::pack ( double * val, int cnt )  
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack ( double * val, int cnt) "<<val<<endl;
#endif
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkdouble ( val, cnt, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}


void PvmOutgoingMessage::pack ( short * val, int cnt )
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack ( short * val, int cnt) "<<val<<endl;
#endif
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkshort ( val, cnt, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}

void PvmOutgoingMessage::pack ( unsigned short * val, int cnt ) 
{
#ifdef _DEBUGPVMMESS
  cerr<<"PvmOutgoingMessage:"<<(void *)this<<":pack (unsigned short * val, int cnt) "<<val<<endl;
#endif
  int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  int info = pvm_pkushort ( val, cnt, 1) ;
  checkForPackErrors ( info ) ;
  pvm_setsbuf ( oldbuf ) ;
}

void PvmOutgoingMessage::flushCurrentBuffer()
{
  // TDTD remplacement par un simple reinit
  //int oldbuf = pvm_setsbuf ( _currentBuffer ) ;
  //reinitAndRevertPvmContext ( oldbuf ) ;
  reinit () ;
}

---