Changeset 1757 for XIOS/dev/dev_ym/XIOS_ONE_SIDED/src/context_server.cpp

Timestamp:

10/18/19 14:55:57 (5 years ago)

Author:

ymipsl

Message:

Implement one sided communication in client/server protocol to avoid dead-lock when some buffer are full.

YM

File:

• XIOS/dev/dev_ym/XIOS_ONE_SIDED/src/context_server.cpp (modified) (12 diffs)

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

@@ -33 +33 @@
     int flag;
     MPI_Comm_test_inter(interComm,&flag);
+
+    if (flag) attachedMode=false ;
+    else  attachedMode=true ;
+    
     if (flag) MPI_Comm_remote_size(interComm,&commSize);
     else  MPI_Comm_size(interComm,&commSize);
 
-    currentTimeLine=0;
+     
+    currentTimeLine=1;
     scheduled=false;
     finished=false;
@@ -44 +49 @@
     else
       hashId=hashString(context->getId());
-  }
-
+
+    if (!isAttachedModeEnabled())
+    {
+      MPI_Intercomm_merge(interComm_,true,&interCommMerged) ;
+// create windows for one sided comm
+      int interCommMergedRank;
+      MPI_Comm winComm ;
+      MPI_Comm_rank(intraComm, &interCommMergedRank);
+      windows.resize(2) ;
+      for(int rank=commSize; rank<commSize+intraCommSize; rank++)
+      {
+        if (rank==commSize+interCommMergedRank) 
+        {
+          MPI_Comm_split(interCommMerged, interCommMergedRank, rank, &winComm);
+          int myRank ;
+          MPI_Comm_rank(winComm,&myRank);
+          MPI_Win_create_dynamic(MPI_INFO_NULL, winComm, &windows[0]);
+          MPI_Win_create_dynamic(MPI_INFO_NULL, winComm, &windows[1]);      
+        }
+        else MPI_Comm_split(interCommMerged, interCommMergedRank, rank, &winComm);
+        MPI_Comm_free(&winComm) ;
+      }
+    }
+    else 
+    {
+      windows.resize(2) ;
+      windows[0]=MPI_WIN_NULL ;
+      windows[1]=MPI_WIN_NULL ;
+    }
+
+
+    
+    MPI_Comm_split(intraComm_,intraCommRank,intraCommRank, &commSelf) ;
+    itLastTimeLine=lastTimeLine.begin() ;
+
+    pureOneSided=CXios::getin<bool>("pure_one_sided",false); // pure one sided communication (for test)
+    if (isAttachedModeEnabled()) pureOneSided=false ; // no one sided in attach mode
+      
+  }
+
+//! Attached mode is used ?
+//! \return true if attached mode is used, false otherwise
+  bool CContextServer::isAttachedModeEnabled() const
+  {
+    return attachedMode ;
+  }
+  
   void CContextServer::setPendingEvent(void)
   {
@@ -65 +115 @@
     listen();
     checkPendingRequest();
-    if (enableEventsProcessing)
-      processEvents();
+    if (enableEventsProcessing)  processEvents();
     return finished;
   }
@@ -117 +166 @@
     if (it==buffers.end()) // Receive the buffer size and allocate the buffer
     {
-       StdSize buffSize = 0;
-       MPI_Recv(&buffSize, 1, MPI_LONG, rank, 20, interComm, &status);
+       MPI_Aint recvBuff[3] ;
+       MPI_Recv(recvBuff, 3, MPI_AINT, rank, 20, interComm, &status);
+       StdSize buffSize = recvBuff[0];
+       vector<MPI_Aint> winAdress(2) ;
+       winAdress[0]=recvBuff[1] ; winAdress[1]=recvBuff[2] ;
        mapBufferSize_.insert(std::make_pair(rank, buffSize));
-       it=(buffers.insert(pair<int,CServerBuffer*>(rank,new CServerBuffer(buffSize)))).first;
+       it=(buffers.insert(pair<int,CServerBuffer*>(rank,new CServerBuffer(windows, winAdress, rank, buffSize)))).first;
+      /*
+       if (!isAttachedModeEnabled())
+       {
+         MPI_Comm OneSidedInterComm, oneSidedComm ;
+         MPI_Intercomm_create(commSelf, 0, interCommMerged, rank, 0, &OneSidedInterComm );
+         MPI_Intercomm_merge(OneSidedInterComm,true,&oneSidedComm);
+         buffers[rank]->createWindows(oneSidedComm) ;
+       }
+       */
+       lastTimeLine[rank]=0 ;
+       itLastTimeLine=lastTimeLine.begin() ;
+
        return true;
     }
@@ -157 +221 @@
       if (flag==true)
       {
+        buffers[rank]->updateCurrentWindows() ;
         recvRequest.push_back(rank);
         MPI_Get_count(&status,MPI_CHAR,&count);
@@ -170 +235 @@
   }
 
+  void CContextServer::getBufferFromClient(size_t timeLine)
+  {
+    if (!isAttachedModeEnabled()) // one sided desactivated in attached mode
+    {  
+      int rank ;
+      char *buffer ;
+      size_t count ; 
+
+      if (itLastTimeLine==lastTimeLine.end()) itLastTimeLine=lastTimeLine.begin() ;
+      for(;itLastTimeLine!=lastTimeLine.end();++itLastTimeLine)
+      {
+        rank=itLastTimeLine->first ;
+        if (itLastTimeLine->second < timeLine &&  pendingRequest.count(rank)==0)
+        {
+          if (buffers[rank]->getBufferFromClient(timeLine, buffer, count))
+          {
+            processRequest(rank, buffer, count);
+            break ;
+          }
+        }
+      }
+    }
+  }
+         
+       
   void CContextServer::processRequest(int rank, char* buff,int count)
   {
@@ -176 +266 @@
     char* startBuffer,endBuffer;
     int size, offset;
-    size_t timeLine;
+    size_t timeLine=0;
     map<size_t,CEventServer*>::iterator it;
 
+    
     CTimer::get("Process request").resume();
     while(count>0)
@@ -185 +276 @@
       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)).first;
@@ -193 +283 @@
       count=buffer.remain();
     }
+
+    if (timeLine>0) lastTimeLine[rank]=timeLine ;
+    
     CTimer::get("Process request").suspend();
   }
@@ -230 +323 @@
         }
       }
-    }
+      else getBufferFromClient(currentTimeLine) ;
+    }
+    else if (pureOneSided) getBufferFromClient(currentTimeLine) ; // if pure one sided check buffer even if no event recorded at current time line
   }
 
@@ -237 +332 @@
     map<int,CServerBuffer*>::iterator it;
     for(it=buffers.begin();it!=buffers.end();++it) delete it->second;
+  }
+
+  void CContextServer::releaseBuffers()
+  {
+    map<int,CServerBuffer*>::iterator it;
+    bool out ;
+    do
+    {
+      out=true ;
+      for(it=buffers.begin();it!=buffers.end();++it)
+      {
+//        out = out && it->second->freeWindows() ;
+
+      }
+    } while (! out) ; 
+  }
+
+  void CContextServer::notifyClientsFinalize(void)
+  {
+    for(auto it=buffers.begin();it!=buffers.end();++it)
+    {
+      it->second->notifyClientFinalize() ;
+    }
   }
 
@@ -254 +372 @@
       finished=true;
       info(20)<<" CContextServer: Receive context <"<<context->getId()<<"> finalize."<<endl;
+//      releaseBuffers() ;
+      notifyClientsFinalize() ;
       context->finalize();
+
+/* don't know where release windows
+      MPI_Win_free(&windows[0]) ;
+      MPI_Win_free(&windows[1]) ;
+*/     
       std::map<int, StdSize>::const_iterator itbMap = mapBufferSize_.begin(),
                            iteMap = mapBufferSize_.end(), itMap;