Ignore:
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/node
Files:
12 edited

Legend:

Unmodified
Added
Removed
  • XIOS/dev/dev_ym/XIOS_COUPLING/src/node/axis.cpp

    r1930 r1934  
    17061706    clientToServerConnector_[client]->transfer(maskIn,client,event3,message3) ;  
    17071707 
     1708    clientFromServerConnector_[client] = new CGathererConnector(clientToServerElement.getView(CElementView::FULL), localElement_->getView(CElementView::WORKFLOW)); 
     1709    clientFromServerConnector_[client]->computeConnector() ; 
    17081710 
    17091711 
     
    17701772      serverFromClientConnector_ = new CGathererConnector(elementFrom_->getView(CElementView::FULL), localElement_->getView(CElementView::WORKFLOW)) ; 
    17711773      serverFromClientConnector_->computeConnector() ; 
     1774 
     1775      serverToClientConnector_ = new CScattererConnector(localElement_->getView(CElementView::WORKFLOW), elementFrom_->getView(CElementView::FULL), 
     1776                                                         context->getIntraComm()) ; 
     1777      serverToClientConnector_->computeConnector() ; 
     1778 
    17721779    } 
    17731780  } 
  • XIOS/dev/dev_ym/XIOS_COUPLING/src/node/axis.hpp

    r1930 r1934  
    277277        CGathererConnector* getServerFromClientConnector(void) { return serverFromClientConnector_ ;} 
    278278 
    279  
     279       private: 
     280         CScattererConnector*  serverToClientConnector_ = nullptr ; 
     281       public:  
     282         CScattererConnector* getServerToClientConnector(void) { return serverToClientConnector_ ;}  
     283 
     284       private: 
     285          map<CContextClient*,CGathererConnector*>  clientFromServerConnector_  ; 
     286       public:  
     287         CGathererConnector* getClientFromServerConnector(CContextClient* client) { return clientFromServerConnector_[client] ;}  
    280288 
    281289         DECLARE_REF_FUNC(Axis,axis) 
  • XIOS/dev/dev_ym/XIOS_COUPLING/src/node/context.cpp

    r1883 r1934  
    12461246      for(auto field : fileInField)  
    12471247      { 
    1248         field->connectToServerInput(garbageCollector) ; // connect tFhe field to server filter 
    1249         field->computeGridIndexToFileServer() ; // compute grid index for transfer to the server context 
     1248        field->connectToServerInput(garbageCollector) ; // connect the field to server filter 
     1249    // obsolete    field->computeGridIndexToFileServer() ; // compute grid index for transfer to the server context 
    12501250        field->sendFieldToInputFileServer() ; 
     1251        fileInFields_.push_back(field) ; 
    12511252      } 
    12521253    } 
  • XIOS/dev/dev_ym/XIOS_COUPLING/src/node/domain.cpp

    r1930 r1934  
    23542354    clientToServerConnector_[client]->transfer(maskIn,client,event3,message3) ;  
    23552355     
     2356    clientFromServerConnector_[client] = new CGathererConnector(clientToServerElement.getView(CElementView::FULL), localElement_->getView(CElementView::WORKFLOW)); 
     2357    clientFromServerConnector_[client]->computeConnector() ; 
     2358 
    23562359  } 
    23572360  CATCH 
     
    24212424      serverFromClientConnector_ = new CGathererConnector(elementFrom_->getView(CElementView::FULL), localElement_->getView(CElementView::WORKFLOW)) ; 
    24222425      serverFromClientConnector_->computeConnector() ; 
     2426       
     2427      serverToClientConnector_ = new CScattererConnector(localElement_->getView(CElementView::WORKFLOW), elementFrom_->getView(CElementView::FULL), 
     2428                                                         context->getIntraComm()) ; 
     2429      serverToClientConnector_->computeConnector() ; 
     2430  
    24232431    } 
    24242432  } 
  • XIOS/dev/dev_ym/XIOS_COUPLING/src/node/domain.hpp

    r1930 r1934  
    349349       public:  
    350350         CScattererConnector* getClientToServerConnector(CContextClient* client) { return clientToServerConnector_[client] ;} 
     351        
    351352       private: 
    352353         CGathererConnector*  gathererConnector_ ; 
     
    354355         CDistributedElement* elementFrom_ ; 
    355356       public: 
    356         CGathererConnector* getServerFromClientConnector(void) { return serverFromClientConnector_ ;} 
    357  
     357         CGathererConnector* getServerFromClientConnector(void) { return serverFromClientConnector_ ;} 
     358 
     359       private: 
     360         CScattererConnector*  serverToClientConnector_ = nullptr ; 
     361       public:  
     362         CScattererConnector* getServerToClientConnector(void) { return serverToClientConnector_ ;}  
     363 
     364       private: 
     365         map<CContextClient*,CGathererConnector*>  clientFromServerConnector_  ; 
     366       public:  
     367         CGathererConnector* getClientFromServerConnector(CContextClient* client) { return clientFromServerConnector_[client] ;}         
    358368          
    359369 
  • XIOS/dev/dev_ym/XIOS_COUPLING/src/node/field.cpp

    r1930 r1934  
    150150  CATCH 
    151151 
    152 /* obsolete old interface 
    153   void CField::sendUpdateData(Time timeStamp, const CArray<double,1>& data, CContextClient* client) 
    154   TRY 
    155   { 
    156     CTimer::get("Field : send data").resume(); 
    157     int receiverSize = client->serverSize; 
    158  
    159     CEventClient event(getType(), EVENT_ID_UPDATE_DATA); 
    160  
    161     map<int, CArray<int,1> >::iterator it; 
    162     list<CMessage> list_msg; 
    163     list<CArray<double,1> > list_data; 
    164  
    165     if (!grid_->doGridHaveDataDistributed(client)) 
    166     { 
    167        if (client->isServerLeader()) 
    168        { 
    169           for (it = grid_->storeIndex_toSrv_[client].begin(); it != grid_->storeIndex_toSrv_[client].end(); it++) 
    170           { 
    171             int rank = it->first; 
    172             CArray<int,1>& index = it->second; 
    173  
    174             list_msg.push_back(CMessage()); 
    175             list_data.push_back(CArray<double,1>(index.numElements())); 
    176  
    177             CArray<double,1>& data_tmp = list_data.back(); 
    178             for (int n = 0; n < data_tmp.numElements(); n++) data_tmp(n) = data(index(n)); 
    179  
    180             list_msg.back() << getId() << timeStamp << data_tmp; 
    181             event.push(rank, 1, list_msg.back()); 
    182           } 
    183           client->sendEvent(event); 
    184         } 
    185       else client->sendEvent(event); 
    186     } 
    187     else 
    188     { 
    189       for (it = grid_->storeIndex_toSrv_[client].begin(); it != grid_->storeIndex_toSrv_[client].end(); it++) 
    190       { 
    191         int rank = it->first; 
    192         CArray<int,1>& index = it->second; 
    193  
    194         list_msg.push_back(CMessage()); 
    195         list_data.push_back(CArray<double,1>(index.numElements())); 
    196  
    197         CArray<double,1>& data_tmp = list_data.back(); 
    198         for (int n = 0; n < data_tmp.numElements(); n++) data_tmp(n) = data(index(n)); 
    199  
    200         list_msg.back() << getId() << timeStamp << data_tmp; 
    201         event.push(rank, grid_->nbSenders_[receiverSize][rank], list_msg.back()); 
    202       } 
    203       client->sendEvent(event); 
    204     } 
    205  
    206     CTimer::get("Field : send data").suspend(); 
    207   } 
    208   CATCH_DUMP_ATTR 
    209 */ 
    210  
    211152  void CField::sendUpdateData(Time timeStamp, const CArray<double,1>& data, CContextClient* client) 
    212153  TRY 
     
    222163  CATCH_DUMP_ATTR 
    223164 
    224   /* old version obsolete 
    225   void CField::recvUpdateData(CEventServer& event) 
    226   TRY 
    227   { 
    228     std::map<int,CBufferIn*> rankBuffers; 
    229  
    230     list<CEventServer::SSubEvent>::iterator it; 
    231     string fieldId; 
    232     CTimer::get("Field : recv data").resume(); 
    233     for (it = event.subEvents.begin(); it != event.subEvents.end(); ++it) 
    234     { 
    235       int rank = it->rank; 
    236       CBufferIn* buffer = it->buffer; 
    237       *buffer >> fieldId; 
    238       rankBuffers[rank] = buffer; 
    239     } 
    240     get(fieldId)->recvUpdateData(rankBuffers); 
    241     CTimer::get("Field : recv data").suspend(); 
    242   } 
    243   CATCH 
    244 */ 
    245  
     165   
    246166  void CField::recvUpdateData(CEventServer& event) 
    247167  TRY 
     
    260180  } 
    261181  CATCH 
    262 /* 
    263   void  CField::recvUpdateDataFromClient(CEventServer& event) 
    264   TRY 
    265   { 
    266     Time timeStamp ; 
    267     for (auto& subEvent : event.subEvents) (*subEvent.buffer) >> timeStamp  ; 
    268  
    269     CArray<double,1> recvData ; 
    270     getGrid()->getServerFromClientConnector()->transfer(event,recvData) ; 
    271     this->setData(recvData); 
    272   } 
    273   CATCH 
    274 */ 
    275  
    276 /*   
    277   void CField::recvUpdateDataFromCoupler(CEventServer& event) 
    278   TRY 
    279   { 
    280     CContext* context = CContext::getCurrent(); 
    281     Time timeStamp ; 
    282     if (wasDataAlreadyReceivedFromServer) 
    283     {   
    284       lastDataReceivedFromServer = lastDataReceivedFromServer + freq_op; 
    285     } 
    286     else 
    287     { 
    288       // unlikely to input from file server where data are received at ts=0 
    289       // for coupling, it would be after the first freq_op, because for now we don't have 
    290       // restart mecanism to send the value at ts=0. It must be changed in future 
    291       lastDataReceivedFromServer = context->getCalendar()->getInitDate(); 
    292       wasDataAlreadyReceivedFromServer = true; 
    293     } 
    294  
    295     CArray<double,1> recvData ; 
    296     getGrid()->getServerFromClientConnector()->transfer(event,recvData) ; 
    297     clientSourceFilter->streamData(lastDataReceivedFromServer, recvData); 
    298  
    299   } 
    300   CATCH_DUMP_ATTR 
    301 */ 
    302  
    303  
    304  
    305  
    306   /* old interface to be removed.... */ 
    307   void  CField::recvUpdateDataFromClient(std::map<int,CBufferIn*>& rankBuffers) 
    308   TRY 
    309   { 
    310     // ym to remove latter 
    311     /* 
    312     CContext* context = CContext::getCurrent(); 
    313     Time timeStamp ; 
    314     size_t sizeData = 0; 
    315     if (0 == recvDataSrv.numElements()) 
    316     {             
    317       CArray<int,1>& storeClient = grid_->getStoreIndex_client(); 
    318  
    319       // Gather all data from different clients       
    320       recvDataSrv.resize(storeClient.numElements()); 
    321       recvFoperationSrv = std::shared_ptr<func::CFunctor>(new func::CInstant(recvDataSrv)); 
    322     } 
    323  
    324     CArray<double,1> recv_data_tmp(recvDataSrv.numElements());     
    325     const CDate& currDate = context->getCalendar()->getCurrentDate(); 
    326     CDuration offsetAllButMonth (freq_offset.getValue().year, 0 , freq_offset.getValue().day, 
    327                                  freq_offset.getValue().hour, freq_offset.getValue().minute, 
    328                                  freq_offset.getValue().second, freq_offset.getValue().timestep); 
    329     const CDate opeDate   = (last_operation_srv - offsetAllButMonth + context->getCalendar()->getTimeStep()) 
    330                              + freq_op + freq_operation_srv - freq_op - context->getCalendar()->getTimeStep() + offsetAllButMonth; 
    331  
    332     if (opeDate <= currDate) 
    333     { 
    334        
    335       auto& outLocalIndexStoreOnClient = grid_-> getOutLocalIndexStoreOnClient() ; 
    336       for (auto it = outLocalIndexStoreOnClient.begin(); it != outLocalIndexStoreOnClient.end(); ++it) 
    337       { 
    338         CArray<double,1> tmp; 
    339         CArray<size_t,1>& indexTmp = it->second; 
    340         *(rankBuffers[it->first]) >> timeStamp >> tmp; 
    341         for (int idx = 0; idx < indexTmp.numElements(); ++idx) recv_data_tmp(indexTmp(idx)) = tmp(idx); 
    342       } 
    343     } 
    344 */ 
    345     Time timeStamp ; 
    346     CArray<int,1>& storeClient = grid_->getStoreIndex_client(); // replace it with local size 
    347     CArray<double,1> recv_data_tmp(storeClient.numElements()); 
    348     auto& outLocalIndexStoreOnClient = grid_-> getOutLocalIndexStoreOnClient() ; 
    349     for (auto it = outLocalIndexStoreOnClient.begin(); it != outLocalIndexStoreOnClient.end(); ++it) 
    350     { 
    351       CArray<double,1> tmp; 
    352       CArray<size_t,1>& indexTmp = it->second; 
    353       *(rankBuffers[it->first]) >> timeStamp >> tmp; 
    354       for (int idx = 0; idx < indexTmp.numElements(); ++idx) recv_data_tmp(indexTmp(idx)) = tmp(idx); 
    355     } 
    356  
    357     this->setData(recv_data_tmp); 
    358 /* 
    359     // delete incomming flux for server only 
    360     recvFoperationSrv.reset() ; 
    361     recvDataSrv.reset() ; 
    362 */ 
    363   } 
    364   CATCH_DUMP_ATTR 
    365  
    366   /* ym : old interface : to be removed... 
    367   void CField::writeUpdateData(const CArray<double,1>& data) 
    368   TRY 
    369   { 
    370     CContext* context = CContext::getCurrent(); 
    371  
    372     const CDate& currDate = context->getCalendar()->getCurrentDate(); 
    373     CDuration offsetAllButMonth (freq_offset.getValue().year, 0 , freq_offset.getValue().day, 
    374                                    freq_offset.getValue().hour, freq_offset.getValue().minute, 
    375                                    freq_offset.getValue().second, freq_offset.getValue().timestep); 
    376     const CDate opeDate   = (last_operation_srv - offsetAllButMonth + context->getCalendar()->getTimeStep()) 
    377                               + freq_op + freq_operation_srv - freq_op - context->getCalendar()->getTimeStep() + offsetAllButMonth; 
    378     const CDate writeDate = last_Write_srv + freq_write_srv; 
    379  
    380     if (opeDate <= currDate) 
    381     { 
    382       (*recvFoperationSrv)(data); 
    383       last_operation_srv = currDate; 
    384     } 
    385  
    386     if (writeDate < (currDate + freq_operation_srv)) 
    387     { 
    388       recvFoperationSrv->final(); 
    389       last_Write_srv = writeDate; 
    390       grid_->computeWrittenIndex(); 
    391       writeField(); 
    392       lastlast_Write_srv = last_Write_srv; 
    393     } 
    394   } 
    395   CATCH_DUMP_ATTR 
    396   */ 
    397  
     182 
     183   
    398184  void CField::writeUpdateData(const CArray<double,1>& data) 
    399185  TRY 
     
    434220    \param [in] tsDataRequested timestamp when the call is made 
    435221  */ 
    436   bool CField::sendReadDataRequest(const CDate& tsDataRequested, CContextClient* client) 
    437   TRY 
    438   { 
    439     CContext* context = CContext::getCurrent(); 
    440  
    441     lastDataRequestedFromServer = tsDataRequested; 
    442  
    443     // No need to send the request if we are sure that we are already at EOF 
    444     if (!isEOF || context->getCalendar()->getCurrentDate() <= dateEOF) 
    445     { 
    446       CEventClient event(getType(), EVENT_ID_READ_DATA); 
    447       if (client->isServerLeader()) 
    448       { 
    449         CMessage msg; 
    450         msg << getId(); 
    451         const std::list<int>& ranks = client->getRanksServerLeader(); 
    452         for (std::list<int>::const_iterator itRank = ranks.begin(), itRankEnd = ranks.end(); itRank != itRankEnd; ++itRank) 
    453           event.push(*itRank, 1, msg); 
    454         client->sendEvent(event); 
    455       } 
    456       else client->sendEvent(event); 
    457     } 
    458     else serverSourceFilter->signalEndOfStream(tsDataRequested); 
    459  
    460     wasDataRequestedFromServer = true; 
    461  
    462     return !isEOF; 
    463   } 
    464   CATCH_DUMP_ATTR 
    465  
     222  bool CField::sendReadDataRequest(const CDate& tsDataRequested) 
     223  TRY 
     224  { 
     225    return clientFromServerSourceFilter_->sendReadDataRequest(tsDataRequested) ; 
     226  } 
     227  CATCH_DUMP_ATTR 
     228  
     229   
    466230  /*! 
    467231  Send request new data read from file if need be, that is the current data is out-of-date. 
     
    471235  TRY 
    472236  { 
    473     const CDate& currentDate = CContext::getCurrent()->getCalendar()->getCurrentDate(); 
    474  
    475     bool dataRequested = false; 
    476  
    477     while (currentDate >= lastDataRequestedFromServer) 
    478     { 
    479       info(20) << "currentDate : " << currentDate << endl ; 
    480       info(20) << "lastDataRequestedFromServer : " << lastDataRequestedFromServer << endl ; 
    481       info(20) << "fileIn_->output_freq.getValue() : " << fileIn_->output_freq.getValue() << endl ; 
    482       info(20) << "lastDataRequestedFromServer + fileIn_->output_freq.getValue() : " << lastDataRequestedFromServer + fileIn_->output_freq << endl ; 
    483  
    484       dataRequested |= sendReadDataRequest(lastDataRequestedFromServer + fileIn_->output_freq, fileIn_->getContextClient()); 
    485     } 
    486  
    487     return dataRequested; 
    488   } 
    489   CATCH_DUMP_ATTR 
     237    return clientFromServerSourceFilter_->sendReadDataRequestIfNeeded() ; 
     238  } 
     239  CATCH_DUMP_ATTR 
     240 
    490241 
    491242  void CField::recvReadDataRequest(CEventServer& event) 
     
    512263  CATCH_DUMP_ATTR   
    513264 
    514 /* old interface -> to remove 
    515   void CField::recvReadDataRequest(CContextServer* server) 
    516   TRY 
    517   { 
    518          
    519     CArray<double,1> data ; 
    520     EReadField hasData = readField(data); 
    521     CDate date = CContext::getCurrent()->getCalendar()->getCurrentDate(); 
    522     if (hasData == RF_DATA) fileServerReaderFilter_->streamData(date,data) ; 
    523     else fileServerReaderFilter_->signalEndOfStream(date) ; 
    524  
    525   } 
    526   CATCH_DUMP_ATTR 
    527 */ 
    528265 
    529266  void CField::sendUpdateDataServerToClient(bool isEOF, const CArray<double,1>& data, CContextClient* client) 
     
    583320  CATCH_DUMP_ATTR 
    584321 
    585   /*! 
    586     Read field from a file. 
    587     A field is read with the distribution of data on the server side 
    588     \return State of field can be read from a file 
    589   */ 
    590   // obsolete to remove 
    591   /* 
    592   CField::EReadField CField::readField(CArray<double,1>& data) 
    593   TRY 
    594   { 
    595     CContext* context = CContext::getCurrent(); 
    596     grid_->computeWrittenIndex(); 
    597     getRelFile()->initRead(); 
    598     EReadField readState = RF_DATA; 
    599  
    600     if (!getRelFile()->isEmptyZone()) 
    601     {       
    602       if (grid_->doGridHaveDataToWrite() || getRelFile()->type == CFile::type_attr::one_file)       
    603       { 
    604           CArray<int,1>& storeClient = grid_->getStoreIndex_client();           
    605           data.resize(storeClient.numElements());           
    606          
    607         getRelFile()->checkReadFile(); 
    608  
    609         if (!nstepMax) 
    610         { 
    611           nstepMax = getRelFile()->getDataInput()->getFieldNbRecords(CField::get(this)); 
    612         } 
    613  
    614         this->incrementNStep(); 
    615  
    616         if (getNStep() > nstepMax && (getRelFile()->cyclic.isEmpty() || !getRelFile()->cyclic) ) 
    617           readState = RF_EOF; 
    618  
    619         if (RF_EOF != readState) 
    620           getRelFile()->getDataInput()->readFieldData(CField::get(this),data); 
    621       } 
    622     } 
    623     else 
    624     { 
    625       this->incrementNStep(); 
    626       if (getNStep() > nstepMax && (getRelFile()->cyclic.isEmpty() || !getRelFile()->cyclic) ) 
    627         readState = RF_EOF; 
    628       else 
    629         readState = RF_NODATA; 
    630  
    631       if (!nstepMaxRead) // This can be a bug if we try to read field from zero time record 
    632         readState = RF_NODATA; 
    633     } 
    634  
    635     if (!nstepMaxRead) 
    636     { 
    637        MPI_Allreduce(MPI_IN_PLACE, &nstepMax, 1, MPI_INT, MPI_MAX, context->intraComm_); 
    638        nstepMaxRead = true; 
    639     } 
    640  
    641     return readState; 
    642   } 
    643   CATCH_DUMP_ATTR 
    644   */ 
    645  
     322   
    646323  /* 
    647324    Receive read data from server. 
     
    655332    string fieldId; 
    656333    for (auto& subEvent : event.subEvents) (*subEvent.buffer) >> fieldId  ; 
    657     get(fieldId)->recvReadDataReady(event); 
     334    get(fieldId)->receiveReadDataReady(event); 
    658335  } 
    659336  CATCH 
    660337 
     338  void CField::receiveReadDataReady(CEventServer& event) 
     339  TRY 
     340  { 
     341    clientFromServerSourceFilter_->streamData(event) ;     
     342  } 
     343  CATCH_DUMP_ATTR 
    661344 
    662345  /* old interface to be removed ..*/ 
     
    742425  } 
    743426  CATCH_DUMP_ATTR 
    744  
    745  
    746  
    747   void CField::receiveReadDataReady(CEventServer& event) 
    748   TRY 
    749   { 
    750     clientFromServerSourceFilter_->streamData(event) ;     
    751   } 
    752   CATCH_DUMP_ATTR 
    753  
    754427 
    755428 
     
    786459  CATCH_DUMP_ATTR 
    787460 
    788  
    789461  void CField::checkForLateDataFromServer(void) 
    790462  TRY 
    791463  { 
    792     CContext* context = CContext::getCurrent(); 
    793     // Check if data previously requested has been received as expected 
    794     if (wasDataRequestedFromServer && !clientFromServerSourceFilter_->isEOF()) 
    795     { 
    796       CTimer timer("CField::checkForLateDataFromServer"); 
    797       timer.resume(); 
    798       traceOff() ; 
    799       timer.suspend(); 
    800        
    801       bool isDataLate; 
    802       do 
    803       { 
    804         isDataLate=clientFromServerSourceFilter_->isDataLate(); 
    805         if (isDataLate) 
    806         { 
    807           timer.resume(); 
    808  
    809 //ym          context->checkBuffersAndListen(); 
    810 //ym            context->eventLoop(); 
    811           context->globalEventLoop(); 
    812  
    813           timer.suspend(); 
    814         } 
    815       } 
    816       while (isDataLate && timer.getCumulatedTime() < CXios::recvFieldTimeout); 
    817       timer.resume(); 
    818       traceOn() ; 
    819       timer.suspend() ; 
    820  
    821  
    822       if (isDataLate) 
    823         ERROR("void CField::checkForLateDataFromServer(void)", 
    824               << "Late data at timestep = " << context->getCalendar()->getCurrentDate()); 
    825     } 
    826   } 
    827   CATCH_DUMP_ATTR 
    828  
     464    clientFromServerSourceFilter_->checkForLateData() ; 
     465  } 
     466  CATCH_DUMP_ATTR  
     467   
     468   
    829469  void CField::triggerLateField(void) 
    830470  TRY 
     
    15081148  void CField::connectToServerInput(CGarbageCollector& gc) 
    15091149  { 
    1510     const bool detectMissingValues = (!detect_missing_value.isEmpty() && !default_value.isEmpty() && detect_missing_value == true); 
    1511     const double defaultValue  = detectMissingValues ? default_value : (!default_value.isEmpty() ? default_value : 0.0); 
    1512  
    15131150    checkTimeAttributes(); 
    1514     serverSourceFilter = std::shared_ptr<CSourceFilter>(new CSourceFilter(gc, grid_, true, false, freq_offset, true, 
    1515                                                                           detectMissingValues, defaultValue)); 
    1516     //serverSourceFilter = std::shared_ptr<CSourceFilter>(new CSourceFilter(gc,  grid_, false, false)); 
    1517     serverSourceFilter -> connectOutput(inputFilter,0) ; 
     1151    clientFromServerSourceFilter_ = std::shared_ptr<CClientFromServerSourceFilter>(new CClientFromServerSourceFilter(gc,this)) ; 
     1152    clientFromServerSourceFilter_ -> connectOutput(inputFilter,0) ; 
    15181153  }  
    15191154 
     
    15611196  { 
    15621197    fileReaderSourceFilter_ = std::shared_ptr<CFileReaderSourceFilter>(new CFileReaderSourceFilter(gc, this)); 
    1563     instantDataFilter->connectOutput(inputFilter, 0); 
     1198    fileReaderSourceFilter_->connectOutput(inputFilter, 0); 
    15641199  } 
    15651200 
     
    15781213  void CField::connectToServerToClient(CGarbageCollector& gc) 
    15791214  { 
    1580     serverToClientFilter_ = std::shared_ptr<CServerToClientFilter>(new CServerToClientFilter(gc, this, client)); 
    1581     instantDataFilter->connectOutput(serverToClientFilter_, 0); 
     1215    serverToClientStoreFilter_ = std::shared_ptr<CServerToClientStoreFilter>(new CServerToClientStoreFilter(gc, this, client)); 
     1216    instantDataFilter->connectOutput(serverToClientStoreFilter_, 0); 
    15821217  } 
    15831218 
     
    22901925  } 
    22911926 
     1927  void CField::sendFieldToInputFileServer(void) 
     1928  { 
     1929    CContext::getCurrent()->sendContextToFileServer(client); 
     1930    getRelFile()->sendFileToFileServer(client); 
     1931    grid_->sendGridToFileServer(client); 
     1932    read_access=true ; // not the best solution, but on server side, the field must be a starting point of the workflow 
     1933                       // must be replace by a better solution when implementing filters for reading and send to client 
     1934                       // on server side 
     1935    this->sendAllAttributesToServer(client); 
     1936    this->sendAddAllVariables(client); 
     1937  } 
     1938 
    22921939  void CField::sendFieldToCouplerOut(void) 
    22931940  { 
     
    23411988 
    23421989 
    2343  
    2344  
    2345  
    2346   void CField::sendFieldToInputFileServer(void) 
    2347   { 
    2348     CContext::getCurrent()->sendContextToFileServer(client); 
    2349     getRelFile()->sendFileToFileServer(client); 
    2350     grid_->sendGridToFileServer(client); 
    2351     read_access=true ; // not the best solution, but on server side, the field must be a starting point of the workflow 
    2352                        // must be replace by a better solution when implementing filters for reading and send to client 
    2353                        // on server side 
    2354     this->sendAllAttributesToServer(client); 
    2355     this->sendAddAllVariables(client); 
    2356   } 
    2357  
    23581990  void CField::sendAddAllVariables(CContextClient* client) 
    23591991  TRY 
  • XIOS/dev/dev_ym/XIOS_COUPLING/src/node/field.hpp

    r1930 r1934  
    2323#include "client_from_server_source_filter.hpp" 
    2424#include "client_to_model_store_filter.hpp" 
     25#include "server_to_client_store_filter.hpp" 
    2526 
    2627 
     
    203204         
    204205        void writeField(const CArray<double,1>& data); 
    205         bool sendReadDataRequest(const CDate& tsDataRequested, CContextClient* client); 
     206        bool sendReadDataRequest(const CDate& tsDataRequested); 
    206207        bool sendReadDataRequestIfNeeded(void); 
    207208        static void recvReadDataRequest(CEventServer& event); 
     
    434435 
    435436         //! The terminal filter which send data from server to client 
    436          std::shared_ptr<CServerToClientFilter> serverToClientFilter_; 
     437         std::shared_ptr<CServerToClientStoreFilter> serverToClientStoreFilter_; 
    437438 
    438439 
  • XIOS/dev/dev_ym/XIOS_COUPLING/src/node/file.cpp

    r1872 r1934  
    10101010     int size = this->enabledFields.size(); 
    10111011     for (int i = 0; i < size; ++i) 
    1012        this->enabledFields[i]->sendReadDataRequest(CContext::getCurrent()->getCalendar()->getCurrentDate(), getContextClient()); 
     1012       this->enabledFields[i]->sendReadDataRequest(CContext::getCurrent()->getCalendar()->getCurrentDate()); 
    10131013   } 
    10141014   CATCH_DUMP_ATTR 
  • XIOS/dev/dev_ym/XIOS_COUPLING/src/node/grid.cpp

    r1930 r1934  
    151151        for (int i = 0; i < size; ++i) 
    152152        { 
    153           if (i < domains.size()) { 
     153          if (i < domains.size()) 
     154          { 
    154155            grid->axis_domain_order(i) = 2; 
    155  
     156            grid->order_.push_back(2) ; 
    156157          } 
    157           else if ((scalars.size() < (size-nb)) < size) { 
     158          else if ((scalars.size() < (size-nb)) < size) 
     159          { 
    158160            grid->axis_domain_order(i) = 1; 
     161            grid->order_.push_back(1) ; 
    159162          } 
    160163          else 
     164          { 
    161165            grid->axis_domain_order(i) = 0; 
     166            grid->order_.push_back(0) ; 
     167          } 
    162168          ++nb; 
    163169        } 
     
    167173        grid->axis_domain_order.resize(axisDomainOrder.numElements()); 
    168174        grid->axis_domain_order = axisDomainOrder; 
     175        grid->order_.clear() ; 
     176        for(int i=0; i<axisDomainOrder.numElements();i++) grid->order_.push_back(axisDomainOrder(i)) ; 
     177 
    169178      } 
    170179 
     
    26632672     
    26642673    vector<CScattererConnector*> clientToServerConnectors ; 
     2674    vector<CGathererConnector*>  clientFromServerConnectors ; 
    26652675    for(int i=0 ; i<elements.size() ; i++) 
    26662676    { 
     
    26712681         domain->distributeToServer(client, gridRemoteConnector.getDistributedGlobalIndex(i)) ; 
    26722682         clientToServerConnectors.push_back(domain->getClientToServerConnector(client)) ; 
     2683         clientFromServerConnectors.push_back(domain->getClientFromServerConnector(client)) ; 
    26732684      } 
    26742685      else if (elements[i].type==TYPE_AXIS) 
     
    26782689        axis->distributeToServer(client, gridRemoteConnector.getDistributedGlobalIndex(i)) ; 
    26792690        clientToServerConnectors.push_back(axis->getClientToServerConnector(client)) ; 
     2691        clientFromServerConnectors.push_back(axis->getClientFromServerConnector(client)) ; 
     2692 
    26802693      } 
    26812694      else if (elements[i].type==TYPE_SCALAR) 
     
    26852698        scalar->distributeToServer(client, gridRemoteConnector.getDistributedGlobalIndex(i)) ; 
    26862699        clientToServerConnectors.push_back(scalar->getClientToServerConnector(client)) ; 
     2700        clientFromServerConnectors.push_back(scalar->getClientFromServerConnector(client)) ; 
    26872701      } 
    26882702    } 
     
    26902704    // compute the grid clientToServerConnector to send flux from client to servers 
    26912705    clientToServerConnector_[client] = new CGridScattererConnector(clientToServerConnectors) ; 
     2706    clientFromServerConnector_[client] = new CGridGathererConnector(clientFromServerConnectors) ; 
     2707 
    26922708 
    26932709  } 
     
    33093325  } 
    33103326 
     3327  void CGrid::computeServerToClientConnector(void) 
     3328  { 
     3329    vector<CScattererConnector*> connectors ; 
     3330    for(auto& element : getElements()) 
     3331    { 
     3332      if (element.type==TYPE_DOMAIN) connectors.push_back(element.domain->getServerToClientConnector()) ; 
     3333      else if (element.type==TYPE_AXIS) connectors.push_back(element.axis->getServerToClientConnector()) ;  
     3334      else if (element.type==TYPE_SCALAR) connectors.push_back(element.scalar->getServerToClientConnector()) ;  
     3335    } 
     3336    serverToClientConnector_ = new CGridScattererConnector(connectors) ; 
     3337  } 
     3338 
    33113339  void CGrid::computeClientFromClientConnector(void) 
    33123340  { 
     
    33203348    clientFromClientConnector_ = new CGridGathererConnector(connectors) ; 
    33213349  } 
     3350 
     3351   
    33223352} // namespace xios 
  • XIOS/dev/dev_ym/XIOS_COUPLING/src/node/grid.hpp

    r1930 r1934  
    548548        void distributeGridToFileServer(CContextClient* client); 
    549549       
     550             
     551      private: 
     552        CGridLocalConnector* workflowToFullConnector_ = nullptr; 
     553      public: 
     554        void computeWorkflowToFullConnector(void) ; 
     555        CGridLocalConnector* getWorkflowToFullConnector(void) { if (workflowToFullConnector_==nullptr) computeWorkflowToFullConnector() ; return workflowToFullConnector_;} 
     556 
     557      private: 
     558        CGridLocalConnector* fullToWorkflowConnector_ = nullptr; 
     559      public: 
     560        void computeFullToWorkflowConnector(void) ; 
     561        CGridLocalConnector* getFullToWorkflowConnector(void) { if (fullToWorkflowConnector_==nullptr) computeFullToWorkflowConnector() ; return fullToWorkflowConnector_;} 
     562 
     563     
     564 
     565      private: 
     566         CGridGathererConnector* clientFromClientConnector_ = nullptr ; 
     567      public: 
     568         CGridGathererConnector* getClientFromClientConnector(void) { if (clientFromClientConnector_==nullptr) computeClientFromClientConnector() ; return clientFromClientConnector_;} 
     569         void computeClientFromClientConnector(void) ; 
     570 
     571      private: 
     572         map<CContextClient*, CGridScattererConnector*> clientToClientConnector_ ; 
     573      public: 
     574         CGridScattererConnector* getClientToClientConnector(CContextClient* client) { return clientToClientConnector_[client] ;} // make some test to see if connector exits for the given client 
     575   
     576 
     577      private: 
     578         map<CContextClient*,CGridGathererConnector*> clientFromServerConnector_  ; 
     579      public: 
     580         CGridGathererConnector* getClientFromServerConnector(CContextClient* client) { return clientFromServerConnector_[client];} 
     581         void computeClientFromServerConnector(void) ; 
     582 
     583      private: 
     584         CGridScattererConnector* serverToClientConnector_=nullptr ; 
     585      public: 
     586         CGridScattererConnector* getServerToClientConnector(void) { if (serverToClientConnector_==nullptr) computeServerToClientConnector() ; return serverToClientConnector_;} 
     587         void computeServerToClientConnector(void) ; 
    550588      private: 
    551589         map<CContextClient*, CGridScattererConnector*> clientToServerConnector_ ; 
     
    558596         CGridGathererConnector* getServerFromClientConnector(void) { if (serverFromClientConnector_==nullptr) computeServerFromClientConnector() ; return serverFromClientConnector_;} 
    559597         void computeServerFromClientConnector(void) ; 
    560           
    561       private: 
    562         CGridLocalConnector* workflowToFullConnector_ = nullptr; 
    563       public: 
    564         void computeWorkflowToFullConnector(void) ; 
    565         CGridLocalConnector* getWorkflowToFullConnector(void) { if (workflowToFullConnector_==nullptr) computeWorkflowToFullConnector() ; return workflowToFullConnector_;} 
    566  
    567       private: 
    568         CGridLocalConnector* fullToWorkflowConnector_ = nullptr; 
    569       public: 
    570         void computeFullToWorkflowConnector(void) ; 
    571         CGridLocalConnector* getFullToWorkflowConnector(void) { if (fullToWorkflowConnector_==nullptr) computeFullToWorkflowConnector() ; return fullToWorkflowConnector_;} 
    572  
    573       private: 
    574          CGridGathererConnector* clientFromClientConnector_ = nullptr ; 
    575       public: 
    576          CGridGathererConnector* getClientFromClientConnector(void) { if (clientFromClientConnector_==nullptr) computeClientFromClientConnector() ; return clientFromClientConnector_;} 
    577          void computeClientFromClientConnector(void) ; 
    578  
    579       private: 
    580          map<CContextClient*, CGridScattererConnector*> clientToClientConnector_ ; 
    581       public: 
    582          CGridScattererConnector* getClientToClientConnector(CContextClient* client) { return clientToClientConnector_[client] ;} // make some test to see if connector exits for the given client 
    583    
    584  
    585       private: 
    586          CGridGathererConnector* clientFromServerConnector_ = nullptr ; 
    587       public: 
    588          CGridGathererConnector* getClientFromServerConnector(void) { if (clientFromServerConnector_==nullptr) computeClientFromServerConnector() ; return clientFromServerConnector_;} 
    589          void computeClientFromServerConnector(void) ; 
    590598 
    591599   }; // class CGrid 
  • XIOS/dev/dev_ym/XIOS_COUPLING/src/node/scalar.cpp

    r1930 r1934  
    296296    CContext* context = CContext::getCurrent(); 
    297297    CDistributedElement scatteredElement(1,globalIndex) ; 
    298     clientToServerConnector_[client] = new CScattererConnector(localElement_->getView(CElementView::FULL), scatteredElement.getView(CElementView::FULL), context->getIntraComm()) ; 
     298    clientToServerConnector_[client] = new CScattererConnector(localElement_->getView(CElementView::FULL), scatteredElement.getView(CElementView::FULL),  
     299                                                               context->getIntraComm()) ; 
    299300    clientToServerConnector_[client] ->computeConnector() ; 
     301 
     302// need to be completed     
     303 
    300304  } 
    301305 
  • XIOS/dev/dev_ym/XIOS_COUPLING/src/node/scalar.hpp

    r1930 r1934  
    182182        CGathererConnector* getServerFromClientConnector(void) { return serverFromClientConnector_ ;} 
    183183 
    184  
    185  
    186  
     184       private: 
     185         CScattererConnector*  serverToClientConnector_ = nullptr ; 
     186       public:  
     187         CScattererConnector* getServerToClientConnector(void) { return serverToClientConnector_ ;}  
     188 
     189       private: 
     190          map<CContextClient*,CGathererConnector*>  clientFromServerConnector_  ; 
     191       public:  
     192         CGathererConnector* getClientFromServerConnector(CContextClient* client) { return clientFromServerConnector_[client] ;}  
    187193 
    188194      private: 
Note: See TracChangeset for help on using the changeset viewer.