Changeset 1757 for XIOS/dev/dev_ym/XIOS_ONE_SIDED/src/buffer_server.cpp

Timestamp:

10/18/19 14:55:57 (5 years ago)

Author:

ymipsl

Message:

Implement one sided communication in client/server protocol to avoid dead-lock when some buffer are full.

YM

File:

• XIOS/dev/dev_ym/XIOS_ONE_SIDED/src/buffer_server.cpp (modified) (6 diffs)

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

@@ -7 +7 @@
 {
 
-  CServerBuffer::CServerBuffer(StdSize buffSize)
+  CServerBuffer::CServerBuffer(vector<MPI_Win>& windows, vector<MPI_Aint>& winAddress, int windowsRank, StdSize buffSize) 
+  : hasWindows(true), windows_(windows), windowsRank_(windowsRank), winAddress_(winAddress)
   {
     size = 3 * buffSize;
@@ -13 +14 @@
     current = 1;
     end = size;
+    used=0 ;
     buffer = new char[size]; // use MPI_ALLOC_MEM later?
+    currentWindows=1 ;
+    if (windows[0]==MPI_WIN_NULL && windows[1]==MPI_WIN_NULL) hasWindows=false ;
   }
 
@@ -21 +25 @@
   }
 
+  void CServerBuffer::updateCurrentWindows(void)
+  {
+    if (currentWindows==0) currentWindows=1 ;
+    else currentWindows=0 ;
+  }
+
+/*
+  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)
@@ -72 +116 @@
   }
 
+  bool CServerBuffer::isBufferEmpty(void)
+  {
+    if (used==0) return true ;
+    else return false;
+  }
 
   void* CServerBuffer::getBuffer(size_t count)
@@ -128 +177 @@
     }
 
+    used+=count ;
     return ret ;
   }
@@ -167 +217 @@
       }
     }
-  }
-
+    used-=count ;
+  }
+
+  bool CServerBuffer::getBufferFromClient(size_t timeLine, char*& buffer, size_t& count)
+  {
+    if (!hasWindows) return false ;
+
+    
+    size_t header[3] ;
+    size_t& clientTimeline=header[0] ;
+    size_t& clientCount=header[1] ;
+    size_t& control=header[2] ;
+    bool ok=false ;
+    
+    MPI_Group group ;
+    int groupSize,groupRank ;
+    MPI_Win_get_group(windows_[currentWindows], &group) ;
+    MPI_Group_size(group, &groupSize) ;
+    MPI_Group_rank(group, &groupRank) ;
+    
+    lockBuffer(); 
+
+// 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_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 ;
+      MPI_Get(buffer, clientCount, MPI_CHAR, windowsRank_, MPI_Aint_add(winAddress_[currentWindows],4*sizeof(size_t)) , clientCount, MPI_CHAR, windows_[currentWindows]) ;
+      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]) ;
+
+// release lock
+     unlockBuffer() ;
+
+      ok=true ;
+      char checksum=0 ;
+      for(size_t i=0;i<count;i++) checksum=checksum+buffer[i] ;
+      char checksumFirst=0 ;
+      for(size_t i=5; i<10 && i<count ;i++) checksumFirst=checksumFirst+buffer[i] ;
+      char checksumLast=0 ;
+      for(size_t i=(count<10)?0:count-10; i<count ; i++) checksumLast=checksumLast+buffer[i] ;
+      
+      info(40)<<"getBufferFromClient timeLine==clientTimeLine: windowsRank "<<windowsRank_<<" timeline "<<timeLine<<" clientTimeline "
+              <<clientTimeline<<" clientCount "<<count<<" checksum "<<(int)checksum<<" "
+              <<(int)buffer[0]<<" "<<(int)buffer[1]<<" "<<(int)buffer[2]<<" "<<(int)buffer[3]<<" "<<(int)buffer[4]<<" "<<(int)buffer[5]<<" " 
+              <<(int)buffer[6]<<" "<<(int)buffer[7]<<" "<<(int)buffer[8]<<" "<<(int)buffer[9]<<" "<<(int)buffer[10]<<" "<<(int)buffer[11]<<endl ;
+
+    }
+    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]) ;
+ 
+ // release lock
+      unlockBuffer() ;
+    }
+
+    if (ok) return true ;
+
+    return false ;
+  }
+  
+  void CServerBuffer::lockBuffer(void)
+  {
+    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]) ;
+    }
+  }
+
+  void CServerBuffer::unlockBuffer(void)
+  {
+    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]) ;
+  }
+  
+  void CServerBuffer::notifyClientFinalize(void)
+  {
+    if (!hasWindows) return ;
+    size_t finalize=1 ;
+    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]) ;
+    unlockBuffer() ;
+  }
 }