source: XIOS/trunk/src/node/field.cpp @ 988

Last change on this file since 988 was 988, checked in by mhnguyen, 6 years ago

Correcting a bug on reading non-distributed data in server mode.

+) Non distributed data read by all servers then transfered back to ALL clients

Test
+) On Curie
+) Work

  • Property copyright set to
    Software name : XIOS (Xml I/O Server)
    http://forge.ipsl.jussieu.fr/ioserver
    Creation date : January 2009
    Licence : CeCCIL version2
    see license file in root directory : Licence_CeCILL_V2-en.txt
    or http://www.cecill.info/licences/Licence_CeCILL_V2-en.html
    Holder : CEA/LSCE (Laboratoire des Sciences du CLimat et de l'Environnement)
    CNRS/IPSL (Institut Pierre Simon Laplace)
    Project Manager : Yann Meurdesoif
    yann.meurdesoif@cea.fr
  • Property svn:executable set to *
File size: 41.4 KB
Line 
1#include "field.hpp"
2
3#include "attribute_template.hpp"
4#include "object_template.hpp"
5#include "group_template.hpp"
6
7#include "node_type.hpp"
8#include "calendar_util.hpp"
9#include "message.hpp"
10#include "xios_spl.hpp"
11#include "type.hpp"
12#include "timer.hpp"
13#include "context_client.hpp"
14#include "context_server.hpp"
15#include <set>
16#include "garbage_collector.hpp"
17#include "source_filter.hpp"
18#include "store_filter.hpp"
19#include "file_writer_filter.hpp"
20#include "pass_through_filter.hpp"
21#include "filter_expr_node.hpp"
22#include "lex_parser.hpp"
23#include "temporal_filter.hpp"
24#include "spatial_transform_filter.hpp"
25
26namespace xios{
27
28   /// ////////////////////// Définitions ////////////////////// ///
29
30   CField::CField(void)
31      : CObjectTemplate<CField>(), CFieldAttributes()
32      , grid(), file()
33      , written(false)
34      , nstep(0), nstepMax(0)
35      , hasOutputFile(false)
36      , domAxisScalarIds_(vector<StdString>(3,"")), areAllReferenceSolved(false), isReferenceSolved(false)
37      , useCompressedOutput(false)
38      , isReadDataRequestPending(false)
39   { setVirtualVariableGroup(CVariableGroup::create(getId() + "_virtual_variable_group")); }
40
41   CField::CField(const StdString& id)
42      : CObjectTemplate<CField>(id), CFieldAttributes()
43      , grid(), file()
44      , written(false)
45      , nstep(0), nstepMax(0)
46      , hasOutputFile(false)
47      , domAxisScalarIds_(vector<StdString>(3,"")), areAllReferenceSolved(false), isReferenceSolved(false)
48      , useCompressedOutput(false)
49      , isReadDataRequestPending(false)
50   { setVirtualVariableGroup(CVariableGroup::create(getId() + "_virtual_variable_group")); }
51
52   CField::~CField(void)
53   {}
54
55  //----------------------------------------------------------------
56
57   void CField::setVirtualVariableGroup(CVariableGroup* newVVariableGroup)
58   {
59      this->vVariableGroup = newVVariableGroup;
60   }
61
62   CVariableGroup* CField::getVirtualVariableGroup(void) const
63   {
64      return this->vVariableGroup;
65   }
66
67   std::vector<CVariable*> CField::getAllVariables(void) const
68   {
69      return this->vVariableGroup->getAllChildren();
70   }
71
72   void CField::solveDescInheritance(bool apply, const CAttributeMap* const parent)
73   {
74      SuperClassAttribute::setAttributes(parent, apply);
75      this->getVirtualVariableGroup()->solveDescInheritance(apply, NULL);
76   }
77
78  //----------------------------------------------------------------
79
80  bool CField::dispatchEvent(CEventServer& event)
81  {
82    if (SuperClass::dispatchEvent(event)) return true;
83    else
84    {
85      switch(event.type)
86      {
87        case EVENT_ID_UPDATE_DATA :
88          recvUpdateData(event);
89          return true;
90          break;
91
92        case EVENT_ID_READ_DATA :
93          recvReadDataRequest(event);
94          return true;
95          break;
96
97        case EVENT_ID_READ_DATA_READY :
98          recvReadDataReady(event);
99          return true;
100          break;
101
102        case EVENT_ID_ADD_VARIABLE :
103          recvAddVariable(event);
104          return true;
105          break;
106
107        case EVENT_ID_ADD_VARIABLE_GROUP :
108          recvAddVariableGroup(event);
109          return true;
110          break;
111
112        default :
113          ERROR("bool CField::dispatchEvent(CEventServer& event)", << "Unknown Event");
114          return false;
115      }
116    }
117  }
118
119  void CField::sendUpdateData(const CArray<double,1>& data)
120  {
121    CTimer::get("XIOS Send Data").resume();
122
123    CContext* context = CContext::getCurrent();
124    CContextClient* client = context->client;
125
126    CEventClient event(getType(), EVENT_ID_UPDATE_DATA);
127
128    map<int, CArray<int,1> >::iterator it;
129    list<CMessage> list_msg;
130    list<CArray<double,1> > list_data;
131
132    if (!grid->doGridHaveDataDistributed())
133    {
134       if (client->isServerLeader())
135       {
136          for (it = grid->storeIndex_toSrv.begin(); it != grid->storeIndex_toSrv.end(); it++)
137          {
138            int rank = it->first;
139            CArray<int,1>& index = it->second;
140
141            list_msg.push_back(CMessage());
142            list_data.push_back(CArray<double,1>(index.numElements()));
143
144            CArray<double,1>& data_tmp = list_data.back();
145            for (int n = 0; n < data_tmp.numElements(); n++) data_tmp(n) = data(index(n));
146
147            list_msg.back() << getId() << data_tmp;
148            event.push(rank, 1, list_msg.back());
149          }
150          client->sendEvent(event);
151       } 
152       else client->sendEvent(event);
153    }
154    else
155    {
156      for (it = grid->storeIndex_toSrv.begin(); it != grid->storeIndex_toSrv.end(); it++)
157      {
158        int rank = it->first;
159        CArray<int,1>& index = it->second;
160
161        list_msg.push_back(CMessage());
162        list_data.push_back(CArray<double,1>(index.numElements()));
163
164        CArray<double,1>& data_tmp = list_data.back();
165        for (int n = 0; n < data_tmp.numElements(); n++) data_tmp(n) = data(index(n));
166
167        list_msg.back() << getId() << data_tmp;
168        event.push(rank, grid->nbSenders[rank], list_msg.back());
169      }
170      client->sendEvent(event);
171    }
172
173    CTimer::get("XIOS Send Data").suspend();
174  }
175
176  void CField::recvUpdateData(CEventServer& event)
177  {
178    vector<int> ranks;
179    vector<CBufferIn*> buffers;
180
181    list<CEventServer::SSubEvent>::iterator it;
182    string fieldId;
183
184    for (it = event.subEvents.begin(); it != event.subEvents.end(); ++it)
185    {
186      int rank = it->rank;
187      CBufferIn* buffer = it->buffer;
188      *buffer >> fieldId;
189      ranks.push_back(rank);
190      buffers.push_back(buffer);
191    }
192    get(fieldId)->recvUpdateData(ranks,buffers);
193  }
194
195  void  CField::recvUpdateData(vector<int>& ranks, vector<CBufferIn*>& buffers)
196  {
197    if (data_srv.empty())
198    {
199      for (map<int, CArray<size_t, 1> >::iterator it = grid->outIndexFromClient.begin(); it != grid->outIndexFromClient.end(); ++it)
200      {
201        int rank = it->first;
202        data_srv.insert(std::make_pair(rank, CArray<double,1>(it->second.numElements())));
203        foperation_srv.insert(pair<int,boost::shared_ptr<func::CFunctor> >(rank,boost::shared_ptr<func::CFunctor>(new func::CInstant(data_srv[rank]))));
204      }
205    }
206
207    CContext* context = CContext::getCurrent();
208    const CDate& currDate = context->getCalendar()->getCurrentDate();
209    const CDate opeDate      = last_operation_srv +freq_op + freq_operation_srv - freq_op;
210    const CDate writeDate    = last_Write_srv     + freq_write_srv;
211
212    if (opeDate <= currDate)
213    {
214      for (int n = 0; n < ranks.size(); n++)
215      {
216        CArray<double,1> data_tmp;
217        *buffers[n] >> data_tmp;
218        (*foperation_srv[ranks[n]])(data_tmp);
219      }
220      last_operation_srv = currDate;
221    }
222
223    if (writeDate < (currDate + freq_operation_srv))
224    {
225      for (int n = 0; n < ranks.size(); n++)
226      {
227        this->foperation_srv[ranks[n]]->final();
228      }
229
230      last_Write_srv = writeDate;
231      writeField();
232      lastlast_Write_srv = last_Write_srv;
233    }
234  }
235
236  void CField::writeField(void)
237  {
238    if (!getRelFile()->allDomainEmpty)
239    {
240      if (grid->doGridHaveDataToWrite() || getRelFile()->type == CFile::type_attr::one_file)
241      {
242        getRelFile()->checkFile();
243        this->incrementNStep();
244        getRelFile()->getDataOutput()->writeFieldData(CField::get(this));
245      }
246    }
247  }
248
249  void CField::sendReadDataRequest(void)
250  {
251    CContext* context = CContext::getCurrent();
252    CContextClient* client = context->client;
253
254    lastDataRequestedFromServer = context->getCalendar()->getCurrentDate();
255    isReadDataRequestPending = true;
256
257    CEventClient event(getType(), EVENT_ID_READ_DATA);
258    if (client->isServerLeader())
259    {
260      CMessage msg;
261      msg << getId();
262      const std::list<int>& ranks = client->getRanksServerLeader();
263      for (std::list<int>::const_iterator itRank = ranks.begin(), itRankEnd = ranks.end(); itRank != itRankEnd; ++itRank)
264        event.push(*itRank, 1, msg);
265      client->sendEvent(event);
266    }
267    else client->sendEvent(event);
268  }
269
270  /*!
271  Send request new data read from file if need be, that is the current data is out-of-date.
272  \return true if and only if some data was requested
273  */
274  bool CField::sendReadDataRequestIfNeeded(void)
275  {
276    const CDate& currentDate = CContext::getCurrent()->getCalendar()->getCurrentDate();
277
278    bool requestData = (currentDate >= lastDataRequestedFromServer + file->output_freq.getValue());
279
280    if (requestData)
281    {
282      cout<<"currentDate : "<<currentDate<<endl ;
283      cout<<"lastDataRequestedFromServer : "<<lastDataRequestedFromServer<<endl ;
284      cout<<"file->output_freq.getValue() : "<<file->output_freq.getValue()<<endl ;
285      cout<<"lastDataRequestedFromServer + file->output_freq.getValue() : "<<lastDataRequestedFromServer + file->output_freq.getValue()<<endl ;
286
287      sendReadDataRequest();
288    }
289
290    return requestData;
291  }
292
293  void CField::recvReadDataRequest(CEventServer& event)
294  {
295    CBufferIn* buffer = event.subEvents.begin()->buffer;
296    StdString fieldId;
297    *buffer >> fieldId;
298    get(fieldId)->recvReadDataRequest();
299  }
300
301  void CField::recvReadDataRequest(void)
302  {
303    CContext* context = CContext::getCurrent();
304    CContextClient* client = context->client;
305
306    CEventClient event(getType(), EVENT_ID_READ_DATA_READY);
307    std::list<CMessage> msgs;
308
309    bool hasData = readField();
310
311    map<int, CArray<double,1> >::iterator it;
312    if (!grid->doGridHaveDataDistributed())
313    {
314       if (client->isServerLeader())
315       {
316          if (!data_srv.empty())
317          {
318            it = data_srv.begin();
319            const std::list<int>& ranks = client->getRanksServerLeader();
320            for (std::list<int>::const_iterator itRank = ranks.begin(), itRankEnd = ranks.end(); itRank != itRankEnd; ++itRank)
321            {
322              msgs.push_back(CMessage());
323              CMessage& msg = msgs.back();
324              msg << getId();
325              if (hasData)
326                msg << getNStep() - 1 << it->second;
327              else
328                msg << int(-1);
329              event.push(*itRank, 1, msg);
330            }
331          }
332          client->sendEvent(event);
333       } 
334       else 
335       {
336          // if (!data_srv.empty())
337          // {
338          //   it = data_srv.begin();
339          //   const std::list<int>& ranks = client->getRanksServerNotLeader();
340          //   for (std::list<int>::const_iterator itRank = ranks.begin(), itRankEnd = ranks.end(); itRank != itRankEnd; ++itRank)
341          //   {
342          //     msgs.push_back(CMessage());
343          //     CMessage& msg = msgs.back();
344          //     msg << getId();
345          //     if (hasData)
346          //       msg << getNStep() - 1 << it->second;
347          //     else
348          //       msg << int(-1);
349          //     event.push(*itRank, 1, msg);
350          //   }
351          // }
352          client->sendEvent(event);
353       }
354    }
355    else
356    {
357      for (it = data_srv.begin(); it != data_srv.end(); it++)
358      {
359        msgs.push_back(CMessage());
360        CMessage& msg = msgs.back();
361        msg << getId();
362        if (hasData)
363          msg << getNStep() - 1 << it->second;
364        else
365          msg << int(-1);
366        event.push(it->first, grid->nbSenders[it->first], msg);
367      }
368      client->sendEvent(event);
369    }
370  }
371
372  bool CField::readField(void)
373  {
374    if (!getRelFile()->allDomainEmpty)
375    {
376      if (grid->doGridHaveDataToWrite() || getRelFile()->type == CFile::type_attr::one_file)
377      {
378        if (data_srv.empty())
379        {
380          for (map<int, CArray<size_t, 1> >::iterator it = grid->outIndexFromClient.begin(); it != grid->outIndexFromClient.end(); ++it)
381            data_srv.insert(std::make_pair(it->first, CArray<double,1>(it->second.numElements())));
382        }
383
384        getRelFile()->checkFile();
385        if (!nstepMax)
386        {
387          nstepMax = getRelFile()->getDataInput()->getFieldNbRecords(CField::get(this));
388        }
389
390        this->incrementNStep();
391
392
393
394        if (getNStep() > nstepMax && (getRelFile()->cyclic.isEmpty() || !getRelFile()->cyclic) )
395          return false;
396
397        getRelFile()->getDataInput()->readFieldData(CField::get(this));
398      }
399    }
400
401    return true;
402  }
403
404  void CField::recvReadDataReady(CEventServer& event)
405  {
406    string fieldId;
407    vector<int> ranks;
408    vector<CBufferIn*> buffers;
409
410    list<CEventServer::SSubEvent>::iterator it;
411    for (it = event.subEvents.begin(); it != event.subEvents.end(); ++it)
412    {
413      ranks.push_back(it->rank);
414      CBufferIn* buffer = it->buffer;
415      *buffer >> fieldId;
416      buffers.push_back(buffer);
417    }
418    get(fieldId)->recvReadDataReady(ranks, buffers);
419  }
420
421  void CField::recvReadDataReady(vector<int> ranks, vector<CBufferIn*> buffers)
422  {
423    CContext* context = CContext::getCurrent();
424    int record;
425    std::map<int, CArray<double,1> > data;
426
427    bool isEOF = false;
428
429    for (int i = 0; i < ranks.size(); i++)
430    {
431      int rank = ranks[i];
432      *buffers[i] >> record;
433      isEOF = (record == int(-1));
434
435      if (!isEOF)
436        *buffers[i] >> data[rank];
437      else
438        break;
439    }
440
441    if (isEOF)
442      serverSourceFilter->signalEndOfStream(lastDataRequestedFromServer);
443    else
444      serverSourceFilter->streamDataFromServer(lastDataRequestedFromServer, data);
445
446    isReadDataRequestPending = false;
447  }
448
449   //----------------------------------------------------------------
450
451   void CField::setRelFile(CFile* _file)
452   {
453      this->file = _file;
454      hasOutputFile = true;
455   }
456
457   //----------------------------------------------------------------
458
459   StdString CField::GetName(void)    { return StdString("field"); }
460   StdString CField::GetDefName(void) { return CField::GetName(); }
461   ENodeType CField::GetType(void)    { return eField; }
462
463   //----------------------------------------------------------------
464
465   CGrid* CField::getRelGrid(void) const
466   {
467      return this->grid;
468   }
469
470   //----------------------------------------------------------------
471
472   CFile* CField::getRelFile(void) const
473   {
474      return this->file;
475   }
476
477   int CField::getNStep(void) const
478   {
479      return this->nstep;
480   }
481
482   func::CFunctor::ETimeType CField::getOperationTimeType() const
483   {
484     return operationTimeType;
485   }
486
487   //----------------------------------------------------------------
488
489   void CField::incrementNStep(void)
490   {
491      this->nstep++;
492   }
493
494   void CField::resetNStep(int nstep /*= 0*/)
495   {
496      this->nstep = nstep;
497   }
498
499   void CField::resetNStepMax(void)
500   {
501      this->nstepMax = 0;
502   }
503
504   //----------------------------------------------------------------
505
506   bool CField::isActive(void) const
507   {
508      return (instantDataFilter != NULL);
509   }
510
511   //----------------------------------------------------------------
512
513   bool CField::wasWritten() const
514   {
515     return written;
516   }
517
518   void CField::setWritten()
519   {
520     written = true;
521   }
522
523   //----------------------------------------------------------------
524
525   bool CField::getUseCompressedOutput() const
526   {
527     return useCompressedOutput;
528   }
529
530   void CField::setUseCompressedOutput()
531   {
532     useCompressedOutput = true;
533   }
534
535   //----------------------------------------------------------------
536
537   boost::shared_ptr<COutputPin> CField::getInstantDataFilter()
538   {
539     return instantDataFilter;
540   }
541
542   //----------------------------------------------------------------
543
544   void CField::solveOnlyReferenceEnabledField(bool doSending2Server)
545   {
546     CContext* context = CContext::getCurrent();
547     if (!isReferenceSolved)
548     {
549        isReferenceSolved = true;
550
551        if (context->hasClient)
552        {
553          solveRefInheritance(true);
554          if (hasDirectFieldReference()) getDirectFieldReference()->solveOnlyReferenceEnabledField(false);
555        }
556        else if (context->hasServer)
557          solveServerOperation();
558
559        solveGridReference();
560
561       if (context->hasClient)
562       {
563         solveGenerateGrid();
564         buildGridTransformationGraph();
565       }
566     }
567   }
568
569   /*!
570     Build up graph of grids which plays role of destination and source in grid transformation
571     This function should be called before \func solveGridReference()
572   */
573   void CField::buildGridTransformationGraph()
574   {
575     CContext* context = CContext::getCurrent();
576     if (context->hasClient)
577     {
578       if (grid && !grid->isTransformed() && hasDirectFieldReference() && grid != getDirectFieldReference()->grid)
579       {
580         grid->addTransGridSource(getDirectFieldReference()->grid);
581       }
582     }
583   }
584
585   /*!
586     Generate a new grid destination if there are more than one grid source pointing to a same grid destination
587   */
588   void CField::generateNewTransformationGridDest()
589   {
590     CContext* context = CContext::getCurrent();
591     if (context->hasClient)
592     {
593       std::map<CGrid*,std::pair<bool,StdString> >& gridSrcMap = grid->getTransGridSource();
594       if (1 < gridSrcMap.size())
595       {
596         // Search for grid source
597         CGrid* gridSrc = grid;
598         CField* currField = this;
599         std::vector<CField*> hieraField;
600
601         while (currField->hasDirectFieldReference() && (gridSrc == grid))
602         {
603           hieraField.push_back(currField);
604           CField* tmp = currField->getDirectFieldReference();
605           currField = tmp;
606           gridSrc = currField->grid;
607         }
608
609         if (gridSrcMap.end() != gridSrcMap.find(gridSrc))
610         {
611           CGrid* gridTmp;
612           std::pair<bool,StdString> newGridDest = gridSrcMap[gridSrc];
613           if (newGridDest.first)
614           {
615             StdString newIdGridDest = newGridDest.second;
616             if (!CGrid::has(newIdGridDest))
617             {
618                ERROR("CGrid* CGrid::generateNewTransformationGridDest()",
619                  << " Something wrong happened! Grid whose id " << newIdGridDest
620                  << "should exist ");
621             }
622             gridTmp = CGrid::get(newIdGridDest);
623           }
624           else
625           {
626             StdString newIdGridDest = CGrid::generateId(gridSrc, grid);
627             gridTmp = CGrid::cloneGrid(newIdGridDest, grid);
628
629             (gridSrcMap[gridSrc]).first = true;
630             (gridSrcMap[gridSrc]).second = newIdGridDest;
631           }
632
633           // Update all descendants
634           for (std::vector<CField*>::iterator it = hieraField.begin(); it != hieraField.end(); ++it)
635           {
636             (*it)->grid = gridTmp;
637             (*it)->updateRef((*it)->grid);
638           }
639         }
640       }
641     }
642   }
643
644   void CField::updateRef(CGrid* grid)
645   {
646     if (!grid_ref.isEmpty()) grid_ref.setValue(grid->getId());
647     else
648     {
649       std::vector<CAxis*> axisTmp = grid->getAxis();
650       std::vector<CDomain*> domainTmp = grid->getDomains();
651       if ((1<axisTmp.size()) || (1<domainTmp.size()))
652         ERROR("void CField::updateRef(CGrid* grid)",
653           << "More than one domain or axis is available for domain_ref/axis_ref of field " << this->getId());
654
655       if ((!domain_ref.isEmpty()) && (domainTmp.empty()))
656         ERROR("void CField::updateRef(CGrid* grid)",
657           << "Incoherent between available domain and domain_ref of field " << this->getId());
658       if ((!axis_ref.isEmpty()) && (axisTmp.empty()))
659         ERROR("void CField::updateRef(CGrid* grid)",
660           << "Incoherent between available axis and axis_ref of field " << this->getId());
661
662       if (!domain_ref.isEmpty()) domain_ref.setValue(domainTmp[0]->getId());
663       if (!axis_ref.isEmpty()) axis_ref.setValue(axisTmp[0]->getId());
664     }
665   }
666
667   void CField::solveAllReferenceEnabledField(bool doSending2Server)
668   {
669     CContext* context = CContext::getCurrent();
670     solveOnlyReferenceEnabledField(doSending2Server);
671
672     if (!areAllReferenceSolved)
673     {
674        areAllReferenceSolved = true;
675
676        if (context->hasClient)
677        {
678          solveRefInheritance(true);
679          if (hasDirectFieldReference()) getDirectFieldReference()->solveAllReferenceEnabledField(false);
680        }
681        else if (context->hasServer)
682          solveServerOperation();
683
684        solveGridReference();
685     }
686
687     solveGridDomainAxisRef(doSending2Server);
688
689     if (context->hasClient)
690     {
691       solveTransformedGrid();
692     }
693
694     solveCheckMaskIndex(doSending2Server);
695   }
696
697   std::map<int, StdSize> CField::getGridAttributesBufferSize()
698   {
699     return grid->getAttributesBufferSize();
700   }
701
702   std::map<int, StdSize> CField::getGridDataBufferSize()
703   {
704     return grid->getDataBufferSize(getId());
705   }
706
707   //----------------------------------------------------------------
708
709   void CField::solveServerOperation(void)
710   {
711      CContext* context = CContext::getCurrent();
712
713      if (!context->hasServer || !hasOutputFile) return;
714
715      if (freq_op.isEmpty())
716        freq_op.setValue(TimeStep);
717
718      if (freq_offset.isEmpty())
719        freq_offset.setValue(NoneDu);
720
721      freq_operation_srv = file->output_freq.getValue();
722      freq_write_srv     = file->output_freq.getValue();
723
724      lastlast_Write_srv = context->getCalendar()->getInitDate();
725      last_Write_srv     = context->getCalendar()->getInitDate();
726      last_operation_srv = context->getCalendar()->getInitDate();
727
728      const CDuration toffset = freq_operation_srv - freq_offset.getValue() - context->getCalendar()->getTimeStep();
729      last_operation_srv     = last_operation_srv - toffset;
730
731      if (operation.isEmpty())
732        ERROR("void CField::solveServerOperation(void)",
733              << "An operation must be defined for field \"" << getId() << "\".");
734
735      boost::shared_ptr<func::CFunctor> functor;
736      CArray<double, 1> dummyData;
737
738#define DECLARE_FUNCTOR(MType, mtype) \
739      if (operation.getValue().compare(#mtype) == 0) \
740      { \
741        functor.reset(new func::C##MType(dummyData)); \
742      }
743
744#include "functor_type.conf"
745
746      if (!functor)
747        ERROR("void CField::solveServerOperation(void)",
748              << "\"" << operation << "\" is not a valid operation.");
749
750      operationTimeType = functor->timeType();
751   }
752
753   //----------------------------------------------------------------
754
755   /*!
756    * Constructs the graph filter for the field, enabling or not the data output.
757    * This method should not be called more than once with enableOutput equal to true.
758    *
759    * \param gc the garbage collector to use when building the filter graph
760    * \param enableOutput must be true when the field data is to be
761    *                     read by the client or/and written to a file
762    */
763   void CField::buildFilterGraph(CGarbageCollector& gc, bool enableOutput)
764   {
765     if (!areAllReferenceSolved) solveAllReferenceEnabledField(false);
766
767     // Start by building a filter which can provide the field's instant data
768     if (!instantDataFilter)
769     {
770       // Check if we have an expression to parse
771       if (!content.empty())
772       {
773         boost::scoped_ptr<IFilterExprNode> expr(parseExpr(content + '\0'));
774         instantDataFilter = expr->reduce(gc, *this);
775       }
776       // Check if we have a reference on another field
777       else if (!field_ref.isEmpty())
778         instantDataFilter = getFieldReference(gc);
779       // Check if the data is to be read from a file
780       else if (file && !file->mode.isEmpty() && file->mode == CFile::mode_attr::read)
781         instantDataFilter = serverSourceFilter = boost::shared_ptr<CSourceFilter>(new CSourceFilter(grid,
782                                                                                                     freq_offset.isEmpty() ? NoneDu : freq_offset));
783       else // The data might be passed from the model
784         instantDataFilter = clientSourceFilter = boost::shared_ptr<CSourceFilter>(new CSourceFilter(grid));
785     }
786
787     // If the field data is to be read by the client or/and written to a file
788     if (enableOutput && !storeFilter && !fileWriterFilter)
789     {
790       if (!read_access.isEmpty() && read_access)
791       {
792         storeFilter = boost::shared_ptr<CStoreFilter>(new CStoreFilter(gc, CContext::getCurrent(), grid));
793         instantDataFilter->connectOutput(storeFilter, 0);
794       }
795
796       if (file && (file->mode.isEmpty() || file->mode == CFile::mode_attr::write))
797       {
798         fileWriterFilter = boost::shared_ptr<CFileWriterFilter>(new CFileWriterFilter(gc, this));
799         getTemporalDataFilter(gc, file->output_freq)->connectOutput(fileWriterFilter, 0);
800       }
801     }
802   }
803
804   /*!
805    * Returns the filter needed to handle the field reference.
806    * This method should only be called when building the filter graph corresponding to the field.
807    *
808    * \param gc the garbage collector to use
809    * \return the output pin corresponding to the field reference
810    */
811   boost::shared_ptr<COutputPin> CField::getFieldReference(CGarbageCollector& gc)
812   {
813     if (instantDataFilter || field_ref.isEmpty())
814       ERROR("COutputPin* CField::getFieldReference(CGarbageCollector& gc)",
815             "Impossible to get the field reference for a field which has already been parsed or which does not have a field_ref.");
816
817     CField* fieldRef = CField::get(field_ref);
818     fieldRef->buildFilterGraph(gc, false);
819
820     std::pair<boost::shared_ptr<CFilter>, boost::shared_ptr<CFilter> > filters;
821     // Check if a spatial transformation is needed
822     if (grid && grid != fieldRef->grid && grid->hasTransform())
823     {
824       double defaultValue = 0.0;
825       if (!default_value.isEmpty()) defaultValue = this->default_value;
826       filters = CSpatialTransformFilter::buildFilterGraph(gc, fieldRef->grid, grid, defaultValue);
827     }
828
829     else
830       filters.first = filters.second = boost::shared_ptr<CFilter>(new CPassThroughFilter(gc));
831
832     fieldRef->getInstantDataFilter()->connectOutput(filters.first, 0);
833
834     return filters.second;
835   }
836
837   /*!
838    * Returns the filter needed to handle a self reference in the field's expression.
839    * If the needed filter does not exist, it is created, otherwise it is reused.
840    * This method should only be called when building the filter graph corresponding
841    * to the field's expression.
842    *
843    * \param gc the garbage collector to use
844    * \return the output pin corresponding to a self reference
845    */
846   boost::shared_ptr<COutputPin> CField::getSelfReference(CGarbageCollector& gc)
847   {
848     if (instantDataFilter || content.empty())
849       ERROR("COutputPin* CField::getSelfReference(CGarbageCollector& gc)",
850             "Impossible to add a self reference to a field which has already been parsed or which does not have an expression.");
851
852     if (!selfReferenceFilter)
853     {
854       if (file && !file->mode.isEmpty() && file->mode == CFile::mode_attr::read)
855       {
856         if (!serverSourceFilter)
857           serverSourceFilter = boost::shared_ptr<CSourceFilter>(new CSourceFilter(grid,
858                                                                                   freq_offset.isEmpty() ? NoneDu : freq_offset));
859
860         selfReferenceFilter = serverSourceFilter;
861       }
862       else if (!field_ref.isEmpty())
863         selfReferenceFilter = getFieldReference(gc);
864       else
865       {
866         if (!clientSourceFilter)
867           clientSourceFilter = boost::shared_ptr<CSourceFilter>(new CSourceFilter(grid));
868
869         selfReferenceFilter = clientSourceFilter;
870       }
871     }
872
873     return selfReferenceFilter;
874   }
875
876   /*!
877    * Returns the temporal filter corresponding to the field's temporal operation
878    * for the specified operation frequency. The filter is created if it does not
879    * exist, otherwise it is reused.
880    *
881    * \param gc the garbage collector to use
882    * \param outFreq the operation frequency, i.e. the frequency at which the output data will be computed
883    * \return the output pin corresponding to the requested temporal filter
884    */
885   boost::shared_ptr<COutputPin> CField::getTemporalDataFilter(CGarbageCollector& gc, CDuration outFreq)
886   {
887     std::map<CDuration, boost::shared_ptr<COutputPin> >::iterator it = temporalDataFilters.find(outFreq);
888
889     if (it == temporalDataFilters.end())
890     {
891       if (operation.isEmpty())
892         ERROR("void CField::getTemporalDataFilter(CGarbageCollector& gc, CDuration outFreq)",
893               << "An operation must be defined for field \"" << getId() << "\".");
894
895       if (freq_op.isEmpty())
896         freq_op.setValue(TimeStep);
897       if (freq_offset.isEmpty())
898         freq_offset.setValue(NoneDu);
899
900       const bool ignoreMissingValue = (!detect_missing_value.isEmpty() && !default_value.isEmpty() && detect_missing_value == true);
901
902       boost::shared_ptr<CTemporalFilter> temporalFilter(new CTemporalFilter(gc, operation,
903                                                                             CContext::getCurrent()->getCalendar()->getInitDate(),
904                                                                             freq_op, freq_offset, outFreq,
905                                                                             ignoreMissingValue, ignoreMissingValue ? default_value : 0.0));
906       instantDataFilter->connectOutput(temporalFilter, 0);
907
908       it = temporalDataFilters.insert(std::make_pair(outFreq, temporalFilter)).first;
909     }
910
911     return it->second;
912   }
913
914   //----------------------------------------------------------------
915/*
916   void CField::fromBinary(StdIStream& is)
917   {
918      SuperClass::fromBinary(is);
919#define CLEAR_ATT(name_)\
920      SuperClassAttribute::operator[](#name_)->reset()
921
922         CLEAR_ATT(domain_ref);
923         CLEAR_ATT(axis_ref);
924#undef CLEAR_ATT
925
926   }
927*/
928   //----------------------------------------------------------------
929
930   void CField::solveGridReference(void)
931   {
932      if (grid_ref.isEmpty() && domain_ref.isEmpty() && axis_ref.isEmpty() && scalar_ref.isEmpty())
933      {
934        ERROR("CField::solveGridReference(void)",
935              << "A grid must be defined for field '" << getFieldOutputName() << "' .");
936      }
937      else if (!grid_ref.isEmpty() && (!domain_ref.isEmpty() || !axis_ref.isEmpty() || !scalar_ref.isEmpty()))
938      {
939        ERROR("CField::solveGridReference(void)",
940              << "Field '" << getFieldOutputName() << "' has both a grid and a domain/axis/scalar." << std::endl
941              << "Please define either 'grid_ref' or 'domain_ref'/'axis_ref'/'scalar_ref'.");
942      }
943
944      if (grid_ref.isEmpty())
945      {
946        std::vector<CDomain*> vecDom;
947        std::vector<CAxis*> vecAxis;
948        std::vector<CScalar*> vecScalar;
949        std::vector<int> axisDomainOrderTmp;
950       
951        if (!domain_ref.isEmpty())
952        {
953          StdString tmp = domain_ref.getValue();
954          if (CDomain::has(domain_ref))
955          {
956            vecDom.push_back(CDomain::get(domain_ref));
957            axisDomainOrderTmp.push_back(2);
958          }
959          else
960            ERROR("CField::solveGridReference(void)",
961                  << "Invalid reference to domain '" << domain_ref.getValue() << "'.");
962        }
963
964        if (!axis_ref.isEmpty())
965        {
966          if (CAxis::has(axis_ref))
967          {
968            vecAxis.push_back(CAxis::get(axis_ref));
969            axisDomainOrderTmp.push_back(1);
970          }
971          else
972            ERROR("CField::solveGridReference(void)",
973                  << "Invalid reference to axis '" << axis_ref.getValue() << "'.");
974        }
975
976        if (!scalar_ref.isEmpty())
977        {
978          if (CScalar::has(scalar_ref))
979          {
980            vecScalar.push_back(CScalar::get(scalar_ref));
981            axisDomainOrderTmp.push_back(0);
982          }
983          else
984            ERROR("CField::solveGridReference(void)",
985                  << "Invalid reference to scalar '" << scalar_ref.getValue() << "'.");
986        }
987       
988        CArray<int,1> axisDomainOrder(axisDomainOrderTmp.size());
989        for (int idx = 0; idx < axisDomainOrderTmp.size(); ++idx)
990        {
991          axisDomainOrder(idx) = axisDomainOrderTmp[idx];
992        }
993
994        // Warning: the gridId shouldn't be set as the grid_ref since it could be inherited
995        StdString gridId = CGrid::generateId(vecDom, vecAxis, vecScalar,axisDomainOrder);
996        if (CGrid::has(gridId))
997          this->grid = CGrid::get(gridId);
998        else
999          this->grid = CGrid::createGrid(gridId, vecDom, vecAxis, vecScalar,axisDomainOrder);
1000      }
1001      else
1002      {
1003        if (CGrid::has(grid_ref))
1004          this->grid = CGrid::get(grid_ref);
1005        else
1006          ERROR("CField::solveGridReference(void)",
1007                << "Invalid reference to grid '" << grid_ref.getValue() << "'.");
1008      }
1009   }
1010
1011   void CField::solveGridDomainAxisRef(bool checkAtt)
1012   {
1013     grid->solveDomainAxisRef(checkAtt);
1014   }
1015
1016   void CField::solveCheckMaskIndex(bool doSendingIndex)
1017   {
1018     grid->checkMaskIndex(doSendingIndex);
1019   }
1020
1021   void CField::solveTransformedGrid()
1022   {
1023     if (grid && !grid->isTransformed() && hasDirectFieldReference() && grid != getDirectFieldReference()->grid)
1024     {
1025       std::vector<CGrid*> grids;
1026       // Source grid
1027       grids.push_back(getDirectFieldReference()->grid);
1028       // Intermediate grids
1029       if (!grid_path.isEmpty())
1030       {
1031         std::string gridId;
1032         size_t start = 0, end;
1033
1034         do
1035         {
1036           end = grid_path.getValue().find(',', start);
1037           if (end != std::string::npos)
1038           {
1039             gridId = grid_path.getValue().substr(start, end - start);
1040             start = end + 1;
1041           }
1042           else
1043             gridId = grid_path.getValue().substr(start);
1044
1045           if (!CGrid::has(gridId))
1046             ERROR("void CField::solveTransformedGrid()",
1047                   << "Invalid grid_path, the grid '" << gridId << "' does not exist.");
1048
1049           grids.push_back(CGrid::get(gridId));
1050         }
1051         while (end != std::string::npos);
1052       }
1053       // Destination grid
1054       grids.push_back(grid);
1055
1056       for (size_t i = 0, count = grids.size() - 1; i < count; ++i)
1057       {
1058         CGrid *gridSrc  = grids[i];
1059         CGrid *gridDest = grids[i + 1];
1060         if (!gridDest->isTransformed())
1061           gridDest->transformGrid(gridSrc);
1062       }
1063     }
1064     else if (grid && grid->hasTransform() && !grid->isTransformed())
1065     {
1066       grid->transformGrid(grid);
1067     }
1068   }
1069
1070   void CField::solveGenerateGrid()
1071   {
1072     if (grid && !grid->isTransformed() && hasDirectFieldReference() && grid != getDirectFieldReference()->grid)
1073       grid->completeGrid(getDirectFieldReference()->grid);
1074     else
1075       grid->completeGrid();
1076   }
1077
1078   void CField::solveGridDomainAxisBaseRef()
1079   {
1080     grid->solveDomainAxisRef(false);
1081     grid->solveDomainAxisBaseRef();
1082   }
1083
1084   ///-------------------------------------------------------------------
1085
1086   template <>
1087   void CGroupTemplate<CField, CFieldGroup, CFieldAttributes>::solveRefInheritance(void)
1088   {
1089      if (this->group_ref.isEmpty()) return;
1090      StdString gref = this->group_ref.getValue();
1091
1092      if (!CFieldGroup::has(gref))
1093         ERROR("CGroupTemplate<CField, CFieldGroup, CFieldAttributes>::solveRefInheritance(void)",
1094               << "[ gref = " << gref << "]"
1095               << " invalid group name !");
1096
1097      CFieldGroup* group = CFieldGroup::get(gref);
1098      CFieldGroup* owner = CFieldGroup::get(boost::polymorphic_downcast<CFieldGroup*>(this));
1099
1100      std::vector<CField*> allChildren  = group->getAllChildren();
1101      std::vector<CField*>::iterator it = allChildren.begin(), end = allChildren.end();
1102
1103      for (; it != end; it++)
1104      {
1105         CField* child = *it;
1106         if (child->hasId()) owner->createChild()->field_ref.setValue(child->getId());
1107
1108      }
1109   }
1110
1111   void CField::scaleFactorAddOffset(double scaleFactor, double addOffset)
1112   {
1113     map<int, CArray<double,1> >::iterator it;
1114     for (it = data_srv.begin(); it != data_srv.end(); it++) it->second = (it->second - addOffset) / scaleFactor;
1115   }
1116
1117   void CField::invertScaleFactorAddOffset(double scaleFactor, double addOffset)
1118   {
1119     map<int, CArray<double,1> >::iterator it;
1120     for (it = data_srv.begin(); it != data_srv.end(); it++) it->second = it->second * scaleFactor + addOffset;
1121   }
1122
1123   void CField::outputField(CArray<double,3>& fieldOut)
1124   {
1125      map<int, CArray<double,1> >::iterator it;
1126      for (it = data_srv.begin(); it != data_srv.end(); it++)
1127      {
1128        grid->outputField(it->first, it->second, fieldOut.dataFirst());
1129      }
1130   }
1131
1132   void CField::outputField(CArray<double,2>& fieldOut)
1133   {
1134      map<int, CArray<double,1> >::iterator it;
1135      for(it=data_srv.begin();it!=data_srv.end();it++)
1136      {
1137         grid->outputField(it->first, it->second, fieldOut.dataFirst());
1138      }
1139   }
1140
1141   void CField::outputField(CArray<double,1>& fieldOut)
1142   {
1143      map<int, CArray<double,1> >::iterator it;
1144
1145      for (it = data_srv.begin(); it != data_srv.end(); it++)
1146      {
1147         grid->outputField(it->first, it->second, fieldOut.dataFirst());
1148      }
1149   }
1150
1151   void CField::inputField(CArray<double,3>& fieldOut)
1152   {
1153      map<int, CArray<double,1> >::iterator it;
1154      for (it = data_srv.begin(); it != data_srv.end(); it++)
1155      {
1156        grid->inputField(it->first, fieldOut.dataFirst(), it->second);
1157      }
1158   }
1159
1160   void CField::inputField(CArray<double,2>& fieldOut)
1161   {
1162      map<int, CArray<double,1> >::iterator it;
1163      for(it = data_srv.begin(); it != data_srv.end(); it++)
1164      {
1165         grid->inputField(it->first, fieldOut.dataFirst(), it->second);
1166      }
1167   }
1168
1169   void CField::inputField(CArray<double,1>& fieldOut)
1170   {
1171      map<int, CArray<double,1> >::iterator it;
1172      for (it = data_srv.begin(); it != data_srv.end(); it++)
1173      {
1174         grid->inputField(it->first, fieldOut.dataFirst(), it->second);
1175      }
1176   }
1177
1178   void CField::outputCompressedField(CArray<double,1>& fieldOut)
1179   {
1180      map<int, CArray<double,1> >::iterator it;
1181
1182      for (it = data_srv.begin(); it != data_srv.end(); it++)
1183      {
1184         grid->outputCompressedField(it->first, it->second, fieldOut.dataFirst());
1185      }
1186   }
1187
1188   ///-------------------------------------------------------------------
1189
1190   void CField::parse(xml::CXMLNode& node)
1191   {
1192      SuperClass::parse(node);
1193      if (!node.getContent(this->content))
1194      {
1195        if (node.goToChildElement())
1196        {
1197          do
1198          {
1199            if (node.getElementName() == "variable" || node.getElementName() == "variable_group") this->getVirtualVariableGroup()->parseChild(node);
1200          } while (node.goToNextElement());
1201          node.goToParentElement();
1202        }
1203      }
1204    }
1205
1206   /*!
1207     This function retrieves Id of corresponding domain_ref and axis_ref (if any)
1208   of a field. In some cases, only domain exists but axis doesn't
1209   \return pair of Domain and Axis id
1210   */
1211   const std::vector<StdString>& CField::getRefDomainAxisIds()
1212   {
1213     CGrid* cgPtr = getRelGrid();
1214     if (NULL != cgPtr)
1215     {
1216       std::vector<StdString>::iterator it;
1217       if (!domain_ref.isEmpty())
1218       {
1219         std::vector<StdString> domainList = cgPtr->getDomainList();
1220         it = std::find(domainList.begin(), domainList.end(), domain_ref.getValue());
1221         if (domainList.end() != it) domAxisScalarIds_[0] = *it;
1222       }
1223
1224       if (!axis_ref.isEmpty())
1225       {
1226         std::vector<StdString> axisList = cgPtr->getAxisList();
1227         it = std::find(axisList.begin(), axisList.end(), axis_ref.getValue());
1228         if (axisList.end() != it) domAxisScalarIds_[1] = *it;
1229       }
1230
1231       if (!scalar_ref.isEmpty())
1232       {
1233         std::vector<StdString> scalarList = cgPtr->getScalarList();
1234         it = std::find(scalarList.begin(), scalarList.end(), scalar_ref.getValue());
1235         if (scalarList.end() != it) domAxisScalarIds_[2] = *it;
1236       }
1237     }
1238     return (domAxisScalarIds_);
1239   }
1240
1241   CVariable* CField::addVariable(const string& id)
1242   {
1243     return vVariableGroup->createChild(id);
1244   }
1245
1246   CVariableGroup* CField::addVariableGroup(const string& id)
1247   {
1248     return vVariableGroup->createChildGroup(id);
1249   }
1250
1251   void CField::sendAddAllVariables()
1252   {
1253     std::vector<CVariable*> allVar = getAllVariables();
1254     std::vector<CVariable*>::const_iterator it = allVar.begin();
1255     std::vector<CVariable*>::const_iterator itE = allVar.end();
1256
1257     for (; it != itE; ++it)
1258     {
1259       this->sendAddVariable((*it)->getId());
1260       (*it)->sendAllAttributesToServer();
1261       (*it)->sendValue();
1262     }
1263   }
1264
1265   void CField::sendAddVariable(const string& id)
1266   {
1267    CContext* context = CContext::getCurrent();
1268
1269    if (!context->hasServer)
1270    {
1271       CContextClient* client = context->client;
1272
1273       CEventClient event(this->getType(),EVENT_ID_ADD_VARIABLE);
1274       if (client->isServerLeader())
1275       {
1276         CMessage msg;
1277         msg << this->getId();
1278         msg << id;
1279         const std::list<int>& ranks = client->getRanksServerLeader();
1280         for (std::list<int>::const_iterator itRank = ranks.begin(), itRankEnd = ranks.end(); itRank != itRankEnd; ++itRank)
1281           event.push(*itRank,1,msg);
1282         client->sendEvent(event);
1283       }
1284       else client->sendEvent(event);
1285    }
1286   }
1287
1288   void CField::sendAddVariableGroup(const string& id)
1289   {
1290    CContext* context = CContext::getCurrent();
1291    if (!context->hasServer)
1292    {
1293       CContextClient* client = context->client;
1294
1295       CEventClient event(this->getType(),EVENT_ID_ADD_VARIABLE_GROUP);
1296       if (client->isServerLeader())
1297       {
1298         CMessage msg;
1299         msg << this->getId();
1300         msg << id;
1301         const std::list<int>& ranks = client->getRanksServerLeader();
1302         for (std::list<int>::const_iterator itRank = ranks.begin(), itRankEnd = ranks.end(); itRank != itRankEnd; ++itRank)
1303           event.push(*itRank,1,msg);
1304         client->sendEvent(event);
1305       }
1306       else client->sendEvent(event);
1307    }
1308   }
1309
1310   void CField::recvAddVariable(CEventServer& event)
1311   {
1312
1313      CBufferIn* buffer = event.subEvents.begin()->buffer;
1314      string id;
1315      *buffer >> id;
1316      get(id)->recvAddVariable(*buffer);
1317   }
1318
1319   void CField::recvAddVariable(CBufferIn& buffer)
1320   {
1321      string id;
1322      buffer >> id;
1323      addVariable(id);
1324   }
1325
1326   void CField::recvAddVariableGroup(CEventServer& event)
1327   {
1328
1329      CBufferIn* buffer = event.subEvents.begin()->buffer;
1330      string id;
1331      *buffer >> id;
1332      get(id)->recvAddVariableGroup(*buffer);
1333   }
1334
1335   void CField::recvAddVariableGroup(CBufferIn& buffer)
1336   {
1337      string id;
1338      buffer >> id;
1339      addVariableGroup(id);
1340   }
1341
1342   DEFINE_REF_FUNC(Field,field)
1343} // namespace xios
Note: See TracBrowser for help on using the repository browser.