Changeset 1547 for XIOS/dev/dev_ym/XIOS_ONE_SIDED/src/buffer_client.cpp

Timestamp:

06/20/18 09:09:23 (6 years ago)

Author:

ymipsl

Message:

New communication protocol between clients and servers, using hybrid mode of p2p mixt with one_sided communication in order to avoid dead-locking. The constraint of the maximum number of event that can be bufferized on client side is released.

Dev branch is created to be tested before merging.

YM

File:

• XIOS/dev/dev_ym/XIOS_ONE_SIDED/src/buffer_client.cpp (modified) (7 diffs)

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

@@ -12 +12 @@
   size_t CClientBuffer::maxRequestSize = 0;
 
-  CClientBuffer::CClientBuffer(MPI_Comm interComm, int serverRank, StdSize bufferSize, StdSize estimatedMaxEventSize, StdSize maxBufferedEvents)
+  CClientBuffer::CClientBuffer(MPI_Comm interComm, int serverRank, StdSize bufferSize, StdSize estimatedMaxEventSize)
     : interComm(interComm)
     , serverRank(serverRank)
@@ -20 +20 @@
     , current(0)
     , count(0)
-    , bufferedEvents(0)
-    , maxBufferedEvents(maxBufferedEvents)
     , pending(false)
-  {
-    buffer[0] = new char[bufferSize]; // transform it with MPI_ALLOC_MEM later
-    buffer[1] = new char[bufferSize];
+    , hasWindows(false) 
+  {
+      MPI_Alloc_mem(bufferSize+headerSize, MPI_INFO_NULL, &bufferHeader[0]) ;
+      MPI_Alloc_mem(bufferSize+headerSize, MPI_INFO_NULL, &bufferHeader[1]) ;
+
+
+    buffer[0] = bufferHeader[0]+headerSize ;
+    buffer[1] = bufferHeader[1]+headerSize ;
+    firstTimeLine[0]=(size_t*)bufferHeader[0] ;
+    firstTimeLine[1]=(size_t*)bufferHeader[1] ;
+    bufferCount[0]=(size_t*)bufferHeader[0] +1 ;
+    bufferCount[1]=(size_t*)bufferHeader[1] +1 ;
+    control[0]=(size_t*)bufferHeader[0] +2 ;
+    control[1]=(size_t*)bufferHeader[1] +2 ;
+
+    *firstTimeLine[0]=0 ;
+    *firstTimeLine[1]=0 ;
+    *bufferCount[0]=0 ;
+    *bufferCount[1]=0 ;
+    *control[0]=0 ;
+    *control[1]=0 ;
+    winState[0]=false ;
+    winState[1]=false ;
     retBuffer = new CBufferOut(buffer[current], bufferSize);
-    info(10) << "CClientBuffer: allocated 2 x " << bufferSize << " bytes for server " << serverRank << " with a maximum of " << maxBufferedEvents << " buffered events" << endl;
+    info(10) << "CClientBuffer: allocated 2 x " << bufferSize << " bytes for server " << serverRank << endl;
   }
 
   CClientBuffer::~CClientBuffer()
   {
-   delete [] buffer[0];
-   delete [] buffer[1];
-   delete retBuffer;
+     freeWindows() ;
+     MPI_Free_mem(bufferHeader[0]) ;
+     MPI_Free_mem(bufferHeader[1]) ;
+     delete retBuffer;
+  }
+
+  void CClientBuffer::createWindows(MPI_Comm oneSidedComm)
+  {
+    MPI_Barrier(oneSidedComm) ;
+    MPI_Win_create(bufferHeader[0], bufferSize+headerSize, 1, MPI_INFO_NULL, oneSidedComm, &(windows[0])) ;
+    MPI_Win_create(bufferHeader[1], bufferSize+headerSize, 1, MPI_INFO_NULL, oneSidedComm, &(windows[1])) ;
+
+    MPI_Win_lock(MPI_LOCK_EXCLUSIVE, 0, 0, windows[0]) ;
+    *firstTimeLine[0]=0 ;
+    *bufferCount[0]=0 ;
+    *control[0]=0 ;
+    MPI_Win_unlock(0, windows[0]) ;
+
+    MPI_Win_lock(MPI_LOCK_EXCLUSIVE, 0, 0, windows[1]) ;
+    *firstTimeLine[1]=0 ;
+    *bufferCount[1]=0 ;
+    *control[1]=0 ;
+    MPI_Win_unlock(0, windows[1]) ;
+    winState[0]=false ;
+    winState[1]=false ;
+    MPI_Barrier(oneSidedComm) ;
+    hasWindows=true ;
+  }
+
+  void CClientBuffer::freeWindows()
+  {
+    if (hasWindows)
+    {
+      MPI_Win_lock(MPI_LOCK_EXCLUSIVE, 0, 0, windows[0]) ;
+      MPI_Win_lock(MPI_LOCK_EXCLUSIVE, 0, 0, windows[1]) ;
+      *control[0]=2 ;
+      *control[1]=2 ;
+      MPI_Win_unlock(0, windows[1]) ;
+      MPI_Win_unlock(0, windows[0]) ;
+      
+      MPI_Win_free(&windows[0]) ;
+      MPI_Win_free(&windows[1]) ;
+      hasWindows=false ;
+    }
+  }
+ 
+  void CClientBuffer::lockBuffer(void)
+  {
+    if (hasWindows)
+    {
+      MPI_Win_lock(MPI_LOCK_EXCLUSIVE, 0, 0, windows[current]) ;
+      winState[current]=true ;
+    }
+  }
+
+  void CClientBuffer::unlockBuffer(void)
+  {
+    if (hasWindows)
+    {
+      MPI_Win_unlock(0, windows[current]) ;
+      winState[current]=false ;
+    }
   }
 
@@ -44 +121 @@
   bool CClientBuffer::isBufferFree(StdSize size)
   {
+    bool loop=true ;
+    while (loop) 
+    {
+      lockBuffer();
+      if (*control[current]==0) loop=false ; // attemp to read from server ?
+      else unlockBuffer() ;
+    }
+    
     if (size > bufferSize)
       ERROR("bool CClientBuffer::isBufferFree(StdSize size)",
@@ -59 +144 @@
     }
 
-
-    return (size <= remain() && bufferedEvents < maxBufferedEvents);
-  }
-
-
-  CBufferOut* CClientBuffer::getBuffer(StdSize size)
+      count=*bufferCount[current] ;
+      return (size <= remain());
+  }
+
+
+  CBufferOut* CClientBuffer::getBuffer(size_t timeLine, StdSize size)
   {
     if (size <= remain())
     {
+      info(100)<<"count "<<count<<"   bufferCount[current]  "<<*bufferCount[current]<<endl ;
       retBuffer->realloc(buffer[current] + count, size);
       count += size;
-      bufferedEvents++;
+      if (*firstTimeLine[current]==0) *firstTimeLine[current]=timeLine ;
+      *bufferCount[current]=count ;
       return retBuffer;
     }
@@ -81 +168 @@
   }
 
-  bool CClientBuffer::checkBuffer(void)
+  bool CClientBuffer::checkBuffer(bool send)
   {
     MPI_Status status;
@@ -96 +183 @@
     if (!pending)
     {
+      if (!send) return false ;
       if (count > 0)
       {
-        MPI_Issend(buffer[current], count, MPI_CHAR, serverRank, 20, interComm, &request);
-        pending = true;
-        if (current == 1) current = 0;
-        else current = 1;
-        count = 0;
-        bufferedEvents = 0;
+        lockBuffer() ;
+        if (*control[current]==0 && bufferCount[current] > 0)
+        {
+          MPI_Issend(buffer[current], count, MPI_CHAR, serverRank, 20, interComm, &request);
+          pending = true;
+          *control[current]=0 ;
+          *firstTimeLine[current]=0 ;
+          *bufferCount[current]=0 ;
+
+           unlockBuffer() ;
+
+          if (current == 1) current = 0;
+          else current = 1;
+          count = 0;
+        }
+        else unlockBuffer() ;
       }
     }
@@ -112 +210 @@
   bool CClientBuffer::hasPendingRequest(void)
   {
+   
+    lockBuffer() ;
+    count=*bufferCount[current] ;
+    unlockBuffer() ;
+
     return (pending || count > 0);
   }
+
+
 }