Changeset 2482 for XIOS3/dev/XIOS_ATTACHED/src/node/field.cpp

Timestamp:

03/28/23 16:42:11 (15 months ago)

Author:

ymipsl

Message:

First guess in supression of attached mode replaced by online reader and write filters

YM

File:

• XIOS3/dev/XIOS_ATTACHED/src/node/field.cpp (modified) (16 diffs)

XIOS3/dev/XIOS_ATTACHED/src/node/field.cpp

@@ -20 +20 @@
 #include "temporal_filter.hpp"
 #include "server_from_client_source_filter.hpp"
+#include "client_online_reader_filter.hpp"
 #include "file_reader_source_filter.hpp"
+#include "grid_redistribute_filter.hpp"
 #include "tracer.hpp"
 #include "graph_package.hpp"
@@ -205 +207 @@
   TRY
   {
-    return clientFromServerSourceFilter_->sendReadDataRequest(tsDataRequested) ;
+    if (clientFromServerSourceFilter_) return clientFromServerSourceFilter_->sendReadDataRequest(tsDataRequested) ;
+    else if (clientOnlineReaderFilter_) return clientOnlineReaderFilter_->sendReadDataRequest(tsDataRequested) ;
+    else ERROR("bool CField::sendReadDataRequest(const CDate& tsDataRequested)", << "uninitialized source filter");
   }
   CATCH_DUMP_ATTR
@@ -217 +221 @@
   TRY
   {
-    return clientFromServerSourceFilter_->sendReadDataRequestIfNeeded() ;
+    if (clientFromServerSourceFilter_) return clientFromServerSourceFilter_->sendReadDataRequestIfNeeded() ;
+    else if (clientOnlineReaderFilter_) return clientOnlineReaderFilter_->sendReadDataRequestIfNeeded() ;
+    else ERROR("bool CField::sendReadDataRequestIfNeeded(void)", << "uninitialized source filter");
   }
   CATCH_DUMP_ATTR
@@ -304 +310 @@
   TRY
   {
-    clientFromServerSourceFilter_->checkForLateData() ;
+    if (clientFromServerSourceFilter_) return clientFromServerSourceFilter_->checkForLateData() ;
+    else if (clientOnlineReaderFilter_) return clientOnlineReaderFilter_->checkForLateData() ;
+    else ERROR("void CField::checkForLateDataFromServer(void)", << "uninitialized source filter");
   }
   CATCH_DUMP_ATTR 
@@ -315 +323 @@
     {
       checkForLateDataFromServer() ;
-      clientFromServerSourceFilter_->trigger(CContext::getCurrent()->getCalendar()->getCurrentDate()) ;
+      if (clientFromServerSourceFilter_) clientFromServerSourceFilter_->trigger(CContext::getCurrent()->getCalendar()->getCurrentDate()) ;
+      else if (clientOnlineReaderFilter_) clientOnlineReaderFilter_->trigger(CContext::getCurrent()->getCalendar()->getCurrentDate()) ;
     } 
     else if (hasCouplerIn())
@@ -325 +334 @@
   CATCH_DUMP_ATTR
 
-
-  void CField::checkIfMustAutoTrigger(void)
-  TRY
-  {
-    mustAutoTrigger = clientFromServerSourceFilter_ ? clientFromServerSourceFilter_->mustAutoTrigger() : false;
-  }
-  CATCH_DUMP_ATTR
-
-  void CField::autoTriggerIfNeeded(void)
-  TRY
-  {
-    if (mustAutoTrigger)
-      clientFromServerSourceFilter_->trigger(CContext::getCurrent()->getCalendar()->getCurrentDate());
-  }
-  CATCH_DUMP_ATTR
-
+  
 
   //----------------------------------------------------------------
@@ -506 +500 @@
                                               bool bufferForWriting)
   {
-    auto& contextBufferSize = bufferSize[client] ;
-    auto& contextMaxEventSize = maxEventSize[client] ;
-    const std::map<int, size_t> mapSize = grid_->getDataBufferSize(client, getId(), bufferForWriting);
+    auto& contextBufferSize = bufferSize[client_] ;
+    auto& contextMaxEventSize = maxEventSize[client_] ;
+    const std::map<int, size_t> mapSize = grid_->getDataBufferSize(client_, getId(), bufferForWriting);
     for(auto& it : mapSize )
     {
@@ -525 +519 @@
                                                    bool bufferForWriting)
   {
-    auto& contextBufferSize = bufferSize[client] ;
-    auto& contextMaxEventSize = maxEventSize[client] ;
-    const std::map<int, size_t> mapSize = grid_->getAttributesBufferSize(client, bufferForWriting);
+    auto& contextBufferSize = bufferSize[client_] ;
+    auto& contextMaxEventSize = maxEventSize[client_] ;
+    const std::map<int, size_t> mapSize = grid_->getAttributesBufferSize(client_, bufferForWriting);
     for(auto& it : mapSize )
     {
@@ -797 +791 @@
   {
     // insert temporal filter before sending to files
-    clientToServerStoreFilter_ = std::shared_ptr<CClientToServerStoreFilter>(new CClientToServerStoreFilter(gc, this, client));
+    clientToServerStoreFilter_ = std::shared_ptr<CClientToServerStoreFilter>(new CClientToServerStoreFilter(gc, this, client_));
     // insert temporal filter before sending to files
     getTemporalDataFilter(gc, fileOut_->output_freq)->connectOutput(clientToServerStoreFilter_, 0);
@@ -809 +803 @@
   } 
 
+  void CField::connectToOnlineWriter(CGarbageCollector& gc)
+  {
+    // insert temporal filter before sending to files
+    CField* fieldOut ;
+    redistributeFilter_ = std::shared_ptr<CGridRedistributeFilter>(new CGridRedistributeFilter(gc, this, fieldOut));
+    fieldOut->setFileOut(this->getFileOut());
+    fileOut_->replaceEnabledFields(this, fieldOut) ;
+    // insert temporal filter before sending to files
+    getTemporalDataFilter(gc, fileOut_->output_freq)->connectOutput(redistributeFilter_, 0);
+    fieldOut->inputFilter = std::shared_ptr<CPassThroughFilter>(new CPassThroughFilter(gc)); 
+    fieldOut->instantDataFilter = fieldOut->inputFilter ;
+    redistributeFilter_->connectOutput(fieldOut->inputFilter, 0);
+    fieldOut->connectToFileWriter(gc) ;
+    fieldOut->solveServerOperation() ; // might not be called, create a new time functor.... find a better solution later
+    const bool buildGraph_ = !build_workflow_graph.isEmpty() && build_workflow_graph == true ;
+  
+    if(buildGraph_) 
+    {
+      clientToServerStoreFilter_->graphPackage = new CGraphPackage;
+      clientToServerStoreFilter_->graphEnabled = true;
+      clientToServerStoreFilter_->graphPackage->inFields.push_back(this);
+    }
+  } 
+
+  void CField::connectToOnlineReader(CGarbageCollector& gc)
+  {
+    // insert temporal filter before sending to files
+    clientOnlineReaderFilter_ = std::shared_ptr<CClientOnlineReaderFilter>(new CClientOnlineReaderFilter(gc,this)) ;
+    clientOnlineReaderFilter_ -> connectOutput(inputFilter,0) ;
+  } 
+
   void CField::connectToCouplerOut(CGarbageCollector& gc)
   {
     // insert temporal filter before sending to files
-    clientToServerStoreFilter_ = std::shared_ptr<CClientToServerStoreFilter>(new CClientToServerStoreFilter(gc, this, client));
+    clientToServerStoreFilter_ = std::shared_ptr<CClientToServerStoreFilter>(new CClientToServerStoreFilter(gc, this, client_));
     instantDataFilter->connectOutput(clientToServerStoreFilter_, 0);
     const bool buildGraph_ = !build_workflow_graph.isEmpty() && build_workflow_graph == true ;
@@ -951 +976 @@
   void CField::connectToServerToClient(CGarbageCollector& gc)
   {
-    serverToClientStoreFilter_ = std::shared_ptr<CServerToClientStoreFilter>(new CServerToClientStoreFilter(gc, this, client));
+    serverToClientStoreFilter_ = std::shared_ptr<CServerToClientStoreFilter>(new CServerToClientStoreFilter(gc, this, client_));
     instantDataFilter->connectOutput(serverToClientStoreFilter_, 0);
     const bool buildGraph_ = !build_workflow_graph.isEmpty() && build_workflow_graph == true ;
@@ -1435 +1460 @@
   {
     CContext* context = CContext::getCurrent();
-    client = contextClient;
+    client_ = contextClient;
   
     // A grid is sent by a client (both for read or write) or by primary server (write only)
@@ -1441 +1466 @@
     {
       if (getRelFile()->mode.isEmpty() || (!getRelFile()->mode.isEmpty() && getRelFile()->mode == CFile::mode_attr::write))
-        grid_->setContextClient(contextClient);
+        /*grid_->setContextClient(contextClient) */; // => nothing to do with thats now, to remove...
     }
     else if (context->getServiceType()==CServicesManager::CLIENT)
     {
       if (grid_)
-        grid_->setContextClient(contextClient);
+        /*grid_->setContextClient(contextClient)*/; // => nothing to do with thats now, to remove...
       else
+
         ERROR( "CField::setContextClient(contextClient)",
                << "Grid not defined for " << getId()
@@ -1459 +1485 @@
   void CField::sendFieldToFileServer(void)
   {
-    CContext::getCurrent()->sendContextToFileServer(client);
-    getRelFile()->sendFileToFileServer(client);
+    CContext::getCurrent()->sendContextToFileServer(client_);
+    getRelFile()->sendFileToFileServer(client_);
     sentGrid_ = grid_-> duplicateSentGrid() ;
-    sentGrid_->sendGridToFileServer(client, false);
+    sentGrid_->sendGridToFileServer(client_, false);
     name = getFieldOutputName() ;
-    this->sendAllAttributesToServer(client);
-    this->sendAddAllVariables(client);
+    this->sendAllAttributesToServer(client_);
+    this->sendAddAllVariables(client_);
   }
 
   void CField::sendFieldToInputFileServer(void)
   {
-    CContext::getCurrent()->sendContextToFileServer(client);
-    getRelFile()->sendFileToFileServer(client);
+    CContext::getCurrent()->sendContextToFileServer(client_);
+    getRelFile()->sendFileToFileServer(client_);
     sentGrid_ = grid_-> duplicateSentGrid() ;
-    sentGrid_->sendGridToFileServer(client, true);
+    sentGrid_->sendGridToFileServer(client_, true);
     read_access=true ; // not the best solution, but on server side, the field must be a starting point of the workflow
                        // must be replace by a better solution when implementing filters for reading and send to client
                        // on server side
-    this->sendAllAttributesToServer(client);
-    this->sendAddAllVariables(client);
+    this->sendAllAttributesToServer(client_);
+    this->sendAddAllVariables(client_);
   }
 
@@ -1486 +1512 @@
     else sendFieldToCouplerOut_done_=true ;
     sentGrid_ = grid_-> duplicateSentGrid() ;
-    sentGrid_->sendGridToCouplerOut(client, this->getId());
+    sentGrid_->sendGridToCouplerOut(client_, this->getId());
     this->sendGridCompleted();
 
@@ -1502 +1528 @@
       CEventClient event(getType(),EVENT_ID_GRID_COMPLETED);
 
-      if (client->isServerLeader())
+      if (client_->isServerLeader())
       {
         CMessage msg;
         msg<<this->getId();
-        for (auto& rank : client->getRanksServerLeader()) event.push(rank,1,msg);
-        client->sendEvent(event);
-      }
-      else client->sendEvent(event);
+        for (auto& rank : client_->getRanksServerLeader()) event.push(rank,1,msg);
+        client_->sendEvent(event);
+      }
+      else client_->sendEvent(event);
    }
    CATCH_DUMP_ATTR