Changeset 1757

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

Location:

XIOS/dev/dev_ym/XIOS_ONE_SIDED/src

Files:

• buffer_client.cpp (modified) (8 diffs)
• buffer_client.hpp (modified) (3 diffs)
• buffer_server.cpp (modified) (6 diffs)
• buffer_server.hpp (modified) (3 diffs)
• context_client.cpp (modified) (15 diffs)
• context_client.hpp (modified) (6 diffs)
• context_server.cpp (modified) (12 diffs)
• context_server.hpp (modified) (5 diffs)
• node/context.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, vector<MPI_Win>& windows, int clientRank, int serverRank, StdSize bufferSize, StdSize estimatedMaxEventSize)
     : interComm(interComm)
+    , clientRank_(clientRank)
     , serverRank(serverRank)
     , bufferSize(bufferSize)
@@ -20 +21 @@
     , 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) 
+    , windows_(windows)
+  {
+    if (windows[0]==MPI_WIN_NULL && windows[1]==MPI_WIN_NULL) hasWindows=false ;
+    else hasWindows=true ;
+
+      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 ;
+      finalize[0]=(size_t*)bufferHeader[0] +3 ;
+      finalize[1]=(size_t*)bufferHeader[1] +3 ;
+
+      *firstTimeLine[0]=0 ;
+      *firstTimeLine[1]=0 ;
+      *bufferCount[0]=0 ;
+      *bufferCount[1]=0 ;
+      *control[0]=0 ;
+      *control[1]=0 ;
+      *finalize[0]=0 ;
+      *finalize[1]=0 ;
+      winState[0]=false ;
+      winState[1]=false ;
+
+
+    if (hasWindows)
+    {  
+    
+      MPI_Win_attach(windows_[0], bufferHeader[0], bufferSize+headerSize) ;
+      MPI_Win_attach(windows_[1], bufferHeader[1], bufferSize+headerSize) ;
+    
+      MPI_Group group ;
+      int groupSize,groupRank ;
+      MPI_Win_get_group(windows_[0], &group) ;
+      MPI_Group_size(group, &groupSize) ;
+      MPI_Group_rank(group, &groupRank) ;
+      if (groupRank!=clientRank_) ERROR("CClientBuffer::CClientBuffer",<< " ClientRank != groupRank "<<clientRank_<<" "<<groupRank);
+
+      MPI_Win_get_group(windows_[1], &group) ;
+      MPI_Group_size(group, &groupSize) ;
+      MPI_Group_rank(group, &groupRank) ;
+      if (groupRank!=clientRank_) ERROR("CClientBuffer::CClientBuffer",<< " ClientRank != groupRank "<<clientRank_<<" "<<groupRank);
+
+      MPI_Win_lock(MPI_LOCK_EXCLUSIVE, clientRank_, 0, windows_[0]) ;
+      MPI_Win_lock(MPI_LOCK_EXCLUSIVE, clientRank_, 0, windows_[1]) ;
+
+      MPI_Win_unlock(clientRank_, windows_[1]) ;
+      MPI_Win_unlock(clientRank_, windows_[0]) ;
+    } 
     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;
+  }
+
+  MPI_Aint CClientBuffer::getWinAddress(int i)
+  {
+     MPI_Aint address ;
+     
+     if (hasWindows) MPI_Get_address(bufferHeader[i], &address) ;
+     else address=0 ;
+
+     return address ;
   }
 
   CClientBuffer::~CClientBuffer()
   {
-   delete [] buffer[0];
-   delete [] buffer[1];
-   delete retBuffer;
+     //freeWindows() ;
+     if (hasWindows)
+     {
+       MPI_Win_detach(windows_[0],bufferHeader[0]);
+       MPI_Win_detach(windows_[1],bufferHeader[1]);
+       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, clientRank_, 0, windows_[current]) ;
+      long long int lock=1 ;
+      long long int zero=0, one=1 ;
+     
+      MPI_Win_lock(MPI_LOCK_EXCLUSIVE,clientRank_, 0, windows_[current]) ;
+     
+      while(lock!=0)
+      {
+        MPI_Compare_and_swap(&one, &zero, &lock, MPI_LONG_LONG_INT, clientRank_, MPI_Aint_add(getWinAddress(current),2*sizeof(size_t)),
+                             windows_[current]) ;
+        MPI_Win_flush(clientRank_, windows_[current]) ;
+      }
+
+//      info(100)<<"Buffer locked "<<&windows_<<"  "<<current<<endl ;
+      winState[current]=true ;
+    }
+  }
+
+  void CClientBuffer::unlockBuffer(void)
+  {
+    if (hasWindows)
+    {
+      long long int lock=1 ;
+      long long int zero=0, one=1 ;
+
+      MPI_Compare_and_swap(&zero, &one, &lock, MPI_LONG_LONG_INT, clientRank_, MPI_Aint_add(getWinAddress(current),2*sizeof(size_t)),
+                             windows_[current]) ;
+      MPI_Win_unlock(clientRank_, windows_[current]) ;
+
+ //     info(100)<<"Buffer unlocked "<<&windows_<<"  "<<current<<endl ;
+      winState[current]=false ;
+    }
   }
 
@@ -44 +191 @@
   bool CClientBuffer::isBufferFree(StdSize size)
   {
+//    bool loop=true ;
+//    while (loop) 
+//    {
+//      lockBuffer();
+//      if (*control[current]==0) loop=false ; // attemp to read from server ?
+//      else unlockBuffer() ;
+//    }
+  
+    lockBuffer();
     if (size > bufferSize)
       ERROR("bool CClientBuffer::isBufferFree(StdSize size)",
@@ -59 +215 @@
     }
 
-
-    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())
@@ -70 +226 @@
       retBuffer->realloc(buffer[current] + count, size);
       count += size;
-      bufferedEvents++;
+      if (*firstTimeLine[current]==0) *firstTimeLine[current]=timeLine ;
+      *bufferCount[current]=count ;
+/*      info(50)<<"CClientBuffer::getBuffer "<<" clientRank_ "<<clientRank_<<" serverRank "<<serverRank <<" current "<<current
+              <<" size "<<size<<" timeLine "<< timeLine <<" firstTimeLine "<<*firstTimeLine[current]<<" count "<<*bufferCount[current]<<endl ;
+      if (!winState[current]) info(40)<<"CClientBuffer::getBuffer "<<" Windows Not Locked... "<<" clientRank_ "<<clientRank_<<" serverRank "<<serverRank <<" current "<<current
+              <<" size "<<size<<" timeLine "<< timeLine <<" firstTimeLine "<<*firstTimeLine[current]<<" count "<<*bufferCount[current]<<endl ;*/
       return retBuffer;
     }
@@ -81 +242 @@
   }
 
-  bool CClientBuffer::checkBuffer(void)
+  void CClientBuffer::infoBuffer(void)
+  {
+      
+      char checksum=0 ;
+      for(size_t i=0;i<*bufferCount[current];i++) checksum=checksum+buffer[current][i] ;
+ 
+      char checksumFirst=0 ;
+      for(size_t i=5; i<10 && i<*bufferCount[current] ;i++) checksumFirst=checksumFirst+buffer[current][i] ;
+ 
+      char checksumLast=0 ;
+      for(size_t i=(*bufferCount[current]<10)?0:*bufferCount[current]-10; i<*bufferCount[current] ; i++) checksumLast=checksumLast+buffer[current][i] ;
+ 
+      info(45)<<"CClientBuffer::infoBuffer "<<" clientRank_ "<<clientRank_<<" serverRank "<<serverRank <<" current "<<current<<" WinState "<<winState[current]
+              <<" firstTimeLine "<<*firstTimeLine[current]<<" count "<<*bufferCount[current]<<" checksum "<<(int)checksum<<" "
+              <<(int)buffer[current][0]<<" "<<(int)buffer[current][1]<<" "<<(int)buffer[current][2]<<" "<<(int)buffer[current][3]<<" "<<(int)buffer[current][4]<<" "<<(int)buffer[current][5]<<" "
+              <<(int)buffer[current][6]<<" "<<(int)buffer[current][7]<<" "<<(int)buffer[current][8]<<" "<<(int)buffer[current][9]<<" "<<(int)buffer[current][10]<<" "<<(int)buffer[current][11]<<endl ;
+
+  }
+
+  bool CClientBuffer::checkBuffer(bool send)
   {
     MPI_Status status;
@@ -96 +276 @@
     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)
+        if (*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 +304 @@
   bool CClientBuffer::hasPendingRequest(void)
   {
+   
+    lockBuffer() ;
+    count=*bufferCount[current] ;
+    unlockBuffer() ;
+
     return (pending || count > 0);
   }
+
+  bool CClientBuffer::isNotifiedFinalized(void)
+  {
+   
+    bool ret ;
+    lockBuffer() ;
+    ret=*finalize[current] == 1 ? true : false ;
+    unlockBuffer() ;
+
+    return ret;
+  }
+
 }

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

@@ -14 +14 @@
       static size_t maxRequestSize;
 
-      CClientBuffer(MPI_Comm intercomm, int serverRank, StdSize bufferSize, StdSize estimatedMaxEventSize, StdSize maxBufferedEvents);
+      CClientBuffer(MPI_Comm intercomm, vector<MPI_Win>& windows, int clientRank, int serverRank, StdSize bufferSize, StdSize estimatedMaxEventSize);
       ~CClientBuffer();
-
+//      void createWindows(MPI_Comm oneSidedComm) ;
+      void freeWindows(void) ;
+      void lockBuffer(void) ;
+      void unlockBuffer(void) ;
+      
       bool isBufferFree(StdSize size);
-      CBufferOut* getBuffer(StdSize size);
-      bool checkBuffer(void);
+      CBufferOut* getBuffer(size_t timeLine, StdSize size);
+      bool checkBuffer(bool send=false);
       bool hasPendingRequest(void);
       StdSize remain(void);
-
+      MPI_Aint getWinAddress(int numWindows) ;
+      void infoBuffer(void) ;
+      bool isNotifiedFinalized(void) ;
     private:
       char* buffer[2];
-
+      char* bufferHeader[2];
+      size_t* firstTimeLine[2] ;
+      size_t* bufferCount[2] ;
+      size_t* control[2] ;
+      size_t* finalize[2] ;
+      bool winState[2] ;
       int current;
 
       StdSize count;
-      StdSize bufferedEvents;
       StdSize maxEventSize;
-      const StdSize maxBufferedEvents;
       const StdSize bufferSize;
       const StdSize estimatedMaxEventSize;
@@ -37 +46 @@
 
       const int serverRank;
+      const int clientRank_;
       bool pending;
 
@@ -43 +53 @@
       CBufferOut* retBuffer;
       const MPI_Comm interComm;
+      std::vector<MPI_Win> windows_ ;
+      bool hasWindows ;
+      static const int headerSize=4*sizeof(size_t);
   };
 }

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() ;
+  }
 }

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

@@ -12 +12 @@
   {
     public:
-      CServerBuffer(StdSize bufSize) ;
+      CServerBuffer(vector<MPI_Win>& windows, vector<MPI_Aint>& winAddress, int windowsRank, StdSize bufSize) ;
       ~CServerBuffer() ;
 
@@ -18 +18 @@
       void* getBuffer(size_t count) ;
       void freeBuffer(size_t count) ;
-
+      void createWindows(MPI_Comm oneSidedComm) ;
+      bool freeWindows(void) ;
+      bool getBufferFromClient(size_t timeLine, char* & buffer, size_t& count) ;
+      bool isBufferEmpty(void) ;
+      void updateCurrentWindows(void) ;
+      void lockBuffer(void) ;
+      void unlockBuffer(void) ;
+      void notifyClientFinalize(void) ;
     private:
       char* buffer;
@@ -25 +32 @@
       size_t end;
       size_t size;
+      size_t used ;  // count of element occupied
+      std::vector<MPI_Win> windows_ ;
+      std::vector<MPI_Aint> winAddress_ ;
+
+      int currentWindows ;
+      bool hasWindows ;
+      int windowsRank_ ;
   };
 }

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

@@ -24 +24 @@
      : mapBufferSize_(), parentServer(cxtSer), maxBufferedEvents(4)
     {
+      pureOneSided=CXios::getin<bool>("pure_one_sided",false); // pure one sided communication (for test)
+      if (isAttachedModeEnabled()) pureOneSided=false ; // no one sided in attach mode
+      
       context = parent;
       intraComm = intraComm_;
@@ -37 +40 @@
       computeLeader(clientRank, clientSize, serverSize, ranksServerLeader, ranksServerNotLeader);
 
-      timeLine = 0;
+      if (flag) MPI_Intercomm_merge(interComm_,false,&interCommMerged) ;
+      
+      if (!isAttachedModeEnabled())
+      {  
+        windows.resize(serverSize) ;
+        MPI_Comm winComm ;
+        for(int rank=0; rank<serverSize; rank++)
+        {
+          windows[rank].resize(2) ;
+          MPI_Comm_split(interCommMerged, rank, clientRank, &winComm);
+          int myRank ;
+          MPI_Comm_rank(winComm,&myRank);
+          MPI_Win_create_dynamic(MPI_INFO_NULL, winComm, &windows[rank][0]);
+          MPI_Win_create_dynamic(MPI_INFO_NULL, winComm, &windows[rank][1]);
+          MPI_Comm_free(&winComm) ;
+        }
+      }
+
+      MPI_Comm_split(intraComm_,clientRank,clientRank, &commSelf) ;
+
+      timeLine = 1;
     }
 
@@ -116 +139 @@
         list<int> sizes = event.getSizes();
 
-        // We force the getBuffers call to be non-blocking on classical servers
+         // We force the getBuffers call to be non-blocking on classical servers
         list<CBufferOut*> buffList;
-        bool couldBuffer = getBuffers(ranks, sizes, buffList, (!CXios::isClient && (CServer::serverLevel == 0) ));
-//        bool couldBuffer = getBuffers(ranks, sizes, buffList, CXios::isServer );
-
-        if (couldBuffer)
-        {
-          event.send(timeLine, sizes, buffList);
-          info(100)<<"Event "<<timeLine<<" of context "<<context->getId()<<"  sent"<<endl ;
-
-          checkBuffers(ranks);
-
-          if (isAttachedModeEnabled()) // couldBuffer is always true in attached mode
-          {
-            waitEvent(ranks);
-            CContext::setCurrent(context->getId());
-          }
-        }
-        else
-        {
-          tmpBufferedEvent.ranks = ranks;
-          tmpBufferedEvent.sizes = sizes;
-
-          for (list<int>::const_iterator it = sizes.begin(); it != sizes.end(); it++)
-            tmpBufferedEvent.buffers.push_back(new CBufferOut(*it));
-          info(100)<<"DEBUG : temporaly event created : timeline "<<timeLine<<endl ;
-          event.send(timeLine, tmpBufferedEvent.sizes, tmpBufferedEvent.buffers);
-          info(100)<<"Event "<<timeLine<<" of context "<<context->getId()<<"  sent"<<endl ;
+        getBuffers(timeLine, ranks, sizes, buffList) ;
+
+        event.send(timeLine, sizes, buffList);
+       
+        //for (auto itRank = ranks.begin(); itRank != ranks.end(); itRank++) buffers[*itRank]->infoBuffer() ;
+
+        unlockBuffers(ranks) ;
+        info(100)<<"Event "<<timeLine<<" of context "<<context->getId()<<"  sent"<<endl ;
+          
+        checkBuffers(ranks);
+
+        if (isAttachedModeEnabled()) // couldBuffer is always true in attached mode
+        {
+          waitEvent(ranks);
+          CContext::setCurrent(context->getId());
         }
       }
 
       timeLine++;
-    }
-
-    /*!
-     * Send the temporarily buffered event (if any).
-     *
-     * \return true if a temporarily buffered event could be sent, false otherwise 
-     */
-    bool CContextClient::sendTemporarilyBufferedEvent()
-    {
-      bool couldSendTmpBufferedEvent = false;
-
-      if (hasTemporarilyBufferedEvent())
-      {
-        list<CBufferOut*> buffList;
-        if (getBuffers(tmpBufferedEvent.ranks, tmpBufferedEvent.sizes, buffList, true)) // Non-blocking call
-        {
-          list<CBufferOut*>::iterator it, itBuffer;
-
-          for (it = tmpBufferedEvent.buffers.begin(), itBuffer = buffList.begin(); it != tmpBufferedEvent.buffers.end(); it++, itBuffer++)
-            (*itBuffer)->put((char*)(*it)->start(), (*it)->count());
-
-          info(100)<<"DEBUG : temporaly event sent "<<endl ;
-          checkBuffers(tmpBufferedEvent.ranks);
-
-          tmpBufferedEvent.clear();
-
-          couldSendTmpBufferedEvent = true;
-        }
-      }
-
-      return couldSendTmpBufferedEvent;
     }
 
@@ -205 +186 @@
      * it is explicitly requested to be non-blocking.
      *
+     *
+     * \param [in] timeLine time line of the event which will be sent to servers
      * \param [in] serverList list of rank of connected server
      * \param [in] sizeList size of message corresponding to each connection
@@ -211 +194 @@
      * \return whether the already allocated buffers could be used
     */
-    bool CContextClient::getBuffers(const list<int>& serverList, const list<int>& sizeList, list<CBufferOut*>& retBuffers,
+    bool CContextClient::getBuffers(const size_t timeLine, const list<int>& serverList, const list<int>& sizeList, list<CBufferOut*>& retBuffers,
                                     bool nonBlocking /*= false*/)
     {
@@ -236 +219 @@
         areBuffersFree = true;
         for (itBuffer = bufferList.begin(), itSize = sizeList.begin(); itBuffer != bufferList.end(); itBuffer++, itSize++)
+        {
           areBuffersFree &= (*itBuffer)->isBufferFree(*itSize);
+        }
 
         if (!areBuffersFree)
         {
+          for (itBuffer = bufferList.begin(); itBuffer != bufferList.end(); itBuffer++) (*itBuffer)->unlockBuffer();
           checkBuffers();
-          if (CServer::serverLevel == 0)
-            context->server->listen();
-
+          if (CServer::serverLevel == 0)  context->server->listen();
           else if (CServer::serverLevel == 1)
           {
             context->server->listen();
-            for (int i = 0; i < context->serverPrimServer.size(); ++i)
-              context->serverPrimServer[i]->listen();
+            for (int i = 0; i < context->serverPrimServer.size(); ++i)  context->serverPrimServer[i]->listen();
             CServer::contextEventLoop(false) ; // avoid dead-lock at finalize...
           }
 
-          else if (CServer::serverLevel == 2)
-            context->server->listen();
+          else if (CServer::serverLevel == 2) context->server->listen();
 
         }
       } while (!areBuffersFree && !nonBlocking);
-
       CTimer::get("Blocking time").suspend();
 
@@ -263 +244 @@
       {
         for (itBuffer = bufferList.begin(), itSize = sizeList.begin(); itBuffer != bufferList.end(); itBuffer++, itSize++)
-          retBuffers.push_back((*itBuffer)->getBuffer(*itSize));
-      }
-
+          retBuffers.push_back((*itBuffer)->getBuffer(timeLine, *itSize));
+      }
       return areBuffersFree;
    }
@@ -281 +261 @@
         maxEventSizes[rank] = CXios::minBufferSize;
       }
-      CClientBuffer* buffer = buffers[rank] = new CClientBuffer(interComm, rank, mapBufferSize_[rank], maxEventSizes[rank], maxBufferedEvents);
+      
+      vector<MPI_Win> Wins(2,MPI_WIN_NULL) ;
+      if (!isAttachedModeEnabled()) Wins=windows[rank] ;
+  
+      CClientBuffer* buffer = buffers[rank] = new CClientBuffer(interComm, Wins, clientRank, rank, mapBufferSize_[rank], maxEventSizes[rank]);
       // Notify the server
-      CBufferOut* bufOut = buffer->getBuffer(sizeof(StdSize));
-      bufOut->put(mapBufferSize_[rank]); // Stupid C++
-      buffer->checkBuffer();
+      CBufferOut* bufOut = buffer->getBuffer(0, 3*sizeof(MPI_Aint));
+      MPI_Aint sendBuff[3] ;
+      sendBuff[0]=mapBufferSize_[rank]; // Stupid C++
+      sendBuff[1]=buffers[rank]->getWinAddress(0); 
+      sendBuff[2]=buffers[rank]->getWinAddress(1); 
+      info(100)<<"CContextClient::newBuffer : rank "<<rank<<" winAdress[0] "<<buffers[rank]->getWinAddress(0)<<" winAdress[1] "<<buffers[rank]->getWinAddress(1)<<endl;
+      bufOut->put(sendBuff, 3); // Stupid C++
+      buffer->checkBuffer(true);
+
+/*
+      if (!isAttachedModeEnabled()) // create windows only in server mode
+      {
+        MPI_Comm OneSidedInterComm, oneSidedComm ;
+        MPI_Intercomm_create(commSelf, 0, interCommMerged, clientSize+rank, 0, &OneSidedInterComm );
+        MPI_Intercomm_merge(OneSidedInterComm,false,&oneSidedComm);
+        buffer->createWindows(oneSidedComm) ;
+      }
+ */      
    }
 
@@ -297 +296 @@
       bool pending = false;
       for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
-        pending |= itBuff->second->checkBuffer();
+        pending |= itBuff->second->checkBuffer(!pureOneSided);
       return pending;
    }
@@ -307 +306 @@
       for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
       {
-          delete itBuff->second;
+         delete itBuff->second;
       }
       buffers.clear();
-   }
-
+
+/* don't know when release windows
+
+      if (!isAttachedModeEnabled())
+      {  
+        for(int rank=0; rank<serverSize; rank++)
+        {
+          MPI_Win_free(&windows[rank][0]);
+          MPI_Win_free(&windows[rank][1]);
+        }
+      } 
+   }
+*/
+      
+  /*!
+   Lock the buffers for one sided communications
+   \param [in] ranks list rank of server to which client connects to
+   */
+   void CContextClient::lockBuffers(list<int>& ranks)
+   {
+      list<int>::iterator it;
+      for (it = ranks.begin(); it != ranks.end(); it++) buffers[*it]->lockBuffer();
+   }
+
+  /*!
+   Unlock the buffers for one sided communications
+   \param [in] ranks list rank of server to which client connects to
+   */
+   void CContextClient::unlockBuffers(list<int>& ranks)
+   {
+      list<int>::iterator it;
+      for (it = ranks.begin(); it != ranks.end(); it++) buffers[*it]->unlockBuffer();
+   }
+      
    /*!
    Verify state of buffers corresponding to a connection
@@ -321 +352 @@
       list<int>::iterator it;
       bool pending = false;
-      for (it = ranks.begin(); it != ranks.end(); it++) pending |= buffers[*it]->checkBuffer();
+      for (it = ranks.begin(); it != ranks.end(); it++) pending |= buffers[*it]->checkBuffer(!pureOneSided);
       return pending;
    }
@@ -335 +366 @@
      mapBufferSize_ = mapSize;
      maxEventSizes = maxEventSize;
-
-     // Compute the maximum number of events that can be safely buffered.
-     double minBufferSizeEventSizeRatio = std::numeric_limits<double>::max();
-     for (std::map<int,StdSize>::const_iterator it = mapSize.begin(), ite = mapSize.end(); it != ite; ++it)
-     {
-       double ratio = double(it->second) / maxEventSizes[it->first];
-       if (ratio < minBufferSizeEventSizeRatio) minBufferSizeEventSizeRatio = ratio;
-     }
-     MPI_Allreduce(MPI_IN_PLACE, &minBufferSizeEventSizeRatio, 1, MPI_DOUBLE, MPI_MIN, intraComm);
-
-     if (minBufferSizeEventSizeRatio < 1.0)
-     {
-       ERROR("void CContextClient::setBufferSize(const std::map<int,StdSize>& mapSize, const std::map<int,StdSize>& maxEventSize)",
-             << "The buffer sizes and the maximum events sizes are incoherent.");
-     }
-     else if (minBufferSizeEventSizeRatio == std::numeric_limits<double>::max())
-       minBufferSizeEventSizeRatio = 1.0; // In this case, maxBufferedEvents will never be used but we want to avoid any floating point exception
-
-     maxBufferedEvents = size_t(2 * minBufferSizeEventSizeRatio) // there is room for two local buffers on the server
-                          + size_t(minBufferSizeEventSizeRatio)  // one local buffer can always be fully used
-                          + 1;                                   // the other local buffer might contain only one event
    }
 
@@ -410 +420 @@
   {
     map<int,CClientBuffer*>::iterator itBuff;
+    std::list<int>::iterator ItServerLeader; 
+    
     bool stop = false;
 
-    CTimer::get("Blocking time").resume();
-    while (hasTemporarilyBufferedEvent())
-    {
-      checkBuffers();
-      sendTemporarilyBufferedEvent();
-    }
-    CTimer::get("Blocking time").suspend();
-
+    int* nbServerConnectionLocal  = new int[serverSize] ;
+    int* nbServerConnectionGlobal  = new int[serverSize] ;
+    for(int i=0;i<serverSize;++i) nbServerConnectionLocal[i]=0 ;
+    for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)  nbServerConnectionLocal[itBuff->first]=1 ;
+    for (ItServerLeader = ranksServerLeader.begin(); ItServerLeader != ranksServerLeader.end(); ItServerLeader++)  nbServerConnectionLocal[*ItServerLeader]=1 ;
+    
+    MPI_Allreduce(nbServerConnectionLocal, nbServerConnectionGlobal, serverSize, MPI_INT, MPI_SUM, intraComm);
+    
     CEventClient event(CContext::GetType(), CContext::EVENT_ID_CONTEXT_FINALIZE);
+    CMessage msg;
+
+    for (int i=0;i<serverSize;++i) if (nbServerConnectionLocal[i]==1) event.push(i, nbServerConnectionGlobal[i], msg) ;
+    sendEvent(event);
+
+    delete[] nbServerConnectionLocal ;
+    delete[] nbServerConnectionGlobal ;
+/*    
     if (isServerLeader())
     {
@@ -433 +453 @@
     }
     else sendEvent(event);
+*/
 
     CTimer::get("Blocking time").resume();
-//    while (!stop)
-    {
-      checkBuffers();
-      if (hasTemporarilyBufferedEvent())
-        sendTemporarilyBufferedEvent();
-
-      stop = true;
-//      for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++) stop &= !itBuff->second->hasPendingRequest();
-    }
+    checkBuffers();
     CTimer::get("Blocking time").suspend();
 
@@ -472 +485 @@
     return pending;
   }
-
+  
+  bool CContextClient::isNotifiedFinalized(void)
+  {
+    if (isAttachedModeEnabled()) return true ;
+
+    bool finalized = true;
+    map<int,CClientBuffer*>::iterator itBuff;
+    for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
+      finalized &= itBuff->second->isNotifiedFinalized();
+    return finalized;
+  }
 
 }

XIOS/dev/dev_ym/XIOS_ONE_SIDED/src/context_client.hpp

@@ -31 +31 @@
       // Send event to server
       void sendEvent(CEventClient& event);
-      bool sendTemporarilyBufferedEvent();
       void waitEvent(list<int>& ranks);
 
       // Functions to set/get buffers
-      bool getBuffers(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);
@@ -48 +47 @@
 
       bool isAttachedModeEnabled() const;
-      bool hasTemporarilyBufferedEvent() const { return !tmpBufferedEvent.isEmpty(); };
 
       static void computeLeader(int clientRank, int clientSize, int serverSize,
@@ -56 +54 @@
       // Close and finalize context client
 //      void closeContext(void);  Never been implemented.
+      bool isNotifiedFinalized(void) ;
       void finalize(void);
 
@@ -71 +70 @@
       int serverSize; //!< Size of server group
 
-      MPI_Comm interComm; //!< Communicator of server group
+      MPI_Comm interComm; //!< Communicator of server group (interCommunicator)
+
+      MPI_Comm interCommMerged; //!< Communicator of the client group + server group (intraCommunicator) needed for one sided communication.
 
       MPI_Comm intraComm; //!< Communicator of client group
 
+      MPI_Comm commSelf; //!< Communicator of the client alone. Needed to create a new communicator between 1 proc client and 1 proc server for one sided communication
+
       map<int,CClientBuffer*> buffers; //!< Buffers for connection to servers
 
+      bool pureOneSided ; //!< if true, client will communicated with servers only trough one sided communication. Otherwise the hybrid mode P2P /One sided is used.
+
     private:
+      void lockBuffers(list<int>& ranks) ;
+      void unlockBuffers(list<int>& ranks) ;
+      
       //! Mapping of server and buffer size for each connection to server
       std::map<int,StdSize> mapBufferSize_;
@@ -84 +92 @@
       //! Maximum number of events that can be buffered
       StdSize maxBufferedEvents;
-
-      struct {
-        std::list<int> ranks, sizes;
-        std::list<CBufferOut*> buffers;
-
-        bool isEmpty() const { return ranks.empty(); };
-        void clear() {
-          ranks.clear();
-          sizes.clear();
-
-          for (std::list<CBufferOut*>::iterator it = buffers.begin(); it != buffers.end(); it++)
-            delete *it;
-
-          buffers.clear();
-        };
-      } tmpBufferedEvent; //! Event temporarily buffered (used only on the server)
 
       //! Context for server (Only used in attached mode)
@@ -110 +102 @@
       std::list<int> ranksServerNotLeader;
 
+      std::vector<std::vector<MPI_Win> >windows ; //! one sided mpi windows to expose client buffers to servers == windows[nbServers][2]
+
+
   };
 }

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

@@ -33 +33 @@
     int flag;
     MPI_Comm_test_inter(interComm,&flag);
+
+    if (flag) attachedMode=false ;
+    else  attachedMode=true ;
+    
     if (flag) MPI_Comm_remote_size(interComm,&commSize);
     else  MPI_Comm_size(interComm,&commSize);
 
-    currentTimeLine=0;
+     
+    currentTimeLine=1;
     scheduled=false;
     finished=false;
@@ -44 +49 @@
     else
       hashId=hashString(context->getId());
-  }
-
+
+    if (!isAttachedModeEnabled())
+    {
+      MPI_Intercomm_merge(interComm_,true,&interCommMerged) ;
+// create windows for one sided comm
+      int interCommMergedRank;
+      MPI_Comm winComm ;
+      MPI_Comm_rank(intraComm, &interCommMergedRank);
+      windows.resize(2) ;
+      for(int rank=commSize; rank<commSize+intraCommSize; rank++)
+      {
+        if (rank==commSize+interCommMergedRank) 
+        {
+          MPI_Comm_split(interCommMerged, interCommMergedRank, rank, &winComm);
+          int myRank ;
+          MPI_Comm_rank(winComm,&myRank);
+          MPI_Win_create_dynamic(MPI_INFO_NULL, winComm, &windows[0]);
+          MPI_Win_create_dynamic(MPI_INFO_NULL, winComm, &windows[1]);      
+        }
+        else MPI_Comm_split(interCommMerged, interCommMergedRank, rank, &winComm);
+        MPI_Comm_free(&winComm) ;
+      }
+    }
+    else 
+    {
+      windows.resize(2) ;
+      windows[0]=MPI_WIN_NULL ;
+      windows[1]=MPI_WIN_NULL ;
+    }
+
+
+    
+    MPI_Comm_split(intraComm_,intraCommRank,intraCommRank, &commSelf) ;
+    itLastTimeLine=lastTimeLine.begin() ;
+
+    pureOneSided=CXios::getin<bool>("pure_one_sided",false); // pure one sided communication (for test)
+    if (isAttachedModeEnabled()) pureOneSided=false ; // no one sided in attach mode
+      
+  }
+
+//! Attached mode is used ?
+//! \return true if attached mode is used, false otherwise
+  bool CContextServer::isAttachedModeEnabled() const
+  {
+    return attachedMode ;
+  }
+  
   void CContextServer::setPendingEvent(void)
   {
@@ -65 +115 @@
     listen();
     checkPendingRequest();
-    if (enableEventsProcessing)
-      processEvents();
+    if (enableEventsProcessing)  processEvents();
     return finished;
   }
@@ -117 +166 @@
     if (it==buffers.end()) // Receive the buffer size and allocate the buffer
     {
-       StdSize buffSize = 0;
-       MPI_Recv(&buffSize, 1, MPI_LONG, rank, 20, interComm, &status);
+       MPI_Aint recvBuff[3] ;
+       MPI_Recv(recvBuff, 3, MPI_AINT, rank, 20, interComm, &status);
+       StdSize buffSize = recvBuff[0];
+       vector<MPI_Aint> winAdress(2) ;
+       winAdress[0]=recvBuff[1] ; winAdress[1]=recvBuff[2] ;
        mapBufferSize_.insert(std::make_pair(rank, buffSize));
-       it=(buffers.insert(pair<int,CServerBuffer*>(rank,new CServerBuffer(buffSize)))).first;
+       it=(buffers.insert(pair<int,CServerBuffer*>(rank,new CServerBuffer(windows, winAdress, rank, buffSize)))).first;
+      /*
+       if (!isAttachedModeEnabled())
+       {
+         MPI_Comm OneSidedInterComm, oneSidedComm ;
+         MPI_Intercomm_create(commSelf, 0, interCommMerged, rank, 0, &OneSidedInterComm );
+         MPI_Intercomm_merge(OneSidedInterComm,true,&oneSidedComm);
+         buffers[rank]->createWindows(oneSidedComm) ;
+       }
+       */
+       lastTimeLine[rank]=0 ;
+       itLastTimeLine=lastTimeLine.begin() ;
+
        return true;
     }
@@ -157 +221 @@
       if (flag==true)
       {
+        buffers[rank]->updateCurrentWindows() ;
         recvRequest.push_back(rank);
         MPI_Get_count(&status,MPI_CHAR,&count);
@@ -170 +235 @@
   }
 
+  void CContextServer::getBufferFromClient(size_t timeLine)
+  {
+    if (!isAttachedModeEnabled()) // one sided desactivated in attached mode
+    {  
+      int rank ;
+      char *buffer ;
+      size_t count ; 
+
+      if (itLastTimeLine==lastTimeLine.end()) itLastTimeLine=lastTimeLine.begin() ;
+      for(;itLastTimeLine!=lastTimeLine.end();++itLastTimeLine)
+      {
+        rank=itLastTimeLine->first ;
+        if (itLastTimeLine->second < timeLine &&  pendingRequest.count(rank)==0)
+        {
+          if (buffers[rank]->getBufferFromClient(timeLine, buffer, count))
+          {
+            processRequest(rank, buffer, count);
+            break ;
+          }
+        }
+      }
+    }
+  }
+         
+       
   void CContextServer::processRequest(int rank, char* buff,int count)
   {
@@ -176 +266 @@
     char* startBuffer,endBuffer;
     int size, offset;
-    size_t timeLine;
+    size_t timeLine=0;
     map<size_t,CEventServer*>::iterator it;
 
+    
     CTimer::get("Process request").resume();
     while(count>0)
@@ -185 +276 @@
       CBufferIn newBuffer(startBuffer,buffer.remain());
       newBuffer>>size>>timeLine;
-
       it=events.find(timeLine);
       if (it==events.end()) it=events.insert(pair<int,CEventServer*>(timeLine,new CEventServer)).first;
@@ -193 +283 @@
       count=buffer.remain();
     }
+
+    if (timeLine>0) lastTimeLine[rank]=timeLine ;
+    
     CTimer::get("Process request").suspend();
   }
@@ -230 +323 @@
         }
       }
-    }
+      else getBufferFromClient(currentTimeLine) ;
+    }
+    else if (pureOneSided) getBufferFromClient(currentTimeLine) ; // if pure one sided check buffer even if no event recorded at current time line
   }
 
@@ -237 +332 @@
     map<int,CServerBuffer*>::iterator it;
     for(it=buffers.begin();it!=buffers.end();++it) delete it->second;
+  }
+
+  void CContextServer::releaseBuffers()
+  {
+    map<int,CServerBuffer*>::iterator it;
+    bool out ;
+    do
+    {
+      out=true ;
+      for(it=buffers.begin();it!=buffers.end();++it)
+      {
+//        out = out && it->second->freeWindows() ;
+
+      }
+    } while (! out) ; 
+  }
+
+  void CContextServer::notifyClientsFinalize(void)
+  {
+    for(auto it=buffers.begin();it!=buffers.end();++it)
+    {
+      it->second->notifyClientFinalize() ;
+    }
   }
 
@@ -254 +372 @@
       finished=true;
       info(20)<<" CContextServer: Receive context <"<<context->getId()<<"> finalize."<<endl;
+//      releaseBuffers() ;
+      notifyClientsFinalize() ;
       context->finalize();
+
+/* don't know where release windows
+      MPI_Win_free(&windows[0]) ;
+      MPI_Win_free(&windows[1]) ;
+*/     
       std::map<int, StdSize>::const_iterator itbMap = mapBufferSize_.begin(),
                            iteMap = mapBufferSize_.end(), itMap;

XIOS/dev/dev_ym/XIOS_ONE_SIDED/src/context_server.hpp

@@ -19 +19 @@
     bool listenPendingRequest(MPI_Status& status) ;
     void checkPendingRequest(void) ;
+    void getBufferFromClient(size_t timeLine) ;
     void processRequest(int rank, char* buff,int count) ;
     void processEvents(void) ;
@@ -25 +26 @@
     void setPendingEvent(void) ;
     bool hasPendingEvent(void) ;
-
+    bool isAttachedModeEnabled() const;
+    void releaseBuffers(void) ;
+    void notifyClientsFinalize(void) ;
+    
     MPI_Comm intraComm ;
     int intraCommSize ;
@@ -33 +37 @@
     int commSize ;
 
+    MPI_Comm interCommMerged; //!< Communicator of the client group + server group (intraCommunicator) needed for one sided communication.
+
+    MPI_Comm commSelf; //!< Communicator of the server alone. Needed to create a new communicator between 1 proc client and 1 proc server for one sided communication
+
     map<int,CServerBuffer*> buffers ;
+    map<int,size_t> lastTimeLine ; //!< last event time line for a processed request
+    map<int,size_t>::iterator itLastTimeLine ; //!< iterator on lastTimeLine
     map<int,MPI_Request> pendingRequest ;
     map<int,char*> bufferRequest ;
@@ -44 +54 @@
     bool pendingEvent ;
     bool scheduled  ;    /*!< event of current timeline is alreading scheduled ? */
+    bool attachedMode ;  //! true if attached mode is enabled otherwise false
+    bool pureOneSided ; //!< if true, client will communicated with servers only trough one sided communication. Otherwise the hybrid mode P2P /One sided is used.
+         
     size_t hashId ;
 
@@ -50 +63 @@
     private:
       std::map<int, StdSize> mapBufferSize_;
+      vector<MPI_Win> windows ; //! one sided mpi windows to expose client buffers to servers ; No memory will be attached on server side.
+
   } ;
 

XIOS/dev/dev_ym/XIOS_ONE_SIDED/src/node/context.cpp

@@ -429 +429 @@
     {
       client->checkBuffers();
-      bool hasTmpBufferedEvent = client->hasTemporarilyBufferedEvent();
-      if (hasTmpBufferedEvent)
-        hasTmpBufferedEvent = !client->sendTemporarilyBufferedEvent();
-      // Don't process events if there is a temporarily buffered event
-      return server->eventLoop(!hasTmpBufferedEvent || !enableEventsProcessing);
+      return server->eventLoop(true);
     }
     else if (CServer::serverLevel == 1)
     {
-      if (!finalized)
-        client->checkBuffers();
+      if (!finalized) client->checkBuffers();
       bool serverFinished = true;
-      if (!finalized)
-        serverFinished = server->eventLoop(enableEventsProcessing);
+      if (!finalized) serverFinished = server->eventLoop(enableEventsProcessing);
       bool serverPrimFinished = true;
       for (int i = 0; i < clientPrimServer.size(); ++i)
       {
-        if (!finalized)
-          clientPrimServer[i]->checkBuffers();
-        if (!finalized)
-          serverPrimFinished *= serverPrimServer[i]->eventLoop(enableEventsProcessing);
+        if (!finalized) clientPrimServer[i]->checkBuffers();
+        if (!finalized) serverPrimFinished *= serverPrimServer[i]->eventLoop(enableEventsProcessing);
       }
       return ( serverFinished && serverPrimFinished);
@@ -484 +476 @@
          ++countChildCtx_;
 
+         info(100)<<"DEBUG: context "<<getId()<<" Send client finalize"<<endl ;
          client->finalize();
-         while (client->havePendingRequests())
-            client->checkBuffers();
-
+         info(100)<<"DEBUG: context "<<getId()<<" Client finalize sent"<<endl ;
+         while (client->havePendingRequests()) client->checkBuffers();
+         
+         info(100)<<"DEBUG: context "<<getId()<<" no pending request ok"<<endl ;
          while (!server->hasFinished())
            server->eventLoop();
+        info(100)<<"DEBUG: context "<<getId()<<" server has finished"<<endl ;
+        
+        bool notifiedFinalized=false ;
+        do
+        {
+          notifiedFinalized=client->isNotifiedFinalized() ;
+        } while (!notifiedFinalized) ;
+        client->releaseBuffers();
 
          if (hasServer) // Mode attache
@@ -499 +501 @@
 
          //! Deallocate client buffers
-         client->releaseBuffers();
-
+//         client->releaseBuffers();
+        info(100)<<"DEBUG: context "<<getId()<<" release client ok"<<endl ;
          //! Free internally allocated communicators
          for (std::list<MPI_Comm>::iterator it = comms.begin(); it != comms.end(); ++it)
@@ -515 +517 @@
        if (countChildCtx_ == 0)
          for (int i = 0; i < clientPrimServer.size(); ++i)
+         {
            clientPrimServer[i]->finalize();
+           bool bufferReleased;
+           do
+           {
+             clientPrimServer[i]->checkBuffers();
+             bufferReleased = !clientPrimServer[i]->havePendingRequests();
+           } while (!bufferReleased);
+           
+           bool notifiedFinalized=false ;
+           do
+           {
+//             clientPrimServer[i]->checkBuffers();
+             notifiedFinalized=clientPrimServer[i]->isNotifiedFinalized() ;
+           } while (!notifiedFinalized) ;
+           clientPrimServer[i]->releaseBuffers();
+         }
+           
 
        // (Last) context finalized message received
@@ -521 +540 @@
        {
          // Blocking send of context finalize message to its client (e.g. primary server or model)
-         info(100)<<"DEBUG: context "<<getId()<<" Send client finalize<<"<<endl ;
+         info(100)<<"DEBUG: context "<<getId()<<" Send client finalize"<<endl ;
          client->finalize();
+         info(100)<<"DEBUG: context "<<getId()<<" Client finalize sent"<<endl ;
          bool bufferReleased;
          do
@@ -529 +549 @@
            bufferReleased = !client->havePendingRequests();
          } while (!bufferReleased);
+         
+         bool notifiedFinalized=false ;
+         do
+         {
+  //         client->checkBuffers();
+           notifiedFinalized=client->isNotifiedFinalized() ;
+         } while (!notifiedFinalized) ;
+         client->releaseBuffers();
+         
          finalized = true;
-
+         info(100)<<"DEBUG: context "<<getId()<<" bufferRelease OK"<<endl ;
+         
          closeAllFile(); // Just move to here to make sure that server-level 1 can close files
          if (hasServer && !hasClient)
@@ -539 +569 @@
 
          //! Deallocate client buffers
-         client->releaseBuffers();
+//         client->releaseBuffers();
+         info(100)<<"DEBUG: context "<<getId()<<" client release"<<endl ;
+
+/*         
          for (int i = 0; i < clientPrimServer.size(); ++i)
            clientPrimServer[i]->releaseBuffers();
-
+*/
          //! Free internally allocated communicators
          for (std::list<MPI_Comm>::iterator it = comms.begin(); it != comms.end(); ++it)