Changeset 1460 for XIOS/dev/branch_openmp/src/node/file.cpp

Timestamp:

03/22/18 10:43:20 (6 years ago)

Author:

yushan

Message:

branch_openmp merged with XIOS_DEV_CMIP6@1459

File:

• XIOS/dev/branch_openmp/src/node/file.cpp (modified) (31 diffs)

XIOS/dev/branch_openmp/src/node/file.cpp

@@ -17 +17 @@
 #include "mpi.hpp"
 #include "timer.hpp"
-
+#include "server.hpp"
 
 namespace xios {
@@ -26 +26 @@
       : CObjectTemplate<CFile>(), CFileAttributes()
       , vFieldGroup(), data_out(), enabledFields()
-      , allDomainEmpty(false), isOpen(false)
+      , isOpen(false), read_client(0), checkRead(false), allZoneEmpty(false)
    {
      setVirtualFieldGroup(CFieldGroup::create(getId() + "_virtual_field_group"));
@@ -35 +35 @@
       : CObjectTemplate<CFile>(id), CFileAttributes()
       , vFieldGroup(), data_out(), enabledFields()
-      , allDomainEmpty(false), isOpen(false)
+      , isOpen(false), read_client(0), checkRead(false), allZoneEmpty(false)
     {
       setVirtualFieldGroup(CFieldGroup::create(getId() + "_virtual_field_group"));
@@ -208 +208 @@
 
    //! Initialize a file in order to write into it
-   void CFile::initFile(void)
+   void CFile::initWrite(void)
    {
       CContext* context = CContext::getCurrent();
@@ -218 +218 @@
       if (!split_freq.isEmpty())
       {
-        StdString keySuffix("CContext_"+CContext::getCurrent()->getId()+"::CFile_"+getFileOutputName()+"::") ; 
+        StdString keySuffix("CFile::"+getFileOutputName()+"::") ; 
         if (context->registryIn->foundKey(keySuffix+"splitStart") && context->registryIn->foundKey(keySuffix+"splitEnd"))
         {
@@ -229 +229 @@
         }
       }
-      isOpen = false;
-
-      allDomainEmpty = true;
+      isOpen = false;      
 
 //      if (!record_offset.isEmpty() && record_offset < 0)
@@ -238 +236 @@
       const int recordOffset = record_offset.isEmpty() ? 0 : record_offset;
 
-      // set<CAxis*> setAxis;
-      // set<CDomain*> setDomains;
       set<StdString> setAxis;
       set<StdString> setDomains;
-      
+
       std::vector<CField*>::iterator it, end = this->enabledFields.end();
       for (it = this->enabledFields.begin(); it != end; it++)
       {
-         CField* field = *it;
-         allDomainEmpty &= !field->grid->doGridHaveDataToWrite();
+         CField* field = *it;         
          std::vector<CAxis*> vecAxis = field->grid->getAxis();
          for (size_t i = 0; i < vecAxis.size(); ++i)
-            setAxis.insert(vecAxis[i]->getAxisOutputName());
-            // setAxis.insert(vecAxis[i]);
+           setAxis.insert(vecAxis[i]->getAxisOutputName());
          std::vector<CDomain*> vecDomains = field->grid->getDomains();
          for (size_t i = 0; i < vecDomains.size(); ++i)
-            setDomains.insert(vecDomains[i]->getDomainOutputName());
-            // setDomains.insert(vecDomains[i]);
+           setDomains.insert(vecDomains[i]->getDomainOutputName());
 
          field->resetNStep(recordOffset);
@@ -263 +256 @@
 
       // create sub communicator for file
-      int color = allDomainEmpty ? 0 : 1;
-      ep_lib::MPI_Comm_split(server->intraComm, color, server->intraCommRank, &fileComm);
-      if (allDomainEmpty) ep_lib::MPI_Comm_free(&fileComm);
+      createSubComFile();
 
       // if (time_counter.isEmpty()) time_counter.setValue(time_counter_attr::centered);
@@ -271 +262 @@
     }
 
-    //! Verify state of a file
-    void CFile::checkFile(void)
+    //! Initialize a file in order to write into it
+    void CFile::initRead(void)
     {
-      if (mode.isEmpty() || mode.getValue() == mode_attr::write)
-      {
-        CTimer::get("Files : create headers").resume();
-        if (!isOpen) createHeader();
-        CTimer::get("Files : create headers").suspend();
-        checkSync();
-      }
-      else
-      {
-        CTimer::get("Files : open headers").resume();
-        if (!isOpen) openInReadMode();
-        CTimer::get("Files : open headers").suspend();
-      }
-      checkSplit();
+      if (checkRead) return;
+      createSubComFile();
+      checkRead = true;
+    }
+
+    /*!
+      Create a sub communicator in which processes participate in reading/opening file
+    */
+    void CFile::createSubComFile()
+    {
+      CContext* context = CContext::getCurrent();
+      CContextServer* server = context->server;
+
+      // create sub communicator for file
+      allZoneEmpty = true;      
+      std::vector<CField*>::iterator it, end = this->enabledFields.end();
+      for (it = this->enabledFields.begin(); it != end; it++)
+      {
+         CField* field = *it;
+         bool nullGrid = (0 == field->grid);
+         allZoneEmpty &= nullGrid ? false : !field->grid->doGridHaveDataToWrite();
+      }
+
+      int color = allZoneEmpty ? 0 : 1;
+      ep_lib::MPI_Comm_split(server->intraComm, color, server->intraCommRank, &fileComm);
+      if (allZoneEmpty) ep_lib::MPI_Comm_free(&fileComm);
+    }
+
+    /*
+       Check condition to write into a file
+       For now, we only use the level-2 server to write files (if this mode is activated)
+       or classical server to do this job.
+    */
+    void CFile::checkWriteFile(void)
+    {
+      CContext* context = CContext::getCurrent();
+      // Done by classical server or secondary server
+      // This condition should be changed soon
+      if (CServer::serverLevel == 0 || CServer::serverLevel == 2)
+      {
+        if (mode.isEmpty() || mode.getValue() == mode_attr::write)
+        {
+          CTimer::get("Files : create headers").resume();
+          if (!isOpen) createHeader();
+          CTimer::get("Files : create headers").suspend();
+          checkSync();
+        }        
+        checkSplit(); // REally need this?
+      }
+    }
+
+    /*
+       Check condition to read from a file
+       For now, we only use the level-1 server to write files (if this mode is activated)
+       or classical server to do this job.
+       This function can be used by client for reading metadata
+    */
+    void CFile::checkReadFile(void)
+    {
+      CContext* context = CContext::getCurrent();
+      // Done by classical server or secondary server
+      // TODO: This condition should be changed soon. It only works with maximum number of level as 2
+      if (CServer::serverLevel == 0 || CServer::serverLevel == 1)
+      {
+        if (!mode.isEmpty() && mode.getValue() == mode_attr::read)
+        {
+          CTimer::get("Files : open headers").resume();
+          
+          if (!isOpen) openInReadMode();
+
+          CTimer::get("Files : open headers").suspend();
+        }
+        //checkSplit(); // Really need for reading?
+      }
+    }
+
+    /*!
+      Verify if a process participates in an opening-file communicator 
+      \return true if the process doesn't participate in opening file
+    */
+    bool CFile::isEmptyZone()
+    {
+      return allZoneEmpty;
     }
 
@@ -352 +412 @@
       CContextServer* server = context->server;
 
-      if (!allDomainEmpty)
+      if (!allZoneEmpty)
       {
          StdString filename = getFileOutputName();
@@ -381 +441 @@
          if (pos2!=std::string::npos)
          {
-           middlePart=filename.substr(pos1,pos2-pos1) ;           
+           middlePart=filename.substr(pos1,pos2-pos1) ;
            pos2+=strEndDate.size() ;
            lastPart=filename.substr(pos2,filename.size()-pos2) ;
@@ -434 +494 @@
            oss << lastPart ;
 
-           StdString keySuffix("CContext_"+CContext::getCurrent()->getId()+"::CFile_"+getFileOutputName()+"::") ; 
+           StdString keySuffix("CFile::"+getFileOutputName()+"::") ; 
            context->registryOut->setKey(keySuffix+"splitStart", lastSplit);
            context->registryOut->setKey(keySuffix+"splitEnd",   splitEnd);
@@ -534 +594 @@
   \brief Open an existing NetCDF file in read-only mode
   */
-  void CFile::openInReadMode(void)
+  void CFile::openInReadMode()
   {
     CContext* context = CContext::getCurrent();
     CContextServer* server = context->server;
-
-    if (!allDomainEmpty)
+    ep_lib::MPI_Comm readComm = this->fileComm;
+
+    if (!allZoneEmpty)
     {
       StdString filename = getFileOutputName();
@@ -578 +639 @@
       {
         int commSize, commRank;
-        ep_lib::MPI_Comm_size(fileComm, &commSize);
-        ep_lib::MPI_Comm_rank(fileComm, &commRank);
+        ep_lib::MPI_Comm_size(readComm, &commSize);
+        ep_lib::MPI_Comm_rank(readComm, &commRank);
 
         if (server->intraCommSize > 1)
@@ -597 +658 @@
       bool isCollective = par_access.isEmpty() || par_access == par_access_attr::collective;
       #ifdef _usingEP
-        //printf("multifile was %d\n", multifile);
-        //multifile = true;
         if (isOpen) data_out->closeFile();
-        if (time_counter_name.isEmpty()) data_in = boost::shared_ptr<CDataInput>(new CNc4DataInput(oss.str(), fileComm, multifile, isCollective));
-        else data_in = boost::shared_ptr<CDataInput>(new CNc4DataInput(oss.str(), fileComm, multifile, isCollective, time_counter_name));
+        if (time_counter_name.isEmpty()) data_in = boost::shared_ptr<CDataInput>(new CNc4DataInput(oss.str(), readComm, multifile, isCollective));
+        else data_in = boost::shared_ptr<CDataInput>(new CNc4DataInput(oss.str(), readComm, multifile, isCollective, time_counter_name));
         isOpen = true;
       #elif _usingMPI
         if (isOpen) data_out->closeFile();
-        if (time_counter_name.isEmpty()) data_in = boost::shared_ptr<CDataInput>(new CNc4DataInput(oss.str(), fileComm, multifile, isCollective));
-        else data_in = boost::shared_ptr<CDataInput>(new CNc4DataInput(oss.str(), fileComm, multifile, isCollective, time_counter_name));
+        if (time_counter_name.isEmpty()) data_in = boost::shared_ptr<CDataInput>(new CNc4DataInput(oss.str(), readComm, multifile, isCollective));
+        else data_in = boost::shared_ptr<CDataInput>(new CNc4DataInput(oss.str(), readComm, multifile, isCollective, time_counter_name));
         isOpen = true;
       #endif
@@ -615 +674 @@
    void CFile::close(void)
    {
-     if (!allDomainEmpty)
+     if (!allZoneEmpty)
        if (isOpen)
        {
@@ -622 +681 @@
          else
           this->data_in->closeFile();
+        isOpen = false;
        }
-//      if (fileComm != MPI_COMM_NULL) MPI_Comm_free(&fileComm);
-      //if (fileComm.mpi_comm != ::MPI_COMM_NULL) MPI_Comm_free(&fileComm);
+      //if (fileComm != MPI_COMM_NULL) MPI_Comm_free(&fileComm);
    }
    //----------------------------------------------------------------
@@ -633 +692 @@
 
      // Just check file and try to open it
-     CContext* context = CContext::getCurrent();
-     CContextClient* client=context->client;
-
-     // It would probably be better to call initFile() somehow
-     ep_lib::MPI_Comm_dup(client->intraComm, &fileComm);
      if (time_counter_name.isEmpty()) time_counter_name = "time_counter";
 
-     checkFile();
+     checkReadFile();
 
      for (int idx = 0; idx < enabledFields.size(); ++idx)
@@ -655 +709 @@
         // Read necessary value from file
         #pragma omp critical (_func)
-        {
-          this->data_in->readFieldAttributesValues(enabledFields[idx]);
-        }
+        this->data_in->readFieldAttributesValues(enabledFields[idx]);
+
         // Fill attributes for base reference
         enabledFields[idx]->solveGridDomainAxisBaseRef();
@@ -737 +790 @@
      {
        this->enabledFields[i]->solveOnlyReferenceEnabledField(sendToServer);
-//       this->enabledFields[i]->buildGridTransformationGraph();
+     }
+   }
+
+   void CFile::checkGridOfEnabledFields()
+   { 
+     int size = this->enabledFields.size();
+     for (int i = 0; i < size; ++i)
+     {
+       this->enabledFields[i]->checkGridOfEnabledFields();
+     }
+   }
+
+   void CFile::sendGridComponentOfEnabledFields()
+   { 
+     int size = this->enabledFields.size();
+     for (int i = 0; i < size; ++i)
+     {
+       this->enabledFields[i]->sendGridComponentOfEnabledFields();
+     }
+   }
+
+   void CFile::sendGridOfEnabledFields()
+   { 
+     int size = this->enabledFields.size();
+     for (int i = 0; i < size; ++i)
+     {
+       this->enabledFields[i]->sendGridOfEnabledFields();
      }
    }
@@ -758 +837 @@
    \param [in] sendToServer: Send all info to server (true) or only a part of it (false)
    */
-   void CFile::solveAllRefOfEnabledFields(bool sendToServer)
+   void CFile::solveAllRefOfEnabledFieldsAndTransform(bool sendToServer)
    {
      int size = this->enabledFields.size();
      for (int i = 0; i < size; ++i)
      {
-       this->enabledFields[i]->solveAllReferenceEnabledField(sendToServer);
+       this->enabledFields[i]->solveAllEnabledFieldsAndTransform();
      }
    }
@@ -782 +861 @@
 
    /*!
+    * Post-process the filter graph for each active field.
+    */
+   void CFile::postProcessFilterGraph()
+   {
+     int size = this->enabledFields.size();
+     for (int i = 0; i < size; ++i)
+     {
+       this->enabledFields[i]->checkIfMustAutoTrigger();
+     }
+   }
+
+   /*!
      Prefetching the data for enabled fields read from file.
    */
@@ -795 +886 @@
 
    /*!
+     Do all pre timestep operations for enabled fields in read mode:
+      - Check that the data excepted from server has been received
+      - Check if some filters must auto-trigger
+   */
+   void CFile::doPreTimestepOperationsForEnabledReadModeFields(void)
+   {
+     if (mode.isEmpty() || mode.getValue() != mode_attr::read)
+       return;
+
+     int size = this->enabledFields.size();
+     for (int i = 0; i < size; ++i)
+     {
+       this->enabledFields[i]->checkForLateDataFromServer();
+       this->enabledFields[i]->autoTriggerIfNeeded();
+     }
+   }
+
+   /*!
      Do all post timestep operations for enabled fields in read mode:
       - Prefetch the data read from file when needed
-      - Check that the data excepted from server has been received
    */
    void CFile::doPostTimestepOperationsForEnabledReadModeFields(void)
@@ -807 +915 @@
      for (int i = 0; i < size; ++i)
      {
-       this->enabledFields[i]->checkForLateDataFromServer();
        this->enabledFields[i]->sendReadDataRequestIfNeeded();
      }
@@ -874 +981 @@
    }
 
+   void CFile::setContextClient(CContextClient* newContextClient)
+   {
+     client = newContextClient;
+     size_t size = this->enabledFields.size();
+     for (size_t i = 0; i < size; ++i)
+     {
+       this->enabledFields[i]->setContextClient(newContextClient);
+     }
+   }
+
+   CContextClient* CFile::getContextClient()
+   {
+     return client;
+   }
+
+   void CFile::setReadContextClient(CContextClient* readContextclient)
+   {
+     read_client = readContextclient;
+   }
+
+   CContextClient* CFile::getReadContextClient()
+   {
+     return read_client;
+   }
+
    /*!
    \brief Send a message to create a field on server side
    \param[in] id String identity of field that will be created on server
    */
-   void CFile::sendAddField(const string& id)
-   {
-    CContext* context = CContext::getCurrent();
-
-    if (! context->hasServer )
-    {
-       CContextClient* client = context->client;
-
-       CEventClient event(this->getType(),EVENT_ID_ADD_FIELD);
-       if (client->isServerLeader())
-       {
-         CMessage msg;
-         msg << this->getId();
-         msg << id;
-         const std::list<int>& ranks = client->getRanksServerLeader();
-         for (std::list<int>::const_iterator itRank = ranks.begin(), itRankEnd = ranks.end(); itRank != itRankEnd; ++itRank)
-           event.push(*itRank,1,msg);
-         client->sendEvent(event);
-       }
-       else client->sendEvent(event);
-    }
-
+   void CFile::sendAddField(const string& id, CContextClient* client)
+   {
+      sendAddItem(id, EVENT_ID_ADD_FIELD, client);
    }
 
@@ -906 +1019 @@
    \param[in] id String identity of field group that will be created on server
    */
-   void CFile::sendAddFieldGroup(const string& id)
-   {
-    CContext* context = CContext::getCurrent();
-    if (! context->hasServer )
-    {
-       CContextClient* client = context->client;
-
-       CEventClient event(this->getType(),EVENT_ID_ADD_FIELD_GROUP);
-       if (client->isServerLeader())
-       {
-         CMessage msg;
-         msg << this->getId();
-         msg << id;
-         const std::list<int>& ranks = client->getRanksServerLeader();
-         for (std::list<int>::const_iterator itRank = ranks.begin(), itRankEnd = ranks.end(); itRank != itRankEnd; ++itRank)
-           event.push(*itRank,1,msg);
-         client->sendEvent(event);
-       }
-       else client->sendEvent(event);
-    }
-
+   void CFile::sendAddFieldGroup(const string& id, CContextClient* client)
+   {
+      sendAddItem(id, (int)EVENT_ID_ADD_FIELD_GROUP, client);
    }
 
@@ -983 +1078 @@
    is to duplicate this value on server, too.
    */
-   void CFile::sendAddAllVariables()
+   void CFile::sendAddAllVariables(CContextClient* client)
    {
      std::vector<CVariable*> allVar = getAllVariables();
@@ -991 +1086 @@
      for (; it != itE; ++it)
      {
-       this->sendAddVariable((*it)->getId());
-       (*it)->sendAllAttributesToServer();
-       (*it)->sendValue();
-     }
-   }
-
-   /*!
-   \brief Send a message to create a variable on server side
-      A variable always belongs to a variable group
-   \param[in] id String identity of variable that will be created on server
-   */
-   void CFile::sendAddVariable(const string& id)
-   {
-    CContext* context = CContext::getCurrent();
-
-    if (! context->hasServer )
-    {
-       CContextClient* client = context->client;
-
-       CEventClient event(this->getType(),EVENT_ID_ADD_VARIABLE);
-       if (client->isServerLeader())
-       {
-         CMessage msg;
-         msg << this->getId();
-         msg << id;
-         const std::list<int>& ranks = client->getRanksServerLeader();
-         for (std::list<int>::const_iterator itRank = ranks.begin(), itRankEnd = ranks.end(); itRank != itRankEnd; ++itRank)
-           event.push(*itRank,1,msg);
-         client->sendEvent(event);
-       }
-       else client->sendEvent(event);
-    }
-
+       this->sendAddVariable((*it)->getId(), client);
+       (*it)->sendAllAttributesToServer(client);
+       (*it)->sendValue(client);
+     }
    }
 
@@ -1029 +1095 @@
    \brief Send a message to create a variable group on server side
    \param[in] id String identity of variable group that will be created on server
-   */
-   void CFile::sendAddVariableGroup(const string& id)
-   {
-    CContext* context = CContext::getCurrent();
-    if (! context->hasServer )
-    {
-       CContextClient* client = context->client;
-
-       CEventClient event(this->getType(),EVENT_ID_ADD_VARIABLE_GROUP);
-       if (client->isServerLeader())
-       {
-         CMessage msg;
-         msg << this->getId();
-         msg << id;
-         const std::list<int>& ranks = client->getRanksServerLeader();
-         for (std::list<int>::const_iterator itRank = ranks.begin(), itRankEnd = ranks.end(); itRank != itRankEnd; ++itRank)
-           event.push(*itRank,1,msg);
-         client->sendEvent(event);
-       }
-       else client->sendEvent(event);
-    }
-
+   \param [in] client client to which we will send this adding action
+   */
+   void CFile::sendAddVariableGroup(const string& id, CContextClient* client)
+   {
+      sendAddItem(id, (int)EVENT_ID_ADD_VARIABLE_GROUP, client);
+   }
+
+   /*
+     Send message to add a variable into a file within a certain client
+     \param [in] id String identity of a variable
+     \param [in] client client to which we will send this adding action
+   */
+   void CFile::sendAddVariable(const string& id, CContextClient* client)
+   {
+      sendAddItem(id, (int)EVENT_ID_ADD_VARIABLE, client);
    }
 
@@ -1108 +1167 @@
    Remark: This function must be called AFTER all active (enabled) files have been created on the server side
    */
-   void CFile::sendEnabledFields()
+   void CFile::sendEnabledFields(CContextClient* client)
    {
      size_t size = this->enabledFields.size();
@@ -1114 +1173 @@
      {
        CField* field = this->enabledFields[i];
-       this->sendAddField(field->getId());
-       field->checkAttributes();
-       field->sendAllAttributesToServer();
-       field->sendAddAllVariables();
-     }
-   }
+       this->sendAddField(field->getId(), client);
+       field->checkTimeAttributes();
+       field->sendAllAttributesToServer(client);
+       field->sendAddAllVariables(client);
+     }
+   }
+
 
    /*!