Changeset 2403

Timestamp:

09/16/22 15:20:24 (20 months ago)

Author:

ymipsl

Message:

Refactor service manager notification
YM

Location:

XIOS3/trunk/src/manager

Files:

• services_manager.cpp (modified) (5 diffs)
• services_manager.hpp (modified) (4 diffs)

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)
+  {
+
+    int leader ;
+    int poolSize ;
+    
+    info(40)<<"CServicesManager : waiting for pool info : "<<poolId<<endl ; ;
+    bool ok=CXios::getRessourcesManager()->getPoolInfo(poolId, poolSize, leader) ;
+    if (wait)
+    {
+      while (!ok) 
+      {
+        CXios::getDaemonsManager()->eventLoop() ;
+        ok=CXios::getRessourcesManager()->getPoolInfo(poolId, poolSize, leader) ;
+      }
+    }
+
+    if (ok) 
+    {
+      info(40)<<"CServicesManager : create service on other, notification to leader "<<leader<<", serviceId : "<<serviceId<<", service onto : "<<OnServiceId<<endl ;
+      createServicesOntoNotify(leader, serviceId, OnServiceId) ;
+      return true ;
+    }
+    else return false ;
+  }
 
   void CServicesManager::createServicesNotify(int rank, const string& serviceId, int type, int size, int nbPartitions)
   {
-    winNotify_->lockWindow(rank,0) ;
-    winNotify_->updateFromWindow(rank, this, &CServicesManager::notificationsDumpIn) ;
-    notifications_.push_back(std::make_tuple(serviceId,type,size,nbPartitions)) ;
-    winNotify_->updateToWindow(rank, this, &CServicesManager::notificationsDumpOut) ;
-    winNotify_->unlockWindow(rank,0) ;
+    notifyType_=NOTIFY_CREATE_SERVICE ;
+    notifyCreateService_=make_tuple(serviceId, type, size, nbPartitions ) ;
+    sendNotification(rank) ;
+  }
+
+
+  void CServicesManager::createServicesOntoNotify(int rank, const string& serviceId, const string& OnServiceId)
+  {
+    notifyType_=NOTIFY_CREATE_SERVICE_ONTO ;
+    notifyCreateServiceOnto_=make_tuple(serviceId, OnServiceId) ;
+    sendNotification(rank) ;
+  }
+
+  void CServicesManager::sendNotification(int rank)
+  {
+    winNotify_->lockWindowExclusive(rank) ;
+    winNotify_->pushToLockedWindow(rank, this, &CServicesManager::notificationsDumpOut) ;
+    winNotify_->unlockWindow(rank) ;
   }
 
   
-  void CServicesManager::checkCreateServicesNotification(void)
-  {
-    int commRank ;
-    MPI_Comm_rank(xiosComm_,&commRank) ;
-    winNotify_->lockWindow(commRank,0) ;
-    winNotify_->updateFromWindow(commRank, this, &CServicesManager::notificationsDumpIn) ;
-    
-    if (!notifications_.empty())
-    {
-      auto info = notifications_.front() ;
-      xios::info(40)<<"CServicesManager : receive create service notification : "<<get<0>(info)<<endl ;
-      CServer::getServersRessource()->getPoolRessource()->createService(get<0>(info), get<1>(info), get<2>(info), get<3>(info)) ;
-      notifications_.pop_front() ;
-      winNotify_->updateToWindow(commRank, this, &CServicesManager::notificationsDumpOut) ;     
-    }
-    winNotify_->unlockWindow(commRank,0) ;
-
-  }
-
   void CServicesManager::eventLoop(void)
   {
@@ -114 +133 @@
     if (time-lastEventLoop_ > eventLoopLatency_) 
     {
-      checkCreateServicesNotification() ;
+      checkNotifications() ;
       lastEventLoop_=time ;
     }
@@ -120 +139 @@
   }
 
-  
+
+
+  void CServicesManager::checkNotifications(void)
+  {
+    int commRank ;
+    MPI_Comm_rank(xiosComm_, &commRank) ;
+    winNotify_->lockWindowExclusive(commRank) ;
+    winNotify_->popFromLockedWindow(commRank, this, &CServicesManager::notificationsDumpIn) ;
+    winNotify_->unlockWindow(commRank) ;
+    if (notifyType_==NOTIFY_CREATE_SERVICE) createService() ;
+    else if (notifyType_==NOTIFY_CREATE_SERVICE_ONTO) createServiceOnto() ;
+  }
+
+  void CServicesManager::createService(void)
+  {
+    auto& arg=notifyCreateService_ ;
+    CServer::getServersRessource()->getPoolRessource()->createService(get<0>(arg), get<1>(arg), get<2>(arg), get<3>(arg)) ;
+  }
+
+  void CServicesManager::createServiceOnto(void)
+  {
+    auto& arg=notifyCreateService_ ;
+    //CServer::getServersRessource()->getPoolRessource()->createService(get<0>(arg), get<1>(arg), get<2>(arg), get<3>(arg)) ;
+  }
+
   void CServicesManager::notificationsDumpOut(CBufferOut& buffer)
   {
-    
+
     buffer.realloc(maxBufferSize_) ;
-   
-    buffer<<(int)notifications_.size();
-    
-    for(auto it=notifications_.begin();it!=notifications_.end(); ++it) 
-      buffer << std::get<0>(*it) << static_cast<int>(std::get<1>(*it))<< std::get<2>(*it) << std::get<3>(*it)  ;
+    
+    if (notifyType_==NOTIFY_CREATE_SERVICE)
+    {
+      auto& arg=notifyCreateService_ ;
+      buffer << notifyType_<< get<0>(arg) << get<1>(arg) << std::get<2>(arg) << get<3>(arg) ;
+    }
+    else if (notifyType_==NOTIFY_CREATE_SERVICE_ONTO)
+    {
+      auto& arg=notifyCreateServiceOnto_ ;
+      buffer << notifyType_<< get<0>(arg) << get<1>(arg)  ;
+    }
   }
 
   void CServicesManager::notificationsDumpIn(CBufferIn& buffer)
   {
-    std::string id ;
-    int type ;
-    int size; 
-    int nbPartitions ;
-
-    notifications_.clear() ;
-    int nbNotifications ;
-    buffer>>nbNotifications ;
-    for(int i=0;i<nbNotifications;i++) 
-    {
-      buffer>>id>>type>>size>>nbPartitions ;
-      notifications_.push_back(std::make_tuple(id,type,size,nbPartitions)) ;
-    }
-  }
-
+    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) ;
+      }
+      else if (notifyType_==NOTIFY_CREATE_SERVICE_ONTO)
+      {
+        auto& arg=notifyCreateServiceOnto_ ;
+        buffer >> get<0>(arg) >> get<1>(arg) ;
+      }
+    }
+  }  
   
   void CServicesManager::servicesDumpOut(CBufferOut& buffer)
@@ -198 +248 @@
     winServices_->unlockWindow(managerGlobalLeader_) ;
 
-/*
-    winServices_->lockWindow(managerGlobalLeader_,0) ;
-    winServices_->updateFromWindow(managerGlobalLeader_, this, &CServicesManager::servicesDumpIn) ;
-    services_[std::tuple<std::string, std::string,int>(poolId,serviceId,partitionId)]=std::make_tuple(type,size,nbPartitions,leader) ;
-    winServices_->updateToWindow(managerGlobalLeader_, this, &CServicesManager::servicesDumpOut) ;
-    winServices_->unlockWindow(managerGlobalLeader_,0) ;*/
   }
 
@@ -213 +257 @@
     winServices_->updateFromLockedWindow(managerGlobalLeader_, this, &CServicesManager::servicesDumpIn) ;
     winServices_->unlockWindow(managerGlobalLeader_) ;
-/*
-    winServices_->lockWindow(managerGlobalLeader_,0) ;
-    winServices_->updateFromWindow(managerGlobalLeader_, this, &CServicesManager::servicesDumpIn) ;
-    winServices_->unlockWindow(managerGlobalLeader_,0) ;*/
 
     auto it=services_.find(std::tuple<std::string,std::string,int>(poolId,serviceId,partitionId)) ;

XIOS3/trunk/src/manager/services_manager.hpp

@@ -21 +21 @@
     static const int CLIENT=0 ;
     static const int GATHERER=1 ;
-    static const int IO_SERVER=2 ;
-    static const int OUT_SERVER=3 ;
-    static const int ALL_SERVICES=4 ;
+    static const int WRITER=2 ;
+    static const int READER=3 ;
+    static const int IO_SERVER=4 ;
+    static const int OUT_SERVER=5 ;
+    static const int ALL_SERVICES=6 ;
 
+    private:
+    const int NOTIFY_NOTHING=0 ;
+    const int NOTIFY_CREATE_SERVICE=1 ;
+    const int NOTIFY_CREATE_SERVICE_ONTO=2 ;
+    
     public:
     
@@ -31 +38 @@
     
     bool createServices(const std::string& poolId, const std::string& serviceId, int type, int size, int nbPartition, bool wait=true) ;
-    void createServicesNotify(int rank, const string& serviceId, int type, int size, int nbPartitions) ;
-    void checkCreateServicesNotification(void) ;
+    bool createServicesOnto(const std::string& poolId, const std::string& serviceId, const std::string& onServiceId, bool wait=true) ;
+    
     void eventLoop(void) ;
-    void notificationsDumpOut(CBufferOut& buffer) ;
-    void notificationsDumpIn(CBufferIn& buffer) ;
     
     void registerService(const std::string& poolId, const std::string& serviceId, const int& partitionId, int type, int size, int nbPartitions, int leader) ;
@@ -45 +50 @@
     void servicesDumpOut(CBufferOut& buffer) ;
     void servicesDumpIn(CBufferIn& buffer) ;
-    
+
+    private:
+
+    void createService(void) ;
+    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 sendNotification(int rank) ;
+    void checkNotifications(void) ;
+    void notificationsDumpOut(CBufferOut& buffer) ;
+    void notificationsDumpIn(CBufferIn& buffer) ;
+
 
     private :
@@ -55 +71 @@
     MPI_Comm xiosComm_ ;
 
-    std::list<std::tuple<std::string, int, int, int> > notifications_;
-    
+    int notifyType_ ;
+    tuple<std::string, int, int, int> notifyCreateService_ ;
+    tuple<std::string, std::string> notifyCreateServiceOnto_ ;
+   
     std::map<tuple<std::string, std::string, int>, std::tuple<int, int, int, int> > services_ ;