src/MPIContainer/MPIObject_i.cxx

   1 // Copyright (C) 2007-2020  CEA/DEN, EDF R&D, OPEN CASCADE
   2 //
   3 // Copyright (C) 2003-2007  OPEN CASCADE, EADS/CCR, LIP6, CEA/DEN,
   4 // CEDRAT, EDF R&D, LEG, PRINCIPIA R&D, BUREAU VERITAS
   5 //
   6 // This library is free software; you can redistribute it and/or
   7 // modify it under the terms of the GNU Lesser General Public
   8 // License as published by the Free Software Foundation; either
   9 // version 2.1 of the License, or (at your option) any later version.
  10 //
  11 // This library is distributed in the hope that it will be useful,
  12 // but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14 // Lesser General Public License for more details.
  15 //
  16 // You should have received a copy of the GNU Lesser General Public
  17 // License along with this library; if not, write to the Free Software
  18 // Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  19 //
  20 // See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
  21 //
  22
  23 #include "MPIObject_i.hxx"
  24 #include "utilities.h"
  25 #include "Utils_SALOME_Exception.hxx"
  26
  27 #define TIMEOUT 5
  28
  29 #ifndef WIN32
  30 #include <unistd.h>
  31 #endif
  32
  33 MPIObject_i::MPIObject_i()
  34 {
  35   MPI_Comm_size( MPI_COMM_WORLD, &_nbproc );
  36   MPI_Comm_rank( MPI_COMM_WORLD, &_numproc );
  37   _tior=NULL;
  38 }
  39
  40 MPIObject_i::~MPIObject_i()
  41 {
  42   if(_tior) delete _tior;
  43 }
  44
  45 Engines::IORTab* MPIObject_i::tior()
  46 {
  47   Engines::IORTab_var tior = new Engines::IORTab;
  48   tior->length(_tior->length());
  49   for(unsigned int ip=0;ip<tior->length();ip++)
  50     tior[ip] = (*_tior)[ip];
  51   return tior._retn();
  52 };
  53
  54 void MPIObject_i::tior(const Engines::IORTab& ior)
  55 {
  56   _tior = new Engines::IORTab;
  57   _tior->length(ior.length());
  58   for(unsigned int ip=0;ip<ior.length();ip++)
  59     (*_tior)[ip] = ior[ip];
  60 }
  61
  62 void MPIObject_i::BCastIOR(CORBA::ORB_ptr orb, Engines::MPIObject_ptr pobj, bool amiCont)
  63 {
  64   int err, ip, n;
  65   char *ior;
  66   MPI_Status status; /* status de reception de message MPI */
  67   std::ostringstream msg;
  68
  69   if( _numproc == 0 )
  70     {
  71
  72       //Allocation du tableau des IOR
  73       Engines::IORTab_var iort = new Engines::IORTab;
  74       iort->length(_nbproc);
  75
  76       iort[0] = pobj;
  77
  78       // Process 0 recupere les ior de l'object sur les autres process
  79       for(ip=1;ip<_nbproc;ip++)
  80         {
  81           err = MPI_Recv(&n,1,MPI_INT,ip,ip,MPI_COMM_WORLD,&status);
  82           if(err)
  83             {
  84               msg << "[" << _numproc << "] MPI_RECV error";
  85               throw SALOME_Exception(msg.str().c_str());
  86             }
  87           // Allocation de la chaine de longueur n
  88           ior = new char[n];
  89           err = MPI_Recv(ior,n,MPI_CHAR,ip,2*ip,MPI_COMM_WORLD,&status);
  90           if(err)
  91             {
  92               msg << "[" << _numproc << "] MPI_RECV error";
  93               throw SALOME_Exception(msg.str().c_str());
  94             }
  95           iort[ip] = orb->string_to_object(ior);
  96           delete [] ior;
  97           if(CORBA::is_nil(iort[ip]))
  98             {
  99               msg << "[" << ip << "] MPI Component not loaded";
 100               throw SALOME_Exception(msg.str().c_str());
 101             }
 102         }
 103       // On donne le tableau des ior a l'objet Corba du process 0
 104       if( amiCont )
 105         tior(*(iort._retn()));
 106       else
 107         pobj->tior(*(iort._retn()));
 108     }
 109   else
 110     {
 111       // Conversion IOR vers string
 112       ior = orb->object_to_string(pobj);
 113       n = strlen(ior) + 1;
 114       // On envoie l'IOR au process 0
 115       err = MPI_Send(&n,1,MPI_INT,0,_numproc,MPI_COMM_WORLD);
 116       if(err)
 117         {
 118           msg << "[" << _numproc << "] MPI_SEND error";
 119           throw SALOME_Exception(msg.str().c_str());
 120         }
 121       err = MPI_Send(ior,n,MPI_CHAR,0,2*_numproc,MPI_COMM_WORLD);
 122       if(err)
 123         {
 124           msg << "[" << _numproc << "] MPI_SEND error";
 125           throw SALOME_Exception(msg.str().c_str());
 126         }
 127       CORBA::string_free(ior);
 128     }
 129
 130 }
 131
 132 #ifdef HAVE_MPI2
 133 void MPIObject_i::remoteMPI2Connect(std::string service)
 134 {
 135   int i;
 136   char port_name[MPI_MAX_PORT_NAME];
 137   char port_name_clt[MPI_MAX_PORT_NAME];
 138   MPI_Info info;
 139   std::ostringstream msg;
 140
 141   if( service.size() == 0 )
 142     {
 143       msg << "[" << _numproc << "] You have to give a service name !";
 144       throw SALOME_Exception(msg.str().c_str());
 145     }
 146
 147   if( _srv.find(service) != _srv.end() )
 148     {
 149       msg << "[" << _numproc << "] service " << service << " already exist !";
 150       throw SALOME_Exception(msg.str().c_str());
 151     }
 152
 153   _srv[service] = false;
 154
 155   MPI_Barrier(MPI_COMM_WORLD);
 156 #if OMPI_MAJOR_VERSION >= 4
 157   MPI_Comm_set_errhandler(MPI_COMM_WORLD, MPI_ERRORS_RETURN);
 158 #else
 159   MPI_Errhandler_set(MPI_COMM_WORLD, MPI_ERRORS_RETURN);
 160 #endif
 161   MPI_Info_create(&info);
 162   MPI_Info_set(info, "ompi_unique", "true");
 163   if( _numproc == 0 )
 164     {
 165       /* rank 0 try to be a server. If service is already published, try to be a cient */
 166       MPI_Open_port(MPI_INFO_NULL, port_name);
 167       if ( MPI_Lookup_name((char*)service.c_str(), MPI_INFO_NULL, port_name_clt) == MPI_SUCCESS )
 168         {
 169           MESSAGE("[" << _numproc << "] I get the connection with " << service << " at " << port_name_clt << std::endl);
 170           MPI_Close_port( port_name );
 171         }
 172       else if ( MPI_Publish_name((char*)service.c_str(), info, port_name) == MPI_SUCCESS )
 173         {
 174           _srv[service] = true;
 175           _port_name[service] = port_name;
 176           MESSAGE("[" << _numproc << "] service " << service << " available at " << port_name << std::endl);
 177         }
 178       else if ( MPI_Lookup_name((char*)service.c_str(), MPI_INFO_NULL, port_name_clt) == MPI_SUCCESS )
 179         {
 180           MESSAGE("[" << _numproc << "] I get the connection with " << service << " at " << port_name_clt << std::endl);
 181           MPI_Close_port( port_name );
 182         }
 183       else
 184         {
 185           msg << "[" << _numproc << "] Error on connection with " << service << " at " << port_name_clt;
 186           throw SALOME_Exception(msg.str().c_str());
 187         }
 188     }
 189   else
 190     {
 191       i=0;
 192       /* Waiting rank 0 publish name and try to be a client */
 193       while ( i != TIMEOUT  )
 194         {
 195           sleep(1);
 196           if ( MPI_Lookup_name((char*)service.c_str(), MPI_INFO_NULL, port_name_clt) == MPI_SUCCESS )
 197             {
 198               MESSAGE("[" << _numproc << "] I get the connection with " << service << " at " << port_name_clt << std::endl);
 199               break;
 200             }
 201           i++;
 202         }
 203       if(i==TIMEOUT)
 204         {
 205           msg << "[" << _numproc << "] Error on connection with " << service << " at " << port_name_clt;
 206           throw SALOME_Exception(msg.str().c_str());
 207         }
 208     }
 209 #if OMPI_MAJOR_VERSION >= 4
 210   MPI_Comm_set_errhandler(MPI_COMM_WORLD, MPI_ERRORS_ARE_FATAL);
 211 #else
 212   MPI_Errhandler_set(MPI_COMM_WORLD, MPI_ERRORS_ARE_FATAL);
 213 #endif
 214
 215   /* If rank 0 is server, all processes call MPI_Comm_accept */
 216   /* If rank 0 is not server, all processes call MPI_Comm_connect */
 217   int srv = (int)_srv[service];
 218   MPI_Bcast(&srv,1,MPI_INT,0,MPI_COMM_WORLD);
 219   _srv[service] = (bool)srv;
 220   if ( _srv[service] )
 221     MPI_Comm_accept( port_name, MPI_INFO_NULL, 0, MPI_COMM_WORLD, &(_icom[service]) );
 222   else
 223     MPI_Comm_connect(port_name_clt, MPI_INFO_NULL, 0, MPI_COMM_WORLD, &(_icom[service]) );
 224
 225   /* create global communicator: servers have low index in global communicator*/
 226   MPI_Intercomm_merge(_icom[service],!_srv[service],&(_gcom[service]));
 227
 228   /* only rank 0 can be server for unpublish name */
 229   if(_numproc != 0) _srv[service] = false;
 230
 231 }
 232
 233 void MPIObject_i::remoteMPI2Disconnect(std::string service)
 234 {
 235   std::ostringstream msg;
 236
 237   if( service.size() == 0 )
 238     {
 239       msg << "[" << _numproc << "] You have to give a service name !";
 240       throw SALOME_Exception(msg.str().c_str());
 241     }
 242
 243   if( _srv.find(service) == _srv.end() )
 244     {
 245       msg << "[" << _numproc << "] service " << service << " don't exist !";
 246       throw SALOME_Exception(msg.str().c_str());
 247     }
 248
 249   MPI_Comm_disconnect( &(_gcom[service]) );
 250   if ( _srv[service] )
 251     {
 252
 253       char port_name[MPI_MAX_PORT_NAME];
 254       strcpy(port_name,_port_name[service].c_str());
 255
 256       MPI_Unpublish_name((char*)service.c_str(), MPI_INFO_NULL, port_name);
 257       MESSAGE("[" << _numproc << "] " << service << ": close port " << _port_name[service] << std::endl);
 258       MPI_Close_port( port_name );
 259       _port_name.erase(service);
 260     }
 261
 262   _gcom.erase(service);
 263   _icom.erase(service);
 264   _srv.erase(service);
 265
 266 }
 267 #endif
 268