Changeset 1871

Timestamp:

04/21/20 16:46:20 (4 years ago)

Author:

ymipsl

Message:

Some update on XIOS_COUPLING branch...

YM

Location:

XIOS/dev/dev_ym/XIOS_COUPLING/src

Files:

• io/nc4_data_output.cpp (modified) (1 diff)
• node/axis.cpp (modified) (1 diff)
• node/context.cpp (modified) (4 diffs)
• node/domain.cpp (modified) (1 diff)
• node/field.cpp (modified) (4 diffs)
• node/field.hpp (modified) (5 diffs)
• node/file.cpp (modified) (3 diffs)
• node/file.hpp (modified) (2 diffs)
• node/grid.cpp (modified) (8 diffs)
• node/grid.hpp (modified) (3 diffs)

XIOS/dev/dev_ym/XIOS_COUPLING/src/io/nc4_data_output.cpp

@@ -1640 +1640 @@
              {
                if (grid->doGridHaveDataDistributed())
-                 grid->getDistributionServer()->computeGlobalIndex(indexes);
+                 grid->getServerDistribution()->computeGlobalIndex(indexes);
 
                std::vector<StdSize> start, count;

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

@@ -1075 +1075 @@
     // Some value should be reset here
     data_begin.setValue(0);
+    data_n.setValue(data_index.numElements());
     globalLocalIndexMap_.rehash(std::ceil(index.numElements()/globalLocalIndexMap_.max_load_factor()));
 //    for (int idx = 0; idx < index.numElements(); ++idx) globalLocalIndexMap_[idx] = index(idx);

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

@@ -1017 +1017 @@
 //     postProcessing();
 
+    
     // Make sure the calendar was correctly created
+    if (serviceType_!=CServicesManager::CLIENT) CCalendarWrapper::get(CCalendarWrapper::GetDefName())->createCalendar();
     if (!calendar)
       ERROR("CContext::postProcessing()", << "A calendar must be defined for the context \"" << getId() << "!\"")
@@ -1058 +1060 @@
     vector<CField*>& fieldWithReadAccess = fieldsWithReadAccess_ ;
     vector<CField*> fieldModelIn ; // fields potentially from model
-      
+     
+    // define if files are on clientSied or serverSide 
+    if (serviceType_==CServicesManager::CLIENT)
+    {
+      for (auto& file : enabledWriteModeFiles) file->setClientSide() ;
+      for (auto& file : enabledReadModeFiles) file->setClientSide() ;
+    }
+    else
+    {
+      for (auto& file : enabledWriteModeFiles) file->setServerSide() ;
+      for (auto& file : enabledReadModeFiles) file->setServerSide() ;
+    }
+
+
+
 // find all field potentially at workflow end
     vector<CField*> endWorkflowFields ;
@@ -1088 +1104 @@
     }
 
+    // workflow endpoint => write to file
+    if (serviceType_==CServicesManager::IO_SERVER || serviceType_==CServicesManager::OUT_SERVER)
+    {
+      for(auto field : fileOutField) 
+      {
+        field->connectToFileWriter(garbageCollector) ; // connect the field to server filter
+      }
+    }
+    
+    // workflow endpoint => Send data from server to client
+    if (serviceType_==CServicesManager::IO_SERVER || serviceType_==CServicesManager::GATHERER)
+    {
+      // no filter to send data from server to client => to be implemented (reading case)
+    }
+
+    // workflow endpoint => sent to model on client side
+    if (serviceType_==CServicesManager::CLIENT)
+    {
+      for(auto field : fieldWithReadAccess) field->connectToModelOutput(garbageCollector) ;
+    }
+
 
     // workflow startpoint => data from model
@@ -1098 +1135 @@
       }
     }
-
-  
+    
+    // workflow startpoint => data from client on server side
+    if (serviceType_==CServicesManager::IO_SERVER || serviceType_==CServicesManager::GATHERER || serviceType_==CServicesManager::OUT_SERVER)
+    {
+      for(auto field : fieldModelIn) 
+      {
+        field->connectToClientInput(garbageCollector) ; // connect the field to server filter
+      }
+    }
+
+     // workflow startpoint => data from server on client side
+    if (serviceType_==CServicesManager::CLIENT)
+    {
+      for(auto field : fileInField) 
+      {
+        field->connectToServerInput(garbageCollector) ; // connect the field to server filter
+        field->computeGridIndexToFileServer() ; // compute grid index for transfer to the server context
+        field->sendFieldToFileServer() ;
+      }
+    }
+
+    // workflow startpoint => data read from file on server side
+    if (serviceType_==CServicesManager::IO_SERVER || serviceType_==CServicesManager::GATHERER)
+    {
+      // no filter for reading data from file => to be implemented
+    }
+
+
+    if (serviceType_==CServicesManager::CLIENT || serviceType_==CServicesManager::GATHERER) this->sendCloseDefinition();
+    if (serviceType_==CServicesManager::IO_SERVER || serviceType_==CServicesManager::OUT_SERVER)  createFileHeader();
+    if (serviceType_==CServicesManager::CLIENT) startPrefetchingOfEnabledReadModeFiles();
+   
+
 
 

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

@@ -1850 +1850 @@
   TRY
   {
-    if (computeConnectedClients_done_.count(client)==0) return ;
+    if (computeConnectedClients_done_.count(client)!=0) return ;
     else computeConnectedClients_done_.insert(client) ;
     

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

@@ -1192 +1192 @@
       {
         grid->makeTransformGrid() ; // make the grid transformation
-        std::pair<std::shared_ptr<CFilter>, std::shared_ptr<CFilter> > filters = CSpatialTransformFilter::buildFilterGraph(gc, gridSrc, grid, detectMissingValues, defaultValue); 
-        lastFilter->connectOutput(filters.first, 0);
-        lastFilter = filters.second;
-        gridSrc=grid ;
+        if (grid->hasTransform()) 
+        {
+          std::pair<std::shared_ptr<CFilter>, std::shared_ptr<CFilter> > filters = CSpatialTransformFilter::buildFilterGraph(gc, gridSrc, grid, detectMissingValues, defaultValue); 
+          lastFilter->connectOutput(filters.first, 0);
+          lastFilter = filters.second;
+          gridSrc=grid ;
+        }
       }
       instantDataFilter = lastFilter ;
@@ -1205 +1208 @@
     {
       if (!grid_->checkIfCompleted()) return false ;
-      grid_->solveElementsRefInheritance() ;
-      grid_->completeGrid(); // grid generation, to be checked
-      grid_->checkElementsAttributes() ;
-      instantDataFilter=inputFilter ;
-      setModelIn() ; // no reference, the field is potentially a source field from model
+      
+      if (hasFileIn()) // input file, attemp to read the grid from file
+      {
+         // must be checked
+         fileIn_->initRead() ;
+         fileIn_->checkReadFile();
+         grid_->solveElementsRefInheritance() ;
+         if (fileIn_->isClientSide()) fileIn_->readFieldAttributesMetaData(this);
+         grid_->completeGrid(); // grid generation, to be checked
+         if (fileIn_->isClientSide()) fileIn_->readFieldAttributesValues(this);
+         grid_->checkElementsAttributes() ;
+         grid_->solveDomainAxisBaseRef();
+         // probably in future tag grid incomplete if coming from a reading
+         instantDataFilter=inputFilter ;
+      }  
+      else 
+      {
+        setModelIn() ; // no reference, the field is potentially a source field from model
+
+        grid_->solveElementsRefInheritance() ;
+        grid_->completeGrid(); // grid generation, to be checked
+        grid_->checkElementsAttributes() ;
+        instantDataFilter=inputFilter ;
+      }
     }
     
@@ -1225 +1247 @@
   {
     // insert temporal filter before sending to files
-    fileWriterFilter = std::shared_ptr<CFileWriterFilter>(new CFileWriterFilter(gc, this, file->getContextClient()));
+    fileWriterFilter = std::shared_ptr<CFileWriterFilter>(new CFileWriterFilter(gc, this, client));
     // insert temporal filter before sending to files
-    getTemporalDataFilter(gc, file->output_freq)->connectOutput(fileWriterFilter, 0);
+    getTemporalDataFilter(gc, fileOut_->output_freq)->connectOutput(fileWriterFilter, 0);
   } 
 
@@ -1251 +1273 @@
   } 
  
+  /*!
+   * Connect field to a source filter to receive data from a client (on server side).
+   */
+  void CField::connectToClientInput(CGarbageCollector& gc)
+  {
+    clientSourceFilter = std::shared_ptr<CSourceFilter>(new CSourceFilter(gc,  grid_, false, false));
+    clientSourceFilter -> connectOutput(inputFilter,0) ;
+  } 
+
+
+  /*!
+   * Connect field to a source filter to receive data from a server (on client side).
+   */
+  void CField::connectToServerInput(CGarbageCollector& gc)
+  {
+    serverSourceFilter = std::shared_ptr<CSourceFilter>(new CSourceFilter(gc,  grid_, false, false));
+    serverSourceFilter -> connectOutput(inputFilter,0) ;
+  } 
+
+  /*!
+   * Connect field to a file writer filter to write data in file (on server side).
+   */
+  void CField::connectToFileWriter(CGarbageCollector& gc)
+  {
+    fileServerWriterFilter = std::shared_ptr<CFileServerWriterFilter>(new CFileServerWriterFilter(gc, this));
+    instantDataFilter->connectOutput(fileServerWriterFilter, 0);
+  } 
+  
+
+  /*!
+   * Connect field to a store filter to output data to model on client Side
+   */
+  void CField::connectToModelOutput(CGarbageCollector& gc)
+  {
+    const bool detectMissingValues = (!detect_missing_value.isEmpty() && !default_value.isEmpty() && detect_missing_value == true);
+    const double defaultValue  = detectMissingValues ? default_value : (!default_value.isEmpty() ? default_value : 0.0);
+
+    storeFilter = std::shared_ptr<CStoreFilter>(new CStoreFilter(gc, CContext::getCurrent(), grid_, detectMissingValues, defaultValue));
+    instantDataFilter->connectOutput(storeFilter, 0);
+  }
+
   /*!
    * Transform the grid_path attribut into vector of grid.

XIOS/dev/dev_ym/XIOS_COUPLING/src/node/field.hpp

@@ -227 +227 @@
         void connectToFileServer(CGarbageCollector& gc) ;
         void connectToModelInput(CGarbageCollector& gc) ;
+        void connectToFileWriter(CGarbageCollector& gc) ;
+        void connectToClientInput(CGarbageCollector& gc) ;
+        void connectToServerInput(CGarbageCollector& gc) ;
+        void connectToModelOutput(CGarbageCollector& gc);
+
         void computeGridIndexToFileServer(void) ;
 
@@ -246 +251 @@
          
          CFile* fileIn_ = nullptr ; //<! pointer to input related file
-         bool hasFileIn(void) { fileIn_==nullptr ? false : true ;}
+         bool hasFileIn(void) { return fileIn_==nullptr ? false : true ;}
          CFile* getFileIn(void) {return fileIn_;}
          void setFileIn(CFile* fileIn) { fileIn_ = fileIn ;}
@@ -252 +257 @@
 
          CFile* fileOut_ = nullptr ; //<! pointer to output related file
-         bool hasFileOut(void) { fileOut_==nullptr ? false : true ;}
+         bool hasFileOut(void) { return fileOut_==nullptr ? false : true ;}
          CFile* getFileOut(void) {return fileOut_;}
          void setFileOut(CFile* fileOut) { fileOut_ = fileOut ;}
@@ -258 +263 @@
 
          CCouplerIn* couplerIn_ = nullptr ; //<!pointer to input related coupler
-         bool hasCouplerIn(void) { couplerIn_==nullptr ? false : true ;}
+         bool hasCouplerIn(void) { return couplerIn_==nullptr ? false : true ;}
          CCouplerIn* getCouplerIn(void) {return couplerIn_;}
          void setCouplerIn(CCouplerIn* couplerIn) { couplerIn_ = couplerIn ;}
@@ -264 +269 @@
 
          CCouplerOut* couplerOut_ = nullptr ; //<!pointer to output related coupler
-         bool hasCouplerOut(void) { couplerOut_==nullptr ? false : true ;}
+         bool hasCouplerOut(void) { return couplerOut_==nullptr ? false : true ;}
          CCouplerOut* getCouplerOut(void) {return couplerOut_;}
          void setCouplerOut(CCouplerOut* couplerOut) { couplerOut_ = couplerOut ;}

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

@@ -345 +345 @@
       // TODO: This condition should be changed soon. It only works with maximum number of level as 2
 
-//ym      if (CServer::serverLevel == 0 || CServer::serverLevel == 1)
-      if (context->getServiceType()==CServicesManager::IO_SERVER || context->getServiceType()==CServicesManager::GATHERER)
-      {
+//ym      if (CServer::serverLevel == 0 || CServer::serverLevel == 1) 
+// ym client must doing it also
+//      if (context->getServiceType()==CServicesManager::IO_SERVER || context->getServiceType()==CServicesManager::GATHERER)
+//      {
         if (!mode.isEmpty() && mode.getValue() == mode_attr::read)
         {
@@ -357 +358 @@
         }
         //checkSplit(); // Really need for reading?
-      }
+ //     }
     }
     CATCH_DUMP_ATTR
@@ -756 +757 @@
    CATCH_DUMP_ATTR
 
+   void CFile::readFieldAttributesMetaData(CField* field)
+   {
+     this->data_in->readFieldAttributesMetaData(field);
+   }
+
+   void CFile::readFieldAttributesValues(CField* field)
+   {
+     this->data_in->readFieldAttributesValues(field);
+   }
    /*!
    \brief Parse xml file and write information into file object

XIOS/dev/dev_ym/XIOS_COUPLING/src/node/file.hpp

@@ -104 +104 @@
          void close(void);
          void readAttributesOfEnabledFieldsInReadMode();
-
+         void readFieldAttributesMetaData(CField* field) ;
+         void readFieldAttributesValues(CField* field) ;
          // Some processing on file
          void solveFieldRefInheritance(bool apply);
@@ -197 +198 @@
          std::vector<CField*> enabledFields;
 
+       private:  
+         bool isClientSide_ ; // the file is on client side or on server side ?
+       public:
+         bool isClientSide(void) { return isClientSide_ ;}
+         bool isServerSide(void) { return !isClientSide_ ;}
+         void setClientSide(void) { isClientSide_=true ;}
+         void setServerSide(void) { isClientSide_=false ;}
+       
+       private:
+
 
       public:

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

@@ -31 +31 @@
       , vScalarGroup_(), scalarList_(), isScalarListSet(false)
       , clientDistribution_(0), isIndexSent(false) , serverDistribution_(0), clientServerMap_(0)
-      , writtenDataSize_(0), numberWrittenIndexes_(0), totalNumberWrittenIndexes_(0), offsetWrittenIndexes_(0)
+      , numberWrittenIndexes_(0), totalNumberWrittenIndexes_(0), offsetWrittenIndexes_(0)
       , connectedDataSize_(), connectedServerRank_(), connectedServerRankRead_(), connectedDataSizeRead_()
             , isCompressible_(false)
@@ -52 +52 @@
       , vScalarGroup_(), scalarList_(), isScalarListSet(false)
       , clientDistribution_(0), isIndexSent(false) , serverDistribution_(0), clientServerMap_(0)
-      , writtenDataSize_(0), numberWrittenIndexes_(0), totalNumberWrittenIndexes_(0), offsetWrittenIndexes_(0)
+      , numberWrittenIndexes_(0), totalNumberWrittenIndexes_(0), offsetWrittenIndexes_(0)
       , connectedDataSize_(), connectedServerRank_(), connectedServerRankRead_(), connectedDataSizeRead_()
             , isCompressible_(false)
@@ -1665 +1665 @@
   TRY
   {
+    CContextClient* client = server->getAssociatedClient();
+    connectedServerRankRead_ = ranks;
+    for (int n = 0; n < ranks.size(); n++)
+    {
+      int rank = ranks[n];
+      CBufferIn& buffer = *buffers[n];
+      buffer >>  isCompressible_; // probably to be removed later
+      CArray<size_t,1> outIndex;
+      buffer >> outIndex;
+      outGlobalIndexFromClient_.insert(std::make_pair(rank, outIndex));
+      connectedDataSizeRead_[rank] = outIndex.numElements();
+    }
+
+    nbReadSenders_[client] = CClientServerMappingDistributed::computeConnectedClients(client->serverSize, client->clientSize,
+                                                                                      client->intraComm, ranks);
+  }
+  CATCH_DUMP_ATTR
+  
+
+  void CGrid::computeServerDistribution(void)
+  TRY
+  {
+    if (computeServerDistribution_done_) return ;
+    else computeServerDistribution_done_=true ;
+
+    CContext* context = CContext::getCurrent();
+      
+    int idx = 0, numElement = axis_domain_order.numElements();
+    int ssize = numElement;
+    std::vector<int> indexMap(numElement);
+    for (int i = 0; i < numElement; ++i)
+    {
+      indexMap[i] = idx;
+      if (2 == axis_domain_order(i))
+      {
+        ++ssize;
+        idx += 2;
+      }
+      else
+        ++idx;
+    }
+
+    for (int n = 0; n < connectedServerRankRead_.size(); n++)
+    {
+      int rank = connectedServerRankRead_[n];
+      size_t dataSize = 0;
+
+      if (0 == serverDistribution_)
+      {
+        int axisId = 0, domainId = 0, scalarId = 0, globalSize = 1;
+        std::vector<CDomain*> domainList = getDomains();
+        std::vector<CAxis*> axisList = getAxis();
+        std::vector<int> nBegin(ssize), nSize(ssize), nGlob(ssize), nBeginGlobal(ssize), nGlobElement(numElement);
+        std::vector<CArray<int,1> > globalIndex(numElement);
+        for (int i = 0; i < numElement; ++i)
+        {
+          nGlobElement[i] = globalSize;
+          if (2 == axis_domain_order(i)) //domain
+          {
+            nBegin[indexMap[i]] = domainList[domainId]->ibegin;
+            nSize[indexMap[i]]  = domainList[domainId]->ni;
+            nBeginGlobal[indexMap[i]] = 0;
+            nGlob[indexMap[i]] = domainList[domainId]->ni_glo;
+
+            nBegin[indexMap[i] + 1] = domainList[domainId]->jbegin;
+            nSize[indexMap[i] + 1] = domainList[domainId]->nj;
+            nBeginGlobal[indexMap[i] + 1] = 0;
+            nGlob[indexMap[i] + 1] = domainList[domainId]->nj_glo;
+
+            {
+              int count = 0;
+              globalIndex[i].resize(nSize[indexMap[i]]*nSize[indexMap[i]+1]);
+              for (int jdx = 0; jdx < nSize[indexMap[i]+1]; ++jdx)
+                for (int idx = 0; idx < nSize[indexMap[i]]; ++idx)
+                {
+                  globalIndex[i](count) = (nBegin[indexMap[i]] + idx) + (nBegin[indexMap[i]+1] + jdx) * nGlob[indexMap[i]];
+                  ++count;
+                }
+            }
+
+            ++domainId;
+          }
+          else if (1 == axis_domain_order(i)) // axis
+          {
+            nBegin[indexMap[i]] = axisList[axisId]->begin;
+            nSize[indexMap[i]]  = axisList[axisId]->n;
+            nBeginGlobal[indexMap[i]] = 0;
+            nGlob[indexMap[i]] = axisList[axisId]->n_glo;     
+            globalIndex[i].resize(nSize[indexMap[i]]);
+            for (int idx = 0; idx < nSize[indexMap[i]]; ++idx)
+              globalIndex[i](idx) = nBegin[indexMap[i]] + idx;
+
+            ++axisId;
+          }
+          else // scalar
+          { 
+            nBegin[indexMap[i]] = 0;
+            nSize[indexMap[i]]  = 1;
+            nBeginGlobal[indexMap[i]] = 0;
+            nGlob[indexMap[i]] = 1;
+            globalIndex[i].resize(1);
+            globalIndex[i](0) = 0;
+            ++scalarId;
+          }
+        }
+        dataSize = 1;
+
+        for (int i = 0; i < nSize.size(); ++i)
+        dataSize *= nSize[i];
+        serverDistribution_ = new CDistributionServer(context->intraCommRank_, 
+                                                      globalIndex, axis_domain_order,
+                                                      nBegin, nSize, nBeginGlobal, nGlob);
+      }
+    }
+  }
+  CATCH_DUMP_ATTR
+
+
+
+
+
+
+/* old interface => transform into compute receivedIndex
+  void CGrid::recvIndex(vector<int> ranks, vector<CBufferIn*> buffers, CContextServer* server)
+  TRY
+  {
     CContext* context = CContext::getCurrent();
     connectedServerRankRead_ = ranks;
@@ -1829 +1955 @@
   }
   CATCH_DUMP_ATTR
+*/
+
 
   /*
@@ -1922 +2050 @@
   TRY
   {
-     return (0 != writtenDataSize_);
+     return (0 != getWrittenDataSize());
   }
   CATCH_DUMP_ATTR
@@ -1932 +2060 @@
     \return size of data written on server
   */
-  size_t CGrid::getWrittenDataSize() const
-  TRY
-  {
-    return writtenDataSize_;
+  size_t CGrid::getWrittenDataSize() 
+  TRY
+  {
+    return getServerDistribution()->getGridSize();
   }
   CATCH
@@ -1972 +2100 @@
   CATCH
 
-  CDistributionServer* CGrid::getDistributionServer()
-  TRY
-  {
-    return serverDistribution_;
-  }
-  CATCH_DUMP_ATTR
-
+  
   CDistributionClient* CGrid::getClientDistribution()
   TRY
@@ -2113 +2235 @@
     gridPtr->sendCreateChild(this->getId(),client);
     this->sendAllAttributesToServer(client);
+    if (isScalarGrid())  sendIndexScalarGrid();
+    else  sendIndex();
     this->sendAllDomains(client);
     this->sendAllAxis(client);

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

@@ -194 +194 @@
          bool doGridHaveDataToWrite();
          bool doGridHaveDataDistributed(CContextClient* client = 0);
-         size_t getWrittenDataSize() const;
+         size_t getWrittenDataSize() ;
          int getNumberWrittenIndexes() const;
          int getTotalNumberWrittenIndexes() const;
          int getOffsetWrittenIndexes() const;
 
-         CDistributionServer* getDistributionServer();
          CGridTransformation* getTransformations();
 
@@ -248 +247 @@
        CDistributionClient* getClientDistribution(void); 
 
+     private:   
+       /** Server-like distribution calculated upon receiving indexes */
+       CDistributionServer* serverDistribution_;
+       void computeServerDistribution(void) ;
+       bool computeServerDistribution_done_=false ;
+     public: 
+       CDistributionServer* getServerDistribution(void) { if (computeServerDistribution_done_) computeServerDistribution() ; return serverDistribution_ ;}
+
+
      private:
        template<int N>
@@ -413 +421 @@
         bool isAxisListSet, isDomListSet, isScalarListSet;
 
-/** Server-like distribution calculated upon receiving indexes */
-        CDistributionServer* serverDistribution_;
-
         CClientServerMapping* clientServerMap_;
-        size_t writtenDataSize_;
         int numberWrittenIndexes_, totalNumberWrittenIndexes_, offsetWrittenIndexes_;