src/ParallelContainer/SALOME_ParallelContainerProxy_i.cxx

   1 // Copyright (C) 2007-2014  CEA/DEN, EDF R&D, OPEN CASCADE
   2 //
   3 // Copyright (C) 2003-2007  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, or (at your option) any later version.
  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.salome-platform.org/ or email : webmaster.salome@opencascade.com
  21 //
  22
  23 //  SALOME_ParallelContainerProxy : implementation of container and engine for Parallel Kernel
  24 //  File   : SALOME_ParallelContainerProxy_i.cxx
  25 //  Author : André RIBES, EDF
  26 //
  27 #include "SALOME_ParallelContainerProxy_i.hxx"
  28
  29 Container_proxy_impl_final::Container_proxy_impl_final(CORBA::ORB_ptr orb,
  30                                                        paco_fabrique_thread * fab_thread,
  31                                                        PortableServer::POA_ptr poa,
  32                                                        std::string containerName,
  33                                                        bool is_a_return_proxy) :
  34   Engines::PACO_Container_proxy_impl(orb, fab_thread, is_a_return_proxy),
  35   Engines::Container_proxy_impl(orb, fab_thread, is_a_return_proxy),
  36   InterfaceManager_impl(orb, fab_thread, is_a_return_proxy)
  37 {
  38   _numInstance = 0;
  39   _hostname = Kernel_Utils::GetHostname();
  40   _containerName = _NS->BuildContainerNameForNS(containerName.c_str(), _hostname.c_str());
  41   _poa = PortableServer::POA::_duplicate(poa);
  42
  43   _fab_thread = fab_thread;
  44
  45   // Add CORBA object to the poa
  46   _id = _poa->activate_object(this);
  47   this->_remove_ref();
  48
  49   // Init SALOME Naming Service
  50   _NS = new SALOME_NamingService();
  51   _NS->init_orb(_orb);
  52
  53   // Init Python container part
  54   CORBA::Object_var container_node = _poa->id_to_reference(*_id);
  55   CORBA::String_var sior =  _orb->object_to_string(container_node);
  56   std::string myCommand="pyCont = SALOME_Container.SALOME_Container_i('";
  57   myCommand += _containerName + "','";
  58   myCommand += sior;
  59   myCommand += "')\n";
  60   Py_ACQUIRE_NEW_THREAD;
  61   PyRun_SimpleString("import SALOME_Container\n");
  62   PyRun_SimpleString((char*)myCommand.c_str());
  63   Py_RELEASE_NEW_THREAD;
  64 }
  65
  66 Container_proxy_impl_final:: ~Container_proxy_impl_final() {
  67   if (_id)
  68     delete _id;
  69   if (_NS)
  70     delete _NS;
  71
  72   // _fab_thread not deleted beacause fab_thread is managed
  73   // by paco_fabrique_manager
  74 }
  75
  76 void
  77 Container_proxy_impl_final::Shutdown()
  78 {
  79   // We Start by destroying all the parallel object
  80   std::list<Container_proxy_impl_final::proxy_object>::iterator itm;
  81   for (itm = _par_obj_inst_list.begin(); itm != _par_obj_inst_list.end(); itm++)
  82   {
  83     try
  84     {
  85       ((*itm).proxy_corba_ref)->destroy();
  86     }
  87     catch(const CORBA::Exception& e)
  88     {
  89       // ignore this entry and continue
  90     }
  91     catch(...)
  92     {
  93       // ignore this entry and continue
  94     }
  95
  96     // Destroy proxy object... parallel object nodes are
  97     // destroyed into the Shutdown of each container nodes
  98     _poa->deactivate_object(*((*itm).proxy_id));
  99     if ((*itm).proxy_id)
 100       delete (*itm).proxy_id;
 101     if ((*itm).proxy_regist)
 102       delete (*itm).proxy_regist;
 103   }
 104
 105   // We call shutdown in each node
 106   for (CORBA::ULong i = 0; i < _infos.nodes.length(); i++)
 107   {
 108     MESSAGE("Shutdown work node : " << i);
 109     CORBA::Object_var object = _orb->string_to_object(_infos.nodes[i]);
 110     Engines::Container_var node = Engines::Container::_narrow(object);
 111     if (!CORBA::is_nil(node))
 112     {
 113       try
 114       {
 115         node->Shutdown();
 116         MESSAGE("Shutdown done node : " << i);
 117       }
 118       catch (...)
 119       {
 120         INFOS("Exception catch during Shutdown of node : " << i);
 121       }
 122     }
 123     else
 124     {
 125       INFOS("Cannot shutdown node " << i << " ref is nil !");
 126     }
 127   }
 128
 129   INFOS("Shutdown Parallel Proxy");
 130   _NS->Destroy_FullDirectory(_containerName.c_str());
 131   _NS->Destroy_Name(_containerName.c_str());
 132   if(!CORBA::is_nil(_orb))
 133     _orb->shutdown(0);
 134 }
 135
 136 // On intercepte cette méthode pour pouvoir ensuite
 137 // déterminer si on doit créer une instance sequentielle
 138 // ou parallèle d'un composant dans la méthode create_component_instance
 139 CORBA::Boolean
 140 Container_proxy_impl_final::load_component_Library(const char* componentName, CORBA::String_out reason)
 141 {
 142   MESSAGE("Begin of load_component_Library on proxy : " << componentName);
 143   reason=CORBA::string_dup("");
 144
 145   std::string aCompName = componentName;
 146
 147   CORBA::Boolean ret = true;
 148   if (_libtype_map.count(aCompName) == 0)
 149   {
 150     _numInstanceMutex.lock(); // lock to be alone
 151
 152     // Default lib is seq
 153     _libtype_map[aCompName] = "seq";
 154
 155     // --- try dlopen C++ component
 156     // If is not a C++ or failed then is maybe
 157     // a seq component...
 158
 159     MESSAGE("Try to load C++ component");
 160 #ifndef WIN32
 161     std::string impl_name = string ("lib") + aCompName + string("Engine.so");
 162 #else
 163     std::string impl_name = aCompName + string("Engine.dll");
 164 #endif
 165     void* handle;
 166 #ifndef WIN32
 167     handle = dlopen( impl_name.c_str() , RTLD_LAZY ) ;
 168 #else
 169     handle = dlopen( impl_name.c_str() , 0 ) ;
 170 #endif
 171     if ( handle )
 172     {
 173       _library_map[impl_name] = handle;
 174       MESSAGE("Library " << impl_name << " loaded");
 175
 176       //Test if lib could contain a parallel component
 177
 178       std::string paco_test_fct_signature = aCompName + std::string("_isAPACO_Component");
 179       INFOS("SIG is : " << paco_test_fct_signature);
 180       PACO_TEST_FUNCTION paco_test_fct = NULL;
 181 #ifndef WIN32
 182       paco_test_fct = (PACO_TEST_FUNCTION)dlsym(handle, paco_test_fct_signature.c_str());
 183 #else
 184       paco_test_fct = (PACO_TEST_FUNCTION)GetProcAddress((HINSTANCE)handle, paco_test_fct_signature.c_str());
 185 #endif
 186       if (paco_test_fct)
 187       {
 188         // PaCO Component found
 189         MESSAGE("PACO LIB FOUND");
 190         _libtype_map[aCompName] = "par";
 191       }
 192       else
 193       {
 194         MESSAGE("SEQ LIB FOUND");
 195 #ifndef WIN32
 196         MESSAGE("dlerror() result is : " << dlerror());
 197 #endif
 198       }
 199     }
 200     else
 201     {
 202       MESSAGE("Error in importing Cpp component : " << impl_name);
 203 #ifndef WIN32
 204       MESSAGE("dlerror() result is : " << dlerror());
 205 #endif
 206
 207       MESSAGE("Try to import Python component "<<componentName);
 208       Py_ACQUIRE_NEW_THREAD;
 209       PyObject *mainmod = PyImport_AddModule("__main__");
 210       PyObject *globals = PyModule_GetDict(mainmod);
 211       PyObject *pyCont = PyDict_GetItemString(globals, "pyCont");
 212       PyObject *result = PyObject_CallMethod(pyCont,
 213                                              (char*)"import_component",
 214                                              (char*)"s",componentName);
 215       std::string ret_p= PyString_AsString(result);
 216       Py_XDECREF(result);
 217       Py_RELEASE_NEW_THREAD;
 218
 219       if (ret_p=="") // import possible: Python component
 220       {
 221         MESSAGE("import Python: " << aCompName <<" OK");
 222       }
 223       else
 224       {
 225         MESSAGE("Error in importing Python component : " << aCompName);
 226         ret = false;
 227       }
 228     }
 229     _numInstanceMutex.unlock();
 230   }
 231
 232   // Call load_component_Library in each node
 233   if (ret)
 234   {
 235     for (CORBA::ULong i = 0; i < _infos.nodes.length(); i++)
 236     {
 237       MESSAGE("Call load_component_Library work node : " << i);
 238       CORBA::Object_var object = _orb->string_to_object(_infos.nodes[i]);
 239       Engines::Container_var node = Engines::Container::_narrow(object);
 240       if (!CORBA::is_nil(node))
 241       {
 242         char* reason;
 243         try
 244         {
 245           node->load_component_Library(componentName,reason);
 246           MESSAGE("Call load_component_Library done node : " << i);
 247           CORBA::string_free(reason);
 248         }
 249         catch (...)
 250         {
 251           INFOS("Exception catch during load_component_Library of node : " << i);
 252           CORBA::string_free(reason);
 253           ret = false;
 254         }
 255       }
 256       else
 257       {
 258         INFOS("Cannot call load_component_Library node " << i << " ref is nil !");
 259         ret = false;
 260       }
 261     }
 262   }
 263
 264   // If ret is false -> lib is not loaded !
 265   if (!ret)
 266   {
 267     INFOS("Cannot call load_component_Library " << aCompName);
 268     _libtype_map.erase(aCompName);
 269   }
 270   return ret;
 271 }
 272
 273 Engines::EngineComponent_ptr
 274 Container_proxy_impl_final::create_component_instance(const char* componentName, ::CORBA::Long studyId)
 275 {
 276   Engines::FieldsDict_var env = new Engines::FieldsDict;
 277   char* reason;
 278   Engines::EngineComponent_ptr compo = create_component_instance_env(componentName, studyId, env, reason);
 279   CORBA::string_free(reason);
 280   return compo;
 281 }
 282
 283 // Il y a deux cas :
 284 // Composant sequentiel -> on le créer sur le noeud 0 (on pourrait faire une répartition de charge)
 285 // Composant parallèle -> création du proxy ici puis appel de la création de chaque objet participant
 286 // au composant parallèle
 287 Engines::EngineComponent_ptr
 288 Container_proxy_impl_final::create_component_instance_env(const char* componentName, ::CORBA::Long studyId,
 289                                                           const Engines::FieldsDict& env, CORBA::String_out reason)
 290 {
 291   reason=CORBA::string_dup("");
 292
 293   std::string aCompName = componentName;
 294   if (_libtype_map.count(aCompName) == 0)
 295   {
 296     // Component is not loaded !
 297     INFOS("Proxy: component is not loaded ! : " << aCompName);
 298     return Engines::EngineComponent::_nil();
 299   }
 300
 301   // If it is a sequential component
 302   if (_libtype_map[aCompName] == "seq")
 303   {
 304     _numInstanceMutex.lock(); // lock on the instance number
 305     _numInstance++;
 306     _numInstanceMutex.unlock();
 307     Engines::PACO_Container_proxy_impl::updateInstanceNumber();
 308     return Engines::Container_proxy_impl::create_component_instance(componentName, studyId);
 309   }
 310
 311   // Parallel Component !
 312   Engines::EngineComponent_var component_proxy = Engines::EngineComponent::_nil();
 313
 314   // On commence par créer le proxy
 315 #ifndef WIN32
 316   std::string impl_name = string ("lib") + aCompName + string("Engine.so");
 317 #else
 318   std::string impl_name = aCompName + string("Engine.dll");
 319 #endif
 320   void* handle = _library_map[impl_name];
 321   std::string factory_name = aCompName + std::string("EngineProxy_factory");
 322
 323   MESSAGE("Creating component proxy : " << factory_name);
 324   FACTORY_FUNCTION component_proxy_factory = (FACTORY_FUNCTION) dlsym(handle, factory_name.c_str());
 325
 326   if (!component_proxy_factory)
 327   {
 328     INFOS("Can't resolve symbol: " + factory_name);
 329 #ifndef WIN32
 330     INFOS("dlerror() result is : " << dlerror());
 331 #endif
 332     return Engines::EngineComponent::_nil() ;
 333   }
 334   try {
 335     _numInstanceMutex.lock() ; // lock on the instance number
 336     _numInstance++ ;
 337     int numInstance = _numInstance ;
 338     _numInstanceMutex.unlock() ;
 339
 340     char aNumI[12];
 341     sprintf( aNumI , "%d" , numInstance ) ;
 342     string instanceName = aCompName + "_inst_" + aNumI ;
 343     string component_registerName = _containerName + "/" + instanceName;
 344
 345     // --- Instanciate required CORBA object
 346     Container_proxy_impl_final::proxy_object * proxy = new Container_proxy_impl_final::proxy_object();
 347
 348     proxy->proxy_id = (component_proxy_factory) (_orb,
 349                                                  _fab_thread,
 350                                                  _poa,
 351                                                  _id,
 352                                                  &(proxy->proxy_regist),
 353                                                  instanceName.c_str(),
 354                                                  _parallel_object_topology.total);
 355
 356     // --- get reference & servant from id
 357     CORBA::Object_var obj = _poa->id_to_reference(*(proxy->proxy_id));
 358     component_proxy = Engines::EngineComponent::_narrow(obj);
 359     proxy->proxy_corba_ref = component_proxy;
 360
 361     if (!CORBA::is_nil(component_proxy))
 362     {
 363       _cntInstances_map[impl_name] += 1;
 364       _par_obj_inst_list.push_back(*proxy);
 365       delete proxy;
 366
 367       // --- register the engine under the name
 368       //     containerName(.dir)/instanceName(.object)
 369       _NS->Register(component_proxy , component_registerName.c_str()) ;
 370       MESSAGE(component_registerName.c_str() << " bound" ) ;
 371     }
 372     else
 373     {
 374       INFOS("The factory returns a nil object !");
 375       return Engines::EngineComponent::_nil();
 376     }
 377
 378   }
 379   catch (...)
 380   {
 381     INFOS( "Exception catched in Proxy creation" );
 382     return Engines::EngineComponent::_nil();
 383   }
 384
 385   // Create on each node a work node
 386   for (CORBA::ULong i = 0; i < _infos.nodes.length(); i++)
 387   {
 388     MESSAGE("Call create_paco_component_node_instance on work node : " << i);
 389     CORBA::Object_var object = _orb->string_to_object(_infos.nodes[i]);
 390     Engines::PACO_Container_var node = Engines::PACO_Container::_narrow(object);
 391     if (!CORBA::is_nil(node))
 392     {
 393       try
 394       {
 395         node->create_paco_component_node_instance(componentName, _containerName.c_str(), studyId);
 396         MESSAGE("Call create_paco_component_node_instance done on node : " << i);
 397       }
 398       catch (SALOME::SALOME_Exception & ex)
 399       {
 400         INFOS("SALOME_EXCEPTION : " << ex.details.text);
 401         return Engines::EngineComponent::_nil();
 402       }
 403       catch (...)
 404       {
 405         INFOS("Unknown Exception catch during create_paco_component_node_instance on node : " << i);
 406         return Engines::EngineComponent::_nil();
 407       }
 408     }
 409     else
 410     {
 411       INFOS("Cannot call create_paco_component_node_instance on node " << i << " ref is nil !");
 412       return Engines::EngineComponent::_nil();
 413     }
 414   }
 415
 416   // Start Parallel object
 417   PaCO::InterfaceManager_var paco_proxy = PaCO::InterfaceManager::_narrow(component_proxy);
 418   paco_proxy->start();
 419
 420   return component_proxy;
 421 }