Changeset 2560

Timestamp:

09/11/23 14:31:18 (8 months ago)

Author:

jderouillat

Message:

Forced the request processing, on servers, to be ordered in the P2P protocol

Location:

XIOS3/trunk/src/transport

Files:

• p2p_context_server.cpp (modified) (2 diffs)
• p2p_context_server.hpp (modified) (2 diffs)

XIOS3/trunk/src/transport/p2p_context_server.cpp

@@ -101 +101 @@
       if (flag==true)
       {
-        requests_.push_back(new CRequest(interCommMerged_, status)) ;
-        if (requests_.back()->test()) 
+        int rank=status.MPI_SOURCE ;
+        requests_[rank].push_back(new CRequest(interCommMerged_, status)) ;
+        // Test 1st request of the list, request treatment must be ordered 
+        if (requests_[rank].front()->test())
         {
-          if ( processRequest( *(requests_.back()) ) )
-          {
-            delete requests_.back();
-            requests_.pop_back() ;
-          }
+          processRequest( *(requests_[rank].front()) );
+          delete requests_[rank].front();
+          requests_[rank].pop_front() ;
         }
       }
@@ -116 +116 @@
   void CP2pContextServer::listenPendingRequest(void)
   {
-    auto it = requests_.begin() ;
-    while (it != requests_.end())
-    {
-      if ((*it)->test())
-      {
-        if (processRequest(*(*it)))
-        {
-          delete (*it);
-          auto it2=it ;
-          ++it ;
-          requests_.erase(it2) ;
-        }
-      }
-      else ++it ;
-    }
-  }
-
-  bool CP2pContextServer::processRequest(CRequest& request)
+    for(auto it_rank=requests_.begin() ; it_rank!=requests_.end() ; ++it_rank)
+    {
+      int rank = it_rank->first;
+      while ( (!requests_[rank].empty()) && (requests_[rank].front()->test()) )
+      {
+        processRequest( *(requests_[rank].front()) );
+        delete requests_[rank].front();
+        requests_[rank].pop_front() ;
+      }
+    }
+  }
+
+  void CP2pContextServer::processRequest(CRequest& request)
   {
     int rank = request.getRank() ;
     auto it=buffers_.find(rank);
-    // getCount(new CP2pServerBuffer) = sizeof(MPI_AINT)
-    // getCount(RESIZE) : size_t timeline, size_t size + size_t EVENT_BUFFER_RESIZE = 24
-    // getCount(HEADER) : size_t timeline, int nbSenders, int nbBlocs
-    //                     + nbBlocs * (sizeof(MPI_Aint) addr + int count + int window) = 16 + nbBlocs * 16
-    if ((it==buffers_.end())&&(request.getCount() < 3*sizeof(size_t)))
+    if (it==buffers_.end())
     {
       buffers_[rank] = new CP2pServerBuffer(rank, commSelf_, interCommMerged_, pendingEvents_, completedEvents_, request.getBuffer()) ;
-      return true;
-    }
-    else if (it!=buffers_.end()) {
+    }
+    else
+    {
       it->second->receivedRequest(request.getBuffer()) ;
-      return true;
-    }
-    else
-      return false;
+    }
   }
 

XIOS3/trunk/src/transport/p2p_context_server.hpp

@@ -63 +63 @@
     void listen(void) ;
     void listenPendingRequest(void) ;
-    bool processRequest(CRequest& request) ;
+    void processRequest(CRequest& request) ;
     void checkBuffers(void) ;
     void processEvents(bool enableEventsProcessing) ;
@@ -106 +106 @@
       MPI_Request processEventRequest_ ;
 
-      std::list<CRequest*> requests_ ;
+      std::map<int,std::list<CRequest*> >requests_ ;
 
       std::map<size_t, SPendingEvent> pendingEvents_   ;