src/Communication/SALOME_Comm_i.cxx

   1 #include <rpc/xdr.h>
   2 #include "SALOME_Comm_i.hxx"
   3 #include "poa.h"
   4 #include "omnithread.h"
   5 #include "Utils_SINGLETON.hxx"
   6 #include "Utils_ORB_INIT.hxx"
   7 #include "utilities.h"
   8
   9 #include "SenderFactory.hxx"
  10 using namespace std;
  11
  12 CORBA::ORB_var &getGlobalORB(){
  13   ORB_INIT &init = *SINGLETON_<ORB_INIT>::Instance();
  14   CORBA::ORB_var &orb = init(0,0);
  15   return orb;
  16 }
  17
  18 /*! Return the C++ data associated to the array to transmit.
  19   Used when sender and receiver are collocalized.
  20  */
  21 const void *SALOME_Sender_i::getData(long &size) const{
  22   size=_lgrTabToSend;
  23   return _tabToSend;
  24 }
  25
  26 /*! Return the sizeof() of each component of the generic array
  27  */
  28 int SALOME_Sender_i::getSizeOf() const {
  29   return _sizeOf;
  30 }
  31
  32 /*! Unique constructor */
  33 SALOME_Sender_i::SALOME_Sender_i(SALOME::TypeOfDataTransmitted type,const void *tabToSend,long lgrTabToSend,int sizeOf):_tabToSend(tabToSend),_lgrTabToSend(lgrTabToSend),_type(type),_sizeOf(sizeOf){
  34 }
  35
  36 /*! Method to establish if the CORBA object refered by pCorba is collocalised.\n
  37   If it is, the pointer to the servant that incarnates the CORBA object is returned.
  38 */
  39 SALOME_Sender_i *SALOME_Sender_i::find(SALOME::Sender_ptr pCorba){
  40   PortableServer::ServantBase *ret;
  41   try {
  42     ret=PortableServer::POA::_the_root_poa()->reference_to_servant(pCorba);
  43   }
  44   catch(...){
  45     return 0;
  46   }
  47   ret->_remove_ref();
  48   return dynamic_cast<SALOME_Sender_i *>(ret);
  49 }
  50
  51 /*! Method for the remote destroy of the current servant. This method is used by the receiver to destroy the sender when the transfert is complete.
  52  */
  53 void SALOME_Sender_i::release()
  54 {
  55   PortableServer::ObjectId_var oid = _default_POA()->servant_to_id(this);
  56   _default_POA()->deactivate_object(oid);
  57   _remove_ref();
  58 }
  59
  60 /*! Return the type of the element that compose the array. Used by receiverfactory to build the correct receiver.
  61  */
  62 SALOME::TypeOfDataTransmitted SALOME_Sender_i::getTypeOfDataTransmitted()
  63 {
  64   return _type;
  65 }
  66
  67 /*! Return a new sender of the same array but with an another protocol.
  68  */
  69 SALOME::Sender_ptr SALOME_Sender_i::buildOtherWithProtocol(SALOME::TypeOfCommunication type)
  70 {
  71   return SenderFactory::buildSender(type,this);
  72 }
  73
  74 SALOME_CorbaDoubleNCSender_i::SALOME_CorbaDoubleNCSender_i(const double *tabToSend,long lgrTabToSend):SALOME_Sender_i(SALOME::DOUBLE_,tabToSend,lgrTabToSend,sizeof(double)){
  75 }
  76
  77 SALOME_CorbaDoubleNCSender_i::~SALOME_CorbaDoubleNCSender_i(){
  78 }
  79
  80 CORBA::ULong SALOME_CorbaDoubleNCSender_i::getSize(){
  81   CORBA::ULong ret=_lgrTabToSend;
  82   return ret;
  83 }
  84
  85 SALOME::vectorOfDouble* SALOME_CorbaDoubleNCSender_i::sendPart(CORBA::ULong offset, CORBA::ULong length){
  86   SALOME::vectorOfDouble_var c1 = new SALOME::vectorOfDouble(length,length,(CORBA::Double *)((double *)_tabToSend+(long)offset),0);
  87   return c1._retn();
  88 }
  89
  90 SALOME::vectorOfDouble* SALOME_CorbaDoubleNCSender_i::send(){
  91   SALOME::vectorOfDouble_var c1 = new SALOME::vectorOfDouble(_lgrTabToSend,_lgrTabToSend,(CORBA::Double *)_tabToSend,0);
  92   return c1._retn();
  93 }
  94
  95 SALOME_CorbaDoubleCSender_i::SALOME_CorbaDoubleCSender_i(const double *tabToSend,long lgrTabToSend):SALOME_Sender_i(SALOME::DOUBLE_,tabToSend,lgrTabToSend,sizeof(double)){
  96 }
  97
  98 SALOME_CorbaDoubleCSender_i::~SALOME_CorbaDoubleCSender_i(){
  99 }
 100
 101 CORBA::ULong SALOME_CorbaDoubleCSender_i::getSize(){
 102   CORBA::ULong ret=_lgrTabToSend;
 103   return ret;
 104 }
 105
 106 SALOME::vectorOfDouble* SALOME_CorbaDoubleCSender_i::sendPart(CORBA::ULong offset, CORBA::ULong length){
 107   SALOME::vectorOfDouble_var c1 = new SALOME::vectorOfDouble;
 108   c1->length(length);
 109   for (long i=0; i<length; i++)
 110     c1[i] = ((double *)_tabToSend)[i+offset];
 111   return c1._retn();
 112 }
 113
 114 ////////////////////////
 115
 116 SALOME_CorbaLongNCSender_i::SALOME_CorbaLongNCSender_i(const int *tabToSend,long lgrTabToSend):SALOME_Sender_i(SALOME::INT_,tabToSend,lgrTabToSend,sizeof(int)){
 117 }
 118
 119 SALOME_CorbaLongNCSender_i::~SALOME_CorbaLongNCSender_i(){
 120 }
 121
 122 CORBA::ULong SALOME_CorbaLongNCSender_i::getSize(){
 123   CORBA::ULong ret=_lgrTabToSend;
 124   return ret;
 125 }
 126
 127 SALOME::vectorOfLong* SALOME_CorbaLongNCSender_i::sendPart(CORBA::ULong offset, CORBA::ULong length){
 128   SALOME::vectorOfLong_var c1 = new SALOME::vectorOfLong(length,length,(CORBA::Long *)((long *)_tabToSend+(long)offset),0);
 129   return c1._retn();
 130 }
 131
 132 SALOME::vectorOfLong* SALOME_CorbaLongNCSender_i::send(){
 133   SALOME::vectorOfLong_var c1 = new SALOME::vectorOfLong(_lgrTabToSend,_lgrTabToSend,(CORBA::Long *)_tabToSend,0);
 134   return c1._retn();
 135 }
 136
 137 SALOME_CorbaLongCSender_i::SALOME_CorbaLongCSender_i(const int *tabToSend,long lgrTabToSend):SALOME_Sender_i(SALOME::INT_,tabToSend,lgrTabToSend,sizeof(int)){
 138 }
 139
 140 SALOME_CorbaLongCSender_i::~SALOME_CorbaLongCSender_i(){
 141 }
 142
 143 CORBA::ULong SALOME_CorbaLongCSender_i::getSize(){
 144   CORBA::ULong ret=_lgrTabToSend;
 145   return ret;
 146 }
 147
 148 SALOME::vectorOfLong* SALOME_CorbaLongCSender_i::sendPart(CORBA::ULong offset, CORBA::ULong length){
 149   SALOME::vectorOfLong_var c1 = new SALOME::vectorOfLong;
 150   c1->length(length);
 151   for (long i=0; i<length; i++)
 152     c1[i] = ((long *)_tabToSend)[i+offset];
 153   return c1._retn();
 154 }
 155
 156 #ifdef HAVE_MPI2
 157
 158 unsigned long SALOME_MPISender_i::_tag1=0;
 159
 160 unsigned long SALOME_MPISender_i::_tag2=1;
 161
 162 SALOME_MPISender_i::SALOME_MPISender_i(SALOME::TypeOfDataTransmitted type,const void *tabToSend,long lgrTabToSend,int sizeOf):SALOME_Sender_i(type,tabToSend,lgrTabToSend,sizeOf){
 163   _portName=new char[MPI_MAX_PORT_NAME];
 164 }
 165
 166 SALOME_MPISender_i::~SALOME_MPISender_i(){
 167   delete [] _portName;
 168 }
 169
 170 SALOME::MPISender::param* SALOME_MPISender_i::getParam()
 171 {
 172   char stag[12];
 173   int myproc,i=0;
 174
 175   SALOME::MPISender::param_var p = new SALOME::MPISender::param;
 176   MPI_Comm_rank(MPI_COMM_WORLD,&_cproc);
 177   p->myproc = _cproc;
 178   p->tag1 = _tag1;
 179   _tag1Inst=_tag1;
 180   p->tag2 =_tag2;
 181   _tag2Inst=_tag2;
 182   std::string service("toto_");
 183   sprintf(stag,"%d_",_tag1);
 184   service += stag;
 185   sprintf(stag,"%d_",p->tag2);
 186   service += stag;
 187   p->service = CORBA::string_dup(service.c_str());
 188   MPI_Open_port(MPI_INFO_NULL, _portName);
 189   MPI_Errhandler_set(MPI_COMM_WORLD,MPI_ERRORS_RETURN);
 190   while ( i != TIMEOUT  && MPI_Publish_name((char*)service.c_str(),MPI_INFO_NULL,_portName) != MPI_SUCCESS) {
 191     i++;
 192   }
 193   MPI_Errhandler_set(MPI_COMM_WORLD,MPI_ERRORS_ARE_FATAL);
 194   if ( i == TIMEOUT  ) {
 195     MPI_Close_port(_portName);
 196     MPI_Finalize();
 197     exit(-1);
 198     }
 199   _tag1 += 2;
 200   _tag2 += 2;
 201   return p._retn();
 202 }
 203
 204 void SALOME_MPISender_i::send()
 205 {
 206   _argsForThr=new (void *)[8];
 207   _argsForThr[0]=_portName;
 208   _argsForThr[1]=&_lgrTabToSend;
 209   _argsForThr[2]=(void *)_tabToSend;
 210   _argsForThr[3]=&_cproc;
 211   _argsForThr[4]=&_tag1Inst;
 212   _argsForThr[5]=&_tag2Inst;
 213   _argsForThr[6]=&_com;
 214   _argsForThr[7]=&_type;
 215
 216   _newThr=new omni_thread(SALOME_MPISender_i::myThread,_argsForThr);
 217   _newThr->start();
 218 }
 219
 220 void* SALOME_MPISender_i::myThread(void *args)
 221 {
 222   void **argsTab=(void **)args;
 223   long *lgrTabToSend=(long *)argsTab[1];
 224   int *cproc=(int *)argsTab[3];
 225   int *tag1=(int *)argsTab[4];
 226   int *tag2=(int *)argsTab[5];
 227   MPI_Comm *com=(MPI_Comm *)argsTab[6];
 228   SALOME::TypeOfDataTransmitted *type=(SALOME::TypeOfDataTransmitted *)argsTab[7];
 229
 230   MPI_Comm_accept((char *)argsTab[0],MPI_INFO_NULL,0,MPI_COMM_SELF,com);
 231   MPI_Send(lgrTabToSend,1,MPI_LONG,*cproc,*tag1,*com);
 232   switch(*type)
 233     {
 234     case SALOME::DOUBLE_:
 235       MPI_Send(argsTab[2],*lgrTabToSend,MPI_DOUBLE,*cproc,*tag2,*com);
 236       break;
 237     case SALOME::INT_:
 238       MPI_Send(argsTab[2],*lgrTabToSend,MPI_INT,*cproc,*tag2,*com);
 239     }
 240   omni_thread::exit();
 241 }
 242
 243 void SALOME_MPISender_i::close(const SALOME::MPISender::param& p)
 244 {
 245   std::string service(p.service);
 246   const char *st=p.service;
 247   void *r;
 248   _newThr->join(&r);
 249   MPI_Comm_free(&_com);
 250   MPI_Unpublish_name((char *)service.c_str(),MPI_INFO_NULL,_portName);
 251   MPI_Close_port(_portName);
 252   delete [] _argsForThr;
 253 }
 254
 255 #endif
 256
 257 #ifdef HAVE_SOCKET
 258
 259 #include <sys/types.h>
 260 #include <sys/socket.h>
 261 #include <netinet/in.h>
 262 #include <arpa/inet.h>
 263 #include <netdb.h>
 264 #include <unistd.h>
 265
 266 SALOME_SocketSender_i::SALOME_SocketSender_i(SALOME::TypeOfDataTransmitted type,const void *tabToSend,long lgrTabToSend,int sizeOf):SALOME_Sender_i(type,tabToSend,lgrTabToSend,sizeOf){
 267   _IPAddress = inetAddress();
 268   _serverSockfd = -1;
 269   _clientSockfd = -1;
 270 }
 271
 272 SALOME_SocketSender_i::~SALOME_SocketSender_i(){
 273 }
 274
 275 std::string SALOME_SocketSender_i::inetAddress()
 276 {
 277    char s[256];
 278    char t[INET_ADDRSTRLEN+1];
 279    struct hostent *host;
 280    struct in_addr saddr;
 281
 282    gethostname(s, 255);
 283
 284    *t = '\0';
 285
 286    saddr.s_addr = inet_addr(s);
 287    if (saddr.s_addr != -1)
 288       inet_ntop(AF_INET, &saddr, t, INET_ADDRSTRLEN);
 289    else {
 290       host = gethostbyname(s);
 291       if (host != NULL)
 292          inet_ntop(AF_INET, (struct in_addr *) *host->h_addr_list,
 293                    t, INET_ADDRSTRLEN);
 294    }
 295    return std::string(t);
 296 }
 297
 298 SALOME::SocketSender::param * SALOME_SocketSender_i::getParam()
 299 {
 300
 301   SALOME::SocketSender::param_var p = new SALOME::SocketSender::param;
 302
 303   p->lstart = 0;
 304   p->lend = _lgrTabToSend - 1;
 305   p->myport = _port;
 306   p->internet_address = CORBA::string_dup(_IPAddress.c_str());
 307
 308   return p._retn();
 309 }
 310
 311 void SALOME_SocketSender_i::send()
 312 {
 313   _argsForThr=new void *[6];
 314   _argsForThr[0]=&_serverSockfd;
 315   _argsForThr[1]=&_clientSockfd;
 316   _argsForThr[2]=&_lgrTabToSend;
 317   _argsForThr[3]=(void *)_tabToSend;
 318   _argsForThr[4]=&_errorFlag;
 319   _argsForThr[5]=&_type;
 320
 321   _newThr=new omni_thread(SALOME_SocketSender_i::myThread,_argsForThr);
 322   _newThr->start();
 323 }
 324
 325 void* SALOME_SocketSender_i::myThread(void *args)
 326 {
 327   int n=0, m;
 328   void **argsTab=(void **)args;
 329   int *serverSockfd=(int *)argsTab[0];
 330   int *clientSockfd=(int *)argsTab[1];
 331   long *lgrTabToSend=(long *)argsTab[2];
 332   void *tabToSend=argsTab[3];
 333   bool *errorFlag=(bool*)argsTab[4];
 334   SALOME::TypeOfDataTransmitted *type=(SALOME::TypeOfDataTransmitted *)argsTab[5];
 335
 336   XDR xp; /* pointeur sur le decodeur XDR */
 337
 338   switch(*type)
 339     {
 340     case SALOME::DOUBLE_:
 341       xdrmem_create(&xp,(char*)tabToSend,(*lgrTabToSend)*sizeof(double),XDR_ENCODE );
 342       xdr_vector( &xp, (char*)tabToSend, *lgrTabToSend, sizeof(double), (xdrproc_t)xdr_double );
 343
 344       *errorFlag = false;
 345       while( n < *lgrTabToSend*sizeof(double) ){
 346         m = write(*clientSockfd, (char*)tabToSend+n, *lgrTabToSend*sizeof(double)-n);
 347         if( m < 0 ){
 348           if( *clientSockfd >= 0 ){
 349             ::close(*clientSockfd);
 350             *clientSockfd = -1;
 351           }
 352           if( *serverSockfd >= 0 ){
 353             ::close(*serverSockfd);
 354             *serverSockfd = -1;
 355           }
 356           *errorFlag = true;
 357         }
 358         n += m;
 359       }
 360       xdr_destroy( &xp );
 361
 362       xdrmem_create(&xp,(char*)tabToSend,(*lgrTabToSend)*sizeof(double),XDR_DECODE );
 363       xdr_vector( &xp, (char*)tabToSend, *lgrTabToSend, sizeof(double), (xdrproc_t)xdr_double );
 364       xdr_destroy( &xp );
 365       break;
 366     case SALOME::INT_:
 367       xdrmem_create(&xp,(char*)tabToSend,(*lgrTabToSend)*sizeof(int),XDR_ENCODE );
 368       xdr_vector( &xp, (char*)tabToSend, *lgrTabToSend, sizeof(int), (xdrproc_t)xdr_int );
 369
 370       *errorFlag = false;
 371       while( n < *lgrTabToSend*sizeof(int) ){
 372         m = write(*clientSockfd, (char*)tabToSend+n, *lgrTabToSend*sizeof(int)-n);
 373         if( m < 0 ){
 374           if( *clientSockfd >= 0 ){
 375             ::close(*clientSockfd);
 376             *clientSockfd = -1;
 377           }
 378           if( *serverSockfd >= 0 ){
 379             ::close(*serverSockfd);
 380             *serverSockfd = -1;
 381           }
 382           *errorFlag = true;
 383         }
 384         n += m;
 385       }
 386       xdr_destroy( &xp );
 387
 388       xdrmem_create(&xp,(char*)tabToSend,(*lgrTabToSend)*sizeof(int),XDR_DECODE );
 389       xdr_vector( &xp, (char*)tabToSend, *lgrTabToSend, sizeof(int), (xdrproc_t)xdr_int );
 390       xdr_destroy( &xp );
 391     }
 392 }
 393
 394 void SALOME_SocketSender_i::initCom() throw(SALOME::SALOME_Exception)
 395 {
 396   struct sockaddr_in serv_addr;
 397   socklen_t n;
 398   SALOME::ExceptionStruct es;
 399
 400   /* Ouverture de la socket */
 401   _serverSockfd = socket(AF_INET , SOCK_STREAM , 0);
 402   if(_serverSockfd < 0) {
 403     es.type = SALOME::COMM;
 404     es.text = "error Socket exception";
 405     throw SALOME::SALOME_Exception(es);
 406   }
 407   /* Socket structure initialisation*/
 408   bzero((char*)&serv_addr,sizeof(serv_addr));
 409   serv_addr.sin_family = AF_INET;
 410   serv_addr.sin_port = 0; /* asking for a free port */
 411   serv_addr.sin_addr.s_addr = INADDR_ANY;
 412
 413   /* Association of socket with a port */
 414   if( ::bind(_serverSockfd, (struct sockaddr *) & serv_addr,
 415            sizeof(struct sockaddr)) < 0 ) {
 416     closeCom();
 417     es.type = SALOME::COMM;
 418     es.text = "error bind Socket exception";
 419     throw SALOME::SALOME_Exception(es);
 420   }
 421   /* Listening to the allocated port */
 422   if( listen(_serverSockfd, 10) < 0 ) {
 423     closeCom();
 424     es.type = SALOME::COMM;
 425     es.text = "error listen Socket exception";
 426     throw SALOME::SALOME_Exception(es);
 427   }
 428   /* Retrieving port number*/
 429   if( getsockname(_serverSockfd, (struct sockaddr *) & serv_addr, &n) < 0 ){
 430     closeCom();
 431     es.type = SALOME::COMM;
 432     es.text = "error getName Socket exception";
 433     throw SALOME::SALOME_Exception(es);
 434   }
 435   _port = htons(serv_addr.sin_port);
 436   SCRUTE(_port);
 437 }
 438
 439 void SALOME_SocketSender_i::acceptCom() throw(SALOME::SALOME_Exception)
 440 {
 441   socklen_t sin_size;
 442   int new_fd;
 443   struct sockaddr_in client_addr;
 444   SALOME::ExceptionStruct es;
 445
 446   sin_size = sizeof(struct sockaddr_in);
 447
 448   _clientSockfd = accept(_serverSockfd, (struct sockaddr *)&client_addr, &sin_size);
 449   if( _clientSockfd < 0 ){
 450     closeCom();
 451     es.type = SALOME::COMM;
 452     es.text = "error accept Socket exception";
 453     throw SALOME::SALOME_Exception(es);
 454   }
 455 }
 456
 457 void SALOME_SocketSender_i::closeCom()
 458 {
 459   if( _clientSockfd >= 0 ){
 460     ::close(_clientSockfd);
 461     _clientSockfd = -1;
 462   }
 463   if( _serverSockfd >= 0 ){
 464     ::close(_serverSockfd);
 465     _serverSockfd = -1;
 466   }
 467
 468 }
 469
 470 void SALOME_SocketSender_i::endOfCom()
 471 {
 472   void *r;
 473   _newThr->join(&r);
 474   if(_errorFlag)
 475     {
 476       SALOME::ExceptionStruct es;
 477       es.type = SALOME::COMM;
 478       es.text = "error write Socket exception";
 479       throw SALOME::SALOME_Exception(es);
 480     }
 481   delete [] _argsForThr;
 482 }
 483
 484 #endif