IntegerMessage.cpp

Go to the documentation of this file.

#include "IntegerMessage.h"
 
 
 
#include <sstream>
#include <algorithm>
 
 
 
tarch::mpi::IntegerMessage::IntegerMessage(int  __value){
setValue( __value);
}
 
 
 
std::string tarch::mpi::IntegerMessage::toString() const {
  std::ostringstream out;
  out << "(";
  out << "value=" << _value;
  out << ")";
  return out.str();
}
 
 
 
 
 
int   tarch::mpi::IntegerMessage::getValue() const {
  return _value;
}
 
 
void   tarch::mpi::IntegerMessage::setValue(int value) {
  _value = value;
}
 
 
 
 
 
 
#ifdef Parallel
 
#if !defined(__MPI_ATTRIBUTES_LANGUAGE_EXTENSION__)
MPI_Datatype tarch::mpi::IntegerMessage::Datatype = MPI_DATATYPE_NULL;
#endif
 
 
[[clang::map_mpi_datatype]]
MPI_Datatype tarch::mpi::IntegerMessage::getForkDatatype() {
  return Datatype;
}
 
 
[[clang::map_mpi_datatype]]
MPI_Datatype tarch::mpi::IntegerMessage::getGlobalCommunciationDatatype() {
  return Datatype;
}
 
 
[[clang::map_mpi_datatype]]
MPI_Datatype tarch::mpi::IntegerMessage::getJoinDatatype() {
  return Datatype;
}
 
 
[[clang::map_mpi_datatype]]
MPI_Datatype tarch::mpi::IntegerMessage::getBoundaryExchangeDatatype() {
  return Datatype;
}
 
 
[[clang::map_mpi_datatype]]
MPI_Datatype tarch::mpi::IntegerMessage::getMultiscaleDataExchangeDatatype() {
  return Datatype;
}
 
 
[[clang::map_mpi_datatype]]
void tarch::mpi::IntegerMessage::freeForkDatatype() {
  if (Datatype != MPI_DATATYPE_NULL){
    MPI_Type_free(&Datatype);
    Datatype = MPI_DATATYPE_NULL;
  }
}
 
 
[[clang::map_mpi_datatype]]
void tarch::mpi::IntegerMessage::freeGlobalCommunciationDatatype() {
  if (Datatype != MPI_DATATYPE_NULL){
    MPI_Type_free(&Datatype);
    Datatype = MPI_DATATYPE_NULL;
  }
}
 
 
[[clang::map_mpi_datatype]]
void tarch::mpi::IntegerMessage::freeJoinDatatype() {
  if (Datatype != MPI_DATATYPE_NULL){
    MPI_Type_free(&Datatype);
    Datatype = MPI_DATATYPE_NULL;
  }
}
 
 
[[clang::map_mpi_datatype]]
void tarch::mpi::IntegerMessage::freeBoundaryExchangeDatatype() {
  if (Datatype != MPI_DATATYPE_NULL){
    MPI_Type_free(&Datatype);
    Datatype = MPI_DATATYPE_NULL;
  }
}
 
 
[[clang::map_mpi_datatype]]
void tarch::mpi::IntegerMessage::freeMultiscaleDataExchangeDatatype() {
  if (Datatype != MPI_DATATYPE_NULL){
    MPI_Type_free(&Datatype);
    Datatype = MPI_DATATYPE_NULL;
  }
}
 
 
int tarch::mpi::IntegerMessage::getSenderRank() const {
  return _senderDestinationRank;
}
 
 
void tarch::mpi::IntegerMessage::initDatatype() {
  #if !defined(__MPI_ATTRIBUTES_LANGUAGE_EXTENSION__)
  tarch::mpi::IntegerMessage  instances[2];
 
  int NumberOfAttributes = 0;
  NumberOfAttributes++;
 
  MPI_Datatype* subtypes = new MPI_Datatype[NumberOfAttributes];
  int*          blocklen = new int[NumberOfAttributes];
  MPI_Aint*     disp     = new MPI_Aint[NumberOfAttributes];
 
  int counter            = 0;
  subtypes[counter] = MPI_INT;
  blocklen[counter] = 1;
  counter++;
 
  MPI_Aint  baseFirstInstance;
  MPI_Aint  baseSecondInstance;
  MPI_Get_address( &instances[0], &baseFirstInstance );
  MPI_Get_address( &instances[1], &baseSecondInstance );
 
  counter = 0;
  MPI_Get_address( &(instances[0]._value), &disp[counter] );
  counter++;
 
  MPI_Aint offset = disp[0] - baseFirstInstance;
  MPI_Aint extent = baseSecondInstance - baseFirstInstance - offset;
  for (int i=NumberOfAttributes-1; i>=0; i--) {
    disp[i] = disp[i] - disp[0];
  }
 
  int errorCode = 0;
  MPI_Datatype tmpType;
  errorCode += MPI_Type_create_struct( NumberOfAttributes, blocklen, disp, subtypes, &tmpType );
  errorCode += MPI_Type_create_resized( tmpType, offset, extent, &Datatype );
  errorCode += MPI_Type_commit( &Datatype );
  errorCode += MPI_Type_free( &tmpType );
  if (errorCode) std::cerr << "error constructing MPI datatype in " << __FILE__ << ":" << __LINE__ << std::endl;
 
  delete[] subtypes;
  delete[] blocklen;
  delete[] disp;
 
  #else
  // invoke routine once to trigger lazy initialisation
  getForkDatatype();
  getJoinDatatype();
  getBoundaryExchangeDatatype();
  getMultiscaleDataExchangeDatatype();
  getGlobalCommunciationDatatype();
  #endif
}
 
 
void tarch::mpi::IntegerMessage::shutdownDatatype() {
  #if !defined(__MPI_ATTRIBUTES_LANGUAGE_EXTENSION__)
  freeForkDatatype();
  freeJoinDatatype();
  freeBoundaryExchangeDatatype();
  freeMultiscaleDataExchangeDatatype();
  freeGlobalCommunciationDatatype();
  #else
  MPI_Datatype type = Datatype;
  MPI_Type_free( &type );
  #endif
}
 
 
void tarch::mpi::IntegerMessage::send(const tarch::mpi::IntegerMessage& buffer, int destination, int tag, MPI_Comm communicator ) {
  MPI_Send( &buffer, 1, Datatype, destination, tag, communicator);
}
 
 
void tarch::mpi::IntegerMessage::receive(tarch::mpi::IntegerMessage& buffer, int source, int tag, MPI_Comm communicator ) {
  MPI_Status status;
  MPI_Recv( &buffer, 1, Datatype, source, tag, communicator, &status);
  buffer._senderDestinationRank = status.MPI_SOURCE;
}
 
 
void tarch::mpi::IntegerMessage::send(
  const tarch::mpi::IntegerMessage& buffer,
  int destination,
  int tag,
  std::function<void()> startCommunicationFunctor,
  std::function<void()> waitFunctor,
  MPI_Comm communicator
) {
  MPI_Request sendRequestHandle;
  int         flag = 0;
  MPI_Isend( &buffer, 1, Datatype, destination, tag, communicator, &sendRequestHandle );
  MPI_Test( &sendRequestHandle, &flag, MPI_STATUS_IGNORE );
  startCommunicationFunctor();
  while (!flag) {
    waitFunctor();
    MPI_Test( &sendRequestHandle, &flag, MPI_STATUS_IGNORE );
  }
}
 
 
void tarch::mpi::IntegerMessage::receive(
  tarch::mpi::IntegerMessage& buffer,
  int source,
  int tag,
  std::function<void()> startCommunicationFunctor,
  std::function<void()> waitFunctor,
  MPI_Comm communicator
) {
  MPI_Status  status;
  MPI_Request receiveRequestHandle;
  int         flag = 0;
  MPI_Irecv( &buffer, 1, Datatype, source, tag, communicator, &receiveRequestHandle );
  MPI_Test( &receiveRequestHandle, &flag, &status );
  startCommunicationFunctor();
  while (!flag) {
    waitFunctor();
    MPI_Test( &receiveRequestHandle, &flag, &status );
  }
  buffer._senderDestinationRank = status.MPI_SOURCE;
}
#endif
 
#ifdef Parallel
void tarch::mpi::IntegerMessage::sendAndPollDanglingMessages(const tarch::mpi::IntegerMessage& message, int destination, int tag, MPI_Comm communicator ) {
  tarch::mpi::IntegerMessage::send(
    message, destination, tag,
    [&]() {
      tarch::mpi::Rank::getInstance().setDeadlockWarningTimeStamp();
      tarch::mpi::Rank::getInstance().setDeadlockTimeOutTimeStamp();
    },
    [&]() {
      tarch::mpi::Rank::getInstance().writeTimeOutWarning( "tarch::mpi::IntegerMessage", "sendAndPollDanglingMessages()",destination, tag );
      tarch::mpi::Rank::getInstance().triggerDeadlockTimeOut( "tarch::mpi::IntegerMessage", "sendAndPollDanglingMessages()", destination, tag );
      tarch::services::ServiceRepository::getInstance().receiveDanglingMessages();
    },
    communicator
  );
}
 
 
void tarch::mpi::IntegerMessage::receiveAndPollDanglingMessages(tarch::mpi::IntegerMessage& message, int source, int tag, MPI_Comm communicator ) {
  tarch::mpi::IntegerMessage::receive(
    message, source, tag,
    [&]() {
      tarch::mpi::Rank::getInstance().setDeadlockWarningTimeStamp();
      tarch::mpi::Rank::getInstance().setDeadlockTimeOutTimeStamp();
    },
    [&]() {
      tarch::mpi::Rank::getInstance().writeTimeOutWarning( "tarch::mpi::IntegerMessage", "receiveAndPollDanglingMessages()", source, tag );
      tarch::mpi::Rank::getInstance().triggerDeadlockTimeOut( "tarch::mpi::IntegerMessage", "receiveAndPollDanglingMessages()", source, tag );
      tarch::services::ServiceRepository::getInstance().receiveDanglingMessages();
    },
    communicator
  );
}
#endif