Changeset 1547 for XIOS

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

Location:

XIOS/dev/dev_ym/XIOS_ONE_SIDED/src

Files:

• buffer_client.cpp (modified) (7 diffs)
• buffer_client.hpp (modified) (3 diffs)
• buffer_server.cpp (modified) (6 diffs)
• buffer_server.hpp (modified) (2 diffs)
• context_client.cpp (modified) (16 diffs)
• context_client.hpp (modified) (4 diffs)
• context_server.cpp (modified) (12 diffs)
• context_server.hpp (modified) (4 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);
   }
+
+
 }

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, 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);
@@ -25 +29 @@
     private:
       char* buffer[2];
-
+      char* bufferHeader[2];
+      size_t* firstTimeLine[2] ;
+      size_t* bufferCount[2] ;
+      size_t* control[2] ;
+      bool winState[2] ;
       int current;
 
       StdSize count;
-      StdSize bufferedEvents;
       StdSize maxEventSize;
-      const StdSize maxBufferedEvents;
       const StdSize bufferSize;
       const StdSize estimatedMaxEventSize;
@@ -43 +49 @@
       CBufferOut* retBuffer;
       const MPI_Comm interComm;
+      MPI_Win windows[2] ;
+      bool hasWindows ;
+      static const int headerSize=3*sizeof(size_t);
   };
 }

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

@@ -7 +7 @@
 {
 
-  CServerBuffer::CServerBuffer(StdSize buffSize)
+  CServerBuffer::CServerBuffer(StdSize buffSize) : hasWindows(false)
   {
     size = 3 * buffSize;
@@ -13 +13 @@
     current = 1;
     end = size;
+    used=0 ;
     buffer = new char[size]; // use MPI_ALLOC_MEM later?
+    currentWindows=0 ;
   }
 
@@ -21 +23 @@
   }
 
+  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 ;
+    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, 0 , 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 +109 @@
   }
 
+  bool CServerBuffer::isBufferEmpty(void)
+  {
+    if (used==0) return true ;
+    else return false;
+  }
 
   void* CServerBuffer::getBuffer(size_t count)
@@ -128 +170 @@
     }
 
+    used+=count ;
     return ret ;
   }
@@ -167 +210 @@
       }
     }
-  }
-
+    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_Win_lock(MPI_LOCK_EXCLUSIVE,0,0,windows[currentWindows]) ;
+
+    MPI_Get(&clientTimeline, 1, MPI_LONG_LONG_INT, 0 , 0, 1, MPI_LONG_LONG_INT,windows[currentWindows]) ;
+    MPI_Get(&clientCount, 1, MPI_LONG_LONG_INT, 0 , 1*sizeof(size_t), 1, MPI_LONG_LONG_INT,windows[currentWindows]) ;
+    control=1 ;
+    MPI_Put(&control, 1, MPI_LONG_LONG_INT, 0 , 2*sizeof(size_t), 1, MPI_LONG_LONG_INT,windows[currentWindows]) ;
+   
+    MPI_Win_unlock(0,windows[currentWindows]) ;
+
+    if (timeLine==clientTimeline)
+    {
+
+      MPI_Win_lock(MPI_LOCK_EXCLUSIVE,0,0,windows[currentWindows]) ;
+      buffer=(char*)getBuffer(clientCount) ;
+      count=clientCount ;
+      MPI_Get(buffer, clientCount, MPI_CHAR, 0, 3*sizeof(size_t) , clientCount, MPI_CHAR, windows[currentWindows]) ;
+      clientTimeline = 0 ;
+      clientCount = 0 ;
+      control=0 ;
+      MPI_Put(&header[0], 3, MPI_LONG_LONG_INT, 0, 0 , 3, MPI_LONG_LONG_INT,windows[currentWindows]) ;
+  
+      MPI_Win_unlock(0,windows[currentWindows]) ;
+      ok=true ;
+    }
+    else
+    {
+      MPI_Win_lock(MPI_LOCK_EXCLUSIVE,0,0,windows[currentWindows]) ;
+      control=0 ;
+      MPI_Put(&control, 1, MPI_LONG_LONG_INT, 0 , 2*sizeof(size_t), 1, MPI_LONG_LONG_INT,windows[currentWindows]) ;
+      MPI_Win_unlock(0,windows[currentWindows]) ;
+    }
+
+    if (ok) return true ;
+
+    return false ;
+  }
+    
+   
 }

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

@@ -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) ;
     private:
       char* buffer;
@@ -25 +29 @@
       size_t end;
       size_t size;
+      size_t used ;  // count of element occupied
+      MPI_Win windows[2] ;
+      int currentWindows ;
+      bool hasWindows ;
   };
 }

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) ;
+
+      MPI_Comm_split(intraComm_,clientRank,clientRank, &commSelf) ;
+
+      timeLine = 1;
     }
 
@@ -116 +123 @@
         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);
-
-          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);
+//        bool couldBuffer = getBuffers(timeLine, ranks, sizes, buffList, (!CXios::isClient && (CServer::serverLevel == 0) ));
+        getBuffers(timeLine, ranks, sizes, buffList) ;
+
+        event.send(timeLine, sizes, buffList);
+        unlockBuffers(ranks) ;
+          
+        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;
     }
 
@@ -203 +167 @@
      * 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
@@ -209 +175 @@
      * \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*/)
     {
@@ -229 +195 @@
       }
 
+      if (CXios::isServer) info(100)<<" getBuffers : entering loop"<<endl ;
       CTimer::get("Blocking time").resume();
       do
@@ -234 +201 @@
         areBuffersFree = true;
         for (itBuffer = bufferList.begin(), itSize = sizeList.begin(); itBuffer != bufferList.end(); itBuffer++, itSize++)
+        {
           areBuffersFree &= (*itBuffer)->isBufferFree(*itSize);
-
+        }
+
+        if (CXios::isServer) info(100)<<" getBuffers : areBuffersFree ? "<<areBuffersFree<<endl ; ;
         if (!areBuffersFree)
         {
+          for (itBuffer = bufferList.begin(); itBuffer != bufferList.end(); itBuffer++) (*itBuffer)->unlockBuffer();
+          if (CXios::isServer) info(100)<<" getBuffers : buffers unlocked "<<endl ;
           checkBuffers();
-          if (CServer::serverLevel == 0)
-            context->server->listen();
-
+          if (CXios::isServer) info(100)<<" getBuffers : buffers checked "<<endl ;
+          if (CServer::serverLevel == 0)  context->server->listen();
+          if (CXios::isServer) info(100)<<" getBuffers : server listened... "<<endl ;
           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);
-
+      if (CXios::isServer) info(100)<<" getBuffers : out of loop"<<endl ;
       CTimer::get("Blocking time").suspend();
 
@@ -261 +231 @@
       {
         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));
+      }
+      if (CXios::isServer) info(100)<<" getBuffers : message pushed"<<endl ;
       return areBuffersFree;
    }
@@ -279 +249 @@
         maxEventSizes[rank] = CXios::minBufferSize;
       }
-      CClientBuffer* buffer = buffers[rank] = new CClientBuffer(interComm, rank, mapBufferSize_[rank], maxEventSizes[rank], maxBufferedEvents);
+      CClientBuffer* buffer = buffers[rank] = new CClientBuffer(interComm, rank, mapBufferSize_[rank], maxEventSizes[rank]);
       // Notify the server
-      CBufferOut* bufOut = buffer->getBuffer(sizeof(StdSize));
+      CBufferOut* bufOut = buffer->getBuffer(0, sizeof(StdSize));
       bufOut->put(mapBufferSize_[rank]); // Stupid C++
-      buffer->checkBuffer();
+      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);
+        info(100)<<"DEBUG: before creating windows (client)"<<endl ;
+        buffer->createWindows(oneSidedComm) ;
+        info(100)<<"DEBUG: after creating windows (client)"<<endl ;
+      }
+       
    }
 
@@ -295 +276 @@
       bool pending = false;
       for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
-        pending |= itBuff->second->checkBuffer();
+        pending |= itBuff->second->checkBuffer(!pureOneSided);
       return pending;
    }
@@ -305 +286 @@
       for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
       {
+//          CEventClient event(CContext::GetType(), CContext::EVENT_ID_CLOSE_P2P_CHANNEL);
+//          CMessage msg;
+//          event.push(itBuff->first, 1, msg);
+//          timeLine = std::numeric_limits<size_t>::max() ;
+//          sendEvent(event);
+//          while (itBuff->second->checkBuffer(!pureOneSided));
           delete itBuff->second;
       }
@@ -310 +297 @@
    }
 
+  /*!
+   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
@@ -319 +326 @@
       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;
    }
@@ -333 +340 @@
      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
    }
 
@@ -408 +394 @@
   {
     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())
     {
@@ -431 +427 @@
     }
     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();
 

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,
@@ -71 +69 @@
       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 +91 @@
       //! 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)

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) ;
+    
+    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)
   {
@@ -63 +84 @@
   bool CContextServer::eventLoop(bool enableEventsProcessing /*= true*/)
   {
+//    info(100)<<"CContextServer::eventLoop : listen"<<endl ;
     listen();
+//    info(100)<<"CContextServer::eventLoop : checkPendingRequest"<<endl ;
     checkPendingRequest();
-    if (enableEventsProcessing)
-      processEvents();
+//    info(100)<<"CContextServer::eventLoop : process events"<<endl ;
+    if (enableEventsProcessing)  processEvents();
+//    info(100)<<"CContextServer::eventLoop : finished "<<finished<<endl ;
     return finished;
   }
@@ -121 +145 @@
        mapBufferSize_.insert(std::make_pair(rank, buffSize));
        it=(buffers.insert(pair<int,CServerBuffer*>(rank,new CServerBuffer(buffSize)))).first;
+       if (!isAttachedModeEnabled())
+       {
+         MPI_Comm OneSidedInterComm, oneSidedComm ;
+         MPI_Intercomm_create(commSelf, 0, interCommMerged, rank, 0, &OneSidedInterComm );
+         MPI_Intercomm_merge(OneSidedInterComm,true,&oneSidedComm);
+         info(100)<<"DEBUG: before creating windows (server)"<<endl ;
+         buffers[rank]->createWindows(oneSidedComm) ;
+         info(100)<<"DEBUG: before creating windows (server)"<<endl ;
+       }
+       lastTimeLine[rank]=0 ;
+       itLastTimeLine=lastTimeLine.begin() ;
+
        return true;
     }
@@ -157 +193 @@
       if (flag==true)
       {
+        buffers[rank]->updateCurrentWindows() ;
         recvRequest.push_back(rank);
         MPI_Get_count(&status,MPI_CHAR,&count);
@@ -170 +207 @@
   }
 
+  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))
+          {
+            info(100)<<"get buffer from client : timeLine "<<timeLine<<endl ;
+            processRequest(rank, buffer, count);
+            break ;
+          }
+        }
+      }
+    }
+  }
+         
+       
   void CContextServer::processRequest(int rank, char* buff,int count)
   {
@@ -176 +239 @@
     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 +249 @@
       CBufferIn newBuffer(startBuffer,buffer.remain());
       newBuffer>>size>>timeLine;
-
+      info(100)<<"new event :   timeLine : "<<timeLine<<"  size : "<<size<<endl ;
       it=events.find(timeLine);
       if (it==events.end()) it=events.insert(pair<int,CEventServer*>(timeLine,new CEventServer)).first;
@@ -193 +257 @@
       count=buffer.remain();
     }
+
+    if (timeLine>0) lastTimeLine[rank]=timeLine ;
+    
     CTimer::get("Process request").suspend();
   }
@@ -230 +297 @@
         }
       }
-    }
+      else getBufferFromClient(currentTimeLine) ;
+    }
+    else if (pureOneSided) getBufferFromClient(currentTimeLine) ; // if pure one sided check buffer even if no event recorded at current time line
   }
 
@@ -237 +306 @@
     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) ; 
   }
 
@@ -254 +338 @@
       finished=true;
       info(20)<<" CContextServer: Receive context <"<<context->getId()<<"> finalize."<<endl;
+      releaseBuffers() ;
       context->finalize();
       std::map<int, StdSize>::const_iterator itbMap = mapBufferSize_.begin(),

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) ;
+    
     MPI_Comm intraComm ;
     int intraCommSize ;
@@ -33 +36 @@
     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 +53 @@
     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 ;