Changeset 2343

Timestamp:

06/29/22 18:19:45 (11 months ago)

Author:

ymipsl

Message:

• Implement new infrastructure for transfert protocol.
• new purelly one sided protocol is now available, the previous protocol (legacy, mix send/recv and one sided) is still available. Other specific protocol could be implemented more easilly in future.
• switch can be operate with "transport_protocol" variable in XIOS context :

ex:
<variable id="transport_protocol" type="string">one_sided</variable>

Available protocols are : one_sided, legacy or default. The default protocol is "legacy".

YM

Location:

XIOS/dev/dev_ym/XIOS_COUPLING/src

Files:

• context_client.cpp (deleted)
• context_client.hpp (deleted)
• context_server.cpp (deleted)
• context_server.hpp (deleted)
• event_client.cpp (modified) (1 diff)
• event_client.hpp (modified) (3 diffs)
• event_server.cpp (modified) (3 diffs)
• event_server.hpp (modified) (2 diffs)
• node/axis.cpp (modified) (3 diffs)
• node/context.cpp (modified) (7 diffs)
• node/domain.cpp (modified) (5 diffs)
• node/grid.cpp (modified) (2 diffs)
• node/scalar.cpp (modified) (1 diff)
• transport/context_client.cpp (added)
• transport/context_client.cpp.old (added)
• transport/context_client.hpp (added)
• transport/context_server.cpp (added)
• transport/context_server.hpp (added)
• transport/legacy_context_client.cpp (added)
• transport/legacy_context_client.hpp (added)
• transport/legacy_context_server.cpp (added)
• transport/legacy_context_server.hpp (added)
• transport/one_sided_client_buffer.cpp (added)
• transport/one_sided_client_buffer.hpp (added)
• transport/one_sided_context_client.cpp (added)
• transport/one_sided_context_client.hpp (added)
• transport/one_sided_context_server.cpp (added)
• transport/one_sided_context_server.hpp (added)
• transport/one_sided_cs_buffer_base.hpp (added)
• transport/one_sided_server_base.hpp (added)
• transport/one_sided_server_buffer.cpp (added)
• transport/one_sided_server_buffer.hpp (added)

XIOS/dev/dev_ym/XIOS_COUPLING/src/event_client.cpp

@@ -56 +56 @@
      }
    }
+
+   void CEventClient::send(size_t timeLine, int size, CBufferOut* buffer) { *buffer << size << timeLine << *itNbSenders << classId << typeId << **itMessages;}
+ 
 }

XIOS/dev/dev_ym/XIOS_COUPLING/src/event_client.hpp

@@ -17 +17 @@
       void push(int rank, int nbSender, CMessage& msg);
       void send(size_t timeLine, const std::list<int>& sizes, std::list<CBufferOut*>&); 
-
       bool isEmpty(void);
       std::list<int> getRanks(void);
@@ -24 +23 @@
       int getTypeId(void) { return typeId; }
       
+      void setFirst(void)
+      {
+        itRanks=ranks.begin() ;
+        itNbSenders=nbSenders.begin() ;
+        itMessages=messages.begin() ;
+      }
+
+      bool isFirst(void)
+      {
+        return itRanks==ranks.begin() ;
+      }
+
+      void next(void)
+      {
+        itRanks++ ;
+        if (itRanks==ranks.end()) itRanks=ranks.begin() ;
+        itNbSenders++ ;
+        if (itNbSenders==nbSenders.end()) itNbSenders=nbSenders.begin() ;
+        itMessages++ ;
+        if (itMessages==messages.end()) itMessages=messages.begin() ;
+      }
+      
+      void remove(void)
+      {
+        auto removedRank = itRanks;
+        itRanks++ ;
+        ranks.erase(removedRank) ;
+        if (itRanks==ranks.end()) itRanks=ranks.begin() ;
+
+        auto removedNbSender = itNbSenders ;
+        itNbSenders++ ;
+        nbSenders.erase(removedNbSender) ;
+        if (itNbSenders==nbSenders.end()) itNbSenders=nbSenders.begin() ;
+
+        auto removedMessage = itMessages ;
+        itMessages++ ;
+        messages.erase(removedMessage) ;
+        if (itMessages==messages.end()) itMessages=messages.begin() ;
+      }
+
+      int getRank(void) { return *itRanks ;}
+      int getNbSender(void) { return *itNbSenders ;}
+      int getSize(void) { return (*itMessages)->size() + headerSize;}
+      void send(size_t timeLine, int size, CBufferOut* buffer) ;
     private:
       int classId;
@@ -30 +73 @@
       std::list<int> nbSenders;
       std::list<CMessage*> messages;
+      
+      std::list<int>::iterator       itRanks;
+      std::list<int>::iterator       itNbSenders;
+      std::list<CMessage*>::iterator itMessages;
   };
 }

XIOS/dev/dev_ym/XIOS_COUPLING/src/event_server.cpp

@@ -35 +35 @@
     ev.buffer = new CBufferIn(buffer.ptr(), buffer.remain());
     ev.size = size;  
+    ev.startBuffer = startBuffer ; // for one sided, take ownership of buffer for now
     subEvents.push_back(ev);
   
@@ -44 +45 @@
     }
   }
+
 
   bool CEventServer::isFull(void)  
@@ -55 +57 @@
     for (it = subEvents.begin(); it != subEvents.end(); it++)
     {
-      it->serverBuffer->freeBuffer(it->size);
-      delete it->buffer;
+      if (it->serverBuffer==nullptr) // one_sided case
+      {
+        delete [] it->startBuffer ;
+        delete it->buffer;
+      }
+      else // legacy case
+      {
+        it->serverBuffer->freeBuffer(it->size);
+        delete it->buffer;
+      }
     }
   }

XIOS/dev/dev_ym/XIOS_COUPLING/src/event_server.hpp

@@ -21 +21 @@
 
     void push(int rank,CServerBuffer* serverBuffer ,char* startBuffer,int size) ;
+    int getNbSender(void) {return nbSender ;}
+
     CContextServer* getContextServer(void) { return contextServer_ ;}
     
@@ -26 +28 @@
     {
       int rank ;
+      char* startBuffer ;
       CServerBuffer* serverBuffer ;
       CBufferIn*  buffer ;

XIOS/dev/dev_ym/XIOS_COUPLING/src/node/axis.cpp

@@ -190 +190 @@
      {
        // size estimation for sendDistributedValue
-       std::unordered_map<int, vector<size_t> >::const_iterator it, ite = indSrv_[client->serverSize].end();
-       for (it = indSrv_[client->serverSize].begin(); it != ite; ++it)
+       std::unordered_map<int, vector<size_t> >::const_iterator it, ite = indSrv_[client->getRemoteSize()].end();
+       for (it = indSrv_[client->getRemoteSize()].begin(); it != ite; ++it)
        {
          size_t size = 6 * sizeof(size_t);
@@ -890 +890 @@
     if (type==EDistributionType::BANDS) // Bands distribution to send to file server
     {
-      int nbServer = client->serverSize;
-      int nbClient = client->clientSize ;
-      int rankClient = client->clientRank ;
+      int nbServer = client->getRemoteSize();
+      int nbClient = client->getIntraCommSize() ;
+      int rankClient = client->getIntraCommRank() ;
       int size = nbServer / nbClient ;
       int start ;
@@ -921 +921 @@
     else if (type==EDistributionType::NONE) // domain is not distributed ie all servers get the same local domain
     {
-      int nbServer = client->serverSize;
+      int nbServer = client->getRemoteSize();
       size_t nglo=n_glo ;
       CArray<size_t,1> indGlo(nglo) ;

XIOS/dev/dev_ym/XIOS_COUPLING/src/node/context.cpp

@@ -7 +7 @@
 #include "duration.hpp"
 
-#include "context_client.hpp"
-#include "context_server.hpp"
+#include "legacy_context_client.hpp"
+#include "legacy_context_server.hpp"
+#include "one_sided_context_client.hpp"
+#include "one_sided_context_server.hpp"
 #include "nc4_data_output.hpp"
 #include "node_type.hpp"
@@ -532 +534 @@
     comms.push_back(intraCommClient);
     // attached_mode=parentServerContext_->isAttachedMode() ; //ym probably inherited from source context
-    server = new CContextServer(this,intraComm_, interCommServer); // check if we need to dupl. intraComm_ ?
-    client = new CContextClient(this,intraCommClient,interCommClient);
+    server = CContextServer::getNew(this,intraComm_, interCommServer); // check if we need to dupl. intraComm_ ?
+    client = CContextClient::getNew(this,intraCommClient,interCommClient);
     client->setAssociatedServer(server) ;  
     server->setAssociatedClient(client) ;  
@@ -594 +596 @@
       intraCommClient=intraComm_ ;
       MPI_Comm_dup(intraComm_, &intraCommServer) ;
-      client = new CContextClient(this, intraCommClient, interCommClient);
-      server = new CContextServer(this, intraCommServer, interCommServer);
+      client = CContextClient::getNew(this, intraCommClient, interCommClient);
+      server = CContextServer::getNew(this, intraCommServer, interCommServer);
       client->setAssociatedServer(server) ;
       server->setAssociatedClient(client) ;
@@ -631 +633 @@
         MPI_Comm_dup(intraComm_, &intraCommServer) ;
 
-        CContextClient* client = new CContextClient(this, intraCommClient, interCommClient) ;
-        CContextServer* server = new CContextServer(this, intraCommServer, interCommServer) ;
+        CContextClient* client = CContextClient::getNew(this, intraCommClient, interCommClient) ;
+        CContextServer* server = CContextServer::getNew(this, intraCommServer, interCommServer) ;
         client->setAssociatedServer(server) ;
         server->setAssociatedClient(client) ;
@@ -728 +730 @@
         MPI_Comm_dup(interComm, &interCommClient) ;
         MPI_Comm_dup(interComm, &interCommServer) ;
-        CContextClient* client = new CContextClient(this, intraCommClient, interCommClient);
-        CContextServer* server = new CContextServer(this, intraCommServer, interCommServer);
+        CContextClient* client = CContextClient::getNew(this, intraCommClient, interCommClient);
+        CContextServer* server = CContextServer::getNew(this, intraCommServer, interCommServer);
         client->setAssociatedServer(server) ;
         server->setAssociatedClient(client) ;
@@ -750 +752 @@
        MPI_Comm_dup(interComm, &interCommServer) ;
        MPI_Comm_dup(interComm, &interCommClient) ;
-       CContextServer* server = new CContextServer(this, intraCommServer, interCommServer);
-       CContextClient* client = new CContextClient(this, intraCommClient, interCommClient);
+       CContextServer* server = CContextServer::getNew(this, intraCommServer, interCommServer);
+       CContextClient* client = CContextClient::getNew(this, intraCommClient, interCommClient);
        client->setAssociatedServer(server) ;
        server->setAssociatedClient(client) ;
@@ -765 +767 @@
    {
       registryOut->hierarchicalGatherRegistry() ;
-      if (server->intraCommRank==0) CXios::getRegistryManager()->merge(*registryOut) ;
+      if (server->getIntraCommRank()==0) CXios::getRegistryManager()->merge(*registryOut) ;
 
       if (serviceType_==CServicesManager::CLIENT)

XIOS/dev/dev_ym/XIOS_COUPLING/src/node/domain.cpp

@@ -162 +162 @@
      }
 
-     std::unordered_map<int, vector<size_t> >::const_iterator itIndexEnd = indSrv_[client->serverSize].end();
+     std::unordered_map<int, vector<size_t> >::const_iterator itIndexEnd = indSrv_[client->getRemoteSize()].end();
      // std::map<int, std::vector<int> >::const_iterator itWrittenIndexEnd = indWrittenSrv_.end();
-     for (size_t k = 0; k < connectedServerRank_[client->serverSize].size(); ++k)
-     {
-       int rank = connectedServerRank_[client->serverSize][k];
-       std::unordered_map<int, std::vector<size_t> >::const_iterator it = indSrv_[client->serverSize].find(rank);
+     for (size_t k = 0; k < connectedServerRank_[client->getRemoteSize()].size(); ++k)
+     {
+       int rank = connectedServerRank_[client->getRemoteSize()][k];
+       std::unordered_map<int, std::vector<size_t> >::const_iterator it = indSrv_[client->getRemoteSize()].find(rank);
        size_t idxCount = (it != itIndexEnd) ? it->second.size() : 0;
 
@@ -1883 +1883 @@
     if (type==EDistributionType::BANDS) // Bands distribution to send to file server
     {
-      int nbServer = client->serverSize;
+      int nbServer = client->getRemoteSize();
       std::vector<int> nGlobDomain(2);
       nGlobDomain[0] = this->ni_glo;
@@ -1919 +1919 @@
     if (distType==EDistributionType::BANDS) // Bands distribution to send to file server
     {
-      int nbServer = client->serverSize;
-      int nbClient = client->clientSize ;
-      int rankClient = client->clientRank ;
+      int nbServer = client->getRemoteSize();
+      int nbClient = client->getIntraCommSize() ;
+      int rankClient = client->getIntraCommRank() ;
       int size = nbServer / nbClient ;
       int start ;
@@ -1952 +1952 @@
     else if (distType==EDistributionType::COLUMNS) // Bands distribution to send to file server
     {
-      int nbServer = client->serverSize;
-      int nbClient = client->clientSize ;
-      int rankClient = client->clientRank ;
+      int nbServer = client->getRemoteSize();
+      int nbClient = client->getIntraCommSize() ;
+      int rankClient = client->getIntraCommRank() ;
       int size = nbServer / nbClient ;
       int start ;
@@ -1988 +1988 @@
     else if (distType==EDistributionType::NONE) // domain is not distributed ie all servers get the same local domain
     {
-      int nbServer = client->serverSize;
+      int nbServer = client->getRemoteSize();
       int nglo=ni_glo*nj_glo ;
       CArray<size_t,1> indGlo ;

XIOS/dev/dev_ym/XIOS_COUPLING/src/node/grid.cpp

@@ -944 +944 @@
 
      std::map<int, StdSize> dataSizes;
-     int receiverSize = client->serverSize;
+     int receiverSize = client->getRemoteSize();
      std::map<int,size_t>& dataSizeMap = bufferForWriting ? connectedDataSize_[receiverSize]: connectedDataSizeRead_;
      std::vector<int>& connectedServerRanks = bufferForWriting ? connectedServerRank_[receiverSize] : connectedServerRankRead_;
@@ -1395 +1395 @@
     else if (0 != client)
     {
-      return  (isDataDistributed() ||  (1 != client->clientSize) || (1 != client->serverSize));
+      return  (isDataDistributed() ||  (1 != client->getRemoteSize()) || (1 != client->getRemoteSize()));
     }
     else

XIOS/dev/dev_ym/XIOS_COUPLING/src/node/scalar.cpp

@@ -411 +411 @@
     else if (type==EDistributionType::NONE) // domain is not distributed ie all servers get the same local domain
     {
-      int nbServer = client->serverSize;
+      int nbServer = client->getRemoteSize();
       CArray<size_t,1> indGlo(nglo) ;
       for(size_t i=0;i<nglo;i++) indGlo(i) = i ;