Changeset 2246 for XIOS/dev/dev_ym/XIOS_COUPLING/src/manager/window_manager.hpp

Timestamp:

10/11/21 14:41:56 (3 years ago)

Author:

ymipsl

Message:

• Update of the tranfer protocol using one sided communication
• Introduce MPI_Improb/MPI_mrecv to listen incomming request
• Introducing latency when looping over managers

YM

File:

• XIOS/dev/dev_ym/XIOS_COUPLING/src/manager/window_manager.hpp (modified) (4 diffs)

XIOS/dev/dev_ym/XIOS_COUPLING/src/manager/window_manager.hpp

@@ -25 +25 @@
     MPI_Win window_ ;
     void * winBuffer_ ;
+    map<int,double> lastTimeLock_ ;
+    const double latency_=0e-2 ; 
 
     public :
@@ -46 +48 @@
     {
       int lock=state ;
-          
+      double time ;
+      auto it=lastTimeLock_.find(rank) ;
+      if (it == lastTimeLock_.end()) 
+      { 
+        lastTimeLock_[rank] = 0. ; 
+        it=lastTimeLock_.find(rank) ;
+      }
+      double& lastTime = it->second ;
+
       do 
       {
+        time=MPI_Wtime() ;
+        while(time-lastTime < latency_) time=MPI_Wtime() ;
         MPI_Win_lock(MPI_LOCK_EXCLUSIVE, rank, 0, window_) ;
         MPI_Compare_and_swap(&WINDOWS_LOCKED, &state, &lock, MPI_INT, rank, OFFSET_LOCK, window_) ;
         MPI_Win_unlock(rank, window_) ;
+        lastTime=MPI_Wtime() ;
       } while (lock!=state) ;
-
-      
-    }
-
+      
+      
+    }
+
+    void lockWindowExclusive(int rank, int state )
+    {
+      int lock=state ;
+      double time ;
+      auto it=lastTimeLock_.find(rank) ;
+      if (it == lastTimeLock_.end()) 
+      { 
+        lastTimeLock_[rank] = 0. ; 
+        it=lastTimeLock_.find(rank) ;
+      }
+      double& lastTime = it->second ;
+
+      do 
+      {
+        time=MPI_Wtime() ;
+        while(time-lastTime < latency_) time=MPI_Wtime() ;
+        MPI_Win_lock(MPI_LOCK_EXCLUSIVE, rank, 0, window_) ;
+        MPI_Compare_and_swap(&WINDOWS_LOCKED, &state, &lock, MPI_INT, rank, OFFSET_LOCK, window_) ;
+        MPI_Win_unlock(rank, window_) ;
+        lastTime=MPI_Wtime() ;
+      } while (lock!=state) ;
+    }
+
+    void lockWindowExclusive(int rank)
+    {
+      MPI_Win_lock(MPI_LOCK_EXCLUSIVE, rank, 0, window_) ;
+    }
+
+    void lockWindowShared(int rank)
+    {
+      MPI_Win_lock(MPI_LOCK_SHARED, rank, 0, window_) ;
+    }
+
+    void unlockWindow(int rank)
+    {
+      MPI_Win_unlock(rank, window_) ;
+    }
+
+    void flushWindow(int rank)
+    {
+      MPI_Win_flush(rank, window_) ;
+    }
 
     void unlockWindow(int rank, int state )
@@ -77 +132 @@
       MPI_Win_unlock(rank, window_) ;
     }
-    
+
+    template< class T >
+    void updateToLockedWindow(int rank, T* object, void (T::*dumpOut)(CBufferOut&) )
+    {
+      CBufferOut buffer ;
+      (object->*dumpOut)(buffer) ;
+      size_t size=buffer.count() ;
+//      MPI_Win_lock(MPI_LOCK_EXCLUSIVE, rank, 0, window_) ;
+      MPI_Put(&size, SIZE_BUFFER_SIZE, MPI_CHAR, rank, OFFSET_BUFFER_SIZE, SIZE_BUFFER_SIZE, MPI_CHAR, window_) ;
+      MPI_Put(buffer.start(), size, MPI_CHAR, rank,OFFSET_BUFFER, size, MPI_CHAR, window_) ;
+//      MPI_Win_unlock(rank, window_) ;
+    }
+
     template< typename T >
     void updateFromWindow(int rank, T* object, void (T::*dumpIn)(CBufferIn&) ) 
@@ -90 +157 @@
       (object->*dumpIn)(buffer) ;
     }
+
+    template< typename T >
+    void updateFromLockedWindow(int rank, T* object, void (T::*dumpIn)(CBufferIn&) ) 
+    {
+      size_t size ;
+//      MPI_Win_lock(MPI_LOCK_SHARED, rank, 0, window_) ;
+      MPI_Get(&size, SIZE_BUFFER_SIZE, MPI_CHAR, rank, OFFSET_BUFFER_SIZE, SIZE_BUFFER_SIZE, MPI_CHAR, window_) ;
+      MPI_Win_flush(rank,window_) ;
+      CBufferIn buffer(size) ;
+      MPI_Get(buffer.start(), size, MPI_CHAR, rank,OFFSET_BUFFER, size, MPI_CHAR, window_) ;
+//      MPI_Win_unlock(rank, window_) ;
+      MPI_Win_flush(rank, window_) ;
+      (object->*dumpIn)(buffer) ;
+    }
+
 
     template< class T >