-/* ------------------------------------------------
-Copyright 2014 AT&T Intellectual Property
- Licensed under the Apache License, Version 2.0 (the "License");
- you may not use this file except in compliance with the License.
- You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
- ------------------------------------------- */
-
-#ifndef __HFTA_H
-#define __HFTA_H
-
-#include "gstypes.h"
-#include "host_tuple.h"
-#include "base_operator.h"
-#include <vector>
-#include <map>
-#include<math.h>
-#include "rdtsc.h"
-using namespace std;
-
-#define hfta_ullong_hashfunc(x) (((int)(*(x)))^((int)((*(x))>>32)))
-#define LLMIN(x,y) ((x)<(y)?(x):(y))
-#define LLMAX(x,y) ((x)<(y)?(y):(x))
-#define UMIN(x,y) ((x)<(y)?(x):(y))
-#define UMAX(x,y) ((x)<(y)?(y):(x))
-#define EQ(x,y) ((x)==(y))
-#define GEQ(x,y) ((x)>=(y))
-#define LEQ(x,y) ((x)<=(y))
-// Cast away temporality
-#define non_temporal(x)(x)
-// Access math libraries
-#define sqrt(x) sqrt(x)
-
-
-extern "C" {
-#include <lapp.h>
-#include <fta.h>
-#include <stdlib.h>
-#include <stdio.h>
-#include <schemaparser.h>
-}
-
-struct param_block {
- gs_int32_t block_length;
- void* data;
-};
-
-// forward declaration of operator_node
-struct operator_node;
-
-struct lfta_info {
- gs_schemahandle_t schema_handle;
- gs_sp_t schema;
- gs_sp_t fta_name;
-#ifdef PLAN_DAG
- list<operator_node*> parent_list;
- list<unsigned> out_channel_list;
-#else
- operator_node* parent;
- unsigned output_channel;
-#endif
- FTAID f;
-
- lfta_info() {
- schema_handle = -1;
- schema = NULL;
- #ifndef PLAN_DAG
- parent = NULL;
- output_channel = 0;
- #endif
- }
-
- ~lfta_info() {
- if (fta_name)
- free (fta_name);
- if (schema)
- free (schema);
- if (schema_handle >= 0)
- ftaschema_free(schema_handle);
- }
-};
-
-
-struct operator_node {
- base_operator* op;
-
-#ifdef PLAN_DAG
- list<operator_node*> parent_list;
- list<unsigned> out_channel_list;
-#else
- operator_node* parent;
-#endif
-
- operator_node* left_child;
- operator_node* right_child;
- lfta_info* left_lfta;
- lfta_info* right_lfta;
-
- list<host_tuple> input_queue;
-
- operator_node(base_operator* o) {
- op = o;
- #ifndef PLAN_DAG
- parent = NULL;
- #endif
- left_child = right_child = NULL;
- left_lfta = right_lfta = NULL;
- }
-
- ~operator_node() {
- delete op;
- }
-
- void set_left_child_node(operator_node* child) {
- left_child = child;
- if (child) {
- #ifdef PLAN_DAG
- child->parent_list.push_back(this);
- child->out_channel_list.push_back(0);
- #else
- child->parent = this;
- child->op->set_output_channel(0);
- #endif
- }
- }
-
- void set_right_child_node(operator_node* child) {
- right_child = child;
- if (child) {
- #ifdef PLAN_DAG
- child->parent_list.push_back(this);
- child->out_channel_list.push_back(1);
- #else
- child->parent = this;
- child->op->set_output_channel(1);
- #endif
- }
- }
-
- void set_left_lfta(lfta_info* l_lfta) {
- left_lfta = l_lfta;
- if (left_lfta) {
- #ifdef PLAN_DAG
- left_lfta->parent_list.push_back(this);
- left_lfta->out_channel_list.push_back(0);
- #else
- left_lfta->parent = this;
- left_lfta->output_channel = 0;
- #endif
- }
- }
-
- void set_right_lfta(lfta_info* r_lfta) {
- right_lfta = r_lfta;
- if (right_lfta) {
- #ifdef PLAN_DAG
- right_lfta->parent_list.push_back(this);
- right_lfta->out_channel_list.push_back(1);
- #else
- right_lfta->parent = this;
- right_lfta->output_channel = 1;
- #endif
- }
- }
-
-};
-
-
-
-int get_lfta_params(gs_int32_t sz, void * value, list<param_block>& lst);
-void finalize_tuple(host_tuple &tup);
-void finalize_tuple(host_tuple &tup);
-gs_retval_t UNOP_HFTA_free_fta (struct FTA *ftap, gs_uint32_t recursive);
-gs_retval_t UNOP_HFTA_control_fta (struct FTA *ftap, gs_int32_t command, gs_int32_t sz, void * value);
-gs_retval_t UNOP_HFTA_accept_packet (struct FTA *ftap, FTAID *ftaid, void * packet, gs_int32_t length);
-gs_retval_t UNOP_HFTA_clock_fta(struct FTA *ftap);
-
-gs_retval_t MULTOP_HFTA_free_fta (struct FTA *ftap, gs_uint32_t recursive);
-gs_retval_t MULTOP_HFTA_control_fta (struct FTA *ftap, gs_int32_t command, gs_int32_t sz, void * value);
-gs_retval_t MULTOP_HFTA_accept_packet (struct FTA *ftap, FTAID *ftaid, void * packet, gs_int32_t length);
-gs_retval_t MULTOP_HFTA_clock_fta(struct FTA *ftap);
-
-struct UNOP_HFTA {
- struct FTA _fta;
- base_operator* oper;
- FTAID f;
- bool failed;
- gs_schemahandle_t schema_handle;
-
- list<host_tuple> output_queue;
-
- // To create an hfta we will need all the parameters passed to alloc_fta by host library plus
- // lfta_name and an instance of the base_operator
- // We don't need to know the output schema as this information is already embeded
- // in create_output_tuple method of operators' functor.
- UNOP_HFTA(struct FTAID ftaid, FTAID child_ftaid, gs_int32_t command, gs_int32_t sz, void * value, base_operator* op,
- gs_csp_t fta_name, char* schema, gs_schemahandle_t sch_handle, bool fta_reusable, gs_uint32_t reuse_option) {
-
- failed = false;
- oper = op;
- f = child_ftaid;
- schema_handle = sch_handle;
-
- // assign streamid
- _fta.ftaid = ftaid;
- _fta.ftaid.streamid = (gs_p_t)this;
-
-#ifdef DEBUG
- fprintf(stderr,"Instantiate a FTA\n");
-#endif
- /* extract lfta param block from hfta param block */
- list<param_block> param_list;
- get_lfta_params(sz, value, param_list);
- param_block param = param_list.front();
-
-
- gs_uint32_t reuse_flag = 2;
- // we will try to create a new instance of child FTA only if it is parameterized
- if (param.block_length > 0 && (reuse_option == 0 || (!fta_reusable && reuse_option == 2))) {
- f.streamid = 0; // not interested in existing instances
- reuse_flag = 0;
- }
-
- if ((fta_alloc_instance(_fta.ftaid, &f,fta_name,schema,reuse_flag, FTA_COMMAND_LOAD_PARAMS,param.block_length,param.data))!=0) {
- fprintf(stderr,"HFTA::error:could instantiate a FTA");
- failed = true;
- return;
- }
-
- free(param.data);
- // set the operator's parameters
- if(oper->set_param_block(sz, (void*)value)) failed = true;;
-
-
- fprintf(stderr,"HFTA::Low level FTA (%s) instanciation done\n", fta_name);
- _fta.stream_subscribed_cnt = 1;
- _fta.stream_subscribed[0] = f;
-
- _fta.alloc_fta = NULL; // why should this be a part of the FTA (it is a factory function)
- _fta.free_fta = UNOP_HFTA_free_fta;
- _fta.control_fta = UNOP_HFTA_control_fta;
- _fta.accept_packet = UNOP_HFTA_accept_packet;
- _fta.clock_fta = UNOP_HFTA_clock_fta;
- }
-
- ~UNOP_HFTA() {
- delete oper; // free operators memory
-
- }
-
- int flush() {
- list<host_tuple> res;
- if (!oper->flush(res)) {
-
- if (!res.empty()) {
- // go through the list of returned tuples and finalyze them
- list<host_tuple>::iterator iter = res.begin();
- while (iter != res.end()) {
- host_tuple& tup = *iter;
-
- // finalize the tuple
- if (tup.tuple_size)
- finalize_tuple(tup);
- iter++;
- }
-
- // append returned list to output_queue
- output_queue.splice(output_queue.end(), res);
-
-
- // post tuples
- while (!output_queue.empty()) {
- host_tuple& tup = output_queue.front();
- #ifdef DEBUG
- fprintf(stderr, "HFTA::about to post tuple\n");
- #endif
- if (hfta_post_tuple(&_fta,tup.tuple_size,tup.data) == 0) {
- tup.free_tuple();
- output_queue.pop_front();
- } else
- break;
- }
- }
- }
-
- return 0;
- }
-
- bool init_failed(){return failed;}
-};
-
-gs_retval_t UNOP_HFTA_free_fta (struct FTA *fta, gs_uint32_t recursive) {
- UNOP_HFTA* ftap = (UNOP_HFTA*)fta; // deallocate the fta and call the destructor
- // will be called on program exit
-
- if (recursive)
- // free instance we are subscribed to
- fta_free_instance(gscpipc_getftaid(), ftap->f, recursive);
-
- delete ftap;
- return 0;
-}
-
-gs_retval_t UNOP_HFTA_control_fta (struct FTA *fta, gs_int32_t command, gs_int32_t sz, void * value) {
- UNOP_HFTA* ftap = (UNOP_HFTA*)fta;
-
- if (command == FTA_COMMAND_FLUSH) {
- // ask lfta to do the flush
- fta_control(gscpipc_getftaid(), ftap->f, FTA_COMMAND_FLUSH, sz, value);
- ftap->flush();
-
- } else if (command == FTA_COMMAND_LOAD_PARAMS) {
- // ask lfta to do the flush
- fta_control(gscpipc_getftaid(), ftap->f, FTA_COMMAND_FLUSH, sz, value);
- ftap->flush();
-
- /* extract lfta param block from hfta param block */
- list<param_block> param_list;
- get_lfta_params(sz, value, param_list);
- param_block param = param_list.front();
- // load new parameters into lfta
- fta_control(gscpipc_getftaid(), ftap->f, FTA_COMMAND_LOAD_PARAMS, param.block_length,param.data);
- free(param.data);
-
- // notify the operator about the change of parameter
- ftap->oper->set_param_block(sz, (void*)value);
-
- } else if (command == FTA_COMMAND_GET_TEMP_STATUS) {
- // we no longer use temp_status commands
- // hearbeat mechanism is used instead
- }
- return 0;
-}
-
-gs_retval_t UNOP_HFTA_accept_packet (struct FTA *fta, FTAID *ftaid, void * packet, gs_int32_t length) {
- UNOP_HFTA* ftap = (UNOP_HFTA*)fta;
-#ifdef DEBUG
- fprintf(stderr, "HFTA::accepted packet\n");
-#endif
- if (!length) /* ignore null tuples */
- return 0;
-
- host_tuple temp;
- temp.tuple_size = length;
- temp.data = packet;
- temp.channel = 0;
- temp.heap_resident = false;
-
- // pass the tuple to operator
- list<host_tuple> res;
- int ret;
- fta_stat* tup_trace = NULL;
- gs_uint32_t tup_trace_sz = 0;
- gs_uint64_t trace_id = 0;
- bool temp_tuple_received = false;
-
-
- // if the tuple is temporal we need to extract the hearbeat payload
- if (ftaschema_is_temporal_tuple(ftap->schema_handle, packet)) {
- temp_tuple_received = true;
- if (ftaschema_get_trace(ftap->schema_handle, packet, length, &trace_id, &tup_trace_sz, &tup_trace))
- fprintf(stderr, "HFTA error: received temporal status tuple with no trace\n");
- }
-
- if (ftaschema_is_eof_tuple(ftap->schema_handle, packet)) {
- /* perform a flush */
- ftap->flush();
-
- /* post eof_tuple to a parent */
- host_tuple eof_tuple;
- ftap->oper->get_temp_status(eof_tuple);
-
- /* last byte of the tuple specifies the tuple type
- * set it to EOF_TUPLE
- */
- *((char*)eof_tuple.data + (eof_tuple.tuple_size - sizeof(char))) = EOF_TUPLE;
- hfta_post_tuple(fta,eof_tuple.tuple_size,eof_tuple.data);
-
- return 0;
- }
-
- ret = ftap->oper->accept_tuple(temp, res);
-
- // go through the list of returned tuples and finalyze them
- list<host_tuple>::iterator iter = res.begin();
- while (iter != res.end()) {
- host_tuple& tup = *iter;
-
- // finalize the tuple
- if (tup.tuple_size)
- finalize_tuple(tup);
- iter++;
- }
-
- // if we received temporal tuple, last tuple of the result must be temporal too
- // we need to extend the trace by adding ftaid and send a hearbeat to clearinghouse
- if (temp_tuple_received) {
- fta_stat stats;
- host_tuple& temp_tup = res.back();
-
-
- int new_tuple_size = temp_tup.tuple_size + sizeof(gs_uint64_t) + (tup_trace_sz + 1)* sizeof(fta_stat);
- char* new_data = (char*)malloc(new_tuple_size);
- memcpy(new_data, temp_tup.data, temp_tup.tuple_size);
- memcpy(new_data + temp_tup.tuple_size, &trace_id, sizeof(gs_uint64_t));
- memcpy(new_data + temp_tup.tuple_size + sizeof(gs_uint64_t), (char*)tup_trace, tup_trace_sz * sizeof(fta_stat));
- memcpy(new_data + temp_tup.tuple_size + sizeof(gs_uint64_t) + tup_trace_sz * sizeof(fta_stat), (char*)ftaid, sizeof(FTAID));
-
- memset((char*)&stats, 0, sizeof(fta_stat));
- memcpy(new_data + temp_tup.tuple_size + sizeof(gs_uint64_t) + tup_trace_sz * sizeof(fta_stat), (char*)&stats, sizeof(fta_stat));
-
- // Send a hearbeat message to clearinghouse.
- fta_heartbeat(ftap->_fta.ftaid, trace_id, tup_trace_sz + 1, (fta_stat *)((char*)new_data + temp_tup.tuple_size + sizeof(gs_uint64_t)));
-
- temp_tup.free_tuple();
- temp_tup.data = new_data;
- temp_tup.tuple_size = new_tuple_size;
- }
-
- // append returned list to output_queue
- ftap->output_queue.splice(ftap->output_queue.end(), res);
-
- // post tuples
- while (!ftap->output_queue.empty()) {
- host_tuple& tup = ftap->output_queue.front();
- #ifdef DEBUG
- fprintf(stderr, "HFTA::about to post tuple\n");
- #endif
- if (hfta_post_tuple(fta,tup.tuple_size,tup.data) == 0) {
- tup.free_tuple();
- ftap->output_queue.pop_front();
- } else
- break;
- }
-
- return 1;
-}
-
-gs_retval_t UNOP_HFTA_clock_fta(struct FTA *ftap) {
-
- // Send a hearbeat message to clearinghouse.to indicate we are alive
- fta_heartbeat(ftap->ftaid, 0, 0, 0);
-
- return 0;
-}
-
-
-struct MULTOP_HFTA {
- struct FTA _fta;
- gs_csp_t fta_name;
- gs_schemahandle_t schema_handle;
- operator_node* root;
- vector<operator_node*> sorted_nodes;
- int num_operators;
- list<lfta_info*> *lfta_list;
- /* number of eof tuples we received so far
- * receiving eof tuples from every source fta will cause a flush
- */
- int num_eof_tuples;
-
- bool failed;
- bool reusable;
-
- list<host_tuple> output_queue;
-
- // Runtime stats
- gs_uint32_t in_tuple_cnt;
- gs_uint32_t out_tuple_cnt;
- gs_uint32_t out_tuple_sz;
- gs_uint64_t cycle_cnt;
-
- gs_uint64_t trace_id;
-
- // memory occupied by output queue
- gs_uint32_t output_queue_mem;
-
-
- // To create an hfta we will need all the parameters passed to alloc_fta by host library plus
- // lfta_name and an instance of the base_operator. We don't need to know the schema for lfta,
- // as the schema handle is already passed during operator creation time.
- // We also don't need to know the output schema as this information is already embeded
- // in create_output_tuple method of operators' functor.
-
-
-
- MULTOP_HFTA(struct FTAID ftaid, gs_csp_t name, gs_int32_t command, gs_int32_t sz, void * value, gs_schemahandle_t sch_handle, operator_node* node,
- list<lfta_info*> *lftas, bool fta_reusable, gs_uint32_t reuse_option) {
-
- fta_name = name;
- failed = false;
-
- root = node;
- lfta_list = lftas;
-
- // assign streamid
- _fta.ftaid = ftaid;
- _fta.ftaid.streamid = (gs_p_t)this;
-
- schema_handle = sch_handle;
-
- output_queue_mem = 0;
-
- // topologically sort the operators in the tree (or DAG)
- // for DAG we make sure we add the node to the sorted list only once
- operator_node* current_node;
- map<operator_node*, int> node_map;
- vector<operator_node*> node_list;
-
- int i = 0;
- node_list.push_back(root);
- node_map[root] = 0;
-
- num_operators = 1;
-
- while (i < node_list.size()) {
- current_node = node_list[i];
- if (current_node->left_child && node_map.find(current_node->left_child) == node_map.end()) {
- node_map[current_node->left_child] = num_operators++;
- node_list.push_back(current_node->left_child);
- }
- if (current_node->right_child && node_map.find(current_node->right_child) == node_map.end()) {
- node_map[current_node->right_child] = num_operators++;
- node_list.push_back(current_node->right_child);
- }
- i++;
- }
- num_operators = i;
-
- // build adjacency lists for query DAG
- list<int>* adj_lists = new list<int>[num_operators];
- bool* leaf_flags = new bool[num_operators];
- memset(leaf_flags, 0, num_operators * sizeof(bool));
- for (i = 0; i < num_operators; ++i) {
- current_node = node_list[i];
- if (current_node->left_child) {
- adj_lists[i].push_back(node_map[current_node->left_child]);
- }
- if (current_node->right_child && current_node->left_child != current_node->right_child) {
- adj_lists[i].push_back(node_map[current_node->right_child]);
- }
- }
-
- // run topolofical sort
- bool leaf_found = true;
- while (leaf_found) {
- leaf_found = false;
- // add all leafs to sorted_nodes
- for (i = 0; i < num_operators; ++i) {
- if (!leaf_flags[i] && adj_lists[i].empty()) {
- leaf_flags[i] = true;
- sorted_nodes.push_back(node_list[i]);
- leaf_found = true;
-
- // remove the node from its parents adjecency lists
- for (int j = 0; j < num_operators; ++j) {
- list<int>::iterator iter;
- for (iter = adj_lists[j].begin(); iter != adj_lists[j].end(); iter++) {
- if (*iter == i) {
- adj_lists[j].erase(iter);
- break;
- }
- }
- }
- }
- }
- }
-
- delete[] adj_lists;
- delete[] leaf_flags;
-
- // set the parameter block for every operator in tree
- for (i = 0; i < num_operators; ++i)
- if(sorted_nodes[i]->op->set_param_block(sz, (void*)value)) failed = true;
-
-#ifdef DEBUG
- fprintf(stderr,"Instantiate FTAs\n");
-#endif
- /* extract lfta param block from hfta param block */
-// NOTE: param_list must line up with lfta_list
- list<param_block> param_list;
- get_lfta_params(sz, value, param_list);
- list<param_block>::iterator iter1;
- list<lfta_info*>::iterator iter2 = lfta_list->begin();
-
- for (iter1 = param_list.begin(), i = 0; iter1 != param_list.end(); iter1++, iter2++, i++) {
- lfta_info* inf = *iter2;
-
- #ifdef DEBUG
- fprintf(stderr,"Instantiate a FTA\n");
- #endif
-
- gs_uint32_t reuse_flag = 2;
-
- // we will try to create a new instance of child FTA only if it is parameterized
- if ((*iter1).block_length > 0 && (reuse_option == 0 || (!fta_reusable && reuse_option == 2))) {
- (*iter2)->f.streamid = 0; // not interested in existing instances
- reuse_flag = 0;
- }
- if (fta_alloc_instance(_fta.ftaid, &(*iter2)->f,(*iter2)->fta_name, (*iter2)->schema, reuse_flag, FTA_COMMAND_LOAD_PARAMS,(*iter1).block_length,(*iter1).data)!=0) {
- fprintf(stderr,"HFTA::error:could instantiate a FTA");
- failed = true;
- return;
- }
-
- free((*iter1).data);
-
- //fprintf(stderr,"HFTA::Low level FTA instanciation done\n");
-
- _fta.stream_subscribed[i]=(*iter2)->f;
- }
- _fta.stream_subscribed_cnt = i;
-
- num_eof_tuples = 0;
-
- _fta.alloc_fta = NULL; // why should this be a part of the FTA (it is a factory function)
- _fta.free_fta = MULTOP_HFTA_free_fta;
- _fta.control_fta = MULTOP_HFTA_control_fta;
- _fta.accept_packet = MULTOP_HFTA_accept_packet;
- _fta.clock_fta = MULTOP_HFTA_clock_fta;
-
- // init runtime stats
- in_tuple_cnt = 0;
- out_tuple_cnt = 0;
- out_tuple_sz = 0;
- cycle_cnt = 0;
-
- }
-
- ~MULTOP_HFTA() {
-
- list<lfta_info*>::iterator iter;
- int i = 0;
-
- for (iter = lfta_list->begin(); i < _fta.stream_subscribed_cnt; iter++, i++) {
- delete *iter;
- }
-
- delete root; // free operators memory
- delete lfta_list;
-
-
- }
-
- int flush() {
-
- list<host_tuple> res;
-
- // go through the list of operators in topological order
- // and flush them
- list<host_tuple>::iterator iter;
- list<host_tuple> temp_output_queue;
-
- for (int i = 0; i < num_operators; ++i) {
- operator_node* node = sorted_nodes[i];
-
-#ifdef PLAN_DAG
- list<host_tuple>& current_output_queue = (i < (num_operators - 1)) ? temp_output_queue : res;
-#else
- // for trees we can put output tuples directly into parent's input buffer
- list<host_tuple>& current_output_queue = (i < (num_operators - 1)) ? node->parent->input_queue : res;
-#endif
- // consume tuples waiting in your queue
- for (iter = node->input_queue.begin(); iter != node->input_queue.end(); iter++) {
- node->op->accept_tuple(*(iter), current_output_queue);
- }
- node->op->flush(current_output_queue);
- node->input_queue.clear();
-
-#ifdef PLAN_DAG
- // copy the tuples from output queue into input queues of all parents
- list<operator_node*>::iterator node_iter;
-
- if (!node->parent_list.empty()) {
- // append the content of the output queue to parent input queue
-
- for (iter = temp_output_queue.begin(); iter != temp_output_queue.end(); iter++) {
- int* ref_cnt = 0;
- if (node->parent_list.size() > 1) {
- ref_cnt = (int*)malloc(sizeof(int));
- *ref_cnt = node->parent_list.size() - 1;
- }
-
- for (node_iter = node->parent_list.begin(); node_iter != node->parent_list.end(); node_iter++) {
- (*iter).ref_cnt = ref_cnt;
- (*node_iter)->input_queue.push_back(*iter);
- }
- }
- }
-#endif
- }
-
- if (!res.empty()) {
- // go through the list of returned tuples and finalyze them
- list<host_tuple>::iterator iter = res.begin();
- while (iter != res.end()) {
- host_tuple& tup = *iter;
-
- // finalize the tuple
- if (tup.tuple_size)
- finalize_tuple(tup);
-
- output_queue_mem += tup.tuple_size;
- iter++;
- }
-
- // append returned list to output_queue
- output_queue.splice(output_queue.end(), res);
-
-
- // post tuples
- while (!output_queue.empty()) {
- host_tuple& tup = output_queue.front();
- #ifdef DEBUG
- fprintf(stderr, "HFTA::about to post tuple\n");
- #endif
- if (hfta_post_tuple(&_fta,tup.tuple_size,tup.data) == 0) {
- output_queue_mem -= tup.tuple_size;
- tup.free_tuple();
- output_queue.pop_front();
- } else
- break;
- }
- }
-
- return 0;
- }
-
- bool init_failed(){return failed;}
-};
-
-
-gs_retval_t MULTOP_HFTA_free_fta (struct FTA *fta, gs_uint32_t recursive) {
- MULTOP_HFTA* ftap = (MULTOP_HFTA*)fta; // deallocate the fta and call the destructor
- // will be called on program exit
-
- if (recursive) {
- // free instance we are subscribed to
- list<lfta_info*>::iterator iter;
- int i = 0;
-
- for (iter = ftap->lfta_list->begin(); i < fta->stream_subscribed_cnt; iter++, i++) {
- fta_free_instance(gscpipc_getftaid(), (*iter)->f, recursive);
- }
- }
-
- delete ftap;
- return 0;
-}
-
-gs_retval_t MULTOP_HFTA_control_fta (struct FTA *fta, gs_int32_t command, gs_int32_t sz, void * value) {
- MULTOP_HFTA* ftap = (MULTOP_HFTA*)fta;
-
- if (command == FTA_COMMAND_FLUSH) {
-
- // ask lftas to do the flush
- list<lfta_info*>::iterator iter;
- for (iter = ftap->lfta_list->begin(); iter != ftap->lfta_list->end(); iter++) {
- fta_control(gscpipc_getftaid(), (*iter)->f, FTA_COMMAND_FLUSH, sz, value);
- }
- // flush hfta operators
- ftap->flush();
-
- } else if (command == FTA_COMMAND_LOAD_PARAMS) {
-
- list<param_block> param_list;
- get_lfta_params(sz, value, param_list);
-
- // ask lftas to do the flush and set new parameters
- list<lfta_info*>::iterator iter;
- list<param_block>::iterator iter2;
- for (iter = ftap->lfta_list->begin(), iter2 = param_list.begin(); iter != ftap->lfta_list->end(); iter++, iter2++) {
- fta_control(gscpipc_getftaid(), (*iter)->f, FTA_COMMAND_FLUSH, 0, NULL);
- fta_control(gscpipc_getftaid(), (*iter)->f, FTA_COMMAND_LOAD_PARAMS, (*iter2).block_length,(*iter2).data);
- free((*iter2).data);
- }
- // flush hfta operators
- ftap->flush();
-
- // set the new parameter block for every operator in tree
- for (int i = 0; i < ftap->num_operators; ++i)
- ftap->sorted_nodes[i]->op->set_param_block(sz, (void*)value);
-
- } else if (command == FTA_COMMAND_GET_TEMP_STATUS) {
- // we no longer use temp_status commands
- // hearbeat mechanism is used instead
- }
- return 0;
-}
-
-gs_retval_t MULTOP_HFTA_accept_packet (struct FTA *fta, FTAID *ftaid, void * packet, gs_int32_t length) {
- MULTOP_HFTA* ftap = (MULTOP_HFTA*)fta;
-
- gs_uint64_t start_cycle = rdtsc();
-#ifdef DEBUG
- fprintf(stderr, "HFTA::accepted packet\n");
-#endif
- if (!length) /* ignore null tuples */
- return 0;
-
- ftap->in_tuple_cnt++;
-
- host_tuple temp;
- temp.tuple_size = length;
- temp.data = packet;
- temp.channel = 0;
- temp.heap_resident = false;
-
-// fprintf(stderr,"created temp, channel=%d, resident=%d\n",temp.channel, (int)temp.heap_resident);
-
- // find from which lfta the tuple came
- list<lfta_info*>::iterator iter;
- lfta_info* inf = NULL;
- int i;
-
- for (i = 0, iter = ftap->lfta_list->begin(); i < ftap->_fta.stream_subscribed_cnt; iter++, i++) {
- if (ftap->_fta.stream_subscribed[i].ip == ftaid->ip &&
- ftap->_fta.stream_subscribed[i].port == ftaid->port &&
- ftap->_fta.stream_subscribed[i].index == ftaid->index &&
- ftap->_fta.stream_subscribed[i].streamid == ftaid->streamid) {
- inf = *iter;
- break;
- }
- }
-
- if (!inf) {
- fprintf(stderr,"HFTA::error:received tuple from unknown stream\n");
- exit(1);
- }
-
- // route tuple through operator chain
- list<host_tuple> result;
- host_tuple tup;
- int ret;
- #ifndef PLAN_DAG
- temp.channel = inf->output_channel;
- #endif
- operator_node* current_node = NULL, *child = NULL;
- list<host_tuple> temp_output_queue;
-
-
- fta_stat* tup_trace = NULL;
- gs_uint32_t tup_trace_sz = 0;
- gs_uint64_t trace_id = 0;
- bool temp_tuple_received = false;
-
- // if the tuple is temporal we need to extract the heartbeat payload
- if (ftaschema_is_temporal_tuple(inf->schema_handle, packet)) {
- temp_tuple_received = true;
- if (ftaschema_get_trace(inf->schema_handle, packet, length, &trace_id, &tup_trace_sz, &tup_trace))
- fprintf(stderr, "HFTA error: received temporal status tuple with no trace\n");
- }
-
- if (ftaschema_is_eof_tuple(inf->schema_handle, packet)) {
-
- if (++ftap->num_eof_tuples < ftap->lfta_list->size())
- return 0;
-
- ftap->num_eof_tuples = 0;
-
- /* perform a flush */
- ftap->flush();
-
- /* post eof_tuple to a parent */
- host_tuple eof_tuple;
- ftap->sorted_nodes[ftap->num_operators - 1]->op->get_temp_status(eof_tuple);
-
- /* last byte of the tuple specify the tuple type
- * set it to EOF_TUPLE
- */
- *((char*)eof_tuple.data + (eof_tuple.tuple_size - sizeof(char))) = EOF_TUPLE;
- hfta_post_tuple(fta,eof_tuple.tuple_size,eof_tuple.data);
- ftap->out_tuple_cnt++;
- ftap->out_tuple_sz+=eof_tuple.tuple_size;
-
- return 0;
- }
-
- list<host_tuple>::iterator iter2;
-
-#ifdef PLAN_DAG
-
- // push tuple to all parent operators of the lfta
- list<operator_node*>::iterator node_iter;
- list<unsigned>::iterator chan_iter;
- for (node_iter = inf->parent_list.begin(), chan_iter = inf->out_channel_list.begin(); node_iter != inf->parent_list.end(); node_iter++, chan_iter++) {
- temp.channel = *chan_iter;
- (*node_iter)->input_queue.push_back(temp);
- }
-
- for (i = 0; i < ftap->num_operators; ++i) {
-
- operator_node* node = ftap->sorted_nodes[i];
- list<host_tuple>& current_output_queue = (i < (ftap->num_operators - 1)) ? temp_output_queue : result;
-
- // consume tuples waiting in your queue
- for (iter2 = node->input_queue.begin(); iter2 != node->input_queue.end(); iter2++) {
- node->op->accept_tuple(*(iter2), current_output_queue);
- }
- node->input_queue.clear();
-
- // copy the tuples from output queue into input queues of all parents
-
- if (!node->parent_list.empty()) {
-
- // append the content of the output queue to parent input queue
- for (iter2 = temp_output_queue.begin(); iter2 != temp_output_queue.end(); iter2++) {
-
- int* ref_cnt = 0;
- if (node->parent_list.size() > 1) {
- ref_cnt = (int*)malloc(sizeof(int));
- *ref_cnt = node->parent_list.size() - 1;
- }
-
- for (node_iter = node->parent_list.begin(), chan_iter = node->out_channel_list.begin(); node_iter != node->parent_list.end(); node_iter++, chan_iter++) {
- (*iter2).ref_cnt = ref_cnt;
- (*iter2).channel = *chan_iter;
- (*node_iter)->input_queue.push_back(*iter2);
- }
- }
- }
- temp_output_queue.clear();
- }
-#else
- current_node = inf->parent;
-
-// fprintf(stderr,"Pushing temp, channel=%d, resident=%d\n",temp.channel, (int)temp.heap_resident);
- current_node->input_queue.push_back(temp);
-
- do {
-//fprintf(stderr,"Routing tuple, current node is %d, parent is %d\n",current_node,current_node->parent);
- list<host_tuple>& current_output_queue = (current_node->parent) ? current_node->parent->input_queue : result;
-
- // consume tuples waiting in your queue
- for (iter2 = current_node->input_queue.begin(); iter2 != current_node->input_queue.end(); iter2++) {
- current_node->op->accept_tuple((*iter2),current_output_queue);
- }
-// All consumed, delete them
- current_node->input_queue.clear();
- current_node = current_node->parent;
-
- } while (current_node);
-#endif
-
-
- host_tuple temp_tup;
-
- bool no_temp_tuple = false;
-
- // if we received temporal tuple, last tuple of the result must be temporal too
- // we need to extend the trace by adding ftaid and send a hearbeat to clearinghouse
- if (temp_tuple_received) {
-
- if (result.empty()) {
- no_temp_tuple = true;
-
- } else {
- fta_stat stats;
- temp_tup = result.back();
- finalize_tuple(temp_tup);
-
- int new_tuple_size = temp_tup.tuple_size + sizeof(gs_uint64_t) + (tup_trace_sz + 1)* sizeof(fta_stat);
- char* new_data = (char*)malloc(new_tuple_size);
- memcpy(new_data, temp_tup.data, temp_tup.tuple_size);
- memcpy(new_data + temp_tup.tuple_size, &trace_id, sizeof(gs_uint64_t));
- memcpy(new_data + temp_tup.tuple_size + sizeof(gs_uint64_t), (char*)tup_trace, tup_trace_sz * sizeof(fta_stat));
-
-
- memset((char*)&stats, 0, sizeof(fta_stat));
- stats.ftaid = fta->ftaid;
- stats.in_tuple_cnt = ftap->in_tuple_cnt;
- stats.out_tuple_cnt = ftap->out_tuple_cnt;
- stats.out_tuple_sz = ftap->out_tuple_sz;
- stats.cycle_cnt = ftap->cycle_cnt;
- memcpy(new_data + temp_tup.tuple_size + sizeof(gs_uint64_t) + tup_trace_sz * sizeof(fta_stat), (char*)&stats, sizeof(fta_stat));
-
- // Send a hearbeat message to clearinghouse.
- fta_heartbeat(fta->ftaid, trace_id, tup_trace_sz + 1, (fta_stat *)((char*)new_data + temp_tup.tuple_size + sizeof(gs_uint64_t)));
-
- // reset the stats
- ftap->in_tuple_cnt = 0;
- ftap->out_tuple_cnt = 0;
- ftap->out_tuple_sz = 0;
- ftap->cycle_cnt = 0;
-
- free(temp_tup.data);
- temp_tup.data = new_data;
- temp_tup.tuple_size = new_tuple_size;
- result.pop_back();
- }
- }
-
- // go through the list of returned tuples and finalyze them
- // since we can produce multiple temporal tuples in DAG plans
- // we can drop all of them except the last one
- iter2 = result.begin();
- while(iter2 != result.end()) {
- host_tuple tup = *iter2;
-
- // finalize the tuple
- if (tup.tuple_size) {
- finalize_tuple(tup);
-
- #ifdef PLAN_DAG
- if (ftaschema_is_temporal_tuple(ftap->schema_handle, tup.data))
- tup.free_tuple();
- else
- #endif
- {
- ftap->output_queue.push_back(tup);
- ftap->output_queue_mem += tup.tuple_size;
- }
-
- }
- iter2++;
- }
-
- // append returned list to output_queue
- // ftap->output_queue.splice(ftap->output_queue.end(), result);
-
- if (temp_tuple_received && !no_temp_tuple) {
- ftap->output_queue.push_back(temp_tup);
- ftap->output_queue_mem += temp_tup.tuple_size;
- }
-
- // post tuples
- while (!ftap->output_queue.empty()) {
- host_tuple& tup = ftap->output_queue.front();
- #ifdef DEBUG
- fprintf(stderr, "HFTA::about to post tuple\n");
- #endif
- if (hfta_post_tuple(fta,tup.tuple_size,tup.data) == 0) {
- ftap->out_tuple_cnt++;
- ftap->out_tuple_sz+=tup.tuple_size;
- ftap->output_queue_mem -= tup.tuple_size;
- tup.free_tuple();
- ftap->output_queue.pop_front();
- } else
- break;
- }
-
- ftap->cycle_cnt += rdtsc() - start_cycle;
-
- return 1;
-}
-
-gs_retval_t MULTOP_HFTA_clock_fta(struct FTA *fta) {
- MULTOP_HFTA* ftap = (MULTOP_HFTA*)fta;
-
-#ifdef HFTA_PROFILE
- /* Print stats */
- fprintf(stderr, "FTA = %s|", ftap->fta_name);
- fprintf(stderr, "in_tuple_cnt = %u|", ftap->in_tuple_cnt);
- fprintf(stderr, "out_tuple_cnt = %u|", ftap->out_tuple_cnt);
- fprintf(stderr, "out_tuple_sz = %u|", ftap->out_tuple_sz);
- fprintf(stderr, "cycle_cnt = %llu|", ftap->cycle_cnt);
-
-
- fprintf(stderr, "mem_footprint %s = %d", ftap->sorted_nodes[0]->op->get_name(), ftap->sorted_nodes[0]->op->get_mem_footprint());
- unsigned int total_mem = ftap->sorted_nodes[0]->op->get_mem_footprint();
-
- for (int i = 1; i < ftap->num_operators; ++i) {
- operator_node* node = ftap->sorted_nodes[i];
- fprintf(stderr, ",%s = %d", node->op->get_name(), node->op->get_mem_footprint());
- total_mem += node->op->get_mem_footprint();
- }
- fprintf(stderr, ", total = %d|", total_mem );
- fprintf(stderr, "output_buffer_size = %d\n", ftap->output_queue_mem );
-#endif
-
- fta_stat stats;
- memset((char*)&stats, 0, sizeof(fta_stat));
- stats.ftaid = fta->ftaid;
- stats.in_tuple_cnt = ftap->in_tuple_cnt;
- stats.out_tuple_cnt = ftap->out_tuple_cnt;
- stats.out_tuple_sz = ftap->out_tuple_sz;
- stats.cycle_cnt = ftap->cycle_cnt;
-
- // Send a hearbeat message to clearinghouse.
- fta_heartbeat(fta->ftaid, ftap->trace_id++, 1, &stats);
-
- // resets runtime stats
- ftap->in_tuple_cnt = 0;
- ftap->out_tuple_cnt = 0;
- ftap->out_tuple_sz = 0;
- ftap->cycle_cnt = 0;
-
- return 0;
-}
-
-
-#endif // __HFTA_H
-
+/* ------------------------------------------------\r
+Copyright 2014 AT&T Intellectual Property\r
+ Licensed under the Apache License, Version 2.0 (the "License");\r
+ you may not use this file except in compliance with the License.\r
+ You may obtain a copy of the License at\r
+\r
+ http://www.apache.org/licenses/LICENSE-2.0\r
+\r
+ Unless required by applicable law or agreed to in writing, software\r
+ distributed under the License is distributed on an "AS IS" BASIS,\r
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
+ See the License for the specific language governing permissions and\r
+ limitations under the License.\r
+ ------------------------------------------- */\r
+\r
+#ifndef __HFTA_H\r
+#define __HFTA_H\r
+\r
+#include "gstypes.h"\r
+#include "host_tuple.h"\r
+#include "base_operator.h"\r
+#include <vector>\r
+#include <map>\r
+#include<math.h>\r
+#include "rdtsc.h"\r
+using namespace std;\r
+\r
+#define hfta_ullong_hashfunc(x) (((int)(*(x)))^((int)((*(x))>>32)))\r
+#define LLMIN(x,y) ((x)<(y)?(x):(y))\r
+#define LLMAX(x,y) ((x)<(y)?(y):(x))\r
+#define UMIN(x,y) ((x)<(y)?(x):(y))\r
+#define UMAX(x,y) ((x)<(y)?(y):(x))\r
+#define EQ(x,y) ((x)==(y))\r
+#define GEQ(x,y) ((x)>=(y))\r
+#define LEQ(x,y) ((x)<=(y))\r
+// Cast away temporality\r
+#define non_temporal(x)(x)\r
+// Access math libraries\r
+#define sqrt(x) sqrt(x)\r
+\r
+\r
+extern "C" {\r
+#include <lapp.h>\r
+#include <fta.h>\r
+#include <stdlib.h>\r
+#include <stdio.h>\r
+#include <schemaparser.h>\r
+}\r
+\r
+struct param_block {\r
+ gs_int32_t block_length;\r
+ void* data;\r
+};\r
+\r
+// forward declaration of operator_node\r
+struct operator_node;\r
+\r
+struct lfta_info {\r
+ gs_schemahandle_t schema_handle;\r
+ gs_sp_t schema;\r
+ gs_sp_t fta_name;\r
+#ifdef PLAN_DAG\r
+ list<operator_node*> parent_list;\r
+ list<unsigned> out_channel_list;\r
+#else\r
+ operator_node* parent;\r
+ unsigned output_channel;\r
+#endif\r
+ FTAID f;\r
+\r
+ lfta_info() {\r
+ schema_handle = -1;\r
+ schema = NULL;\r
+ #ifndef PLAN_DAG\r
+ parent = NULL;\r
+ output_channel = 0;\r
+ #endif\r
+ }\r
+\r
+ ~lfta_info() {\r
+ if (fta_name)\r
+ free (fta_name);\r
+ if (schema)\r
+ free (schema);\r
+ if (schema_handle >= 0)\r
+ ftaschema_free(schema_handle);\r
+ }\r
+};\r
+\r
+\r
+struct operator_node {\r
+ base_operator* op;\r
+\r
+#ifdef PLAN_DAG\r
+ list<operator_node*> parent_list;\r
+ list<unsigned> out_channel_list;\r
+#else\r
+ operator_node* parent;\r
+#endif\r
+\r
+ operator_node* left_child;\r
+ operator_node* right_child;\r
+ lfta_info* left_lfta;\r
+ lfta_info* right_lfta;\r
+\r
+ list<host_tuple> input_queue;\r
+\r
+ operator_node(base_operator* o) {\r
+ op = o;\r
+ #ifndef PLAN_DAG\r
+ parent = NULL;\r
+ #endif\r
+ left_child = right_child = NULL;\r
+ left_lfta = right_lfta = NULL;\r
+ }\r
+\r
+ ~operator_node() {\r
+ delete op;\r
+ }\r
+\r
+ void set_left_child_node(operator_node* child) {\r
+ left_child = child;\r
+ if (child) {\r
+ #ifdef PLAN_DAG\r
+ child->parent_list.push_back(this);\r
+ child->out_channel_list.push_back(0);\r
+ #else\r
+ child->parent = this;\r
+ child->op->set_output_channel(0);\r
+ #endif\r
+ }\r
+ }\r
+\r
+ void set_right_child_node(operator_node* child) {\r
+ right_child = child;\r
+ if (child) {\r
+ #ifdef PLAN_DAG\r
+ child->parent_list.push_back(this);\r
+ child->out_channel_list.push_back(1);\r
+ #else\r
+ child->parent = this;\r
+ child->op->set_output_channel(1);\r
+ #endif\r
+ }\r
+ }\r
+\r
+ void set_left_lfta(lfta_info* l_lfta) {\r
+ left_lfta = l_lfta;\r
+ if (left_lfta) {\r
+ #ifdef PLAN_DAG\r
+ left_lfta->parent_list.push_back(this);\r
+ left_lfta->out_channel_list.push_back(0);\r
+ #else\r
+ left_lfta->parent = this;\r
+ left_lfta->output_channel = 0;\r
+ #endif\r
+ }\r
+ }\r
+\r
+ void set_right_lfta(lfta_info* r_lfta) {\r
+ right_lfta = r_lfta;\r
+ if (right_lfta) {\r
+ #ifdef PLAN_DAG\r
+ right_lfta->parent_list.push_back(this);\r
+ right_lfta->out_channel_list.push_back(1);\r
+ #else\r
+ right_lfta->parent = this;\r
+ right_lfta->output_channel = 1;\r
+ #endif\r
+ }\r
+ }\r
+\r
+};\r
+\r
+\r
+\r
+int get_lfta_params(gs_int32_t sz, void * value, list<param_block>& lst);\r
+void finalize_tuple(host_tuple &tup);\r
+void finalize_tuple(host_tuple &tup);\r
+gs_retval_t UNOP_HFTA_free_fta (struct FTA *ftap, gs_uint32_t recursive);\r
+gs_retval_t UNOP_HFTA_control_fta (struct FTA *ftap, gs_int32_t command, gs_int32_t sz, void * value);\r
+gs_retval_t UNOP_HFTA_accept_packet (struct FTA *ftap, FTAID *ftaid, void * packet, gs_int32_t length);\r
+gs_retval_t UNOP_HFTA_clock_fta(struct FTA *ftap);\r
+\r
+gs_retval_t MULTOP_HFTA_free_fta (struct FTA *ftap, gs_uint32_t recursive);\r
+gs_retval_t MULTOP_HFTA_control_fta (struct FTA *ftap, gs_int32_t command, gs_int32_t sz, void * value);\r
+gs_retval_t MULTOP_HFTA_accept_packet (struct FTA *ftap, FTAID *ftaid, void * packet, gs_int32_t length);\r
+gs_retval_t MULTOP_HFTA_clock_fta(struct FTA *ftap);\r
+\r
+struct UNOP_HFTA {\r
+ struct FTA _fta;\r
+ base_operator* oper;\r
+ FTAID f;\r
+ bool failed;\r
+ gs_schemahandle_t schema_handle;\r
+\r
+ list<host_tuple> output_queue;\r
+\r
+ // To create an hfta we will need all the parameters passed to alloc_fta by host library plus\r
+ // lfta_name and an instance of the base_operator\r
+ // We don't need to know the output schema as this information is already embeded\r
+ // in create_output_tuple method of operators' functor.\r
+ UNOP_HFTA(struct FTAID ftaid, FTAID child_ftaid, gs_int32_t command, gs_int32_t sz, void * value, base_operator* op,\r
+ gs_csp_t fta_name, char* schema, gs_schemahandle_t sch_handle, bool fta_reusable, gs_uint32_t reuse_option) {\r
+\r
+ failed = false;\r
+ oper = op;\r
+ f = child_ftaid;\r
+ schema_handle = sch_handle;\r
+\r
+ // assign streamid\r
+ _fta.ftaid = ftaid;\r
+ _fta.ftaid.streamid = (gs_p_t)this;\r
+\r
+#ifdef DEBUG\r
+ fprintf(stderr,"Instantiate a FTA\n");\r
+#endif\r
+ /* extract lfta param block from hfta param block */\r
+ list<param_block> param_list;\r
+ get_lfta_params(sz, value, param_list);\r
+ param_block param = param_list.front();\r
+\r
+\r
+ gs_uint32_t reuse_flag = 2;\r
+ // we will try to create a new instance of child FTA only if it is parameterized\r
+ if (param.block_length > 0 && (reuse_option == 0 || (!fta_reusable && reuse_option == 2))) {\r
+ f.streamid = 0; // not interested in existing instances\r
+ reuse_flag = 0;\r
+ }\r
+\r
+ if ((fta_alloc_instance(_fta.ftaid, &f,fta_name,schema,reuse_flag, FTA_COMMAND_LOAD_PARAMS,param.block_length,param.data))!=0) {\r
+ fprintf(stderr,"HFTA::error:could instantiate a FTA");\r
+ failed = true;\r
+ return;\r
+ }\r
+\r
+ free(param.data);\r
+ // set the operator's parameters\r
+ if(oper->set_param_block(sz, (void*)value)) failed = true;;\r
+\r
+\r
+ fprintf(stderr,"HFTA::Low level FTA (%s) instanciation done\n", fta_name);\r
+ _fta.stream_subscribed_cnt = 1;\r
+ _fta.stream_subscribed[0] = f;\r
+\r
+ _fta.alloc_fta = NULL; // why should this be a part of the FTA (it is a factory function)\r
+ _fta.free_fta = UNOP_HFTA_free_fta;\r
+ _fta.control_fta = UNOP_HFTA_control_fta;\r
+ _fta.accept_packet = UNOP_HFTA_accept_packet;\r
+ _fta.clock_fta = UNOP_HFTA_clock_fta;\r
+ }\r
+\r
+ ~UNOP_HFTA() {\r
+ delete oper; // free operators memory\r
+\r
+ }\r
+\r
+ int flush() {\r
+ list<host_tuple> res;\r
+ if (!oper->flush(res)) {\r
+\r
+ if (!res.empty()) {\r
+ // go through the list of returned tuples and finalyze them\r
+ list<host_tuple>::iterator iter = res.begin();\r
+ while (iter != res.end()) {\r
+ host_tuple& tup = *iter;\r
+\r
+ // finalize the tuple\r
+ if (tup.tuple_size)\r
+ finalize_tuple(tup);\r
+ iter++;\r
+ }\r
+\r
+ // append returned list to output_queue\r
+ output_queue.splice(output_queue.end(), res);\r
+\r
+\r
+ // post tuples\r
+ while (!output_queue.empty()) {\r
+ host_tuple& tup = output_queue.front();\r
+ #ifdef DEBUG\r
+ fprintf(stderr, "HFTA::about to post tuple\n");\r
+ #endif\r
+ if (hfta_post_tuple(&_fta,tup.tuple_size,tup.data) == 0) {\r
+ tup.free_tuple();\r
+ output_queue.pop_front();\r
+ } else\r
+ break;\r
+ }\r
+ }\r
+ }\r
+\r
+ return 0;\r
+ }\r
+\r
+ bool init_failed(){return failed;}\r
+};\r
+\r
+gs_retval_t UNOP_HFTA_free_fta (struct FTA *fta, gs_uint32_t recursive) {\r
+ UNOP_HFTA* ftap = (UNOP_HFTA*)fta; // deallocate the fta and call the destructor\r
+ // will be called on program exit\r
+\r
+ if (recursive)\r
+ // free instance we are subscribed to\r
+ fta_free_instance(gscpipc_getftaid(), ftap->f, recursive);\r
+\r
+ delete ftap;\r
+ return 0;\r
+}\r
+\r
+gs_retval_t UNOP_HFTA_control_fta (struct FTA *fta, gs_int32_t command, gs_int32_t sz, void * value) {\r
+ UNOP_HFTA* ftap = (UNOP_HFTA*)fta;\r
+\r
+ if (command == FTA_COMMAND_FLUSH) {\r
+ // ask lfta to do the flush\r
+ fta_control(gscpipc_getftaid(), ftap->f, FTA_COMMAND_FLUSH, sz, value);\r
+ ftap->flush();\r
+\r
+ } else if (command == FTA_COMMAND_LOAD_PARAMS) {\r
+ // ask lfta to do the flush\r
+ fta_control(gscpipc_getftaid(), ftap->f, FTA_COMMAND_FLUSH, sz, value);\r
+ ftap->flush();\r
+\r
+ /* extract lfta param block from hfta param block */\r
+ list<param_block> param_list;\r
+ get_lfta_params(sz, value, param_list);\r
+ param_block param = param_list.front();\r
+ // load new parameters into lfta\r
+ fta_control(gscpipc_getftaid(), ftap->f, FTA_COMMAND_LOAD_PARAMS, param.block_length,param.data);\r
+ free(param.data);\r
+\r
+ // notify the operator about the change of parameter\r
+ ftap->oper->set_param_block(sz, (void*)value);\r
+\r
+ } else if (command == FTA_COMMAND_GET_TEMP_STATUS) {\r
+ // we no longer use temp_status commands\r
+ // hearbeat mechanism is used instead\r
+ }\r
+ return 0;\r
+}\r
+\r
+gs_retval_t UNOP_HFTA_accept_packet (struct FTA *fta, FTAID *ftaid, void * packet, gs_int32_t length) {\r
+ UNOP_HFTA* ftap = (UNOP_HFTA*)fta;\r
+#ifdef DEBUG\r
+ fprintf(stderr, "HFTA::accepted packet\n");\r
+#endif\r
+ if (!length) /* ignore null tuples */\r
+ return 0;\r
+\r
+ host_tuple temp;\r
+ temp.tuple_size = length;\r
+ temp.data = packet;\r
+ temp.channel = 0;\r
+ temp.heap_resident = false;\r
+\r
+ // pass the tuple to operator\r
+ list<host_tuple> res;\r
+ int ret;\r
+ fta_stat* tup_trace = NULL;\r
+ gs_uint32_t tup_trace_sz = 0;\r
+ gs_uint64_t trace_id = 0;\r
+ bool temp_tuple_received = false;\r
+\r
+\r
+ // if the tuple is temporal we need to extract the hearbeat payload\r
+ if (ftaschema_is_temporal_tuple(ftap->schema_handle, packet)) {\r
+ temp_tuple_received = true;\r
+ if (ftaschema_get_trace(ftap->schema_handle, packet, length, &trace_id, &tup_trace_sz, &tup_trace))\r
+ fprintf(stderr, "HFTA error: received temporal status tuple with no trace\n");\r
+ }\r
+\r
+ if (ftaschema_is_eof_tuple(ftap->schema_handle, packet)) {\r
+ /* perform a flush */\r
+ ftap->flush();\r
+\r
+ /* post eof_tuple to a parent */\r
+ host_tuple eof_tuple;\r
+ ftap->oper->get_temp_status(eof_tuple);\r
+\r
+ /* last byte of the tuple specifies the tuple type\r
+ * set it to EOF_TUPLE\r
+ */\r
+ *((char*)eof_tuple.data + (eof_tuple.tuple_size - sizeof(char))) = EOF_TUPLE;\r
+ hfta_post_tuple(fta,eof_tuple.tuple_size,eof_tuple.data);\r
+\r
+ return 0;\r
+ }\r
+\r
+ ret = ftap->oper->accept_tuple(temp, res);\r
+\r
+ // go through the list of returned tuples and finalyze them\r
+ list<host_tuple>::iterator iter = res.begin();\r
+ while (iter != res.end()) {\r
+ host_tuple& tup = *iter;\r
+\r
+ // finalize the tuple\r
+ if (tup.tuple_size)\r
+ finalize_tuple(tup);\r
+ iter++;\r
+ }\r
+\r
+ // if we received temporal tuple, last tuple of the result must be temporal too\r
+ // we need to extend the trace by adding ftaid and send a hearbeat to clearinghouse\r
+ if (temp_tuple_received) {\r
+ fta_stat stats;\r
+ host_tuple& temp_tup = res.back();\r
+\r
+\r
+ int new_tuple_size = temp_tup.tuple_size + sizeof(gs_uint64_t) + (tup_trace_sz + 1)* sizeof(fta_stat);\r
+ char* new_data = (char*)malloc(new_tuple_size);\r
+ memcpy(new_data, temp_tup.data, temp_tup.tuple_size);\r
+ memcpy(new_data + temp_tup.tuple_size, &trace_id, sizeof(gs_uint64_t));\r
+ memcpy(new_data + temp_tup.tuple_size + sizeof(gs_uint64_t), (char*)tup_trace, tup_trace_sz * sizeof(fta_stat));\r
+ memcpy(new_data + temp_tup.tuple_size + sizeof(gs_uint64_t) + tup_trace_sz * sizeof(fta_stat), (char*)ftaid, sizeof(FTAID));\r
+\r
+ memset((char*)&stats, 0, sizeof(fta_stat));\r
+ memcpy(new_data + temp_tup.tuple_size + sizeof(gs_uint64_t) + tup_trace_sz * sizeof(fta_stat), (char*)&stats, sizeof(fta_stat));\r
+\r
+ // Send a hearbeat message to clearinghouse.\r
+ fta_heartbeat(ftap->_fta.ftaid, trace_id, tup_trace_sz + 1, (fta_stat *)((char*)new_data + temp_tup.tuple_size + sizeof(gs_uint64_t)));\r
+\r
+ temp_tup.free_tuple();\r
+ temp_tup.data = new_data;\r
+ temp_tup.tuple_size = new_tuple_size;\r
+ }\r
+\r
+ // append returned list to output_queue\r
+ ftap->output_queue.splice(ftap->output_queue.end(), res);\r
+\r
+ // post tuples\r
+ while (!ftap->output_queue.empty()) {\r
+ host_tuple& tup = ftap->output_queue.front();\r
+ #ifdef DEBUG\r
+ fprintf(stderr, "HFTA::about to post tuple\n");\r
+ #endif\r
+ if (hfta_post_tuple(fta,tup.tuple_size,tup.data) == 0) {\r
+ tup.free_tuple();\r
+ ftap->output_queue.pop_front();\r
+ } else\r
+ break;\r
+ }\r
+\r
+ return 1;\r
+}\r
+\r
+gs_retval_t UNOP_HFTA_clock_fta(struct FTA *ftap) {\r
+\r
+ // Send a hearbeat message to clearinghouse.to indicate we are alive\r
+ fta_heartbeat(ftap->ftaid, 0, 0, 0);\r
+\r
+ return 0;\r
+}\r
+\r
+\r
+struct MULTOP_HFTA {\r
+ struct FTA _fta;\r
+ gs_csp_t fta_name;\r
+ gs_schemahandle_t schema_handle;\r
+ operator_node* root;\r
+ vector<operator_node*> sorted_nodes;\r
+ int num_operators;\r
+ list<lfta_info*> *lfta_list;\r
+ /* number of eof tuples we received so far\r
+ * receiving eof tuples from every source fta will cause a flush\r
+ */\r
+ int num_eof_tuples;\r
+\r
+ bool failed;\r
+ bool reusable;\r
+\r
+ list<host_tuple> output_queue;\r
+\r
+ // Runtime stats\r
+ gs_uint32_t in_tuple_cnt;\r
+ gs_uint32_t out_tuple_cnt;\r
+ gs_uint32_t out_tuple_sz;\r
+ gs_uint64_t cycle_cnt;\r
+\r
+ gs_uint64_t trace_id;\r
+\r
+ // memory occupied by output queue\r
+ gs_uint32_t output_queue_mem;\r
+\r
+\r
+ // To create an hfta we will need all the parameters passed to alloc_fta by host library plus\r
+ // lfta_name and an instance of the base_operator. We don't need to know the schema for lfta,\r
+ // as the schema handle is already passed during operator creation time.\r
+ // We also don't need to know the output schema as this information is already embeded\r
+ // in create_output_tuple method of operators' functor.\r
+\r
+\r
+\r
+ MULTOP_HFTA(struct FTAID ftaid, gs_csp_t name, gs_int32_t command, gs_int32_t sz, void * value, gs_schemahandle_t sch_handle, operator_node* node,\r
+ list<lfta_info*> *lftas, bool fta_reusable, gs_uint32_t reuse_option) {\r
+\r
+ fta_name = name;\r
+ failed = false;\r
+\r
+ root = node;\r
+ lfta_list = lftas;\r
+\r
+ // assign streamid\r
+ _fta.ftaid = ftaid;\r
+ _fta.ftaid.streamid = (gs_p_t)this;\r
+\r
+ schema_handle = sch_handle;\r
+\r
+ output_queue_mem = 0;\r
+\r
+ // topologically sort the operators in the tree (or DAG)\r
+ // for DAG we make sure we add the node to the sorted list only once\r
+ operator_node* current_node;\r
+ map<operator_node*, int> node_map;\r
+ vector<operator_node*> node_list;\r
+\r
+ int i = 0;\r
+ node_list.push_back(root);\r
+ node_map[root] = 0;\r
+\r
+ num_operators = 1;\r
+\r
+ while (i < node_list.size()) {\r
+ current_node = node_list[i];\r
+ if (current_node->left_child && node_map.find(current_node->left_child) == node_map.end()) {\r
+ node_map[current_node->left_child] = num_operators++;\r
+ node_list.push_back(current_node->left_child);\r
+ }\r
+ if (current_node->right_child && node_map.find(current_node->right_child) == node_map.end()) {\r
+ node_map[current_node->right_child] = num_operators++;\r
+ node_list.push_back(current_node->right_child);\r
+ }\r
+ i++;\r
+ }\r
+ num_operators = i;\r
+\r
+ // build adjacency lists for query DAG\r
+ list<int>* adj_lists = new list<int>[num_operators];\r
+ bool* leaf_flags = new bool[num_operators];\r
+ memset(leaf_flags, 0, num_operators * sizeof(bool));\r
+ for (i = 0; i < num_operators; ++i) {\r
+ current_node = node_list[i];\r
+ if (current_node->left_child) {\r
+ adj_lists[i].push_back(node_map[current_node->left_child]);\r
+ }\r
+ if (current_node->right_child && current_node->left_child != current_node->right_child) {\r
+ adj_lists[i].push_back(node_map[current_node->right_child]);\r
+ }\r
+ }\r
+\r
+ // run topolofical sort\r
+ bool leaf_found = true;\r
+ while (leaf_found) {\r
+ leaf_found = false;\r
+ // add all leafs to sorted_nodes\r
+ for (i = 0; i < num_operators; ++i) {\r
+ if (!leaf_flags[i] && adj_lists[i].empty()) {\r
+ leaf_flags[i] = true;\r
+ sorted_nodes.push_back(node_list[i]);\r
+ leaf_found = true;\r
+\r
+ // remove the node from its parents adjecency lists\r
+ for (int j = 0; j < num_operators; ++j) {\r
+ list<int>::iterator iter;\r
+ for (iter = adj_lists[j].begin(); iter != adj_lists[j].end(); iter++) {\r
+ if (*iter == i) {\r
+ adj_lists[j].erase(iter);\r
+ break;\r
+ }\r
+ }\r
+ }\r
+ }\r
+ }\r
+ }\r
+\r
+ delete[] adj_lists;\r
+ delete[] leaf_flags;\r
+\r
+ // set the parameter block for every operator in tree\r
+ for (i = 0; i < num_operators; ++i)\r
+ if(sorted_nodes[i]->op->set_param_block(sz, (void*)value)) failed = true;\r
+\r
+#ifdef DEBUG\r
+ fprintf(stderr,"Instantiate FTAs\n");\r
+#endif\r
+ /* extract lfta param block from hfta param block */\r
+// NOTE: param_list must line up with lfta_list\r
+ list<param_block> param_list;\r
+ get_lfta_params(sz, value, param_list);\r
+ list<param_block>::iterator iter1;\r
+ list<lfta_info*>::iterator iter2 = lfta_list->begin();\r
+\r
+ for (iter1 = param_list.begin(), i = 0; iter1 != param_list.end(); iter1++, iter2++, i++) {\r
+ lfta_info* inf = *iter2;\r
+\r
+ #ifdef DEBUG\r
+ fprintf(stderr,"Instantiate a FTA\n");\r
+ #endif\r
+\r
+ gs_uint32_t reuse_flag = 2;\r
+\r
+ // we will try to create a new instance of child FTA only if it is parameterized\r
+ if ((*iter1).block_length > 0 && (reuse_option == 0 || (!fta_reusable && reuse_option == 2))) {\r
+ (*iter2)->f.streamid = 0; // not interested in existing instances\r
+ reuse_flag = 0;\r
+ }\r
+ if (fta_alloc_instance(_fta.ftaid, &(*iter2)->f,(*iter2)->fta_name, (*iter2)->schema, reuse_flag, FTA_COMMAND_LOAD_PARAMS,(*iter1).block_length,(*iter1).data)!=0) {\r
+ fprintf(stderr,"HFTA::error:could instantiate a FTA");\r
+ failed = true;\r
+ return;\r
+ }\r
+\r
+ free((*iter1).data);\r
+\r
+ //fprintf(stderr,"HFTA::Low level FTA instanciation done\n");\r
+\r
+ _fta.stream_subscribed[i]=(*iter2)->f;\r
+ }\r
+ _fta.stream_subscribed_cnt = i;\r
+\r
+ num_eof_tuples = 0;\r
+\r
+ _fta.alloc_fta = NULL; // why should this be a part of the FTA (it is a factory function)\r
+ _fta.free_fta = MULTOP_HFTA_free_fta;\r
+ _fta.control_fta = MULTOP_HFTA_control_fta;\r
+ _fta.accept_packet = MULTOP_HFTA_accept_packet;\r
+ _fta.clock_fta = MULTOP_HFTA_clock_fta;\r
+\r
+ // init runtime stats\r
+ in_tuple_cnt = 0;\r
+ out_tuple_cnt = 0;\r
+ out_tuple_sz = 0;\r
+ cycle_cnt = 0;\r
+\r
+ }\r
+\r
+ ~MULTOP_HFTA() {\r
+\r
+ list<lfta_info*>::iterator iter;\r
+ int i = 0;\r
+\r
+ for (iter = lfta_list->begin(); i < _fta.stream_subscribed_cnt; iter++, i++) {\r
+ delete *iter;\r
+ }\r
+\r
+ delete root; // free operators memory\r
+ delete lfta_list;\r
+\r
+\r
+ }\r
+\r
+ int flush() {\r
+\r
+ list<host_tuple> res;\r
+\r
+ // go through the list of operators in topological order\r
+ // and flush them\r
+ list<host_tuple>::iterator iter;\r
+ list<host_tuple> temp_output_queue;\r
+\r
+ for (int i = 0; i < num_operators; ++i) {\r
+ operator_node* node = sorted_nodes[i];\r
+\r
+#ifdef PLAN_DAG\r
+ list<host_tuple>& current_output_queue = (i < (num_operators - 1)) ? temp_output_queue : res;\r
+#else\r
+ // for trees we can put output tuples directly into parent's input buffer\r
+ list<host_tuple>& current_output_queue = (i < (num_operators - 1)) ? node->parent->input_queue : res;\r
+#endif\r
+ // consume tuples waiting in your queue\r
+ for (iter = node->input_queue.begin(); iter != node->input_queue.end(); iter++) {\r
+ node->op->accept_tuple(*(iter), current_output_queue);\r
+ }\r
+ node->op->flush(current_output_queue);\r
+ node->input_queue.clear();\r
+\r
+#ifdef PLAN_DAG\r
+ // copy the tuples from output queue into input queues of all parents\r
+ list<operator_node*>::iterator node_iter;\r
+\r
+ if (!node->parent_list.empty()) {\r
+ // append the content of the output queue to parent input queue\r
+\r
+ for (iter = temp_output_queue.begin(); iter != temp_output_queue.end(); iter++) {\r
+ int* ref_cnt = 0;\r
+ if (node->parent_list.size() > 1) {\r
+ ref_cnt = (int*)malloc(sizeof(int));\r
+ *ref_cnt = node->parent_list.size() - 1;\r
+ }\r
+\r
+ for (node_iter = node->parent_list.begin(); node_iter != node->parent_list.end(); node_iter++) {\r
+ (*iter).ref_cnt = ref_cnt;\r
+ (*node_iter)->input_queue.push_back(*iter);\r
+ }\r
+ }\r
+ }\r
+#endif\r
+ }\r
+\r
+ if (!res.empty()) {\r
+ // go through the list of returned tuples and finalyze them\r
+ list<host_tuple>::iterator iter = res.begin();\r
+ while (iter != res.end()) {\r
+ host_tuple& tup = *iter;\r
+\r
+ // finalize the tuple\r
+ if (tup.tuple_size)\r
+ finalize_tuple(tup);\r
+\r
+ output_queue_mem += tup.tuple_size;\r
+ iter++;\r
+ }\r
+\r
+ // append returned list to output_queue\r
+ output_queue.splice(output_queue.end(), res);\r
+\r
+\r
+ // post tuples\r
+ while (!output_queue.empty()) {\r
+ host_tuple& tup = output_queue.front();\r
+ #ifdef DEBUG\r
+ fprintf(stderr, "HFTA::about to post tuple\n");\r
+ #endif\r
+ if (hfta_post_tuple(&_fta,tup.tuple_size,tup.data) == 0) {\r
+ output_queue_mem -= tup.tuple_size;\r
+ tup.free_tuple();\r
+ output_queue.pop_front();\r
+ } else\r
+ break;\r
+ }\r
+ }\r
+\r
+ return 0;\r
+ }\r
+\r
+ bool init_failed(){return failed;}\r
+};\r
+\r
+\r
+gs_retval_t MULTOP_HFTA_free_fta (struct FTA *fta, gs_uint32_t recursive) {\r
+ MULTOP_HFTA* ftap = (MULTOP_HFTA*)fta; // deallocate the fta and call the destructor\r
+ // will be called on program exit\r
+\r
+ if (recursive) {\r
+ // free instance we are subscribed to\r
+ list<lfta_info*>::iterator iter;\r
+ int i = 0;\r
+\r
+ for (iter = ftap->lfta_list->begin(); i < fta->stream_subscribed_cnt; iter++, i++) {\r
+ fta_free_instance(gscpipc_getftaid(), (*iter)->f, recursive);\r
+ }\r
+ }\r
+\r
+ delete ftap;\r
+ return 0;\r
+}\r
+\r
+gs_retval_t MULTOP_HFTA_control_fta (struct FTA *fta, gs_int32_t command, gs_int32_t sz, void * value) {\r
+ MULTOP_HFTA* ftap = (MULTOP_HFTA*)fta;\r
+\r
+ if (command == FTA_COMMAND_FLUSH) {\r
+\r
+ // ask lftas to do the flush\r
+ list<lfta_info*>::iterator iter;\r
+ for (iter = ftap->lfta_list->begin(); iter != ftap->lfta_list->end(); iter++) {\r
+ fta_control(gscpipc_getftaid(), (*iter)->f, FTA_COMMAND_FLUSH, sz, value);\r
+ }\r
+ // flush hfta operators\r
+ ftap->flush();\r
+\r
+ } else if (command == FTA_COMMAND_LOAD_PARAMS) {\r
+\r
+ list<param_block> param_list;\r
+ get_lfta_params(sz, value, param_list);\r
+\r
+ // ask lftas to do the flush and set new parameters\r
+ list<lfta_info*>::iterator iter;\r
+ list<param_block>::iterator iter2;\r
+ for (iter = ftap->lfta_list->begin(), iter2 = param_list.begin(); iter != ftap->lfta_list->end(); iter++, iter2++) {\r
+ fta_control(gscpipc_getftaid(), (*iter)->f, FTA_COMMAND_FLUSH, 0, NULL);\r
+ fta_control(gscpipc_getftaid(), (*iter)->f, FTA_COMMAND_LOAD_PARAMS, (*iter2).block_length,(*iter2).data);\r
+ free((*iter2).data);\r
+ }\r
+ // flush hfta operators\r
+ ftap->flush();\r
+\r
+ // set the new parameter block for every operator in tree\r
+ for (int i = 0; i < ftap->num_operators; ++i)\r
+ ftap->sorted_nodes[i]->op->set_param_block(sz, (void*)value);\r
+\r
+ } else if (command == FTA_COMMAND_GET_TEMP_STATUS) {\r
+ // we no longer use temp_status commands\r
+ // hearbeat mechanism is used instead\r
+ }\r
+ return 0;\r
+}\r
+\r
+gs_retval_t MULTOP_HFTA_accept_packet (struct FTA *fta, FTAID *ftaid, void * packet, gs_int32_t length) {\r
+ MULTOP_HFTA* ftap = (MULTOP_HFTA*)fta;\r
+\r
+ gs_uint64_t start_cycle = rdtsc();\r
+#ifdef DEBUG\r
+ fprintf(stderr, "HFTA::accepted packet\n");\r
+#endif\r
+ if (!length) /* ignore null tuples */\r
+ return 0;\r
+\r
+ ftap->in_tuple_cnt++;\r
+\r
+ host_tuple temp;\r
+ temp.tuple_size = length;\r
+ temp.data = packet;\r
+ temp.channel = 0;\r
+ temp.heap_resident = false;\r
+\r
+// fprintf(stderr,"created temp, channel=%d, resident=%d\n",temp.channel, (int)temp.heap_resident);\r
+\r
+ // find from which lfta the tuple came\r
+ list<lfta_info*>::iterator iter;\r
+ lfta_info* inf = NULL;\r
+ int i;\r
+\r
+ for (i = 0, iter = ftap->lfta_list->begin(); i < ftap->_fta.stream_subscribed_cnt; iter++, i++) {\r
+ if (ftap->_fta.stream_subscribed[i].ip == ftaid->ip &&\r
+ ftap->_fta.stream_subscribed[i].port == ftaid->port &&\r
+ ftap->_fta.stream_subscribed[i].index == ftaid->index &&\r
+ ftap->_fta.stream_subscribed[i].streamid == ftaid->streamid) {\r
+ inf = *iter;\r
+ break;\r
+ }\r
+ }\r
+\r
+ if (!inf) {\r
+ fprintf(stderr,"HFTA::error:received tuple from unknown stream\n");\r
+ exit(1);\r
+ }\r
+\r
+ // route tuple through operator chain\r
+ list<host_tuple> result;\r
+ host_tuple tup;\r
+ int ret;\r
+ #ifndef PLAN_DAG\r
+ temp.channel = inf->output_channel;\r
+ #endif\r
+ operator_node* current_node = NULL, *child = NULL;\r
+ list<host_tuple> temp_output_queue;\r
+\r
+\r
+ fta_stat* tup_trace = NULL;\r
+ gs_uint32_t tup_trace_sz = 0;\r
+ gs_uint64_t trace_id = 0;\r
+ bool temp_tuple_received = false;\r
+\r
+ // if the tuple is temporal we need to extract the heartbeat payload\r
+ if (ftaschema_is_temporal_tuple(inf->schema_handle, packet)) {\r
+ temp_tuple_received = true;\r
+ if (ftaschema_get_trace(inf->schema_handle, packet, length, &trace_id, &tup_trace_sz, &tup_trace))\r
+ fprintf(stderr, "HFTA error: received temporal status tuple with no trace\n");\r
+ }\r
+\r
+ if (ftaschema_is_eof_tuple(inf->schema_handle, packet)) {\r
+\r
+ if (++ftap->num_eof_tuples < ftap->lfta_list->size())\r
+ return 0;\r
+\r
+ ftap->num_eof_tuples = 0;\r
+\r
+ /* perform a flush */\r
+ ftap->flush();\r
+\r
+ /* post eof_tuple to a parent */\r
+ host_tuple eof_tuple;\r
+ ftap->sorted_nodes[ftap->num_operators - 1]->op->get_temp_status(eof_tuple);\r
+\r
+ /* last byte of the tuple specify the tuple type\r
+ * set it to EOF_TUPLE\r
+ */\r
+ *((char*)eof_tuple.data + (eof_tuple.tuple_size - sizeof(char))) = EOF_TUPLE;\r
+ hfta_post_tuple(fta,eof_tuple.tuple_size,eof_tuple.data);\r
+ ftap->out_tuple_cnt++;\r
+ ftap->out_tuple_sz+=eof_tuple.tuple_size;\r
+\r
+ return 0;\r
+ }\r
+\r
+ list<host_tuple>::iterator iter2;\r
+\r
+#ifdef PLAN_DAG\r
+\r
+ // push tuple to all parent operators of the lfta\r
+ list<operator_node*>::iterator node_iter;\r
+ list<unsigned>::iterator chan_iter;\r
+ for (node_iter = inf->parent_list.begin(), chan_iter = inf->out_channel_list.begin(); node_iter != inf->parent_list.end(); node_iter++, chan_iter++) {\r
+ temp.channel = *chan_iter;\r
+ (*node_iter)->input_queue.push_back(temp);\r
+ }\r
+\r
+ for (i = 0; i < ftap->num_operators; ++i) {\r
+\r
+ operator_node* node = ftap->sorted_nodes[i];\r
+ list<host_tuple>& current_output_queue = (i < (ftap->num_operators - 1)) ? temp_output_queue : result;\r
+\r
+ // consume tuples waiting in your queue\r
+ for (iter2 = node->input_queue.begin(); iter2 != node->input_queue.end(); iter2++) {\r
+ node->op->accept_tuple(*(iter2), current_output_queue);\r
+ }\r
+ node->input_queue.clear();\r
+\r
+ // copy the tuples from output queue into input queues of all parents\r
+\r
+ if (!node->parent_list.empty()) {\r
+\r
+ // append the content of the output queue to parent input queue\r
+ for (iter2 = temp_output_queue.begin(); iter2 != temp_output_queue.end(); iter2++) {\r
+\r
+ int* ref_cnt = 0;\r
+ if (node->parent_list.size() > 1) {\r
+ ref_cnt = (int*)malloc(sizeof(int));\r
+ *ref_cnt = node->parent_list.size() - 1;\r
+ }\r
+\r
+ for (node_iter = node->parent_list.begin(), chan_iter = node->out_channel_list.begin(); node_iter != node->parent_list.end(); node_iter++, chan_iter++) {\r
+ (*iter2).ref_cnt = ref_cnt;\r
+ (*iter2).channel = *chan_iter;\r
+ (*node_iter)->input_queue.push_back(*iter2);\r
+ }\r
+ }\r
+ }\r
+ temp_output_queue.clear();\r
+ }\r
+#else\r
+ current_node = inf->parent;\r
+\r
+// fprintf(stderr,"Pushing temp, channel=%d, resident=%d\n",temp.channel, (int)temp.heap_resident);\r
+ current_node->input_queue.push_back(temp);\r
+\r
+ do {\r
+//fprintf(stderr,"Routing tuple, current node is %d, parent is %d\n",current_node,current_node->parent);\r
+ list<host_tuple>& current_output_queue = (current_node->parent) ? current_node->parent->input_queue : result;\r
+\r
+ // consume tuples waiting in your queue\r
+ for (iter2 = current_node->input_queue.begin(); iter2 != current_node->input_queue.end(); iter2++) {\r
+ current_node->op->accept_tuple((*iter2),current_output_queue);\r
+ }\r
+// All consumed, delete them\r
+ current_node->input_queue.clear();\r
+ current_node = current_node->parent;\r
+\r
+ } while (current_node);\r
+#endif\r
+\r
+\r
+ host_tuple temp_tup;\r
+\r
+ bool no_temp_tuple = false;\r
+\r
+ // if we received temporal tuple, last tuple of the result must be temporal too\r
+ // we need to extend the trace by adding ftaid and send a hearbeat to clearinghouse\r
+ if (temp_tuple_received) {\r
+\r
+ if (result.empty()) {\r
+ no_temp_tuple = true;\r
+\r
+ } else {\r
+ fta_stat stats;\r
+ temp_tup = result.back();\r
+ finalize_tuple(temp_tup);\r
+\r
+ int new_tuple_size = temp_tup.tuple_size + sizeof(gs_uint64_t) + (tup_trace_sz + 1)* sizeof(fta_stat);\r
+ char* new_data = (char*)malloc(new_tuple_size);\r
+ memcpy(new_data, temp_tup.data, temp_tup.tuple_size);\r
+ memcpy(new_data + temp_tup.tuple_size, &trace_id, sizeof(gs_uint64_t));\r
+ memcpy(new_data + temp_tup.tuple_size + sizeof(gs_uint64_t), (char*)tup_trace, tup_trace_sz * sizeof(fta_stat));\r
+\r
+\r
+ memset((char*)&stats, 0, sizeof(fta_stat));\r
+ stats.ftaid = fta->ftaid;\r
+ stats.in_tuple_cnt = ftap->in_tuple_cnt;\r
+ stats.out_tuple_cnt = ftap->out_tuple_cnt;\r
+ stats.out_tuple_sz = ftap->out_tuple_sz;\r
+ stats.cycle_cnt = ftap->cycle_cnt;\r
+ memcpy(new_data + temp_tup.tuple_size + sizeof(gs_uint64_t) + tup_trace_sz * sizeof(fta_stat), (char*)&stats, sizeof(fta_stat));\r
+\r
+ // Send a hearbeat message to clearinghouse.\r
+ fta_heartbeat(fta->ftaid, trace_id, tup_trace_sz + 1, (fta_stat *)((char*)new_data + temp_tup.tuple_size + sizeof(gs_uint64_t)));\r
+\r
+ // reset the stats\r
+ ftap->in_tuple_cnt = 0;\r
+ ftap->out_tuple_cnt = 0;\r
+ ftap->out_tuple_sz = 0;\r
+ ftap->cycle_cnt = 0;\r
+\r
+ free(temp_tup.data);\r
+ temp_tup.data = new_data;\r
+ temp_tup.tuple_size = new_tuple_size;\r
+ result.pop_back();\r
+ }\r
+ }\r
+\r
+ // go through the list of returned tuples and finalyze them\r
+ // since we can produce multiple temporal tuples in DAG plans\r
+ // we can drop all of them except the last one\r
+ iter2 = result.begin();\r
+ while(iter2 != result.end()) {\r
+ host_tuple tup = *iter2;\r
+\r
+ // finalize the tuple\r
+ if (tup.tuple_size) {\r
+ finalize_tuple(tup);\r
+\r
+ #ifdef PLAN_DAG\r
+ if (ftaschema_is_temporal_tuple(ftap->schema_handle, tup.data))\r
+ tup.free_tuple();\r
+ else\r
+ #endif\r
+ {\r
+ ftap->output_queue.push_back(tup);\r
+ ftap->output_queue_mem += tup.tuple_size;\r
+ }\r
+\r
+ }\r
+ iter2++;\r
+ }\r
+\r
+ // append returned list to output_queue\r
+ // ftap->output_queue.splice(ftap->output_queue.end(), result);\r
+\r
+ if (temp_tuple_received && !no_temp_tuple) {\r
+ ftap->output_queue.push_back(temp_tup);\r
+ ftap->output_queue_mem += temp_tup.tuple_size;\r
+ }\r
+\r
+ // post tuples\r
+ while (!ftap->output_queue.empty()) {\r
+ host_tuple& tup = ftap->output_queue.front();\r
+ #ifdef DEBUG\r
+ fprintf(stderr, "HFTA::about to post tuple\n");\r
+ #endif\r
+ if (hfta_post_tuple(fta,tup.tuple_size,tup.data) == 0) {\r
+ ftap->out_tuple_cnt++;\r
+ ftap->out_tuple_sz+=tup.tuple_size;\r
+ ftap->output_queue_mem -= tup.tuple_size;\r
+ tup.free_tuple();\r
+ ftap->output_queue.pop_front();\r
+ } else\r
+ break;\r
+ }\r
+\r
+ ftap->cycle_cnt += rdtsc() - start_cycle;\r
+\r
+ return 1;\r
+}\r
+\r
+gs_retval_t MULTOP_HFTA_clock_fta(struct FTA *fta) {\r
+ MULTOP_HFTA* ftap = (MULTOP_HFTA*)fta;\r
+\r
+#ifdef HFTA_PROFILE\r
+ /* Print stats */\r
+ fprintf(stderr, "FTA = %s|", ftap->fta_name);\r
+ fprintf(stderr, "in_tuple_cnt = %u|", ftap->in_tuple_cnt);\r
+ fprintf(stderr, "out_tuple_cnt = %u|", ftap->out_tuple_cnt);\r
+ fprintf(stderr, "out_tuple_sz = %u|", ftap->out_tuple_sz);\r
+ fprintf(stderr, "cycle_cnt = %llu|", ftap->cycle_cnt);\r
+\r
+\r
+ fprintf(stderr, "mem_footprint %s = %d", ftap->sorted_nodes[0]->op->get_name(), ftap->sorted_nodes[0]->op->get_mem_footprint());\r
+ unsigned int total_mem = ftap->sorted_nodes[0]->op->get_mem_footprint();\r
+\r
+ for (int i = 1; i < ftap->num_operators; ++i) {\r
+ operator_node* node = ftap->sorted_nodes[i];\r
+ fprintf(stderr, ",%s = %d", node->op->get_name(), node->op->get_mem_footprint());\r
+ total_mem += node->op->get_mem_footprint();\r
+ }\r
+ fprintf(stderr, ", total = %d|", total_mem );\r
+ fprintf(stderr, "output_buffer_size = %d\n", ftap->output_queue_mem );\r
+#endif\r
+\r
+ fta_stat stats;\r
+ memset((char*)&stats, 0, sizeof(fta_stat));\r
+ stats.ftaid = fta->ftaid;\r
+ stats.in_tuple_cnt = ftap->in_tuple_cnt;\r
+ stats.out_tuple_cnt = ftap->out_tuple_cnt;\r
+ stats.out_tuple_sz = ftap->out_tuple_sz;\r
+ stats.cycle_cnt = ftap->cycle_cnt;\r
+\r
+ // Send a hearbeat message to clearinghouse.\r
+ fta_heartbeat(fta->ftaid, ftap->trace_id++, 1, &stats);\r
+\r
+ // resets runtime stats\r
+ ftap->in_tuple_cnt = 0;\r
+ ftap->out_tuple_cnt = 0;\r
+ ftap->out_tuple_sz = 0;\r
+ ftap->cycle_cnt = 0;\r
+\r
+ return 0;\r
+}\r
+\r
+\r
+#endif // __HFTA_H\r
+\r
Code Review / com / gs-lite.git / blobdiff