Changeset 2404

Timestamp:

09/19/22 10:38:09 (21 months ago)

Author:

ymipsl

Message:

Add the possibility to launch a service on same ressource than an other.
YM

Location:

XIOS3/trunk/src

Files:

• manager/pool_ressource.cpp (modified) (9 diffs)
• manager/pool_ressource.hpp (modified) (2 diffs)
• manager/services.cpp (modified) (4 diffs)
• manager/services.hpp (modified) (4 diffs)
• manager/services_manager.cpp (modified) (6 diffs)
• manager/services_manager.hpp (modified) (3 diffs)
• node/context.hpp (modified) (1 diff)
• server.cpp (modified) (1 diff)
• transport/context_server.hpp (modified) (1 diff)

XIOS3/trunk/src/manager/pool_ressource.cpp

@@ -27 +27 @@
     
     winNotify_->lockWindow(commRank,0) ;
-    winNotify_->updateToWindow(commRank, this, &CPoolRessource::createServiceDumpOut) ;  
+    winNotify_->updateToWindow(commRank, this, &CPoolRessource::notificationsDumpOut) ;
     winNotify_->unlockWindow(commRank,0) ;       
     MPI_Barrier(poolComm_) ;
@@ -60 +60 @@
   }
 
-  
+  void CPoolRessource::createServiceOnto(const std::string& serviceId, int type, const std::string& onServiceId)
+  {
+    // for now suppose nbPartitions=1
+    
+    auto it=occupancy_.begin() ;
+    int commSize ;
+    MPI_Comm_size(poolComm_, &commSize) ;
+    
+    info(40)<<"CPoolRessource::createService  : notify createServiceOnto to all pool members ; serviceId : "<<serviceId
+            <<"  onto service Id  :"<< serviceId<<endl ;
+    for(int rank=0; rank<commSize; rank++) createServiceOntoNotify(rank, serviceId, type, onServiceId) ;
+  }
+
+/*  
   void CPoolRessource::createServiceNotify(int rank, const std::string& serviceId, int type, int size, int nbPartitions, 
                                            bool in)
@@ -70 +83 @@
     winNotify_->unlockWindow(rank,0) ;   
   }
-
-
+*/
+  
+  void CPoolRessource::createServiceNotify(int rank, const string& serviceId, int type, int size, int nbPartitions, bool in)
+  {
+    notifyType_=NOTIFY_CREATE_SERVICE ;
+    notifyCreateService_=make_tuple(serviceId, type, size, nbPartitions, in ) ;
+    sendNotification(rank) ;
+  }
+
+
+  void CPoolRessource::createServiceOntoNotify(int rank, const string& serviceId, int type, const string& onServiceId)
+  {
+    notifyType_=NOTIFY_CREATE_SERVICE_ONTO ;
+    notifyCreateServiceOnto_=make_tuple(serviceId, type, onServiceId) ;
+    sendNotification(rank) ;
+  }
+
+
+  void CPoolRessource::sendNotification(int rank)
+  {
+    winNotify_->lockWindowExclusive(rank) ;
+    winNotify_->pushToLockedWindow(rank, this, &CPoolRessource::notificationsDumpOut) ;
+    winNotify_->unlockWindow(rank) ;
+  }
+
+  void CPoolRessource::checkNotifications(void)
+  {
+    int commRank ;
+    MPI_Comm_rank(poolComm_, &commRank) ;
+    winNotify_->lockWindowExclusive(commRank) ;
+    winNotify_->popFromLockedWindow(commRank, this, &CPoolRessource::notificationsDumpIn) ;
+    winNotify_->unlockWindow(commRank) ;
+    if (notifyType_==NOTIFY_CREATE_SERVICE) createService() ;
+    else if (notifyType_==NOTIFY_CREATE_SERVICE_ONTO) createServiceOnto() ;
+  }
+
+
+  void CPoolRessource::notificationsDumpOut(CBufferOut& buffer)
+  {
+
+    buffer.realloc(maxBufferSize_) ;
+    
+    if (notifyType_==NOTIFY_CREATE_SERVICE)
+    {
+      auto& arg=notifyCreateService_ ;
+      buffer << notifyType_<< get<0>(arg) << get<1>(arg) << std::get<2>(arg) << get<3>(arg) << get<4>(arg);
+    }
+    else if (notifyType_==NOTIFY_CREATE_SERVICE_ONTO)
+    {
+      auto& arg=notifyCreateServiceOnto_ ;
+      buffer << notifyType_<< get<0>(arg) << get<1>(arg)<< get<2>(arg)  ;
+    }
+  }
+
+  void CPoolRessource::notificationsDumpIn(CBufferIn& buffer)
+  {
+    if (buffer.bufferSize() == 0) notifyType_= NOTIFY_NOTHING ;
+    else
+    {
+      buffer>>notifyType_;
+      if (notifyType_==NOTIFY_CREATE_SERVICE)
+      {
+        auto& arg=notifyCreateService_ ;
+        buffer >> get<0>(arg) >> get<1>(arg) >> std::get<2>(arg)>> get<3>(arg)>> get<4>(arg) ;
+      }
+      else if (notifyType_==NOTIFY_CREATE_SERVICE_ONTO)
+      {
+        auto& arg=notifyCreateServiceOnto_ ;
+        buffer >> get<0>(arg) >> get<1>(arg) >> get<2>(arg) ;
+      }
+    }
+  }  
+
+  void CPoolRessource::createService(void)
+  {
+    auto& arg = notifyCreateService_ ;
+    createNewService(get<0>(arg), get<1>(arg), get<2>(arg), get<3>(arg), get<4>(arg)) ;
+  }
+  
+  void CPoolRessource::createServiceOnto(void)
+  {
+    auto& arg = notifyCreateServiceOnto_ ;
+    createNewServiceOnto(get<0>(arg), get<1>(arg), get<2>(arg)) ;
+  }
+
+/*
   void CPoolRessource::createServiceDumpOut(CBufferOut& buffer)
   {
@@ -82 +179 @@
   }
 
-
+*/
+
+/*
   void CPoolRessource::createServiceDumpIn(CBufferIn& buffer)
   {
@@ -100 +199 @@
     }
   }
+*/
 
   bool CPoolRessource::eventLoop(bool serviceOnly)
@@ -110 +210 @@
     if (time-lastEventLoop_ > eventLoopLatency_) 
     {
-      checkCreateServiceNotification() ;
+      //checkCreateServiceNotification() ;
+      checkNotifications() ;
       lastEventLoop_=time ;
     }
@@ -128 +229 @@
     else return false ;
   }
-
+/*
   void CPoolRessource::checkCreateServiceNotification(void)
   {
@@ -146 +247 @@
 
   }
+*/
 
   void CPoolRessource::createNewService(const std::string& serviceId, int type, int size, int nbPartitions, bool in)
@@ -186 +288 @@
      MPI_Comm_free(&serviceComm) ;
   }
+  
+  void CPoolRessource::createNewServiceOnto(const std::string& serviceId, int type, const std::string& onServiceId)
+  {
+     
+    info(40)<<"CPoolRessource::createNewServiceOnto  : receive createServiceOnto notification ; serviceId : "
+            <<serviceId<<"  ontoServiceId : "<<onServiceId<<endl ;
+    for(auto& service : services_) 
+    {
+      if (std::get<0>(service.first)==onServiceId)
+      {
+        const MPI_Comm& serviceComm = service.second->getCommunicator() ;
+        MPI_Comm newServiceComm ;
+        MPI_Comm_dup(serviceComm, &newServiceComm) ;
+        int nbPartitions = service.second->getNbPartitions() ;
+        int partitionId = service.second->getPartitionId() ;
+        shared_ptr<CEventScheduler>  eventScheduler = service.second->getEventScheduler() ;
+        info(40)<<"CPoolRessource::createNewServiceOnto ; found onServiceId : "<<onServiceId<<endl  ;
+        services_[std::make_tuple(serviceId,partitionId)] = new CService(newServiceComm, Id_, serviceId, partitionId, type,
+                                                                         nbPartitions, eventScheduler) ;       
+      }
+    }
+    
+  }
 
   void CPoolRessource::createService(MPI_Comm serviceComm, const std::string& serviceId, int partitionId, int type, int nbPartitions) // for clients & attached

XIOS3/trunk/src/manager/pool_ressource.hpp

@@ -21 +21 @@
     CWindowManager* winNotify_ ;
     
-   public:
+    private:
+    const int NOTIFY_NOTHING=0 ;
+    const int NOTIFY_CREATE_SERVICE=1 ;
+    const int NOTIFY_CREATE_SERVICE_ONTO=2 ;
+
+    public:
     CPoolRessource(MPI_Comm poolComm, const std::string& Id) ;
     ~CPoolRessource() ;
@@ -27 +32 @@
     void createService(const std::string& serviceId, int type, int size, int nbPartition) ;
     void createService(MPI_Comm serviceComm, const std::string& serviceId, int partitionId, int type, int nbPartitions) ; 
-    void createServiceNotify(int rank, const std::string& serviceId, int type, int size, int nbPartitions, bool in) ;
-    void createServiceDumpOut(CBufferOut& buffer) ;
-    void createServiceDumpIn(CBufferIn& buffer) ;
-    void checkCreateServiceNotification(void) ;
-    void createNewService(const std::string& serviceId, int type, int size, int nbPartitions, bool in) ;
-     bool eventLoop(bool serviceOnly=false) ;
+    void createServiceOnto(const std::string& serviceId, int type, const std::string& OnServiceId) ;
+    bool eventLoop(bool serviceOnly=false) ;
     CService* getService(const std::string serviceId, int partitionId) { return services_[make_tuple(serviceId,partitionId)]; }
     void finalizeSignal(void) ;
     string getId(void) { return Id_; }
+
+  private:
+    void createServiceNotify(int rank, const string& serviceId, int type, int size, int nbPartitions, bool in) ;
+    void createServiceOntoNotify(int rank, const string& serviceId, int type, const string& onServiceId) ;
+    void sendNotification(int rank) ;
+    void checkNotifications(void) ;
+    void notificationsDumpOut(CBufferOut& buffer) ;
+    void notificationsDumpIn(CBufferIn& buffer) ;
+    void createService(void) ;
+    void createServiceOnto(void) ;
     
-   private:
+//    void createServiceNotify(int rank, const std::string& serviceId, int type, int size, int nbPartitions, bool in) ;
+//    void createServiceDumpOut(CBufferOut& buffer) ;
+//    void createServiceDumpIn(CBufferIn& buffer) ;
+//    void checkCreateServiceNotification(void) ;
+    void createNewService(const std::string& serviceId, int type, int size, int nbPartitions, bool in) ;
+    void createNewServiceOnto(const std::string& serviceId, int type, const string& onServiceId) ;
+    
+    private:
     MPI_Comm poolComm_ ;
     
     std::multimap<int,int> occupancy_ ;
-    std::list<std::tuple<std::string, int, int, int, bool> > notifications_;
+
+//    std::list<std::tuple<std::string, int, int, int, bool> > notifications_;
+    
+    int notifyType_ ;
+    tuple<std::string, int, int, int, bool> notifyCreateService_ ;
+    tuple<std::string, int, std::string> notifyCreateServiceOnto_ ;
+
+
     std::map< std::tuple<std::string, int>, CService*> services_ ;
     std::string Id_ ;

XIOS3/trunk/src/manager/services.cpp

@@ -10 +10 @@
 {
   CService::CService(MPI_Comm serviceComm, const std::string& poolId, const std::string& serviceId, const int& partitionId, 
-                     int type, int nbPartitions) : finalizeSignal_(false), eventScheduler_(nullptr), poolId_(poolId), serviceId_(serviceId),
-                                                   partitionId_(partitionId), type_(type), nbPartitions_(nbPartitions), hasNotification_(false)
+                     int type, int nbPartitions, shared_ptr<CEventScheduler> eventScheduler) 
+                         : finalizeSignal_(false), eventScheduler_(nullptr), poolId_(poolId), serviceId_(serviceId),
+                           partitionId_(partitionId), type_(type), nbPartitions_(nbPartitions), hasNotification_(false)
 
 
@@ -36 +37 @@
       CXios::getServicesManager()->registerService(poolId, serviceId, partitionId, type, commSize, nbPartitions, globalLeader_) ;
     }
-    eventScheduler_ = new CEventScheduler(serviceComm_) ;
+    if (eventScheduler) eventScheduler_ = eventScheduler ;
+    eventScheduler_ = make_shared<CEventScheduler>(serviceComm_) ;
 
     ostringstream oss;
@@ -45 +47 @@
   CService::~CService()
   {
-    delete eventScheduler_ ;
     delete winNotify_ ;
     for(auto& it : contexts_) delete it.second ;
@@ -262 +263 @@
   }
 
-  CEventScheduler* CService::getEventScheduler(void)
+  shared_ptr<CEventScheduler> CService::getEventScheduler(void)
   {
     return eventScheduler_ ;

XIOS3/trunk/src/manager/services.hpp

@@ -20 +20 @@
 
     CService(MPI_Comm serviceComm, const std::string& poolId, const std::string& serviceId, const int& partitionId, 
-             int type, int nbPartitions) ;
+             int type, int nbPartitions, shared_ptr<CEventScheduler> = nullptr) ;
     ~CService() ;
 
@@ -32 +32 @@
     CServerContext* getServerContext(const std::string& contextId) { return contexts_[contextId]; }
     void finalizeSignal(void) ;
-    CEventScheduler* getEventScheduler(void) ;
+    shared_ptr<CEventScheduler> getEventScheduler(void) ;
 
     std::string getPoolId(void) {return poolId_;}
@@ -39 +39 @@
     int getType(void) {return type_;}
     int getNbPartitions(void) {return nbPartitions_;}
+    const MPI_Comm& getCommunicator(void) { return serviceComm_ ;}
     
     private:
@@ -65 +66 @@
     std::map<std::string, CServerContext*> contexts_ ;
     bool finalizeSignal_ ;
-    CEventScheduler* eventScheduler_ ;
+    shared_ptr<CEventScheduler> eventScheduler_ ;
 
     std::string poolId_ ;

XIOS3/trunk/src/manager/services_manager.cpp

@@ -76 +76 @@
   }
 
-  bool CServicesManager::createServicesOnto(const std::string& poolId, const std::string& serviceId, const std::string& OnServiceId, bool wait)
+  bool CServicesManager::createServicesOnto(const std::string& poolId, const std::string& serviceId, int type, const std::string& OnServiceId, bool wait)
   {
 
@@ -96 +96 @@
     {
       info(40)<<"CServicesManager : create service on other, notification to leader "<<leader<<", serviceId : "<<serviceId<<", service onto : "<<OnServiceId<<endl ;
-      createServicesOntoNotify(leader, serviceId, OnServiceId) ;
+      createServicesOntoNotify(leader, serviceId, type, OnServiceId) ;
       return true ;
     }
@@ -110 +110 @@
 
 
-  void CServicesManager::createServicesOntoNotify(int rank, const string& serviceId, const string& OnServiceId)
+  void CServicesManager::createServicesOntoNotify(int rank, const string& serviceId, int type, const string& OnServiceId)
   {
     notifyType_=NOTIFY_CREATE_SERVICE_ONTO ;
-    notifyCreateServiceOnto_=make_tuple(serviceId, OnServiceId) ;
+    notifyCreateServiceOnto_=make_tuple(serviceId, type, OnServiceId) ;
     sendNotification(rank) ;
   }
@@ -160 +160 @@
   void CServicesManager::createServiceOnto(void)
   {
-    auto& arg=notifyCreateService_ ;
-    //CServer::getServersRessource()->getPoolRessource()->createService(get<0>(arg), get<1>(arg), get<2>(arg), get<3>(arg)) ;
+    auto& arg=notifyCreateServiceOnto_ ;
+    CServer::getServersRessource()->getPoolRessource()->createServiceOnto(get<0>(arg), get<1>(arg), get<2>(arg)) ;
   }
 
@@ -177 +177 @@
     {
       auto& arg=notifyCreateServiceOnto_ ;
-      buffer << notifyType_<< get<0>(arg) << get<1>(arg)  ;
+      buffer << notifyType_<< get<0>(arg) << get<1>(arg) << get<2>(arg)  ;
     }
   }
@@ -195 +195 @@
       {
         auto& arg=notifyCreateServiceOnto_ ;
-        buffer >> get<0>(arg) >> get<1>(arg) ;
+        buffer >> get<0>(arg) >> get<1>(arg) >> get<2>(arg) ;
       }
     }

XIOS3/trunk/src/manager/services_manager.hpp

@@ -38 +38 @@
     
     bool createServices(const std::string& poolId, const std::string& serviceId, int type, int size, int nbPartition, bool wait=true) ;
-    bool createServicesOnto(const std::string& poolId, const std::string& serviceId, const std::string& onServiceId, bool wait=true) ;
+    bool createServicesOnto(const std::string& poolId, const std::string& serviceId, int type, const std::string& onServiceId, bool wait=true) ;
     
     void eventLoop(void) ;
@@ -56 +56 @@
     void createServiceOnto(void) ;    
     void createServicesNotify(int rank, const string& serviceId, int type, int size, int nbPartitions) ;
-    void createServicesOntoNotify(int rank, const string& serviceId, const string& OnServiceId) ;
+    void createServicesOntoNotify(int rank, const string& serviceId, int type, const string& OnServiceId) ;
     void sendNotification(int rank) ;
     void checkNotifications(void) ;
@@ -73 +73 @@
     int notifyType_ ;
     tuple<std::string, int, int, int> notifyCreateService_ ;
-    tuple<std::string, std::string> notifyCreateServiceOnto_ ;
+    tuple<std::string, int, std::string> notifyCreateServiceOnto_ ;
    
     std::map<tuple<std::string, std::string, int>, std::tuple<int, int, int, int> > services_ ;

XIOS3/trunk/src/node/context.hpp

@@ -342 +342 @@
         
       private:
-         CEventScheduler* eventScheduler_ ; //! The local event scheduler for context
+         shared_ptr<CEventScheduler> eventScheduler_ ; //! The local event scheduler for context
          size_t hashId_ ; //! the local hashId for scheduler
          size_t timeLine_=0 ;

XIOS3/trunk/src/server.cpp

@@ -190 +190 @@
           ressourcesManager->createPool(CXios::defaultPoolId, nbRessources) ;
           servicesManager->createServices(CXios::defaultPoolId, CXios::defaultServerId, CServicesManager::IO_SERVER,nbRessources,1) ;
+          servicesManager->createServicesOnto(CXios::defaultPoolId, "default_reader", CServicesManager::READER, CXios::defaultServerId) ;
         }
         else

XIOS3/trunk/src/transport/context_server.hpp

@@ -48 +48 @@
 
       size_t hashId ;
-      CEventScheduler* eventScheduler_=nullptr ;
+      shared_ptr<CEventScheduler> eventScheduler_=nullptr ;
   } ;