Changeset 2547

Timestamp:

08/29/23 17:24:04 (9 months ago)

Author:

ymipsl

Message:

Major update :

• New method to lock and unlock one-sided windows (window_dynamic) to avoid network overhead
• Introducing multithreading on server sided to manage more efficiently dead-lock occuring (similar to co-routine which will be available and implemented in futur c++ standard), based on c++ threads
• Suprression of old "attached mode" which is replaced by online writer and reder filters

YM

Location:

XIOS3/trunk/src

Files:

• buffer_client.cpp (modified) (11 diffs)
• buffer_client.hpp (modified) (6 diffs)
• buffer_server.cpp (modified) (8 diffs)
• buffer_server.hpp (modified) (3 diffs)
• client.cpp (modified) (1 diff)
• cxios.cpp (modified) (2 diffs)
• cxios.hpp (modified) (3 diffs)
• manager/daemons_manager.cpp (modified) (3 diffs)
• manager/pool_ressource.cpp (modified) (5 diffs)
• manager/pool_ressource.hpp (modified) (4 diffs)
• manager/server_context.cpp (modified) (8 diffs)
• manager/server_context.hpp (modified) (2 diffs)
• manager/servers_ressource.cpp (modified) (5 diffs)
• manager/servers_ressource.hpp (modified) (1 diff)
• manager/services.cpp (modified) (4 diffs)
• manager/services.hpp (modified) (2 diffs)
• manager/thread_manager.cpp (added)
• manager/thread_manager.hpp (added)
• manager/token_manager.hpp (modified) (2 diffs)
• manager/window_base.hpp (modified) (1 diff)
• manager/window_dynamic.hpp (added)
• node/context.cpp (modified) (14 diffs)
• node/context.hpp (modified) (2 diffs)
• node/pool_node.cpp (modified) (2 diffs)
• node/service_node.cpp (modified) (1 diff)
• server.cpp (modified) (7 diffs)
• transport/context_client.cpp (modified) (1 diff)
• transport/context_client.hpp (modified) (2 diffs)
• transport/context_server.cpp (modified) (2 diffs)
• transport/context_server.hpp (modified) (2 diffs)
• transport/legacy_context_client.cpp (modified) (12 diffs)
• transport/legacy_context_client.hpp (modified) (4 diffs)
• transport/legacy_context_server.cpp (modified) (11 diffs)
• transport/legacy_context_server.hpp (modified) (1 diff)
• transport/one_sided_client_buffer.hpp (modified) (1 diff)
• transport/one_sided_context_client.cpp (modified) (5 diffs)
• transport/one_sided_context_client.hpp (modified) (2 diffs)
• transport/one_sided_context_server.cpp (modified) (8 diffs)

XIOS3/trunk/src/buffer_client.cpp

@@ -14 +14 @@
   size_t CClientBuffer::maxRequestSize = 0;
 
-  CClientBuffer::CClientBuffer(MPI_Comm interComm, int serverRank, StdSize bufferSize, StdSize estimatedMaxEventSize)
+  CClientBuffer::CClientBuffer(MPI_Comm interComm, int serverRank, StdSize bufferSize, bool hasWindows)
     : interComm(interComm)
     , clientRank_(0)
     , serverRank(serverRank)
     , bufferSize(bufferSize)
-    , estimatedMaxEventSize(estimatedMaxEventSize)
     , maxEventSize(0)
     , current(0)
     , count(0)
     , pending(false)
-    , hasWindows(false) 
-  {
-     /*
-      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]) ;
+    , hasWindows_(hasWindows) 
+  {
+      if (hasWindows_)
+      { 
+        windows_.resize(2) ;
+        windows_[0] = new CWindowDynamic() ;
+        windows_[0]->allocateBuffer(bufferSize+headerSize_) ;
+        bufferHeader[0] = (char*) windows_[0]->getBufferAddress() ;
+        windows_[1] = new CWindowDynamic() ;
+        windows_[1]->allocateBuffer(bufferSize+headerSize_) ;
+        bufferHeader[1] = (char*) windows_[1]->getBufferAddress() ;
+      }
+      else
+      {
+        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_ ;
@@ -55 +63 @@
       winState[1]=false ;
 
-
-    if (hasWindows)
-    {  
-    
-      MPI_Aint buffSize=bufferSize+headerSize_ ;
-      MPI_Win_attach(windows_[0], bufferHeader[0], buffSize) ;
-      MPI_Win_attach(windows_[1], bufferHeader[1], buffSize) ;
-    
-      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 << endl;
+    
+      retBuffer = new CBufferOut(buffer[current], bufferSize);
+      info(10) << "CClientBuffer: allocated 2 x " << bufferSize << " bytes for server " << serverRank << endl;
   }
 
@@ -91 +74 @@
     return address ;
   }
-
-  void CClientBuffer::attachWindows(vector<MPI_Win>& windows)
+  
+  MPI_Aint CClientBuffer::getWinBufferAddress(int i)
+  {
+    return windows_[i]->getWinBufferAddress() ;
+  }
+
+  void CClientBuffer::attachWindows(MPI_Comm& winComm)
   {
     isAttachedWindows_=true ;
-    windows_=windows ;
-    if (windows_[0]==MPI_WIN_NULL && windows_[1]==MPI_WIN_NULL) hasWindows=false ;
-    else hasWindows=true ;
-
-    if (hasWindows)
+
+    if (hasWindows_)
     {  
       MPI_Aint buffSize=bufferSize+headerSize_ ;
-      MPI_Win_attach(windows_[0], bufferHeader[0], buffSize) ;
-      MPI_Win_attach(windows_[1], bufferHeader[1], buffSize) ;
-    
-      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]) ;
+      windows_[0]->create(winComm) ;
+      windows_[0]->attach() ;
+      windows_[1]->create(winComm) ;
+      windows_[1]->attach() ;
+
+      windows_[0]->lockExclusive(clientRank_) ;
+      windows_[1]->lockExclusive(clientRank_) ;
+
+      windows_[0]->unlockExclusive(clientRank_) ;
+      windows_[1]->unlockExclusive(clientRank_) ;
+
     } 
 
@@ -129 +105 @@
   CClientBuffer::~CClientBuffer()
   {
-     //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;
+    if (hasWindows_)
+    {
+      windows_[0]->detach() ;
+      windows_[1]->detach() ;
+      delete windows_[0] ;
+      delete windows_[1] ;
+    }
+    else 
+    {
+      MPI_Free_mem(bufferHeader[0]) ;
+      MPI_Free_mem(bufferHeader[1]) ;
+    }
+    delete retBuffer;
   }
 
@@ -143 +123 @@
   {
     CTimer::get("lock buffer").resume();
-    if (hasWindows)
+    if (isAttachedWindows_)    
     {
       if (winState[current]==true) ERROR("CClientBuffer::lockBuffer(void)",<<"Try lo lock client buffer but winState said it is already locked") ;
-      MPI_Win_lock(MPI_LOCK_EXCLUSIVE,clientRank_, 0, windows_[current]) ;
+      //MPI_Win_lock(MPI_LOCK_EXCLUSIVE,clientRank_, 0, windows_[current]) ;
+      windows_[current]->lockExclusive(clientRank_) ;
       winState[current]=true ;
     }
@@ -155 +136 @@
   {
     CTimer::get("unlock buffer").resume();
-    if (hasWindows)
+    if (isAttachedWindows_)
     {
       if (winState[current]==false) ERROR("CClientBuffer::lockBuffer(void)",<<"Try lo unlock client buffer but winState said it is already unlocked") ;
-      MPI_Win_unlock(clientRank_, windows_[current]) ;
+      //MPI_Win_unlock(clientRank_, windows_[current]) ;
+      windows_[current]->unlockExclusive(clientRank_) ;
       winState[current]=false ;
     }
@@ -189 +171 @@
     {
       maxEventSize = size;
-
-      if (size > estimatedMaxEventSize)
-        error(0) << "WARNING: Unexpected event of size " << size << " for server " << serverRank
-                 << " (estimated max event size = " << estimatedMaxEventSize << ")" << std::endl;
 
       if (size > maxRequestSize) maxRequestSize = size;
@@ -319 +297 @@
   {
 
-    if (hasWindows)
+    bufferSize=newSize ;
+
+    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]) ;
-
-    bufferSize=newSize ;
-    MPI_Alloc_mem(bufferSize+headerSize_, MPI_INFO_NULL, &bufferHeader[0]) ;
-    MPI_Alloc_mem(bufferSize+headerSize_, MPI_INFO_NULL, &bufferHeader[1]) ;
+      windows_[0]->detach();
+      windows_[1]->detach();
+      
+      windows_[0]->attach(bufferSize+headerSize_) ;
+      bufferHeader[0] = (char*) windows_[0] -> getBufferAddress() ;
+      windows_[1]->attach(bufferSize+headerSize_) ;
+      bufferHeader[1] = (char*) windows_[1] -> getBufferAddress() ;
+    }
+    
+    
     buffer[0] = bufferHeader[0]+headerSize_ ;
     buffer[1] = bufferHeader[1]+headerSize_ ;
@@ -353 +334 @@
     current=0 ;
     
-    if (hasWindows)
+    if (hasWindows_)
     {  
-    
-      MPI_Win_attach(windows_[0], bufferHeader[0], bufferSize+headerSize_) ;
-      MPI_Win_attach(windows_[1], bufferHeader[1], bufferSize+headerSize_) ;
-          
-      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]) ;
+
+      windows_[0]->lockExclusive(clientRank_) ;
+      windows_[1]->lockExclusive(clientRank_) ;
+      
+      windows_[1]->unlockExclusive(clientRank_) ;
+      windows_[0]->unlockExclusive(clientRank_) ;
     } 
 
@@ -373 +351 @@
     bufOut->put(timelineEventChangeBufferSize);
     bufOut->put(newBufferSize_);
-    bufOut->put(this->getWinAddress(0));
-    bufOut->put(this->getWinAddress(1));
+    
+    bufOut->put(this->getWinBufferAddress(0));
+    bufOut->put(this->getWinBufferAddress(1));
 
     resizingBufferStep_=4;
     unlockBuffer() ;
-    info(100)<<"CClientBuffer::resizeBuffer(size_t newSize) : resizing buffer of server "<<serverRank<<" ; new size : "<<newSize<<" ; winAdress[0] "<<this->getWinAddress(0)<<" winAdress[1] "<<this->getWinAddress(1)<<endl;
+    info(100)<<"CClientBuffer::resizeBuffer(size_t newSize) : resizing buffer of server "<<serverRank<<" ; new size : "<<newSize<<" ; winAdress[0] "<<this->getWinBufferAddress(0)<<" winAdress[1] "<<this->getWinBufferAddress(1)<<endl;
   }
 
@@ -397 +376 @@
     lockBuffer() ;
     ret=*notify[current] == notifyResizeBuffer_ ? true : false ;
-    if (ret || !hasWindows) 
+    if (ret || !hasWindows_) 
     {
       *notify[current] = notifyNothing_ ;

XIOS3/trunk/src/buffer_client.hpp

@@ -7 +7 @@
 #include "mpi.hpp"
 #include "cxios.hpp"
+#include "window_dynamic.hpp"
 
 namespace xios
@@ -15 +16 @@
       static size_t maxRequestSize;
 
-      CClientBuffer(MPI_Comm intercomm, int serverRank, StdSize bufferSize, StdSize estimatedMaxEventSize);
+      CClientBuffer(MPI_Comm intercomm, int serverRank, StdSize bufferSize, bool hasWindows);
       ~CClientBuffer();
 //      void createWindows(MPI_Comm oneSidedComm) ;
@@ -28 +29 @@
       StdSize remain(void);
       MPI_Aint getWinAddress(int numWindows) ;
+      MPI_Aint getWinBufferAddress(int numWindows) ;
       void infoBuffer(void) ;
       bool isNotifiedFinalized(void) ;
@@ -33 +35 @@
       void fixBufferSize(size_t bufferSize) { newBufferSize_=bufferSize ; isGrowableBuffer_=false ; resizingBufferStep_=1 ;}
       void fixBuffer(void) { isGrowableBuffer_=false ;}
-      void attachWindows(vector<MPI_Win>& windows) ;
+      void attachWindows(MPI_Comm& winComm) ;
       bool isAttachedWindows(void) { return isAttachedWindows_ ;}
     private:
@@ -58 +60 @@
       StdSize maxEventSize;
       StdSize bufferSize;
-      const StdSize estimatedMaxEventSize;
       bool isFinalized_=false ;
 
@@ -69 +70 @@
       CBufferOut* retBuffer;
       const MPI_Comm interComm;
-      std::vector<MPI_Win> windows_ ;
-      bool hasWindows=false ;
+      std::vector<CWindowDynamic*> windows_ ;
+      bool hasWindows_=false ;
       bool isAttachedWindows_=false ;
       double latency_=0 ;

XIOS3/trunk/src/buffer_server.cpp

@@ -3 +3 @@
 #include "buffer_server.hpp"
 #include "timer.hpp"
+#include "window_dynamic.hpp"
 
 
@@ -8 +9 @@
 {
 
-  CServerBuffer::CServerBuffer(vector<MPI_Win>& windows, vector<MPI_Aint>& winAddress, int windowsRank, StdSize buffSize) 
+  CServerBuffer::CServerBuffer(vector<CWindowDynamic*>& windows, vector<MPI_Aint>& winAddress, int windowsRank, StdSize buffSize) 
   : hasWindows(true), windows_(windows), windowsRank_(windowsRank), winAddress_(winAddress)
   {
@@ -18 +19 @@
     MPI_Alloc_mem(size, MPI_INFO_NULL, &buffer) ;
     currentWindows=1 ;
-    if (windows[0]==MPI_WIN_NULL && windows[1]==MPI_WIN_NULL) hasWindows=false ;
+    if (windows[0]==nullptr && windows[1]==nullptr) hasWindows=false ;
   }
 
@@ -239 +240 @@
     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(); 
     CTimer::get("getBufferFromClient_locked").resume() ;   
 // lock is acquired
 
-    MPI_Get(&clientTimeline, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows],timeLineOffset_*sizeof(size_t)), 1, MPI_LONG_LONG_INT,windows_[currentWindows]) ;
-    MPI_Get(&clientCount, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows],countOffset_*sizeof(size_t)), 1, MPI_LONG_LONG_INT,windows_[currentWindows]) ;
-    MPI_Win_flush(windowsRank_, windows_[currentWindows]) ;
-
+    windows_[currentWindows]->get(&clientTimeline, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows],timeLineOffset_*sizeof(size_t)), 1, MPI_LONG_LONG_INT) ;
+    windows_[currentWindows]->get(&clientCount, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows],countOffset_*sizeof(size_t)), 1, MPI_LONG_LONG_INT) ;
+    windows_[currentWindows]->flush(windowsRank_) ;
+   
     if (timeLine==clientTimeline)
     {
       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]) ;
+      windows_[currentWindows]->get(buffer, clientCount, MPI_CHAR, windowsRank_, MPI_Aint_add(winAddress_[currentWindows],4*sizeof(size_t)) , clientCount, MPI_CHAR) ;
+      
       clientTimeline = 0 ;
       clientCount = 0 ;
-      MPI_Put(&clientTimeline, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows],timeLineOffset_*sizeof(size_t)), 1, MPI_LONG_LONG_INT,windows_[currentWindows]) ;
-      MPI_Put(&clientCount, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows],countOffset_*sizeof(size_t)), 1, MPI_LONG_LONG_INT,windows_[currentWindows]) ;
+      windows_[currentWindows]->put(&clientTimeline, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows],timeLineOffset_*sizeof(size_t)), 1, MPI_LONG_LONG_INT) ;
+      windows_[currentWindows]->put(&clientCount, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows],countOffset_*sizeof(size_t)), 1, MPI_LONG_LONG_INT) ;
 
 // release lock
@@ -298 +295 @@
   {
     if (!hasWindows) return ;
-    MPI_Win_lock(MPI_LOCK_EXCLUSIVE,windowsRank_,0,windows_[currentWindows]) ;
+    //MPI_Win_lock(MPI_LOCK_EXCLUSIVE,windowsRank_,0,windows_[currentWindows]) ;
+    windows_[currentWindows]->lockExclusive(windowsRank_) ;
   }
 
@@ -304 +302 @@
   {
     if (!hasWindows) return ;
-    MPI_Win_unlock(windowsRank_,windows_[currentWindows]) ;
+    windows_[currentWindows]->unlockExclusive(windowsRank_) ;
   }
   
@@ -313 +311 @@
     lockBuffer(); 
 // lock is acquired
-    MPI_Put(&notify, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows], notifyOffset_*sizeof(size_t)), 1, MPI_LONG_LONG_INT,windows_[currentWindows]) ;
+    windows_[currentWindows]->put(&notify, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows], notifyOffset_*sizeof(size_t)), 1, MPI_LONG_LONG_INT) ;
     unlockBuffer() ;
   }
@@ -324 +322 @@
     lockBuffer(); 
 // lock is acquired
-    MPI_Put(&notify, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows], notifyOffset_*sizeof(size_t)), 1, MPI_LONG_LONG_INT,windows_[currentWindows]) ;
+    windows_[currentWindows]->put(&notify, 1, MPI_LONG_LONG_INT, windowsRank_ , MPI_Aint_add(winAddress_[currentWindows], notifyOffset_*sizeof(size_t)), 1, MPI_LONG_LONG_INT) ;
     unlockBuffer() ;
   }

XIOS3/trunk/src/buffer_server.hpp

@@ -7 +7 @@
 #include "mpi.hpp"
 #include "cxios.hpp"
+#include "window_dynamic.hpp"
+
 
 namespace xios
@@ -13 +15 @@
   {
     public:
-      CServerBuffer(vector<MPI_Win>& windows, vector<MPI_Aint>& winAddress, int windowsRank, StdSize bufSize) ;
+      CServerBuffer(vector<CWindowDynamic*>& windows, vector<MPI_Aint>& winAddress, int windowsRank, StdSize bufSize) ;
       ~CServerBuffer() ;
 
@@ -37 +39 @@
       size_t size;
       size_t used ;  // count of element occupied
-      std::vector<MPI_Win> windows_ ;
+      std::vector<CWindowDynamic*> windows_ ;
       std::vector<MPI_Aint> winAddress_ ;
       bool resizingBuffer_ = false ;

XIOS3/trunk/src/client.cpp

@@ -180 +180 @@
       MPI_Comm_size(CXios::getXiosComm(), &xiosCommSize) ;
       MPI_Comm_size(clientsComm_, &clientsCommSize) ;
-      if (xiosCommSize==clientsCommSize) CXios::setUsingServer() ;
-      else CXios::setNotUsingServer() ;
+      if (xiosCommSize==clientsCommSize) CXios::setNotUsingServer() ;
+      else CXios::setUsingServer() ;
 
       /////////////////////////////////////////

XIOS3/trunk/src/cxios.cpp

@@ -313 +313 @@
   }
 
-  
+  void CXios::launchThreadManager(bool isXiosServer)
+  {
+    CThreadManager::initialize(isXiosServer) ;
+  }
+
   void CXios::finalizeRegistryManager()
   {
@@ -339 +343 @@
   }
   
+  void CXios::finalizeThreadManager()
+  {
+    CThreadManager::finalize()  ;
+  }
+
   void CXios::finalizeDaemonsManager()
   {

XIOS3/trunk/src/cxios.hpp

@@ -11 +11 @@
 #include "coupler_manager.hpp"
 #include "registry_manager.hpp"
+#include "thread_manager.hpp"
 #include "mpi_garbage_collector.hpp"
 
@@ -106 +107 @@
      static void launchCouplerManager(bool isXiosServer) ;
      static void launchRegistryManager(bool isXiosServer) ;
+     static void launchThreadManager(bool isXiosServer) ;
     
      static void finalizeServicesManager() ;
@@ -113 +115 @@
      static void finalizeCouplerManager() ;
      static void finalizeRegistryManager() ;
+     static void finalizeThreadManager() ;
 
      static CRegistryManager*   getRegistryManager(void) { return registryManager_ ;}

XIOS3/trunk/src/manager/daemons_manager.cpp

@@ -24 +24 @@
     CXios::launchContextsManager(isXiosServer) ;
     CXios::launchCouplerManager(isXiosServer) ;
+    CXios::launchThreadManager(isXiosServer) ;
  
     if (isXiosServer) CServer::launchServersRessource(splitComm) ;
@@ -41 +42 @@
     CXios::getServicesManager()->eventLoop() ;
     CXios::getContextsManager()->eventLoop() ;
-    if (isServer_) {
-        if (CServer::isRoot) {
-            CServer::listenOasisEnddef() ;
-            CServer::listenRootOasisEnddef() ;
-        }
-        else {
-            CServer::listenRootOasisEnddef() ;
-        }
-        return CServer::getServersRessource()->eventLoop(false) ;
+    if (isServer_) 
+    {
+      if (CServer::isRoot) 
+      {
+        CServer::listenOasisEnddef() ;
+        CServer::listenRootOasisEnddef() ;
+      }
+      else CServer::listenRootOasisEnddef() ;
+      
+      if (CThreadManager::isUsingThreads()) return CServer::getServersRessource()->isFinished() ;
+      else return CServer::getServersRessource()->eventLoop(false) ;
     }
     else  return CXios::getPoolRessource()->eventLoop(false) ;
@@ -76 +79 @@
       CXios::finalizeRessourcesManager() ;
       CXios::finalizeRegistryManager() ;
+      CXios::finalizeThreadManager() ;
       isFinalized_=true ;
     }

XIOS3/trunk/src/manager/pool_ressource.cpp

@@ -8 +8 @@
 #include "timer.hpp"
 #include "event_scheduler.hpp"
+#include "thread_manager.hpp"
 
 namespace xios
@@ -33 +34 @@
     else eventScheduler_= make_shared<CEventScheduler>(poolComm) ;
     freeRessourceEventScheduler_ = eventScheduler_ ;
+    std::hash<string> hashString ;
+    hashId_ = hashString("CPoolRessource::"+Id) ;
+    if (CThreadManager::isUsingThreads()) CThreadManager::spawnThread(&CPoolRessource::threadEventLoop, this) ;
+  }
+
+  void CPoolRessource::synchronize(void)
+  {
+    bool out=false ; 
+    size_t timeLine=0 ;
+          
+    eventScheduler_->registerEvent(timeLine, hashId_) ;
+    while (!out)
+    {
+      CThreadManager::yield() ;
+      out = eventScheduler_->queryEvent(timeLine,hashId_) ;
+      if (out) eventScheduler_->popEvent() ;
+    }
   }
 
@@ -122 +140 @@
     MPI_Comm_rank(poolComm_, &commRank) ;
     winNotify_->popFromExclusiveWindow(commRank, this, &CPoolRessource::notificationsDumpIn) ;
-    if (notifyType_==NOTIFY_CREATE_SERVICE) createService() ;
-    else if (notifyType_==NOTIFY_CREATE_SERVICE_ONTO) createServiceOnto() ;
+    if (notifyType_==NOTIFY_CREATE_SERVICE) 
+    {
+      if (CThreadManager::isUsingThreads()) synchronize() ;
+      createService() ;
+    }
+    else if (notifyType_==NOTIFY_CREATE_SERVICE_ONTO) 
+    {
+      if (CThreadManager::isUsingThreads()) synchronize() ;
+      createServiceOnto() ;
+    }
   }
 
@@ -233 +259 @@
     }
     CTimer::get("CPoolRessource::eventLoop").suspend();
-    if (services_.empty() && finalizeSignal_) return true ;
-    else return false ;
-  }
+    if (services_.empty() && finalizeSignal_) finished_=true ;
+    return finished_ ;
+  }
+
+  void CPoolRessource::threadEventLoop(void)
+  {
+    CTimer::get("CPoolRessource::eventLoop").resume();
+    info(100)<<"Launch Thread for  CPoolRessource::threadEventLoop, pool id = "<<Id_<<endl ;
+    CThreadManager::threadInitialize() ; 
+    
+    do
+    {
+
+      double time=MPI_Wtime() ;
+      int flag ;
+      MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &flag, MPI_STATUS_IGNORE);
+      if (time-lastEventLoop_ > eventLoopLatency_) 
+      {
+        //checkCreateServiceNotification() ;
+        checkNotifications() ;
+        lastEventLoop_=time ;
+      }
+    
+      for(auto it=services_.begin();it!=services_.end();++it) 
+      {
+        if (it->second->isFinished())
+        {
+          delete it->second ; 
+          services_.erase(it) ;
+          // destroy server_context -> to do later
+          break ;
+        } ;
+      }
+
+      CTimer::get("CPoolRessource::eventLoop").suspend();
+      if (services_.empty() && finalizeSignal_) finished_=true ;
+      
+      if (!finished_) CThreadManager::yield() ;
+    
+    } while (!finished_) ;
+
+    CThreadManager::threadFinalize() ;
+    info(100)<<"Close thread for  CPoolRessource::threadEventLoop, pool id = "<<Id_<<endl ;
+  }
+
 /*
   void CPoolRessource::checkCreateServiceNotification(void)
@@ -347 +415 @@
   }
 
-  void CPoolRessource::createService(MPI_Comm serviceComm, shared_ptr<CEventScheduler> eventScheduler, const std::string& serviceId, int partitionId, int type, int nbPartitions) // for clients & attached
+  void CPoolRessource::createService(MPI_Comm serviceComm, shared_ptr<CEventScheduler> eventScheduler, const std::string& serviceId, int partitionId, int type, int nbPartitions) // for clients
   {
     services_[std::make_tuple(serviceId,partitionId)] = new CService(serviceComm, eventScheduler, Id_, serviceId, partitionId, type, nbPartitions) ;

XIOS3/trunk/src/manager/pool_ressource.hpp

@@ -35 +35 @@
     void createServiceOnto(const std::string& serviceId, int type, const std::string& OnServiceId) ;
     bool eventLoop(bool serviceOnly=false) ;
+    void threadEventLoop(void) ;
     bool hasService(const std::string serviceId, int partitionId) {return services_.count(make_tuple(serviceId,partitionId))>0 ;}
     CService* getService(const std::string serviceId, int partitionId) { return services_[make_tuple(serviceId,partitionId)]; }
@@ -50 +51 @@
     void createService(void) ;
     void createServiceOnto(void) ;
+    void synchronize(void) ;
     
 //    void createServiceNotify(int rank, const std::string& serviceId, int type, int size, int nbPartitions, bool in) ;
@@ -58 +60 @@
   public:
     void createNewServiceOnto(const std::string& serviceId, int type, const string& onServiceId) ;
-    
-    private:
+  
+  private:
+    bool finished_=false ;
+  public:
+    bool isFinished(void) { return finished_ ;}
+
+  private:
     MPI_Comm poolComm_ ;
     
@@ -74 +81 @@
     std::string Id_ ;
     bool finalizeSignal_ ;
-    
+        
     const double eventLoopLatency_=0; 
     double lastEventLoop_=0. ;
 
     private:
+      size_t hashId_ ;
       shared_ptr<CEventScheduler> eventScheduler_ ;
       shared_ptr<CEventScheduler> freeRessourceEventScheduler_ ;

XIOS3/trunk/src/manager/server_context.cpp

@@ -6 +6 @@
 #include "register_context_info.hpp"
 #include "services.hpp"
+#include "thread_manager.hpp"
 #include "timer.hpp"
 
@@ -50 +51 @@
     info(10)<<"Context "<< CXios::getContextsManager()->getServerContextName(poolId, serviceId, partitionId, type, contextId)<<" created, on local rank "<<localRank
                         <<" and global rank "<<globalRank<<endl  ;
+   
+    if (CThreadManager::isUsingThreads()) CThreadManager::spawnThread(&CServerContext::threadEventLoop, this) ;
   }
 
@@ -84 +87 @@
     }
     
-    MPI_Request req ;
-    MPI_Status status ;
-    MPI_Ibarrier(intraComm,&req) ;
-    
-    int flag=false ;
-    while(!flag) 
-    {
-      CXios::getDaemonsManager()->servicesEventLoop() ;
-      MPI_Test(&req,&flag,&status) ;
-    }
-
+    if (wait)
+    {
+      MPI_Request req ;
+      MPI_Status status ;
+      MPI_Ibarrier(intraComm,&req) ;
+    
+      int flag=false ;
+      while(!flag) 
+      {
+        CXios::getDaemonsManager()->servicesEventLoop() ;
+        MPI_Test(&req,&flag,&status) ;
+      }
+    }
+    
     MPI_Bcast(&ok, 1, MPI_INT, 0, intraComm) ;
 
@@ -117 +123 @@
       MPI_Comm_size(contextComm_,&commSize ) ;
 */
-      if (nOverlap> 0 )
-      {
-        while (get<0>(overlapedComm_[name_])==false) CXios::getDaemonsManager()->servicesEventLoop() ;
-        isAttachedMode_=true ;
-        cout<<"CServerContext::createIntercomm : total overlap ==> context in attached mode"<<endl ;
-        interCommClient=newInterCommClient ;
-        interCommServer=newInterCommServer ;
-      }
-      else if (nOverlap==0)
+      if (nOverlap==0)
       { 
-        cout<<"CServerContext::createIntercomm : No overlap ==> context in server mode"<<endl ;
-        isAttachedMode_=false ;
         MPI_Intercomm_create(intraComm, 0, xiosComm_, contextLeader, 3141, &interCommClient) ;
         MPI_Comm_dup(interCommClient, &interCommServer) ;
@@ -136 +132 @@
       else
       {
-        cout<<"CServerContext::createIntercomm : partial overlap ==> not managed"<<endl ;
+        ERROR("void CServerContext::createIntercomm(void)",<<"CServerContext::createIntercomm : overlap ==> not managed") ;
       }
     }
@@ -259 +255 @@
   }
 
+  void CServerContext::threadEventLoop(void)
+  {
+    
+    info(100)<<"Launch Thread for CServerContext::threadEventLoop, context id = "<<context_->getId()<<endl ;
+    CThreadManager::threadInitialize() ; 
+    do
+    {
+      CTimer::get("CServerContext::eventLoop").resume();
+      int flag ;
+      MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &flag, MPI_STATUS_IGNORE);
+
+      if (winNotify_!=nullptr) checkNotifications() ;
+
+
+      if (context_!=nullptr)  
+      {
+        if (context_->eventLoop())
+        {
+          info(100)<<"Remove context server with id "<<context_->getId()<<endl ;
+          CContext::removeContext(context_->getId()) ;
+          context_=nullptr ;
+          // destroy context ??? --> later
+        }
+      }
+      CTimer::get("CServerContext::eventLoop").suspend();
+      if (context_==nullptr && finalizeSignal_) finished_=true ;
+ 
+      if (!finished_) CThreadManager::yield() ;
+    }
+    while (!finished_) ;
+    
+    CThreadManager::threadFinalize() ;
+    info(100)<<"Close thread for CServerContext::threadEventLoop"<<endl ;
+  }
+
   void CServerContext::createIntercomm(void)
   {
@@ -280 +311 @@
      MPI_Comm_size(contextComm_,&commSize ) ;
 
-    if (nOverlap==commSize)
-    {
-      info(10)<<"CServerContext::createIntercomm : total overlap ==> context in attached mode"<<endl ;
-      isAttachedMode_=true ;
-      interCommClient=get<2>(it->second) ;
-      interCommServer=get<1>(it->second) ;
-      context_ -> createClientInterComm(interCommClient, interCommServer ) ;
-      clientsInterComm_.push_back(interCommClient) ;
-      clientsInterComm_.push_back(interCommServer) ;
-    }
-    else if (nOverlap==0)
+    if (nOverlap==0)
     { 
       info(10)<<"CServerContext::createIntercomm : No overlap ==> context in server mode"<<endl ;
-      isAttachedMode_=false ;
       MPI_Intercomm_create(contextComm_, 0, xiosComm_, remoteLeader, 3141, &interCommServer) ;
       MPI_Comm_dup(interCommServer,&interCommClient) ;
@@ -302 +322 @@
     else
     {
-      ERROR("void CServerContext::createIntercomm(void)",<<"CServerContext::createIntercomm : partial overlap ==> not managed") ;
+      ERROR("void CServerContext::createIntercomm(void)",<<"CServerContext::createIntercomm : overlap ==> not managed") ;
     }
    

XIOS3/trunk/src/manager/server_context.hpp

@@ -29 +29 @@
 
     bool eventLoop(bool serviceOnly=false) ;
+    void threadEventLoop(void) ;
     void notificationsDumpOut(CBufferOut& buffer) ;
     void notificationsDumpIn(CBufferIn& buffer) ;
     void finalizeSignal(void) ;
     void freeComm(void) ;
-    bool isAttachedMode(void) { return isAttachedMode_ ;}
     CService* getParentService(void) {return parentService_ ; }
-
+    
+    private : 
+      bool finished_=false ;
+    public:
+      bool isFinished(void) { return finished_ ; }
     private:
     void createIntercomm(void) ;
@@ -60 +64 @@
     bool finalizeSignal_ ;
     bool hasNotification_ ;
-    bool isAttachedMode_ ;
 
     const double eventLoopLatency_=0; 

XIOS3/trunk/src/manager/servers_ressource.cpp

@@ -9 +9 @@
 #include <vector>
 #include <string>
+#include "thread_manager.hpp"
 
 
@@ -43 +44 @@
     eventScheduler_ = make_shared<CEventScheduler>(freeRessourcesComm_) ;
     freeRessourceEventScheduler_ = eventScheduler_ ;
+    if (CThreadManager::isUsingThreads()) CThreadManager::spawnThread(&CServersRessource::threadEventLoop, this) ;
   }
 
@@ -131 +133 @@
     if (poolRessource_!=nullptr) 
     {
-      if (poolRessource_->eventLoop(serviceOnly))
+      poolRessource_->eventLoop(serviceOnly) ;
+      if (poolRessource_->isFinished())
       {
         delete poolRessource_ ;
@@ -139 +142 @@
     }
     CTimer::get("CServersRessource::eventLoop").suspend();
-    if (poolRessource_==nullptr && finalizeSignal_) return true ;
-    else return false ;
-  }
+    if (poolRessource_==nullptr && finalizeSignal_) finished_=true ;
+    return finished_ ;
+  }
+
+  void CServersRessource::threadEventLoop(void)
+  {
+    CTimer::get("CServersRessource::eventLoop").resume();
+    info(100)<<"Launch Thread for  CServersRessource::threadEventLoop"<<endl ;
+    CThreadManager::threadInitialize() ; 
+
+    do
+    {
+      double time=MPI_Wtime() ;
+      int flag ;
+      MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &flag, MPI_STATUS_IGNORE);
+
+      if (time-lastEventLoop_ > eventLoopLatency_) 
+      {
+        checkNotifications() ;
+        lastEventLoop_=time ;
+      }
+
+      if (poolRessource_!=nullptr) 
+      {
+        if (poolRessource_->isFinished())
+        {
+          delete poolRessource_ ;
+          poolRessource_=nullptr ;
+          // don't forget to free pool ressource later
+        } 
+      }
+      CTimer::get("CServersRessource::eventLoop").suspend();
+      if (poolRessource_==nullptr && finalizeSignal_) finished_=true ;
+      if (!finished_) CThreadManager::yield() ;
+    
+    } while (!finished_) ;
+
+    CThreadManager::threadFinalize() ;
+    info(100)<<"Close thread for CServersRessource::threadEventLoop"<<endl ; ;
+  }
+
 
   void CServersRessource::checkNotifications(void)
@@ -148 +189 @@
     MPI_Comm_rank(serverComm_, &commRank) ;
     winNotify_->popFromExclusiveWindow(commRank, this, &CServersRessource::notificationsDumpIn) ;
-    if (notifyInType_==NOTIFY_CREATE_POOL) createPool() ;
+    if (notifyInType_==NOTIFY_CREATE_POOL) 
+    {
+      if (CThreadManager::isUsingThreads()) synchronize() ;
+      createPool() ;
+    }
     else if (notifyInType_==NOTIFY_FINALIZE) finalizeSignal() ;
+  }
+
+  void CServersRessource::synchronize(void)
+  {
+    bool out=false ; 
+    size_t timeLine=0 ;
+    std::hash<string> hashString ;
+    int commSize ;
+    MPI_Comm_size(freeRessourcesComm_,&commSize) ;
+    size_t hashId = hashString("CServersRessource::"+to_string(commSize)) ;
+    freeRessourceEventScheduler_->registerEvent(timeLine, hashId) ;
+    while (!out)
+    {
+      CThreadManager::yield() ;
+      out = eventScheduler_->queryEvent(timeLine,hashId) ;
+      if (out) eventScheduler_->popEvent() ;
+    }
   }
 

XIOS3/trunk/src/manager/servers_ressource.hpp

@@ -55 +55 @@
     const double eventLoopLatency_=0; 
     double lastEventLoop_=0. ;
-
+    
     private:
+      bool finished_=false;
+    public:
+      bool isFinished(void) { return finished_ ;}
+    private:
+      void synchronize(void) ;
+      void threadEventLoop(void) ;
       shared_ptr<CEventScheduler> eventScheduler_ ;
       shared_ptr<CEventScheduler> freeRessourceEventScheduler_ ;

XIOS3/trunk/src/manager/services.cpp

@@ -5 +5 @@
 #include "server_context.hpp"
 #include "event_scheduler.hpp"
+#include "thread_manager.hpp"
 #include "timer.hpp"
 
@@ -41 +42 @@
     oss<<partitionId;
     name_= poolId+"__"+serviceId+"_"+oss.str();
+    
+    if (CThreadManager::isUsingThreads()) CThreadManager::spawnThread(&CService::threadEventLoop, this) ;
   }
 
@@ -124 +127 @@
 
     eventScheduler_->checkEvent() ;
+   
     for(auto it=contexts_.begin();it!=contexts_.end();++it) 
     {
@@ -134 +138 @@
       } ;
     }
+  
     CTimer::get("CService::eventLoop").suspend();
     if (contexts_.empty() && finalizeSignal_) return true ;
     else return false ;
   }
+
+  void CService::threadEventLoop(void)
+  {
+    info(100)<<"Launch Thread for  CService::threadEventLoop, service id = "<<name_<<endl ;
+    CThreadManager::threadInitialize() ; 
+    
+    do
+    {
+      CTimer::get("CService::eventLoop").resume();
+      int flag ;
+      MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &flag, MPI_STATUS_IGNORE);
+   
+//    double time=MPI_Wtime() ;
+//    if (time-lastEventLoop_ > eventLoopLatency_) 
+//    {
+      checkNotifications() ;
+//      lastEventLoop_=time ;
+//    }
+
+
+      eventScheduler_->checkEvent() ;
+   
+      for(auto it=contexts_.begin();it!=contexts_.end();++it) 
+      {
+        if (it->second->isFinished())
+        {
+          delete it->second ; 
+          contexts_.erase(it) ;
+          // destroy server_context -> to do later
+          break ;
+        } ;
+      }
+
+      CTimer::get("CService::eventLoop").suspend();
+      if (contexts_.empty() && finalizeSignal_) finished_=true ;
+      if (!finished_) CThreadManager::yield() ;
+    } while (!finished_) ;
+    
+    CThreadManager::threadFinalize() ;
+    info(100)<<"Close thread for  CService::threadEventLoop, service id = "<<name_<<endl ;
+  }
+
 
   void CService::sendNotification(int rank)

XIOS3/trunk/src/manager/services.hpp

@@ -24 +24 @@
 
     bool eventLoop(bool serviceOnly=false) ;
+    void threadEventLoop(void) ;
     void createContext(const std::string& poolId, const std::string& serviceId, const int& partitionId, const std::string& contextId) ;
     void checkCreateContextNotification(void) ;
@@ -41 +42 @@
     const MPI_Comm& getCommunicator(void) { return serviceComm_ ;}
     
+    private: 
+    bool finished_=false ;
+    public:
+    bool isFinished(void) { return finished_; }
+
     private:
     void sendNotification(int rank) ;

XIOS3/trunk/src/manager/token_manager.hpp

@@ -26 +26 @@
           memset( winBufferRetrieved_, 0, windowSize );
         }
+        MPI_Win_lock_all(0, winCurrentToken_) ;
+        MPI_Win_lock_all(0, winRetrievedToken_) ;
       }
-
+      
+      ~CTokenManager()
+      {
+        MPI_Win_unlock_all(winCurrentToken_) ;
+        MPI_Win_unlock_all(winRetrievedToken_) ;
+        MPI_Win_free(&winCurrentToken_) ;
+        MPI_Win_free(&winRetrievedToken_) ;
+      }
+      
       size_t getToken(void)
       {
         size_t inc=1 ;
         size_t token ;
-        MPI_Win_lock(MPI_LOCK_EXCLUSIVE, leader_, 0, winCurrentToken_) ;
         MPI_Fetch_and_op(&inc, &token, MPI_SIZE_T, leader_, 0, MPI_SUM, winCurrentToken_) ;
-        MPI_Win_unlock(leader_, winCurrentToken_) ;
+        MPI_Win_flush(leader_, winCurrentToken_);
         return token ;
       }
 
-      bool lockToken(size_t token)
+      bool checkToken(size_t token)
       {
         size_t tokenRead ;
-        MPI_Win_lock(MPI_LOCK_SHARED, leader_, 0, winRetrievedToken_) ;
-        MPI_Get(&tokenRead, 1, MPI_SIZE_T, leader_, 0, 1, MPI_SIZE_T, winRetrievedToken_ ) ;
-        MPI_Win_unlock(leader_, winRetrievedToken_) ;
-        if (token==tokenRead) return true ;
-        else return false ;
+        size_t inc=0 ;
+        MPI_Fetch_and_op(&inc, &tokenRead, MPI_SIZE_T, leader_, 0, MPI_NO_OP, winRetrievedToken_) ;
+        MPI_Win_flush(leader_, winRetrievedToken_);
+        return tokenRead==token ;
       }
-
-      void unlockToken(size_t token)
+      
+      void updateToken(size_t token)
+      {
+        size_t inc=1 ;
+        size_t tokenRead ;
+        MPI_Fetch_and_op(&inc, &tokenRead, MPI_SIZE_T, leader_, 0, MPI_SUM, winRetrievedToken_) ;
+        MPI_Win_flush(leader_, winRetrievedToken_);
+        if (token!=tokenRead)  ERROR("void CTokenManager::unlockToken(size_t token)",<<"Cannot release token="<<token<<
+                                     " that is not corresponding to the locked token="<<tokenRead) ;     
+      }
+/*      void unlockToken(size_t token)
       {
         size_t inc=1 ;
@@ -59 +76 @@
                                      " that is not corresponding to the locked token="<<tokenRead) ;     
       }
-
+*/
     private:
 

XIOS3/trunk/src/manager/window_base.hpp

@@ -137 +137 @@
       return MPI_Get(origin_addr, origin_count, origin_datatype, target_rank, target_disp + OFFSET_BUFFER, target_count, target_datatype, window_) ;
     }
-
+    
+    int fetchAndOp(const void *origin_addr, void *result_addr, MPI_Datatype datatype, int target_rank, MPI_Aint target_disp, MPI_Op op)
+    {
+      return MPI_Fetch_and_op(origin_addr, result_addr, datatype, target_rank, target_disp + OFFSET_BUFFER, op, window_ ) ;
+    }
+    
     int compareAndSwap(const void *origin_addr, const void *compare_addr, void *result_addr, MPI_Datatype datatype,
                        int target_rank, MPI_Aint target_disp)

XIOS3/trunk/src/node/context.cpp

@@ -34 +34 @@
 #include "services.hpp"
 #include "contexts_manager.hpp"
+#include "thread_manager.hpp"
 #include <chrono>
 #include <random>
@@ -488 +489 @@
       }
       contextId_ = getId() ;
-      
-      attached_mode=true ;
-      if (!CXios::isUsingServer()) attached_mode=false ;
-
 
       string contextRegistryId=getId() ;
@@ -544 +541 @@
     MPI_Comm_dup(intraComm_, &intraCommClient);
     comms.push_back(intraCommClient);
-    // attached_mode=parentServerContext_->isAttachedMode() ; //ym probably inherited from source context
 
     CContextServer* server ;
@@ -578 +574 @@
     if (commRank==0)
     {
-      CXios::getServicesManager()->getServiceNbPartitions(poolId, serverId, 0, nbPartitions, true) ;
-      for(int i=0 ; i<nbPartitions; i++) CXios::getContextsManager()->createServerContext(poolId, serverId, i, getContextId()) ;
-    }
-    setCurrent(getId()) ; // getCurrent/setCurrent may be supress, it can cause a lot of trouble (attached ???)
+      while (! CXios::getServicesManager()->getServiceNbPartitions(poolId, serverId, 0, nbPartitions)) yield() ;
+      for(int i=0 ; i<nbPartitions; i++)  
+        while (!CXios::getContextsManager()->createServerContext(poolId, serverId, i, getContextId())) yield() ;
+    }
+    synchronize() ;
+    setCurrent(getId()) ; // getCurrent/setCurrent may be supress, it can cause a lot of trouble 
     MPI_Bcast(&nbPartitions, 1, MPI_INT, 0, intraComm_) ;
       
@@ -587 +585 @@
     for(int i=0 ; i<nbPartitions; i++)
     {
-      parentServerContext_->createIntercomm(poolId, serverId, i, getContextId(), intraComm_, interCommClient, interCommServer) ;
+      while (!parentServerContext_->createIntercomm(poolId, serverId, i, getContextId(), intraComm_, interCommClient, interCommServer)) yield() ;
       int type ; 
-      if (commRank==0) CXios::getServicesManager()->getServiceType(poolId, serverId, 0, type, true) ;
+      if (commRank==0) while (!CXios::getServicesManager()->getServiceType(poolId, serverId, 0, type)) yield();
+      synchronize() ;
       MPI_Bcast(&type,1,MPI_INT,0,intraComm_) ;
+
       string fullServerId=CXios::getContextsManager()->getServerContextName(poolId, serverId, i, type, getContextId()) ;
 
@@ -620 +620 @@
     if (serviceType_ == CServicesManager::CLIENT)
     {
-      if (attached_mode) createServerInterComm(CClient::getPoolRessource()->getId(), getContextId()+"_"+CXios::defaultWriterId, clientServers) ;
-      else if (CXios::usingServer2) createServerInterComm(CXios::defaultPoolId, CXios::defaultGathererId, clientServers) ;
+      if (CXios::usingServer2) createServerInterComm(CXios::defaultPoolId, CXios::defaultGathererId, clientServers) ;
       else createServerInterComm(CXios::defaultPoolId, CXios::defaultWriterId, clientServers) ;
       
@@ -629 +628 @@
       clientServers.clear() ;
    
-      if (attached_mode) createServerInterComm(CClient::getPoolRessource()->getId(), getContextId()+"_"+CXios::defaultReaderId, clientServers) ;
-      else createServerInterComm(CXios::defaultPoolId, CXios::defaultReaderId, clientServers) ;
+      createServerInterComm(CXios::defaultPoolId, CXios::defaultReaderId, clientServers) ;
       readerClientOut_.push_back(clientServers[0].second.first) ; 
       readerServerOut_.push_back(clientServers[0].second.second) ;
@@ -658 +656 @@
     else
     {
-      if (attached_mode) createServerInterComm(CClient::getPoolRessource()->getId(), getContextId()+"_"+serviceId, retClientServers) ;
-      else createServerInterComm(poolId, serviceId, retClientServers) ;
+      createServerInterComm(poolId, serviceId, retClientServers) ;
       for(auto& retClientServer : retClientServers)  clientServers.push_back(retClientServer.second) ;
       
       int serviceType ;
-      if (intraCommRank_==0) CXios::getServicesManager()->getServiceType(poolId, serviceId, 0, serviceType, true) ;
+      if (intraCommRank_==0) while(!CXios::getServicesManager()->getServiceType(poolId, serviceId, 0, serviceType)) yield(); 
+      synchronize() ;
       MPI_Bcast(&serviceType,1,MPI_INT,0,intraComm_) ;
       
@@ -694 +692 @@
   void CContext::globalEventLoop(void)
   {
-    lockContext() ;
-    CXios::getDaemonsManager()->eventLoop() ;
-    unlockContext() ;
-    setCurrent(getId()) ;
+    if (CThreadManager::isUsingThreads()) CThreadManager::yield();
+    else
+    {
+      lockContext() ;
+      CXios::getDaemonsManager()->eventLoop() ;
+      unlockContext() ;
+      setCurrent(getId()) ;
+    }
   }
 
+  void CContext::yield(void)
+  {
+    if (CThreadManager::isUsingThreads()) 
+    {
+      CThreadManager::yield();
+      setCurrent(getId()) ;
+    }
+    else
+    {
+      lockContext() ;
+      CXios::getDaemonsManager()->eventLoop() ;
+      unlockContext() ;
+      setCurrent(getId()) ;
+    }
+  }
+
+  void CContext::synchronize(void)
+  {
+    bool out, finished; 
+    size_t timeLine=timeLine_ ;
+      
+    timeLine_++ ;
+    eventScheduler_->registerEvent(timeLine, hashId_) ;
+      
+    out = eventScheduler_->queryEvent(timeLine,hashId_) ;
+    if (out) eventScheduler_->popEvent() ;
+    while (!out)
+    {
+      yield() ;
+      out = eventScheduler_->queryEvent(timeLine,hashId_) ;
+      if (out) eventScheduler_->popEvent() ;
+    }
+  }
+  
   bool CContext::scheduledEventLoop(bool enableEventsProcessing) 
   {
@@ -761 +797 @@
        if (couplerOutClient_.find(fullContextId)==couplerOutClient_.end()) 
        {
-         bool ok=CXios::getContextsManager()->getContextLeader(fullContextId, contextLeader, getIntraComm()) ;
+         while(!CXios::getContextsManager()->getContextLeader(fullContextId, contextLeader, getIntraComm())) yield();
+         synchronize() ;
      
          MPI_Comm interComm, interCommClient, interCommServer  ;
          MPI_Comm intraCommClient, intraCommServer ;
 
-         if (ok) MPI_Intercomm_create(getIntraComm(), 0, CXios::getXiosComm(), contextLeader, 0, &interComm) ;
+         MPI_Intercomm_create(getIntraComm(), 0, CXios::getXiosComm(), contextLeader, 0, &interComm) ;
 
         MPI_Comm_dup(intraComm_, &intraCommClient) ;
@@ -783 +820 @@
     else if (couplerInClient_.find(fullContextId)==couplerInClient_.end())
     {
-      bool ok=CXios::getContextsManager()->getContextLeader(fullContextId, contextLeader, getIntraComm()) ;
+      while(!CXios::getContextsManager()->getContextLeader(fullContextId, contextLeader, getIntraComm())) yield() ;
+      synchronize() ;
      
        MPI_Comm interComm, interCommClient, interCommServer  ;
        MPI_Comm intraCommClient, intraCommServer ;
 
-       if (ok) MPI_Intercomm_create(getIntraComm(), 0, CXios::getXiosComm(), contextLeader, 0, &interComm) ;
+       MPI_Intercomm_create(getIntraComm(), 0, CXios::getXiosComm(), contextLeader, 0, &interComm) ;
 
        MPI_Comm_dup(intraComm_, &intraCommClient) ;
@@ -824 +862 @@
           couplersInFinalized=true ;
           for(auto& couplerOutClient : couplerOutClient_) couplersInFinalized &= isCouplerInContextFinalized(couplerOutClient.second) ; 
-          globalEventLoop() ;
+          if (CThreadManager::isUsingThreads()) yield() ;
+          else globalEventLoop() ;
         } while (!couplersInFinalized) ;
 
@@ -900 +939 @@
           info(100)<<"DEBUG: context "<<getId()<<" release client reader ok"<<endl ;
         }
+        closeAllFile() ;
       }
       else if (serviceType_==CServicesManager::GATHERER)
       {
-         for(auto& client : writerClientOut_)
-         {
-           client->finalize();
-           bool bufferReleased;
-           do
-           {
-             client->eventLoop();
-             bufferReleased = !client->havePendingRequests();
-           } while (!bufferReleased);
+        CContextClient* client ;
+        CContextServer* server ;
+        
+        for(int n=0; n<writerClientOut_.size() ; n++)
+        {
+          client=writerClientOut_[n] ;
+          server=writerServerOut_[n] ;
+        
+          client->finalize();
+          bool bufferReleased;
+          do
+          {
+            client->eventLoop();
+            bufferReleased = !client->havePendingRequests();
+          } while (!bufferReleased);
            
-           bool notifiedFinalized=false ;
-           do
-           {
-             notifiedFinalized=client->isNotifiedFinalized() ;
-           } while (!notifiedFinalized) ;
-           client->releaseBuffers();
+          bool notifiedFinalized=false ;
+          do
+          {
+            notifiedFinalized=client->isNotifiedFinalized() ;
+          } while (!notifiedFinalized) ;
+          server->releaseBuffers();
+          client->releaseBuffers();
          }
          closeAllFile();
+         writerClientIn_[0]->releaseBuffers();
+         writerServerIn_[0]->releaseBuffers();         
          //ym writerClientIn & writerServerIn not released here ==> to check !!
       }
@@ -957 +1006 @@
    void CContext::setDefaultServices(void)
    {
-     defaultPoolWriterId_ = CXios::defaultPoolId ;
-     defaultPoolReaderId_ = CXios::defaultPoolId ;
-     defaultPoolGathererId_ = CXios::defaultPoolId ;
-     defaultWriterId_ = CXios::defaultWriterId ;
-     defaultReaderId_ = CXios::defaultReaderId ;
-     defaultGathererId_ = CXios::defaultGathererId ;
-     defaultUsingServer2_ = CXios::usingServer2 ;
+     if (!CXios::isUsingServer())
+     {
+       defaultPoolWriterId_ = CXios::defaultPoolId ;
+       defaultPoolReaderId_ = CXios::defaultPoolId ;
+       defaultPoolGathererId_ = CXios::defaultPoolId ;
+       defaultWriterId_ = "attached" ;
+       defaultReaderId_ = "attached" ;
+       defaultGathererId_ =  "attached" ;
+       defaultUsingServer2_ = false;
+     }
+     else
+     {
+       defaultPoolWriterId_ = CXios::defaultPoolId ;
+       defaultPoolReaderId_ = CXios::defaultPoolId ;
+       defaultPoolGathererId_ = CXios::defaultPoolId ;
+       defaultWriterId_ = CXios::defaultWriterId ;
+       defaultReaderId_ = CXios::defaultReaderId ;
+       defaultGathererId_ = CXios::defaultGathererId ;
+       defaultUsingServer2_ = CXios::usingServer2 ;
      
-     if (!default_pool.isEmpty())  defaultPoolWriterId_ = defaultPoolReaderId_= defaultPoolGathererId_= default_pool ;
-     if (!default_pool_writer.isEmpty()) defaultPoolWriterId_ = default_pool_writer ;
-     if (!default_pool_reader.isEmpty()) defaultPoolReaderId_ = default_pool_reader ;
-     if (!default_pool_gatherer.isEmpty()) defaultPoolGathererId_ = default_pool_gatherer ;
-     if (!default_writer.isEmpty()) defaultWriterId_ = default_writer ;
-     if (!default_reader.isEmpty()) defaultWriterId_ = default_reader ;
-     if (!default_gatherer.isEmpty()) defaultGathererId_ = default_gatherer ;
-     if (!default_using_server2.isEmpty()) defaultUsingServer2_ = default_using_server2 ;
+       if (!default_pool.isEmpty())  defaultPoolWriterId_ = defaultPoolReaderId_= defaultPoolGathererId_= default_pool ;
+       if (!default_pool_writer.isEmpty()) defaultPoolWriterId_ = default_pool_writer ;
+       if (!default_pool_reader.isEmpty()) defaultPoolReaderId_ = default_pool_reader ;
+       if (!default_pool_gatherer.isEmpty()) defaultPoolGathererId_ = default_pool_gatherer ;
+       if (!default_writer.isEmpty()) defaultWriterId_ = default_writer ;
+       if (!default_reader.isEmpty()) defaultWriterId_ = default_reader ;
+       if (!default_gatherer.isEmpty()) defaultGathererId_ = default_gatherer ;
+       if (!default_using_server2.isEmpty()) defaultUsingServer2_ = default_using_server2 ;
+     }
    }
 

XIOS3/trunk/src/node/context.hpp

@@ -112 +112 @@
          bool scheduledEventLoop(bool enableEventsProcessing=true) ; 
          void globalEventLoop(void);
+         void yield(void) ;
+         void synchronize(void) ;
 
          // Finalize a context
@@ -368 +370 @@
         int getIntraCommRank(void) {return intraCommRank_;}
         int getIntraCommSize(void) {return intraCommSize_;}
+      
+      public:
+        shared_ptr<CEventScheduler> getEventScheduler(void) {return eventScheduler_ ;}
       private:
          shared_ptr<CEventScheduler> eventScheduler_ ; //! The local event scheduler for context

XIOS3/trunk/src/node/pool_node.cpp

@@ -1 +1 @@
 #include "pool_node.hpp"
 #include "cxios.hpp"
+#include "thread_manager.hpp"
 #include<cmath>
 
@@ -61 +62 @@
     else ERROR("void CPoolNode::allocateRessources(void)",<<"Pool has no name or id, attributes <id> or <name> must be specified")
     ressourcesManager->createPool(poolId, nbRessources) ;
-    ressourcesManager->waitPoolRegistration(poolId) ;
+    if (CThreadManager::isUsingThreads()) 
+      while(!ressourcesManager->hasPool(CXios::defaultPoolId)) 
+      {
+        CXios::getDaemonsManager()->eventLoop() ;
+        CThreadManager::yield() ;
+      }
+    else ressourcesManager->waitPoolRegistration(poolId) ;
     auto services=this->getAllServiceNodes() ;
     for(auto& service : services) service->allocateRessources(poolId) ;

XIOS3/trunk/src/node/service_node.cpp

@@ -60 +60 @@
     else if (!hasAutoGeneratedId() ) serviceId=getId() ;
     else ERROR("void CServiceNode::allocateRessources(const string& poolId)",<<"Service has no name or id, attributes <id> or <name> must be specified")
+    
     servicesManager->createServices(poolId, serviceId, serviceType, nbRessources, nb_partitions, true) ;
+    if (CThreadManager::isUsingThreads())
+      for(int i=0; i<nb_partitions; i++)
+        while(!servicesManager->hasService(poolId, serviceId, i)) 
+        {
+          CXios::getDaemonsManager()->eventLoop() ;
+          CThreadManager::yield() ;
+        }
+    else servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultWriterId) ;
+
   }
 

XIOS3/trunk/src/server.cpp

@@ -25 +25 @@
 #include "workflow_graph.hpp"
 #include "release_static_allocation.hpp"
+#include "thread_manager.hpp"
 #include <sys/stat.h>
 #include <unistd.h>
@@ -68 +69 @@
       if (!CXios::usingOasis)
       {
-        if (!is_MPI_Initialized) MPI_Init(NULL, NULL);
+        if (!is_MPI_Initialized) 
+        {
+          int required = MPI_THREAD_SERIALIZED ;
+          int provided ;
+          MPI_Init_thread(NULL,NULL, required, &provided) ;
+        }
        
         // split the global communicator
@@ -99 +105 @@
       else // using OASIS
       {
-        if (!is_MPI_Initialized) driver_ = new CThirdPartyDriver();
+        
+        if (!is_MPI_Initialized) 
+        {
+          int required = MPI_THREAD_SERIALIZED ;
+          int provided ;
+          MPI_Init_thread(NULL,NULL, required, &provided) ;
+        }
+
+        driver_ = new CThirdPartyDriver();
 
         driver_->getComponentCommunicator( serverComm );
@@ -200 +214 @@
           {
             ressourcesManager->createPool(CXios::defaultPoolId, nbRessources) ;
-            ressourcesManager->waitPoolRegistration(CXios::defaultPoolId) ;
+            if (CThreadManager::isUsingThreads()) 
+              while(!ressourcesManager->hasPool(CXios::defaultPoolId)) 
+              {
+                daemonsManager->eventLoop() ;
+                CThreadManager::yield() ;
+              }
+            else ressourcesManager->waitPoolRegistration(CXios::defaultPoolId) ;
+          
             servicesManager->createServices(CXios::defaultPoolId, CXios::defaultWriterId, CServicesManager::WRITER,nbRessources,1) ;
-            servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultWriterId) ;
+            if (CThreadManager::isUsingThreads()) 
+              while(!servicesManager->hasService(CXios::defaultPoolId, CXios::defaultWriterId,0)) 
+              {
+                daemonsManager->eventLoop() ;
+                CThreadManager::yield() ;
+              }
+            else servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultWriterId) ;
+            
             servicesManager->createServicesOnto(CXios::defaultPoolId, CXios::defaultReaderId, CServicesManager::READER, CXios::defaultWriterId) ;
-            servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultReaderId) ;
+            if (CThreadManager::isUsingThreads()) 
+            {
+              daemonsManager->eventLoop() ;
+              while(!servicesManager->hasService(CXios::defaultPoolId, CXios::defaultReaderId, 0)) CThreadManager::yield() ;
+            }
+            else servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultReaderId) ;
           }
           else
@@ -214 +247 @@
             if (nbPoolsServer2 == 0) nbPoolsServer2 = nprocsServer;
             ressourcesManager->createPool(CXios::defaultPoolId, nbRessources) ;
-            ressourcesManager->waitPoolRegistration(CXios::defaultPoolId) ;
+            if (CThreadManager::isUsingThreads()) 
+              while(!ressourcesManager->hasPool(CXios::defaultPoolId)) 
+              {
+                daemonsManager->eventLoop() ;
+                CThreadManager::yield() ;
+              }
+            else ressourcesManager->waitPoolRegistration(CXios::defaultPoolId) ;
+
             servicesManager->createServices(CXios::defaultPoolId,  CXios::defaultGathererId, CServicesManager::GATHERER, nprocsGatherer, 1) ;
-            servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultGathererId) ;
+            if (CThreadManager::isUsingThreads()) 
+              while(!servicesManager->hasService(CXios::defaultPoolId, CXios::defaultGathererId,0)) 
+              {
+                daemonsManager->eventLoop() ;
+                CThreadManager::yield() ;
+              }
+            else servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultGathererId) ;
+
             servicesManager->createServicesOnto(CXios::defaultPoolId, CXios::defaultReaderId, CServicesManager::READER, CXios::defaultGathererId) ;
-            servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultReaderId) ;
+            if (CThreadManager::isUsingThreads()) 
+              while(!servicesManager->hasService(CXios::defaultPoolId, CXios::defaultReaderId, 0)) 
+              {
+                daemonsManager->eventLoop() ;
+                CThreadManager::yield() ;
+              }
+            else servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultReaderId) ;
+            
             servicesManager->createServices(CXios::defaultPoolId,  CXios::defaultWriterId, CServicesManager::WRITER, nprocsServer, nbPoolsServer2) ;
-            servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultWriterId) ;
+            if (CThreadManager::isUsingThreads())
+              for(int i=0; i<nbPoolsServer2; i++)
+                while(!servicesManager->hasService(CXios::defaultPoolId, CXios::defaultWriterId,i)) 
+                {
+                  daemonsManager->eventLoop() ;
+                  CThreadManager::yield() ;
+                }
+            else servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultWriterId) ;
           }
         }
@@ -233 +294 @@
       {
         daemonsManager->eventLoop() ;
+        if (CThreadManager::isUsingThreads()) CThreadManager::yield();
         MPI_Test(&req,&ok,&status) ;
       }
 
 
-      testingEventScheduler() ;
+      //testingEventScheduler() ;
 /*
       MPI_Request req ;
@@ -262 +324 @@
       {
         finished=daemonsManager->eventLoop() ;
+        if (CThreadManager::isUsingThreads()) CThreadManager::yield() ;
       }
       CTimer::get("XIOS event loop").suspend() ;

XIOS3/trunk/src/transport/context_client.cpp

@@ -40 +40 @@
 
       int flag;
-      MPI_Comm_test_inter(interComm, &flag);
-      if (flag) isAttached_=false ;
-      else  isAttached_=true ;
-
-      if (flag) MPI_Comm_remote_size(interComm, &serverSize);
-      else  MPI_Comm_size(interComm, &serverSize);
-
+      
+      MPI_Comm_remote_size(interComm, &serverSize);
+      
       computeLeader(clientRank, clientSize, serverSize, ranksServerLeader, ranksServerNotLeader);
 

XIOS3/trunk/src/transport/context_client.hpp

@@ -40 +40 @@
       const std::list<int>& getRanksServerLeader(void) const;
       const std::list<int>& getRanksServerNotLeader(void) const;
-      bool isAttachedModeEnabled() const { return isAttached_ ; } 
       static void computeLeader(int clientRank, int clientSize, int serverSize,
                                 std::list<int>& rankRecvLeader,
@@ -93 +92 @@
       size_t hashId_ ; //!< hash id on the context client that will be used for context server to identify the remote calling context client.
 
-      bool isAttached_ ;
       CContextServer* associatedServer_ ; //!< The server associated to the pair client/server
   };

XIOS3/trunk/src/transport/context_server.cpp

@@ -44 +44 @@
 
     interComm=interComm_;
-    int flag;
-    MPI_Comm_test_inter(interComm,&flag);
-
-    if (flag) attachedMode=false ;
-    else  attachedMode=true ;
+    MPI_Comm_remote_size(interComm,&clientSize_);
     
-    if (flag) MPI_Comm_remote_size(interComm,&clientSize_);
-    else  MPI_Comm_size(interComm,&clientSize_);
-
     SRegisterContextInfo contextInfo ;
     CXios::getContextsManager()->getContextInfo(context->getId(), contextInfo, intraComm) ;
@@ -89 +82 @@
   }
 
-//! Attached mode is used ?
-//! \return true if attached mode is used, false otherwise
-  bool CContextServer::isAttachedModeEnabled() const
-  {
-    return attachedMode ;
-  }
-
 }

XIOS3/trunk/src/transport/context_server.hpp

@@ -23 +23 @@
     CContextServer(CContext* parent,MPI_Comm intraComm,MPI_Comm interComm) ;
     ~CContextServer() {} 
-    bool isAttachedModeEnabled() const ;
     void setAssociatedClient(CContextClient* associatedClient) {associatedClient_=associatedClient ;}
     CContextClient* getAssociatedClient(void) { return associatedClient_ ;}
@@ -41 +40 @@
       int commSize ;
       int clientSize_ ;
-
-      bool attachedMode ;  //! true if attached mode is enabled otherwise false
 
       CContext* context ;

XIOS3/trunk/src/transport/legacy_context_client.cpp

@@ -24 +24 @@
     \param [in] intraComm_ communicator of group client
     \param [in] interComm_ communicator of group server
-    \cxtSer [in] cxtSer Pointer to context of server side. (It is only used in case of attached mode).
+    \cxtSer [in] cxtSer Pointer to context of server side. (It is only used in case of attached mode --> obsolete).
     */
     CLegacyContextClient::CLegacyContextClient(CContext* parent, MPI_Comm intraComm_, MPI_Comm interComm_, CContext* cxtSer)
@@ -31 +31 @@
     {
       pureOneSided=CXios::getin<bool>("pure_one_sided",false); // pure one sided communication (for test)
-      if (isAttachedModeEnabled()) pureOneSided=false ; // no one sided in attach mode
-
-      if (!isAttachedModeEnabled()) MPI_Intercomm_merge(interComm_,false, &interCommMerged_) ;
-      else interCommMerged_ = interComm_; // interComm_ is yet an intracommunicator in attached
-      
+      MPI_Intercomm_merge(interComm_,false, &interCommMerged_) ;
       MPI_Comm_split(intraComm_,clientRank,clientRank, &commSelf_) ; // for windows
-
+      eventScheduler_ = parent->getEventScheduler() ;  
       timeLine = 1;
     }
@@ -44 +40 @@
 
     /*!
-    In case of attached mode, the current context must be reset to context for client
     \param [in] event Event sent to server
     */
@@ -105 +100 @@
       }
       
-      if (isAttachedModeEnabled()) // couldBuffer is always true in attached mode
-      {
-        while (checkBuffers(ranks)) callGlobalEventLoop() ;
-      
-        CXios::getDaemonsManager()->scheduleContext(hashId_) ;
-        while (CXios::getDaemonsManager()->isScheduledContext(hashId_)) callGlobalEventLoop() ;
-      }
-      
-      MPI_Request req ;
-      MPI_Status status ;
-      MPI_Ibarrier(intraComm,&req) ;
-      int flag ;
-      MPI_Test(&req,&flag,&status) ;
-      while(!flag) 
-      {
-        callGlobalEventLoop() ;
-        MPI_Test(&req,&flag,&status) ;
-      }
-
-
+      synchronize() ;
       timeLine++;
     }
@@ -148 +124 @@
       list<CClientBuffer*>::iterator itBuffer;
       bool areBuffersFree;
-     
+/*     
       for (itServer = serverList.begin(); itServer != serverList.end(); itServer++)
       {
@@ -156 +132 @@
           CTokenManager* tokenManager = CXios::getRessourcesManager()->getTokenManager() ;
           size_t token = tokenManager->getToken() ;
-          while (!tokenManager->lockToken(token)) callGlobalEventLoop() ;
+          while (!tokenManager->checkToken(token)) callGlobalEventLoop() ;
           newBuffer(*itServer);
           it = buffers.find(*itServer);
           checkAttachWindows(it->second,it->first) ;
-          tokenManager->unlockToken(token) ;
+          tokenManager->updateToken(token) ;
         }
         bufferList.push_back(it->second);
       }
+*/
+      map<int,MPI_Request> attachList ;
+     
+      for (itServer = serverList.begin(); itServer != serverList.end(); itServer++)
+      {
+        it = buffers.find(*itServer);
+        if (it == buffers.end())
+        {
+          newBuffer(*itServer);
+          it = buffers.find(*itServer);
+          checkAttachWindows(it->second, it->first, attachList) ;
+        }
+        bufferList.push_back(it->second);
+      }
+      
+      while(!attachList.empty())
+      {
+        auto it = attachList.begin() ;
+        while(it!=attachList.end())
+        {
+          if (checkAttachWindows(buffers[it->first], it->first, attachList)) it=attachList.erase(it) ;
+          else ++it ;
+        }
+
+        yield() ;
+      }
+
 
       double lastTimeBuffersNotFree=0. ;
@@ -193 +196 @@
           checkBuffers();
 
-          callGlobalEventLoop() ;
+          yield() ;
         }
 
@@ -203 +206 @@
    }
 
+
+   bool CLegacyContextClient::checkAttachWindows(CClientBuffer* buffer, int rank, map<int, MPI_Request>& attachList)
+   {
+      int dummy;
+      bool ret=true; 
+
+      if (!buffer->isAttachedWindows())
+      {
+           // create windows dynamically for one-sided
+          /*
+          CTimer::get("create Windows").resume() ;
+          MPI_Comm interComm ;
+          int tag = 0 ;
+          MPI_Intercomm_create(commSelf_, 0, interCommMerged_, clientSize+rank, tag, &interComm) ;
+          MPI_Intercomm_merge(interComm, false, &winComm_[rank]) ;
+          MPI_Comm_free(&interComm) ;
+                
+          buffer->attachWindows(winComm_[rank]) ;
+          CXios::getMpiGarbageCollector().registerCommunicator(winComm_[rank]) ;
+          MPI_Barrier(winComm_[rank]) ;
+        */
+        if (attachList.count(rank)==0) 
+        {
+          MPI_Irecv(&dummy,0,MPI_INT,clientSize+rank, 21, interCommMerged_, &attachList[rank]) ;
+          ret = false ;
+        }
+        else
+        {
+          MPI_Status status ;
+          int flag ;
+          MPI_Test(&attachList[rank],&flag, &status) ;
+          if (flag)
+          {
+            CTimer::get("create Windows").resume() ;
+            MPI_Comm interComm ;
+            int tag = 0 ;
+            MPI_Intercomm_create(commSelf_, 0, interCommMerged_, clientSize+rank, tag, &interComm) ;
+            MPI_Intercomm_merge(interComm, false, &winComm_[rank]) ;
+            MPI_Comm_free(&interComm) ;
+              
+            buffer->attachWindows(winComm_[rank]) ;
+            CXios::getMpiGarbageCollector().registerCommunicator(winComm_[rank]) ;
+            MPI_Barrier(winComm_[rank]) ;
+            ret = true ;
+          }
+          else ret=false ;
+        }
+      }
+      return ret ;
+    }
+
+
    void CLegacyContextClient::eventLoop(void)
    {
@@ -211 +266 @@
    {
      locked_=true ;
-     context_->globalEventLoop() ;
+     context_->yield() ;
      locked_=false ;
+   }
+
+   void CLegacyContextClient::yield(void)
+   {
+     locked_=true ;
+     context_->yield() ;
+     locked_=false ;
+   }
+
+   void CLegacyContextClient::synchronize(void)
+   {
+     if (context_->getServiceType()!=CServicesManager::CLIENT)
+     {
+       locked_=true ;
+       context_->synchronize() ;
+       locked_=false ;
+     }    
    }
    /*!
@@ -226 +298 @@
         maxEventSizes[rank] = CXios::minBufferSize;
       }
-      
-      int considerServers = 1;
-      if (isAttachedModeEnabled()) considerServers = 0;
-      CClientBuffer* buffer = buffers[rank] = new CClientBuffer(interCommMerged_, considerServers*clientSize+rank, mapBufferSize_[rank], maxEventSizes[rank]);
+      bool hasWindows = true ;
+      CClientBuffer* buffer = buffers[rank] = new CClientBuffer(interCommMerged_, clientSize+rank, mapBufferSize_[rank], hasWindows);
       if (isGrowableBuffer_) buffer->setGrowableBuffer(1.2) ;
       else buffer->fixBuffer() ;
       // Notify the server
+     
       CBufferOut* bufOut = buffer->getBuffer(0, 4*sizeof(MPI_Aint));
       MPI_Aint sendBuff[4] ;
       sendBuff[0]=hashId_;
       sendBuff[1]=mapBufferSize_[rank];
-      sendBuff[2]=buffers[rank]->getWinAddress(0); 
-      sendBuff[3]=buffers[rank]->getWinAddress(1); 
-      info(100)<<"CLegacyContextClient::newBuffer : rank "<<rank<<" winAdress[0] "<<buffers[rank]->getWinAddress(0)<<" winAdress[1] "<<buffers[rank]->getWinAddress(1)<<endl;
-      bufOut->put(sendBuff, 4); 
+      sendBuff[2]=buffers[rank]->getWinBufferAddress(0); 
+      sendBuff[3]=buffers[rank]->getWinBufferAddress(1); 
+      info(100)<<"CLegacyContextClient::newBuffer : rank "<<rank<<" winAdress[0] "<<buffers[rank]->getWinBufferAddress(0)<<" winAdress[1] "<<buffers[rank]->getWinBufferAddress(1)<<endl;
+      bufOut->put(sendBuff,4); 
       buffer->checkBuffer(true);
-/*
-       // create windows dynamically for one-sided
-      if (!isAttachedModeEnabled())
-      { 
-        CTimer::get("create Windows").resume() ;
-        MPI_Comm interComm ;
-        MPI_Intercomm_create(commSelf_, 0, interCommMerged_, clientSize+rank, 0, &interComm) ;
-        MPI_Intercomm_merge(interComm, false, &winComm_[rank]) ;
-        CXios::getMpiGarbageCollector().registerCommunicator(winComm_[rank]) ;
-        MPI_Comm_free(&interComm) ;
-        windows_[rank].resize(2) ;
-        
-        MPI_Win_create_dynamic(MPI_INFO_NULL, winComm_[rank], &windows_[rank][0]);
-        CXios::getMpiGarbageCollector().registerWindow(windows_[rank][0]) ;
-        
-        MPI_Win_create_dynamic(MPI_INFO_NULL, winComm_[rank], &windows_[rank][1]);   
-        CXios::getMpiGarbageCollector().registerWindow(windows_[rank][1]) ;
-
-        CTimer::get("create Windows").suspend() ;
-      }
-      else
-      {
-        winComm_[rank] = MPI_COMM_NULL ;
-        windows_[rank].resize(2) ;
-        windows_[rank][0] = MPI_WIN_NULL ;
-        windows_[rank][1] = MPI_WIN_NULL ;
-      }
-      buffer->attachWindows(windows_[rank]) ;
-      if (!isAttachedModeEnabled()) MPI_Barrier(winComm_[rank]) ;
-  */     
-   }
-
-   void CLegacyContextClient::checkAttachWindows(CClientBuffer* buffer, int rank)
-   {
-      if (!buffer->isAttachedWindows())
-      {
-           // create windows dynamically for one-sided
-        if (!isAttachedModeEnabled())
-        { 
-          CTimer::get("create Windows").resume() ;
-          MPI_Comm interComm ;
-          MPI_Intercomm_create(commSelf_, 0, interCommMerged_, clientSize+rank, 0, &interComm) ;
-          MPI_Intercomm_merge(interComm, false, &winComm_[rank]) ;
-          CXios::getMpiGarbageCollector().registerCommunicator(winComm_[rank]) ;
-          MPI_Comm_free(&interComm) ;
-          windows_[rank].resize(2) ;
-      
-          MPI_Win_create_dynamic(MPI_INFO_NULL, winComm_[rank], &windows_[rank][0]);
-          CXios::getMpiGarbageCollector().registerWindow(windows_[rank][0]) ;
-      
-          MPI_Win_create_dynamic(MPI_INFO_NULL, winComm_[rank], &windows_[rank][1]);   
-          CXios::getMpiGarbageCollector().registerWindow(windows_[rank][1]) ;
-
-          CTimer::get("create Windows").suspend() ;
-          buffer->attachWindows(windows_[rank]) ;
-          MPI_Barrier(winComm_[rank]) ;
-        }
-        else
-        {
-          winComm_[rank] = MPI_COMM_NULL ;
-          windows_[rank].resize(2) ;
-          windows_[rank][0] = MPI_WIN_NULL ;
-          windows_[rank][1] = MPI_WIN_NULL ;
-          buffer->attachWindows(windows_[rank]) ;
-        }
-
-      }
-    }
-
-
+
+   }
+
+  
   
    /*!
@@ -336 +341 @@
       buffers.clear();
 
-// don't know when release windows
-
-      //if (!isAttachedModeEnabled())
-      //{  
-      //  for(auto& it : winComm_)
-      //  {
-      //    int rank = it.first ;
-      //    MPI_Win_free(&windows_[rank][0]);
-      //    MPI_Win_free(&windows_[rank][1]);
-      //    MPI_Comm_free(&winComm_[rank]) ;
-      //  }
-      //} 
+      for(auto& it : winComm_)
+      {
+        int rank = it.first ;
+      }
    }
 
@@ -469 +466 @@
   bool CLegacyContextClient::isNotifiedFinalized(void)
   {
-    if (isAttachedModeEnabled()) return true ;
-
     bool finalized = true;
     map<int,CClientBuffer*>::iterator itBuff;

XIOS3/trunk/src/transport/legacy_context_client.hpp

@@ -11 +11 @@
 #include "registry.hpp"
 #include "context_client.hpp"
+#include "window_dynamic.hpp"
 
 namespace xios
@@ -48 +49 @@
       void getBuffers(const size_t timeLine, const list<int>& serverList, const list<int>& sizeList, list<CBufferOut*>& retBuffers);
       void newBuffer(int rank);
-      void checkAttachWindows(CClientBuffer* buffer , int rank) ;
+      bool checkAttachWindows(CClientBuffer* buffer , int rank, map<int,MPI_Request>& attachList) ;
       bool checkBuffers(list<int>& ranks);
       bool checkBuffers(void);
       void callGlobalEventLoop() ;
+      void yield(void) ;
+      void synchronize(void) ;
       bool havePendingRequests(list<int>& ranks) ;
       void setGrowableBuffer(void) { isGrowableBuffer_=true;}
@@ -76 +79 @@
 
       std::map<int, MPI_Comm> winComm_ ; //! Window communicators
-      std::map<int, std::vector<MPI_Win> >windows_ ; //! one sided mpi windows to expose client buffers to servers == windows[nbServers][2]
+      std::map<int, std::vector<CWindowDynamic*> >windows_ ; //! one sided mpi windows to expose client buffers to servers == windows[nbServers][2]
       bool isGrowableBuffer_ = true ;
 
@@ -82 +85 @@
 
       bool locked_ = false ; //!< The context client is locked to avoid recursive checkBuffer
+      shared_ptr<CEventScheduler> eventScheduler_ ;
   };
 }

XIOS3/trunk/src/transport/legacy_context_server.cpp

@@ -43 +43 @@
     finished=false;
 
-    if (!isAttachedModeEnabled()) MPI_Intercomm_merge(interComm_,true,&interCommMerged_) ;
-    else interCommMerged_ = interComm_; // interComm_ is yet an intracommunicator in attached
+    MPI_Intercomm_merge(interComm_,true,&interCommMerged_) ;
     MPI_Comm_split(intraComm_, intraCommRank, intraCommRank, &commSelf_) ; // for windows
     
@@ -50 +49 @@
 
     pureOneSided=CXios::getin<bool>("pure_one_sided",false); // pure one sided communication (for test)
-    if (isAttachedModeEnabled()) pureOneSided=false ; // no one sided in attach mode
-      
   }
  
@@ -115 +112 @@
       remoteHashId_ = recvBuff[0] ;
       StdSize buffSize = recvBuff[1];
-      vector<MPI_Aint> winAdress(2) ;
-      winAdress[0]=recvBuff[2] ; winAdress[1]=recvBuff[3] ;
+      vector<MPI_Aint> winBufferAddress(2) ;
+      winBufferAddress[0]=recvBuff[2] ; winBufferAddress[1]=recvBuff[3] ;
       mapBufferSize_.insert(std::make_pair(rank, buffSize));
 
       // create windows dynamically for one-sided
-      if (!isAttachedModeEnabled())
-      { 
-        CTimer::get("create Windows").resume() ;
-        MPI_Comm interComm ;
-        MPI_Intercomm_create(commSelf_, 0, interCommMerged_, rank, 0 , &interComm) ;
-        MPI_Intercomm_merge(interComm, true, &winComm_[rank]) ;
-        CXios::getMpiGarbageCollector().registerCommunicator(winComm_[rank]) ;
-        MPI_Comm_free(&interComm) ;
-        windows_[rank].resize(2) ;
-        MPI_Win_create_dynamic(MPI_INFO_NULL, winComm_[rank], &windows_[rank][0]);
-        CXios::getMpiGarbageCollector().registerWindow(windows_[rank][0]) ;
-        MPI_Win_create_dynamic(MPI_INFO_NULL, winComm_[rank], &windows_[rank][1]);
-        CXios::getMpiGarbageCollector().registerWindow(windows_[rank][1]) ;
-        CTimer::get("create Windows").suspend() ;
-        MPI_Barrier(winComm_[rank]) ;
-      }
-      else
-      {
-        winComm_[rank] = MPI_COMM_NULL ;
-        windows_[rank].resize(2) ;
-        windows_[rank][0] = MPI_WIN_NULL ;
-        windows_[rank][1] = MPI_WIN_NULL ;
-      }   
-
-      it=(buffers.insert(pair<int,CServerBuffer*>(rank,new CServerBuffer(windows_[rank], winAdress, 0, buffSize)))).first;
+      int dummy ;
+      MPI_Send(&dummy, 0, MPI_INT, rank, 21,interCommMerged_) ;
+      CTimer::get("create Windows").resume() ;
+      MPI_Comm interComm ;
+      int tag = 0 ;
+      MPI_Intercomm_create(commSelf_, 0, interCommMerged_, rank, tag , &interComm) ;
+      MPI_Intercomm_merge(interComm, true, &winComm_[rank]) ;
+      MPI_Comm_free(&interComm) ;
+      windows_[rank].resize(2) ;
+      //MPI_Win_create_dynamic(MPI_INFO_NULL, winComm_[rank], &windows_[rank][0]);
+      //CXios::getMpiGarbageCollector().registerWindow(windows_[rank][0]) ;
+      //MPI_Win_create_dynamic(MPI_INFO_NULL, winComm_[rank], &windows_[rank][1]);
+      //CXios::getMpiGarbageCollector().registerWindow(windows_[rank][1]) ;
+      windows_[rank][0] = new CWindowDynamic() ;
+      windows_[rank][1] = new CWindowDynamic() ;
+      windows_[rank][0] -> create(winComm_[rank]) ;
+      windows_[rank][1] -> create(winComm_[rank]) ;
+      windows_[rank][0] -> setWinBufferAddress(winBufferAddress[0],0) ;
+      windows_[rank][1] -> setWinBufferAddress(winBufferAddress[1],0) ;
+      CTimer::get("create Windows").suspend() ;
+      CXios::getMpiGarbageCollector().registerCommunicator(winComm_[rank]) ;
+      MPI_Barrier(winComm_[rank]) ;
+
+      it=(buffers.insert(pair<int,CServerBuffer*>(rank,new CServerBuffer(windows_[rank], winBufferAddress, 0, buffSize)))).first;
       lastTimeLine[rank]=0 ;
       itLastTimeLine=lastTimeLine.begin() ;
@@ -230 +226 @@
   {
     CTimer::get("CLegacyContextServer::getBufferFromClient").resume() ;
-    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 && buffers[rank]->isBufferEmpty())
-        {
-          if (buffers[rank]->getBufferFromClient(timeLine, buffer, count)) processRequest(rank, buffer, count);
-          if (count >= 0) ++itLastTimeLine ;
-          break ;
-        }
+
+    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 && buffers[rank]->isBufferEmpty())
+      {
+        if (buffers[rank]->getBufferFromClient(timeLine, buffer, count)) processRequest(rank, buffer, count);
+        if (count >= 0) ++itLastTimeLine ;
+        break ;
       }
     }
@@ -280 +274 @@
       {
         size_t newSize ;
-        vector<MPI_Aint> winAdress(2) ;
-        newBuffer>>newSize>>winAdress[0]>>winAdress[1] ;
+        vector<MPI_Aint> winBufferAdress(2) ;
+        newBuffer>>newSize>>winBufferAdress[0]>>winBufferAdress[1] ;
         buffers[rank]->freeBuffer(count) ;
         delete buffers[rank] ;
-        buffers[rank] = new CServerBuffer(windows_[rank], winAdress, 0, newSize) ;
+        windows_[rank][0] -> setWinBufferAddress(winBufferAdress[0],0) ;
+        windows_[rank][1] -> setWinBufferAddress(winBufferAdress[1],0) ;
+        buffers[rank] = new CServerBuffer(windows_[rank], winBufferAdress, 0, newSize) ;
         info(100)<<"Context id "<<context->getId()<<" : Receive ChangeBufferSize from client rank "<<rank
-                 <<"  newSize : "<<newSize<<" Address : "<<winAdress[0]<<" & "<<winAdress[1]<<endl ;
+                 <<"  newSize : "<<newSize<<" Address : "<<winBufferAdress[0]<<" & "<<winBufferAdress[1]<<endl ;
       }
       else
@@ -309 +305 @@
     CEventServer* event;
     
-//    if (context->isProcessingEvent()) return ;
     if (isProcessingEvent_) return ;
-    if (isAttachedModeEnabled())
-      if (!CXios::getDaemonsManager()->isScheduledContext(remoteHashId_)) return ;
 
     it=events.find(currentTimeLine);
@@ -321 +314 @@
       if (event->isFull())
       {
-        if (!scheduled && !isAttachedModeEnabled()) // Skip event scheduling for attached mode and reception on client side
+        if (!scheduled)
         {
           eventScheduler_->registerEvent(currentTimeLine,hashId);
@@ -327 +320 @@
           scheduled=true;
         }
-        else if (isAttachedModeEnabled() || eventScheduler_->queryEvent(currentTimeLine,hashId) )
+        else if (eventScheduler_->queryEvent(currentTimeLine,hashId) )
         {
           if (!enableEventsProcessing && isCollectiveEvent(*event)) return ;
@@ -346 +339 @@
           }
           
-          if (CXios::checkEventSync)
+          if (CXios::checkEventSync && context->getServiceType()!=CServicesManager::CLIENT)
           {
             int typeId, classId, typeId_in, classId_in;
@@ -364 +357 @@
           }
 
-          if (!isAttachedModeEnabled()) eventScheduler_->popEvent() ;
-          //MPI_Barrier(intraComm) ;
-         // When using attached mode, synchronise the processes to avoid that differents event be scheduled by differents processes
-         // The best way to properly solve this problem will be to use the event scheduler also in attached mode
-         // for now just set up a MPI barrier
-//ym to be check later
-//         if (!eventScheduler_ && CXios::isServer) MPI_Barrier(intraComm) ;
-
-//         context->setProcessingEvent() ;
-         isProcessingEvent_=true ;
-         CTimer::get("Process events").resume();
-         info(100)<<"Context id "<<context->getId()<<" : Process Event "<<currentTimeLine<<" of class "<<event->classId<<" of type "<<event->type<<endl ;
-         dispatchEvent(*event);
-         CTimer::get("Process events").suspend();
-         isProcessingEvent_=false ;
-//         context->unsetProcessingEvent() ;
-         pendingEvent=false;
-         delete event;
-         events.erase(it);
-         currentTimeLine++;
-         scheduled = false;
-         if (isAttachedModeEnabled()) CXios::getDaemonsManager()->unscheduleContext() ;
+          isProcessingEvent_=true ;
+          CTimer::get("Process events").resume();
+          info(100)<<"Context id "<<context->getId()<<" : Process Event "<<currentTimeLine<<" of class "<<event->classId<<" of type "<<event->type<<endl ;
+          eventScheduler_->popEvent() ;
+          dispatchEvent(*event);
+          CTimer::get("Process events").suspend();
+          isProcessingEvent_=false ;
+          pendingEvent=false;
+          delete event;
+          events.erase(it);
+          currentTimeLine++;
+          scheduled = false;
         }
       }
@@ -409 +392 @@
   void CLegacyContextServer::freeWindows()
   {
-    //if (!isAttachedModeEnabled())
-    //{
-    //  for(auto& it : winComm_)
-    //  {
-    //    int rank = it.first ;
-    //    MPI_Win_free(&windows_[rank][0]);
-    //    MPI_Win_free(&windows_[rank][1]);
-    //    MPI_Comm_free(&winComm_[rank]) ;
-    //  }
-    //}
+    for(auto& it : winComm_)
+    {
+      int rank = it.first ;
+      delete windows_[rank][0];
+      delete windows_[rank][1];
+    }
   }
 

XIOS3/trunk/src/transport/legacy_context_server.hpp

@@ -61 +61 @@
       std::map<int, StdSize> mapBufferSize_;
       std::map<int,MPI_Comm> winComm_ ; //! Window communicators
-      std::map<int,std::vector<MPI_Win> >windows_ ; //! one sided mpi windows to expose client buffers to servers ; No memory will be attached on server side.
+      std::map<int,std::vector<CWindowDynamic*> >windows_ ; //! one sided mpi windows to expose client buffers to servers ; No memory will be attached on server side.
       bool isProcessingEvent_ ;
       size_t remoteHashId_; //!< the hash is of the calling context client

XIOS3/trunk/src/transport/one_sided_client_buffer.hpp

@@ -182 +182 @@
       int serverRank_ ;
 
-      MPI_Comm interCommMerged_; 
-      int intraServerRank_ ;
+      MPI_Comm interCommMerged_;  
+      int intraServerRank_ ; 
 
       std::list<CBuffer*> buffers_ ;

XIOS3/trunk/src/transport/one_sided_context_client.cpp

@@ -22 +22 @@
     \param [in] intraComm_ communicator of group client
     \param [in] interComm_ communicator of group server
-    \cxtSer [in] cxtSer Pointer to context of server side. (It is only used in case of attached mode).
+    \cxtSer [in] cxtSer Pointer to context of server side. (It is only used in case of attached mode --> obsolete ).
     */
     COneSidedContextClient::COneSidedContextClient(CContext* parent, MPI_Comm intraComm_, MPI_Comm interComm_, CContext* cxtSer)
@@ -30 +30 @@
       
       pureOneSided=CXios::getin<bool>("pure_one_sided",false); // pure one sided communication (for test)
-      if (isAttachedModeEnabled()) pureOneSided=false ; // no one sided in attach mode
-
-      if (!isAttachedModeEnabled()) MPI_Intercomm_merge(interComm_,false, &interCommMerged_) ;
+
+      MPI_Intercomm_merge(interComm_,false, &interCommMerged_) ;
       
       MPI_Comm_split(intraComm_,clientRank,clientRank, &commSelf_) ; // for windows
-
+      eventScheduler_ = parent->getEventScheduler() ;  
       timeLine = 1;
     }
@@ -109 +108 @@
         {
           if (CTimer::get("Blocking time").isSuspended()) CTimer::get("Blocking time").resume() ;
-          callGlobalEventLoop() ;
+          yield() ;
         } 
       }
       if (!CTimer::get("Blocking time").isSuspended()) CTimer::get("Blocking time").suspend() ;
 
-      if (isAttachedModeEnabled()) // couldBuffer is always true in attached mode
-      {
-        while (checkBuffers(ranks)) callGlobalEventLoop() ;
-      
-        CXios::getDaemonsManager()->scheduleContext(hashId_) ;
-        while (CXios::getDaemonsManager()->isScheduledContext(hashId_)) callGlobalEventLoop() ;
-      }
+
+      synchronize() ;
       
       timeLine++;
@@ -137 +131 @@
      locked_=false ;
    }
+
+   void COneSidedContextClient::yield(void)
+   {
+     locked_=true ;
+     context_->yield() ;
+     locked_=false ;
+   }
+
+   void COneSidedContextClient::synchronize(void)
+   {
+     if (context_->getServiceType()!=CServicesManager::CLIENT)
+     {
+       locked_=true ;
+       context_->synchronize() ;
+       locked_=false ;
+     }    
+   }
+
    /*!
    Make a new buffer for a certain connection to server with specific rank
@@ -271 +283 @@
   bool COneSidedContextClient::isNotifiedFinalized(void)
   {
-    if (isAttachedModeEnabled()) return true ;
 
     bool finalized = true;

XIOS3/trunk/src/transport/one_sided_context_client.hpp

@@ -48 +48 @@
       void eventLoop(void) ;
       void callGlobalEventLoop() ;
+      void yield() ;
+      void synchronize() ;
       bool havePendingRequests(list<int>& ranks) ;
 
@@ -81 +83 @@
 
       bool locked_ = false ; //!< The context client is locked to avoid recursive checkBuffer
+      shared_ptr<CEventScheduler> eventScheduler_ ;
   };
 }

XIOS3/trunk/src/transport/one_sided_context_server.cpp

@@ -42 +42 @@
     finished=false;
 
-    if (!isAttachedModeEnabled()) MPI_Intercomm_merge(interComm_,true,&interCommMerged_) ;
+    MPI_Intercomm_merge(interComm_,true,&interCommMerged_) ;
     MPI_Comm_split(intraComm_, intraCommRank, intraCommRank, &commSelf_) ; // for windows
     
@@ -48 +48 @@
 
     pureOneSided=CXios::getin<bool>("pure_one_sided",false); // pure one sided communication (for test)
-    if (isAttachedModeEnabled()) pureOneSided=false ; // no one sided in attach mode
       
   }
@@ -165 +164 @@
   
     if (isProcessingEvent_) return ;
-    if (isAttachedModeEnabled())
-      if (!CXios::getDaemonsManager()->isScheduledContext(remoteHashId_)) return ;
 
     auto it=completedEvents_.find(currentTimeLine);
@@ -174 +171 @@
       if (it->second.nbSenders == it->second.currentNbSenders)
       {
-        if (!scheduled && !isAttachedModeEnabled()) // Skip event scheduling for attached mode and reception on client side
+        if (!scheduled) 
         {
           eventScheduler_->registerEvent(currentTimeLine,hashId);
           scheduled=true;
         }
-        else if (isAttachedModeEnabled() || eventScheduler_->queryEvent(currentTimeLine,hashId) )
+        else if (eventScheduler_->queryEvent(currentTimeLine,hashId) )
         {
           //if (!enableEventsProcessing && isCollectiveEvent(event)) return ;
@@ -198 +195 @@
           }
 
-          if (!isAttachedModeEnabled()) eventScheduler_->popEvent() ;
+          eventScheduler_->popEvent() ;
 
           isProcessingEvent_=true ;
@@ -214 +211 @@
           currentTimeLine++;
           scheduled = false;
-          if (isAttachedModeEnabled()) CXios::getDaemonsManager()->unscheduleContext() ;
         }
       }
@@ -235 +231 @@
   void COneSidedContextServer::freeWindows()
   {
-    //if (!isAttachedModeEnabled())
-    //{
     //  for(auto& it : winComm_)
     //  {
@@ -244 +238 @@
     //    MPI_Comm_free(&winComm_[rank]) ;
     //  }
-    //}
   }