1 // Copyright (C) 2007-2010 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.
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];
176 float telps, tuser, tsys, tcpu;
182 MPI_Comm_rank(MPI_COMM_WORLD, &myproc);
183 SALOME::MPISender::param_var p =_mySender->getParam();
186 MPI_Errhandler_set(MPI_COMM_WORLD, MPI_ERRORS_RETURN);
187 while ( i != TIMEOUT && MPI_Lookup_name((char*)p->service,MPI_INFO_NULL,port_name_clt) != MPI_SUCCESS) {
190 MPI_Errhandler_set(MPI_COMM_WORLD, MPI_ERRORS_ARE_FATAL);
191 if ( i == TIMEOUT ) {
196 // Connect to service, get the inter-communicator server
197 // Attention MPI_Comm_connect est un appel collectif :
198 // - Si lancement mpirun -c n -----> uniquement MPI_COMM_SELF fonctionne
199 // - Si lancement client_server&client_server ----> MPI_COMM_WORLD fonctionne
201 // TIMEOUT is inefficient since MPI_Comm_Connect doesn't return if we asked for
202 // a service that has been unpublished !
203 MPI_Errhandler_set(MPI_COMM_WORLD, MPI_ERRORS_RETURN);
205 while ( i != TIMEOUT && MPI_Comm_connect(port_name_clt, MPI_INFO_NULL, 0, MPI_COMM_WORLD, &com)!=MPI_SUCCESS ) {
208 MPI_Errhandler_set(MPI_COMM_WORLD, MPI_ERRORS_ARE_FATAL);
209 if ( i == TIMEOUT ) {
214 MPI_Recv( &_n, 1, MPI_LONG, sproc,p->tag1,com,&status);
216 MPI_Recv( _v, _n, MPITRAITS<T>::MpiType, sproc,p->tag2,com,&status);
218 MPI_Comm_disconnect( &com );
223 template<class T,class CorbaSender,class servForT,class ptrForT>
224 T *MPIReceiver<T,CorbaSender,servForT,ptrForT>::getValue(long &size)
226 return Receiver<T,servForT,ptrForT>::getValue(size,_mySender);
232 #include <sys/types.h>
233 #include <sys/socket.h>
234 #include <netinet/in.h>
235 #include <arpa/inet.h>
240 template<class T,int (*myFunc)(XDR*,T*),class CorbaSender,class servForT,class ptrForT>
241 SocketReceiver<T,myFunc,CorbaSender,servForT,ptrForT>::SocketReceiver(CorbaSender mySender) : _mySender(mySender)
247 template<class T,int (*myFunc)(XDR*,T*),class CorbaSender,class servForT,class ptrForT>
248 SocketReceiver<T,myFunc,CorbaSender,servForT,ptrForT>::~SocketReceiver()
252 _mySender->release();
256 template<class T,int (*myFunc)(XDR*,T*),class CorbaSender,class servForT,class ptrForT>
257 T *SocketReceiver<T,myFunc,CorbaSender,servForT,ptrForT>::getValue(long &size)
259 return Receiver<T,servForT,ptrForT>::getValue(size,_mySender);
262 template<class T,int (*myFunc)(XDR*,T*),class CorbaSender,class servForT,class ptrForT>
263 T* SocketReceiver<T,myFunc,CorbaSender,servForT,ptrForT>::getDistValue(long &size)
267 XDR xp; /* pointeur sur le decodeur XDR */
272 SALOME::SocketSender::param_var p = _mySender->getParam();
274 size = p->lend - p->lstart + 1;
277 connectCom(p->internet_address, p->myport);
281 xdrmem_create(&xp,(char*)v,size*sizeof(T),XDR_DECODE );
282 while( n < size*sizeof(T) ){
283 m = read(_clientSockfd, (char*)v+n, size*sizeof(T)-n);
287 SALOME::ExceptionStruct es;
288 es.type = SALOME::COMM;
289 es.text = "error read Socket exception";
290 throw SALOME::SALOME_Exception(es);
294 xdr_vector( &xp, (char*)v, size, sizeof(T), (xdrproc_t)myFunc);
297 _mySender->endOfCom();
300 catch(SALOME::SALOME_Exception &ex){
301 if( ex.details.type == SALOME::COMM )
303 _senderDestruc=false;
304 std::cout << ex.details.text << std::endl;
305 throw MultiCommException("Unknown sender protocol");
314 template<class T,int (*myFunc)(XDR*,T*),class CorbaSender,class servForT,class ptrForT>
315 void SocketReceiver<T,myFunc,CorbaSender,servForT,ptrForT>::initCom()
318 _mySender->initCom();
320 /* Ouverture de la socket */
321 _clientSockfd = socket(AF_INET, SOCK_STREAM, 0);
322 if (_clientSockfd < 0) {
324 SALOME::ExceptionStruct es;
325 es.type = SALOME::COMM;
326 es.text = "error Socket exception";
327 throw SALOME::SALOME_Exception(es);
330 catch(SALOME::SALOME_Exception &ex){
331 if( ex.details.type == SALOME::COMM )
333 _senderDestruc=false;
334 std::cout << ex.details.text << std::endl;
335 throw MultiCommException("Unknown sender protocol");
343 template<class T,int (*myFunc)(XDR*,T*),class CorbaSender,class servForT,class ptrForT>
344 void SocketReceiver<T,myFunc,CorbaSender,servForT,ptrForT>::connectCom(const char *dest_address, int port)
346 struct sockaddr_in serv_addr;
347 struct hostent * server;
348 SALOME::ExceptionStruct es;
351 /* reception of the host structure on the remote process */
352 server = gethostbyname(dest_address);
353 if( server == NULL ) {
355 es.type = SALOME::COMM;
356 es.text = "error unknown host Socket exception";
357 _senderDestruc=false;
358 throw SALOME::SALOME_Exception(es);
361 /* Initialisation of the socket structure */
362 bzero((char*)&serv_addr,sizeof(serv_addr));
363 serv_addr.sin_family = AF_INET;
364 serv_addr.sin_addr.s_addr = INADDR_ANY;
365 bcopy((char *)server->h_addr,
366 (char *)&serv_addr.sin_addr.s_addr,
368 serv_addr.sin_port = htons(port);
370 if( connect(_clientSockfd, (struct sockaddr *) & serv_addr, sizeof(struct sockaddr)) < 0 ){
372 es.type = SALOME::COMM;
373 es.text = "error connect Socket exception";
374 _senderDestruc=false;
375 throw SALOME::SALOME_Exception(es);
378 _mySender->acceptCom();
381 catch(SALOME::SALOME_Exception &ex){
382 if( ex.details.type == SALOME::COMM )
384 _senderDestruc=false;
385 std::cout << ex.details.text << std::endl;
386 throw MultiCommException("Unknown sender protocol");
395 template<class T,int (*myFunc)(XDR*,T*),class CorbaSender,class servForT,class ptrForT>
396 void SocketReceiver<T,myFunc,CorbaSender,servForT,ptrForT>::closeCom()
398 _mySender->closeCom();
399 if( _clientSockfd >= 0 ){
400 close(_clientSockfd);