Changeset 1934 for XIOS

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

Files:

• config/field_attribute.conf (modified) (1 diff)
• distribution/grid_scatterer_connector.hpp (modified) (2 diffs)
• filter/client_from_client_source_filter.cpp (modified) (1 diff)
• filter/client_from_client_source_filter.hpp (modified) (1 diff)
• filter/client_from_server_source_filter.cpp (modified) (5 diffs)
• filter/client_from_server_source_filter.hpp (modified) (2 diffs)
• filter/file_reader_source_filter.cpp (modified) (5 diffs)
• filter/server_to_client_filter.cpp (deleted)
• filter/server_to_client_filter.hpp (deleted)
• filter/server_to_client_store_filter.cpp (added)
• filter/server_to_client_store_filter.hpp (added)
• node/axis.cpp (modified) (2 diffs)
• node/axis.hpp (modified) (1 diff)
• node/context.cpp (modified) (1 diff)
• node/domain.cpp (modified) (2 diffs)
• node/domain.hpp (modified) (2 diffs)
• node/field.cpp (modified) (15 diffs)
• node/field.hpp (modified) (3 diffs)
• node/file.cpp (modified) (1 diff)
• node/grid.cpp (modified) (9 diffs)
• node/grid.hpp (modified) (2 diffs)
• node/scalar.cpp (modified) (1 diff)
• node/scalar.hpp (modified) (1 diff)
• test/generic_testcase.f90 (modified) (1 diff)

XIOS/dev/dev_ym/XIOS_COUPLING/src/config/field_attribute.conf

@@ -24 +24 @@
 DECLARE_ATTRIBUTE(StdString, scalar_ref, false)
 DECLARE_ATTRIBUTE(StdString, grid_ref)
-DECLARE_ATTRIBUTE(StdString, field_ref)
+DECLARE_ATTRIBUTE(StdString, field_ref, false)
 DECLARE_ATTRIBUTE(StdString, grid_path)
 

XIOS/dev/dev_ym/XIOS_COUPLING/src/distribution/grid_scatterer_connector.hpp

@@ -68 +68 @@
       {
         list<CMessage> messages;
-        for(auto ranksData : dataOut)
+        for(auto& ranksData : dataOut)
         {
           int rank = ranksData.first ;
@@ -78 +78 @@
         }
         client->sendEvent(event) ;
-      }
- 
+      } 
+      
+      void transfer(CContextClient* client, CEventClient& event, const CMessage& messageHeader)
+      {
+        list<CMessage> messages;
+        for(auto& it : nbSenders_)
+        {
+          int rank = it.first ;
+          auto& nbSender = it.second ;
+
+          messages.push_back(CMessage(messageHeader));
+          event.push(rank, nbSenders_[rank], messages.back());
+        }
+        client->sendEvent(event) ;
+      }  
   };
 }

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) ; 
 

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

@@ -1706 +1706 @@
     clientToServerConnector_[client]->transfer(maskIn,client,event3,message3) ; 
 
+    clientFromServerConnector_[client] = new CGathererConnector(clientToServerElement.getView(CElementView::FULL), localElement_->getView(CElementView::WORKFLOW));
+    clientFromServerConnector_[client]->computeConnector() ;
 
 
@@ -1770 +1772 @@
       serverFromClientConnector_ = new CGathererConnector(elementFrom_->getView(CElementView::FULL), localElement_->getView(CElementView::WORKFLOW)) ;
       serverFromClientConnector_->computeConnector() ;
+
+      serverToClientConnector_ = new CScattererConnector(localElement_->getView(CElementView::WORKFLOW), elementFrom_->getView(CElementView::FULL),
+                                                         context->getIntraComm()) ;
+      serverToClientConnector_->computeConnector() ;
+
     }
   }

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

@@ -277 +277 @@
         CGathererConnector* getServerFromClientConnector(void) { return serverFromClientConnector_ ;}
 
-
+       private:
+         CScattererConnector*  serverToClientConnector_ = nullptr ;
+       public: 
+         CScattererConnector* getServerToClientConnector(void) { return serverToClientConnector_ ;} 
+
+       private:
+          map<CContextClient*,CGathererConnector*>  clientFromServerConnector_  ;
+       public: 
+         CGathererConnector* getClientFromServerConnector(CContextClient* client) { return clientFromServerConnector_[client] ;} 
 
          DECLARE_REF_FUNC(Axis,axis)

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

@@ -1246 +1246 @@
       for(auto field : fileInField) 
       {
-        field->connectToServerInput(garbageCollector) ; // connect tFhe field to server filter
-        field->computeGridIndexToFileServer() ; // compute grid index for transfer to the server context
+        field->connectToServerInput(garbageCollector) ; // connect the field to server filter
+    // obsolete    field->computeGridIndexToFileServer() ; // compute grid index for transfer to the server context
         field->sendFieldToInputFileServer() ;
+        fileInFields_.push_back(field) ;
       }
     }

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

@@ -2354 +2354 @@
     clientToServerConnector_[client]->transfer(maskIn,client,event3,message3) ; 
     
+    clientFromServerConnector_[client] = new CGathererConnector(clientToServerElement.getView(CElementView::FULL), localElement_->getView(CElementView::WORKFLOW));
+    clientFromServerConnector_[client]->computeConnector() ;
+
   }
   CATCH
@@ -2421 +2424 @@
       serverFromClientConnector_ = new CGathererConnector(elementFrom_->getView(CElementView::FULL), localElement_->getView(CElementView::WORKFLOW)) ;
       serverFromClientConnector_->computeConnector() ;
+      
+      serverToClientConnector_ = new CScattererConnector(localElement_->getView(CElementView::WORKFLOW), elementFrom_->getView(CElementView::FULL),
+                                                         context->getIntraComm()) ;
+      serverToClientConnector_->computeConnector() ;
+ 
     }
   }

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

@@ -349 +349 @@
        public: 
          CScattererConnector* getClientToServerConnector(CContextClient* client) { return clientToServerConnector_[client] ;}
+       
        private:
          CGathererConnector*  gathererConnector_ ;
@@ -354 +355 @@
          CDistributedElement* elementFrom_ ;
        public:
-        CGathererConnector* getServerFromClientConnector(void) { return serverFromClientConnector_ ;}
-
+         CGathererConnector* getServerFromClientConnector(void) { return serverFromClientConnector_ ;}
+
+       private:
+         CScattererConnector*  serverToClientConnector_ = nullptr ;
+       public: 
+         CScattererConnector* getServerToClientConnector(void) { return serverToClientConnector_ ;} 
+
+       private:
+         map<CContextClient*,CGathererConnector*>  clientFromServerConnector_  ;
+       public: 
+         CGathererConnector* getClientFromServerConnector(CContextClient* client) { return clientFromServerConnector_[client] ;}        
          
 

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

@@ -150 +150 @@
   CATCH
 
-/* obsolete old interface
-  void CField::sendUpdateData(Time timeStamp, const CArray<double,1>& data, CContextClient* client)
-  TRY
-  {
-    CTimer::get("Field : send data").resume();
-    int receiverSize = client->serverSize;
-
-    CEventClient event(getType(), EVENT_ID_UPDATE_DATA);
-
-    map<int, CArray<int,1> >::iterator it;
-    list<CMessage> list_msg;
-    list<CArray<double,1> > list_data;
-
-    if (!grid_->doGridHaveDataDistributed(client))
-    {
-       if (client->isServerLeader())
-       {
-          for (it = grid_->storeIndex_toSrv_[client].begin(); it != grid_->storeIndex_toSrv_[client].end(); it++)
-          {
-            int rank = it->first;
-            CArray<int,1>& index = it->second;
-
-            list_msg.push_back(CMessage());
-            list_data.push_back(CArray<double,1>(index.numElements()));
-
-            CArray<double,1>& data_tmp = list_data.back();
-            for (int n = 0; n < data_tmp.numElements(); n++) data_tmp(n) = data(index(n));
-
-            list_msg.back() << getId() << timeStamp << data_tmp;
-            event.push(rank, 1, list_msg.back());
-          }
-          client->sendEvent(event);
-        }
-      else client->sendEvent(event);
-    }
-    else
-    {
-      for (it = grid_->storeIndex_toSrv_[client].begin(); it != grid_->storeIndex_toSrv_[client].end(); it++)
-      {
-        int rank = it->first;
-        CArray<int,1>& index = it->second;
-
-        list_msg.push_back(CMessage());
-        list_data.push_back(CArray<double,1>(index.numElements()));
-
-        CArray<double,1>& data_tmp = list_data.back();
-        for (int n = 0; n < data_tmp.numElements(); n++) data_tmp(n) = data(index(n));
-
-        list_msg.back() << getId() << timeStamp << data_tmp;
-        event.push(rank, grid_->nbSenders_[receiverSize][rank], list_msg.back());
-      }
-      client->sendEvent(event);
-    }
-
-    CTimer::get("Field : send data").suspend();
-  }
-  CATCH_DUMP_ATTR
-*/
-
   void CField::sendUpdateData(Time timeStamp, const CArray<double,1>& data, CContextClient* client)
   TRY
@@ -222 +163 @@
   CATCH_DUMP_ATTR
 
-  /* old version obsolete
-  void CField::recvUpdateData(CEventServer& event)
-  TRY
-  {
-    std::map<int,CBufferIn*> rankBuffers;
-
-    list<CEventServer::SSubEvent>::iterator it;
-    string fieldId;
-    CTimer::get("Field : recv data").resume();
-    for (it = event.subEvents.begin(); it != event.subEvents.end(); ++it)
-    {
-      int rank = it->rank;
-      CBufferIn* buffer = it->buffer;
-      *buffer >> fieldId;
-      rankBuffers[rank] = buffer;
-    }
-    get(fieldId)->recvUpdateData(rankBuffers);
-    CTimer::get("Field : recv data").suspend();
-  }
-  CATCH
-*/
-
+  
   void CField::recvUpdateData(CEventServer& event)
   TRY
@@ -260 +180 @@
   }
   CATCH
-/*
-  void  CField::recvUpdateDataFromClient(CEventServer& event)
-  TRY
-  {
-    Time timeStamp ;
-    for (auto& subEvent : event.subEvents) (*subEvent.buffer) >> timeStamp  ;
-
-    CArray<double,1> recvData ;
-    getGrid()->getServerFromClientConnector()->transfer(event,recvData) ;
-    this->setData(recvData);
-  }
-  CATCH
-*/
-
-/*  
-  void CField::recvUpdateDataFromCoupler(CEventServer& event)
-  TRY
-  {
-    CContext* context = CContext::getCurrent();
-    Time timeStamp ;
-    if (wasDataAlreadyReceivedFromServer)
-    {  
-      lastDataReceivedFromServer = lastDataReceivedFromServer + freq_op;
-    }
-    else
-    {
-      // unlikely to input from file server where data are received at ts=0
-      // for coupling, it would be after the first freq_op, because for now we don't have
-      // restart mecanism to send the value at ts=0. It must be changed in future
-      lastDataReceivedFromServer = context->getCalendar()->getInitDate();
-      wasDataAlreadyReceivedFromServer = true;
-    }
-
-    CArray<double,1> recvData ;
-    getGrid()->getServerFromClientConnector()->transfer(event,recvData) ;
-    clientSourceFilter->streamData(lastDataReceivedFromServer, recvData);
-
-  }
-  CATCH_DUMP_ATTR
-*/
-
-
-
-
-  /* old interface to be removed.... */
-  void  CField::recvUpdateDataFromClient(std::map<int,CBufferIn*>& rankBuffers)
-  TRY
-  {
-    // ym to remove latter
-    /*
-    CContext* context = CContext::getCurrent();
-    Time timeStamp ;
-    size_t sizeData = 0;
-    if (0 == recvDataSrv.numElements())
-    {            
-      CArray<int,1>& storeClient = grid_->getStoreIndex_client();
-
-      // Gather all data from different clients      
-      recvDataSrv.resize(storeClient.numElements());
-      recvFoperationSrv = std::shared_ptr<func::CFunctor>(new func::CInstant(recvDataSrv));
-    }
-
-    CArray<double,1> recv_data_tmp(recvDataSrv.numElements());    
-    const CDate& currDate = context->getCalendar()->getCurrentDate();
-    CDuration offsetAllButMonth (freq_offset.getValue().year, 0 , freq_offset.getValue().day,
-                                 freq_offset.getValue().hour, freq_offset.getValue().minute,
-                                 freq_offset.getValue().second, freq_offset.getValue().timestep);
-    const CDate opeDate   = (last_operation_srv - offsetAllButMonth + context->getCalendar()->getTimeStep())
-                             + freq_op + freq_operation_srv - freq_op - context->getCalendar()->getTimeStep() + offsetAllButMonth;
-
-    if (opeDate <= currDate)
-    {
-      
-      auto& outLocalIndexStoreOnClient = grid_-> getOutLocalIndexStoreOnClient() ;
-      for (auto it = outLocalIndexStoreOnClient.begin(); it != outLocalIndexStoreOnClient.end(); ++it)
-      {
-        CArray<double,1> tmp;
-        CArray<size_t,1>& indexTmp = it->second;
-        *(rankBuffers[it->first]) >> timeStamp >> tmp;
-        for (int idx = 0; idx < indexTmp.numElements(); ++idx) recv_data_tmp(indexTmp(idx)) = tmp(idx);
-      }
-    }
-*/
-    Time timeStamp ;
-    CArray<int,1>& storeClient = grid_->getStoreIndex_client(); // replace it with local size
-    CArray<double,1> recv_data_tmp(storeClient.numElements());
-    auto& outLocalIndexStoreOnClient = grid_-> getOutLocalIndexStoreOnClient() ;
-    for (auto it = outLocalIndexStoreOnClient.begin(); it != outLocalIndexStoreOnClient.end(); ++it)
-    {
-      CArray<double,1> tmp;
-      CArray<size_t,1>& indexTmp = it->second;
-      *(rankBuffers[it->first]) >> timeStamp >> tmp;
-      for (int idx = 0; idx < indexTmp.numElements(); ++idx) recv_data_tmp(indexTmp(idx)) = tmp(idx);
-    }
-
-    this->setData(recv_data_tmp);
-/*
-    // delete incomming flux for server only
-    recvFoperationSrv.reset() ;
-    recvDataSrv.reset() ;
-*/
-  }
-  CATCH_DUMP_ATTR
-
-  /* ym : old interface : to be removed...
-  void CField::writeUpdateData(const CArray<double,1>& data)
-  TRY
-  {
-    CContext* context = CContext::getCurrent();
-
-    const CDate& currDate = context->getCalendar()->getCurrentDate();
-    CDuration offsetAllButMonth (freq_offset.getValue().year, 0 , freq_offset.getValue().day,
-                                   freq_offset.getValue().hour, freq_offset.getValue().minute,
-                                   freq_offset.getValue().second, freq_offset.getValue().timestep);
-    const CDate opeDate   = (last_operation_srv - offsetAllButMonth + context->getCalendar()->getTimeStep())
-                              + freq_op + freq_operation_srv - freq_op - context->getCalendar()->getTimeStep() + offsetAllButMonth;
-    const CDate writeDate = last_Write_srv + freq_write_srv;
-
-    if (opeDate <= currDate)
-    {
-      (*recvFoperationSrv)(data);
-      last_operation_srv = currDate;
-    }
-
-    if (writeDate < (currDate + freq_operation_srv))
-    {
-      recvFoperationSrv->final();
-      last_Write_srv = writeDate;
-      grid_->computeWrittenIndex();
-      writeField();
-      lastlast_Write_srv = last_Write_srv;
-    }
-  }
-  CATCH_DUMP_ATTR
-  */
-
+
+  
   void CField::writeUpdateData(const CArray<double,1>& data)
   TRY
@@ -434 +220 @@
     \param [in] tsDataRequested timestamp when the call is made
   */
-  bool CField::sendReadDataRequest(const CDate& tsDataRequested, CContextClient* client)
-  TRY
-  {
-    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(getType(), EVENT_ID_READ_DATA);
-      if (client->isServerLeader())
-      {
-        CMessage msg;
-        msg << getId();
-        const std::list<int>& ranks = client->getRanksServerLeader();
-        for (std::list<int>::const_iterator itRank = ranks.begin(), itRankEnd = ranks.end(); itRank != itRankEnd; ++itRank)
-          event.push(*itRank, 1, msg);
-        client->sendEvent(event);
-      }
-      else client->sendEvent(event);
-    }
-    else serverSourceFilter->signalEndOfStream(tsDataRequested);
-
-    wasDataRequestedFromServer = true;
-
-    return !isEOF;
-  }
-  CATCH_DUMP_ATTR
-
+  bool CField::sendReadDataRequest(const CDate& tsDataRequested)
+  TRY
+  {
+    return clientFromServerSourceFilter_->sendReadDataRequest(tsDataRequested) ;
+  }
+  CATCH_DUMP_ATTR
+ 
+  
   /*!
   Send request new data read from file if need be, that is the current data is out-of-date.
@@ -471 +235 @@
   TRY
   {
-    const CDate& currentDate = CContext::getCurrent()->getCalendar()->getCurrentDate();
-
-    bool dataRequested = false;
-
-    while (currentDate >= lastDataRequestedFromServer)
-    {
-      info(20) << "currentDate : " << currentDate << endl ;
-      info(20) << "lastDataRequestedFromServer : " << lastDataRequestedFromServer << endl ;
-      info(20) << "fileIn_->output_freq.getValue() : " << fileIn_->output_freq.getValue() << endl ;
-      info(20) << "lastDataRequestedFromServer + fileIn_->output_freq.getValue() : " << lastDataRequestedFromServer + fileIn_->output_freq << endl ;
-
-      dataRequested |= sendReadDataRequest(lastDataRequestedFromServer + fileIn_->output_freq, fileIn_->getContextClient());
-    }
-
-    return dataRequested;
-  }
-  CATCH_DUMP_ATTR
+    return clientFromServerSourceFilter_->sendReadDataRequestIfNeeded() ;
+  }
+  CATCH_DUMP_ATTR
+
 
   void CField::recvReadDataRequest(CEventServer& event)
@@ -512 +263 @@
   CATCH_DUMP_ATTR  
 
-/* old interface -> to remove
-  void CField::recvReadDataRequest(CContextServer* server)
-  TRY
-  {
-        
-    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)
@@ -583 +320 @@
   CATCH_DUMP_ATTR
 
-  /*!
-    Read field from a file.
-    A field is read with the distribution of data on the server side
-    \return State of field can be read from a file
-  */
-  // obsolete to remove
-  /*
-  CField::EReadField CField::readField(CArray<double,1>& data)
-  TRY
-  {
-    CContext* context = CContext::getCurrent();
-    grid_->computeWrittenIndex();
-    getRelFile()->initRead();
-    EReadField readState = RF_DATA;
-
-    if (!getRelFile()->isEmptyZone())
-    {      
-      if (grid_->doGridHaveDataToWrite() || getRelFile()->type == CFile::type_attr::one_file)      
-      {
-          CArray<int,1>& storeClient = grid_->getStoreIndex_client();          
-          data.resize(storeClient.numElements());          
-        
-        getRelFile()->checkReadFile();
-
-        if (!nstepMax)
-        {
-          nstepMax = getRelFile()->getDataInput()->getFieldNbRecords(CField::get(this));
-        }
-
-        this->incrementNStep();
-
-        if (getNStep() > nstepMax && (getRelFile()->cyclic.isEmpty() || !getRelFile()->cyclic) )
-          readState = RF_EOF;
-
-        if (RF_EOF != readState)
-          getRelFile()->getDataInput()->readFieldData(CField::get(this),data);
-      }
-    }
-    else
-    {
-      this->incrementNStep();
-      if (getNStep() > nstepMax && (getRelFile()->cyclic.isEmpty() || !getRelFile()->cyclic) )
-        readState = RF_EOF;
-      else
-        readState = RF_NODATA;
-
-      if (!nstepMaxRead) // This can be a bug if we try to read field from zero time record
-        readState = RF_NODATA;
-    }
-
-    if (!nstepMaxRead)
-    {
-       MPI_Allreduce(MPI_IN_PLACE, &nstepMax, 1, MPI_INT, MPI_MAX, context->intraComm_);
-       nstepMaxRead = true;
-    }
-
-    return readState;
-  }
-  CATCH_DUMP_ATTR
-  */
-
+  
   /*
     Receive read data from server.
@@ -655 +332 @@
     string fieldId;
     for (auto& subEvent : event.subEvents) (*subEvent.buffer) >> fieldId  ;
-    get(fieldId)->recvReadDataReady(event);
+    get(fieldId)->receiveReadDataReady(event);
   }
   CATCH
 
+  void CField::receiveReadDataReady(CEventServer& event)
+  TRY
+  {
+    clientFromServerSourceFilter_->streamData(event) ;    
+  }
+  CATCH_DUMP_ATTR
 
   /* old interface to be removed ..*/
@@ -742 +425 @@
   }
   CATCH_DUMP_ATTR
-
-
-
-  void CField::receiveReadDataReady(CEventServer& event)
-  TRY
-  {
-    clientFromServerSourceFilter_->streamData(event) ;    
-  }
-  CATCH_DUMP_ATTR
-
 
 
@@ -786 +459 @@
   CATCH_DUMP_ATTR
 
-
   void CField::checkForLateDataFromServer(void)
   TRY
   {
-    CContext* context = CContext::getCurrent();
-    // Check if data previously requested has been received as expected
-    if (wasDataRequestedFromServer && !clientFromServerSourceFilter_->isEOF())
-    {
-      CTimer timer("CField::checkForLateDataFromServer");
-      timer.resume();
-      traceOff() ;
-      timer.suspend();
-      
-      bool isDataLate;
-      do
-      {
-        isDataLate=clientFromServerSourceFilter_->isDataLate();
-        if (isDataLate)
-        {
-          timer.resume();
-
-//ym          context->checkBuffersAndListen();
-//ym            context->eventLoop();
-          context->globalEventLoop();
-
-          timer.suspend();
-        }
-      }
-      while (isDataLate && timer.getCumulatedTime() < CXios::recvFieldTimeout);
-      timer.resume();
-      traceOn() ;
-      timer.suspend() ;
-
-
-      if (isDataLate)
-        ERROR("void CField::checkForLateDataFromServer(void)",
-              << "Late data at timestep = " << context->getCalendar()->getCurrentDate());
-    }
-  }
-  CATCH_DUMP_ATTR
-
+    clientFromServerSourceFilter_->checkForLateData() ;
+  }
+  CATCH_DUMP_ATTR 
+  
+  
   void CField::triggerLateField(void)
   TRY
@@ -1508 +1148 @@
   void CField::connectToServerInput(CGarbageCollector& gc)
   {
-    const bool detectMissingValues = (!detect_missing_value.isEmpty() && !default_value.isEmpty() && detect_missing_value == true);
-    const double defaultValue  = detectMissingValues ? default_value : (!default_value.isEmpty() ? default_value : 0.0);
-
     checkTimeAttributes();
-    serverSourceFilter = std::shared_ptr<CSourceFilter>(new CSourceFilter(gc, grid_, true, false, freq_offset, true,
-                                                                          detectMissingValues, defaultValue));
-    //serverSourceFilter = std::shared_ptr<CSourceFilter>(new CSourceFilter(gc,  grid_, false, false));
-    serverSourceFilter -> connectOutput(inputFilter,0) ;
+    clientFromServerSourceFilter_ = std::shared_ptr<CClientFromServerSourceFilter>(new CClientFromServerSourceFilter(gc,this)) ;
+    clientFromServerSourceFilter_ -> connectOutput(inputFilter,0) ;
   } 
 
@@ -1561 +1196 @@
   {
     fileReaderSourceFilter_ = std::shared_ptr<CFileReaderSourceFilter>(new CFileReaderSourceFilter(gc, this));
-    instantDataFilter->connectOutput(inputFilter, 0);
+    fileReaderSourceFilter_->connectOutput(inputFilter, 0);
   }
 
@@ -1578 +1213 @@
   void CField::connectToServerToClient(CGarbageCollector& gc)
   {
-    serverToClientFilter_ = std::shared_ptr<CServerToClientFilter>(new CServerToClientFilter(gc, this, client));
-    instantDataFilter->connectOutput(serverToClientFilter_, 0);
+    serverToClientStoreFilter_ = std::shared_ptr<CServerToClientStoreFilter>(new CServerToClientStoreFilter(gc, this, client));
+    instantDataFilter->connectOutput(serverToClientStoreFilter_, 0);
   }
 
@@ -2290 +1925 @@
   }
 
+  void CField::sendFieldToInputFileServer(void)
+  {
+    CContext::getCurrent()->sendContextToFileServer(client);
+    getRelFile()->sendFileToFileServer(client);
+    grid_->sendGridToFileServer(client);
+    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);
+  }
+
   void CField::sendFieldToCouplerOut(void)
   {
@@ -2341 +1988 @@
 
 
-
-
-
-  void CField::sendFieldToInputFileServer(void)
-  {
-    CContext::getCurrent()->sendContextToFileServer(client);
-    getRelFile()->sendFileToFileServer(client);
-    grid_->sendGridToFileServer(client);
-    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);
-  }
-
   void CField::sendAddAllVariables(CContextClient* client)
   TRY

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

@@ -23 +23 @@
 #include "client_from_server_source_filter.hpp"
 #include "client_to_model_store_filter.hpp"
+#include "server_to_client_store_filter.hpp"
 
 
@@ -203 +204 @@
         
         void writeField(const CArray<double,1>& data);
-        bool sendReadDataRequest(const CDate& tsDataRequested, CContextClient* client);
+        bool sendReadDataRequest(const CDate& tsDataRequested);
         bool sendReadDataRequestIfNeeded(void);
         static void recvReadDataRequest(CEventServer& event);
@@ -434 +435 @@
 
          //! The terminal filter which send data from server to client
-         std::shared_ptr<CServerToClientFilter> serverToClientFilter_;
+         std::shared_ptr<CServerToClientStoreFilter> serverToClientStoreFilter_;
 
 

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

@@ -1010 +1010 @@
      int size = this->enabledFields.size();
      for (int i = 0; i < size; ++i)
-       this->enabledFields[i]->sendReadDataRequest(CContext::getCurrent()->getCalendar()->getCurrentDate(), getContextClient());
+       this->enabledFields[i]->sendReadDataRequest(CContext::getCurrent()->getCalendar()->getCurrentDate());
    }
    CATCH_DUMP_ATTR

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

@@ -151 +151 @@
         for (int i = 0; i < size; ++i)
         {
-          if (i < domains.size()) {
+          if (i < domains.size())
+          {
             grid->axis_domain_order(i) = 2;
-
+            grid->order_.push_back(2) ;
           }
-          else if ((scalars.size() < (size-nb)) < size) {
+          else if ((scalars.size() < (size-nb)) < size)
+          {
             grid->axis_domain_order(i) = 1;
+            grid->order_.push_back(1) ;
           }
           else
+          {
             grid->axis_domain_order(i) = 0;
+            grid->order_.push_back(0) ;
+          }
           ++nb;
         }
@@ -167 +173 @@
         grid->axis_domain_order.resize(axisDomainOrder.numElements());
         grid->axis_domain_order = axisDomainOrder;
+        grid->order_.clear() ;
+        for(int i=0; i<axisDomainOrder.numElements();i++) grid->order_.push_back(axisDomainOrder(i)) ;
+
       }
 
@@ -2663 +2672 @@
     
     vector<CScattererConnector*> clientToServerConnectors ;
+    vector<CGathererConnector*>  clientFromServerConnectors ;
     for(int i=0 ; i<elements.size() ; i++)
     {
@@ -2671 +2681 @@
          domain->distributeToServer(client, gridRemoteConnector.getDistributedGlobalIndex(i)) ;
          clientToServerConnectors.push_back(domain->getClientToServerConnector(client)) ;
+         clientFromServerConnectors.push_back(domain->getClientFromServerConnector(client)) ;
       }
       else if (elements[i].type==TYPE_AXIS)
@@ -2678 +2689 @@
         axis->distributeToServer(client, gridRemoteConnector.getDistributedGlobalIndex(i)) ;
         clientToServerConnectors.push_back(axis->getClientToServerConnector(client)) ;
+        clientFromServerConnectors.push_back(axis->getClientFromServerConnector(client)) ;
+
       }
       else if (elements[i].type==TYPE_SCALAR)
@@ -2685 +2698 @@
         scalar->distributeToServer(client, gridRemoteConnector.getDistributedGlobalIndex(i)) ;
         clientToServerConnectors.push_back(scalar->getClientToServerConnector(client)) ;
+        clientFromServerConnectors.push_back(scalar->getClientFromServerConnector(client)) ;
       }
     }
@@ -2690 +2704 @@
     // compute the grid clientToServerConnector to send flux from client to servers
     clientToServerConnector_[client] = new CGridScattererConnector(clientToServerConnectors) ;
+    clientFromServerConnector_[client] = new CGridGathererConnector(clientFromServerConnectors) ;
+
 
   }
@@ -3309 +3325 @@
   }
 
+  void CGrid::computeServerToClientConnector(void)
+  {
+    vector<CScattererConnector*> connectors ;
+    for(auto& element : getElements())
+    {
+      if (element.type==TYPE_DOMAIN) connectors.push_back(element.domain->getServerToClientConnector()) ;
+      else if (element.type==TYPE_AXIS) connectors.push_back(element.axis->getServerToClientConnector()) ; 
+      else if (element.type==TYPE_SCALAR) connectors.push_back(element.scalar->getServerToClientConnector()) ; 
+    }
+    serverToClientConnector_ = new CGridScattererConnector(connectors) ;
+  }
+
   void CGrid::computeClientFromClientConnector(void)
   {
@@ -3320 +3348 @@
     clientFromClientConnector_ = new CGridGathererConnector(connectors) ;
   }
+
+  
 } // namespace xios

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

@@ -548 +548 @@
         void distributeGridToFileServer(CContextClient* client);
       
+            
+      private:
+        CGridLocalConnector* workflowToFullConnector_ = nullptr;
+      public:
+        void computeWorkflowToFullConnector(void) ;
+        CGridLocalConnector* getWorkflowToFullConnector(void) { if (workflowToFullConnector_==nullptr) computeWorkflowToFullConnector() ; return workflowToFullConnector_;}
+
+      private:
+        CGridLocalConnector* fullToWorkflowConnector_ = nullptr;
+      public:
+        void computeFullToWorkflowConnector(void) ;
+        CGridLocalConnector* getFullToWorkflowConnector(void) { if (fullToWorkflowConnector_==nullptr) computeFullToWorkflowConnector() ; return fullToWorkflowConnector_;}
+
+    
+
+      private:
+         CGridGathererConnector* clientFromClientConnector_ = nullptr ;
+      public:
+         CGridGathererConnector* getClientFromClientConnector(void) { if (clientFromClientConnector_==nullptr) computeClientFromClientConnector() ; return clientFromClientConnector_;}
+         void computeClientFromClientConnector(void) ;
+
+      private:
+         map<CContextClient*, CGridScattererConnector*> clientToClientConnector_ ;
+      public:
+         CGridScattererConnector* getClientToClientConnector(CContextClient* client) { return clientToClientConnector_[client] ;} // make some test to see if connector exits for the given client
+  
+
+      private:
+         map<CContextClient*,CGridGathererConnector*> clientFromServerConnector_  ;
+      public:
+         CGridGathererConnector* getClientFromServerConnector(CContextClient* client) { return clientFromServerConnector_[client];}
+         void computeClientFromServerConnector(void) ;
+
+      private:
+         CGridScattererConnector* serverToClientConnector_=nullptr ;
+      public:
+         CGridScattererConnector* getServerToClientConnector(void) { if (serverToClientConnector_==nullptr) computeServerToClientConnector() ; return serverToClientConnector_;}
+         void computeServerToClientConnector(void) ;
       private:
          map<CContextClient*, CGridScattererConnector*> clientToServerConnector_ ;
@@ -558 +596 @@
          CGridGathererConnector* getServerFromClientConnector(void) { if (serverFromClientConnector_==nullptr) computeServerFromClientConnector() ; return serverFromClientConnector_;}
          void computeServerFromClientConnector(void) ;
-         
-      private:
-        CGridLocalConnector* workflowToFullConnector_ = nullptr;
-      public:
-        void computeWorkflowToFullConnector(void) ;
-        CGridLocalConnector* getWorkflowToFullConnector(void) { if (workflowToFullConnector_==nullptr) computeWorkflowToFullConnector() ; return workflowToFullConnector_;}
-
-      private:
-        CGridLocalConnector* fullToWorkflowConnector_ = nullptr;
-      public:
-        void computeFullToWorkflowConnector(void) ;
-        CGridLocalConnector* getFullToWorkflowConnector(void) { if (fullToWorkflowConnector_==nullptr) computeFullToWorkflowConnector() ; return fullToWorkflowConnector_;}
-
-      private:
-         CGridGathererConnector* clientFromClientConnector_ = nullptr ;
-      public:
-         CGridGathererConnector* getClientFromClientConnector(void) { if (clientFromClientConnector_==nullptr) computeClientFromClientConnector() ; return clientFromClientConnector_;}
-         void computeClientFromClientConnector(void) ;
-
-      private:
-         map<CContextClient*, CGridScattererConnector*> clientToClientConnector_ ;
-      public:
-         CGridScattererConnector* getClientToClientConnector(CContextClient* client) { return clientToClientConnector_[client] ;} // make some test to see if connector exits for the given client
-  
-
-      private:
-         CGridGathererConnector* clientFromServerConnector_ = nullptr ;
-      public:
-         CGridGathererConnector* getClientFromServerConnector(void) { if (clientFromServerConnector_==nullptr) computeClientFromServerConnector() ; return clientFromServerConnector_;}
-         void computeClientFromServerConnector(void) ;
 
    }; // class CGrid

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

@@ -296 +296 @@
     CContext* context = CContext::getCurrent();
     CDistributedElement scatteredElement(1,globalIndex) ;
-    clientToServerConnector_[client] = new CScattererConnector(localElement_->getView(CElementView::FULL), scatteredElement.getView(CElementView::FULL), context->getIntraComm()) ;
+    clientToServerConnector_[client] = new CScattererConnector(localElement_->getView(CElementView::FULL), scatteredElement.getView(CElementView::FULL), 
+                                                               context->getIntraComm()) ;
     clientToServerConnector_[client] ->computeConnector() ;
+
+// need to be completed    
+
   }
 

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

@@ -182 +182 @@
         CGathererConnector* getServerFromClientConnector(void) { return serverFromClientConnector_ ;}
 
-
-
-
+       private:
+         CScattererConnector*  serverToClientConnector_ = nullptr ;
+       public: 
+         CScattererConnector* getServerToClientConnector(void) { return serverToClientConnector_ ;} 
+
+       private:
+          map<CContextClient*,CGathererConnector*>  clientFromServerConnector_  ;
+       public: 
+         CGathererConnector* getClientFromServerConnector(CContextClient* client) { return clientFromServerConnector_[client] ;} 
 
       private:

XIOS/dev/dev_ym/XIOS_COUPLING/src/test/generic_testcase.f90

@@ -186 +186 @@
     INTEGER :: i,j,k,xy,x,y,z,w
     DOUBLE PRECISION :: scale,dist
-    LOGICAL :: ierr,ok      
+    LOGICAL :: ok
+    INTEGER :: ierr      
 
     LOGICAL :: ok_field2D, ok_field3D, ok_pressure, ok_field2D_sub, ok_field3D_sub,ok_field3D_recv, ok_field3D_send