Changeset 2246 for XIOS/dev/dev_ym/XIOS_COUPLING/src/buffer_server.cpp

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/buffer_server.cpp (modified) (9 diffs)

XIOS/dev/dev_ym/XIOS_COUPLING/src/buffer_server.cpp

@@ -2 +2 @@
 #include "exception.hpp"
 #include "buffer_server.hpp"
+#include "timer.hpp"
 
 
@@ -31 +32 @@
   }
 
-/*
-  void CServerBuffer::createWindows(MPI_Comm oneSidedComm)
-  {
-    MPI_Barrier(oneSidedComm) ;
-    MPI_Win_create(NULL, 0, 1, MPI_INFO_NULL, oneSidedComm, &(windows[0])) ;
-    MPI_Win_create(NULL, 0, 1, MPI_INFO_NULL, oneSidedComm, &(windows[1])) ;
-    hasWindows=true ;
-    updateCurrentWindows() ;
-    MPI_Barrier(oneSidedComm) ;
-  }
-*/
-
-/*
-  bool CServerBuffer::freeWindows()
-  {
-    if (hasWindows)
-    {
-      size_t header[3] ;
-      size_t& control=header[2] ;
-      MPI_Win_lock(MPI_LOCK_EXCLUSIVE,0,0,windows_[0]) ;
-      MPI_Get(&control, 1, MPI_LONG_LONG_INT, windowsRank , 2*sizeof(size_t), 1, MPI_LONG_LONG_INT,windows[0]) ;
-      MPI_Win_unlock(0,windows[0]) ;
-      if (control==2)  // ok for free windows
-      {
-        MPI_Win_free( &(windows[0])) ;
-        MPI_Win_free( &(windows[1])) ;
-        hasWindows=false ;
-        return true ;
-      }
-      else return false ;
-    }
-    else return true ;
-  }
-*/
 
   bool CServerBuffer::isBufferFree(size_t count)
@@ -222 +189 @@
   bool CServerBuffer::getBufferFromClient(size_t timeLine, char*& buffer, size_t& count)
   {
+    count = -1 ;
     if (!hasWindows || resizingBuffer_) return false ;
-
-    
-    size_t header[3] ;
-    size_t& clientTimeline=header[0] ;
-    size_t& clientCount=header[1] ;
-    size_t& control=header[2] ;
+    double time=MPI_Wtime() ;
+    if (time-bufferFromClientTime_ < bufferFromClientLatency_ ) return false;
+    bufferFromClientTime_ = time ;
+    CTimer::get("getBufferFromClient").resume() ;   
+    size_t clientTimeline ;
+    size_t clientCount ;
     bool ok=false ;
     
@@ -238 +206 @@
     
     lockBuffer(); 
-
+    CTimer::get("getBufferFromClient_locked").resume() ;   
 // lock is acquired
 
-    MPI_Get(&clientTimeline, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows],0), 1, MPI_LONG_LONG_INT,windows_[currentWindows]) ;
-    MPI_Get(&clientCount, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows],sizeof(size_t)), 1, MPI_LONG_LONG_INT,windows_[currentWindows]) ;
+    MPI_Get(&clientTimeline, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows],timeLineOffset_*sizeof(size_t)), 1, MPI_LONG_LONG_INT,windows_[currentWindows]) ;
+    MPI_Get(&clientCount, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows],countOffset_*sizeof(size_t)), 1, MPI_LONG_LONG_INT,windows_[currentWindows]) ;
     MPI_Win_flush(windowsRank_, windows_[currentWindows]) ;
 
-//    control=1 ;
-//    MPI_Put(&control, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows],2*sizeof(size_t)), 1, MPI_LONG_LONG_INT,windows_[currentWindows]) ;
-   
-//    MPI_Win_unlock(windowsRank_, windows_[currentWindows]) ;
-    MPI_Win_flush(windowsRank_, windows_[currentWindows]) ;
-//    info(100)<<"getBufferFromClient : windowsRank "<<windowsRank_<<" timeline "<<timeLine<<" clientTimeline "<<clientTimeline<<" clientCount "<<clientCount<<endl ;
     if (timeLine==clientTimeline)
     {
-//      info(50)<<"getBufferFromClient timeLine==clientTimeLine: windowsRank "<<windowsRank_<<" timeline "<<timeLine<<" clientTimeline "<<clientTimeline<<" clientCount "<<clientCount<<endl ;
- 
-//      MPI_Win_lock(MPI_LOCK_EXCLUSIVE,windowsRank_,0,windows_[currentWindows]) ;
       buffer=(char*)getBuffer(clientCount) ;
       count=clientCount ;
@@ -261 +220 @@
       clientTimeline = 0 ;
       clientCount = 0 ;
-//      control=0 ;
-      MPI_Put(&header[0], 2, MPI_LONG_LONG_INT, windowsRank_, MPI_Aint_add(winAddress_[currentWindows],0) , 2, MPI_LONG_LONG_INT,windows_[currentWindows]) ;
+      MPI_Put(&clientTimeline, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows],timeLineOffset_*sizeof(size_t)), 1, MPI_LONG_LONG_INT,windows_[currentWindows]) ;
+      MPI_Put(&clientCount, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows],countOffset_*sizeof(size_t)), 1, MPI_LONG_LONG_INT,windows_[currentWindows]) ;
 
 // release lock
-     unlockBuffer() ;
+      CTimer::get("getBufferFromClient_locked").suspend() ;   
+      unlockBuffer() ;
 
       ok=true ;
@@ -283 +243 @@
     else
     {
-      //MPI_Win_lock(MPI_LOCK_EXCLUSIVE,windowsRank_,0,windows_[currentWindows]) ;
-      //control=0 ;
-      //MPI_Put(&control, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows],2*sizeof(size_t)), 1, MPI_LONG_LONG_INT,windows_[currentWindows]) ;
+      count=0 ;
  
  // release lock
+      CTimer::get("getBufferFromClient_locked").suspend() ; 
       unlockBuffer() ;
     }
-
+    CTimer::get("getBufferFromClient").suspend() ;   
     if (ok) return true ;
 
@@ -299 +258 @@
   {
     if (!hasWindows) return ;
-
-    long long int lock=1 ;
-    long long int zero=0, one=1 ;
-//    control=1 ;
     MPI_Win_lock(MPI_LOCK_EXCLUSIVE,windowsRank_,0,windows_[currentWindows]) ;
-    while(lock!=0)
-    {
-      MPI_Compare_and_swap(&one, &zero, &lock, MPI_LONG_LONG_INT, windowsRank_, MPI_Aint_add(winAddress_[currentWindows],2*sizeof(size_t)),
-                           windows_[currentWindows]) ;
-      MPI_Win_flush(windowsRank_, windows_[currentWindows]) ;
-    }
   }
 
@@ -315 +264 @@
   {
     if (!hasWindows) return ;
-    long long int lock=1 ;
-    long long int zero=0, one=1 ;
-    
-    MPI_Compare_and_swap(&zero, &one, &lock, MPI_LONG_LONG_INT, windowsRank_, MPI_Aint_add(winAddress_[currentWindows],2*sizeof(size_t)),
-                          windows_[currentWindows]) ;
-    MPI_Win_flush(windowsRank_, windows_[currentWindows]) ; 
     MPI_Win_unlock(windowsRank_,windows_[currentWindows]) ;
   }
@@ -327 +270 @@
   {
     if (!hasWindows) return ;
-    size_t finalize=1 ;
+    size_t notify=notifyFinalize_ ;
     lockBuffer(); 
 // lock is acquired
-    MPI_Put(&finalize, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows],3*sizeof(size_t)), 1, MPI_LONG_LONG_INT,windows_[currentWindows]) ;
+    MPI_Put(&notify, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows], notifyOffset_*sizeof(size_t)), 1, MPI_LONG_LONG_INT,windows_[currentWindows]) ;
     unlockBuffer() ;
   }
+
+  void CServerBuffer::notifyBufferResizing(void)
+  {
+    resizingBuffer_=true ;
+    if (!hasWindows) return ;
+    size_t notify=notifyResizeBuffer_ ;
+    lockBuffer(); 
+// lock is acquired
+    MPI_Put(&notify, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows], notifyOffset_*sizeof(size_t)), 1, MPI_LONG_LONG_INT,windows_[currentWindows]) ;
+    unlockBuffer() ;
+  }
 }