Changeset 640

Timestamp:

07/17/15 13:58:12 (9 years ago)

Author:

rlacroix

Message:

Start using the filter infrastructure to read and write fields.

Note that currently only the simple cases are working. References and all types of operations are disabled.

Location:

XIOS/trunk/src/node

Files:

• context.cpp (modified) (6 diffs)
• context.hpp (modified) (2 diffs)
• field.cpp (modified) (7 diffs)
• field.hpp (modified) (3 diffs)
• field_impl.hpp (modified) (5 diffs)
• file.cpp (modified) (1 diff)
• file.hpp (modified) (2 diffs)

XIOS/trunk/src/node/context.cpp

@@ -46 +46 @@
 
    //----------------------------------------------------------------
+
    /*!
    \brief Get context group (context root)
@@ -56 +57 @@
    }
 
-
    //----------------------------------------------------------------
+
    /*!
    \brief Get calendar of a context
@@ -68 +69 @@
 
    //----------------------------------------------------------------
+
    /*!
    \brief Set a context with a calendar
@@ -386 +388 @@
     if (hasClient)
     {
+      //this->buildAllExpressionOfEnabledFields();
+      //buildAllExpressionOfFieldsWithReadAccess();
+      this->buildFilterGraphOfEnabledFields();
+      buildFilterGraphOfFieldsWithReadAccess();
       this->solveAllRefOfEnabledFields(true);
-      this->buildAllExpressionOfEnabledFields();
-      buildAllExpressionOfFieldsWithReadAccess();
     }
 
@@ -426 +430 @@
      {
        this->enabledFiles[i]->buildAllExpressionOfEnabledFields();
+     }
+   }
+
+   void CContext::buildFilterGraphOfEnabledFields()
+   {
+     int size = this->enabledFiles.size();
+     for (int i = 0; i < size; ++i)
+     {
+       this->enabledFiles[i]->buildFilterGraphOfEnabledFields(garbageCollector);
      }
    }
@@ -474 +487 @@
     for (size_t i = 0; i < fieldsWithReadAccess.size(); ++i)
       fieldsWithReadAccess[i]->buildAllExpressionEnabledField();
+  }
+
+  void CContext::buildFilterGraphOfFieldsWithReadAccess()
+  {
+    for (size_t i = 0; i < fieldsWithReadAccess.size(); ++i)
+      fieldsWithReadAccess[i]->buildFilterGraph(garbageCollector, true);
   }
 

XIOS/trunk/src/node/context.hpp

@@ -112 +112 @@
          void solveAllRefOfEnabledFields(bool sendToServer);
          void buildAllExpressionOfEnabledFields();
+         void buildFilterGraphOfEnabledFields();
          void startPrefetchingOfEnabledReadModeFiles();
          void checkPrefetchingOfEnabledReadModeFiles();
@@ -117 +118 @@
          void solveAllRefOfFieldsWithReadAccess();
          void buildAllExpressionOfFieldsWithReadAccess();
+         void buildFilterGraphOfFieldsWithReadAccess();
          void postProcessing();
 

XIOS/trunk/src/node/field.cpp

@@ -14 +14 @@
 #include "context_server.hpp"
 #include <set>
+#include "garbage_collector.hpp"
+#include "source_filter.hpp"
+#include "store_filter.hpp"
+#include "file_writer_filter.hpp"
 
 namespace xios{
@@ -31 +35 @@
       , isReadDataRequestPending(false)
       , filterSources_()
-      { setVirtualVariableGroup(); }
+   { setVirtualVariableGroup(); }
 
    CField::CField(const StdString& id)
@@ -456 +460 @@
     CContext* context = CContext::getCurrent();
     StdSize record;
+    std::map<int, CArray<double,1> > data;
+
     for (int i = 0; i < ranks.size(); i++)
+    {
+      int rank = ranks[i];
+      *buffers[i] >> record;
+      isEOF = (record == size_t(-1));
+
+      if (!isEOF)
+        *buffers[i] >> data[rank];
+      else
+        break;
+    }
+
+    if (isEOF)
+      serverSourceFilter->signalEndOfStream(lastDataRequestedFromServer);
+    else
+      serverSourceFilter->streamDataFromServer(lastDataRequestedFromServer, data);
+
+    /*for (int i = 0; i < ranks.size(); i++)
     {
       int rank = ranks[i];
@@ -487 +510 @@
         if (hasFieldOut) updateDataFromExpression(instantData);
       }
-    }
+    }*/
 
     isReadDataRequestPending = false;
@@ -633 +656 @@
       using namespace func;
 
-      if (!hasOutputFile && !hasFieldOut) return;
+      CContext* context = CContext::getCurrent();
+
+      if (!context->hasServer || !hasOutputFile) return;
 
       StdString id;
@@ -639 +664 @@
       else if (!name.isEmpty()) id = name;
       else if (hasDirectFieldReference()) id = baseRefObject->getId();
-
-      CContext* context = CContext::getCurrent();
 
       if (freq_op.isEmpty()) freq_op.setValue(TimeStep);
@@ -722 +745 @@
                << "The operation is not defined !");
 //      }
+   }
+
+   //----------------------------------------------------------------
+
+   /*!
+    * Constructs the graph filter for the field, enabling or not the data output.
+    * This method should not be called more than once with enableOutput equal to true.
+    *
+    * \param gc the garbage collector to use when building the filter graph
+    * \param enableOutput must be true when the field data is to be
+    *                     read by the client or/and written to a file
+    */
+   void CField::buildFilterGraph(CGarbageCollector& gc, bool enableOutput)
+   {
+     // Start by building a filter which can provide the field's instant data
+     if (!instantDataFilter)
+     {
+       // Check if the data is to be read from a file
+       if (file && !file->mode.isEmpty() && file->mode == CFile::mode_attr::read)
+         instantDataFilter = serverSourceFilter = boost::shared_ptr<CSourceFilter>(new CSourceFilter(grid));
+       else // The data might be passed from the model
+         instantDataFilter = clientSourceFilter = boost::shared_ptr<CSourceFilter>(new CSourceFilter(grid));
+     }
+
+     // If the field data is to be read by the client or/and written to a file
+     if (enableOutput && !storeFilter && !fileWriterFilter)
+     {
+       if (!read_access.isEmpty() && read_access.getValue())
+       {
+         storeFilter = boost::shared_ptr<CStoreFilter>(new CStoreFilter(gc, CContext::getCurrent(), grid));
+         instantDataFilter->connectOutput(storeFilter, 0);
+       }
+
+       if (file && (file->mode.isEmpty() || file->mode == CFile::mode_attr::write))
+       {
+         fileWriterFilter = boost::shared_ptr<CFileWriterFilter>(new CFileWriterFilter(gc, this));
+         instantDataFilter->connectOutput(fileWriterFilter, 0);
+       }
+     }
    }
 

XIOS/trunk/src/node/field.hpp

@@ -28 +28 @@
    class CFile;
    class CGrid;
-   class CContext ;
+   class CContext;
    class CGenericFilter;
+
+   class CGarbageCollector;
+   class COutputPin;
+   class CSourceFilter;
+   class CStoreFilter;
+   class CFileWriterFilter;
 
    ///--------------------------------------------------------------
@@ -124 +130 @@
          void solveTransformedGrid();
          CGrid* getGridRefOfBaseReference();
+
+         void buildFilterGraph(CGarbageCollector& gc, bool enableOutput);
 
 //         virtual void fromBinary(StdIStream & is);
@@ -233 +241 @@
          DECLARE_REF_FUNC(Field,field)
 
+      private:
+         //! The output pin of the filter providing the instant data for the field
+         boost::shared_ptr<COutputPin> instantDataFilter;
+         //! The source filter for data provided by the client
+         boost::shared_ptr<CSourceFilter> clientSourceFilter;
+         //! The source filter for data provided by the server
+         boost::shared_ptr<CSourceFilter> serverSourceFilter;
+         //! The terminal filter which stores the instant data
+         boost::shared_ptr<CStoreFilter> storeFilter;
+         //! The terminal filter which writes the data to file
+         boost::shared_ptr<CFileWriterFilter> fileWriterFilter;
    }; // class CField
 

XIOS/trunk/src/node/field_impl.hpp

@@ -9 +9 @@
 #include "timer.hpp"
 #include "array_new.hpp"
+#include "source_filter.hpp"
+#include "store_filter.hpp"
 
 
@@ -16 +18 @@
   void CField::setData(const CArray<double, N>& _data)
   {
-    if (hasInstantData)
+    if (clientSourceFilter)
+      clientSourceFilter->streamData(CContext::getCurrent()->getCalendar()->getCurrentDate(), _data);
+    else if (!field_ref.isEmpty() || !content.empty())
+      ERROR("void CField::setData(const CArray<double, N>& _data)",
+            << "Impossible to receive data from the model for a field [ id = " << getId() << " ] with a reference or an arithmetic operation.");
+
+    /*if (hasInstantData)
     {
       grid->inputField(_data, instantData);
@@ -52 +60 @@
       }
       if (hasOutputFile || hasFieldOut) updateData(_data);
-    }
+    }*/
   }
 
@@ -279 +287 @@
   void CField::getData(CArray<double, N>& _data) const
   {
-    if (!read_access.isEmpty() && read_access.getValue() && hasInstantData)
-    {
-      CContext* context = CContext::getCurrent();
+    if (storeFilter)
+    {
+      CDataPacket::StatusCode status = storeFilter->getData(CContext::getCurrent()->getCalendar()->getCurrentDate(), _data);
+
+      if (status == CDataPacket::END_OF_STREAM)
+        ERROR("void CField::getData(CArray<double, N>& _data) const",
+              << "Impossible to access field data, all the records of the field [ id = " << getId() << " ] have been already read.");
+
+      /*CContext* context = CContext::getCurrent();
       const CDate& currentDate = context->getCalendar()->getCurrentDate();
 
@@ -291 +305 @@
               << "Impossible to access field data, all the records of the field [ id = " << getId() << " ] have been already read.");
 
-      grid->outputField(instantData, _data);
+      grid->outputField(instantData, _data);*/
     }
     else

XIOS/trunk/src/node/file.cpp

@@ -622 +622 @@
 
    /*!
+    * Constructs the filter graph for each active field.
+    *
+    * \param gc the garbage collector to use when building the filter graph
+    */
+   void CFile::buildFilterGraphOfEnabledFields(CGarbageCollector& gc)
+   {
+     int size = this->enabledFields.size();
+     for (int i = 0; i < size; ++i)
+     {
+       this->enabledFields[i]->buildFilterGraph(gc, true);
+     }
+   }
+
+   /*!
      Prefetching the data for enabled fields read from file.
    */

XIOS/trunk/src/node/file.hpp

@@ -22 +22 @@
    class CFileAttributes;
    class CFile;
+
+   class CGarbageCollector;
 
    ///--------------------------------------------------------------
@@ -108 +110 @@
          void solveAllRefOfEnabledFields(bool sendToServer);
          void buildAllExpressionOfEnabledFields();
+         void buildFilterGraphOfEnabledFields(CGarbageCollector& gc);
          void prefetchEnabledReadModeFields();
          void prefetchEnabledReadModeFieldsIfNeeded();