Changeset 2594 for XIOS3/trunk/src/transport/p2p_client_buffer.cpp

Timestamp:

10/25/23 11:40:07 (8 months ago)

Author:

jderouillat

Message:

Update the p2p protocol as a mirror protocol : the servers buffers will strictly mirror (number of buffers, positions of messages in the buffers) the clients buffers. The memory consumption of servers will be capped impplicitly by the clients behavior where the time spent to wait for free buffers could be present again.

File:

• XIOS3/trunk/src/transport/p2p_client_buffer.cpp (modified) (11 diffs)

XIOS3/trunk/src/transport/p2p_client_buffer.cpp

@@ -41 +41 @@
     //}
     currentWindow_=-1 ;
+    currentMirror_=-1 ;
     
     //MPI_Win_create_dynamic(MPI_INFO_NULL, winComm_, &winControl_);
@@ -59 +60 @@
   void CP2pClientBuffer::newBuffer(size_t size, bool fixed)
   { 
+    currentMirror_++;
     currentWindow_=(currentWindow_+1)%maxWindows_ ;
-    if (usedWindows_[currentWindow_]) 
-    {
-      ERROR("void CP2pClientBuffer::newBuffer(size_t size, bool fixed)",<<"Try to alloc buffer to a window already in use"<<endl ) ;
-    }
-    else usedWindows_[currentWindow_]=true ;
-    buffers_.push_back(new CBuffer(windows_[currentWindow_], size, fixed)); 
+    //if (usedWindows_[currentWindow_]) 
+    //{
+    //  ERROR("void CP2pClientBuffer::newBuffer(size_t size, bool fixed)",<<"Try to alloc buffer to a window already in use"<<endl ) ;
+    //}
+    //else usedWindows_[currentWindow_]=true ;
+    buffers_.push_back(new CBuffer(windows_[currentMirror_], size, fixed)); 
     currentBuffer_=buffers_.back() ;
     info(logProtocol)<<"   Nb attached memory blocs="<<buffers_.size()<<endl ;
@@ -106 +108 @@
         if (buffers_.empty())
         {
-          if (fixed_) newBuffer(fixedSize_,fixed_) ;
+          if (fixed_) {
+            currentBufferSize_=fixedSize_;
+            newBuffer(fixedSize_,fixed_) ;
+          }
           else
           { 
@@ -119 +124 @@
         if (count > 0) 
         {
-          blocs_.push_back({addr,currentBuffer_, start, static_cast<int>(count), currentWindow_}) ;
+          //info(logProtocol) << "Using currentMirror_ 1 : "<<currentMirror_ << endl;
+          blocs_.push_back({addr,currentBuffer_, start, static_cast<int>(count), currentMirror_}) ;
           nbBlocs++ ; 
         }
@@ -128 +134 @@
         if (count > 0) 
         {
-          blocs_.push_back({addr,currentBuffer_, start, static_cast<int>(count), currentWindow_}) ;
+          //info(logProtocol) << "Using currentMirror_ 2 : "<<currentMirror_ << endl;
+          blocs_.push_back({addr,currentBuffer_, start, static_cast<int>(count), currentMirror_}) ;
           nbBlocs++ ; 
         }
@@ -181 +188 @@
       if (buffers_.size()>1) 
       {  
-        usedWindows_[bloc.window]=false ;
+        //usedWindows_[bloc.window]=false ;
         delete buffers_.front() ;
         buffers_.pop_front() ;
@@ -279 +286 @@
     ostringstream outStr ;
     SRequest request ;
-    request.buffer = new CBufferOut(sizeof(timeline)+sizeof(nbSenders)+sizeof(nbBlocs)+(sizeof(MPI_Aint)+sizeof(int)+sizeof(int))*nbBlocs) ; 
+    request.buffer = new CBufferOut(sizeof(timeline)+sizeof(nbSenders)+sizeof(nbBlocs)+(sizeof(MPI_Aint)+sizeof(int)+sizeof(int)+sizeof(size_t))*nbBlocs) ; 
     *(request.buffer)<<timeline<<nbSenders<<nbBlocs ;
     if (info.isActive(logProtocol))  outStr<<"New timeline event sent to server rank "<<serverRank_<<" : timeLine="<<timeline<<"  nbSenders="<<nbSenders<<"  nbBlocs="<<nbBlocs<<endl ;
@@ -286 +293 @@
     for(int i=0 ; i<nbBlocs; ++i,++it) 
     {
-      *(request.buffer) << it->addr << it->count << it->window;
+      *(request.buffer) << it->addr << it->count << it->window << it->start;
     
       if (info.isActive(logProtocol))
@@ -292 +299 @@
         size_t checksum=0 ;
         for(size_t j=0;j<it->count;j++) checksum+=((unsigned char*)(it->addr))[j] ;
-        outStr<<"Bloc "<<i<<"  addr="<<it->addr<<"  count="<<it->count<<"  checksum="<<checksum<<"  ;  " ;
+        outStr<<"Bloc "<<i<<"  addr="<<it->addr<<"  count="<<it->count<<"  checksum="<<checksum<<"  window="<<it->window<<"  start="<<it->start<<"  ;  " ;
       }
 
@@ -300 +307 @@
     }
     if (info.isActive(logProtocol)) CTimer::get("sendTimelineEvent : MPI_Isend").resume() ;
+    //info(logProtocol) << "Send event : " << request.buffer->count() << endl;
     MPI_Isend(request.buffer->start(),request.buffer->count(), MPI_CHAR, intraServerRank_, 20, interCommMerged_, &request.mpiRequest ) ;
     if (info.isActive(logProtocol)) CTimer::get("sendTimelineEvent : MPI_Isend").suspend() ;
@@ -311 +319 @@
     request.buffer = new CBufferOut(sizeof(EVENT_BUFFER_RESIZE)+sizeof(timeline)+sizeof(size)) ; 
     *(request.buffer)<<EVENT_BUFFER_RESIZE<<timeline<<size ;
+    //info(logProtocol) << "Send resize : " << request.buffer->count() << endl;
     MPI_Isend(request.buffer->start(),request.buffer->count(), MPI_CHAR, intraServerRank_, 20, interCommMerged_, &request.mpiRequest ) ;
     requests_.push_back(request) ;