Changeset 1784 for XIOS/dev/dev_ym/XIOS_COUPLING/src/node/context.cpp

Timestamp:

12/12/19 18:15:14 (5 years ago)

Author:

ymipsl

Message:

• Preparing coupling functionalities.
• Make some cleaner things

YM

File:

• XIOS/dev/dev_ym/XIOS_COUPLING/src/node/context.cpp (modified) (29 diffs)

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

@@ -24 +24 @@
 #include "cxios.hpp"
 #include "client.hpp"
+#include "coupler_in.hpp"
+#include "coupler_out.hpp"
 
 namespace xios {
@@ -353 +355 @@
    {
       intraComm_=intraComm ;
+      MPI_Comm_rank(intraComm_, &intraCommRank_) ;
+      MPI_Comm_size(intraComm_, &intraCommSize_) ;
+
       serviceType_ = CServicesManager::CLIENT ;
       if (serviceType_==CServicesManager::CLIENT)
@@ -385 +390 @@
      hasServer=true;
      intraComm_=intraComm ;
+     MPI_Comm_rank(intraComm_, &intraCommRank_) ;
+     MPI_Comm_size(intraComm_, &intraCommSize_) ;
+
      serviceType_=serviceType ;
 
@@ -544 +552 @@
     }
 
+    for (auto it=couplerClient_.begin(); it!=couplerClient_.end(); ++it)
+    {
+      if (!finalized) it->second->checkBuffers();
+    }
+
+    for (auto it=couplerServer_.begin(); it!=couplerServer_.end(); ++it)
+    {
+      if (!finalized) it->second->eventLoop(enableEventsProcessing);
+    }
+
     if (server!=nullptr) if (!finalized) finished &= server->eventLoop(enableEventsProcessing);
   
@@ -549 +567 @@
   }
 
- 
+  void CContext::addCouplingChanel(const std::string& context, bool out)
+  {
+     vector<string> vectStr=splitRegex(context,"::") ;
+     string poolId=vectStr[0] ;
+     string serviceId=poolId ;
+     string contextId=vectStr[1] ;
+
+     int contextLeader ;
+     int type = CServicesManager::CLIENT ;
+     string contextName=CXios::getContextsManager()->getServerContextName(poolId, serviceId, 0, type, contextId) ;
+     
+     if (couplerClient_.find(contextName)==couplerClient_.end())
+     {
+       bool ok=CXios::getContextsManager()->getContextLeader(contextName, contextLeader, getIntraComm()) ;
+     
+       MPI_Comm interComm, interCommClient, interCommServer  ;
+       MPI_Comm intraCommClient, intraCommServer ;
+
+       if (ok) MPI_Intercomm_create(getIntraComm(), 0, CXios::getXiosComm(), contextLeader, 0, &interComm) ;
+
+       MPI_Comm_dup(intraComm_, &intraCommClient) ;
+       MPI_Comm_dup(intraComm_, &intraCommServer) ;
+       if (out)
+       {
+         MPI_Comm_dup(interComm, &interCommClient) ;
+         MPI_Comm_dup(interComm, &interCommServer) ;
+         CContextClient* client = new CContextClient(this, intraCommClient, interCommClient);
+         CContextServer* server = new CContextServer(this, intraCommServer, interCommServer);
+       }
+       else
+       {
+          MPI_Comm_dup(interComm, &interCommServer) ;
+          MPI_Comm_dup(interComm, &interCommClient) ;
+          CContextServer* server = new CContextServer(this, intraCommServer, interCommServer);
+          CContextClient* client = new CContextClient(this, intraCommClient, interCommClient);
+       }
+       MPI_Comm_free(&interComm) ;
+
+
+      // for now, we don't now which beffer size must be used for client coupler
+      // It will be evaluated later. Fix a constant size for now...
+      // set to 10Mb for development
+       map<int,size_t> bufferSize, maxEventSize ;
+       for(int i=0;i<client->getRemoteSize();i++)
+       {
+         bufferSize[i]=10000000 ;
+         maxEventSize[i]=10000000 ;
+       }
+
+       client->setBufferSize(bufferSize, maxEventSize);    
+       
+       couplerClient_[contextName] = client ;
+       couplerServer_[contextName] = server ;
+     }
+  }
   
   void CContext::globalEventLoop(void)
@@ -649 +721 @@
      postProcessing();
 
+     // Distribute files between secondary servers according to the data size
+     distributeFiles();
+
      // Check grid and calculate its distribution
      checkGridEnabledFields();
- 
-     // Distribute files between secondary servers according to the data size
-     distributeFiles();
 
      setClientServerBuffer(client, (hasClient && !hasServer));
@@ -659 +731 @@
          setClientServerBuffer(clientPrimServer[i], true);
 
+    
      if (hasClient)
-     {
-      // Send all attributes of current context to server
-      this->sendAllAttributesToServer();
-
-      // Send all attributes of current calendar
-      CCalendarWrapper::get(CCalendarWrapper::GetDefName())->sendAllAttributesToServer();
+     { 
+       if (hasServer)
+       { 
+         for (auto it=clientPrimServer.begin(); it!=clientPrimServer.end();++it) 
+         {
+           this->sendAllAttributesToServer(*it); // Send all attributes of current context to server
+           CCalendarWrapper::get(CCalendarWrapper::GetDefName())->sendAllAttributesToServer(*it); // Send all attributes of current calendar
+         }
+       }
+       else
+       {
+         this->sendAllAttributesToServer(client);   // Send all attributes of current context to server
+         CCalendarWrapper::get(CCalendarWrapper::GetDefName())->sendAllAttributesToServer(client); // Send all attributes of current calendar
+       }
+
 
       // We have enough information to send to server
@@ -779 +861 @@
     {
       buildFilterGraphOfFieldsWithReadAccess();
-      postProcessFilterGraph();
+      postProcessFilterGraph(); // For coupling in, modify this later
     }
     
@@ -849 +931 @@
        enabledFiles[i]->checkGridOfEnabledFields();       
      }
+
+     size = enabledCouplerOut.size();
+     for (int i = 0; i < size; ++i)
+     {
+       enabledCouplerOut[i]->checkGridOfEnabledFields();       
+     }
    }
    CATCH_DUMP_ATTR
@@ -872 +960 @@
       \param [in] sendToServer Flag to indicate whether calculated information will be sent
    */
-   void CContext::solveOnlyRefOfEnabledFields(bool sendToServer)
+   void CContext::solveOnlyRefOfEnabledFields(void)
    TRY
    {
@@ -878 +966 @@
      for (int i = 0; i < size; ++i)
      {
-       this->enabledFiles[i]->solveOnlyRefOfEnabledFields(sendToServer);
+       this->enabledFiles[i]->solveOnlyRefOfEnabledFields();
      }
 
@@ -884 +972 @@
      {
        this->enabledFiles[i]->generateNewTransformationGridDest();
+     }
+
+     size = this->enabledCouplerOut.size();
+     for (int i = 0; i < size; ++i)
+     {
+       this->enabledCouplerOut[i]->solveOnlyRefOfEnabledFields();
+     }
+
+     for (int i = 0; i < size; ++i)
+     {
+       this->enabledCouplerOut[i]->generateNewTransformationGridDest();
      }
    }
@@ -894 +993 @@
       \param [in] sendToServer Flag to indicate whether calculated information will be sent
    */
-   void CContext::solveAllRefOfEnabledFieldsAndTransform(bool sendToServer)
+   void CContext::solveAllRefOfEnabledFieldsAndTransform(void)
    TRY
    {
@@ -900 +999 @@
      for (int i = 0; i < size; ++i)
      {
-       this->enabledFiles[i]->solveAllRefOfEnabledFieldsAndTransform(sendToServer);
-     }
+       this->enabledFiles[i]->solveAllRefOfEnabledFieldsAndTransform();
+     }
+
+     size = this->enabledCouplerOut.size();
+     for (int i = 0; i < size; ++i)
+     {
+       this->enabledCouplerOut[i]->solveAllRefOfEnabledFieldsAndTransform();
+     }
+
    }
    CATCH_DUMP_ATTR
@@ -912 +1018 @@
      {
        this->enabledFiles[i]->buildFilterGraphOfEnabledFields(garbageCollector);
+     }
+
+     size = this->enabledCouplerOut.size();
+     for (int i = 0; i < size; ++i)
+     {
+       this->enabledCouplerOut[i]->buildFilterGraphOfEnabledFields(garbageCollector);
      }
    }
@@ -1002 +1114 @@
      // Résolution des héritages par référence au niveau des fichiers.
       const vector<CFile*> allFiles=CFile::getAll();
+      const vector<CCouplerIn*> allCouplerIn=CCouplerIn::getAll();
+      const vector<CCouplerOut*> allCouplerOut=CCouplerOut::getAll();
       const vector<CGrid*> allGrids= CGrid::getAll();
 
@@ -1009 +1123 @@
         for (unsigned int i = 0; i < allFiles.size(); i++)
           allFiles[i]->solveFieldRefInheritance(apply);
+
+        for (unsigned int i = 0; i < allCouplerIn.size(); i++)
+          allCouplerIn[i]->solveFieldRefInheritance(apply);
+
+        for (unsigned int i = 0; i < allCouplerOut.size(); i++)
+          allCouplerOut[i]->solveFieldRefInheritance(apply);
       }
 
@@ -1071 +1191 @@
    CATCH_DUMP_ATTR
 
+   void CContext::findEnabledCouplerIn(void)
+   TRY
+   {
+      const std::vector<CCouplerIn*> allCouplerIn = CCouplerIn::getAll();
+      bool enabled ;
+      for (size_t i = 0; i < allCouplerIn.size(); i++)
+      {
+        if (allCouplerIn[i]->enabled.isEmpty()) enabled=true ;
+        else enabled=allCouplerIn[i]->enabled ;
+        if (enabled) enabledCouplerIn.push_back(allCouplerIn[i]) ;
+      }
+   }
+   CATCH_DUMP_ATTR
+
+   void CContext::findEnabledCouplerOut(void)
+   TRY
+   {
+      const std::vector<CCouplerOut*> allCouplerOut = CCouplerOut::getAll();
+      bool enabled ;
+      for (size_t i = 0; i < allCouplerOut.size(); i++)
+      {
+        if (allCouplerOut[i]->enabled.isEmpty()) enabled=true ;
+        else enabled=allCouplerOut[i]->enabled ;
+        if (enabled) enabledCouplerOut.push_back(allCouplerOut[i]) ;
+      }
+   }
+   CATCH_DUMP_ATTR
+
+
+
+
    void CContext::distributeFiles(void)
    TRY
@@ -1669 +1820 @@
       findEnabledWriteModeFiles();
       findEnabledReadModeFiles();
+      findEnabledCouplerIn();
+      findEnabledCouplerOut();
+      
+      createCouplerInterCommunicator() ;
 
       // For now, only read files with client and only one level server
@@ -1689 +1844 @@
 
       // Only search and rebuild all reference objects of enable fields, don't transform
-      this->solveOnlyRefOfEnabledFields(false);
+      this->solveOnlyRefOfEnabledFields();
 
       // Search and rebuild all reference object of enabled fields, and transform
-      this->solveAllRefOfEnabledFieldsAndTransform(false);
+      this->solveAllRefOfEnabledFieldsAndTransform();
 
       // Find all fields with read access from the public API
@@ -1700 +1855 @@
 
       isPostProcessed = true;
+   }
+   CATCH_DUMP_ATTR
+
+   void CContext::createCouplerInterCommunicator(void)
+   TRY
+   {
+      // juste for test now, in future need an scheduler to avoid dead-lock
+      for(auto it=enabledCouplerOut.begin();it!=enabledCouplerOut.end();++it)
+      {
+        (*it)->createInterCommunicator() ;
+      }
+
+      for(auto it=enabledCouplerIn.begin();it!=enabledCouplerIn.end();++it)
+      {
+        (*it)->createInterCommunicator() ;
+      }
    }
    CATCH_DUMP_ATTR
@@ -1953 +2124 @@
    TRY
    {
-     std::set<StdString> gridIds;
+     std::set<pair<StdString,CContextClient*>> gridIds;
+
      int sizeFile = activeFiles.size();
      CFile* filePtr(NULL);
@@ -1966 +2138 @@
        {
          if (0 != enabledFields[numField]->getRelGrid())
-           gridIds.insert(CGrid::get(enabledFields[numField]->getRelGrid())->getId());
+           gridIds.insert(make_pair(CGrid::get(enabledFields[numField]->getRelGrid())->getId(),enabledFields[numField]->getContextClient()));
        }
      }
@@ -1973 +2145 @@
      StdString gridDefRoot("grid_definition");
      CGridGroup* gridPtr = CGridGroup::get(gridDefRoot);
-     std::set<StdString>::const_iterator it, itE = gridIds.end();
-     for (it = gridIds.begin(); it != itE; ++it)
-     {
-       gridPtr->sendCreateChild(*it);
-       CGrid::get(*it)->sendAllAttributesToServer();
-       CGrid::get(*it)->sendAllDomains();
-       CGrid::get(*it)->sendAllAxis();
-       CGrid::get(*it)->sendAllScalars();
+     for (auto it = gridIds.begin(); it != gridIds.end(); ++it)
+     {
+       gridPtr->sendCreateChild(it->first,it->second);
+       CGrid::get(it->first)->sendAllAttributesToServer(it->second);
+       CGrid::get(it->first)->sendAllDomains(it->second);
+       CGrid::get(it->first)->sendAllAxis(it->second);
+       CGrid::get(it->first)->sendAllScalars(it->second);
      }
    }
@@ -1989 +2160 @@
    TRY
    {
-     std::set<StdString> domainIds, axisIds, scalarIds;
+     std::set<pair<StdString,CContextClient*>> domainIds, axisIds, scalarIds;
 
      // Find all reference domain and axis of all active fields
@@ -1999 +2170 @@
        for (int j = 0; j < numEnabledFields; ++j)
        {
+         CContextClient* contextClient=enabledFields[j]->getContextClient() ;
          const std::vector<StdString>& prDomAxisScalarId = enabledFields[j]->getRefDomainAxisIds();
-         if ("" != prDomAxisScalarId[0]) domainIds.insert(prDomAxisScalarId[0]);
-         if ("" != prDomAxisScalarId[1]) axisIds.insert(prDomAxisScalarId[1]);
-         if ("" != prDomAxisScalarId[2]) scalarIds.insert(prDomAxisScalarId[2]);
+         if ("" != prDomAxisScalarId[0]) domainIds.insert(make_pair(prDomAxisScalarId[0],contextClient));
+         if ("" != prDomAxisScalarId[1]) axisIds.insert(make_pair(prDomAxisScalarId[1],contextClient));
+         if ("" != prDomAxisScalarId[2]) scalarIds.insert(make_pair(prDomAxisScalarId[2],contextClient));
        }
      }
@@ -2012 +2184 @@
      StdString scalarDefRoot("scalar_definition");
      CScalarGroup* scalarPtr = CScalarGroup::get(scalarDefRoot);
-     itE = scalarIds.end();
-     for (itScalar = scalarIds.begin(); itScalar != itE; ++itScalar)
-     {
-       if (!itScalar->empty())
+     
+     for (auto itScalar = scalarIds.begin(); itScalar != scalarIds.end(); ++itScalar)
+     {
+       if (!itScalar->first.empty())
        {
-         scalarPtr->sendCreateChild(*itScalar);
-         CScalar::get(*itScalar)->sendAllAttributesToServer();
+         scalarPtr->sendCreateChild(itScalar->first,itScalar->second);
+         CScalar::get(itScalar->first)->sendAllAttributesToServer(itScalar->second);
        }
      }
@@ -2024 +2196 @@
      StdString axiDefRoot("axis_definition");
      CAxisGroup* axisPtr = CAxisGroup::get(axiDefRoot);
-     itE = axisIds.end();
-     for (itAxis = axisIds.begin(); itAxis != itE; ++itAxis)
-     {
-       if (!itAxis->empty())
+     
+     for (auto itAxis = axisIds.begin(); itAxis != axisIds.end(); ++itAxis)
+     {
+       if (!itAxis->first.empty())
        {
-         axisPtr->sendCreateChild(*itAxis);
-         CAxis::get(*itAxis)->sendAllAttributesToServer();
+         axisPtr->sendCreateChild(itAxis->first, itAxis->second);
+         CAxis::get(itAxis->first)->sendAllAttributesToServer(itAxis->second);
        }
      }
@@ -2037 +2209 @@
      StdString domDefRoot("domain_definition");
      CDomainGroup* domPtr = CDomainGroup::get(domDefRoot);
-     itE = domainIds.end();
-     for (itDom = domainIds.begin(); itDom != itE; ++itDom)
-     {
-       if (!itDom->empty()) {
-          domPtr->sendCreateChild(*itDom);
-          CDomain::get(*itDom)->sendAllAttributesToServer();
+     
+     for (auto itDom = domainIds.begin(); itDom != domainIds.end(); ++itDom)
+     {
+       if (!itDom->first.empty()) {
+          domPtr->sendCreateChild(itDom->first, itDom->second);
+          CDomain::get(itDom->first)->sendAllAttributesToServer(itDom->second);
        }
      }