Changeset 1934 for XIOS/dev/dev_ym/XIOS_COUPLING/src/filter

Timestamp:

09/16/20 18:34:23 (4 years ago)

Author:

ymipsl

Message:

Big update on on going work related to data distribution and transfer between clients and servers.
Revisite of the source and store filter using "connectors".

-> inputs work again

YM

Location:

XIOS/dev/dev_ym/XIOS_COUPLING/src/filter

Files:

• client_from_client_source_filter.cpp (modified) (1 diff)
• client_from_client_source_filter.hpp (modified) (1 diff)
• client_from_server_source_filter.cpp (modified) (5 diffs)
• client_from_server_source_filter.hpp (modified) (2 diffs)
• file_reader_source_filter.cpp (modified) (5 diffs)
• server_to_client_filter.cpp (deleted)
• server_to_client_filter.hpp (deleted)
• server_to_client_store_filter.cpp (added)
• server_to_client_store_filter.hpp (added)

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

@@ -14 +14 @@
     {
       CContext* context = CContext::getCurrent();
+      field_=field ;
       grid_= field-> getGrid();
 

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

@@ -37 +37 @@
 
     private:
+      CField* field_ ;
       CGrid* grid_;             //!< The grid attached to the data the filter can accept
       CDuration freqOp_ ;

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

@@ -6 +6 @@
 #include "calendar_util.hpp"
 #include "context.hpp"
+#include "event_client.hpp"
+#include "timer.hpp"
+#include "tracer.hpp"
 #include <limits> 
 
@@ -14 +17 @@
   {
     CContext* context = CContext::getCurrent();
+    field_ = field ;
     grid_= field->getGrid();
-    freqOp_ = field->fileIn_->output_freq ;
+    freqOp_ = field->getRelFile()->output_freq ;
+    client_= field->getRelFile()->getContextClient() ;
     lastDateReceived_ = context->getCalendar()->getInitDate();
     offset_ = field->freq_offset ;
@@ -39 +44 @@
       if (!wasEOF) dateEOF_ = lastDateReceived_;
       packet->status = CDataPacket::END_OF_STREAM;
+      info(20)<<"Receiv Data from server to client: FieldId : "<<field_->getId()<<endl ;
+      info(20)<<"lastDateReceived_ "<<lastDateReceived_<< "  date "<<packet->date<<"  ----> EOF"<<endl; 
+
     }
     else 
     {
-      grid_->getServerFromClientConnector()->transfer(event, packet->data) ;
+      CContextClient* client = event.getContextServer()->getAssociatedClient() ;
+      grid_->getClientFromServerConnector(client)->transfer(event, packet->data) ; // to avoid to make a search in map for corresponding client connector, 
+     
+      info(20)<<"Receiv Data from server to client: FieldId : "<<field_->getId()<<endl ;
+      info(20)<<"lastDateReceived_ "<<lastDateReceived_<< "  date "<<packet->date<<endl;                                                                                    // make a registration at initialization once
       packet->status = CDataPacket::NO_ERROR;
     }
@@ -49 +61 @@
   }
  
+  int CClientFromServerSourceFilter::sendReadDataRequest(const CDate& tsDataRequested)
+  {
+    CContext* context = CContext::getCurrent();
+    lastDataRequestedFromServer_ = tsDataRequested;
+
+    // No need to send the request if we are sure that we are already at EOF
+    if (!isEOF_ || context->getCalendar()->getCurrentDate() <= dateEOF_)
+    {
+      CEventClient event(field_->getType(), CField::EVENT_ID_READ_DATA);
+      if (client_->isServerLeader())
+      {
+        CMessage msg;
+        msg << field_->getId();
+        for(auto& rank : client_->getRanksServerLeader()) event.push(rank, 1, msg);
+        client_->sendEvent(event);
+      }
+      else client_->sendEvent(event);
+    }
+    else 
+    {
+      CDataPacketPtr packet(new CDataPacket);
+      packet->date = tsDataRequested;
+      packet->timestamp = packet->date ;
+      packet->status = CDataPacket::END_OF_STREAM;
+      onOutputReady(packet);
+    }
+
+    wasDataRequestedFromServer_ = true;
+
+    return !isEOF_;
+  }
+
+  bool CClientFromServerSourceFilter::sendReadDataRequestIfNeeded(void)
+  TRY
+  {
+    const CDate& currentDate = CContext::getCurrent()->getCalendar()->getCurrentDate();
+
+    bool dataRequested = false;
+
+    while (currentDate >= lastDataRequestedFromServer_)
+    {
+      info(20) << "currentDate : " << currentDate << endl ;
+      info(20) << "Field : " << field_->getId() << endl ;
+      info(20) << "lastDataRequestedFromServer : " << lastDataRequestedFromServer_ << endl ;
+      info(20) << "freqOp : " << freqOp_ << endl ;
+      info(20) << "lastDataRequestedFromServer + fileIn_->output_freq.getValue() : " << lastDataRequestedFromServer_ + freqOp_ << endl ;
+
+      dataRequested |= sendReadDataRequest(lastDataRequestedFromServer_ + freqOp_);
+    }
+
+    return dataRequested;
+  }
+  CATCH
+
+  void CClientFromServerSourceFilter::checkForLateData(void)
+  TRY
+  {
+    CContext* context = CContext::getCurrent();
+    // Check if data previously requested has been received as expected
+    if (wasDataRequestedFromServer_ && ! isEOF_)
+    {
+      CTimer timer("CClientFromServerSourceFilter::checkForLateDataFromServer");
+      timer.resume();
+      traceOff() ;
+      timer.suspend();
+      
+      bool isLate;
+      do
+      {
+        isLate = isDataLate();
+        if (isLate)
+        {
+          timer.resume();
+          context->globalEventLoop();
+          timer.suspend();
+        }
+      }
+      while (isLate && timer.getCumulatedTime() < CXios::recvFieldTimeout);
+      timer.resume();
+      traceOn() ;
+      timer.suspend() ;
+
+
+      if (isLate)
+        ERROR("void CClientFromServerSourceFilter::checkForLateDataFromServer(void)",
+              << "Late data at timestep = " << context->getCalendar()->getCurrentDate());
+    }
+  }
+  CATCH
+
+  
   bool CClientFromServerSourceFilter::isDataLate(void)
   {
@@ -60 +163 @@
     
   }
+
+
 } // namespace xios

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

@@ -6 +6 @@
 #include "output_pin.hpp"
 #include "event_server.hpp"
+#include "context_client.hpp"
 #include "calendar_util.hpp"
 
@@ -38 +39 @@
       bool isDataLate(void) ;
       bool isEOF() {return isEOF_ ;}
+      int  sendReadDataRequest(const CDate& tsDataRequested) ;
+      bool sendReadDataRequestIfNeeded(void) ;
+      void checkForLateData(void) ;
 
      private:
+      CField* field_;
       CGrid* grid_;             //!< The grid attached to the data the filter can accept
       CDuration freqOp_ ;
       CDuration offset_ ;
-
+      CContextClient* client_ = nullptr ;
       bool wasDataAlreadyReceived_= false ;
       CDate lastDateReceived_ ;
+      bool wasDataRequestedFromServer_ = false ;
+      CDate lastDataRequestedFromServer_ ;
+
       bool isEOF_ = false ;
       CDate dateEOF_ ;

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

@@ -30 +30 @@
     packet->status = CDataPacket::NO_ERROR;
 
-    CArray<double,1> data ;
     if (!isInitialized_)  initialize() ;
     CField::EReadField readState = CField::RF_DATA;
-
     if ( nStepMax_==0 || (nStep_ >= nStepMax_ && !isCyclic_)) readState = CField::RF_EOF;
   
     if (CField::RF_EOF != readState)
     {
-      if (file_->isEmptyZone()) readData(data) ;
+      if (!file_->isEmptyZone()) readData(packet->data) ;
       else readState = CField::RF_NODATA;
     }
@@ -45 +43 @@
     if (readState == CField::RF_DATA) packet->status = CDataPacket::NO_ERROR;
     else packet->status = CDataPacket::END_OF_STREAM;
-            
+    
+    info(20)<<"Read data from file : FieldId "<<field_->getId()<<"  nStep "<<nStep_<<"  date : "<<packet->date<<endl ;
+           
     onOutputReady(packet);
   }
@@ -53 +53 @@
     CContext* context = CContext::getCurrent();
     file_->initRead();
-    if (file_->isEmptyZone())
+    if (!file_->isEmptyZone())
     {      
       file_->checkReadFile();
@@ -60 +60 @@
     }
     MPI_Allreduce(MPI_IN_PLACE, &nStepMax_, 1, MPI_INT, MPI_MAX, context->getIntraComm());
+    isInitialized_=true;
   }
 
@@ -65 +66 @@
   {
     CGridLocalConnector*  connector = grid_->getFullToWorkflowConnector() ;
-    CArray<double,1> dataIn(connector->getDstSize()) ;
+    CArray<double,1> dataIn(connector->getSrcSize()) ;
     file_->getDataInput()->readFieldData(field_, nStep_%nStepMax_, dataIn);
+    data.resize(connector->getDstSize()) ;
     connector->transfer(dataIn, data) ;