src/MPIContainer/MPIContainer_i.cxx

   1 //  SALOME MPIContainer : implemenation of container based on MPI libraries
   2 //
   3 //  Copyright (C) 2003  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.
  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.opencascade.org/SALOME/ or email : webmaster.salome@opencascade.org
  21 //
  22 //
  23 //
  24 //  File   : MPIContainer_i.cxx
  25 //  Module : SALOME
  26
  27 #include <iostream.h>
  28 #include <dlfcn.h>
  29 #include <stdio.h>
  30 #include "SALOME_Component_i.hxx"
  31 #include "MPIContainer_i.hxx"
  32 #include "SALOME_NamingService.hxx"
  33 #include "Utils_SINGLETON.hxx"
  34 #include "OpUtil.hxx"
  35 #include "utilities.h"
  36 #include <pthread.h>  // must be before Python.h !
  37 #include <Python.h>
  38 #include "Container_init_python.hxx"
  39 using namespace std;
  40
  41 // L'appel au registry SALOME ne se fait que pour le process 0
  42 Engines_MPIContainer_i::Engines_MPIContainer_i(int nbproc, int numproc,
  43                                                CORBA::ORB_ptr orb,
  44                                                PortableServer::POA_ptr poa,
  45                                                char * containerName,
  46                                                int argc, char *argv[])
  47   : Engines_Container_i(orb,poa,containerName,argc,argv,false), MPIObject_i(nbproc,numproc)
  48 {
  49   long id=0;
  50   string IdContainerinNS;
  51   char idc[3*sizeof(long)];
  52
  53   MESSAGE("[" << numproc << "] activate object");
  54   _id = _poa->activate_object(this);
  55
  56   if(argc>1)
  57     {
  58       for(int i=0;i<argc;i++)
  59         {
  60           if(strcmp(argv[i],"-id")==NULL)
  61             {
  62               id = atoi(argv[i+1]);
  63               continue;
  64             }
  65         }
  66     }
  67   SCRUTE(id);
  68
  69   if(numproc==0){
  70
  71     _NS = new SALOME_NamingService();
  72     _NS->init_orb( CORBA::ORB::_duplicate(_orb) ) ;
  73
  74     CORBA::Object_var obj=_poa->id_to_reference(*_id);
  75     Engines::Container_var pCont = Engines::Container::_narrow(obj);
  76
  77     string hostname = GetHostname();
  78     _containerName = _NS->BuildContainerNameForNS(containerName,hostname.c_str());
  79     SCRUTE(_containerName);
  80     _NS->Register(pCont, _containerName.c_str());
  81
  82     // A parallel container registers in Naming Service
  83     // on the machine where is process 0. ContainerManager does'nt know the name
  84     // of this machine before the launch of the parallel container. So to get
  85     // the IOR of the parallel container in Naming Service, ContainerManager
  86     // gives a unique Id. The parallel container registers his name under
  87     // /ContainerManager/Id directory in NamingService
  88
  89     IdContainerinNS = "/ContainerManager/id";
  90     sprintf(idc,"%ld",id);
  91     IdContainerinNS += idc;
  92     SCRUTE(IdContainerinNS);
  93     _NS->Register(pCont, IdContainerinNS.c_str());
  94
  95   }
  96
  97   // Root recupere les ior des container des autre process
  98   Engines::MPIObject_var pobj = POA_Engines::MPIContainer::_this();
  99   BCastIOR(_orb,pobj,true);
 100 }
 101
 102 Engines_MPIContainer_i::Engines_MPIContainer_i(int nbproc, int numproc)
 103   : Engines_Container_i(), MPIObject_i(nbproc,numproc)
 104 {
 105 }
 106
 107 Engines_MPIContainer_i::~Engines_MPIContainer_i(void)
 108 {
 109   MESSAGE("[" << _numproc << "] Engines_MPIContainer_i::~Engines_MPIContainer_i()");
 110 }
 111
 112 // Load component
 113 void Engines_MPIContainer_i::Shutdown()
 114 {
 115   int ip;
 116   MESSAGE("[" << _numproc << "] shutdown of MPI Corba Server");
 117   if( _numproc == 0 ){
 118     _NS->Destroy_FullDirectory(_containerName.c_str());
 119     for(ip= 1;ip<_nbproc;ip++)
 120       (Engines::MPIContainer::_narrow((*_tior)[ip]))->Shutdown();
 121   }
 122   _orb->shutdown(0);
 123
 124 }
 125
 126 // Load a component library
 127 bool Engines_MPIContainer_i::load_component_Library(const char* componentName)
 128 {
 129   if( _numproc == 0 ){
 130     // Invocation du chargement du composant dans les autres process
 131     for(int ip= 1;ip<_nbproc;ip++)
 132       (Engines::MPIContainer::_narrow((*_tior)[ip]))->Asload_component_Library(componentName);
 133   }
 134
 135   return Lload_component_Library(componentName);
 136 }
 137
 138 void Engines_MPIContainer_i::Asload_component_Library(const char* componentName)
 139 {
 140   Lload_component_Library(componentName);
 141 }
 142
 143 bool Engines_MPIContainer_i::Lload_component_Library(const char* componentName)
 144 {
 145   string aCompName = componentName;
 146
 147   // --- try dlopen C++ component
 148
 149   string impl_name = string ("lib") + aCompName + string("Engine.so");
 150   SCRUTE(impl_name);
 151
 152   _numInstanceMutex.lock(); // lock to be alone
 153   // (see decInstanceCnt, finalize_removal))
 154   if (_toRemove_map[impl_name]) _toRemove_map.erase(impl_name);
 155   if (_library_map[impl_name])
 156     {
 157       MESSAGE("[" << _numproc << "] Library " << impl_name << " already loaded");
 158       _numInstanceMutex.unlock();
 159       return true;
 160     }
 161
 162   void* handle;
 163   handle = dlopen( impl_name.c_str() , RTLD_LAZY ) ;
 164   if ( handle )
 165     {
 166       _library_map[impl_name] = handle;
 167       _numInstanceMutex.unlock();
 168       return true;
 169     }
 170   else
 171     {
 172       INFOS("[" << _numproc << "] Can't load shared library : " << impl_name);
 173       INFOS("[" << _numproc << "] error dlopen: " << dlerror());
 174     }
 175   _numInstanceMutex.unlock();
 176
 177   // --- try import Python component
 178
 179   INFOS("[" << _numproc << "] try import Python component "<<componentName);
 180   if (_isSupervContainer)
 181     {
 182       INFOS("[" << _numproc << "] Supervision Container does not support Python Component Engines");
 183       return false;
 184     }
 185   if (_library_map[aCompName])
 186     {
 187       return true; // Python Component, already imported
 188     }
 189   else
 190     {
 191       Py_ACQUIRE_NEW_THREAD;
 192       PyObject *mainmod = PyImport_AddModule("__main__");
 193       PyObject *globals = PyModule_GetDict(mainmod);
 194       PyObject *pyCont = PyDict_GetItemString(globals, "pyCont");
 195       PyObject *result = PyObject_CallMethod(pyCont,
 196                                              "import_component",
 197                                              "s",componentName);
 198       int ret= PyInt_AsLong(result);
 199       SCRUTE(ret);
 200       Py_RELEASE_NEW_THREAD;
 201
 202       if (ret) // import possible: Python component
 203         {
 204           _library_map[aCompName] = (void *)pyCont; // any non O value OK
 205           MESSAGE("[" << _numproc << "] import Python: "<<aCompName<<" OK");
 206           return true;
 207         }
 208     }
 209   return false;
 210 }
 211
 212 // Create an instance of component
 213 Engines::Component_ptr
 214 Engines_MPIContainer_i::create_component_instance( const char* componentName,
 215                                                    CORBA::Long studyId)
 216 {
 217   if( _numproc == 0 ){
 218     // Invocation du chargement du composant dans les autres process
 219     for(int ip= 1;ip<_nbproc;ip++)
 220       (Engines::MPIContainer::_narrow((*_tior)[ip]))->Ascreate_component_instance(componentName,studyId);
 221   }
 222
 223   return Lcreate_component_instance(componentName,studyId);
 224 }
 225
 226 void Engines_MPIContainer_i::Ascreate_component_instance( const char* componentName,
 227                                                           CORBA::Long studyId)
 228 {
 229   Lcreate_component_instance(componentName,studyId);
 230 }
 231
 232 Engines::Component_ptr
 233 Engines_MPIContainer_i::Lcreate_component_instance( const char* genericRegisterName, CORBA::Long studyId)
 234 {
 235   if (studyId < 0) {
 236     INFOS("studyId must be > 0 for mono study instance, =0 for multiStudy");
 237     return Engines::Component::_nil() ;
 238   }
 239
 240   Engines::Component_var iobject = Engines::Component::_nil() ;
 241   Engines::MPIObject_var pobj;
 242
 243   string aCompName = genericRegisterName;
 244   if (_library_map[aCompName]) { // Python component
 245     if (_isSupervContainer) {
 246       INFOS("Supervision Container does not support Python Component Engines");
 247       return Engines::Component::_nil();
 248     }
 249     _numInstanceMutex.lock() ; // lock on the instance number
 250     _numInstance++ ;
 251     int numInstance = _numInstance ;
 252     _numInstanceMutex.unlock() ;
 253
 254     char aNumI[12];
 255     sprintf( aNumI , "%d" , numInstance ) ;
 256     string instanceName = aCompName + "_inst_" + aNumI ;
 257     string component_registerName =
 258       _containerName + "/" + instanceName;
 259
 260     Py_ACQUIRE_NEW_THREAD;
 261     PyObject *mainmod = PyImport_AddModule("__main__");
 262     PyObject *globals = PyModule_GetDict(mainmod);
 263     PyObject *pyCont = PyDict_GetItemString(globals, "pyCont");
 264     PyObject *result = PyObject_CallMethod(pyCont,
 265                                            "create_component_instance",
 266                                            "ssl",
 267                                            aCompName.c_str(),
 268                                            instanceName.c_str(),
 269                                            studyId);
 270     string iors = PyString_AsString(result);
 271     SCRUTE(iors);
 272     Py_RELEASE_NEW_THREAD;
 273
 274     CORBA::Object_var obj = _orb->string_to_object(iors.c_str());
 275     iobject = Engines::Component::_narrow( obj ) ;
 276     pobj = Engines::MPIObject::_narrow(obj) ;
 277     if( _numproc == 0 )
 278       _NS->Register(iobject, component_registerName.c_str()) ;
 279     // Root recupere les ior des composants des autre process
 280     BCastIOR(_orb,pobj,false);
 281
 282     return iobject._retn();
 283   }
 284
 285   //--- try C++
 286
 287   string impl_name = string ("lib") + genericRegisterName +string("Engine.so");
 288   void* handle = _library_map[impl_name];
 289   if ( !handle ) {
 290     INFOS("shared library " << impl_name <<"must be loaded before instance");
 291     return Engines::Component::_nil() ;
 292   }
 293   else {
 294     iobject = createMPIInstance(genericRegisterName,
 295                                 handle,
 296                                 studyId);
 297     return iobject._retn();
 298   }
 299 }
 300
 301 Engines::Component_ptr
 302 Engines_MPIContainer_i::createMPIInstance(string genericRegisterName,
 303                                           void *handle,
 304                                           int studyId)
 305 {
 306   Engines::Component_var iobject;
 307   Engines::MPIObject_var pobj;
 308   // --- find the factory
 309
 310   string aGenRegisterName = genericRegisterName;
 311   string factory_name = aGenRegisterName + string("Engine_factory");
 312   SCRUTE(factory_name) ;
 313
 314   typedef  PortableServer::ObjectId * (*MPIFACTORY_FUNCTION)
 315     (int,int,
 316      CORBA::ORB_ptr,
 317      PortableServer::POA_ptr,
 318      PortableServer::ObjectId *,
 319      const char *,
 320      const char *) ;
 321
 322   MPIFACTORY_FUNCTION MPIComponent_factory
 323     = (MPIFACTORY_FUNCTION) dlsym(handle, factory_name.c_str());
 324
 325   char *error ;
 326   if ( (error = dlerror() ) != NULL) {
 327     // Try to load a sequential component
 328     MESSAGE("[" << _numproc << "] Try to load a sequential component");
 329     _numInstanceMutex.unlock() ;
 330     iobject = Engines_Container_i::createInstance(genericRegisterName,handle,studyId);
 331     if( CORBA::is_nil(iobject) ) return Engines::Component::_duplicate(iobject);
 332   }
 333
 334   // --- create instance
 335
 336   iobject = Engines::Component::_nil() ;
 337
 338   try
 339     {
 340       _numInstanceMutex.lock() ; // lock on the instance number
 341       _numInstance++ ;
 342       int numInstance = _numInstance ;
 343       _numInstanceMutex.unlock() ;
 344
 345       char aNumI[12];
 346       sprintf( aNumI , "%d" , numInstance ) ;
 347       string instanceName = aGenRegisterName + "_inst_" + aNumI ;
 348       string component_registerName =
 349         _containerName + "/" + instanceName;
 350
 351       // --- Instanciate required CORBA object
 352
 353       PortableServer::ObjectId *id ; //not owner, do not delete (nore use var)
 354       id = (MPIComponent_factory) ( _nbproc,_numproc,_orb, _poa, _id, instanceName.c_str(),
 355                                  aGenRegisterName.c_str() ) ;
 356
 357       // --- get reference & servant from id
 358
 359       CORBA::Object_var obj = _poa->id_to_reference(*id);
 360       iobject = Engines::Component::_narrow( obj ) ;
 361       pobj = Engines::MPIObject::_narrow(obj) ;
 362
 363       Engines_Component_i *servant =
 364         dynamic_cast<Engines_Component_i*>(_poa->reference_to_servant(iobject));
 365       ASSERT(servant);
 366       //SCRUTE(servant->pd_refCount);
 367       servant->_remove_ref(); // compensate previous id_to_reference
 368       //SCRUTE(servant->pd_refCount);
 369       _listInstances_map[instanceName] = iobject;
 370       _cntInstances_map[aGenRegisterName] += 1;
 371       SCRUTE(aGenRegisterName);
 372       SCRUTE(_cntInstances_map[aGenRegisterName]);
 373       //SCRUTE(servant->pd_refCount);
 374       bool ret_studyId = servant->setStudyId(studyId);
 375       ASSERT(ret_studyId);
 376
 377       // --- register the engine under the name
 378       //     containerName(.dir)/instanceName(.object)
 379
 380       if( _numproc == 0 ){
 381         _NS->Register( iobject , component_registerName.c_str() ) ;
 382         MESSAGE( component_registerName.c_str() << " bound" ) ;
 383       }
 384       // Root recupere les ior des composants des autre process
 385       BCastIOR(_orb,pobj,false);
 386
 387     }
 388   catch (...)
 389     {
 390       INFOS( "Container_i::createInstance exception catched" ) ;
 391     }
 392   return iobject._retn();
 393 }
 394
 395 // Load component
 396 Engines::Component_ptr Engines_MPIContainer_i::load_impl(const char* nameToRegister,
 397                                                  const char* componentName)
 398 {
 399   if( _numproc == 0 ){
 400     // Invocation du chargement du composant dans les autres process
 401     for(int ip= 1;ip<_nbproc;ip++)
 402       (Engines::MPIContainer::_narrow((*_tior)[ip]))->Asload_impl(nameToRegister,
 403                                                                 componentName);
 404   }
 405
 406   return Lload_impl(nameToRegister,componentName);
 407
 408 }
 409
 410 // Load component
 411 void Engines_MPIContainer_i::Asload_impl(const char* nameToRegister,
 412                                          const char* componentName)
 413 {
 414   Lload_impl(nameToRegister,componentName);
 415 }
 416
 417 Engines::Component_ptr Engines_MPIContainer_i::Lload_impl(
 418                                    const char* nameToRegister,
 419                                    const char* componentName)
 420 {
 421   Engines::Component_var iobject;
 422   Engines::MPIObject_var pobj;
 423   char cproc[4];
 424
 425   sprintf(cproc,"_%d",_numproc);
 426
 427   BEGIN_OF("[" << _numproc << "] MPIContainer_i::Lload_impl");
 428
 429   _numInstanceMutex.lock() ; // lock on the instance number
 430   _numInstance++ ;
 431   char _aNumI[12];
 432   sprintf(_aNumI,"%d",_numInstance) ;
 433
 434   string _impl_name = componentName;
 435   string _nameToRegister = nameToRegister;
 436   string instanceName = _nameToRegister + "_inst_" + _aNumI + cproc;
 437   MESSAGE("[" << _numproc << "] instanceName=" << instanceName);
 438
 439   string absolute_impl_name(_impl_name);
 440   MESSAGE("[" << _numproc << "] absolute_impl_name=" << absolute_impl_name);
 441   void * handle = dlopen(absolute_impl_name.c_str(), RTLD_LAZY);
 442   if(!handle){
 443     INFOS("[" << _numproc << "] Can't load shared library : " << absolute_impl_name);
 444     INFOS("[" << _numproc << "] error dlopen: " << dlerror());
 445     return Engines::Component::_nil() ;
 446   }
 447
 448   string factory_name = _nameToRegister + string("Engine_factory");
 449   MESSAGE("[" << _numproc << "] factory_name=" << factory_name) ;
 450
 451   PortableServer::ObjectId * (*MPIComponent_factory) (int,int,
 452                                                   CORBA::ORB_ptr,
 453                                                   PortableServer::POA_ptr,
 454                                                   PortableServer::ObjectId *,
 455                                                   const char *,
 456                                                   const char *) =
 457     (PortableServer::ObjectId * (*) (int,int,
 458                                      CORBA::ORB_ptr,
 459                                      PortableServer::POA_ptr,
 460                                      PortableServer::ObjectId *,
 461                                      const char *,
 462                                      const char *))
 463     dlsym(handle, factory_name.c_str());
 464
 465   char *error ;
 466   if ((error = dlerror()) != NULL){
 467     // Try to load a sequential component
 468     MESSAGE("[" << _numproc << "] Try to load a sequential component");
 469     _numInstanceMutex.unlock() ;
 470     iobject = Engines_Container_i::load_impl(nameToRegister,componentName);
 471     if( CORBA::is_nil(iobject) ) return Engines::Component::_duplicate(iobject);
 472   }
 473   else{
 474     // Instanciation du composant parallele
 475     MESSAGE("[" << _numproc << "] Try to load a parallel component");
 476     PortableServer::ObjectId * id = (MPIComponent_factory)
 477       (_nbproc,_numproc,_orb, _poa, _id, instanceName.c_str(), _nameToRegister.c_str());
 478     // get reference from id
 479     CORBA::Object_var o = _poa->id_to_reference(*id);
 480     pobj = Engines::MPIObject::_narrow(o) ;
 481     iobject = Engines::Component::_narrow(o) ;
 482   }
 483
 484   if( _numproc == 0 ){
 485     // utiliser + tard le registry ici :
 486     // register the engine under the name containerName.dir/nameToRegister.object
 487     string component_registerName = _containerName + "/" + _nameToRegister;
 488     _NS->Register(iobject, component_registerName.c_str()) ;
 489   }
 490
 491   _numInstanceMutex.unlock() ;
 492
 493   // Root recupere les ior des composants des autre process
 494   BCastIOR(_orb,pobj,false);
 495
 496   END_OF("[" <<_numproc << "] MPIContainer_i::Lload_impl");
 497   return Engines::Component::_duplicate(iobject);
 498
 499 }
 500
 501 void Engines_MPIContainer_i::remove_impl(Engines::Component_ptr component_i)
 502 {
 503   int ip;
 504   Engines::Component_ptr cptr;
 505   Engines::MPIObject_ptr pcptr;
 506   Engines::MPIObject_ptr spcptr;
 507
 508   ASSERT(! CORBA::is_nil(component_i));
 509
 510   if( _numproc == 0 ){
 511     // Invocation de la destruction du composant dans les autres process
 512     pcptr = (Engines::MPIObject_ptr)component_i;
 513     for(ip= 1;ip<_nbproc;ip++){
 514       spcptr = Engines::MPIObject::_narrow((*(pcptr->tior()))[ip]);
 515       cptr = (Engines::Component_ptr)spcptr;
 516       (Engines::MPIContainer::_narrow((*_tior)[ip]))->Asremove_impl(cptr);
 517     }
 518   }
 519
 520   Lremove_impl(component_i);
 521 }
 522
 523 void Engines_MPIContainer_i::Asremove_impl(Engines::Component_ptr component_i)
 524 {
 525   Lremove_impl(component_i);
 526 }
 527
 528 void Engines_MPIContainer_i::Lremove_impl(Engines::Component_ptr component_i)
 529 {
 530   BEGIN_OF("[" << _numproc << "] MPIContainer_i::Lremove_impl");
 531
 532   ASSERT(! CORBA::is_nil(component_i));
 533
 534   string instanceName = component_i->instanceName() ;
 535   MESSAGE("[" << _numproc << "] unload component " << instanceName);
 536   component_i->destroy() ;
 537   MESSAGE("[" << _numproc << "] test key handle_map");
 538   _numInstanceMutex.lock() ; // lock on the remove on handle_map
 539   _numInstanceMutex.unlock() ;
 540   MESSAGE("[" << _numproc << "] list handle_map");
 541
 542   END_OF("[" << _numproc << "] MPIContainer_i::Lremove_impl");
 543
 544 }
 545
 546 void Engines_MPIContainer_i::finalize_removal()
 547 {
 548   int ip;
 549
 550   if( _numproc == 0 ){
 551     // Invocation de la destruction du composant dans les autres process
 552     for(ip= 1;ip<_nbproc;ip++)
 553       (Engines::MPIContainer::_narrow((*_tior)[ip]))->Asfinalize_removal();
 554   }
 555
 556   Lfinalize_removal();
 557 }
 558
 559 void Engines_MPIContainer_i::Asfinalize_removal()
 560 {
 561   Lfinalize_removal();
 562 }
 563
 564 void Engines_MPIContainer_i::Lfinalize_removal()
 565 {
 566   BEGIN_OF("[" << _numproc << "] MPIContainer_i::Lfinalize_removal");
 567
 568 //   map<string, void *>::iterator im ;
 569 //   // lock on the explore remove_map & dlclose
 570 //   _numInstanceMutex.lock() ;
 571 //   for (im = remove_map.begin() ; im != remove_map.end() ; im ++)
 572 //     {
 573 //       void * handle = (*im).second ;
 574 //       MESSAGE("[" << _numproc << "] dlclose " << (*im).first);
 575 //       dlclose(handle) ;
 576 //     }
 577 //   MESSAGE("[" << _numproc << "] remove_map.clear()");
 578 //   remove_map.clear() ;
 579 //   _numInstanceMutex.unlock() ;
 580
 581   END_OF("[" << _numproc << "] MPIContainer_i::Lfinalize_removal");
 582 }