Changeset 1961

Timestamp:

10/12/20 11:26:40 (4 years ago)

Author:

ymipsl

Message:

Rewrite file writing filter.
YM

Location:

XIOS/dev/dev_ym/XIOS_COUPLING/src

Files:

• data_output.cpp (modified) (1 diff)
• data_output.hpp (modified) (2 diffs)
• filter/file_writer_store_filter.cpp (modified) (2 diffs)
• filter/file_writer_store_filter.hpp (modified) (2 diffs)
• io/nc4_data_output.cpp (modified) (14 diffs)
• io/nc4_data_output.hpp (modified) (1 diff)
• node/field.cpp (modified) (3 diffs)
• node/field.hpp (modified) (2 diffs)
• node/file.cpp (modified) (3 diffs)
• node/file.hpp (modified) (1 diff)

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

@@ -122 +122 @@
       //----------------------------------------------------------------
 
-      void CDataOutput::writeFieldData(CField* field, const CArray<double,1>& data)
+      int CDataOutput::writeFieldData(CField* field, const CArray<double,1>& data, const CDate& lastWrite, const CDate& currentWrite, int nstep)
       TRY
       {
-//         CGrid* grid = CGrid::get(field->grid_ref.getValue());
-//         CDomain* domain = CDomain::get(grid->domain_ref.getValue());
-         this->writeFieldData_(field, data);
+        return this->writeFieldData_(field, data, lastWrite, currentWrite, nstep);
       }
       CATCH
+
 
       ///----------------------------------------------------------------

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

@@ -28 +28 @@
             void writeFieldGrid(CField* field);
             void writeTimeDimension(void);
-            void writeFieldData(CField* field, const CArray<double,1>& data);
+            int  writeFieldData(CField* field, const CArray<double,1>& data, const CDate& lastWrite, const CDate& currentWrite, int nstep);
 
             virtual void definition_start(void) = 0;
@@ -52 +52 @@
             virtual void syncFile_      (void)               = 0;
             virtual void writeField_    (CField*     field)  = 0;
-            virtual void writeFieldData_(CField*     field, const CArray<double,1>& data)  = 0;
+            virtual int writeFieldData_ (CField*     field, const CArray<double,1>& data, const CDate& lastWrite, const CDate& currentWrite, int nstep)  = 0;
             virtual void writeDomain_   (CDomain*    domain) = 0;
             virtual void writeAxis_     (CAxis*      axis)   = 0;

XIOS/dev/dev_ym/XIOS_COUPLING/src/filter/file_writer_store_filter.cpp

@@ -2 +2 @@
 #include "exception.hpp"
 #include "field.hpp"
+#include "file.hpp"
+#include "context.hpp"
 
 namespace xios
@@ -7 +9 @@
   CFileWriterStoreFilter::CFileWriterStoreFilter(CGarbageCollector& gc, CField* field)
     : CInputPin(gc, 1)
-    , field(field)
+    , field_(field)
+
   {
-    if (!field)
-      ERROR("CFileWriterStoreFilter::CFileWriterStoreFilter(CField* field)",
-            "The field cannot be null.");
+    CContext* context = CContext::getCurrent();
+
+    if (!field) ERROR("CFileWriterStoreFilter::CFileWriterStoreFilter(CField* field)", "The field cannot be null.");
+    file_ = field->getFileOut() ;
+    grid_= field->getGrid() ;
+    freqWrite_ = file_->output_freq ;
+    lastWrite_ = context->getCalendar()->getInitDate();
+    if (!file_->isEmptyZone() && (field->getGrid()->doGridHaveDataToWrite() || file_->type == CFile::type_attr::one_file))
+         needToWrite_=true ;
+    else needToWrite_=false;
+    lastFileSplit_ = file_->getLastSplit() ;
+    nstep_ = 0 ;
+    if (!field->scale_factor.isEmpty()) { scaleFactor_ = field->scale_factor ; hasScaleFactor_ = true ; }
+    if (!field->add_offset.isEmpty()) { addOffset_ = field->add_offset ; hasAddOffset_ = true ; }
+    if (!field->prec.isEmpty() && field->prec == 2) hasRounding_ = true ;
+    if (!field->default_value.isEmpty()) {hasDefaultValue_=true ; defaultValue_ = field->default_value ;}
   }
 
   void CFileWriterStoreFilter::onInputReady(std::vector<CDataPacketPtr> data)
   {
-    field->writeUpdateData(data[0]->data);
+    const CDate currentWrite = lastWrite_ + freqWrite_;
+    if (needToWrite_)
+    {
+      file_->checkWriteFile();
+      if (file_->getLastSplit() != lastFileSplit_)
+      {
+        nstep_ = 0 ;
+        lastFileSplit_ = file_->getLastSplit() ;
+      }
+      nstep_ = nstep_+1;
+      
+      CArray<double,1> dataIn = data[0]->data ;
+      CArray<double,1> fieldData ;
+
+      if (hasAddOffset_ || hasScaleFactor_ || hasRounding_) dataIn = data[0]->data ;
+      else dataIn.reference(data[0]->data) ;
+
+      if (hasAddOffset_)   dataIn = dataIn - addOffset_ ;
+      if (hasScaleFactor_) dataIn = dataIn / scaleFactor_;
+      if (hasRounding_)    dataIn = round(dataIn);
+
+      if (hasDefaultValue_)
+      {
+        size_t nbData=dataIn.numElements() ;
+        for (size_t idx = 0; idx < nbData; ++idx) if ( NumTraits<double>::isNan(dataIn(idx)) ) dataIn(idx)=defaultValue_ ;
+      }
+
+      if (field_->getUseCompressedOutput()) fieldData.reference(dataIn) ;
+      else
+      {
+        fieldData.resize(grid_->getWorkflowToFullConnector()->getDstSize());
+        if (hasDefaultValue_) grid_->getWorkflowToFullConnector()->transfer(dataIn, fieldData, defaultValue_ ) ;
+        else grid_->getWorkflowToFullConnector()->transfer(dataIn, fieldData ) ;
+      }  
+      nstep_ = file_->getDataOutput()->writeFieldData(field_, fieldData, lastWrite_,currentWrite, nstep_);
+    }
+
+    lastWrite_ = currentWrite ;
+
   }
 

XIOS/dev/dev_ym/XIOS_COUPLING/src/filter/file_writer_store_filter.hpp

@@ -7 +7 @@
 {
   class CField;
+  class CFile;
+  class CGrid;
 
   /*!
@@ -46 +48 @@
 
     private:
-      CField* field; //<! The associated field
-      std::map<Time, CDataPacketPtr> packets; //<! The stored packets
+      CField* field_; //<! The associated field
+      CFile* file_ ;
+      CGrid* grid_ ;
+      CDate lastWrite_ ;
+      CDate lastFileSplit_ ;
+      CDuration freqWrite_ ;
+      int nstep_ ;
+      bool needToWrite_ ;
+      double scaleFactor_ = 1.0;
+      double addOffset_ = 0.0;
+      bool hasScaleFactor_ = false ;
+      bool hasAddOffset_ = false ;
+      double defaultValue_ ;
+      bool hasDefaultValue_=false;
+      bool hasRounding_=false ;
+
   }; // class CFileWriterStoreFilter
 } // namespace xios

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

@@ -1713 +1713 @@
         }
 
-        if (compressedOutput && grid->isCompressible())
+        while (!dimIdList.empty())
         {
-          dims.push_back(grid->getId() + "_points");
-          field->setUseCompressedOutput();
-        }
-        else
-        {
-          while (!dimIdList.empty())
-          {
-            dims.push_back(dimIdList.back());
-            dimIdList.pop_back();
-          }
+          dims.push_back(dimIdList.back());
+          dimIdList.pop_back();
         }
 
@@ -2070 +2062 @@
       //---------------------------------------------------------------
 
-      void CNc4DataOutput::writeFieldData_ (CField*  field, const CArray<double,1>& data)
+      int CNc4DataOutput::writeFieldData_ (CField*  field, const CArray<double,1>& data, const CDate& lastWrite,
+                                           const CDate& currentWrite, int nstep)
       {
         CContext* context = CContext::getCurrent();
         CGrid* grid = field->getGrid();
-        CArray<double,1> dataIn(data.copy()) ;
-
-        if (field->getNStep()<1) 
+       
+        if (nstep<1) 
         {
-          return;
+          return nstep;
         }
         
@@ -2084 +2076 @@
           if (SuperClass::type == MULTI_FILE || !isCollective)
           {
-            return;
+            return nstep;
           }
 
@@ -2109 +2101 @@
             factorUnit=context->getCalendar()->getDayLengthInSeconds() ;
             else factorUnit=1 ;
-            field->resetNStep(getRecordFromTime(field->last_Write_srv,factorUnit) + 1);
+            nstep = getRecordFromTime(field->last_Write_srv,factorUnit) + 1;
           }
 
@@ -2128 +2120 @@
         if (wtime)
         {
-          Time lastWrite = field->last_Write_srv;
-          Time lastLastWrite = field->lastlast_Write_srv;
-
           
           if (field->hasTimeInstant)
           {
-            time_data(0) = time_data_bound(1) = lastWrite;
-            time_data_bound(0) = time_data_bound(1) = lastWrite;
+            time_data(0) = time_data_bound(1) = (Time) lastWrite;
+            time_data_bound(0) = time_data_bound(1) = (Time) currentWrite;
             if (timeCounterType==instant)
             {
@@ -2147 +2136 @@
           else if (field->hasTimeCentered)
           {
-            time_data(0) = (lastWrite + lastLastWrite) / 2;
-            time_data_bound(0) = lastLastWrite;
-            time_data_bound(1) = lastWrite;
+            time_data(0) = ((Time)currentWrite + (Time)lastWrite) / 2;
+            time_data_bound(0) = (Time)lastWrite;
+            time_data_bound(1) = (Time)currentWrite;
             if (timeCounterType==centered)
             {
@@ -2162 +2151 @@
           if (timeCounterType==record)
           {
-            time_counter(0) = field->getNStep() - 1;
-            time_counter_bound(0) = time_counter_bound(1) = field->getNStep() - 1;
+            time_counter(0) = nstep - 1;
+            time_counter_bound(0) = time_counter_bound(1) = nstep - 1;
             wtimeCounter=true ;
           }
@@ -2179 +2168 @@
          bool isRoot = (context->intraCommRank_ == 0);
 
-         if (!field->scale_factor.isEmpty() || !field->add_offset.isEmpty())
-         {
-           double scaleFactor = 1.0;
-           double addOffset = 0.0;
-           if (!field->scale_factor.isEmpty()) scaleFactor = field->scale_factor;
-           if (!field->add_offset.isEmpty()) addOffset = field->add_offset;
-           field->scaleFactorAddOffset(dataIn, scaleFactor, addOffset);
-         }
-
          try
          {
-           size_t writtenSize;
-           if (field->getUseCompressedOutput())
-             writtenSize = grid->getNumberWrittenIndexes();
-           else
-             writtenSize = grid->getWrittenDataSize();
-
-           CArray<double,1> fieldData(writtenSize);
-           if (!field->default_value.isEmpty()) fieldData = field->default_value;
-
-           if (field->getUseCompressedOutput()) fieldData.reference(dataIn) ;
-             // field->outputCompressedField(dataIn, fieldData);
-           else
-           {
-             //field->outputField(dataIn, fieldData);
-             if (!field->default_value.isEmpty()) field->getGrid()->getWorkflowToFullConnector()->transfer(dataIn, fieldData, (double)field->default_value ) ;
-             else field->getGrid()->getWorkflowToFullConnector()->transfer(dataIn, fieldData ) ;
-           }
-
-           if (!field->prec.isEmpty() && field->prec == 2) fieldData = round(fieldData);
-
            switch (SuperClass::type)
            {
@@ -2215 +2175 @@
               {
                  CTimer::get("Files : writing data").resume();
-                 SuperClassWriter::writeData(fieldData, fieldid, isCollective, field->getNStep() - 1);
+                 SuperClassWriter::writeData(data, fieldid, isCollective, nstep - 1);
                  CTimer::get("Files : writing data").suspend();
                  if (wtime)
@@ -2222 +2182 @@
                    if ( wtimeData)
                    {
-                       SuperClassWriter::writeTimeAxisData(time_data, timeAxisId, isCollective, field->getNStep() - 1, isRoot);
-                       SuperClassWriter::writeTimeAxisDataBounds(time_data_bound, timeAxisBoundId, isCollective, field->getNStep() - 1, isRoot);
+                       SuperClassWriter::writeTimeAxisData(time_data, timeAxisId, isCollective, nstep - 1, isRoot);
+                       SuperClassWriter::writeTimeAxisDataBounds(time_data_bound, timeAxisBoundId, isCollective, nstep - 1, isRoot);
                   }
                    if (wtimeCounter)
                    {
-                     SuperClassWriter::writeTimeAxisData(time_counter, getTimeCounterName(), isCollective, field->getNStep() - 1,isRoot);
-                     if (timeCounterType!=record) SuperClassWriter::writeTimeAxisDataBounds(time_counter_bound, timeBoundId, isCollective, field->getNStep() - 1, isRoot);
+                     SuperClassWriter::writeTimeAxisData(time_counter, getTimeCounterName(), isCollective, nstep - 1,isRoot);
+                     if (timeCounterType!=record) SuperClassWriter::writeTimeAxisDataBounds(time_counter_bound, timeBoundId, isCollective, nstep - 1, isRoot);
                    }
                    CTimer::get("Files : writing time axis").suspend();
@@ -2316 +2276 @@
                     }
                   }
-              }
+                }
                 else
                 {
@@ -2368 +2328 @@
 
                 CTimer::get("Files : writing data").resume();
-                SuperClassWriter::writeData(fieldData, fieldid, isCollective, field->getNStep() - 1, &start, &count);
+                SuperClassWriter::writeData(data, fieldid, isCollective, nstep - 1, &start, &count);
                 CTimer::get("Files : writing data").suspend();
 
@@ -2376 +2336 @@
                    if ( wtimeData)
                    {
-                     SuperClassWriter::writeTimeAxisData(time_data, timeAxisId, isCollective, field->getNStep() - 1, isRoot);
-                     SuperClassWriter::writeTimeAxisDataBounds(time_data_bound, timeAxisBoundId, isCollective, field->getNStep() - 1, isRoot);
+                     SuperClassWriter::writeTimeAxisData(time_data, timeAxisId, isCollective, nstep - 1, isRoot);
+                     SuperClassWriter::writeTimeAxisDataBounds(time_data_bound, timeAxisBoundId, isCollective, nstep - 1, isRoot);
                    }
                    if (wtimeCounter)
                    {
-                     SuperClassWriter::writeTimeAxisData(time_counter, getTimeCounterName(), isCollective, field->getNStep() - 1,isRoot);
-                     if (timeCounterType!=record) SuperClassWriter::writeTimeAxisDataBounds(time_counter_bound, timeBoundId, isCollective, field->getNStep() - 1, isRoot);
+                     SuperClassWriter::writeTimeAxisData(time_counter, getTimeCounterName(), isCollective, nstep - 1,isRoot);
+                     if (timeCounterType!=record) SuperClassWriter::writeTimeAxisDataBounds(time_counter_bound, timeBoundId, isCollective, nstep - 1, isRoot);
 
                    }
@@ -2391 +2351 @@
               }
             }
+            return nstep ;
          }
          catch (CNetCdfException& e)

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

@@ -56 +56 @@
             virtual void writeAttribute_(CVariable* var);
             virtual void writeAttribute_(CVariable* var, const string& fieldId);
-            virtual void writeFieldData_(CField* field, const CArray<double,1>& data);
+            virtual int  writeFieldData_(CField* field, const CArray<double,1>& data, const CDate& lastWrite, const CDate& currentWrite, int nstep);
             virtual void writeFile_     (CFile* file);
             virtual void closeFile_     (void);

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

@@ -163 +163 @@
   CATCH
 
-  
+/*  
   void CField::writeUpdateData(const CArray<double,1>& data)
   TRY
@@ -169 +169 @@
     const CDate writeDate = last_Write_srv + freq_write_srv;
     last_Write_srv = writeDate;
-    // grid_->computeWrittenIndex(); -> obselete function need to be removed
-    /*
-    recvDataSrv.resize(data.numElements()) ;
-    recvDataSrv = data ;
-    */
     writeField(data);
     lastlast_Write_srv = last_Write_srv;
   }
   CATCH_DUMP_ATTR
-
-
+*/
+
+/*
   void CField::writeField(const CArray<double,1>& data)
   TRY
@@ -194 +190 @@
   }
   CATCH_DUMP_ATTR
-
+*/
+  
   /*
     Send a request for reading data.

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

@@ -203 +203 @@
         void recvUpdateDataFromCoupler(std::map<int,CBufferIn*>& rankBuffers); // old interface to be removed
         
-        void writeField(const CArray<double,1>& data);
+//        void writeField(const CArray<double,1>& data);
         bool sendReadDataRequest(const CDate& tsDataRequested);
         bool sendReadDataRequestIfNeeded(void);
@@ -246 +246 @@
         void recvAddVariableGroup(CBufferIn& buffer);        
         void sendAddAllVariables(CContextClient* client);
-        void writeUpdateData(const CArray<double,1>& data);
+        //void writeUpdateData(const CArray<double,1>& data);
 
         const std::vector<StdString>& getRefDomainAxisIds();

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

@@ -318 +318 @@
     {
       CContext* context = CContext::getCurrent();
-      // Done by classical server or secondary server
-      // This condition should be changed soon
-//ym      if (CServer::serverLevel == 0 || CServer::serverLevel == 2)
-       if (context->getServiceType()==CServicesManager::IO_SERVER || context->getServiceType()==CServicesManager::OUT_SERVER)
-       {
-        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?
-      }
+
+      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(); 
     }
     CATCH_DUMP_ATTR
@@ -417 +412 @@
           lastSplit = lastSplit + split_freq.getValue();
           std::vector<CField*>::iterator it, end = this->enabledFields.end();
-          for (it = this->enabledFields.begin(); it != end; it++)
+/*          for (it = this->enabledFields.begin(); it != end; it++)
           {
             (*it)->resetNStep();
             (*it)->resetNStepMax();
-          }
+          }*/
           if (mode.isEmpty() || mode.getValue() == mode_attr::write)
             createHeader();
@@ -432 +427 @@
     }
     CATCH_DUMP_ATTR
+
 
    /*!

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

@@ -179 +179 @@
          CDate lastSync;
          CDate lastSplit;
+         
+         const CDate& getLastSplit(void) { return lastSplit ; }
+        
          int nbAxis, nbDomains;
          bool isOpen;