Changeset 2455

Timestamp:

01/09/23 18:41:19 (2 years ago)

Author:

ymipsl

Message:

Make coupling working again.
YM

Location:

XIOS3/dev/XIOS_FILE_SERVICES/src

Files:

• buffer_client.cpp (modified) (2 diffs)
• buffer_client.hpp (modified) (2 diffs)
• manager/ressources_manager.cpp (modified) (2 diffs)
• manager/ressources_manager.hpp (modified) (3 diffs)
• manager/token_manager.cpp (added)
• manager/token_manager.hpp (added)
• node/context.cpp (modified) (1 diff)
• transport/legacy_context_client.cpp (modified) (8 diffs)
• transport/legacy_context_client.hpp (modified) (1 diff)
• transport/legacy_context_server.cpp (modified) (1 diff)
• transport/one_sided_context_client.hpp (modified) (1 diff)

XIOS3/dev/XIOS_FILE_SERVICES/src/buffer_client.cpp

@@ -94 +94 @@
   void CClientBuffer::attachWindows(vector<MPI_Win>& windows)
   {
+    isAttachedWindows_=true ;
     windows_=windows ;
     if (windows_[0]==MPI_WIN_NULL && windows_[1]==MPI_WIN_NULL) hasWindows=false ;
@@ -170 +171 @@
   bool CClientBuffer::isBufferFree(StdSize size)
   {
-  
+    if (!isAttachedWindows_) return false;
+
     lockBuffer();
     count=*bufferCount[current] ;

XIOS3/dev/XIOS_FILE_SERVICES/src/buffer_client.hpp

@@ -34 +34 @@
       void fixBuffer(void) { isGrowableBuffer_=false ;}
       void attachWindows(vector<MPI_Win>& windows) ;
+      bool isAttachedWindows(void) { return isAttachedWindows_ ;}
     private:
        void resizeBuffer(size_t newSize) ;
@@ -69 +70 @@
       const MPI_Comm interComm;
       std::vector<MPI_Win> windows_ ;
-      bool hasWindows ;
-
+      bool hasWindows=false ;
+      bool isAttachedWindows_=false ;
       double latency_=0 ;
       double lastCheckedWithNothing_=0 ;

XIOS3/dev/XIOS_FILE_SERVICES/src/manager/ressources_manager.cpp

@@ -2 +2 @@
 #include "server.hpp"
 #include "servers_ressource.hpp"
+#include "token_manager.hpp"
 #include "timer.hpp"
 
@@ -21 +22 @@
     if (commRank==0 && isXiosServer) MPI_Comm_rank(xiosComm_, &commRank) ; 
     else commRank=0 ;
+    tokenManager_ = new CTokenManager(xiosComm_,commRank) ;
+
     MPI_Allreduce(&commRank, &managerGlobalLeader_, 1, MPI_INT, MPI_SUM, xiosComm_) ;
 

XIOS3/dev/XIOS_FILE_SERVICES/src/manager/ressources_manager.hpp

@@ -11 +11 @@
 #include "window_manager.hpp"
 #include "pool_ressource.hpp"
+#include "token_manager.hpp"
 
 
@@ -53 +54 @@
     void waitPoolRegistration(const string& poolId) ;
     
+
     void registerServerLeader(int leaderRank) ;
     void registerRessourcesSize(int size) ;
     void registerPoolClient(const std::string& poolId,int size,int leader) ;
     void registerPoolServer(const std::string& poolId,int size,int leader) ;
+    CTokenManager* getTokenManager(void) {return tokenManager_ ;} 
 
     int managerGlobalLeader_ ;
@@ -63 +66 @@
 
     CWindowManager* winNotify_ ;
+    CTokenManager* tokenManager_ ;
 
     const size_t maxBufferSize_=1024*1024 ;

XIOS3/dev/XIOS_FILE_SERVICES/src/node/context.cpp

@@ -738 +738 @@
       for(auto couplerOut : couplerOutClient_) couplerOut.second->eventLoop();
       for(auto couplerIn : couplerInClient_) couplerIn.second->eventLoop();
-      for(auto couplerOut : couplerOutServer_) couplerOut.second->eventLoop(enableEventsProcessing);
-      for(auto couplerIn : couplerInServer_) couplerIn.second->eventLoop(enableEventsProcessing);
+      //for(auto couplerOut : couplerOutServer_) couplerOut.second->eventLoop(enableEventsProcessing);
+      //for(auto couplerIn : couplerInServer_) couplerIn.second->eventLoop(enableEventsProcessing);
+      for(auto couplerOut : couplerOutServer_) couplerOut.second->eventLoop();
+      for(auto couplerIn : couplerInServer_) couplerIn.second->eventLoop();
     }
     setCurrent(getId()) ;

XIOS3/dev/XIOS_FILE_SERVICES/src/transport/legacy_context_client.cpp

@@ -13 +13 @@
 #include "server.hpp"
 #include "services.hpp"
+#include "ressources_manager.hpp"
 #include <boost/functional/hash.hpp>
 #include <random>
@@ -110 +111 @@
       }
       
+      MPI_Request req ;
+      MPI_Status status ;
+      MPI_Ibarrier(intraComm,&req) ;
+      int flag ;
+      MPI_Test(&req,&flag,&status) ;
+      while(!flag) 
+      {
+        callGlobalEventLoop() ;
+        MPI_Test(&req,&flag,&status) ;
+      }
+
+
       timeLine++;
     }
@@ -126 +139 @@
      * \return whether the already allocated buffers could be used
     */
-    bool CLegacyContextClient::getBuffers(const size_t timeLine, const list<int>& serverList, const list<int>& sizeList, list<CBufferOut*>& retBuffers,
-                                    bool nonBlocking /*= false*/)
+    void CLegacyContextClient::getBuffers(const size_t timeLine, const list<int>& serverList, const list<int>& sizeList, list<CBufferOut*>& retBuffers)
     {
       list<int>::const_iterator itServer, itSize;
@@ -134 +146 @@
       list<CClientBuffer*>::iterator itBuffer;
       bool areBuffersFree;
-
+     
       for (itServer = serverList.begin(); itServer != serverList.end(); itServer++)
       {
@@ -140 +152 @@
         if (it == buffers.end())
         {
+          CTokenManager* tokenManager = CXios::getRessourcesManager()->getTokenManager() ;
+          size_t token = tokenManager->getToken() ;
+          while (!tokenManager->lockToken(token)) callGlobalEventLoop() ;
           newBuffer(*itServer);
           it = buffers.find(*itServer);
+          checkAttachWindows(it->second,it->first) ;
+          tokenManager->unlockToken(token) ;
         }
         bufferList.push_back(it->second);
@@ -177 +194 @@
         }
 
-      } while (!areBuffersFree && !nonBlocking);
+      } while (!areBuffersFree);
       CTimer::get("Blocking time").suspend();
 
-      if (areBuffersFree)
-      {
-        for (itBuffer = bufferList.begin(), itSize = sizeList.begin(); itBuffer != bufferList.end(); itBuffer++, itSize++)
-          retBuffers.push_back((*itBuffer)->getBuffer(timeLine, *itSize));
-      }
-      return areBuffersFree;
+      for (itBuffer = bufferList.begin(), itSize = sizeList.begin(); itBuffer != bufferList.end(); itBuffer++, itSize++)
+        retBuffers.push_back((*itBuffer)->getBuffer(timeLine, *itSize));
    }
 
@@ -225 +238 @@
       bufOut->put(sendBuff, 4); 
       buffer->checkBuffer(true);
-      
+/*
        // create windows dynamically for one-sided
       if (!isAttachedModeEnabled())
@@ -254 +267 @@
       buffer->attachWindows(windows_[rank]) ;
       if (!isAttachedModeEnabled()) MPI_Barrier(winComm_[rank]) ;
-       
-   }
-
+  */     
+   }
+
+   void CLegacyContextClient::checkAttachWindows(CClientBuffer* buffer, int rank)
+   {
+      if (!buffer->isAttachedWindows())
+      {
+           // create windows dynamically for one-sided
+        if (!isAttachedModeEnabled())
+        { 
+          CTimer::get("create Windows").resume() ;
+          MPI_Comm interComm ;
+          MPI_Intercomm_create(commSelf_, 0, interCommMerged_, clientSize+rank, 0, &interComm) ;
+          MPI_Intercomm_merge(interComm, false, &winComm_[rank]) ;
+          CXios::getMpiGarbageCollector().registerCommunicator(winComm_[rank]) ;
+          MPI_Comm_free(&interComm) ;
+          windows_[rank].resize(2) ;
+      
+          MPI_Win_create_dynamic(MPI_INFO_NULL, winComm_[rank], &windows_[rank][0]);
+          CXios::getMpiGarbageCollector().registerWindow(windows_[rank][0]) ;
+      
+          MPI_Win_create_dynamic(MPI_INFO_NULL, winComm_[rank], &windows_[rank][1]);   
+          CXios::getMpiGarbageCollector().registerWindow(windows_[rank][1]) ;
+
+          CTimer::get("create Windows").suspend() ;
+          buffer->attachWindows(windows_[rank]) ;
+          MPI_Barrier(winComm_[rank]) ;
+        }
+        else
+        {
+          winComm_[rank] = MPI_COMM_NULL ;
+          windows_[rank].resize(2) ;
+          windows_[rank][0] = MPI_WIN_NULL ;
+          windows_[rank][1] = MPI_WIN_NULL ;
+          buffer->attachWindows(windows_[rank]) ;
+        }
+
+      }
+    }
+
+
+  
    /*!
    Verify state of buffers. Buffer is under pending state if there is no message on it

XIOS3/dev/XIOS_FILE_SERVICES/src/transport/legacy_context_client.hpp

@@ -45 +45 @@
 
       // Functions to set/get buffers
-      bool getBuffers(const size_t timeLine, const list<int>& serverList, const list<int>& sizeList, list<CBufferOut*>& retBuffers, bool nonBlocking = false);
+      void getBuffers(const size_t timeLine, const list<int>& serverList, const list<int>& sizeList, list<CBufferOut*>& retBuffers);
       void newBuffer(int rank);
+      void checkAttachWindows(CClientBuffer* buffer , int rank) ;
       bool checkBuffers(list<int>& ranks);
       bool checkBuffers(void);

XIOS3/dev/XIOS_FILE_SERVICES/src/transport/legacy_context_server.cpp

@@ -95 +95 @@
 
     traceOff();
-    MPI_Improbe(MPI_ANY_SOURCE, 20,interComm,&flag,&message, &status);
+    MPI_Improbe(MPI_ANY_SOURCE, 20, interComm,&flag,&message, &status);
     traceOn();
     if (flag==true) listenPendingRequest(message, status) ;

XIOS3/dev/XIOS_FILE_SERVICES/src/transport/one_sided_context_client.hpp

@@ -41 +41 @@
 
       // Functions to set/get buffers
-      bool getBuffers(const size_t timeLine, const list<int>& serverList, const list<int>& sizeList, list<CBufferOut*>& retBuffers, bool nonBlocking = false);
+//      bool getBuffers(const size_t timeLine, const list<int>& serverList, const list<int>& sizeList, list<CBufferOut*>& retBuffers, bool nonBlocking = false);
       void newBuffer(int rank);
       bool checkBuffers(list<int>& ranks);