Changeset 1883

Timestamp:

05/19/20 16:44:47 (4 years ago)

Author:

ymipsl

Message:

XIOS coupling branch
Adopt infrastructure based on filter for reading data on server side and sending it to the client, in a similar way on what is done for other case.

YM

Location:

XIOS/dev/dev_ym/XIOS_COUPLING/src

Files:

• filter/file_server_reader_filter.cpp (added)
• filter/file_server_reader_filter.hpp (added)
• filter/server_to_client_filter.cpp (added)
• filter/server_to_client_filter.hpp (added)
• node/context.cpp (modified) (3 diffs)
• node/field.cpp (modified) (6 diffs)
• node/field.hpp (modified) (4 diffs)

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

@@ -1166 +1166 @@
     else if (serviceType_==CServicesManager::CLIENT) for(auto file : this->enabledWriteModeFiles) file->setContextClient(client) ;
 
+    // client side, assign context for file reading
     if (serviceType_==CServicesManager::CLIENT) for(auto file : this->enabledReadModeFiles) file->setContextClient(client) ;
+    
+    // server side, assign context where to send file data read
+    if (serviceType_==CServicesManager::CServicesManager::GATHERER || serviceType_==CServicesManager::IO_SERVER) 
+      for(auto file : this->enabledReadModeFiles) file->setContextClient(client) ;
    
     // workflow endpoint => sent to IO/SERVER
@@ -1192 +1197 @@
     if (serviceType_==CServicesManager::IO_SERVER || serviceType_==CServicesManager::GATHERER)
     {
-      // no filter to send data from server to client => to be implemented (reading case)
+      for(auto field : fileInField) 
+      {
+        field->connectToServerToClient(garbageCollector) ;
+      }
     }
 
@@ -1247 +1255 @@
     if (serviceType_==CServicesManager::IO_SERVER || serviceType_==CServicesManager::GATHERER)
     {
-      // no filter for reading data from file => to be implemented
+      for(auto field : fileInField) 
+      {
+        field->connectToFileReader(garbageCollector) ;
+      }
     }
     

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

@@ -24 +24 @@
 #include "spatial_transform_filter.hpp"
 #include "file_server_writer_filter.hpp"
+#include "file_server_reader_filter.hpp"
+#include "server_to_client_filter.hpp"
 #include "tracer.hpp"
 
@@ -435 +437 @@
   TRY
   {
-    CContextClient* client = server->getAssociatedClient() ;
+        
+    CArray<double,1> data ;
+    EReadField hasData = readField(data);
+    CDate date = CContext::getCurrent()->getCalendar()->getCurrentDate();
+    if (hasData == RF_DATA) fileServerReaderFilter_->streamData(date,data) ;
+    else fileServerReaderFilter_->signalEndOfStream(date) ;
+
+  }
+  CATCH_DUMP_ATTR
+
+
+  void CField::sendUpdateDataServerToClient(bool isEOF, const CArray<double,1>& data, CContextClient* client)
+  TRY
+  {
     CEventClient event(getType(), EVENT_ID_READ_DATA_READY);
     std::list<CMessage> msgs;
-    
-    CArray<double,1> data ;
-    EReadField hasData = readField(data);
-
-    map<int, CArray<double,1> >::iterator it;
     if (!grid_->doGridHaveDataDistributed(client))
     {
@@ -455 +465 @@
               CMessage& msg = msgs.back();
               msg << getId();
-              switch (hasData)
-              {
-                case RF_DATA:
-                  msg << getNStep() - 1 << data;
-                  break;
-                case RF_NODATA:
-                  msg << int(-2) << data;
-                  break;
-                case RF_EOF:                  
-                default:
-                  msg << int(-1);
-                  break;
-              }
-
+              
+              if (isEOF)  msg << int(-1);
+              else msg << getNStep() - 1 << data;
+         
               event.push(*itRank, 1, msg);
             }
@@ -484 +484 @@
       for (auto it = outLocalIndexStoreOnClient.begin(); it != outLocalIndexStoreOnClient.end(); ++it)
       {
-        CArray<size_t,1>& indexTmp = it->second;
-        CArray<double,1> tmp(indexTmp.numElements());
-        for (int idx = 0; idx < indexTmp.numElements(); ++idx)
-        {
-          tmp(idx) = data(indexTmp(idx));
-        } 
+        
 
         msgs.push_back(CMessage());
         CMessage& msg = msgs.back();
         msg << getId();
-        switch (hasData)
+        if (isEOF) msg << int(-1);
+        else 
         {
-          case RF_DATA:
-            msg << getNStep() - 1 << tmp;
-            break;
-          case RF_NODATA:
-            msg << int(-2) << tmp;
-            break;
-          case RF_EOF:                  
-          default:
-            msg << int(-1);
-            break;
+          CArray<size_t,1>& indexTmp = it->second;
+          CArray<double,1> tmp(indexTmp.numElements());
+          for (int idx = 0; idx < indexTmp.numElements(); ++idx) tmp(idx) = data(indexTmp(idx));
+          msg << getNStep() - 1 << tmp;
         }
-
         event.push(it->first, grid_->getNbReadSenders(client)[it->first], msg);
       }
@@ -1481 +1470 @@
   } 
   
+  /*!
+   * Connect field to a file reader filter to read data from file (on server side).
+   */
+  void CField::connectToFileReader(CGarbageCollector& gc)
+  {
+    fileServerReaderFilter_ = std::shared_ptr<CFileServerReaderFilter>(new CFileServerReaderFilter(gc, this));
+    fileServerReaderFilter_->connectOutput(inputFilter, 0);
+  } 
 
   /*!
@@ -1492 +1489 @@
     storeFilter = std::shared_ptr<CStoreFilter>(new CStoreFilter(gc, CContext::getCurrent(), grid_, detectMissingValues, defaultValue));
     instantDataFilter->connectOutput(storeFilter, 0);
+  }
+
+
+ 
+  void CField::connectToServerToClient(CGarbageCollector& gc)
+  {
+    serverToClientFilter_ = std::shared_ptr<CServerToClientFilter>(new CServerToClientFilter(gc, this, client));
+    instantDataFilter->connectOutput(serverToClientFilter_, 0);
   }
 

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

@@ -43 +43 @@
    class CFileWriterFilter;
    class CFileServerWriterFilter;
-
+   class CFileServerReaderFilter;
+   class CServerToClientFilter;
    ///--------------------------------------------------------------
 
@@ -183 +184 @@
         static bool dispatchEvent(CEventServer& event);
         void sendAllAttributesToServer(CContextClient* client) ; 
-        void sendUpdateData(const CArray<double,1>& data);
         void sendUpdateData(Time timestamp, const CArray<double,1>& data, CContextClient* client);
+        void sendUpdateDataServerToClient(bool isEOF, const CArray<double,1>& data, CContextClient* client) ;
+
         static void recvUpdateData(CEventServer& event);
         void recvUpdateData(std::map<int,CBufferIn*>& rankBuffers);
@@ -250 +252 @@
         void connectToServerInput(CGarbageCollector& gc) ;
         void connectToModelOutput(CGarbageCollector& gc);
+        void connectToFileReader(CGarbageCollector& gc) ;
+        void connectToServerToClient(CGarbageCollector& gc) ;
 
         void computeGridIndexToFileServer(void) ;
@@ -402 +406 @@
          //! The terminal filter which writes data to file
          std::shared_ptr<CFileServerWriterFilter> fileServerWriterFilter;
+
+         //! The source filter which read data from file
+         std::shared_ptr<CFileServerReaderFilter> fileServerReaderFilter_;
+
+         //! The terminal filter which send data from file server to client
+         std::shared_ptr<CServerToClientFilter> serverToClientFilter_;
+
+
    }; // class CField