Changeset 2558

Timestamp:

09/06/23 14:53:33 (8 months ago)

Author:

ymipsl

Message:

p2p transport protocol

• bug fix
• more diagnostics
• set buffer parameters to realistic values

YM

Location:

XIOS3/trunk/src

Files:

• buffer_server.cpp (modified) (4 diffs)
• buffer_server.hpp (modified) (2 diffs)
• transport/legacy_context_server.cpp (modified) (2 diffs)
• transport/one_sided_client_buffer.hpp (modified) (1 diff)
• transport/one_sided_server_buffer.hpp (modified) (1 diff)
• transport/p2p_client_buffer.cpp (modified) (3 diffs)
• transport/p2p_client_buffer.hpp (modified) (2 diffs)
• transport/p2p_server_buffer.cpp (modified) (5 diffs)
• transport/p2p_server_buffer.hpp (modified) (3 diffs)

XIOS3/trunk/src/buffer_server.cpp

@@ -9 +9 @@
 {
 
-  CServerBuffer::CServerBuffer(vector<CWindowDynamic*>& windows, vector<MPI_Aint>& winAddress, int windowsRank, StdSize buffSize) 
-  : hasWindows(true), windows_(windows), windowsRank_(windowsRank), winAddress_(winAddress)
+  CServerBuffer::CServerBuffer(int clientRank, vector<CWindowDynamic*>& windows, vector<MPI_Aint>& winAddress, int windowsRank, StdSize buffSize) 
+  : hasWindows(true), clientRank_(clientRank), windows_(windows), windowsRank_(windowsRank), winAddress_(winAddress)
   {
     size = 3 * buffSize;
@@ -240 +240 @@
     bool ok=false ;
     
-   
+    info(100)<<"getBufferFromClient : check data in client buffer  : clientRank "<<clientRank_<<" timeline "<<timeLine<<" ??"<< endl ;
     lockBuffer(); 
     CTimer::get("getBufferFromClient_locked").resume() ;   
@@ -272 +272 @@
       for(size_t i=(count<10)?0:count-10; i<count ; i++) checksumLast=checksumLast+buffer[i] ;
       
-      info(40)<<"getBufferFromClient timeLine==clientTimeLine: windowsRank "<<windowsRank_<<" timeline "<<timeLine<<" clientTimeline "
+      info(40)<<"getBufferFromClient timeLine==clientTimeLine: clientRank "<<clientRank_<<" timeline "<<timeLine<<" clientTimeline "
               <<clientTimeline<<" clientCount "<<count<<" checksum "<<(int)checksum<<" "
               <<(int)buffer[0]<<" "<<(int)buffer[1]<<" "<<(int)buffer[2]<<" "<<(int)buffer[3]<<" "<<(int)buffer[4]<<" "<<(int)buffer[5]<<" " 
@@ -286 +286 @@
       unlockBuffer() ;
     }
-    CTimer::get("getBufferFromClient").suspend() ;   
+    CTimer::get("getBufferFromClient").suspend() ;  
+    info(100)<<"getBufferFromClient : check data in client buffer ==> done"<<endl ;
+
     if (ok) return true ;
 

XIOS3/trunk/src/buffer_server.hpp

@@ -15 +15 @@
   {
     public:
-      CServerBuffer(vector<CWindowDynamic*>& windows, vector<MPI_Aint>& winAddress, int windowsRank, StdSize bufSize) ;
+      CServerBuffer(int clientRank, vector<CWindowDynamic*>& windows, vector<MPI_Aint>& winAddress, int windowsRank, StdSize bufSize) ;
       ~CServerBuffer() ;
 
@@ -44 +44 @@
       int currentWindows ;
       bool hasWindows ;
+      int clientRank_ ; // for debugging
       int windowsRank_ ;
       double bufferFromClientLatency_=1e-1 ;

XIOS3/trunk/src/transport/legacy_context_server.cpp

@@ -140 +140 @@
       MPI_Barrier(winComm_[rank]) ;
 
-      it=(buffers.insert(pair<int,CServerBuffer*>(rank,new CServerBuffer(windows_[rank], winBufferAddress, 0, buffSize)))).first;
+      it=(buffers.insert(pair<int,CServerBuffer*>(rank, new CServerBuffer(rank, windows_[rank], winBufferAddress, 0, buffSize)))).first;
       lastTimeLine[rank]=0 ;
       itLastTimeLine=lastTimeLine.begin() ;
@@ -280 +280 @@
         windows_[rank][0] -> setWinBufferAddress(winBufferAdress[0],0) ;
         windows_[rank][1] -> setWinBufferAddress(winBufferAdress[1],0) ;
-        buffers[rank] = new CServerBuffer(windows_[rank], winBufferAdress, 0, newSize) ;
+        buffers[rank] = new CServerBuffer(rank, windows_[rank], winBufferAdress, 0, newSize) ;
         info(100)<<"Context id "<<context->getId()<<" : Receive ChangeBufferSize from client rank "<<rank
                  <<"  newSize : "<<newSize<<" Address : "<<winBufferAdress[0]<<" & "<<winBufferAdress[1]<<endl ;

XIOS3/trunk/src/transport/one_sided_client_buffer.hpp

@@ -194 +194 @@
       size_t fixedSize_ = 0 ;
       size_t currentBufferSize_= 0  ;
-      double growingFactor_ = 2. ; 
+      double growingFactor_ = 1.2 ; 
       MPI_Aint lastFreedBloc_=0 ;
       bool isFinalized_ = false ;

XIOS3/trunk/src/transport/one_sided_server_buffer.hpp

@@ -131 +131 @@
       size_t fixedSize_ = 0 ;
       size_t currentBufferSize_=0 ;
-      double growingFactor_ = 2. ;
-      double bufferServerFactor_=10. ;
+      double growingFactor_ = 1.2 ;
+      double bufferServerFactor_=1. ;
       
       std::list<CBuffer*> buffers_ ;

XIOS3/trunk/src/transport/p2p_client_buffer.cpp

@@ -72 +72 @@
   bool CP2pClientBuffer::isBufferFree(size_t size)
   {
+    if (sentBlocRequest_.size()> maxSentBlocRequests_) return false ;
     if (buffers_.size()>maxWindows_-1) return false ; 
     CBuffer* buffer ;
@@ -168 +169 @@
   {
     SBloc& bloc = blocs_.front() ;
+    
+    if (info.isActive(logProtocol))
+    {
+      size_t checksum=0 ;
+      for(size_t j=0;j<bloc.count;j++) checksum+=((unsigned char*)(bloc.addr))[j] ;
+      info(logProtocol)<<"free bloc sent to server rank "<<serverRank_<<" : addr="<<bloc.addr<<"  start="<<bloc.start<<"  count="<<bloc.count<<"  checksum="<<checksum<<endl ;
+    }
+
     bloc.buffer->free(bloc.start, bloc.count) ;
     if (bloc.buffer->getCount()==0) 
@@ -243 +252 @@
       if (flag==false) 
       {
-        ++it ;
+//        ++it ;
         break ;
       }

XIOS3/trunk/src/transport/p2p_client_buffer.hpp

@@ -41 +41 @@
           if (count_>0) ERROR("COneSidedClientBuffer::~CBuffer()",<<"Try to delete buffer that is not empty"<<std::endl) ;
           //MPI_Win_detach(window_, buffer_) ;
-          //MPI_Free_mem(buffer_) ;
+          MPI_Free_mem(buffer_) ;
           info(logProtocol)<<"Detach memory from windows : addr="<<(MPI_Aint)buffer_<<"   count="<<size_<<endl ;
         }
@@ -205 +205 @@
       size_t fixedSize_ = 0 ;
       size_t currentBufferSize_= 0  ;
-      double growingFactor_ = 2. ; 
+      double growingFactor_ = 1.2 ; 
       MPI_Aint lastFreedBloc_=0 ;
       bool isFinalized_ = false ;
+      int maxSentBlocRequests_ = 10000 ;
 
   } ;

XIOS3/trunk/src/transport/p2p_server_buffer.cpp

@@ -97 +97 @@
     if (pendingRmaRequests_.empty()) transferEvents() ;
 
-    if (!isLocked_)
-    {
+    //if (!isLocked_)
+    //{
       if (lastBlocToFree_!=0)
       {
@@ -110 +110 @@
         lastBlocToFree_ = 0 ;        
       }
-    }
+    //}
 
     if (buffers_.size()>1) 
@@ -164 +164 @@
                            << "  Bandwith : "<< transferedSize/time<< "byte/s"<<endl ;
           CTimer::get("lastTransfer from "+std::to_string(clientRank_)).reset() ;
+          for(int i=0;i<pendingRmaAddr_.size();i++)
+          {
+            size_t checksum=0 ;
+            unsigned char* buffer = (unsigned char*) pendingRmaAddr_[i] ;
+            for(size_t j=0;j<pendingRmaCount_[i];j++) checksum += buffer[j] ;
+            info(logProtocol)<<"Bloc transfered to adrr="<<(void*) buffer<<"  count="<<pendingRmaCount_[i]<<"  checksum="<<checksum<<endl ;
+          }
+
          }
 
@@ -170 +178 @@
         pendingRmaStatus_.clear() ;
         pendingRmaCount_.clear() ;
+        pendingRmaAddr_.clear() ;
         completedEvents_.insert(onTransferEvents_.begin(),onTransferEvents_.end()) ;
         
@@ -381 +390 @@
     pendingRmaRequests_.push_back(request) ;
     pendingRmaCount_.push_back(count) ;
+    pendingRmaAddr_.push_back(buffer->getBuffer()+start) ;
     onTransferEvents_[timeline].push_back({buffer,start,count,addr}) ;
   }

XIOS3/trunk/src/transport/p2p_server_buffer.hpp

@@ -133 +133 @@
       size_t fixedSize_ = 0 ;
       size_t currentBufferSize_=0 ;
-      double growingFactor_ = 2. ;
-      double bufferServerFactor_=10. ;
+      double growingFactor_ = 1.2 ;
+      double bufferServerFactor_=1. ;
       
       std::list<CBuffer*> buffers_ ;
@@ -147 +147 @@
       vector<MPI_Request> pendingRmaRequests_ ;
       vector<MPI_Status> pendingRmaStatus_ ;
+      vector<char*> pendingRmaAddr_ ;
       vector<int> pendingRmaCount_ ;
 
@@ -164 +165 @@
 
       MPI_Win winControl_ ;
-      bool isLocked_=false ;
+      //bool isLocked_=false ;
       const int windowRank_=0 ;
       MPI_Aint lastBlocToFree_=0 ;