src/ParaMEDMEMComponent/ParaMEDMEMComponent_i.cxx

   1 // Copyright (C) 2007-2013  CEA/DEN, EDF R&D
   2 //
   3 // This library is free software; you can redistribute it and/or
   4 // modify it under the terms of the GNU Lesser General Public
   5 // License as published by the Free Software Foundation; either
   6 // version 2.1 of the License.
   7 //
   8 // This library is distributed in the hope that it will be useful,
   9 // but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  11 // Lesser General Public License for more details.
  12 //
  13 // You should have received a copy of the GNU Lesser General Public
  14 // License along with this library; if not, write to the Free Software
  15 // Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  16 //
  17 // See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
  18 //
  19
  20 #include "ParaMEDMEMComponent_i.hxx"
  21 #include "utilities.h"
  22 #include "Utils_SALOME_Exception.hxx"
  23 using namespace std;
  24 using namespace ParaMEDMEM;
  25
  26 typedef struct
  27 {
  28   bool exception;
  29   string msg;
  30 } except_st;
  31
  32 pthread_mutex_t m1 = PTHREAD_MUTEX_INITIALIZER;
  33 pthread_mutex_t m2 = PTHREAD_MUTEX_INITIALIZER;
  34
  35 ParaMEDMEMComponent_i::ParaMEDMEMComponent_i()
  36 {
  37   _interface = new CommInterface;
  38 }
  39
  40 ParaMEDMEMComponent_i::ParaMEDMEMComponent_i(CORBA::ORB_ptr orb,
  41                                              PortableServer::POA_ptr poa,
  42                                              PortableServer::ObjectId * contId,
  43                                              const char *instanceName,
  44                                              const char *interfaceName,
  45                                              bool regist)
  46   : Engines_Component_i(orb,poa,contId,instanceName,interfaceName,false,regist)
  47 {
  48   _interface = new CommInterface;
  49 }
  50
  51 ParaMEDMEMComponent_i::~ParaMEDMEMComponent_i()
  52 {
  53   MESSAGE("* [" << _numproc << "] ParaMEDMEMComponent destructor");
  54   delete _interface;
  55   pthread_mutex_destroy (&m1);
  56   pthread_mutex_destroy (&m2);
  57 }
  58
  59 void ParaMEDMEMComponent_i::initializeCoupling(const char * coupling, const char * ior) throw(SALOME::SALOME_Exception)
  60 {
  61   int gsize, grank;
  62   except_st *est;
  63   void *ret_th;
  64   pthread_t *th;
  65   ostringstream msg;
  66
  67   pthread_mutex_lock(&m1);
  68   if(_numproc == 0)
  69     {
  70       th = new pthread_t[_nbproc];
  71       for(int ip=1;ip<_nbproc;ip++)
  72         {
  73           thread_st *st = new thread_st;
  74           st->ip = ip;
  75           st->tior = _tior;
  76           st->coupling = coupling;
  77           st->ior = ior;
  78           pthread_create(&(th[ip]),NULL,th_initializecoupling,(void*)st);
  79         }
  80     }
  81
  82   try{
  83     string service = coupling;
  84     if( service.size() == 0 )
  85       throw SALOME_Exception("You have to give a service name !");
  86
  87     if( _gcom.find(service) != _gcom.end() )
  88       {
  89         msg << "service " << service << " already exists !";
  90         throw SALOME_Exception(msg.str().c_str());
  91       }
  92
  93     // Connection to distributed parallel component
  94 #ifdef HAVE_MPI2
  95     remoteMPI2Connect(coupling);
  96 #else
  97     throw SALOME_Exception("You have to use a MPI2 compliant mpi implementation !");
  98 #endif
  99
 100     MPI_Comm_size( _gcom[coupling], &gsize );
 101     MPI_Comm_rank( _gcom[coupling], &grank );
 102     MESSAGE("[" << grank << "] new communicator of " << gsize << " processes");
 103
 104     // Creation of processors group for ParaMEDMEM
 105     // source is always the lower processor numbers
 106     // target is always the upper processor numbers
 107     if(_numproc==grank)
 108       {
 109         _source[coupling] = new MPIProcessorGroup(*_interface,0,_nbproc-1,_gcom[coupling]);
 110         _target[coupling] = new MPIProcessorGroup(*_interface,_nbproc,gsize-1,_gcom[coupling]);
 111         _commgroup[coupling] = _source[coupling];
 112       }
 113     else
 114       {
 115         _source[coupling] = new MPIProcessorGroup(*_interface,0,gsize-_nbproc-1,_gcom[coupling]);
 116         _target[coupling] = new MPIProcessorGroup(*_interface,gsize-_nbproc,gsize-1,_gcom[coupling]);
 117         _commgroup[coupling] = _target[coupling];
 118       }
 119     _connectto [coupling] = ior;
 120     _dec[coupling] = NULL;
 121     _dec_options[coupling] = NULL;
 122
 123   }
 124   catch(const std::exception &ex)
 125     {
 126       MESSAGE(ex.what());
 127       THROW_SALOME_CORBA_EXCEPTION(ex.what(),SALOME::INTERNAL_ERROR);
 128     }
 129
 130   pthread_mutex_unlock(&m1);
 131   if(_numproc == 0)
 132     {
 133       for(int ip=1;ip<_nbproc;ip++)
 134         {
 135           pthread_join(th[ip],&ret_th);
 136           est = (except_st*)ret_th;
 137           if(est->exception)
 138             {
 139               msg << "[" << ip << "] " << est->msg;
 140               THROW_SALOME_CORBA_EXCEPTION(msg.str().c_str(),SALOME::INTERNAL_ERROR);
 141             }
 142           delete est;
 143         }
 144       delete[] th;
 145     }
 146 }
 147
 148 void ParaMEDMEMComponent_i::terminateCoupling(const char * coupling) throw(SALOME::SALOME_Exception)
 149 {
 150   except_st *est;
 151   void *ret_th;
 152   pthread_t *th;
 153   ostringstream msg;
 154
 155   pthread_mutex_lock(&m2);
 156   if(_numproc == 0)
 157     {
 158       th = new pthread_t[_nbproc];
 159       for(int ip=1;ip<_nbproc;ip++)
 160         {
 161           thread_st *st = new thread_st;
 162           st->ip = ip;
 163           st->tior = _tior;
 164           st->coupling = coupling;
 165           pthread_create(&(th[ip]),NULL,th_terminatecoupling,(void*)st);
 166         }
 167     }
 168
 169   try{
 170     string service = coupling;
 171     if( service.size() == 0 )
 172       throw SALOME_Exception("You have to give a service name !");
 173
 174     if( _gcom.find(service) == _gcom.end() )
 175       {
 176         msg << "service " << service << " doesn't exist !";
 177         throw SALOME_Exception(msg.str().c_str());
 178       }
 179
 180     // Disconnection to distributed parallel component
 181 #ifdef HAVE_MPI2
 182     remoteMPI2Disconnect(coupling);
 183 #else
 184     throw SALOME_Exception("You have to use a MPI2 compliant mpi implementation !");
 185 #endif
 186
 187     /* Processors groups and DEC destruction */
 188     delete _source[coupling];
 189     _source.erase(coupling);
 190     delete _target[coupling];
 191     _target.erase(coupling);
 192     delete _dec[coupling];
 193     _dec.erase(coupling);
 194     _commgroup.erase(coupling);
 195     if(_dec_options[coupling])
 196       {
 197         delete _dec_options[coupling];
 198         _dec_options.erase(coupling);
 199       }
 200     _connectto.erase(coupling);
 201   }
 202   catch(const std::exception &ex)
 203     {
 204       MESSAGE(ex.what());
 205       THROW_SALOME_CORBA_EXCEPTION(ex.what(),SALOME::INTERNAL_ERROR);
 206     }
 207   pthread_mutex_unlock(&m2);
 208   if(_numproc == 0)
 209     {
 210       for(int ip=1;ip<_nbproc;ip++)
 211         {
 212           pthread_join(th[ip],&ret_th);
 213           est = (except_st*)ret_th;
 214           if(est->exception)
 215             {
 216               ostringstream msg;
 217               msg << "[" << ip << "] " << est->msg;
 218               THROW_SALOME_CORBA_EXCEPTION(msg.str().c_str(),SALOME::INTERNAL_ERROR);
 219             }
 220           delete est;
 221         }
 222       delete[] th;
 223     }
 224 }
 225
 226 void ParaMEDMEMComponent_i::setInterpolationOptions(const char * coupling,
 227                                                     CORBA::Long print_level,
 228                                                     const char * intersection_type,
 229                                                     CORBA::Double precision,
 230                                                     CORBA::Double median_plane,
 231                                                     CORBA::Boolean do_rotate,
 232                                                     CORBA::Double bounding_box_adjustment,
 233                                                     CORBA::Double bounding_box_adjustment_abs,
 234                                                     CORBA::Double max_distance_for_3Dsurf_intersect,
 235                                                     CORBA::Long orientation,
 236                                                     CORBA::Boolean measure_abs,
 237                                                     const char * splitting_policy,
 238                                                     CORBA::Boolean P1P0_bary_method ) throw(SALOME::SALOME_Exception)
 239 {
 240   except_st *est;
 241   void *ret_th;
 242   pthread_t *th;
 243   ostringstream msg;
 244
 245   if(_numproc == 0)
 246     {
 247       th = new pthread_t[_nbproc];
 248       for(int ip=1;ip<_nbproc;ip++)
 249         {
 250           thread_st *st = new thread_st;
 251           st->ip = ip;
 252           st->tior = _tior;
 253           st->coupling = coupling;
 254           st->print_level = print_level;
 255           st->intersection_type = intersection_type;
 256           st->precision = precision;
 257           st->median_plane = median_plane;
 258           st->do_rotate = do_rotate;
 259           st->bounding_box_adjustment = bounding_box_adjustment;
 260           st->bounding_box_adjustment_abs = bounding_box_adjustment_abs;
 261           st->max_distance_for_3Dsurf_intersect = max_distance_for_3Dsurf_intersect;
 262           st->orientation = orientation;
 263           st->measure_abs = measure_abs;
 264           st->splitting_policy = splitting_policy;
 265           st->P1P0_bary_method = P1P0_bary_method;
 266           pthread_create(&(th[ip]),NULL,th_setinterpolationoptions,(void*)st);
 267         }
 268     }
 269
 270   if(!_dec_options[coupling])
 271     _dec_options[coupling] = new INTERP_KERNEL::InterpolationOptions;
 272
 273   bool ret = _dec_options[coupling]->setInterpolationOptions(print_level,
 274                                                              intersection_type,
 275                                                              precision,
 276                                                              median_plane,
 277                                                              do_rotate,
 278                                                              bounding_box_adjustment,
 279                                                              bounding_box_adjustment_abs,
 280                                                              max_distance_for_3Dsurf_intersect,
 281                                                              orientation,
 282                                                              measure_abs,
 283                                                              splitting_policy,
 284                                                              P1P0_bary_method );
 285
 286   if(!ret)
 287     {
 288       MESSAGE("Error on setting interpolation options");
 289       THROW_SALOME_CORBA_EXCEPTION("Error on setting interpolation options",SALOME::INTERNAL_ERROR);
 290     }
 291
 292   if(_numproc == 0)
 293     {
 294       for(int ip=1;ip<_nbproc;ip++)
 295         {
 296           pthread_join(th[ip],&ret_th);
 297           est = (except_st*)ret_th;
 298           if(est->exception)
 299             {
 300               msg << "[" << ip << "] " << est->msg;
 301               THROW_SALOME_CORBA_EXCEPTION(msg.str().c_str(),SALOME::INTERNAL_ERROR);
 302             }
 303           delete est;
 304         }
 305       delete[] th;
 306     }
 307 }
 308
 309 void ParaMEDMEMComponent_i::_setInputField(SALOME_MED::MPIMEDCouplingFieldDoubleCorbaInterface_ptr fieldptr, MEDCouplingFieldDouble *field)
 310 {
 311   int grank;
 312   except_st *est;
 313   void *ret_th;
 314   pthread_t th;
 315   ostringstream msg;
 316   string coupling;
 317
 318   std::map<std::string,std::string>::const_iterator it = mapSearchByValue(_connectto, fieldptr->getRef());
 319   if(it != _connectto.end())
 320     coupling = (*it).first.c_str();
 321   else
 322     throw SALOME_Exception("Reference of remote component doesn't find in connectto map !");
 323
 324   if(_numproc == 0)
 325     {
 326       thread_st *st = new thread_st;
 327       st->fieldptr = fieldptr;
 328       st->coupling = coupling;
 329       pthread_create(&th,NULL,th_getdata,(void*)st);
 330     }
 331
 332   if( coupling.size() == 0 )
 333     throw SALOME_Exception("You have to give a service name !");
 334
 335   if( _gcom.find(coupling) == _gcom.end() )
 336     {
 337       msg << "service " << coupling << " doesn't exist !";
 338       throw SALOME_Exception(msg.str().c_str());
 339     }
 340
 341   if(!_dec[coupling])
 342     {
 343
 344       MPI_Comm_rank( _gcom[coupling], &grank );
 345
 346       // Creating the intersection Data Exchange Channel
 347       // Processors which received the field are always the second argument of InterpKernelDEC object
 348       if(_numproc==grank)
 349         _dec[coupling] = new InterpKernelDEC(*_target[coupling], *_source[coupling]);
 350       else
 351         _dec[coupling] = new InterpKernelDEC(*_source[coupling], *_target[coupling]);
 352
 353       if(_dec_options[coupling])
 354         _dec[coupling]->copyOptions(*(_dec_options[coupling]));
 355
 356       //Attaching the field to the DEC
 357       _dec[coupling]->attachLocalField(field);
 358
 359       // computing the interpolation matrix
 360       _dec[coupling]->synchronize();
 361
 362     }
 363   else
 364     //Attaching the field to the DEC
 365     _dec[coupling]->attachLocalField(field);
 366
 367   //Receiving data
 368   _dec[coupling]->recvData();
 369
 370   if(_numproc == 0)
 371     {
 372       pthread_join(th,&ret_th);
 373       est = (except_st*)ret_th;
 374       if(est->exception)
 375         throw SALOME_Exception(est->msg.c_str());
 376       delete est;
 377     }
 378
 379 }
 380
 381 void ParaMEDMEMComponent_i::_getOutputField(const char * coupling, MEDCouplingFieldDouble *field)
 382 {
 383   int grank;
 384   string service = coupling;
 385   ostringstream msg;
 386
 387   if( service.size() == 0 )
 388     throw SALOME_Exception("You have to give a service name !");
 389
 390   if( _gcom.find(service) == _gcom.end() )
 391     {
 392       msg << "service " << service << " doesn't exist !";
 393       throw SALOME_Exception(msg.str().c_str());
 394     }
 395
 396   if(!_dec[coupling])
 397     {
 398
 399       MPI_Comm_rank( _gcom[coupling], &grank );
 400
 401       // Creating the intersection Data Exchange Channel
 402       // Processors which sent the field are always the first argument of InterpKernelDEC object
 403       if(_numproc==grank)
 404         _dec[coupling] = new InterpKernelDEC(*_source[coupling], *_target[coupling]);
 405       else
 406         _dec[coupling] = new InterpKernelDEC(*_target[coupling], *_source[coupling]);
 407
 408       if(_dec_options[coupling])
 409         _dec[coupling]->copyOptions(*(_dec_options[coupling]));
 410
 411       //Attaching the field to the DEC
 412       _dec[coupling]->attachLocalField(field);
 413
 414       // computing the interpolation matrix
 415       _dec[coupling]->synchronize();
 416     }
 417   else
 418     //Attaching the field to the DEC
 419     _dec[coupling]->attachLocalField(field);
 420
 421   //Sending data
 422   _dec[coupling]->sendData();
 423 }
 424
 425 void ParaMEDMEMComponent_i::_initializeCoupling(SALOME_MED::MPIMEDCouplingFieldDoubleCorbaInterface_ptr fieldptr)
 426 {
 427   except_st *est;
 428   void *ret_th;
 429   pthread_t *th;
 430   //this string specifies the coupling
 431   string coupling;
 432   //getting IOR string of the remote object
 433   string rcompo = fieldptr->getRef();
 434   if(_numproc == 0){
 435     //getting IOR string of the local object
 436     CORBA::Object_var my_ref = _poa->servant_to_reference (_thisObj);
 437     string lcompo = _orb->object_to_string(my_ref);
 438     //the component does not communicate with itself, a connection is required
 439     if( rcompo.find(lcompo) == std::string::npos ){
 440       th = new pthread_t[1];
 441       //finding the IOR of the remote object in the map
 442       std::map<std::string,std::string>::const_iterator it = mapSearchByValue(_connectto, rcompo);
 443       //if it is not found : connecting two objects : this is the first (and the only) connection between these objects
 444       if (it == _connectto.end()){
 445         //generating the coupling string : concatenation of two IOR strings
 446         coupling = lcompo + rcompo;
 447
 448         //initializing the coupling on the remote object in a thread
 449         thread_st *st = new thread_st;
 450         CORBA::Object_var obj = _orb->string_to_object (rcompo.c_str());
 451         SALOME_MED::ParaMEDMEMComponent_var compo = SALOME_MED::ParaMEDMEMComponent::_narrow(obj);
 452         st->compo = compo._retn();
 453         st->coupling = coupling;
 454         st->ior = lcompo;
 455
 456         pthread_create(&(th[0]),NULL,th_initializecouplingdist,(void*)st);
 457
 458         //initializing the coupling on the local object
 459         initializeCoupling (coupling.c_str(), rcompo.c_str());
 460         pthread_join (th[0], &ret_th);
 461         est = (except_st*)ret_th;
 462         if(est->exception)
 463           THROW_SALOME_CORBA_EXCEPTION(est->msg.c_str(),SALOME::INTERNAL_ERROR);
 464         delete est;
 465         delete[] th;
 466       }
 467     }
 468   }
 469 }
 470
 471 std::map<std::string,std::string>::const_iterator ParaMEDMEMComponent_i::mapSearchByValue(std::map<std::string,std::string> & search_map, std::string search_val)
 472 {
 473   std::map<std::string,std::string>::const_iterator iRet = search_map.end();
 474   for (std::map<std::string,std::string>::const_iterator iTer = search_map.begin(); iTer != search_map.end(); iTer ++)
 475     {
 476       if( iTer->second.find(search_val) != std::string::npos )
 477         {
 478           iRet = iTer;
 479           break;
 480         }
 481     }
 482   return iRet;
 483 }
 484
 485 bool ParaMEDMEMComponent_i::amICoupledWithThisComponent(const char* cref)
 486 {
 487   std::map<std::string,std::string>::const_iterator it = mapSearchByValue(_connectto, cref);
 488   if(it != _connectto.end())
 489     return true;
 490   else
 491     return false;
 492 }
 493
 494 void *th_setinterpolationoptions(void *s)
 495 {
 496   ostringstream msg;
 497   thread_st *st = (thread_st*)s;
 498   except_st *est = new except_st;
 499   est->exception = false;
 500   try
 501     {
 502       SALOME_MED::ParaMEDMEMComponent_var compo=SALOME_MED::ParaMEDMEMComponent::_narrow((*(st->tior))[st->ip]);
 503       compo->setInterpolationOptions(st->coupling.c_str(),
 504                                      st->print_level,
 505                                      st->intersection_type,
 506                                      st->precision,
 507                                      st->median_plane,
 508                                      st->do_rotate,
 509                                      st->bounding_box_adjustment,
 510                                      st->bounding_box_adjustment_abs,
 511                                      st->max_distance_for_3Dsurf_intersect,
 512                                      st->orientation,
 513                                      st->measure_abs,
 514                                      st->splitting_policy,
 515                                      st->P1P0_bary_method);
 516     }
 517   catch(const SALOME::SALOME_Exception &ex)
 518     {
 519       est->exception = true;
 520       est->msg = ex.details.text;
 521     }
 522   catch(const CORBA::Exception &ex)
 523     {
 524       est->exception = true;
 525       msg << "CORBA::Exception: " << ex;
 526       est->msg = msg.str();
 527     }
 528   delete st;
 529   return((void*)est);
 530 }
 531
 532 void *th_initializecoupling(void *s)
 533 {
 534   ostringstream msg;
 535   thread_st *st = (thread_st*)s;
 536   except_st *est = new except_st;
 537   est->exception = false;
 538
 539   try
 540     {
 541       SALOME_MED::ParaMEDMEMComponent_var compo=SALOME_MED::ParaMEDMEMComponent::_narrow((*(st->tior))[st->ip]);
 542       compo->initializeCoupling(st->coupling.c_str(),st->ior.c_str());
 543     }
 544   catch(const SALOME::SALOME_Exception &ex)
 545     {
 546       est->exception = true;
 547       est->msg = ex.details.text;
 548     }
 549   catch(const CORBA::Exception &ex)
 550     {
 551       est->exception = true;
 552       msg << "CORBA::Exception: " << ex;
 553       est->msg = msg.str();
 554     }
 555   delete st;
 556   return((void*)est);
 557 }
 558
 559 void *th_terminatecoupling(void *s)
 560 {
 561   ostringstream msg;
 562   thread_st *st = (thread_st*)s;
 563   except_st *est = new except_st;
 564   est->exception = false;
 565
 566   try
 567     {
 568       SALOME_MED::ParaMEDMEMComponent_var compo=SALOME_MED::ParaMEDMEMComponent::_narrow((*(st->tior))[st->ip]);
 569       compo->terminateCoupling(st->coupling.c_str());
 570     }
 571   catch(const SALOME::SALOME_Exception &ex)
 572     {
 573       est->exception = true;
 574       est->msg = ex.details.text;
 575     }
 576   catch(const CORBA::Exception &ex)
 577     {
 578       est->exception = true;
 579       msg << "CORBA::Exception: " << ex;
 580       est->msg = msg.str();
 581     }
 582   delete st;
 583   return((void*)est);
 584 }
 585
 586 void *th_getdata(void *s)
 587 {
 588   ostringstream msg;
 589   thread_st *st = (thread_st*)s;
 590   except_st *est = new except_st;
 591   est->exception = false;
 592
 593   try
 594     {
 595       st->fieldptr->getDataByMPI(st->coupling.c_str());
 596     }
 597   catch(const SALOME::SALOME_Exception &ex)
 598     {
 599       est->exception = true;
 600       est->msg = ex.details.text;
 601     }
 602   catch(const CORBA::Exception &ex)
 603     {
 604       est->exception = true;
 605       msg << "CORBA::Exception: " << ex;
 606       est->msg = msg.str();
 607     }
 608   delete st;
 609   return((void*)est);
 610 }
 611
 612 void *th_initializecouplingdist(void *s)
 613 {
 614   ostringstream msg;
 615   thread_st *st = (thread_st*)s;
 616   except_st *est = new except_st;
 617   est->exception = false;
 618
 619   try
 620     {
 621       st->compo->initializeCoupling(st->coupling.c_str(), st->ior.c_str());
 622     }
 623   catch(const SALOME::SALOME_Exception &ex)
 624     {
 625       est->exception = true;
 626       est->msg = ex.details.text;
 627     }
 628   catch(const CORBA::Exception &ex)
 629     {
 630       est->exception = true;
 631       msg << "CORBA::Exception: " << ex;
 632       est->msg = msg.str();
 633     }
 634   delete st;
 635   return((void*)est);
 636 }
 637
 638 void *th_terminatecouplingdist(void *s)
 639 {
 640   ostringstream msg;
 641   thread_st *st = (thread_st*)s;
 642   except_st *est = new except_st;
 643   est->exception = false;
 644
 645   try
 646     {
 647       st->compo->terminateCoupling(st->coupling.c_str());
 648     }
 649   catch(const SALOME::SALOME_Exception &ex)
 650     {
 651       est->exception = true;
 652       est->msg = ex.details.text;
 653     }
 654   catch(const CORBA::Exception &ex)
 655     {
 656       est->exception = true;
 657       msg << "CORBA::Exception: " << ex;
 658       est->msg = msg.str();
 659     }
 660   delete st;
 661   return((void*)est);
 662 }