1 // Copyright (C) 2007-2016 CEA/DEN, EDF R&D, OPEN CASCADE
3 // Copyright (C) 2003-2007 OPEN CASCADE, EADS/CCR, LIP6, CEA/DEN,
4 // CEDRAT, EDF R&D, LEG, PRINCIPIA R&D, BUREAU VERITAS
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.
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.
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
20 // See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
23 #include "omniORB4/poa.h"
24 #include "utilities.h"
26 #define TAILLE_SPLIT 100000
29 template<class T,class TCorba,class TSeqCorba,class CorbaSender,class servForT,class ptrForT>
30 CorbaNCNoCopyReceiver<T,TCorba,TSeqCorba,CorbaSender,servForT,ptrForT>::CorbaNCNoCopyReceiver(CorbaSender mySender):_mySender(mySender){
33 template<class T,class TCorba,class TSeqCorba,class CorbaSender,class servForT,class ptrForT>
34 CorbaNCNoCopyReceiver<T,TCorba,TSeqCorba,CorbaSender,servForT,ptrForT>::~CorbaNCNoCopyReceiver(){
38 template<class T,class TCorba,class TSeqCorba,class CorbaSender,class servForT,class ptrForT>
39 T *CorbaNCNoCopyReceiver<T,TCorba,TSeqCorba,CorbaSender,servForT,ptrForT>::getDistValue(long &size)
41 TSeqCorba seq=_mySender->send();
43 return (T *)seq->get_buffer(1);
46 template<class T,class TCorba,class TSeqCorba,class CorbaSender,class servForT,class ptrForT>
47 T *CorbaNCNoCopyReceiver<T,TCorba,TSeqCorba,CorbaSender,servForT,ptrForT>::getValue(long &size)
49 return Receiver<T,servForT,ptrForT>::getValue(size,_mySender);
52 template<class T,class TCorba,class TSeqCorba,class CorbaSender,class servForT,class ptrForT>
53 CorbaNCWithCopyReceiver<T,TCorba,TSeqCorba,CorbaSender,servForT,ptrForT>::CorbaNCWithCopyReceiver(CorbaSender mySender):_mySender(mySender){
56 template<class T,class TCorba,class TSeqCorba,class CorbaSender,class servForT,class ptrForT>
57 CorbaNCWithCopyReceiver<T,TCorba,TSeqCorba,CorbaSender,servForT,ptrForT>::~CorbaNCWithCopyReceiver(){
61 template<class T,class TCorba,class TSeqCorba,class CorbaSender,class servForT,class ptrForT>
62 T *CorbaNCWithCopyReceiver<T,TCorba,TSeqCorba,CorbaSender,servForT,ptrForT>::getDistValue(long &size){
63 size=_mySender->getSize();
67 for(long i=0;i<size;i+=TAILLE_SPLIT)
69 if(size-i>TAILLE_SPLIT)
73 TSeqCorba seq=_mySender->sendPart(i,n);
74 T *seqd=(T *)seq->get_buffer(0);
81 template<class T,class TCorba,class TSeqCorba,class CorbaSender,class servForT,class ptrForT>
82 T *CorbaNCWithCopyReceiver<T,TCorba,TSeqCorba,CorbaSender,servForT,ptrForT>::getValue(long &size)
84 return Receiver<T,servForT,ptrForT>::getValue(size,_mySender);
87 template<class T,class TCorba,class TSeqCorba,class CorbaSender,class servForT,class ptrForT>
88 CorbaWCNoCopyReceiver<T,TCorba,TSeqCorba,CorbaSender,servForT,ptrForT>::CorbaWCNoCopyReceiver(CorbaSender mySender):_mySender(mySender){
91 template<class T,class TCorba,class TSeqCorba,class CorbaSender,class servForT,class ptrForT>
92 CorbaWCNoCopyReceiver<T,TCorba,TSeqCorba,CorbaSender,servForT,ptrForT>::~CorbaWCNoCopyReceiver(){
96 template<class T,class TCorba,class TSeqCorba,class CorbaSender,class servForT,class ptrForT>
97 T *CorbaWCNoCopyReceiver<T,TCorba,TSeqCorba,CorbaSender,servForT,ptrForT>::getDistValue(long &size){
98 size=_mySender->getSize();
102 for(long i=0;i<size;i+=TAILLE_SPLIT)
104 if(size-i>TAILLE_SPLIT)
108 TSeqCorba seq=_mySender->sendPart(i,n);
109 TCorba *seqd=seq->get_buffer(0);
110 for(long j=0;j<n;j++)
116 template<class T,class TCorba,class TSeqCorba,class CorbaSender,class servForT,class ptrForT>
117 T *CorbaWCNoCopyReceiver<T,TCorba,TSeqCorba,CorbaSender,servForT,ptrForT>::getValue(long &size)
119 return Receiver<T,servForT,ptrForT>::getValue(size,_mySender);
122 template<class T,class TCorba,class TSeqCorba,class CorbaSender,class servForT,class ptrForT>
123 CorbaWCWithCopyReceiver<T,TCorba,TSeqCorba,CorbaSender,servForT,ptrForT>::CorbaWCWithCopyReceiver(CorbaSender mySender):_mySender(mySender){
126 template<class T,class TCorba,class TSeqCorba,class CorbaSender,class servForT,class ptrForT>
127 CorbaWCWithCopyReceiver<T,TCorba,TSeqCorba,CorbaSender,servForT,ptrForT>::~CorbaWCWithCopyReceiver(){
128 _mySender->release();
131 template<class T,class TCorba,class TSeqCorba,class CorbaSender,class servForT,class ptrForT>
132 T *CorbaWCWithCopyReceiver<T,TCorba,TSeqCorba,CorbaSender,servForT,ptrForT>::getDistValue(long &size){
133 size=_mySender->getSize();
137 for(long i=0;i<size;i+=TAILLE_SPLIT)
139 if(size-i>TAILLE_SPLIT)
143 TSeqCorba seq=_mySender->sendPart(i,n);
144 TCorba *seqd=seq->get_buffer(0);
145 for(long j=0;j<n;j++)
151 template<class T,class TCorba,class TSeqCorba,class CorbaSender,class servForT,class ptrForT>
152 T *CorbaWCWithCopyReceiver<T,TCorba,TSeqCorba,CorbaSender,servForT,ptrForT>::getValue(long &size)
154 return Receiver<T,servForT,ptrForT>::getValue(size,_mySender);
159 template<class T,class CorbaSender,class servForT,class ptrForT>
160 MPIReceiver<T,CorbaSender,servForT,ptrForT>::MPIReceiver(CorbaSender mySender):_mySender(mySender){
163 template<class T,class CorbaSender,class servForT,class ptrForT>
164 MPIReceiver<T,CorbaSender,servForT,ptrForT>::~MPIReceiver(){
165 _mySender->release();
168 template<class T,class CorbaSender,class servForT,class ptrForT>
169 T *MPIReceiver<T,CorbaSender,servForT,ptrForT>::getDistValue(long &size){
175 char port_name_clt [MPI_MAX_PORT_NAME];
181 MPI_Comm_rank(MPI_COMM_WORLD, &myproc);
182 SALOME::MPISender::param_var p =_mySender->getParam();
185 MPI_Errhandler_set(MPI_COMM_WORLD, MPI_ERRORS_RETURN);
186 while ( i != TIMEOUT && MPI_Lookup_name((char*)p->service,MPI_INFO_NULL,port_name_clt) != MPI_SUCCESS) {
189 MPI_Errhandler_set(MPI_COMM_WORLD, MPI_ERRORS_ARE_FATAL);
190 if ( i == TIMEOUT ) {
195 // Connect to service, get the inter-communicator server
196 // Attention MPI_Comm_connect est un appel collectif :
197 // - Si lancement mpirun -c n -----> uniquement MPI_COMM_SELF fonctionne
198 // - Si lancement client_server&client_server ----> MPI_COMM_WORLD fonctionne
200 // TIMEOUT is inefficient since MPI_Comm_Connect doesn't return if we asked for
201 // a service that has been unpublished !
202 MPI_Errhandler_set(MPI_COMM_WORLD, MPI_ERRORS_RETURN);
204 while ( i != TIMEOUT && MPI_Comm_connect(port_name_clt, MPI_INFO_NULL, 0, MPI_COMM_WORLD, &com)!=MPI_SUCCESS ) {
207 MPI_Errhandler_set(MPI_COMM_WORLD, MPI_ERRORS_ARE_FATAL);
208 if ( i == TIMEOUT ) {
213 MPI_Recv( &_n, 1, MPI_LONG, sproc,p->tag1,com,&status);
215 MPI_Recv( _v, _n, MPITRAITS<T>::MpiType, sproc,p->tag2,com,&status);
217 MPI_Comm_disconnect( &com );
222 template<class T,class CorbaSender,class servForT,class ptrForT>
223 T *MPIReceiver<T,CorbaSender,servForT,ptrForT>::getValue(long &size)
225 return Receiver<T,servForT,ptrForT>::getValue(size,_mySender);
231 #include <sys/types.h>
232 #include <sys/socket.h>
233 #include <netinet/in.h>
234 #include <arpa/inet.h>
239 template<class T,int (*myFunc)(XDR*,T*),class CorbaSender,class servForT,class ptrForT>
240 SocketReceiver<T,myFunc,CorbaSender,servForT,ptrForT>::SocketReceiver(CorbaSender mySender) : _mySender(mySender)
246 template<class T,int (*myFunc)(XDR*,T*),class CorbaSender,class servForT,class ptrForT>
247 SocketReceiver<T,myFunc,CorbaSender,servForT,ptrForT>::~SocketReceiver()
251 _mySender->release();
255 template<class T,int (*myFunc)(XDR*,T*),class CorbaSender,class servForT,class ptrForT>
256 T *SocketReceiver<T,myFunc,CorbaSender,servForT,ptrForT>::getValue(long &size)
258 return Receiver<T,servForT,ptrForT>::getValue(size,_mySender);
261 template<class T,int (*myFunc)(XDR*,T*),class CorbaSender,class servForT,class ptrForT>
262 T* SocketReceiver<T,myFunc,CorbaSender,servForT,ptrForT>::getDistValue(long &size)
266 XDR xp; /* pointeur sur le decodeur XDR */
271 SALOME::SocketSender::param_var p = _mySender->getParam();
273 size = p->lend - p->lstart + 1;
276 connectCom(p->internet_address, p->myport);
280 xdrmem_create(&xp,(char*)v,size*sizeof(T),XDR_DECODE );
281 while( n < size*sizeof(T) ){
282 m = read(_clientSockfd, (char*)v+n, size*sizeof(T)-n);
286 SALOME::ExceptionStruct es;
287 es.type = SALOME::COMM;
288 es.text = "error read Socket exception";
289 throw SALOME::SALOME_Exception(es);
293 xdr_vector( &xp, (char*)v, size, sizeof(T), (xdrproc_t)myFunc);
296 _mySender->endOfCom();
299 catch(SALOME::SALOME_Exception &ex){
300 if( ex.details.type == SALOME::COMM )
302 _senderDestruc=false;
303 std::cout << ex.details.text << std::endl;
304 throw MultiCommException("Unknown sender protocol");
313 template<class T,int (*myFunc)(XDR*,T*),class CorbaSender,class servForT,class ptrForT>
314 void SocketReceiver<T,myFunc,CorbaSender,servForT,ptrForT>::initCom()
317 _mySender->initCom();
319 /* Ouverture de la socket */
320 _clientSockfd = socket(AF_INET, SOCK_STREAM, 0);
321 if (_clientSockfd < 0) {
323 SALOME::ExceptionStruct es;
324 es.type = SALOME::COMM;
325 es.text = "error Socket exception";
326 throw SALOME::SALOME_Exception(es);
329 catch(SALOME::SALOME_Exception &ex){
330 if( ex.details.type == SALOME::COMM )
332 _senderDestruc=false;
333 std::cout << ex.details.text << std::endl;
334 throw MultiCommException("Unknown sender protocol");
342 template<class T,int (*myFunc)(XDR*,T*),class CorbaSender,class servForT,class ptrForT>
343 void SocketReceiver<T,myFunc,CorbaSender,servForT,ptrForT>::connectCom(const char *dest_address, int port)
345 struct sockaddr_in serv_addr;
346 struct hostent * server;
347 SALOME::ExceptionStruct es;
350 /* reception of the host structure on the remote process */
351 server = gethostbyname(dest_address);
352 if( server == NULL ) {
354 es.type = SALOME::COMM;
355 es.text = "error unknown host Socket exception";
356 _senderDestruc=false;
357 throw SALOME::SALOME_Exception(es);
360 /* Initialisation of the socket structure */
361 bzero((char*)&serv_addr,sizeof(serv_addr));
362 serv_addr.sin_family = AF_INET;
363 serv_addr.sin_addr.s_addr = INADDR_ANY;
364 bcopy((char *)server->h_addr,
365 (char *)&serv_addr.sin_addr.s_addr,
367 serv_addr.sin_port = htons(port);
369 if( connect(_clientSockfd, (struct sockaddr *) & serv_addr, sizeof(struct sockaddr)) < 0 ){
371 es.type = SALOME::COMM;
372 es.text = "error connect Socket exception";
373 _senderDestruc=false;
374 throw SALOME::SALOME_Exception(es);
377 _mySender->acceptCom();
380 catch(SALOME::SALOME_Exception &ex){
381 if( ex.details.type == SALOME::COMM )
383 _senderDestruc=false;
384 std::cout << ex.details.text << std::endl;
385 throw MultiCommException("Unknown sender protocol");
394 template<class T,int (*myFunc)(XDR*,T*),class CorbaSender,class servForT,class ptrForT>
395 void SocketReceiver<T,myFunc,CorbaSender,servForT,ptrForT>::closeCom()
397 _mySender->closeCom();
398 if( _clientSockfd >= 0 ){
399 close(_clientSockfd);