Ignore:
Timestamp:
09/11/19 10:22:09 (5 years ago)
Author:
yushan
Message:

Introducing the new graph functionality. Attribute build_workflow_graph=.TRUE. is used in the field definition section in the xml file to enable the workflow graph of the field and other fields referecing to it. A more detailed document will be available soon on the graph fuctionality.

File:
1 edited

Legend:

Unmodified
Added
Removed

  • XIOS/trunk/src/filter/binary_arithmetic_filter.cpp

    r804 r1704  
    11#include "binary_arithmetic_filter.hpp" 
     2#include "workflow_graph.hpp" 
     3#include "yacc_var.hpp" 
     4#include "file.hpp" 
     5 
    26 
    37namespace xios 
    … …  
    711    , op(operatorExpr.getOpScalarField(op)) 
    812    , value(value) 
    9   { /* Nothing to do */ }; 
     13  {  
     14    expression.assign(*yacc_globalInputText_ptr, 0, yacc_globalInputText_ptr->size()-1); 
     15  }; 
     16 
     17  std::tuple<int, int, int> CScalarFieldArithmeticFilter::buildGraph(std::vector<CDataPacketPtr> data) 
     18  { 
     19    bool building_graph = this->tag ? data[0]->timestamp >= this->start_graph && data[0]->timestamp <= this->end_graph : false; 
     20    // bool building_graph = true; 
     21    int unique_filter_id; 
     22    bool firstround; 
     23 
     24    if(building_graph) 
     25    { 
     26      CWorkflowGraph::allocNodeEdge(); 
     27 
     28      size_t filterhash = std::hash<StdString>{}(expression+to_string(data[0]->timestamp)+this->field->getId()); 
     29 
     30      // first round 
     31      if(CWorkflowGraph::mapHashFilterID_ptr->find(filterhash) == CWorkflowGraph::mapHashFilterID_ptr->end()) 
     32      { 
     33        firstround = true; 
     34        this->filterID = InvalidableObject::filterIdGenerator++; 
     35        int edgeID = InvalidableObject::edgeIdGenerator++; 
     36 
     37        CWorkflowGraph::addNode(this->filterID, "Arithmetic Filter\\n("+expression+")", 3, 1, 0, data[0]); 
     38        (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].filter_tag = this->tag; 
     39        (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].distance = data[0]->distance+1; 
     40 
     41 
     42        (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].attributes = this->field->record4graphXiosAttributes(); 
     43        if(this->field->file) (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].attributes += "</br>file attributes : </br>" +this->field->file->record4graphXiosAttributes(); 
     44       
     45 
     46        if(CWorkflowGraph::build_begin) 
     47        { 
     48          CWorkflowGraph::addEdge(edgeID, this->filterID, data[0]); 
     49          (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].expected_entry_nb ++; 
     50 
     51          (*CWorkflowGraph::mapFilters_ptr_with_info)[data[0]->src_filterID].filter_filled = 0 ; 
     52        } 
     53        else CWorkflowGraph::build_begin = true; 
     54 
     55        (*CWorkflowGraph::mapHashFilterID_ptr)[filterhash] = this->filterID;  
     56        unique_filter_id = this->filterID; 
     57      } 
     58      // not first round 
     59      else  
     60      { 
     61        firstround=false; 
     62        unique_filter_id = (*CWorkflowGraph::mapHashFilterID_ptr)[filterhash]; 
     63        if(data[0]->src_filterID != unique_filter_id) 
     64        { 
     65          int edgeID = InvalidableObject::edgeIdGenerator++; 
     66          CWorkflowGraph::addEdge(edgeID, unique_filter_id, data[0]);   
     67          (*CWorkflowGraph::mapFilters_ptr_with_info)[data[0]->src_filterID].filter_filled = 0 ;  
     68          (*CWorkflowGraph::mapFilters_ptr_with_info)[unique_filter_id].expected_entry_nb ++; 
     69        } 
     70      }   
     71    } 
     72 
     73    return std::make_tuple(building_graph, firstround, unique_filter_id); 
     74  } 
     75 
    1076 
    1177  CDataPacketPtr CScalarFieldArithmeticFilter::apply(std::vector<CDataPacketPtr> data) 
    … …  
    1682    packet->status = data[0]->status; 
    1783 
     84    std::tuple<int, int, int> graph = buildGraph(data); 
     85 
     86    if(std::get<0>(graph)) packet->src_filterID = std::get<2>(graph); 
     87    if(std::get<0>(graph) && std::get<1>(graph)) packet->distance = data[0]->distance+1; 
     88    if(std::get<0>(graph) && !std::get<1>(graph)) packet->distance = data[0]->distance; 
     89 
     90    packet->field = this->field; 
     91 
    1892    if (packet->status == CDataPacket::NO_ERROR) 
    1993      packet->data.reference(op(value, data[0]->data)); 
    … …  
    26100    , op(operatorExpr.getOpFieldScalar(op)) 
    27101    , value(value) 
    28   { /* Nothing to do */ }; 
     102  {  
     103    expression.assign(*yacc_globalInputText_ptr, 0, yacc_globalInputText_ptr->size()-1); 
     104  }; 
     105 
     106  std::tuple<int, int, int> CFieldScalarArithmeticFilter::buildGraph(std::vector<CDataPacketPtr> data) 
     107  { 
     108    bool building_graph = this->tag ? data[0]->timestamp >= this->start_graph && data[0]->timestamp <= this->end_graph : false; 
     109    // bool building_graph = true; 
     110    int unique_filter_id; 
     111    bool firstround; 
     112 
     113    if(building_graph) 
     114    { 
     115      CWorkflowGraph::allocNodeEdge(); 
     116 
     117      size_t filterhash = std::hash<StdString>{}(expression+to_string(data[0]->timestamp)+this->field->getId()); 
     118 
     119      // first round 
     120      if(CWorkflowGraph::mapHashFilterID_ptr->find(filterhash) == CWorkflowGraph::mapHashFilterID_ptr->end()) 
     121      { 
     122        firstround = true; 
     123        this->filterID = InvalidableObject::filterIdGenerator++; 
     124        int edgeID = InvalidableObject::edgeIdGenerator++; 
     125 
     126        CWorkflowGraph::addNode(this->filterID, "Arithmetic Filter\\n("+expression+")", 3, 1, 0, data[0]); 
     127        (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].filter_tag = this->tag; 
     128        (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].distance = data[0]->distance+1; 
     129 
     130 
     131        (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].attributes = this->field->record4graphXiosAttributes(); 
     132        if(this->field->file) (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].attributes += "</br>file attributes : </br>" +this->field->file->record4graphXiosAttributes(); 
     133       
     134 
     135        if(CWorkflowGraph::build_begin) 
     136        { 
     137          CWorkflowGraph::addEdge(edgeID, this->filterID, data[0]); 
     138          (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].expected_entry_nb ++; 
     139 
     140          (*CWorkflowGraph::mapFilters_ptr_with_info)[data[0]->src_filterID].filter_filled = 0 ; 
     141        } 
     142        else CWorkflowGraph::build_begin = true; 
     143 
     144        (*CWorkflowGraph::mapHashFilterID_ptr)[filterhash] = this->filterID;  
     145        unique_filter_id = this->filterID; 
     146      } 
     147      // not first round 
     148      else  
     149      { 
     150        firstround=false; 
     151        unique_filter_id = (*CWorkflowGraph::mapHashFilterID_ptr)[filterhash]; 
     152        if(data[0]->src_filterID != unique_filter_id) 
     153        { 
     154          int edgeID = InvalidableObject::edgeIdGenerator++; 
     155          CWorkflowGraph::addEdge(edgeID, unique_filter_id, data[0]);   
     156          (*CWorkflowGraph::mapFilters_ptr_with_info)[data[0]->src_filterID].filter_filled = 0 ;  
     157          (*CWorkflowGraph::mapFilters_ptr_with_info)[unique_filter_id].expected_entry_nb ++; 
     158        } 
     159      }   
     160    } 
     161 
     162    return std::make_tuple(building_graph, firstround, unique_filter_id); 
     163  } 
    29164 
    30165  CDataPacketPtr CFieldScalarArithmeticFilter::apply(std::vector<CDataPacketPtr> data) 
    … …  
    35170    packet->status = data[0]->status; 
    36171 
     172    std::tuple<int, int, int> graph = buildGraph(data); 
     173 
     174    if(std::get<0>(graph)) packet->src_filterID = std::get<2>(graph); 
     175    if(std::get<0>(graph) && std::get<1>(graph)) packet->distance = data[0]->distance+1; 
     176    if(std::get<0>(graph) && !std::get<1>(graph)) packet->distance = data[0]->distance; 
     177 
     178    packet->field = this->field; 
     179 
    37180    if (packet->status == CDataPacket::NO_ERROR) 
    38181      packet->data.reference(op(data[0]->data, value)); 
    … …  
    44187    : CFilter(gc, 2, this) 
    45188    , op(operatorExpr.getOpFieldField(op)) 
    46   { /* Nothing to do */ }; 
     189  {  
     190    expression.assign(*yacc_globalInputText_ptr, 0, yacc_globalInputText_ptr->size()-1); 
     191  }; 
     192 
     193  std::tuple<int, int, int> CFieldFieldArithmeticFilter::buildGraph(std::vector<CDataPacketPtr> data) 
     194  { 
     195    bool building_graph = this->tag ? ((data[0]->timestamp >= this->field->field_graph_start && data[0]->timestamp <= this->field->field_graph_end) && (data[0]->timestamp == data[1]->timestamp)) : false; 
     196 
     197    int unique_filter_id; 
     198 
     199    bool firstround; 
     200 
     201    if(building_graph) 
     202    {   
     203      CWorkflowGraph::allocNodeEdge(); 
     204 
     205      // std::cout<<"CFieldFieldArithmeticFilter::apply filter tag = "<<this->tag<<std::endl; 
     206 
     207      size_t filterhash = std::hash<StdString>{}(expression+to_string(data[0]->timestamp)+this->field->getId()); 
     208 
     209      // first round 
     210      if(CWorkflowGraph::mapHashFilterID_ptr->find(filterhash) == CWorkflowGraph::mapHashFilterID_ptr->end()) 
     211      { 
     212        firstround = true; 
     213        this->filterID = InvalidableObject::filterIdGenerator++; 
     214        int edgeID = InvalidableObject::edgeIdGenerator++; 
     215     
     216        CWorkflowGraph::addNode(this->filterID, "Arithmetic Filter\\n("+expression+")", 3, 1, 0, data[0]); 
     217        (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].attributes = this->field->record4graphXiosAttributes(); 
     218        (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].distance = data[0]->distance+1; 
     219 
     220        if(this->field->file) (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].attributes += "</br>file attributes : </br>" +this->field->file->record4graphXiosAttributes(); 
     221     
     222        (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].filter_tag = this->tag; 
     223        if(CWorkflowGraph::build_begin) 
     224        { 
     225 
     226          CWorkflowGraph::addEdge(edgeID, this->filterID, data[0]); 
     227          (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].expected_entry_nb ++; 
     228 
     229          edgeID = InvalidableObject::edgeIdGenerator++; 
     230 
     231          CWorkflowGraph::addEdge(edgeID, this->filterID, data[1]); 
     232          (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].expected_entry_nb ++; 
     233 
     234          (*CWorkflowGraph::mapFilters_ptr_with_info)[data[0]->src_filterID].filter_filled = 0 ; 
     235          (*CWorkflowGraph::mapFilters_ptr_with_info)[data[1]->src_filterID].filter_filled = 0 ; 
     236        } 
     237        CWorkflowGraph::build_begin = true; 
     238 
     239        (*CWorkflowGraph::mapHashFilterID_ptr)[filterhash] = this->filterID;  
     240        unique_filter_id = this->filterID; 
     241  
     242      } 
     243      // not first round 
     244      else  
     245      { 
     246        firstround = false; 
     247        unique_filter_id = (*CWorkflowGraph::mapHashFilterID_ptr)[filterhash]; 
     248        if(data[0]->src_filterID != unique_filter_id) 
     249        { 
     250          int edgeID = InvalidableObject::edgeIdGenerator++; 
     251          CWorkflowGraph::addEdge(edgeID, unique_filter_id, data[0]);  
     252          (*CWorkflowGraph::mapFilters_ptr_with_info)[data[0]->src_filterID].filter_filled = 0 ;  
     253          (*CWorkflowGraph::mapFilters_ptr_with_info)[unique_filter_id].expected_entry_nb ++; 
     254        } 
     255        if(data[1]->src_filterID != unique_filter_id) 
     256        {  
     257          int edgeID = InvalidableObject::edgeIdGenerator++; 
     258          CWorkflowGraph::addEdge(edgeID, unique_filter_id, data[1]);   
     259          (*CWorkflowGraph::mapFilters_ptr_with_info)[data[1]->src_filterID].filter_filled = 0 ; 
     260          (*CWorkflowGraph::mapFilters_ptr_with_info)[unique_filter_id].expected_entry_nb ++; 
     261        } 
     262         
     263      }   
     264    } 
     265 
     266    return std::make_tuple(building_graph, firstround, unique_filter_id); 
     267  } 
    47268 
    48269  CDataPacketPtr CFieldFieldArithmeticFilter::apply(std::vector<CDataPacketPtr> data) 
    … …  
    51272    packet->date = data[0]->date; 
    52273    packet->timestamp = data[0]->timestamp; 
     274 
     275    std::tuple<int, int, int> graph = buildGraph(data); 
     276 
     277    if(std::get<0>(graph)) packet->src_filterID = std::get<2>(graph); 
     278    if(std::get<0>(graph) && std::get<1>(graph)) packet->distance = data[0]->distance+1; 
     279    if(std::get<0>(graph) && !std::get<1>(graph)) packet->distance = data[0]->distance; 
     280     
     281    packet->field = this->field; 
     282     
    53283 
    54284    if (data[0]->status != CDataPacket::NO_ERROR) 
    … …  
    64294    return packet; 
    65295  } 
     296 
     297  StdString CScalarFieldArithmeticFilter::GetName(void)    { return StdString("CScalarFieldArithmeticFilter"); } 
     298  StdString CFieldScalarArithmeticFilter::GetName(void)    { return StdString("CFieldScalarArithmeticFilter"); } 
     299  StdString CFieldFieldArithmeticFilter::GetName(void)     { return StdString("CFieldFieldArithmeticFilter"); } 
     300 
     301 
    66302} // namespace xios 
Note: See TracChangeset for help on using the changeset viewer.