Changeset 2015 for XIOS/dev/dev_ym/XIOS_ASSIM2K/src

Timestamp:

01/13/21 19:09:23 (4 years ago)

Author:

ymipsl

Message:

ASSIM2K Branch :
Improve transfer protocol using mpi matching Probe / matching receive
YM

Location:

XIOS/dev/dev_ym/XIOS_ASSIM2K/src

Files:

• context_server.cpp (modified) (3 diffs)
• context_server.hpp (modified) (2 diffs)

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

@@ -64 +64 @@
   {
     listen();
+    checkPendingProbe();
     checkPendingRequest();
     if (enableEventsProcessing)
@@ -69 +70 @@
     return finished;
   }
-
+/*
   void CContextServer::listen(void)
   {
@@ -138 +139 @@
   }
 
+*/
+
+  void CContextServer::listen(void)
+  {
+    int rank;
+    int flag;
+    int count;
+    char * addr;
+    MPI_Status status;
+    MPI_Message message ;
+    map<int,CServerBuffer*>::iterator it;
+    bool okLoop;
+
+    traceOff();
+    MPI_Improbe(MPI_ANY_SOURCE, 20,interComm,&flag,&message, &status);
+    traceOn();
+    if (flag==true) listenPendingRequest(message, status) ;
+  }
+
+  bool CContextServer::listenPendingRequest( MPI_Message &message, MPI_Status& status)
+  {
+    int count;
+    char * addr;
+    map<int,CServerBuffer*>::iterator it;
+    int rank=status.MPI_SOURCE ;
+
+    it=buffers.find(rank);
+    if (it==buffers.end()) // Receive the buffer size and allocate the buffer
+    {
+       StdSize buffSize = 0;
+       MPI_Mrecv(&buffSize, 1, MPI_LONG, &message, &status);
+       mapBufferSize_.insert(std::make_pair(rank, buffSize));
+       it=(buffers.insert(pair<int,CServerBuffer*>(rank,new CServerBuffer(buffSize)))).first;
+       return true;
+    }
+    else
+    {
+//      MPI_Get_count(&status,MPI_CHAR,&count);
+//      if (it->second->isBufferFree(count))
+//      {
+//        addr=(char*)it->second->getBuffer(count);
+//         MPI_Imrecv(addr,count,MPI_CHAR, &message, &pendingRequest[rank]);
+//         bufferRequest[rank]=addr;
+//         return true;
+//       }
+//      else
+//      {
+        pendingProbe[rank].push_back(make_pair<MPI_Message,MPI_Status>(message,status)) ;
+        return false;
+//      }
+    }
+  }
+
+  void CContextServer::checkPendingProbe(void)
+  {
+    
+    list<int> recvProbe ;
+    list<int>::iterator itRecv ;
+    map<int, list<std::pair<MPI_Message,MPI_Status> > >::iterator itProbe;
+
+    for(itProbe=pendingProbe.begin();itProbe!=pendingProbe.end();itProbe++)
+    {
+      int rank=itProbe->first ;
+      if (pendingRequest.count(rank)==0)
+      {
+        MPI_Message& message = itProbe->second.front().first ;
+        MPI_Status& status = itProbe->second.front().second ;
+        int count ;
+        MPI_Get_count(&status,MPI_CHAR,&count);
+        map<int,CServerBuffer*>::iterator it = buffers.find(rank);
+        if (it->second->isBufferFree(count))
+        {
+          char * addr;
+          addr=(char*)it->second->getBuffer(count);
+          MPI_Imrecv(addr,count,MPI_CHAR, &message, &pendingRequest[rank]);
+          bufferRequest[rank]=addr;
+          recvProbe.push_back(rank) ;
+          itProbe->second.pop_front() ;
+        }
+      }
+    }
+
+    for(itRecv=recvProbe.begin(); itRecv!=recvProbe.end(); itRecv++) if (pendingProbe[*itRecv].empty()) pendingProbe.erase(*itRecv) ;
+  }
 
   void CContextServer::checkPendingRequest(void)

XIOS/dev/dev_ym/XIOS_ASSIM2K/src/context_server.hpp

@@ -17 +17 @@
     bool eventLoop(bool enableEventsProcessing = true);
     void listen(void) ;
-    bool listenPendingRequest(MPI_Status& status) ;
+    bool listenPendingRequest(MPI_Message &message, MPI_Status& status) ;
+    void checkPendingProbe(void) ;
     void checkPendingRequest(void) ;
     void processRequest(int rank, char* buff,int count) ;
@@ -34 +35 @@
 
     map<int,CServerBuffer*> buffers ;
+    map<int, list<std::pair<MPI_Message,MPI_Status> > > pendingProbe;
     map<int,MPI_Request> pendingRequest ;
     map<int,char*> bufferRequest ;