Changeset 2130 for XIOS/dev/dev_ym/XIOS_COUPLING/src/context_server.cpp

Timestamp:

04/20/21 09:49:47 (3 years ago)

Author:

ymipsl

Message:

New management of client-server buffers.

• buffers can grow automatically in intialization phase
• buffers is evaluated after the close context definition phase and fixed at optimal value.

YM

File:

• XIOS/dev/dev_ym/XIOS_COUPLING/src/context_server.cpp (modified) (5 diffs)

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

@@ -20 +20 @@
 #include "services.hpp"
 #include "contexts_manager.hpp"
+#include "timeline_events.hpp"
 
 #include <boost/functional/hash.hpp>
@@ -188 +189 @@
     if (it==buffers.end()) // Receive the buffer size and allocate the buffer
     {
-       MPI_Aint recvBuff[3] ;
-       MPI_Recv(recvBuff, 3, MPI_AINT, rank, 20, interComm, &status);
-       StdSize buffSize = recvBuff[0];
+       MPI_Aint recvBuff[4] ;
+       MPI_Recv(recvBuff, 4, MPI_AINT, rank, 20, interComm, &status);
+       remoteHashId_ = recvBuff[0] ;
+       StdSize buffSize = recvBuff[1];
        vector<MPI_Aint> winAdress(2) ;
-       winAdress[0]=recvBuff[1] ; winAdress[1]=recvBuff[2] ;
+       winAdress[0]=recvBuff[2] ; winAdress[1]=recvBuff[3] ;
        mapBufferSize_.insert(std::make_pair(rank, buffSize));
        it=(buffers.insert(pair<int,CServerBuffer*>(rank,new CServerBuffer(windows, winAdress, rank, buffSize)))).first;
@@ -290 +292 @@
       CBufferIn newBuffer(startBuffer,buffer.remain());
       newBuffer>>size>>timeLine;
-      it=events.find(timeLine);
-      if (it==events.end()) it=events.insert(pair<int,CEventServer*>(timeLine,new CEventServer(this))).first;
-      it->second->push(rank,buffers[rank],startBuffer,size);
-
+
+      if (timeLine==timelineEventNotifyChangeBufferSize)
+      {
+        buffers[rank]->notifyBufferResizing() ;
+        buffers[rank]->updateCurrentWindows() ;
+      } 
+      else if (timeLine==timelineEventChangeBufferSize)
+      {
+        size_t newSize ;
+        vector<MPI_Aint> winAdress(2) ;
+        newBuffer>>newSize>>winAdress[0]>>winAdress[1] ;
+        buffers.erase(rank) ;
+        buffers.insert(pair<int,CServerBuffer*>(rank,new CServerBuffer(windows, winAdress, rank, newSize)));
+      }
+      else
+      {
+        it=events.find(timeLine);
+        if (it==events.end()) it=events.insert(pair<int,CEventServer*>(timeLine,new CEventServer(this))).first;
+        it->second->push(rank,buffers[rank],startBuffer,size);
+        if (timeLine>0) lastTimeLine[rank]=timeLine ;
+      }
       buffer.advance(size);
       count=buffer.remain();
     }
-
-    if (timeLine>0) lastTimeLine[rank]=timeLine ;
     
     CTimer::get("Process request").suspend();
@@ -310 +327 @@
 //    if (context->isProcessingEvent()) return ;
     if (isProcessingEvent_) return ;
+    if (isAttachedModeEnabled())
+      if (!CXios::getDaemonsManager()->isScheduledContext(remoteHashId_)) return ;
 
     it=events.find(currentTimeLine);
@@ -356 +375 @@
          currentTimeLine++;
          scheduled = false;
+         if (isAttachedModeEnabled()) CXios::getDaemonsManager()->unscheduleContext() ;
         }
       }