-/* ------------------------------------------------
-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.
- ------------------------------------------- */
-
-#include<unistd.h> // for gethostname
-
-#include <string>
-#include "parse_fta.h"
-#include "parse_schema.h"
-#include "parse_ext_fcns.h"
-#include"analyze_fta.h"
-#include"query_plan.h"
-#include"generate_lfta_code.h"
-#include"stream_query.h"
-#include"generate_utils.h"
-#include"nic_def.h"
-#include"generate_nic_code.h"
-
-#include <stdlib.h>
-#include <stdio.h>
-#include<ctype.h>
-#include<glob.h>
-#include<string.h>
-
-#include<list>
-
-// for the scandir
- #include <sys/types.h>
- #include <dirent.h>
-
-
-#include<errno.h>
-
-// to verify that some files exist.
- #include <sys/types.h>
- #include <sys/stat.h>
-
-#include "parse_partn.h"
-
-#include "print_plan.h"
-
-// Interface to the xml parser
-
-#include"xml_t.h"
-#include"field_list.h"
-
-extern int xmlParserparse(void);
-extern FILE *xmlParserin;
-extern int xmlParserdebug;
-
-std::vector<std::string> xml_attr_vec;
-std::vector<std::string> xml_val_vec;
-std::string xml_a, xml_v;
-xml_t *xml_leaves = NULL;
-
-// Interface to the field list verifier
-field_list *field_verifier = NULL;
-
-#define TMPSTRLEN 1000
-
-#ifndef PATH_DELIM
- #define PATH_DELIM '/'
-#endif
-
-char tmp_schema_str[10000];
-
-// maximum delay between two hearbeats produced
-// by UDOP. Used when its not explicity
-// provided in udop definition
-#define DEFAULT_UDOP_LIVENESS_TIMEOUT 5
-
-// Default lfta hash table size, must be power of 2.
-int DEFAULT_LFTA_HASH_TABLE_SIZE = 4096;
-
-// Interface to FTA definition lexer and parser ...
-
-extern int FtaParserparse(void);
-extern FILE *FtaParserin;
-extern int FtaParserdebug;
-
-fta_parse_t *fta_parse_result;
-var_defs_t *fta_parse_defines;
-
-
-
-// Interface to external function lexer and parser ...
-
-extern int Ext_fcnsParserparse(void);
-extern FILE *Ext_fcnsParserin;
-extern int Ext_fcnsParserdebug;
-
-ext_fcn_list *Ext_fcns;
-
-
-// Interface to partition definition parser
-extern int PartnParserparse();
-partn_def_list_t *partn_parse_result = NULL;
-
-
-
-using namespace std;
-//extern int errno;
-
-
-// forward delcaration of local utility function
-void generate_makefile(vector<string> &input_file_names, int nfiles,
- vector<string> &hfta_names, opview_set &opviews,
- vector<string> &machine_names,
- string schema_file_name,
- vector<string> &interface_names,
- ifq_t *ifdb, string &config_dir_path,
- bool use_pads,
- string extra_libs,
- map<string, vector<int> > &rts_hload
- );
-
-//static int split_string(char *instr,char sep, char **words,int max_words);
-#define MAXFLDS 100
-
- FILE *schema_summary_output = NULL; // query names
-
-// Dump schema summary
-void dump_summary(stream_query *str){
- fprintf(schema_summary_output,"%s\n",str->query_name.c_str());
-
- table_def *sch = str->get_output_tabledef();
-
- vector<field_entry *> flds = sch->get_fields();
- int f;
- for(f=0;f<flds.size();++f){
- if(f>0) fprintf(schema_summary_output,"|");
- fprintf(schema_summary_output,"%s",flds[f]->get_name().c_str());
- }
- fprintf(schema_summary_output,"\n");
- for(f=0;f<flds.size();++f){
- if(f>0) fprintf(schema_summary_output,"|");
- fprintf(schema_summary_output,"%s",flds[f]->get_type().c_str());
- }
- fprintf(schema_summary_output,"\n");
-}
-
-// Globals
-string hostname; // name of current host.
-int hostname_len;
-bool generate_stats = false;
-string root_path = "../..";
-
-
-int main(int argc, char **argv){
- char tmpstr[TMPSTRLEN];
- string err_str;
- int q,s,h,f;
-
- set<int>::iterator si;
-
- vector<string> query_names; // for lfta.c registration
- map<string, vector<int> > mach_query_names; // list queries of machine
- vector<int> snap_lengths; // for lfta.c registration
- vector<string> interface_names; // for lfta.c registration
- vector<string> machine_names; // machine of interface
- vector<bool> lfta_reuse_options; // for lfta.c registration
- vector<int> lfta_liveness_timeouts; // fot qtree.xml generation
- vector<string> hfta_names; // hfta cource code names, for
- // creating make file.
- vector<string> qnames; // ensure unique names
- map<string, int> lfta_names; // keep track of unique lftas.
-
-
-// set these to 1 to debug the parser
- FtaParserdebug = 0;
- Ext_fcnsParserdebug = 0;
-
- FILE *lfta_out; // lfta.c output.
- FILE *fta_in; // input file
- FILE *table_schemas_in; // source tables definition file
- FILE *query_name_output; // query names
- FILE *qtree_output; // interconnections of query nodes
-
- // -------------------------------
- // Handling of Input Arguments
- // -------------------------------
- char optstr[] = "BDpLC:l:HNQMf:PSh:n:cR:";
- const char *usage_str = "Usage: %s [-B] [-D] [-p] [-L] [-N] [-H] [-Q] [-M] [-C <config directory>] [-l <library directory>] [-f] [-P] [-S] [-n n_virtual_interfaces] [-h hostname] [-R root_path] [schema_file] input_file [input file ...]\n"
- "\t[-B] : debug only (don't create output files)\n"
- "\t[-D] : distributed mode (will use cluster.ifq instead of local .ifq file)\n"
- "\t[-p] : partitioned mode (will use partition information in partition.txt to drive query optimization)\n"
- "\t[-L] : use live_hosts.txt file to restrict queries to a set of live hosts\n"
- "\t[-C] : use <config directory> for definition files\n"
- "\t[-l] : use <library directory> for library queries\n"
- "\t[-N] : output query names in query_names.txt\n"
- "\t[-H] : create HFTA only (no schema_file)\n"
- "\t[-Q] : use query name for hfta suffix\n"
- "\t[-M] : generate make file and runit, stopit scripts\n"
- "\t[-S] : enable LFTA statistics (alters Makefile).\n"
- "\t[-f] : Output schema summary to schema_summary.txt\n"
- "\t[-P] : link with PADS\n"
- "\t[-h] : override host name.\n"
- "\t[-c] : clean out Makefile and hfta_*.cc first.\n"
- "\t[-R] : path to root of GS-lite\n"
-;
-
-// parameters gathered from command line processing
- string external_fcns_path;
-// string internal_fcn_path;
- string config_dir_path;
- string library_path = "./";
- vector<string> input_file_names;
- string schema_file_name;
- bool debug_only = false;
- bool hfta_only = false;
- bool output_query_names = false;
- bool output_schema_summary=false;
- bool numeric_hfta_flname = true;
- bool create_makefile = false;
- bool distributed_mode = false;
- bool partitioned_mode = false;
- bool use_live_hosts_file = false;
- bool use_pads = false;
- bool clean_make = false;
- int n_virtual_interfaces = 1;
-
- char chopt;
- while((chopt = getopt(argc,argv,optstr)) != -1){
- switch(chopt){
- case 'B':
- debug_only = true;
- break;
- case 'D':
- distributed_mode = true;
- break;
- case 'p':
- partitioned_mode = true;
- break;
- case 'L':
- use_live_hosts_file = true;
- break;
- case 'C':
- if(optarg != NULL)
- config_dir_path = string(optarg) + string("/");
- break;
- case 'l':
- if(optarg != NULL)
- library_path = string(optarg) + string("/");
- break;
- case 'N':
- output_query_names = true;
- break;
- case 'Q':
- numeric_hfta_flname = false;
- break;
- case 'H':
- if(schema_file_name == ""){
- hfta_only = true;
- }
- break;
- case 'f':
- output_schema_summary=true;
- break;
- case 'M':
- create_makefile=true;
- break;
- case 'S':
- generate_stats=true;
- break;
- case 'P':
- use_pads = true;
- break;
- case 'c':
- clean_make = true;
- break;
- case 'h':
- if(optarg != NULL)
- hostname = optarg;
- break;
- case 'R':
- if(optarg != NULL)
- root_path = optarg;
- break;
- case 'n':
- if(optarg != NULL){
- n_virtual_interfaces = atoi(optarg);
- if(n_virtual_interfaces < 1 || n_virtual_interfaces > 128){
- fprintf(stderr,"Warning, %d virtual interfaces specified, valid values are between 1 and 128 inclusive\n",n_virtual_interfaces);
- n_virtual_interfaces = 1;
- }
- }
- break;
- case '?':
- fprintf(stderr,"Error, argument %c not recognized.\n",optopt);
- fprintf(stderr,"%s\n", usage_str);
- exit(1);
- default:
- fprintf(stderr, "Argument was %c\n", optopt);
- fprintf(stderr,"Invalid arguments\n");
- fprintf(stderr,"%s\n", usage_str);
- exit(1);
- }
- }
- argc -= optind;
- argv += optind;
- for (int i = 0; i < argc; ++i) {
- if((schema_file_name == "") && !hfta_only){
- schema_file_name = argv[i];
- }else{
- input_file_names.push_back(argv[i]);
- }
- }
-
- if(input_file_names.size() == 0){
- fprintf(stderr,"%s\n", usage_str);
- exit(1);
- }
-
- if(clean_make){
- string clean_cmd = "rm Makefile hfta_*.cc";
- int clean_ret = system(clean_cmd.c_str());
- if(clean_ret){
- fprintf(stderr,"Warning, return value %d when trying to clean out old source files.\n", clean_ret);
- }
- }
-
-
- nic_prop_db *npdb = new nic_prop_db(config_dir_path);
-
-// Open globally used file names.
-
- // prepend config directory to schema file
- schema_file_name = config_dir_path + schema_file_name;
- external_fcns_path = config_dir_path + string("external_fcns.def");
- string ifx_fname = config_dir_path + string("ifres.xml");
-
-// Find interface query file(s).
- if(hostname == ""){
- gethostname(tmpstr,TMPSTRLEN);
- hostname = tmpstr;
- }
- hostname_len = strlen(tmpstr);
- string ifq_fname = config_dir_path + (distributed_mode ? "cluster" : hostname) + string(".ifq");
- vector<string> ifq_fls;
-
- ifq_fls.push_back(ifq_fname);
-
-
-// Get the field list, if it exists
- string flist_fl = config_dir_path + "field_list.xml";
- FILE *flf_in = NULL;
- if((flf_in = fopen(flist_fl.c_str(), "r")) != NULL) {
- fprintf(stderr,"Found field list file %s\n",flist_fl.c_str());
- xml_leaves = new xml_t();
- xmlParser_setfileinput(flf_in);
- if(xmlParserparse()){
- fprintf(stderr,"WARNING, could not parse field list file %s\n",flist_fl.c_str());
- }else{
- field_verifier = new field_list(xml_leaves);
- }
- }
-
- if(!hfta_only){
- if((table_schemas_in = fopen(schema_file_name.c_str(), "r")) == NULL) {
- fprintf(stderr,"Can't open schema file %s\n%s\n",schema_file_name.c_str(),strerror(errno));
- exit(1);
- }
- }
-
-/*
- if(!(debug_only || hfta_only)){
- if((lfta_out = fopen("lfta.c","w")) == NULL){
- fprintf(stderr,"Can't open output file %s\n%s\n","lfta.c",strerror(errno));
- exit(1);
- }
- }
-*/
-
-// Get the output specification file.
-// format is
-// query, operator, operator_param, directory, bucketwidth, partitioning_se, n_partitions
- string ospec_fl = "output_spec.cfg";
- FILE *osp_in = NULL;
- vector<ospec_str *> output_specs;
- multimap<string, int> qname_to_ospec;
- if((osp_in = fopen(ospec_fl.c_str(), "r")) != NULL) {
- char *flds[MAXFLDS];
- int o_lineno = 0;
- while(fgets(tmpstr,TMPSTRLEN,osp_in)){
- o_lineno++;
- int nflds = split_string(tmpstr,',',flds,MAXFLDS);
- if(nflds == 7){
-// make operator type lowercase
- char *tmpc;
- for(tmpc=flds[1];*tmpc!='\0';++tmpc)
- *tmpc = tolower(*tmpc);
-
- ospec_str *tmp_ospec = new ospec_str();
- tmp_ospec->query = flds[0];
- tmp_ospec->operator_type = flds[1];
- tmp_ospec->operator_param = flds[2];
- tmp_ospec->output_directory = flds[3];
- tmp_ospec->bucketwidth = atoi(flds[4]);
- tmp_ospec->partitioning_flds = flds[5];
- tmp_ospec->n_partitions = atoi(flds[6]);
- qname_to_ospec.insert(pair<string,int>(tmp_ospec->query,output_specs.size()));
- output_specs.push_back(tmp_ospec);
- }else{
- fprintf(stderr,"Warning, line %d corrupted in output_spec.cfg, has %d fields.\n",o_lineno,nflds);
- }
- }
- fclose(osp_in);
- }else{
- fprintf(stderr,"output_spec.cfg not found. The query set has no output. exiting.\n");
- exit(1);
- }
-
-// hfta parallelism
- string pspec_fl = "hfta_parallelism.cfg";
- FILE *psp_in = NULL;
- map<string, int> hfta_parallelism;
- if((psp_in = fopen(pspec_fl.c_str(), "r")) != NULL){
- char *flds[MAXFLDS];
- int o_lineno = 0;
- while(fgets(tmpstr,TMPSTRLEN,psp_in)){
- bool good_entry = true;
- o_lineno++;
- int nflds = split_string(tmpstr,',',flds,MAXFLDS);
- if(nflds == 2){
- string hname = flds[0];
- int par = atoi(flds[1]);
- if(par <= 0 || par > n_virtual_interfaces){
- fprintf(stderr,"Warning, line %d of %s is incorrectly formatted, parallelism is %d, must be between 1 and %d\n",o_lineno,pspec_fl.c_str(),par,n_virtual_interfaces);
- good_entry = false;
- }
- if(good_entry && n_virtual_interfaces % par != 0){
- fprintf(stderr,"Warning, line %d of %s is incorrectly formatted, parallelism is %d, must divide the number of virtual interfaces (%d), ignoring.\n",o_lineno,pspec_fl.c_str(),par,n_virtual_interfaces);
- good_entry = false;
- }
- if(good_entry)
- hfta_parallelism[hname] = par;
- }
- }
- }else{
- fprintf(stderr,"WARNING, no file %s found, using single copies of hftas.\n",pspec_fl.c_str());
- }
-
-
-// LFTA hash table sizes
- string htspec_fl = "lfta_htsize.cfg";
- FILE *htsp_in = NULL;
- map<string, int> lfta_htsize;
- if((htsp_in = fopen(htspec_fl.c_str(), "r")) != NULL){
- char *flds[MAXFLDS];
- int o_lineno = 0;
- while(fgets(tmpstr,TMPSTRLEN,htsp_in)){
- bool good_entry = true;
- o_lineno++;
- int nflds = split_string(tmpstr,',',flds,MAXFLDS);
- if(nflds == 2){
- string lfta_name = flds[0];
- int htsz = atoi(flds[1]);
- if(htsz>0){
- lfta_htsize[lfta_name] = htsz;
- }else{
- fprintf(stderr,"Warning, line %d of %s is incorrectly formatted, htsize is %d, must be larger than 0.\n",o_lineno,htspec_fl.c_str(),htsz);
- }
- }
- }
- }else{
- fprintf(stderr,"WARNING, no file %s found, using default LFTA hash table sizes.\n",htspec_fl.c_str());
- }
-
-// LFTA vitual interface hash split
- string rtlspec_fl = "rts_load.cfg";
- FILE *rtl_in = NULL;
- map<string, vector<int> > rts_hload;
- if((rtl_in = fopen(rtlspec_fl.c_str(), "r")) != NULL){
- char *flds[MAXFLDS];
- int r_lineno = 0;
- string iface_name;
- vector<int> hload;
- while(fgets(tmpstr,TMPSTRLEN,rtl_in)){
- bool good_entry = true;
- r_lineno++;
- iface_name = "";
- hload.clear();
- int nflds = split_string(tmpstr,',',flds,MAXFLDS);
- if(nflds >1){
- iface_name = flds[0];
- int cumm_h = 0;
- int j;
- for(j=1;j<nflds;++j){
- int h = atoi(flds[j]);
- if(h<=0)
- good_entry = false;
- cumm_h += h;
- hload.push_back(cumm_h);
- }
- }else{
- good_entry = false;
- }
- if(good_entry){
- rts_hload[iface_name] = hload;
- }else{
- fprintf(stderr,"Warning, line %d of %s is malformed, skipping.\n",r_lineno, rtlspec_fl.c_str());
- }
- }
- }
-
-
-
- if(output_query_names){
- if((query_name_output = fopen("query_names.txt","w")) == NULL){
- fprintf(stderr,"Can't open output file %s\n%s\n","query_names.txt",strerror(errno));
- exit(1);
- }
- }
-
- if(output_schema_summary){
- if((schema_summary_output = fopen("schema_summary.txt","w")) == NULL){
- fprintf(stderr,"Can't open output file %s\n%s\n","schema_summary.txt",strerror(errno));
- exit(1);
- }
- }
-
- if((qtree_output = fopen("qtree.xml","w")) == NULL){
- fprintf(stderr,"Can't open output file %s\n%s\n","qtree.xml",strerror(errno));
- exit(1);
- }
- fprintf(qtree_output,"<?xml version='1.0' encoding='ISO-8859-1'?>\n");
- fprintf(qtree_output,"<?xml-stylesheet type='text/xsl' href='qtree.xsl'?>\n");
- fprintf(qtree_output,"<QueryNodes>\n");
-
-
-// Get an initial Schema
- table_list *Schema;
- if(!hfta_only){
-// Parse the table schema definitions.
- fta_parse_result = new fta_parse_t();
- FtaParser_setfileinput(table_schemas_in);
- if(FtaParserparse()){
- fprintf(stderr,"Table schema parse failed.\n");
- exit(1);
- }
- if(fta_parse_result->parse_type != TABLE_PARSE){
- fprintf(stderr,"ERROR, file %s is not a table definition file.\n",schema_file_name.c_str());
- exit(1);
- }
- Schema = fta_parse_result->tables;
-
-// Process schema field inheritance
- int retval;
- retval = Schema->unroll_tables(err_str);
- if(retval){
- fprintf(stderr,"Error processing schema filed inheritance:\n %s\n", err_str.c_str() );
- exit(1);
- }
- }else{
-// hfta only => we will try to fetch schemas from the registry.
-// therefore, start off with an empty schema.
- Schema = new table_list();
- }
-
-
-// Open and parse the external functions file.
- Ext_fcnsParserin = fopen(external_fcns_path.c_str(),"r");
- if(Ext_fcnsParserin == NULL){
- fprintf(stderr,"Warning, can't find external functions definition file (external_fcns.def).\n");
- Ext_fcns = new ext_fcn_list();
- }else{
- if(Ext_fcnsParserparse()){
- fprintf(stderr,"Warning, error parsing external functions definition file (external_fcns.def).\n");
- Ext_fcns = new ext_fcn_list();
- }
- }
- if(Ext_fcns->validate_fcns(err_str)){
- fprintf(stderr,"Error processing external functions definition file (external_fcns.def):\n%s\n",err_str.c_str());
- exit(1);
- }
-
-// Open and parse the interface resources file.
-// ifq_t *ifaces_db = new ifq_t();
-// string ierr;
-// if(ifaces_db->load_ifaces(ifx_fname,use_live_hosts_file,distributed_mode,ierr)){
-// fprintf(stderr,"ERROR, can't load interface resource file %s :\n%s",
-// ifx_fname.c_str(), ierr.c_str());
-// exit(1);
-// }
-// if(ifaces_db->load_ifqs(ifq_fname, ierr)){
-// fprintf(stderr,"ERROR, can't load interface query file %s :\n%s",
-// ifq_fname.c_str(), ierr.c_str());
-// exit(1);
-// }
-
-
-// The LFTA code string.
-// Put the standard preamble here.
-// NOTE: the hash macros, fcns should go into the run time
- map<string, string> lfta_val;
- map<string, string> lfta_prefilter_val;
-
- string lfta_header =
-"#include <limits.h>\n\n"
-"#include \"rts.h\"\n"
-"#include \"fta.h\"\n"
-"#include \"lapp.h\"\n"
-"#include \"rts_udaf.h\"\n\n"
-;
-// Get any locally defined parsing headers
- glob_t glob_result;
- memset(&glob_result, 0, sizeof(glob_result));
-
- // do the glob operation
- int return_value = glob("../../include/lfta/local/*h", GLOB_TILDE, NULL, &glob_result);
- if(return_value == 0){
- for(size_t i = 0; i < glob_result.gl_pathc; ++i) {
- char *flds[1000];
- int nflds = split_string(glob_result.gl_pathv[i],'/',flds,1000);
- lfta_header += "#include \"local/"+string(flds[nflds-1])+"\"\n\n";
- }
- }else{
- fprintf(stderr,"Warning, glob on ../../include/lfta/local/*h failed.\n");
- }
-
-/*
-"#define IS_FILLED(bitmap,bucket) (bitmap[bucket >> 4] & (0x80000000 >> ((bucket & 15)<<1)))\n"
-"#define IS_NEW(bitmap,bucket) (bitmap[bucket >> 4] & (0x80000000 >> (((bucket & 15) << 1) + 1)))\n"
-"#define SET_EMPTY(bitmap,bucket) (bitmap[bucket >> 4] &= (~(0x80000000 >> ((bucket & 15)<<1))))\n"
-"#define SET_FILLED_AND_NEW(bitmap,bucket) (bitmap[bucket >> 4] |= (0xC0000000 >> ((bucket & 15)<<1)))\n"
-*/
-
- lfta_header +=
-"\n"
-"gs_uint64_t (*lfta_prefilter)(void *pkt) = NULL; // set at fta_init\n"
-"\n"
-"#define SLOT_FILLED 0x04\n"
-"#define SLOT_GEN_BITS 0x03\n"
-"#define SLOT_HASH_BITS 0xfffffff8\n"
-"#define SET_BF_BIT(bf_table, bf_num, bf_index, bucket) (bf_table[(((bucket) * (bf_num)) + (bf_index)) >> 3] |= (0x80 >> ((((bucket) * (bf_num)) + (bf_index)) & 7)))\n"
-"#define IS_BF_SET(bf_table, bf_num, bf_index, bucket) (bf_table[(((bucket) * (bf_num))+(bf_index)) >> 3] & (0x80 >> ((((bucket) * (bf_num))+(bf_index)) & 7)))\n"
-"#define SET_BF_EMPTY(bf_table, bf_num, bf_index, bucket) (bf_table[((bucket * bf_num)+bf_index) >> 3] &= (~0x80 >> (((bucket * bf_num)+bf_index) & 7)))\n"
-"\n\n"
-
-"#define lfta_BOOL_to_hash(x) (x)\n"
-"#define lfta_USHORT_to_hash(x) (x)\n"
-"#define lfta_UINT_to_hash(x) (x)\n"
-"#define lfta_IP_to_hash(x) (x)\n"
-"#define lfta_IPV6_to_hash(x) ( (x.v[0]) ^ (x.v[1]) ^ (x.v[2]) ^ (x.v[3]))\n"
-"#define lfta_INT_to_hash(x) (gs_uint32_t)(x)\n"
-"#define lfta_ULLONG_to_hash(x) ( (( (x) >>32)&0xffffffff) ^ ((x)&0xffffffff) )\n"
-"#define lfta_LLONG_to_hash(x) ( (( (x) >>32)&0xffffffff) ^ ((x)&0xffffffff) )\n"
-"#define lfta_FLOAT_to_hash(x) ( (( ((gs_uint64_t)(x)) >>32)&0xffffffff) ^ (((gs_uint64_t)(x))&0xffffffff) )\n"
-"static gs_uint32_t lfta_V_STR_to_hash(struct gs_string x){\n"
-" gs_uint32_t i,ret=0,tmp_sum = 0;\n"
-" for(i=0;i<x.length;++i){\n"
-" tmp_sum |= (x.data[i]) << (8*(i%4));\n"
-" if((i%4) == 3){\n"
-" ret ^= tmp_sum;\n"
-" tmp_sum = 0;\n"
-" }\n"
-" }\n"
-" if((i%4)!=0) ret ^=tmp_sum;\n"
-" return(ret);\n"
-"}\n\n\n";
-
-
-
-//////////////////////////////////////////////////////////////////
-///// Get all of the query parse trees
-
-
- int i,p;
- int hfta_count = 0; // for numeric suffixes to hfta .cc files
-
-//---------------------------
-// Global info needed for post processing.
-
-// Set of operator views ref'd in the query set.
- opview_set opviews;
-// lftas on a per-machine basis.
- map<string, vector<stream_query *> > lfta_mach_lists;
- int nfiles = input_file_names.size();
- vector<stream_query *> hfta_list; // list of hftas.
- map<string, stream_query *> sq_map; // map from query name to stream query.
-
-
-//////////////////////////////////////////
-
-// Open and parse the interface resources file.
- ifq_t *ifaces_db = new ifq_t();
- string ierr;
- if(ifaces_db->load_ifaces(ifx_fname,use_live_hosts_file,distributed_mode,ierr)){
- fprintf(stderr,"ERROR, can't load interface resource file %s :\n%s",
- ifx_fname.c_str(), ierr.c_str());
- exit(1);
- }
- if(ifaces_db->load_ifqs(ifq_fls[0], ierr)){
- fprintf(stderr,"ERROR, can't load interface query file %s :\n%s",
- ifq_fls[0].c_str(), ierr.c_str());
- exit(1);
- }
-
- map<string, string> qname_to_flname; // for detecting duplicate query names
-
-
-
-// Parse the files to create a vector of parse trees.
-// Load qnodes with information to perform a topo sort
-// based on query dependencies.
- vector<query_node *> qnodes; // for topo sort.
- map<string,int> name_node_map; // map query name to qnodes entry
- for(i=0;i<input_file_names.size();i++){
-
- if((fta_in = fopen(input_file_names[i].c_str(), "r")) == NULL) {
- fprintf(stderr,"Can't open input file %s\n%s",input_file_names[i].c_str(),strerror(errno));
- continue;
- }
-fprintf(stderr,"Parsing file %s\n",input_file_names[i].c_str());
-
-// Parse the FTA query
- fta_parse_result = new fta_parse_t();
- FtaParser_setfileinput(fta_in);
- if(FtaParserparse()){
- fprintf(stderr,"FTA parse failed.\n");
- exit(1);
- }
- if(fta_parse_result->parse_type != QUERY_PARSE){
- fprintf(stderr,"ERROR, file %s is not a query file.\n",input_file_names[i].c_str());
- exit(1);
- }
-
-// returns a list of parse trees
- vector<table_exp_t *> qlist = fta_parse_result->parse_tree_list->qlist;
- for(p=0;p<qlist.size();++p){
- table_exp_t *fta_parse_tree = qlist[p];
-// query_parse_trees.push_back(fta_parse_tree);
-
-// compute the default name -- extract from query name
- strcpy(tmpstr,input_file_names[i].c_str());
- char *qname = strrchr(tmpstr,PATH_DELIM);
- if(qname == NULL)
- qname = tmpstr;
- else
- qname++;
- char *qname_end = strchr(qname,'.');
- if(qname_end != NULL) *qname_end = '\0';
- string qname_str = qname;
- string imputed_qname = impute_query_name(fta_parse_tree, qname_str);
-
-// Deternmine visibility. Should I be attaching all of the output methods?
- if(qname_to_ospec.count(imputed_qname)>0)
- fta_parse_tree->set_visible(true);
- else
- fta_parse_tree->set_visible(false);
-
-
-// Create a manipulable repesentation of the parse tree.
-// the qnode inherits the visibility assigned to the parse tree.
- int pos = qnodes.size();
- qnodes.push_back( new query_node(pos, imputed_qname, input_file_names[i], fta_parse_tree ));
- name_node_map[ qnodes[pos]->name ] = pos;
-//printf("name_node_map[ %s ] = %d\n",qnodes[pos]->name.c_str(),pos);
-// qnames.push_back(impute_query_name(fta_parse_tree, qname_str));
-// qfiles.push_back(i);
-
-// Check for duplicate query names
-// NOTE : in hfta-only generation, I should
-// also check with the names of the registered queries.
- if(qname_to_flname.count(qnodes[pos]->name) > 0){
- fprintf(stderr,"ERROR duplicate query name %s in files %s and %s.\n",
- qnodes[pos]->name.c_str(), (qname_to_flname[qnodes[pos]->name]).c_str(), input_file_names[i].c_str());
- exit(1);
- }
- if(Schema->find_tbl(qnodes[pos]->name) >= 0){
- fprintf(stderr,"ERROR duplicate query name %s (file %s) was already defined as a PROTOCOL.\n",
- qnodes[pos]->name.c_str(), input_file_names[i].c_str());
- exit(1);
- }
- qname_to_flname[qnodes[pos]->name] = input_file_names[i].c_str();
-
-
- }
- }
-
-// Add the library queries
-
- int pos;
- for(pos=0;pos<qnodes.size();++pos){
- int fi;
- for(fi = 0;fi<qnodes[pos]->refd_tbls.size();++fi){
- string src_tbl = qnodes[pos]->refd_tbls[fi];
- if(qname_to_flname.count(src_tbl) == 0){
- int last_sep = src_tbl.find_last_of('/');
- if(last_sep != string::npos){
-fprintf(stderr,"Finding library query %s for query %s\n",src_tbl.c_str(),qnodes[pos]->name.c_str());
- string target_qname = src_tbl.substr(last_sep+1);
- string qpathname = library_path + src_tbl + ".gsql";
- if((fta_in = fopen(qpathname.c_str(), "r")) == NULL) {
- fprintf(stderr,"Can't open library file %s, referenced by query %s in file %s\n\t%s\n",qpathname.c_str(),qnodes[pos]->name.c_str(), qname_to_flname[qnodes[pos]->name].c_str(), strerror(errno));
- exit(1);
- fprintf(stderr,"After exit\n");
- }
-fprintf(stderr,"Parsing file %s\n",qpathname.c_str());
-// Parse the FTA query
- fta_parse_result = new fta_parse_t();
- FtaParser_setfileinput(fta_in);
- if(FtaParserparse()){
- fprintf(stderr,"FTA parse failed.\n");
- exit(1);
- }
- if(fta_parse_result->parse_type != QUERY_PARSE){
- fprintf(stderr,"ERROR, file %s is not a query file.\n",input_file_names[i].c_str());
- exit(1);
- }
-
- map<string, int> local_query_map;
- vector<string> local_query_names;
- vector<table_exp_t *> qlist = fta_parse_result->parse_tree_list->qlist;
- for(p=0;p<qlist.size();++p){
- table_exp_t *fta_parse_tree = qlist[p];
- fta_parse_tree->set_visible(false); // assumed to not produce output
- string imputed_qname = impute_query_name(fta_parse_tree, target_qname);
- if(imputed_qname == target_qname)
- imputed_qname = src_tbl;
- if(local_query_map.count(imputed_qname)>0){
- fprintf(stderr,"ERROR, multiple queries named %s in library file %s\n",imputed_qname.c_str(), qpathname.c_str());
- exit(1);
- }
- local_query_map[ imputed_qname ] = p;
- local_query_names.push_back(imputed_qname);
- }
-
- if(local_query_map.count(src_tbl)==0){
- fprintf(stderr,"ERROR, library query file %s has no query named %s\n",qpathname.c_str(),target_qname.c_str());
- exit(1);
- }
-
- vector<int> worklist;
- set<int> added_queries;
- vector<query_node *> new_qnodes;
- worklist.push_back(local_query_map[target_qname]);
- added_queries.insert(local_query_map[target_qname]);
- int qq;
- int qpos = qnodes.size();
- for(qq=0;qq<worklist.size();++qq){
- int q_id = worklist[qq];
- query_node *new_qnode = new query_node(qpos+qq, local_query_names[q_id], qpathname, qlist[q_id] );
- new_qnodes.push_back( new_qnode);
- vector<string> refd_tbls = new_qnode->refd_tbls;
- int ff;
- for(ff = 0;ff<refd_tbls.size();++ff){
- if(local_query_map.count(refd_tbls[ff])>0 && added_queries.count(local_query_map[refd_tbls[ff]])==0){
-
- if(name_node_map.count(refd_tbls[ff])>0){
- fprintf(stderr,"ERROR, query %s occurs both in the regular query set, file %s, and in library file %s\n",refd_tbls[ff].c_str(), qname_to_flname[refd_tbls[ff]].c_str(), qpathname.c_str() );
- exit(1);
- }else{
- worklist.push_back(local_query_map[refd_tbls[ff]]);
- }
- }
- }
- }
-
- for(qq=0;qq<new_qnodes.size();++qq){
- int qpos = qnodes.size();
- qnodes.push_back(new_qnodes[qq]);
- name_node_map[qnodes[qpos]->name ] = qpos;
- qname_to_flname[qnodes[qpos]->name ] = qpathname;
- }
- }
- }
- }
- }
-
-
-
-
-
-
-
-
-//---------------------------------------
-
-
-// Add the UDOPS.
-
- string udop_missing_sources;
- for(i=0;i<qnodes.size();++i){
- int fi;
- for(fi = 0;fi<qnodes[i]->refd_tbls.size();++fi){
- int sid = Schema->find_tbl(qnodes[i]->refd_tbls[fi]);
- if(sid >= 0){
- if(Schema->get_schema_type(sid) == OPERATOR_VIEW_SCHEMA){
- if(name_node_map.count(qnodes[i]->refd_tbls[fi]) == 0){
- int pos = qnodes.size();
- qnodes.push_back( new query_node(pos, qnodes[i]->refd_tbls[fi], Schema));
- name_node_map[ qnodes[pos]->name ] = pos;
- qnodes[pos]->is_externally_visible = false; // its visible
- // Need to mark the source queries as visible.
- int si;
- string missing_sources = "";
- for(si=0;si<qnodes[pos]->refd_tbls.size();++si){
- string src_tbl = qnodes[pos]->refd_tbls[si];
- if(name_node_map.count(src_tbl)==0){
- missing_sources += src_tbl + " ";
- }
- }
- if(missing_sources != ""){
- udop_missing_sources += "\tUDOP "+qnodes[pos]->name+" references undefined tables "+missing_sources+"\n";
- }
- }
- }
- }
- }
- }
- if(udop_missing_sources != ""){
- fprintf(stderr,"ERROR, User Defined OPerators reference source tables that are not part of the query set:\n%s",udop_missing_sources.c_str());
- exit(1);
- }
-
-
-
-////////////////////////////////////////////////////////////////////
-/// Check parse trees to verify that some
-/// global properties are met :
-/// if q1 reads from q2, then
-/// q2 is processed before q1
-/// q1 can supply q2's parameters
-/// Verify there is no cycle in the reads-from graph.
-
-// Compute an order in which to process the
-// queries.
-
-// Start by building the reads-from lists.
-//
-
- for(i=0;i<qnodes.size();++i){
- int qi, fi;
- vector<string> refd_tbls = qnodes[i]->refd_tbls;
- for(fi = 0;fi<refd_tbls.size();++fi){
- if(name_node_map.count(refd_tbls[fi])>0){
-//printf("query %d (%s) reads from %s (%d)\n", i, qnodes[i]->name.c_str(),refd_tbls[fi].c_str(),name_node_map[refd_tbls[fi]]);
- (qnodes[i]->reads_from).insert(name_node_map[refd_tbls[fi]]);
- }
- }
- }
-
-
-// If one query reads the result of another,
-// check for parameter compatibility. Currently it must
-// be an exact match. I will move to requiring
-// containment after re-ordering, but will require
-// some analysis for code generation which is not
-// yet in place.
-//printf("There are %d query nodes.\n",qnodes.size());
-
-
- for(i=0;i<qnodes.size();++i){
- vector<var_pair_t *> target_params = qnodes[i]->params;
- for(si=qnodes[i]->reads_from.begin();si!=qnodes[i]->reads_from.end();++si){
- vector<var_pair_t *> source_params = qnodes[(*si)]->params;
- if(target_params.size() != source_params.size()){
- fprintf(stderr,"ERROR, query %s (in file %s) reads from %s (in file %s), but they do not have identical parameters.\n",qnodes[i]->name.c_str(), qnodes[i]->file.c_str(), qnodes[(*si)]->name.c_str(), qnodes[(*si)]->file.c_str());
- exit(1);
- }
- int p;
- for(p=0;p<target_params.size();++p){
- if(! (target_params[p]->name == source_params[p]->name &&
- target_params[p]->val == source_params[p]->val ) ){
- fprintf(stderr,"ERROR, query %s (in file %s) reads from %s (in file %s), but they do not have identical parameters.\n",qnodes[i]->name.c_str(), qnodes[i]->file.c_str(), qnodes[(*si)]->name.c_str(), qnodes[(*si)]->file.c_str());
- exit(1);
- }
- }
- }
- }
-
-
-// Find the roots.
-// Start by counting inedges.
- for(i=0;i<qnodes.size();++i){
- for(si=qnodes[i]->reads_from.begin();si!=qnodes[i]->reads_from.end();++si){
- qnodes[(*si)]->n_consumers++;
- }
- }
-
-// The roots are the nodes with indegree zero.
- set<int> roots;
- for(i=0;i<qnodes.size();++i){
- if(qnodes[i]->n_consumers == 0){
- if(qnodes[i]->is_externally_visible == false){
- fprintf(stderr,"WARNING: query %s (file %s) is a root query but it isn't visible. Ignored.\n",qnodes[i]->name.c_str(), qnodes[i]->file.c_str());
- }
- roots.insert(i);
- }
- }
-
-// Remove the parts of the subtree that produce no output.
- set<int> valid_roots;
- set<int> discarded_nodes;
- set<int> candidates;
- while(roots.size() >0){
- for(si=roots.begin();si!=roots.end();++si){
- if(qnodes[(*si)]->is_externally_visible){
- valid_roots.insert((*si));
- }else{
- discarded_nodes.insert((*si));
- set<int>::iterator sir;
- for(sir=qnodes[(*si)]->reads_from.begin(); sir!=qnodes[(*si)]->reads_from.end();++sir){
- qnodes[(*sir)]->n_consumers--;
- if(qnodes[(*sir)]->n_consumers == 0)
- candidates.insert( (*sir));
- }
- }
- }
- roots = candidates;
- candidates.clear();
- }
- roots = valid_roots;
- if(discarded_nodes.size()>0){
- fprintf(stderr,"Warning, the following queries were discarded because they produce no output:\n");
- int di = 0;
- for(si=discarded_nodes.begin();si!=discarded_nodes.end();++si){
- if(di>0 && (di%8)==0) fprintf(stderr,"\n");
- di++;
- fprintf(stderr," %s",qnodes[(*si)]->name.c_str());
- }
- fprintf(stderr,"\n");
- }
-
-// Compute the sources_to set, ignoring discarded nodes.
- for(i=0;i<qnodes.size();++i){
- if(discarded_nodes.count(i)==0)
- for(si=qnodes[i]->reads_from.begin();si!=qnodes[i]->reads_from.end();++si){
- qnodes[(*si)]->sources_to.insert(i);
- }
- }
-
-
-// Find the nodes that are shared by multiple visible subtrees.
-// THe roots become inferred visible nodes.
-
-// Find the visible nodes.
- vector<int> visible_nodes;
- for(i=0;i<qnodes.size();i++){
- if(qnodes[i]->is_externally_visible){
- visible_nodes.push_back(i);
- }
- }
-
-// Find UDOPs referenced by visible nodes.
- list<int> workq;
- for(i=0;i<visible_nodes.size();++i){
- workq.push_back(visible_nodes[i]);
- }
- while(!workq.empty()){
- int node = workq.front();
- workq.pop_front();
- set<int>::iterator children;
- if(qnodes[node]->is_udop && qnodes[node]->is_externally_visible == false){
- qnodes[node]->is_externally_visible = true;
- visible_nodes.push_back(node);
- for(children=qnodes[node]->reads_from.begin();children!=qnodes[node]->reads_from.end();++children){
- if(qnodes[(*children)]->is_externally_visible == false){
- qnodes[(*children)]->is_externally_visible = true;
- visible_nodes.push_back((*children));
- }
- }
- }
- for(children=qnodes[node]->reads_from.begin();children!=qnodes[node]->reads_from.end();++children){
- workq.push_back((*children));
- }
- }
-
- bool done = false;
- while(!done){
-// reset the nodes
- for(i=0;i<qnodes.size();i++){
- qnodes[i]->subtree_roots.clear();
- }
-
-// Walk the tree defined by a visible node, not descending into
-// subtrees rooted by a visible node. Mark the node visited with
-// the visible node ID.
- for(i=0;i<visible_nodes.size();++i){
- set<int> vroots;
- vroots.insert(visible_nodes[i]);
- while(vroots.size()>0){
- for(si=vroots.begin();si!=vroots.end();++si){
- qnodes[(*si)]->subtree_roots.insert(visible_nodes[i]);
-
- set<int>::iterator sir;
- for(sir=qnodes[(*si)]->reads_from.begin(); sir!=qnodes[(*si)]->reads_from.end();++sir){
- if(! qnodes[(*sir)]->is_externally_visible){
- candidates.insert( (*sir));
- }
- }
- }
- vroots = candidates;
- candidates.clear();
- }
- }
-// Find the nodes in multiple visible node subtrees, but with no parent
-// that has is in multile visible node subtrees. Mark these as inferred visible nodes.
- done = true; // until proven otherwise
- for(i=0;i<qnodes.size();i++){
- if(qnodes[i]->subtree_roots.size()>1){
- bool is_new_root = true;
- set<int>::iterator sir;
- for(sir=qnodes[i]->sources_to.begin(); sir!=qnodes[i]->sources_to.end();++sir){
- if(qnodes[(*sir)]->subtree_roots.size()>1)
- is_new_root = false;
- }
- if(is_new_root){
- qnodes[i]->is_externally_visible = true;
- qnodes[i]->inferred_visible_node = true;
- visible_nodes.push_back(i);
- done = false;
- }
- }
- }
- }
-
-
-
-
-
-// get visible nodes in topo ordering.
-// for(i=0;i<qnodes.size();i++){
-// qnodes[i]->n_consumers = qnodes[i]->sources_to.size();
-// }
- vector<int> process_order;
- while(roots.size() >0){
- for(si=roots.begin();si!=roots.end();++si){
- if(discarded_nodes.count((*si))==0){
- process_order.push_back( (*si) );
- }
- set<int>::iterator sir;
- for(sir=qnodes[(*si)]->reads_from.begin(); sir!=qnodes[(*si)]->reads_from.end();++sir){
- qnodes[(*sir)]->n_consumers--;
- if(qnodes[(*sir)]->n_consumers == 0)
- candidates.insert( (*sir));
- }
- }
- roots = candidates;
- candidates.clear();
- }
-
-
-//printf("process_order.size() =%d\n",process_order.size());
-
-// Search for cyclic dependencies
- string found_dep;
- for(i=0;i<qnodes.size();++i){
- if(discarded_nodes.count(i)==0 && qnodes[i]->n_consumers > 0){
- if(found_dep.size() != 0) found_dep += ", ";
- found_dep += "query "+qnodes[i]->name+" (file "+qnodes[i]->file+")";
- }
- }
- if(found_dep.size()>0){
- fprintf(stderr,"ERROR, the following queries contain a cyclic reads-from dependency:\n%s\n",found_dep.c_str());
- exit(1);
- }
-
-// Get a list of query sets, in the order to be processed.
-// Start at visible root and do bfs.
-// The query set includes queries referenced indirectly,
-// as sources for user-defined operators. These are needed
-// to ensure that they are added to the schema, but are not part
-// of the query tree.
-
-// stream_node_sets contains queries reachable only through the
-// FROM clause, so I can tell which queries to add to the stream
-// query. (DISABLED, UDOPS are integrated, does this cause problems?)
-
-// NOTE: this code works because in order for data to be
-// read by multiple hftas, the node must be externally visible.
-// But visible nodes define roots of process sets.
-// internally visible nodes can feed data only
-// to other nodes in the same query file.
-// Therefore, any access can be restricted to a file,
-// hfta output sharing is done only on roots
-// never on interior nodes.
-
-
-
-
-// Conpute the base collection of hftas.
- vector<hfta_node *> hfta_sets;
- map<string, int> hfta_name_map;
-// vector< vector<int> > process_sets;
-// vector< set<int> > stream_node_sets;
- reverse(process_order.begin(), process_order.end()); // get listing in reverse order.
- // i.e. process leaves 1st.
- for(i=0;i<process_order.size();++i){
- if(qnodes[process_order[i]]->is_externally_visible == true){
-//printf("Visible.\n");
- int root = process_order[i];
- hfta_node *hnode = new hfta_node();
- hnode->name = qnodes[root]-> name;
- hnode->source_name = qnodes[root]-> name;
- hnode->is_udop = qnodes[root]->is_udop;
- hnode->inferred_visible_node = qnodes[root]->inferred_visible_node;
-
- vector<int> proc_list; proc_list.push_back(root);
-// Ensure that nodes are added only once.
- set<int> proc_set; proc_set.insert(root);
- roots.clear(); roots.insert(root);
- candidates.clear();
- while(roots.size()>0){
- for(si=roots.begin();si!=roots.end();++si){
-//printf("Processing root %d\n",(*si));
- set<int>::iterator sir;
- for(sir=qnodes[(*si)]->reads_from.begin(); sir!=qnodes[(*si)]->reads_from.end();++sir){
-//printf("reads fom %d\n",(*sir));
- if(qnodes[(*sir)]->is_externally_visible==false){
- candidates.insert( (*sir) );
- if(proc_set.count( (*sir) )==0){
- proc_set.insert( (*sir) );
- proc_list.push_back( (*sir) );
- }
- }
- }
- }
- roots = candidates;
- candidates.clear();
- }
-
- reverse(proc_list.begin(), proc_list.end());
- hnode->query_node_indices = proc_list;
- hfta_name_map[hnode->name] = hfta_sets.size();
- hfta_sets.push_back(hnode);
- }
- }
-
-// Compute the reads_from / sources_to graphs for the hftas.
-
- for(i=0;i<hfta_sets.size();++i){
- hfta_node *hnode = hfta_sets[i];
- for(q=0;q<hnode->query_node_indices.size();q++){
- query_node *qnode = qnodes[hnode->query_node_indices[q]];
- for(s=0;s<qnode->refd_tbls.size();++s){
- if(hfta_name_map.count(qnode->refd_tbls[s])){
- int other_hfta = hfta_name_map[qnode->refd_tbls[s]];
- hnode->reads_from.insert(other_hfta);
- hfta_sets[other_hfta]->sources_to.insert(i);
- }
- }
- }
- }
-
-// Compute a topological sort of the hfta_sets.
-
- vector<int> hfta_topsort;
- workq.clear();
- int hnode_srcs[hfta_sets.size()];
- for(i=0;i<hfta_sets.size();++i){
- hnode_srcs[i] = 0;
- if(hfta_sets[i]->sources_to.size() == 0)
- workq.push_back(i);
- }
-
- while(! workq.empty()){
- int node = workq.front();
- workq.pop_front();
- hfta_topsort.push_back(node);
- set<int>::iterator stsi;
- for(stsi=hfta_sets[node]->reads_from.begin();stsi!=hfta_sets[node]->reads_from.end();++stsi){
- int parent = (*stsi);
- hnode_srcs[parent]++;
- if(hnode_srcs[parent] == hfta_sets[parent]->sources_to.size()){
- workq.push_back(parent);
- }
- }
- }
-
-// Decorate hfta nodes with the level of parallelism given as input.
-
- map<string, int>::iterator msii;
- for(msii=hfta_parallelism.begin();msii!=hfta_parallelism.end();++msii){
- string hfta_name = (*msii).first;
- int par = (*msii).second;
- if(hfta_name_map.count(hfta_name) > 0){
- hfta_sets[ hfta_name_map[hfta_name] ]->n_parallel = par;
- }else{
- fprintf(stderr,"Warning, hfta_parallelism.cfg has an entry for %s, but its not a hfta.\n",hfta_name.c_str());
- }
- }
-
-// Propagate levels of parallelism: children should have a level of parallelism
-// as large as any of its parents. Adjust children upwards to compensate.
-// Start at parents and adjust children, auto-propagation will occur.
-
- for(i=hfta_sets.size()-1;i>=0;i--){
- set<int>::iterator stsi;
- for(stsi=hfta_sets[i]->reads_from.begin();stsi!=hfta_sets[i]->reads_from.end();++stsi){
- if(hfta_sets[i]->n_parallel > hfta_sets[ (*stsi) ]->n_parallel){
- hfta_sets[ (*stsi) ]->n_parallel = hfta_sets[i]->n_parallel;
- }
- }
- }
-
-// Before all the name mangling, check if therey are any output_spec.cfg
-// or hfta_parallelism.cfg entries that do not have a matching query.
-
- string dangling_ospecs = "";
- for(msii=qname_to_ospec.begin();msii!=qname_to_ospec.end();++msii){
- string oq = (*msii).first;
- if(hfta_name_map.count(oq) == 0){
- dangling_ospecs += " "+(*msii).first;
- }
- }
- if(dangling_ospecs!=""){
- fprintf(stderr,"WARNING, the following entries in output_spec.cfg don't have a matching query: %s\n",dangling_ospecs.c_str());
- }
-
- string dangling_par = "";
- for(msii=hfta_parallelism.begin();msii!=hfta_parallelism.end();++msii){
- string oq = (*msii).first;
- if(hfta_name_map.count(oq) == 0){
- dangling_par += " "+(*msii).first;
- }
- }
- if(dangling_par!=""){
- fprintf(stderr,"WARNING, the following entries in hfta_parallelism.cfg don't have a matching query: %s\n",dangling_par.c_str());
- }
-
-
-
-// Replicate parallelized hftas. Do __copyX name mangling. Adjust
-// FROM clauses: retarget any name which is an internal node, and
-// any which is in hfta_sets (and which is parallelized). Add Merge nodes
-// when the source hfta has more parallelism than the target node.
-// Add new nodes to qnodes and hfta_nodes wth the appropriate linkages.
-
-
- int n_original_hfta_sets = hfta_sets.size();
- for(i=0;i<n_original_hfta_sets;++i){
- if(hfta_sets[i]->n_parallel > 1){
- hfta_sets[i]->do_generation =false; // set the deletion flag for this entry.
- set<string> local_nodes; // names of query nodes in the hfta.
- for(h=0;h<hfta_sets[i]->query_node_indices.size();++h){
- local_nodes.insert(qnodes[ hfta_sets[i]->query_node_indices[h] ]->name);
- }
-
- for(p=0;p<hfta_sets[i]->n_parallel;++p){
- string mangler = "__copy"+int_to_string(p);
- hfta_node *par_hfta = new hfta_node();
- par_hfta->name = hfta_sets[i]->name + mangler;
- par_hfta->source_name = hfta_sets[i]->name;
- par_hfta->is_udop = hfta_sets[i]->is_udop;
- par_hfta->inferred_visible_node = hfta_sets[i]->inferred_visible_node;
- par_hfta->n_parallel = hfta_sets[i]->n_parallel;
- par_hfta->parallel_idx = p;
-
- map<string, int> par_qnode_map; // qnode name-to-idx, aids dependency tracking.
-
-// Is it a UDOP?
- if(hfta_sets[i]->is_udop){
- int root = hfta_sets[i]->query_node_indices[0];
-
- string unequal_par_sources;
- set<int>::iterator rfsii;
- for(rfsii=hfta_sets[i]->reads_from.begin();rfsii!=hfta_sets[i]->reads_from.end();++rfsii){
- if(hfta_sets[(*rfsii)]->n_parallel != par_hfta->n_parallel){
- unequal_par_sources = hfta_sets[(*rfsii)]->name+" ("+int_to_string(hfta_sets[(*rfsii)]->n_parallel)+") ";
- }
- }
- if(unequal_par_sources != ""){
- fprintf(stderr,"ERROR, UDOP %s has parallelism %d, but some of its sources have a different parallelism: %s\n",hfta_sets[i]->name.c_str(), hfta_sets[i]->n_parallel, unequal_par_sources.c_str());
- exit(1);
- }
-
- int rti;
- vector<string> new_sources;
- for(rti=0;rti<qnodes[root]->refd_tbls.size();++rti){
- new_sources.push_back(qnodes[root]->refd_tbls[rti]+mangler);
- }
-
- query_node *new_qn = new query_node(qnodes.size(), qnodes[root]->name, Schema);
- new_qn->name += mangler;
- new_qn->mangler = mangler;
- new_qn->refd_tbls = new_sources;
- par_hfta->query_node_indices.push_back(qnodes.size());
- par_qnode_map[new_qn->name] = qnodes.size();
- name_node_map[ new_qn->name ] = qnodes.size();
- qnodes.push_back(new_qn);
- }else{
-// regular query node
- for(h=0;h<hfta_sets[i]->query_node_indices.size();++h){
- int hqn_idx = hfta_sets[i]->query_node_indices[h];
- table_exp_t *dup_pt = dup_table_exp(qnodes[hqn_idx]->parse_tree);
-// rehome the from clause on mangled names.
-// create merge nodes as needed for external sources.
- for(f=0;f<dup_pt->fm->tlist.size();++f){
- if(local_nodes.count(dup_pt->fm->tlist[f]->schema_name)>0){
- dup_pt->fm->tlist[f]->schema_name += mangler;
- }else if(hfta_name_map.count(dup_pt->fm->tlist[f]->schema_name)>0){
-// Ref's an external HFTA. No parallelism => leave as is. Else, if level of parallelism of the two hftas is the same, rename by the mangler. Else, there mnust be more sources, so create a merge node.
- int other_hidx = hfta_name_map[dup_pt->fm->tlist[f]->schema_name];
- if(par_hfta->n_parallel == hfta_sets[other_hidx]->n_parallel){
- dup_pt->fm->tlist[f]->schema_name += mangler;
- }else{
- vector<string> src_tbls;
- int stride = hfta_sets[other_hidx]->n_parallel / par_hfta->n_parallel;
- if(stride == 0){
- fprintf(stderr,"INTERNAL ERROR, parent hfta %s has a larger parallelism than its child %s\n",par_hfta->name.c_str(), hfta_sets[other_hidx]->name.c_str());
- exit(1);
- }
- for(s=0;s<stride;++s){
- string ext_src_name = hfta_sets[other_hidx]->name+"__copy"+int_to_string(s+stride*p);
- src_tbls.push_back(ext_src_name);
- }
- table_exp_t *merge_pt = table_exp_t::make_deferred_merge(src_tbls);
- string merge_node_name = qnodes[hqn_idx]->name+"__copy"+int_to_string(p)+"__merge"+int_to_string(f);
- dup_pt->fm->tlist[f]->schema_name = merge_node_name;
-// Make a qnode to represent the new merge node
- query_node *qn_pt = new query_node(qnodes.size(),merge_node_name, qnodes[hqn_idx]->file,merge_pt);
- qn_pt->refd_tbls = src_tbls;
- qn_pt->is_udop = false;
- qn_pt->is_externally_visible = false;
- qn_pt->inferred_visible_node = false;
- par_hfta->query_node_indices.push_back(qnodes.size());
- par_qnode_map[merge_node_name] = qnodes.size();
- name_node_map[ merge_node_name ] = qnodes.size();
- qnodes.push_back(qn_pt);
- }
- }
- }
- query_node *new_qn = new query_node(qnodes.size(),qnodes[hqn_idx]->name+"__copy"+int_to_string(p),qnodes[hqn_idx]->file,dup_pt);
- for(f=0;f<dup_pt->fm->tlist.size();++f){
- new_qn->refd_tbls.push_back(dup_pt->fm->tlist[f]->schema_name);
- }
- new_qn->params = qnodes[hqn_idx]->params;
- new_qn->is_udop = false;
- new_qn->is_externally_visible = qnodes[hqn_idx]->is_externally_visible;
- new_qn->inferred_visible_node = qnodes[hqn_idx]->inferred_visible_node;
- par_hfta->query_node_indices.insert(par_hfta->query_node_indices.begin(),qnodes.size());
- par_qnode_map[new_qn->name] = qnodes.size();
- name_node_map[ new_qn->name ] = qnodes.size();
- qnodes.push_back(new_qn);
- }
- }
- hfta_name_map[par_hfta->name] = hfta_sets.size();
- hfta_sets.push_back(par_hfta);
- }
- }else{
-// This hfta isn't being parallelized, but add merge nodes for any parallelized
-// hfta sources.
- if(!hfta_sets[i]->is_udop){
- for(h=0;h<hfta_sets[i]->query_node_indices.size();++h){
- int hqn_idx = hfta_sets[i]->query_node_indices[h];
- for(f=0;f<qnodes[hqn_idx]->parse_tree->fm->tlist.size();++f){
- if(hfta_name_map.count(qnodes[hqn_idx]->parse_tree->fm->tlist[f]->schema_name)>0){
-// Ref's an external HFTA. No parallelism => leave as is. Else, if level of parallelism of the two hftas is the same, rename by the mangler. Else, there mnust be more sources, so create a merge node.
- int other_hidx = hfta_name_map[qnodes[hqn_idx]->parse_tree->fm->tlist[f]->schema_name];
- if(hfta_sets[i]->n_parallel != hfta_sets[other_hidx]->n_parallel){
- vector<string> src_tbls;
- for(s=0;s<hfta_sets[other_hidx]->n_parallel;++s){
- string ext_src_name = hfta_sets[other_hidx]->name+"__copy"+int_to_string(s);
- src_tbls.push_back(ext_src_name);
- }
- table_exp_t *merge_pt = table_exp_t::make_deferred_merge(src_tbls);
- string merge_node_name = qnodes[hqn_idx]->name+"__merge"+int_to_string(f);
- qnodes[hqn_idx]->parse_tree->fm->tlist[f]->schema_name = merge_node_name;
-// Make a qnode to represent the new merge node
- query_node *qn_pt = new query_node(qnodes.size(),merge_node_name, qnodes[hqn_idx]->file,merge_pt);
- qn_pt->refd_tbls = src_tbls;
- qn_pt->is_udop = false;
- qn_pt->is_externally_visible = false;
- qn_pt->inferred_visible_node = false;
- hfta_sets[i]->query_node_indices.insert(hfta_sets[i]->query_node_indices.begin(),qnodes.size());
- name_node_map[ merge_node_name ] = qnodes.size();
- qnodes.push_back(qn_pt);
- }
- }
- }
- }
- }
- }
- }
-
-// Rebuild the reads_from / sources_to lists in the qnodes
- for(q=0;q<qnodes.size();++q){
- qnodes[q]->reads_from.clear();
- qnodes[q]->sources_to.clear();
- }
- for(q=0;q<qnodes.size();++q){
- for(s=0;s<qnodes[q]->refd_tbls.size();++s){
- if(name_node_map.count(qnodes[q]->refd_tbls[s])>0){
- int rf = name_node_map[qnodes[q]->refd_tbls[s]];
- qnodes[q]->reads_from.insert(rf);
- qnodes[rf]->sources_to.insert(q);
- }
- }
- }
-
-// Rebuild the reads_from / sources_to lists in hfta_sets
- for(q=0;q<hfta_sets.size();++q){
- hfta_sets[q]->reads_from.clear();
- hfta_sets[q]->sources_to.clear();
- }
- for(q=0;q<hfta_sets.size();++q){
- for(s=0;s<hfta_sets[q]->query_node_indices.size();++s){
- int node = hfta_sets[q]->query_node_indices[s];
- set<int>::iterator rfsii;
- for(rfsii=qnodes[node]->reads_from.begin();rfsii!=qnodes[node]->reads_from.end();++rfsii){
- if(hfta_name_map.count(qnodes[(*rfsii)]->name)>0){
- hfta_sets[q]->reads_from.insert(hfta_name_map[qnodes[(*rfsii)]->name]);
- hfta_sets[hfta_name_map[qnodes[(*rfsii)]->name]]->sources_to.insert(q);
- }
- }
- }
- }
-
-/*
-for(q=0;q<qnodes.size();++q){
- printf("qnode %d reads-from %d:",q,qnodes[q]->reads_from.size());
- set<int>::iterator rsii;
- for(rsii=qnodes[q]->reads_from.begin();rsii!=qnodes[q]->reads_from.end();++rsii)
- printf(" %d",(*rsii));
- printf(", and sources-to %d:",qnodes[q]->sources_to.size());
- for(rsii=qnodes[q]->sources_to.begin();rsii!=qnodes[q]->sources_to.end();++rsii)
- printf(" %d",(*rsii));
- printf("\n");
-}
-
-for(q=0;q<hfta_sets.size();++q){
- if(hfta_sets[q]->do_generation==false)
- continue;
- printf("hfta %d (%s) reads-from %d:",q,hfta_sets[q]->name.c_str(),hfta_sets[q]->reads_from.size());
- set<int>::iterator rsii;
- for(rsii=hfta_sets[q]->reads_from.begin();rsii!=hfta_sets[q]->reads_from.end();++rsii)
- printf(" %d",(*rsii));
- printf(", and sources-to %d:",hfta_sets[q]->sources_to.size());
- for(rsii=hfta_sets[q]->sources_to.begin();rsii!=hfta_sets[q]->sources_to.end();++rsii)
- printf(" %d",(*rsii));
- printf("\n");
-}
-*/
-
-
-
-// Re-topo sort the hftas
- hfta_topsort.clear();
- workq.clear();
- int hnode_srcs_2[hfta_sets.size()];
- for(i=0;i<hfta_sets.size();++i){
- hnode_srcs_2[i] = 0;
- if(hfta_sets[i]->sources_to.size() == 0 && hfta_sets[i]->do_generation){
- workq.push_back(i);
- }
- }
-
- while(workq.empty() == false){
- int node = workq.front();
- workq.pop_front();
- hfta_topsort.push_back(node);
- set<int>::iterator stsii;
- for(stsii=hfta_sets[node]->reads_from.begin();stsii!=hfta_sets[node]->reads_from.end();++stsii){
- int child = (*stsii);
- hnode_srcs_2[child]++;
- if(hnode_srcs_2[child] == hfta_sets[child]->sources_to.size()){
- workq.push_back(child);
- }
- }
- }
-
-// Ensure that all of the query_node_indices in hfta_sets are topologically
-// sorted, don't rely on assumptions that all transforms maintain some kind of order.
- for(i=0;i<hfta_sets.size();++i){
- if(hfta_sets[i]->do_generation){
- map<int,int> n_accounted;
- vector<int> new_order;
- workq.clear();
- vector<int>::iterator vii;
- for(vii=hfta_sets[i]->query_node_indices.begin();vii!=hfta_sets[i]->query_node_indices.end();++vii){
- n_accounted[(*vii)]= 0;
- }
- for(vii=hfta_sets[i]->query_node_indices.begin();vii!=hfta_sets[i]->query_node_indices.end();++vii){
- set<int>::iterator rfsii;
- for(rfsii=qnodes[(*vii)]->reads_from.begin();rfsii!=qnodes[(*vii)]->reads_from.end();++rfsii){
- if(n_accounted.count((*rfsii)) == 0){
- n_accounted[(*vii)]++;
- }
- }
- if(n_accounted[(*vii)] == qnodes[(*vii)]->reads_from.size()){
- workq.push_back((*vii));
- }
- }
-
- while(workq.empty() == false){
- int node = workq.front();
- workq.pop_front();
- new_order.push_back(node);
- set<int>::iterator stsii;
- for(stsii=qnodes[node]->sources_to.begin();stsii!=qnodes[node]->sources_to.end();++stsii){
- if(n_accounted.count((*stsii))){
- n_accounted[(*stsii)]++;
- if(n_accounted[(*stsii)] == qnodes[(*stsii)]->reads_from.size()){
- workq.push_back((*stsii));
- }
- }
- }
- }
- hfta_sets[i]->query_node_indices = new_order;
- }
- }
-
-
-
-
-
-/// Global checkng is done, start the analysis and translation
-/// of the query parse tree in the order specified by process_order
-
-
-// Get a list of the LFTAs for global lfta optimization
-// TODO: separate building operators from spliting lftas,
-// that will make optimizations such as predicate pushing easier.
- vector<stream_query *> lfta_list;
-
- stream_query *rootq;
-
- int qi,qj;
-
- for(qi=hfta_topsort.size()-1;qi>=0;--qi){
-
- int hfta_id = hfta_topsort[qi];
- vector<int> curr_list = hfta_sets[hfta_id]->query_node_indices;
-
-
-
-// Two possibilities, either its a UDOP, or its a collection of queries.
-// if(qnodes[curr_list.back()]->is_udop)
- if(hfta_sets[hfta_id]->is_udop){
- int node_id = curr_list.back();
- int udop_schref = Schema->find_tbl(qnodes[node_id]->file);
- opview_entry *opv = new opview_entry();
-
-// Many of the UDOP properties aren't currently used.
- opv->parent_qname = "no_parent";
- opv->root_name = qnodes[node_id]->name;
- opv->view_name = qnodes[node_id]->file;
- opv->pos = qi;
- sprintf(tmpstr,"%s_UDOP%d_%s",qnodes[node_id]->name.c_str(),qi,opv->view_name.c_str());
- opv->udop_alias = tmpstr;
- opv->mangler = qnodes[node_id]->mangler;
-
- if(opv->mangler != ""){
- int new_udop_schref = Schema->add_duplicate_table(opv->view_name,opv->root_name);
- Schema->mangle_subq_names(new_udop_schref,opv->mangler);
- }
-
-// This piece of code makes each hfta which referes to the same udop
-// reference a distinct running udop. Do this at query optimization time?
-// fmtbl->set_udop_alias(opv->udop_alias);
-
- opv->exec_fl = Schema->get_op_prop(udop_schref, string("file"));
- opv->liveness_timeout = atoi(Schema->get_op_prop(udop_schref, string("liveness_timeout")).c_str());
-
- vector<subquery_spec *> subq = Schema->get_subqueryspecs(udop_schref);
- int s,f,q;
- for(s=0;s<subq.size();++s){
-// Validate that the fields match.
- subquery_spec *sqs = subq[s];
- string subq_name = sqs->name + opv->mangler;
- vector<field_entry *> flds = Schema->get_fields(subq_name);
- if(flds.size() == 0){
- fprintf(stderr,"INTERNAL ERROR: the subquery %s of view %s not found in Schema.\n",subq_name.c_str(), opv->view_name.c_str());
- return(1);
- }
- if(flds.size() < sqs->types.size()){
- fprintf(stderr,"ERROR: subquery %s of view %s does not have enough fields (%lu found, %lu expected).\n",subq_name.c_str(), opv->view_name.c_str(),flds.size(), sqs->types.size());
- return(1);
- }
- bool failed = false;
- for(f=0;f<sqs->types.size();++f){
- data_type dte(sqs->types[f],sqs->modifiers[f]);
- data_type dtf(flds[f]->get_type(),flds[f]->get_modifier_list());
- if(! dte.subsumes_type(&dtf) ){
- fprintf(stderr,"ERROR: subquery %s of view %s does not have the correct type for field %d (%s found, %s expected).\n",subq_name.c_str(), opv->view_name.c_str(),f,dtf.to_string().c_str(), dte.to_string().c_str());
- failed = true;
- }
-/*
- if(dte.is_temporal() && (dte.get_temporal() != dtf.get_temporal()) ){
- string pstr = dte.get_temporal_string();
- fprintf(stderr,"ERROR: subquery %s of view %s does not have the expected temporal value %s of field %d.\n",sqs->name.c_str(), opv->view_name.c_str(),pstr.c_str(),f);
- failed = true;
- }
-*/
- }
- if(failed)
- return(1);
-/// Validation done, find the subquery, make a copy of the
-/// parse tree, and add it to the return list.
- for(q=0;q<qnodes.size();++q)
- if(qnodes[q]->name == subq_name)
- break;
- if(q==qnodes.size()){
- fprintf(stderr,"INTERNAL ERROR: subquery %s of view %s not found in list of query names.\n",subq_name.c_str(), opv->view_name.c_str());
- return(1);
- }
-
- }
-
-// Cross-link to from entry(s) in all sourced-to tables.
- set<int>::iterator sii;
- for(sii=qnodes[curr_list.back()]->sources_to.begin() ;sii!=qnodes[curr_list.back()]->sources_to.end();++sii){
-//printf("\tUDOP %s sources_to %d (%s)\n",hfta_sets[hfta_id]->name.c_str(),(*sii),hfta_sets[(*sii)]->name.c_str());
- vector<tablevar_t *> tblvars = qnodes[(*sii)]->parse_tree->get_from()->get_table_list();
- int ii;
- for(ii=0;ii<tblvars.size();++ii){
- if(tblvars[ii]->schema_name == opv->root_name){
- tblvars[ii]->set_opview_idx(opviews.size());
- }
-
- }
- }
-
- opviews.append(opv);
- }else{
-
-// Analyze the parse trees in this query,
-// put them in rootq
-// vector<int> curr_list = process_sets[qi];
-
-
-////////////////////////////////////////
-
- rootq = NULL;
-//printf("Process set %d, has %d queries\n",qi,curr_list.size());
- for(qj=0;qj<curr_list.size();++qj){
- i = curr_list[qj];
- fprintf(stderr,"Processing query %s (file %s) is_udop = %d\n",qnodes[i]->name.c_str(), qnodes[i]->file.c_str(),qnodes[i]->is_udop);
-
-// Select the current query parse tree
- table_exp_t *fta_parse_tree = qnodes[i]->parse_tree;
-
-// if hfta only, try to fetch any missing schemas
-// from the registry (using the print_schema program).
-// Here I use a hack to avoid analyzing the query -- all referenced
-// tables must be in the from clause
-// If there is a problem loading any table, just issue a warning,
-//
- tablevar_list_t *fm = fta_parse_tree->get_from();
- vector<string> refd_tbls = fm->get_src_tbls(Schema);
-// iterate over all referenced tables
- int t;
- for(t=0;t<refd_tbls.size();++t){
- int tbl_ref = Schema->get_table_ref(refd_tbls[t]);
-
- if(tbl_ref < 0){ // if this table is not in the Schema
-
- if(hfta_only){
- string cmd="print_schema "+refd_tbls[t];
- FILE *schema_in = popen(cmd.c_str(), "r");
- if(schema_in == NULL){
- fprintf(stderr,"ERROR: cannot execute command %s\n",cmd.c_str());
- }else{
- string schema_instr;
- while(fgets(tmpstr,TMPSTRLEN,schema_in)){
- schema_instr += tmpstr;
- }
- fta_parse_result = new fta_parse_t();
- strcpy(tmp_schema_str,schema_instr.c_str());
- FtaParser_setstringinput(tmp_schema_str);
- if(FtaParserparse()){
- fprintf(stderr,"FTA parse failed on result from command %s.\n",cmd.c_str());
- }else{
- if( fta_parse_result->tables != NULL){
- int tl;
- for(tl=0;tl<fta_parse_result->tables->size();++tl){
- Schema->add_table(fta_parse_result->tables->get_table(tl));
- }
- }else{
- fprintf(stderr,"ERROR command %s returned no tables.\n",cmd.c_str());
- }
- }
- }
- }else{
- fprintf(stderr,"ERROR, query %s (file %s) references stream %s, which is neither a PROTOCOL nor an externally visible query stream.\n", qnodes[i]->name.c_str(), qnodes[i]->file.c_str(), refd_tbls[t].c_str());
- exit(1);
- }
-
- }
- }
-
-
-// Analyze the query.
- query_summary_class *qs = analyze_fta(fta_parse_tree, Schema, Ext_fcns, qnodes[i]->name);
- if(qs == NULL){
- fprintf(stderr,"analyze_fta failed on query %s (file %s) \n",qnodes[i]->name.c_str(),qnodes[i]->file.c_str());
- exit(1);
- }
-
- stream_query new_sq(qs, Schema);
- if(new_sq.error_code){
- fprintf(stderr,"ERROR, can't create query node for query %s (file %s):\n%s\n",qnodes[i]->name.c_str(),qnodes[i]->file.c_str(),new_sq.err_str.c_str());
- exit(1);
- }
-
-// Add it to the Schema
- table_def *output_td = new_sq.get_output_tabledef();
- Schema->add_table(output_td);
-
-// Create a query plan from the analyzed parse tree.
-// If its a query referneced via FROM, add it to the stream query.
- if(rootq){
- rootq->add_query(new_sq);
- }else{
- rootq = new stream_query(new_sq);
-// have the stream query object inherit properties form the analyzed
-// hfta_node object.
- rootq->set_nparallel(hfta_sets[hfta_id]->n_parallel,hfta_sets[hfta_id]->parallel_idx);
- rootq->n_successors = hfta_sets[hfta_id]->sources_to.size();
- }
-
-
- }
-
-// This stream query has all its parts
-// Build and optimize it.
-//printf("translate_fta: generating plan.\n");
- if(rootq->generate_plan(Schema)){
- fprintf(stderr,"INTERNAL ERROR: Can't generate query plan for query %s.\n",rootq->query_name.c_str());
- continue;
- }
-
-// If we've found the query plan head, so now add the output operators
- if(qname_to_ospec.count(hfta_sets[hfta_id]->source_name)){
- pair< multimap<string, int>::iterator, multimap<string, int>::iterator > oset;
- multimap<string, int>::iterator mmsi;
- oset = qname_to_ospec.equal_range(hfta_sets[hfta_id]->source_name);
- for(mmsi=oset.first; mmsi!=oset.second; ++mmsi){
- rootq->add_output_operator(output_specs[(*mmsi).second]);
- }
- }
-
-
-
-// Perform query splitting if necessary.
- bool hfta_returned;
- vector<stream_query *> split_queries = rootq->split_query(Ext_fcns, Schema, hfta_returned, ifaces_db, n_virtual_interfaces, hfta_sets[hfta_id]->n_parallel, hfta_sets[hfta_id]->parallel_idx);
-
- int l;
-//for(l=0;l<split_queries.size();++l){
-//printf("split query %d is %s\n",l,split_queries[l]->q`uery_name.c_str());
-//}
-
-
-
-
- if(split_queries.size() > 0){ // should be at least one component.
-
-// Compute the number of LFTAs.
- int n_lfta = split_queries.size();
- if(hfta_returned) n_lfta--;
-
-
-// Process the LFTA components.
- for(l=0;l<n_lfta;++l){
- if(lfta_names.count(split_queries[l]->query_name) == 0){
-// Grab the lfta for global optimization.
- vector<tablevar_t *> tvec = split_queries[l]->query_plan[0]->get_input_tbls();
- string liface = tvec[0]->get_interface();
- string lmach = tvec[0]->get_machine();
- if (lmach == "")
- lmach = hostname;
- interface_names.push_back(liface);
- machine_names.push_back(lmach);
-//printf("Machine is %s\n",lmach.c_str());
-
-// Set the ht size from the recommendation, if there is one in the rec file
- if(lfta_htsize.count(split_queries[l]->query_name)>0){
- split_queries[l]->query_plan[0]->set_definition("aggregate_slots",int_to_string(lfta_htsize[split_queries[l]->query_name]));
- }
-
-
- lfta_names[split_queries[l]->query_name] = lfta_list.size();
- split_queries[l]->set_gid(lfta_list.size()); // set lfta global id
- lfta_list.push_back(split_queries[l]);
- lfta_mach_lists[lmach].push_back(split_queries[l]);
-
-// THe following is a hack,
-// as I should be generating LFTA code through
-// the stream_query object.
- split_queries[l]->query_plan[0]->bind_to_schema(Schema);
-// split_queries[l]->query_plan[0]->definitions = split_queries[l]->defines;
-
-/*
-// Create query description to embed in lfta.c
- string lfta_schema_str = split_queries[l]->make_schema();
- string lfta_schema_embed = make_C_embedded_string(lfta_schema_str);
-
-// get NIC capabilities.
- int erri;
- nic_property *nicprop = NULL;
- vector<string> iface_codegen_type = ifaces_db->get_iface_vals(lmach,liface,"iface_codegen_type",erri,err_str);
- if(iface_codegen_type.size()){
- nicprop = npdb->get_nic_property(iface_codegen_type[0],erri);
- if(!nicprop){
- fprintf(stderr,"ERROR cannot load nic properties %s for interface %s of machine %s for lfta codegen\n",iface_codegen_type[0].c_str(), liface.c_str(), lmach.c_str());
- exit(1);
- }
- }
-
- lfta_val[lmach] += generate_lfta_block(split_queries[l]->query_plan[0], Schema, split_queries[l]->get_gid(), Ext_fcns, lfta_schema_embed, ifaces_db,nicprop);
-*/
-
- snap_lengths.push_back(compute_snap_len(split_queries[l]->query_plan[0], Schema));
- query_names.push_back(split_queries[l]->query_name);
- mach_query_names[lmach].push_back(query_names.size()-1);
-// NOTE: I will assume a 1-1 correspondance between
-// mach_query_names[lmach] and lfta_mach_lists[lmach]
-// where mach_query_names[lmach][i] contains the index into
-// query_names, which names the lfta, and
-// mach_query_names[lmach][i] is the stream_query * of the
-// corresponding lfta.
-// Later, lfta_iface_qnames are the query names matching lfta_iface_lists
-
-
-
- // check if lfta is reusable
- // FTA is reusable if reusable option is set explicitly or it doesn't have any parameters
-
- bool lfta_reusable = false;
- if (split_queries[l]->query_plan[0]->get_val_of_def("reusable") == "yes" ||
- split_queries[l]->query_plan[0]->get_param_tbl()->size() == 0) {
- lfta_reusable = true;
- }
- lfta_reuse_options.push_back(lfta_reusable);
-
- // LFTA will inherit the liveness timeout specification from the containing query
- // it is too conservative as lfta are expected to spend less time per tuple
- // then full query
-
- // extract liveness timeout from query definition
- int liveness_timeout = atoi(split_queries[l]->query_plan[0]->get_val_of_def("liveness_timeout").c_str());
- if (!liveness_timeout) {
-// fprintf(stderr, "WARNING: No liveness timeout specified for lfta %s, using default value of %d\n",
-// split_queries[l]->query_name.c_str(), DEFAULT_UDOP_LIVENESS_TIMEOUT);
- liveness_timeout = DEFAULT_UDOP_LIVENESS_TIMEOUT;
- }
- lfta_liveness_timeouts.push_back(liveness_timeout);
-
-// Add it to the schema
- table_def *td = split_queries[l]->get_output_tabledef();
- Schema->append_table(td);
-//printf("added lfta %d (%s)\n",l,split_queries[l]->query_name.c_str());
-
- }
- }
-
-// If the output is lfta-only, dump out the query name.
- if(split_queries.size() == 1 && !hfta_returned){
- if(output_query_names ){
- fprintf(query_name_output,"%s L\n",split_queries[0]->query_name.c_str());
- }
-/*
-else{
- fprintf(stderr,"query name is %s\n",split_queries[0]->query_name.c_str());
- }
-*/
-
-/*
-// output schema summary
- if(output_schema_summary){
- dump_summary(split_queries[0]);
- }
-*/
- }
-
-
- if(hfta_returned){ // query also has an HFTA component
- int hfta_nbr = split_queries.size()-1;
-
- hfta_list.push_back(split_queries[hfta_nbr]);
-
-// report on generated query names
- if(output_query_names){
- string hfta_name =split_queries[hfta_nbr]->query_name;
- fprintf(query_name_output,"%s H\n",hfta_name.c_str());
- for(l=0;l<hfta_nbr;++l){
- string lfta_name =split_queries[l]->query_name;
- fprintf(query_name_output,"%s L\n",lfta_name.c_str());
- }
- }
-// else{
-// fprintf(stderr,"query names are ");
-// for(l=0;l<hfta_nbr;++l){
-// if(l>0) fprintf(stderr,",");
-// string fta_name =split_queries[l]->query_name;
-// fprintf(stderr," %s",fta_name.c_str());
-// }
-// fprintf(stderr,"\n");
-// }
- }
-
- }else{
- fprintf(stderr,"ERROR, query in file %s has a bug.\n",qnodes[i]->file.c_str());
- fprintf(stderr,"%s\n",rootq->get_error_str().c_str());
- exit(1);
- }
- }
-}
-
-
-//-----------------------------------------------------------------
-// Compute and propagate the SE in PROTOCOL fields compute a field.
-//-----------------------------------------------------------------
-
-for(i=0;i<lfta_list.size();i++){
- lfta_list[i]->generate_protocol_se(sq_map, Schema);
- sq_map[lfta_list[i]->get_sq_name()] = lfta_list[i];
-}
-for(i=0;i<hfta_list.size();i++){
- hfta_list[i]->generate_protocol_se(sq_map, Schema);
- sq_map[hfta_list[i]->get_sq_name()] = hfta_list[i];
-}
-
-
-
-//------------------------------------------------------------------------
-// Perform individual FTA optimizations
-//-----------------------------------------------------------------------
-
-if (partitioned_mode) {
-
- // open partition definition file
- string part_fname = config_dir_path + "partition.txt";
-
- FILE* partfd = fopen(part_fname.c_str(), "r");
- if (!partfd) {
- fprintf(stderr,"ERROR, unable to open partition definition file %s\n",part_fname.c_str());
- exit(1);
- }
- PartnParser_setfileinput(partfd);
- if (PartnParserparse()) {
- fprintf(stderr,"ERROR, unable to parse partition definition file %s\n",part_fname.c_str());
- exit(1);
- }
- fclose(partfd);
-}
-
-
-print_hftas("preopt_hfta_info.txt", hfta_list, lfta_names, query_names, interface_names);
-
-int num_hfta = hfta_list.size();
-for(i=0; i < hfta_list.size(); ++i){
- hfta_list[i]->optimize(hfta_list, lfta_names, interface_names, machine_names, Ext_fcns, Schema, ifaces_db, partn_parse_result);
-}
-
-// Add all new hftas to schema
-for(i=num_hfta; i < hfta_list.size(); ++i){
- table_def *td = hfta_list[i]->get_output_tabledef();
- Schema->append_table(td);
-}
-
-print_hftas("postopt_hfta_info.txt", hfta_list, lfta_names, query_names, interface_names);
-
-
-
-//------------------------------------------------------------------------
-// Do global (cross-fta) optimization
-//-----------------------------------------------------------------------
-
-
-
-
-
-
-set<string> extra_external_libs;
-
-for(i=0;i<hfta_list.size();++i){ // query also has an HFTA component
-
- if(! debug_only){
-// build hfta file name, create output
- if(numeric_hfta_flname){
- sprintf(tmpstr,"hfta_%d",hfta_count);
- hfta_names.push_back(tmpstr);
- sprintf(tmpstr,"hfta_%d.cc",hfta_count);
- }else{
- sprintf(tmpstr,"hfta_%s",hfta_list[i]->query_name.c_str());
- hfta_names.push_back(tmpstr);
- sprintf(tmpstr,"hfta_%s.cc",hfta_list[i]->query_name.c_str());
- }
- FILE *hfta_fl = fopen(tmpstr,"w");
- if(hfta_fl == NULL){
- fprintf(stderr,"ERROR can't open fta output file %s\n",tmpstr);
- exit(1);
- }
- fprintf(hfta_fl,"%s\n\n",hfta_list[i]->generate_hfta(Schema, Ext_fcns, opviews, distributed_mode).c_str());
-
-// If there is a field verifier, warn about
-// lack of compatability
-// NOTE : this code assumes that visible non-lfta queries
-// are those at the root of a stream query.
- string hfta_comment;
- string hfta_title;
- string hfta_namespace;
- if(hfta_list[i]->defines.count("comment")>0)
- hfta_comment = hfta_list[i]->defines["comment"];
- if(hfta_list[i]->defines.count("Comment")>0)
- hfta_comment = hfta_list[i]->defines["Comment"];
- if(hfta_list[i]->defines.count("COMMENT")>0)
- hfta_comment = hfta_list[i]->defines["COMMENT"];
- if(hfta_list[i]->defines.count("title")>0)
- hfta_title = hfta_list[i]->defines["title"];
- if(hfta_list[i]->defines.count("Title")>0)
- hfta_title = hfta_list[i]->defines["Title"];
- if(hfta_list[i]->defines.count("TITLE")>0)
- hfta_title = hfta_list[i]->defines["TITLE"];
- if(hfta_list[i]->defines.count("namespace")>0)
- hfta_namespace = hfta_list[i]->defines["namespace"];
- if(hfta_list[i]->defines.count("Namespace")>0)
- hfta_namespace = hfta_list[i]->defines["Namespace"];
- if(hfta_list[i]->defines.count("Namespace")>0)
- hfta_namespace = hfta_list[i]->defines["Namespace"];
-
- if(field_verifier != NULL){
- string warning_str;
- if(hfta_comment == "")
- warning_str += "\tcomment not found.\n";
- if(hfta_title == "")
- warning_str += "\ttitle not found.\n";
- if(hfta_namespace == "")
- warning_str += "\tnamespace not found.\n";
-
- vector<field_entry *> flds = hfta_list[i]->get_output_tabledef()->get_fields();
- int fi;
- for(fi=0;fi<flds.size();fi++){
- field_verifier->verify_field(flds[fi]->get_name(),flds[fi]->get_type(),warning_str);
- }
- if(warning_str != "")
- fprintf(stderr,"Warning, in HFTA stream %s:\n%s",
- hfta_list[i]->get_output_tabledef()->get_tbl_name().c_str(),warning_str.c_str());
- }
-
- fprintf(qtree_output,"\t<HFTA name='%s'>\n",hfta_list[i]->get_output_tabledef()->get_tbl_name().c_str());
- if(hfta_comment != "")
- fprintf(qtree_output,"\t\t<Description value='%s' />\n",hfta_comment.c_str());
- if(hfta_title != "")
- fprintf(qtree_output,"\t\t<Title value='%s' />\n",hfta_title.c_str());
- if(hfta_namespace != "")
- fprintf(qtree_output,"\t\t<Namespace value='%s' />\n",hfta_namespace.c_str());
- fprintf(qtree_output,"\t\t<FileName value='%s' />\n",tmpstr);
- fprintf(qtree_output,"\t\t<Rate value='100' />\n");
-
-// write info about fields to qtree.xml
- vector<field_entry *> flds = hfta_list[i]->get_output_tabledef()->get_fields();
- int fi;
- for(fi=0;fi<flds.size();fi++){
- fprintf(qtree_output,"\t\t<Field name='%s' pos='%d' type='%s' ",flds[fi]->get_name().c_str(), fi, flds[fi]->get_type().c_str());
- if(flds[fi]->get_modifier_list()->size()){
- fprintf(qtree_output,"mods='%s' ", flds[fi]->get_modifier_list()->to_string().c_str());
- }
- fprintf(qtree_output," />\n");
- }
-
- // extract liveness timeout from query definition
- int liveness_timeout = atoi(hfta_list[i]->query_plan[hfta_list[i]->qhead]->get_val_of_def("liveness_timeout").c_str());
- if (!liveness_timeout) {
-// fprintf(stderr, "WARNING: No liveness timeout specified for hfta %s, using default value of %d\n",
-// hfta_list[i]->get_output_tabledef()->get_tbl_name().c_str(), DEFAULT_UDOP_LIVENESS_TIMEOUT);
- liveness_timeout = DEFAULT_UDOP_LIVENESS_TIMEOUT;
- }
- fprintf(qtree_output,"\t\t<LivenessTimeout value='%d' />\n", liveness_timeout);
-
- vector<tablevar_t *> tmp_tv = hfta_list[i]->get_input_tables();
- int itv;
- for(itv=0;itv<tmp_tv.size();++itv){
- fprintf(qtree_output,"\t\t<ReadsFrom value='%s' />\n",tmp_tv[itv]->get_schema_name().c_str());
- }
- string ifrs = hfta_list[i]->collect_refd_ifaces();
- if(ifrs != ""){
- fprintf(qtree_output,"\t\t<Ifaces_referenced value='%s'/>\n",ifrs.c_str());
- }
- fprintf(qtree_output,"\t</HFTA>\n");
-
- fclose(hfta_fl);
- }else{
-// debug only -- do code generation to catch generation-time errors.
- hfta_list[i]->generate_hfta(Schema, Ext_fcns,opviews, distributed_mode);
- }
-
- hfta_count++; // for hfta file names with numeric suffixes
-
- hfta_list[i]->get_external_libs(extra_external_libs);
-
- }
-
-string ext_lib_string;
-set<string>::iterator ssi_el;
-for(ssi_el=extra_external_libs.begin();ssi_el!=extra_external_libs.end();++ssi_el)
- ext_lib_string += (*ssi_el)+" ";
-
-
-
-// Report on the set of operator views
- for(i=0;i<opviews.size();++i){
- opview_entry *opve = opviews.get_entry(i);
- fprintf(qtree_output,"\t<UDOP name='%s' >\n",opve->view_name.c_str());
- fprintf(qtree_output,"\t\t<FileName value='%s' />\n",opve->exec_fl.c_str());
- fprintf(qtree_output,"\t\t<Parent value='%s' />\n",opve->root_name.c_str());
- fprintf(qtree_output,"\t\t<Alias value='%s' />\n",opve->udop_alias.c_str());
- fprintf(qtree_output,"\t\t<Rate value='100' />\n");
-
- if (!opve->liveness_timeout) {
-// fprintf(stderr, "WARNING: No liveness timeout specified for view %s, using default value of %d\n",
-// opve->view_name.c_str(), DEFAULT_UDOP_LIVENESS_TIMEOUT);
- opve->liveness_timeout = DEFAULT_UDOP_LIVENESS_TIMEOUT;
- }
- fprintf(qtree_output,"\t\t<LivenessTimeout value='%d' />\n", opve->liveness_timeout);
- int j;
- for(j=0;j<opve->subq_names.size();j++)
- fprintf(qtree_output,"\t\t<ReadsFrom value='%s' />\n",opve->subq_names[j].c_str());
- fprintf(qtree_output,"\t</UDOP>\n");
- }
-
-
-//-----------------------------------------------------------------
-
-// Create interface-specific meta code files.
-// first, open and parse the interface resources file.
- ifaces_db = new ifq_t();
- ierr = "";
- if(ifaces_db->load_ifaces(ifx_fname,use_live_hosts_file,distributed_mode,ierr)){
- fprintf(stderr,"ERROR, can't load interface resource file %s :\n%s",
- ifx_fname.c_str(), ierr.c_str());
- exit(1);
- }
-
- map<string, vector<stream_query *> >::iterator svsi;
- for(svsi=lfta_mach_lists.begin(); svsi!=lfta_mach_lists.end(); ++svsi){
- string lmach = (*svsi).first;
-
- // For this machine, create a set of lftas per interface.
- vector<stream_query *> mach_lftas = (*svsi).second;
- map<string, vector<stream_query *> > lfta_iface_lists;
- int li;
- for(li=0;li<mach_lftas.size();++li){
- vector<tablevar_t *> tvec = mach_lftas[li]->query_plan[0]->get_input_tbls();
- string lfta_iface = tvec[0]->get_interface();
- lfta_iface_lists[lfta_iface].push_back(mach_lftas[li]);
- }
-
- map<string, vector<stream_query *> >::iterator lsvsi;
- for(lsvsi=lfta_iface_lists.begin(); lsvsi!=lfta_iface_lists.end(); ++lsvsi){
- int erri;
- string liface = (*lsvsi).first;
- vector<stream_query *> iface_lftas = (*lsvsi).second;
- vector<string> iface_codegen_type = ifaces_db->get_iface_vals(lmach,liface,"iface_codegen_type",erri,err_str);
- if(iface_codegen_type.size()){
- nic_property *nicprop = npdb->get_nic_property(iface_codegen_type[0],erri);
- if(!nicprop){
- fprintf(stderr,"ERROR cannot load nic properties %s for interface %s of machine %s\n",iface_codegen_type[0].c_str(), liface.c_str(), lmach.c_str());
- exit(1);
- }
- string mcs = generate_nic_code(iface_lftas, nicprop);
- string mcf_flnm;
- if(lmach != "")
- mcf_flnm = lmach + "_"+liface+".mcf";
- else
- mcf_flnm = hostname + "_"+liface+".mcf";
- FILE *mcf_fl ;
- if((mcf_fl = fopen(mcf_flnm.c_str(),"w")) == NULL){
- fprintf(stderr,"Can't open output file %s\n%s\n",mcf_flnm.c_str(),strerror(errno));
- exit(1);
- }
- fprintf(mcf_fl,"%s",mcs.c_str());
- fclose(mcf_fl);
-//printf("mcs of machine %s, iface %s of type %s is \n%s\n",
-//lmach.c_str(), liface.c_str(),iface_codegen_type[0].c_str(), mcs.c_str());
- }
- }
-
-
- }
-
-
-
-//-----------------------------------------------------------------
-
-
-// Find common filter predicates in the LFTAs.
-// in addition generate structs to store the temporal attributes unpacked by prefilter
-
- map<string, vector<stream_query *> >::iterator ssqi;
- for(ssqi=lfta_mach_lists.begin(); ssqi!=lfta_mach_lists.end(); ++ssqi){
-
- string lmach = (*ssqi).first;
- bool packed_return = false;
- int li, erri;
-
-
-// The LFTAs of this machine.
- vector<stream_query *> mach_lftas = (*ssqi).second;
-// break up on a per-interface basis.
- map<string, vector<stream_query *> > lfta_iface_lists;
- map<string, vector<int> > lfta_iface_qname_ix; // need the query name
- // for fta_init
- for(li=0;li<mach_lftas.size();++li){
- vector<tablevar_t *> tvec = mach_lftas[li]->query_plan[0]->get_input_tbls();
- string lfta_iface = tvec[0]->get_interface();
- lfta_iface_lists[lfta_iface].push_back(mach_lftas[li]);
- lfta_iface_qname_ix[lfta_iface].push_back(mach_query_names[lmach][li]);
- }
-
-
-// Are the return values "packed"?
-// This should be done on a per-interface basis.
-// But this is defunct code for gs-lite
- for(li=0;li<mach_lftas.size();++li){
- vector<tablevar_t *> tvec = mach_lftas[li]->query_plan[0]->get_input_tbls();
- string liface = tvec[0]->get_interface();
- vector<string> iface_codegen_type = ifaces_db->get_iface_vals(lmach,liface,"iface_codegen_type",erri,err_str);
- if(iface_codegen_type.size()){
- if(npdb->get_nic_prop_val(iface_codegen_type[0],"Return",erri) == "Packed"){
- packed_return = true;
- }
- }
- }
-
-
-// Separate lftas by interface, collect results on a per-interface basis.
-
- vector<cnf_set *> no_preds; // fallback if there is no prefilter
- map<string, vector<cnf_set *> > prefilter_preds;
- set<unsigned int> pred_ids; // this can be global for all interfaces
- for(auto mvsi=lfta_iface_lists.begin(); mvsi!=lfta_iface_lists.end(); ++mvsi){
- string liface = (*mvsi).first;
- vector<cnf_set *> empty_list;
- prefilter_preds[liface] = empty_list;
- if(! packed_return){
- get_common_lfta_filter(lfta_iface_lists[liface], Schema,Ext_fcns, prefilter_preds[liface], pred_ids);
- }
-
-// get NIC capabilities. (Is this needed?)
- nic_property *nicprop = NULL;
- vector<string> iface_codegen_type = ifaces_db->get_iface_vals(lmach,liface,"iface_codegen_type",erri,err_str);
- if(iface_codegen_type.size()){
- nicprop = npdb->get_nic_property(iface_codegen_type[0],erri);
- if(!nicprop){
- fprintf(stderr,"ERROR cannot load nic properties %s for interface %s of machine %s for lfta codegen\n",iface_codegen_type[0].c_str(), liface.c_str(), lmach.c_str());
- exit(1);
- }
- }
- }
-
-
-// Now that we know the prefilter preds, generate the lfta code.
-// Do this for all lftas in this machine.
- for(li=0;li<mach_lftas.size();++li){
- set<unsigned int> subsumed_preds;
- set<unsigned int>::iterator sii;
-#ifdef PREFILTER_OK
- for(sii=pred_ids.begin();sii!=pred_ids.end();++sii){
- int pid = (*sii);
- if((pid>>16) == li){
- subsumed_preds.insert(pid & 0xffff);
- }
- }
-#endif
- string lfta_schema_str = mach_lftas[li]->make_schema();
- string lfta_schema_embed = make_C_embedded_string(lfta_schema_str);
- nic_property *nicprop = NULL; // no NIC properties?
- lfta_val[lmach] += generate_lfta_block(mach_lftas[li]->query_plan[0], Schema, mach_lftas[li]->get_gid(), Ext_fcns, lfta_schema_embed, ifaces_db,nicprop,subsumed_preds);
- }
-
-
-// generate structs to store the temporal attributes
-// unpacked by prefilter
- col_id_set temp_cids;
- get_prefilter_temporal_cids(lfta_mach_lists[lmach], temp_cids);
- lfta_prefilter_val[lmach] = generate_lfta_prefilter_struct(temp_cids, Schema);
-
-// Compute the lfta bit signatures and the lfta colrefs
-// do this on a per-interface basis
-#ifdef PREFILTER_OK
- lfta_prefilter_val[lmach] += "#define PREFILTER_DEFINED 1;\n\n";
-#endif
- map<string, vector<long long int> > lfta_sigs; // used again later
- for(auto mvsi=lfta_iface_lists.begin(); mvsi!=lfta_iface_lists.end(); ++mvsi){
- string liface = (*mvsi).first;
- vector<long long int> empty_list;
- lfta_sigs[liface] = empty_list;
-
- vector<col_id_set> lfta_cols;
- vector<int> lfta_snap_length;
- for(li=0;li<lfta_iface_lists[liface].size();++li){
- unsigned long long int mask=0, bpos=1;
- int f_pos;
- for(f_pos=0;f_pos<prefilter_preds[liface].size();++f_pos){
- if(prefilter_preds[liface][f_pos]->lfta_id.count(li))
- mask |= bpos;
- bpos = bpos << 1;
- }
- lfta_sigs[liface].push_back(mask);
- lfta_cols.push_back(lfta_iface_lists[liface][li]->query_plan[0]->get_colrefs(true,Schema));
- lfta_snap_length.push_back(compute_snap_len(lfta_iface_lists[liface][li]->query_plan[0], Schema));
- }
-
-//for(li=0;li<mach_lftas.size();++li){
-//printf("lfta %d, msak is %llu\n",li,lfta_sigs[li]);
-//col_id_set::iterator tcisi;
-//for(tcisi=lfta_cols[li].begin(); tcisi!=lfta_cols[li].end();++tcisi){
-//printf("\t%s %d (%d)\n",(*tcisi).field.c_str(),(*tcisi).schema_ref, (*tcisi).tblvar_ref);
-//}
-//}
-
-
-// generate the prefilter
-// Do this on a per-interface basis, except for the #define
-#ifdef PREFILTER_OK
-// lfta_prefilter_val[lmach] += "#define PREFILTER_DEFINED 1;\n\n";
- lfta_val[lmach] += generate_lfta_prefilter(prefilter_preds[liface], temp_cids, Schema, Ext_fcns, lfta_cols, lfta_sigs[liface], lfta_snap_length, liface);
-#else
- lfta_val[lmach] += generate_lfta_prefilter(no_preds, temp_cids, Schema,Ext_fcns, lfta_cols, lfta_sigs[liface], lfta_snap_length, liface);
-
-#endif
- }
-
-// Generate interface parameter lookup function
- lfta_val[lmach] += "// lookup interface properties by name\n";
- lfta_val[lmach] += "// mulitple values of given property returned in the same string separated by commas\n";
- lfta_val[lmach] += "// returns NULL if given property does not exist\n";
- lfta_val[lmach] += "gs_sp_t get_iface_properties (const gs_sp_t iface_name, const gs_sp_t property_name) {\n";
-
-// collect a lit of interface names used by queries running on this host
- set<std::string> iface_names;
- for(i=0;i<mach_query_names[lmach].size();i++){
- int mi = mach_query_names[lmach][i];
- stream_query *lfta_sq = lfta_mach_lists[lmach][i];
-
- if(interface_names[mi]=="")
- iface_names.insert("DEFAULTDEV");
- else
- iface_names.insert(interface_names[mi]);
- }
-
-// generate interface property lookup code for every interface
- set<std::string>::iterator sir;
- for (sir = iface_names.begin(); sir != iface_names.end(); sir++) {
- if (sir == iface_names.begin())
- lfta_val[lmach] += "\tif (!strcmp(iface_name, \"" + *sir + "\")) {\n";
- else
- lfta_val[lmach] += "\t} else if (!strcmp(iface_name, \"" + *sir + "\")) {\n";
-
- // iterate through interface properties
- vector<string> iface_properties = ifaces_db->get_iface_properties(lmach,*sir,erri,err_str);
- if (erri) {
- fprintf(stderr,"ERROR cannot retrieve interface properties for %s.%s, %s\n",lmach.c_str(), sir->c_str(), err_str.c_str());
- exit(1);
- }
- if (iface_properties.empty())
- lfta_val[lmach] += "\t\treturn NULL;\n";
- else {
- for (int i = 0; i < iface_properties.size(); ++i) {
- if (i == 0)
- lfta_val[lmach] += "\t\tif (!strcmp(property_name, \"" + iface_properties[i] + "\")) {\n";
- else
- lfta_val[lmach] += "\t\t} else if (!strcmp(property_name, \"" + iface_properties[i] + "\")) {\n";
-
- // combine all values for the interface property using comma separator
- vector<string> vals = ifaces_db->get_iface_vals(lmach,*sir,iface_properties[i],erri,err_str);
- for (int j = 0; j < vals.size(); ++j) {
- lfta_val[lmach] += "\t\t\treturn \"" + vals[j];
- if (j != vals.size()-1)
- lfta_val[lmach] += ",";
- }
- lfta_val[lmach] += "\";\n";
- }
- lfta_val[lmach] += "\t\t} else\n";
- lfta_val[lmach] += "\t\t\treturn NULL;\n";
- }
- }
- lfta_val[lmach] += "\t} else\n";
- lfta_val[lmach] += "\t\treturn NULL;\n";
- lfta_val[lmach] += "}\n\n";
-
-
-// Generate a full list of FTAs for clearinghouse reference
- lfta_val[lmach] += "// list of FTAs clearinghouse expects to register themselves\n";
- lfta_val[lmach] += "const gs_sp_t fta_names[] = {";
-
- for (i = 0; i < query_names.size(); ++i) {
- if (i)
- lfta_val[lmach] += ", ";
- lfta_val[lmach] += "\"" + query_names[i] + "\"";
- }
- for (i = 0; i < hfta_list.size(); ++i) {
- lfta_val[lmach] += ", \"" + hfta_list[i]->query_name + "\"";
- }
- lfta_val[lmach] += ", NULL};\n\n";
-
-
-// Add the initialization function to lfta.c
-// Change to accept the interface name, and
-// set the prefilter function accordingly.
-// see the example in demo/err2
- lfta_val[lmach] += "void fta_init(gs_sp_t device){\n";
-
-// for(i=0;i<mach_query_names[lmach].size();i++)
-// int mi = mach_query_names[lmach][i];
-// stream_query *lfta_sq = lfta_mach_lists[lmach][i];
-
- for(auto mvsi=lfta_iface_lists.begin(); mvsi!=lfta_iface_lists.end(); ++mvsi){
- string liface = (*mvsi).first;
- vector<stream_query *> lfta_list = (*mvsi).second;
- for(i=0;i<lfta_list.size();i++){
- stream_query *lfta_sq = lfta_list[i];
- int mi = lfta_iface_qname_ix[liface][i];
-
- fprintf(stderr,"interface %s, LFTA %d, snap length is %d\n",liface.c_str(),mi,snap_lengths[mi]);
-
- string this_iface = "DEFAULTDEV";
- if(interface_names[mi]!="")
- this_iface = '"'+interface_names[mi]+'"';
- lfta_val[lmach] += "\tif(!strcmp(device,"+this_iface+"))\n";
- lfta_val[lmach] += "\t\tfta_register(\""+query_names[mi]+"\", " + (lfta_reuse_options[mi]?"1":"0") + ", ";
-// if(interface_names[mi]=="")
-// lfta_val[lmach]+="DEFAULTDEV";
-// else
-// lfta_val[lmach]+='"'+interface_names[mi]+'"';
- lfta_val[lmach] += this_iface;
-
-
- lfta_val[lmach] += ", "+generate_alloc_name(query_names[mi])
- +"\n#ifndef LFTA_IN_NIC\n\t,"+generate_schema_string_name(query_names[mi])
- +"\n#endif\n";
- sprintf(tmpstr,",%d",snap_lengths[mi]);
- lfta_val[lmach] += tmpstr;
-
-// unsigned long long int mask=0, bpos=1;
-// int f_pos;
-// for(f_pos=0;f_pos<prefilter_preds.size();++f_pos){
-// if(prefilter_preds[f_pos]->lfta_id.count(i))
-// mask |= bpos;
-// bpos = bpos << 1;
-// }
-
-#ifdef PREFILTER_OK
-// sprintf(tmpstr,",%lluull",mask);
- sprintf(tmpstr,",%lluull",lfta_sigs[liface][i]);
- lfta_val[lmach]+=tmpstr;
-#else
- lfta_val[lmach] += ",0ull";
-#endif
-
- lfta_val[lmach] += ");\n";
-
-
-
-// End of lfta prefilter stuff
-// --------------------------------------------------
-
-// If there is a field verifier, warn about
-// lack of compatability
- string lfta_comment;
- string lfta_title;
- string lfta_namespace;
- map<string,string> ldefs = lfta_sq->query_plan[0]->get_definitions();
- if(ldefs.count("comment")>0)
- lfta_comment = lfta_sq->defines["comment"];
- if(ldefs.count("Comment")>0)
- lfta_comment = lfta_sq->defines["Comment"];
- if(ldefs.count("COMMENT")>0)
- lfta_comment = lfta_sq->defines["COMMENT"];
- if(ldefs.count("title")>0)
- lfta_title = lfta_sq->defines["title"];
- if(ldefs.count("Title")>0)
- lfta_title = lfta_sq->defines["Title"];
- if(ldefs.count("TITLE")>0)
- lfta_title = lfta_sq->defines["TITLE"];
- if(ldefs.count("NAMESPACE")>0)
- lfta_namespace = lfta_sq->defines["NAMESPACE"];
- if(ldefs.count("Namespace")>0)
- lfta_namespace = lfta_sq->defines["Namespace"];
- if(ldefs.count("namespace")>0)
- lfta_namespace = lfta_sq->defines["namespace"];
-
- string lfta_ht_size;
- if(lfta_sq->query_plan[0]->node_type()== "sgah_qpn" || lfta_sq->query_plan[0]->node_type()== "rsgah_qpn")
- lfta_ht_size = int_to_string(DEFAULT_LFTA_HASH_TABLE_SIZE);
- if(ldefs.count("aggregate_slots")>0){
- lfta_ht_size = ldefs["aggregate_slots"];
- }
-
-// NOTE : I'm assuming that visible lftas do not start with _fta.
-// -- will fail for non-visible simple selection queries.
- if(field_verifier != NULL && query_names[mi].substr(0,1) != "_"){
- string warning_str;
- if(lfta_comment == "")
- warning_str += "\tcomment not found.\n";
- if(lfta_title == "")
- warning_str += "\ttitle not found.\n";
- if(lfta_namespace == "")
- warning_str += "\tnamespace not found.\n";
-
- vector<field_entry *> flds = lfta_sq->get_output_tabledef()->get_fields();
- int fi;
- for(fi=0;fi<flds.size();fi++){
- field_verifier->verify_field(flds[fi]->get_name(),flds[fi]->get_type(),warning_str);
- }
- if(warning_str != "")
- fprintf(stderr,"Warning, in LFTA stream %s:\n%s",
- query_names[mi].c_str(),warning_str.c_str());
- }
-
-
-// Create qtree output
- fprintf(qtree_output,"\t<LFTA name='%s' >\n",query_names[mi].c_str());
- if(lfta_comment != "")
- fprintf(qtree_output,"\t\t<Description value='%s' />\n",lfta_comment.c_str());
- if(lfta_title != "")
- fprintf(qtree_output,"\t\t<Title value='%s' />\n",lfta_title.c_str());
- if(lfta_namespace != "")
- fprintf(qtree_output,"\t\t<Namespace value='%s' />\n",lfta_namespace.c_str());
- if(lfta_ht_size != "")
- fprintf(qtree_output,"\t\t<HtSize value='%s' />\n",lfta_ht_size.c_str());
- if(lmach != "")
- fprintf(qtree_output,"\t\t<Host value='%s' />\n",lmach.c_str());
- else
- fprintf(qtree_output,"\t\t<Host value='%s' />\n",hostname.c_str());
- fprintf(qtree_output,"\t\t<Interface value='%s' />\n",interface_names[mi].c_str());
- fprintf(qtree_output,"\t\t<ReadsFrom value='%s' />\n",interface_names[mi].c_str());
- fprintf(qtree_output,"\t\t<Rate value='100' />\n");
- fprintf(qtree_output,"\t\t<LivenessTimeout value='%d' />\n", lfta_liveness_timeouts[mi]);
-// write info about fields to qtree.xml
- vector<field_entry *> flds = lfta_sq->get_output_tabledef()->get_fields();
- int fi;
- for(fi=0;fi<flds.size();fi++){
- fprintf(qtree_output,"\t\t<Field name='%s' pos='%d' type='%s' ",flds[fi]->get_name().c_str(), fi, flds[fi]->get_type().c_str());
- if(flds[fi]->get_modifier_list()->size()){
- fprintf(qtree_output,"mods='%s' ", flds[fi]->get_modifier_list()->to_string().c_str());
- }
- fprintf(qtree_output," />\n");
- }
- fprintf(qtree_output,"\t</LFTA>\n");
-
-
- }
- }
-
- for(auto mvsi=lfta_iface_lists.begin(); mvsi!=lfta_iface_lists.end(); ++mvsi){
- string liface = (*mvsi).first;
- lfta_val[lmach] +=
-" if (!strcmp(device, \""+liface+"\")) \n"
-" lfta_prefilter = &lfta_prefilter_"+liface+"; \n"
-;
- }
- lfta_val[lmach] +=
-" if(lfta_prefilter == NULL){\n"
-" fprintf(stderr, \"Error, interface %s not found in fta_init.\\n\",device);\n"
-" exit(1);\n"
-" }\n"
-;
-
-
-
- lfta_val[lmach] += "}\n\n";
-
- if(!(debug_only || hfta_only) ){
- string lfta_flnm;
- if(lmach != "")
- lfta_flnm = lmach + "_lfta.c";
- else
- lfta_flnm = hostname + "_lfta.c";
- if((lfta_out = fopen(lfta_flnm.c_str(),"w")) == NULL){
- fprintf(stderr,"Can't open output file %s\n%s\n","lfta.c",strerror(errno));
- exit(1);
- }
- fprintf(lfta_out,"%s",lfta_header.c_str());
- fprintf(lfta_out,"%s",lfta_prefilter_val[lmach].c_str());
- fprintf(lfta_out,"%s",lfta_val[lmach].c_str());
- fclose(lfta_out);
- }
- }
-
-// Say what are the operators which must execute
- if(opviews.size()>0)
- fprintf(stderr,"The queries use the following external operators:\n");
- for(i=0;i<opviews.size();++i){
- opview_entry *opv = opviews.get_entry(i);
- fprintf(stderr,"\t%s\n",opv->view_name.c_str());
- }
-
- if(create_makefile)
- generate_makefile(input_file_names, nfiles, hfta_names, opviews,
- machine_names, schema_file_name,
- interface_names,
- ifaces_db, config_dir_path,use_pads,ext_lib_string, rts_hload);
-
-
- fprintf(qtree_output,"</QueryNodes>\n");
-
- return(0);
-}
-
-////////////////////////////////////////////////////////////
-
-void generate_makefile(vector<string> &input_file_names, int nfiles,
- vector<string> &hfta_names, opview_set &opviews,
- vector<string> &machine_names,
- string schema_file_name,
- vector<string> &interface_names,
- ifq_t *ifdb, string &config_dir_path,
- bool use_pads,
- string extra_libs,
- map<string, vector<int> > &rts_hload
- ){
- int i,j;
-
- if(config_dir_path != ""){
- config_dir_path = "-C "+config_dir_path;
- }
-
- struct stat sb;
- bool libz_exists = stat((root_path+"/lib/libz.a").c_str(),&sb) >= 0;
- bool libast_exists = stat((root_path+"/lib/libast.a").c_str(),&sb) >= 0;
-
-// if(libz_exists && !libast_exists){
-// fprintf(stderr,"Configuration ERROR: ../../lib/libz.a exists but ../../lib/libast.a dows not.\n");
-// exit(1);
-// }
-
-// Get set of operator executable files to run
- set<string> op_fls;
- set<string>::iterator ssi;
- for(i=0;i<opviews.size();++i){
- opview_entry *opv = opviews.get_entry(i);
- if(opv->exec_fl != "") op_fls.insert(opv->exec_fl);
- }
-
- FILE *outfl = fopen("Makefile", "w");
- if(outfl==NULL){
- fprintf(stderr,"Can't open Makefile for write, exiting.\n");
- exit(0);
- }
-
- fputs(
-("CPP= g++ -O3 -g -I "+root_path+"/include -I "+root_path+"/include/hfta\n"
-"CC= gcc -O3 -g -I . -I "+root_path+"/include -I "+root_path+"/include/lfta"
-).c_str(), outfl
-);
- if(generate_stats)
- fprintf(outfl," -DLFTA_STATS");
-
-// Gather the set of interfaces
-// Also, gather "base interface names" for use in computing
-// the hash splitting to virtual interfaces.
-// TODO : must update to hanndle machines
- set<string> ifaces;
- set<string> base_vifaces; // base interfaces of virtual interfaces
- map<string, string> ifmachines;
- map<string, string> ifattrs;
- for(i=0;i<interface_names.size();++i){
- ifaces.insert(interface_names[i]);
- ifmachines[interface_names[i]] = machine_names[i];
-
- size_t Xpos = interface_names[i].find_last_of("X");
- if(Xpos!=string::npos){
- string iface = interface_names[i].substr(0,Xpos);
- base_vifaces.insert(iface);
- }
- // get interface attributes and add them to the list
- }
-
-// Do we need to include protobuf libraries?
- bool use_proto = false;
- int erri;
- string err_str;
- for(ssi=ifaces.begin();ssi!=ifaces.end();++ssi){
- string ifnm = (*ssi);
- vector<string> ift = ifdb->get_iface_vals(ifmachines[ifnm],ifnm, "InterfaceType", erri, err_str);
- for(int ift_i=0;ift_i<ift.size();ift_i++){
- if(ift[ift_i]=="PROTO"){
- use_proto = true;
- }
- }
- }
-
- fprintf(outfl,
-"\n"
-"\n"
-"all: rts");
- for(i=0;i<hfta_names.size();++i)
- fprintf(outfl," %s",hfta_names[i].c_str());
- fputs(
-("\n"
-"\n"
-"rts: lfta.o "+root_path+"/lib/libgscphost.a "+root_path+"/lib/libgscplftaaux.a "+root_path+"/lib/libgscprts.a "+root_path+"/lib/libclearinghouse.a\n"
-"\tg++ -O3 -g -o rts lfta.o ").c_str(), outfl);
- if(use_pads)
- fprintf(outfl,"-L. ");
- fputs(
-("-L"+root_path+"/lib -lgscplftaaux ").c_str(), outfl);
- if(use_pads)
- fprintf(outfl,"-lgscppads -lpads ");
- fprintf(outfl,
-"-lgscprts -lgscphost -lm -lgscpaux -lgscplftaaux -lclearinghouse -lresolv -lpthread -lgscpinterface -lz");
- if(use_pads)
- fprintf(outfl, " -lpz -lz -lbz ");
- if(libz_exists && libast_exists)
- fprintf(outfl," -last ");
- if(use_pads)
- fprintf(outfl, " -ldll -ldl ");
- if(use_proto)
- fprintf(outfl, " -L/usr/local/lib/ -lprotobuf-c ");
- fprintf(outfl," -lgscpaux");
-#ifdef GCOV
- fprintf(outfl," -fprofile-arcs");
-#endif
- fprintf(outfl,
-"\n"
-"\n"
-"lfta.o: %s_lfta.c\n"
-"\t$(CC) -o lfta.o -c %s_lfta.c\n"
-"\n"
-"%s_lfta.c: external_fcns.def %s ",hostname.c_str(), hostname.c_str(), hostname.c_str(),schema_file_name.c_str());
- for(i=0;i<nfiles;++i)
- fprintf(outfl," %s",input_file_names[i].c_str());
- if(hostname == ""){
- fprintf(outfl,"\n\t%s/bin/translate_fta %s %s ",root_path.c_str(), config_dir_path.c_str(),schema_file_name.c_str());
- }else{
- fprintf(outfl,"\n\t%s/bin/translate_fta -h %s %s %s ", root_path.c_str(), hostname.c_str(), config_dir_path.c_str(),schema_file_name.c_str());
- }
- for(i=0;i<nfiles;++i)
- fprintf(outfl," %s",input_file_names[i].c_str());
- fprintf(outfl,"\n");
-
- for(i=0;i<hfta_names.size();++i)
- fprintf(outfl,
-("%s: %s.o\n"
-"\t$(CPP) -o %s %s.o -L"+root_path+"/lib -lgscpapp -lgscphostaux -lgscphost -lgscpinterface -lgscphftaaux -lgscphostaux -lm -lgscpaux -lclearinghouse -lresolv -lpthread -lgscpaux -lgscphftaaux -lgscpaux %s\n"
-"\n"
-"%s.o: %s.cc\n"
-"\t$(CPP) -o %s.o -c %s.cc\n"
-"\n"
-"\n").c_str(),
- hfta_names[i].c_str(), hfta_names[i].c_str(),
- hfta_names[i].c_str(), hfta_names[i].c_str(), extra_libs.c_str(),
- hfta_names[i].c_str(), hfta_names[i].c_str(),
- hfta_names[i].c_str(), hfta_names[i].c_str()
- );
-
- fprintf(outfl,
-("\n"
-"packet_schema.txt:\n"
-"\tln -s "+root_path+"/cfg/packet_schema.txt .\n"
-"\n"
-"external_fcns.def:\n"
-"\tln -s "+root_path+"/cfg/external_fcns.def .\n"
-"\n"
-"clean:\n"
-"\trm -rf core rts *.o %s_lfta.c external_fcns.def packet_schema.txt").c_str(),hostname.c_str());
- for(i=0;i<hfta_names.size();++i)
- fprintf(outfl," %s %s.cc",hfta_names[i].c_str(),hfta_names[i].c_str());
- fprintf(outfl,"\n");
-
- fclose(outfl);
-
-
-
-// Gather the set of interfaces
-// TODO : must update to hanndle machines
-// TODO : lookup interface attributes and add them as a parameter to rts process
- outfl = fopen("runit", "w");
- if(outfl==NULL){
- fprintf(stderr,"Can't open runit for write, exiting.\n");
- exit(0);
- }
-
-
- fputs(
-("#!/bin/sh\n"
-"./stopit\n"
-+root_path+"/bin/gshub.py> /dev/null 2>&1 &\n"
-"sleep 5\n"
-"if [ ! -f gshub.log ]\n"
-"then\n"
-"\techo \"Failed to start bin/gshub.py\"\n"
-"\texit -1\n"
-"fi\n"
-"ADDR=`cat gshub.log`\n"
-"ps opgid= $! >> gs.pids\n"
-"./rts $ADDR default ").c_str(), outfl);
-// int erri;
-// string err_str;
- for(ssi=ifaces.begin();ssi!=ifaces.end();++ssi){
- string ifnm = (*ssi);
- fprintf(outfl, "%s ",ifnm.c_str());
- vector<string> ifv = ifdb->get_iface_vals(ifmachines[ifnm],ifnm, "Command", erri, err_str);
- for(j=0;j<ifv.size();++j)
- fprintf(outfl, "%s ",ifv[j].c_str());
- }
- fprintf(outfl, " &\n");
- fprintf(outfl, "echo $! >> gs.pids\n");
- for(i=0;i<hfta_names.size();++i)
- fprintf(outfl,"./%s $ADDR default &\n",hfta_names[i].c_str());
-
- for(j=0;j<opviews.opview_list.size();++j){
- fprintf(outfl,"%s/views/%s %s &\n",root_path.c_str(),opviews.opview_list[j]->exec_fl.c_str(), opviews.opview_list[j]->mangler.c_str());
- }
-
- fclose(outfl);
- system("chmod +x runit");
-
- outfl = fopen("stopit", "w");
- if(outfl==NULL){
- fprintf(stderr,"Can't open stopit for write, exiting.\n");
- exit(0);
- }
-
- fprintf(outfl,"#!/bin/sh\n"
-"rm -f gshub.log\n"
-"if [ ! -f gs.pids ]\n"
-"then\n"
-"exit\n"
-"fi\n"
-"for pgid in `cat gs.pids`\n"
-"do\n"
-"kill -TERM -$pgid\n"
-"done\n"
-"sleep 1\n"
-"for pgid in `cat gs.pids`\n"
-"do\n"
-"kill -9 -$pgid\n"
-"done\n"
-"rm gs.pids\n");
-
- fclose(outfl);
- system("chmod +x stopit");
-
-//-----------------------------------------------
-
-/* For now disable support for virtual interfaces
- outfl = fopen("set_vinterface_hash.bat", "w");
- if(outfl==NULL){
- fprintf(stderr,"Can't open set_vinterface_hash.bat for write, exiting.\n");
- exit(0);
- }
-
-// The format should be determined by an entry in the ifres.xml file,
-// but for now hardcode the only example I have.
- for(ssi=base_vifaces.begin();ssi!=base_vifaces.end();++ssi){
- if(rts_hload.count((*ssi))){
- string iface_name = (*ssi);
- string iface_number = "";
- for(j=iface_name.size()-1; j>0 && iface_number == ""; j--){
- if(isdigit(iface_name[j])){
- iface_number = iface_name[j];
- if(j>0 && isdigit(iface_name[j-1]))
- iface_number = iface_name[j-1] + iface_number;
- }
- }
-
- fprintf(outfl,"dagconfig -d%s -S hat_range=",iface_number.c_str());
- vector<int> halloc = rts_hload[iface_name];
- int prev_limit = 0;
- for(j=0;j<halloc.size();++j){
- if(j>0)
- fprintf(outfl,":");
- fprintf(outfl,"%d-%d",prev_limit,halloc[j]);
- prev_limit = halloc[j];
- }
- fprintf(outfl,"\n");
- }
- }
- fclose(outfl);
- system("chmod +x set_vinterface_hash.bat");
-*/
-}
-
-// Code for implementing a local schema
-/*
- table_list qpSchema;
-
-// Load the schemas of any LFTAs.
- int l;
- for(l=0;l<hfta_nbr;++l){
- stream_query *sq0 = split_queries[l];
- table_def *td = sq0->get_output_tabledef();
- qpSchema.append_table(td);
- }
-// load the schemas of any other ref'd tables.
-// (e.g., hftas)
- vector<tablevar_t *> input_tbl_names = split_queries[hfta_nbr]->get_input_tables();
- int ti;
- for(ti=0;ti<input_tbl_names.size();++ti){
- int tbl_ref = qpSchema.get_table_ref(input_tbl_names[ti]->get_schema_name());
- if(tbl_ref < 0){
- tbl_ref = Schema->get_table_ref(input_tbl_names[ti]->get_schema_name());
- if(tbl_ref < 0){
- fprintf(stderr,"ERROR file %s references table %s, which is not in the schema.\n",input_file_names[i].c_str(), (input_tbl_names[ti]->get_schema_name()).c_str());
- exit(1);
- }
- qpSchema.append_table(Schema->get_table(tbl_ref));
- }
- }
-*/
-
-// Functions related to parsing.
-
-/*
-static int split_string(char *instr,char sep, char **words,int max_words){
- char *loc;
- char *str;
- int nwords = 0;
-
- str = instr;
- words[nwords++] = str;
- while( (loc = strchr(str,sep)) != NULL){
- *loc = '\0';
- str = loc+1;
- if(nwords >= max_words){
- fprintf(stderr,"Error in split_string, too many words discovered (max is %d)\n",max_words);
- nwords = max_words-1;
- }
- words[nwords++] = str;
- }
-
- return(nwords);
-}
-
-*/
-
+/* ------------------------------------------------\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
+#include<unistd.h> // for gethostname\r
+\r
+#include <string>\r
+#include "parse_fta.h"\r
+#include "parse_schema.h"\r
+#include "parse_ext_fcns.h"\r
+#include"analyze_fta.h"\r
+#include"query_plan.h"\r
+#include"generate_lfta_code.h"\r
+#include"stream_query.h"\r
+#include"generate_utils.h"\r
+#include"nic_def.h"\r
+#include"generate_nic_code.h"\r
+\r
+#include <stdlib.h>\r
+#include <stdio.h>\r
+#include<ctype.h>\r
+#include<glob.h>\r
+#include<string.h>\r
+\r
+#include<list>\r
+\r
+// for the scandir\r
+ #include <sys/types.h>\r
+ #include <dirent.h>\r
+\r
+\r
+#include<errno.h>\r
+\r
+// to verify that some files exist.\r
+ #include <sys/types.h>\r
+ #include <sys/stat.h>\r
+\r
+#include "parse_partn.h"\r
+\r
+#include "print_plan.h"\r
+\r
+// Interface to the xml parser\r
+\r
+#include"xml_t.h"\r
+#include"field_list.h"\r
+\r
+extern int xmlParserparse(void);\r
+extern FILE *xmlParserin;\r
+extern int xmlParserdebug;\r
+\r
+std::vector<std::string> xml_attr_vec;\r
+std::vector<std::string> xml_val_vec;\r
+std::string xml_a, xml_v;\r
+xml_t *xml_leaves = NULL;\r
+\r
+// Interface to the field list verifier\r
+field_list *field_verifier = NULL;\r
+\r
+#define TMPSTRLEN 1000\r
+\r
+#ifndef PATH_DELIM\r
+ #define PATH_DELIM '/'\r
+#endif\r
+\r
+char tmp_schema_str[10000];\r
+\r
+// maximum delay between two hearbeats produced\r
+// by UDOP. Used when its not explicity\r
+// provided in udop definition\r
+#define DEFAULT_UDOP_LIVENESS_TIMEOUT 5\r
+\r
+// Default lfta hash table size, must be power of 2.\r
+int DEFAULT_LFTA_HASH_TABLE_SIZE = 4096;\r
+\r
+// Interface to FTA definition lexer and parser ...\r
+\r
+extern int FtaParserparse(void);\r
+extern FILE *FtaParserin;\r
+extern int FtaParserdebug;\r
+\r
+fta_parse_t *fta_parse_result;\r
+var_defs_t *fta_parse_defines;\r
+\r
+\r
+\r
+// Interface to external function lexer and parser ...\r
+\r
+extern int Ext_fcnsParserparse(void);\r
+extern FILE *Ext_fcnsParserin;\r
+extern int Ext_fcnsParserdebug;\r
+\r
+ext_fcn_list *Ext_fcns;\r
+\r
+\r
+// Interface to partition definition parser\r
+extern int PartnParserparse();\r
+partn_def_list_t *partn_parse_result = NULL;\r
+\r
+\r
+\r
+using namespace std;\r
+//extern int errno;\r
+\r
+\r
+// forward delcaration of local utility function\r
+void generate_makefile(vector<string> &input_file_names, int nfiles,\r
+ vector<string> &hfta_names, opview_set &opviews,\r
+ vector<string> &machine_names,\r
+ string schema_file_name,\r
+ vector<string> &interface_names,\r
+ ifq_t *ifdb, string &config_dir_path,\r
+ bool use_pads,\r
+ string extra_libs,\r
+ map<string, vector<int> > &rts_hload\r
+ );\r
+\r
+//static int split_string(char *instr,char sep, char **words,int max_words);\r
+#define MAXFLDS 100\r
+\r
+ FILE *schema_summary_output = NULL; // query names\r
+\r
+// Dump schema summary\r
+void dump_summary(stream_query *str){\r
+ fprintf(schema_summary_output,"%s\n",str->query_name.c_str());\r
+\r
+ table_def *sch = str->get_output_tabledef();\r
+\r
+ vector<field_entry *> flds = sch->get_fields();\r
+ int f;\r
+ for(f=0;f<flds.size();++f){\r
+ if(f>0) fprintf(schema_summary_output,"|");\r
+ fprintf(schema_summary_output,"%s",flds[f]->get_name().c_str());\r
+ }\r
+ fprintf(schema_summary_output,"\n");\r
+ for(f=0;f<flds.size();++f){\r
+ if(f>0) fprintf(schema_summary_output,"|");\r
+ fprintf(schema_summary_output,"%s",flds[f]->get_type().c_str());\r
+ }\r
+ fprintf(schema_summary_output,"\n");\r
+}\r
+\r
+// Globals\r
+string hostname; // name of current host.\r
+int hostname_len;\r
+bool generate_stats = false;\r
+string root_path = "../..";\r
+\r
+\r
+int main(int argc, char **argv){\r
+ char tmpstr[TMPSTRLEN];\r
+ string err_str;\r
+ int q,s,h,f;\r
+\r
+ set<int>::iterator si;\r
+\r
+ vector<string> query_names; // for lfta.c registration\r
+ map<string, vector<int> > mach_query_names; // list queries of machine\r
+ vector<int> snap_lengths; // for lfta.c registration\r
+ vector<string> interface_names; // for lfta.c registration\r
+ vector<string> machine_names; // machine of interface\r
+ vector<bool> lfta_reuse_options; // for lfta.c registration\r
+ vector<int> lfta_liveness_timeouts; // fot qtree.xml generation\r
+ vector<string> hfta_names; // hfta cource code names, for\r
+ // creating make file.\r
+ vector<string> qnames; // ensure unique names\r
+ map<string, int> lfta_names; // keep track of unique lftas.\r
+\r
+\r
+// set these to 1 to debug the parser\r
+ FtaParserdebug = 0;\r
+ Ext_fcnsParserdebug = 0;\r
+\r
+ FILE *lfta_out; // lfta.c output.\r
+ FILE *fta_in; // input file\r
+ FILE *table_schemas_in; // source tables definition file\r
+ FILE *query_name_output; // query names\r
+ FILE *qtree_output; // interconnections of query nodes\r
+\r
+ // -------------------------------\r
+ // Handling of Input Arguments\r
+ // -------------------------------\r
+ char optstr[] = "BDpLC:l:HNQMf:PSh:n:cR:";\r
+ const char *usage_str = "Usage: %s [-B] [-D] [-p] [-L] [-N] [-H] [-Q] [-M] [-C <config directory>] [-l <library directory>] [-f] [-P] [-S] [-n n_virtual_interfaces] [-h hostname] [-R root_path] [schema_file] input_file [input file ...]\n"\r
+ "\t[-B] : debug only (don't create output files)\n"\r
+ "\t[-D] : distributed mode (will use cluster.ifq instead of local .ifq file)\n"\r
+ "\t[-p] : partitioned mode (will use partition information in partition.txt to drive query optimization)\n"\r
+ "\t[-L] : use live_hosts.txt file to restrict queries to a set of live hosts\n"\r
+ "\t[-C] : use <config directory> for definition files\n"\r
+ "\t[-l] : use <library directory> for library queries\n"\r
+ "\t[-N] : output query names in query_names.txt\n"\r
+ "\t[-H] : create HFTA only (no schema_file)\n"\r
+ "\t[-Q] : use query name for hfta suffix\n"\r
+ "\t[-M] : generate make file and runit, stopit scripts\n"\r
+ "\t[-S] : enable LFTA statistics (alters Makefile).\n"\r
+ "\t[-f] : Output schema summary to schema_summary.txt\n"\r
+ "\t[-P] : link with PADS\n"\r
+ "\t[-h] : override host name.\n"\r
+ "\t[-c] : clean out Makefile and hfta_*.cc first.\n"\r
+ "\t[-R] : path to root of GS-lite\n"\r
+;\r
+\r
+// parameters gathered from command line processing\r
+ string external_fcns_path;\r
+// string internal_fcn_path;\r
+ string config_dir_path;\r
+ string library_path = "./";\r
+ vector<string> input_file_names;\r
+ string schema_file_name;\r
+ bool debug_only = false;\r
+ bool hfta_only = false;\r
+ bool output_query_names = false;\r
+ bool output_schema_summary=false;\r
+ bool numeric_hfta_flname = true;\r
+ bool create_makefile = false;\r
+ bool distributed_mode = false;\r
+ bool partitioned_mode = false;\r
+ bool use_live_hosts_file = false;\r
+ bool use_pads = false;\r
+ bool clean_make = false;\r
+ int n_virtual_interfaces = 1;\r
+\r
+ char chopt;\r
+ while((chopt = getopt(argc,argv,optstr)) != -1){\r
+ switch(chopt){\r
+ case 'B':\r
+ debug_only = true;\r
+ break;\r
+ case 'D':\r
+ distributed_mode = true;\r
+ break;\r
+ case 'p':\r
+ partitioned_mode = true;\r
+ break;\r
+ case 'L':\r
+ use_live_hosts_file = true;\r
+ break;\r
+ case 'C':\r
+ if(optarg != NULL)\r
+ config_dir_path = string(optarg) + string("/");\r
+ break;\r
+ case 'l':\r
+ if(optarg != NULL)\r
+ library_path = string(optarg) + string("/");\r
+ break;\r
+ case 'N':\r
+ output_query_names = true;\r
+ break;\r
+ case 'Q':\r
+ numeric_hfta_flname = false;\r
+ break;\r
+ case 'H':\r
+ if(schema_file_name == ""){\r
+ hfta_only = true;\r
+ }\r
+ break;\r
+ case 'f':\r
+ output_schema_summary=true;\r
+ break;\r
+ case 'M':\r
+ create_makefile=true;\r
+ break;\r
+ case 'S':\r
+ generate_stats=true;\r
+ break;\r
+ case 'P':\r
+ use_pads = true;\r
+ break;\r
+ case 'c':\r
+ clean_make = true;\r
+ break;\r
+ case 'h':\r
+ if(optarg != NULL)\r
+ hostname = optarg;\r
+ break;\r
+ case 'R':\r
+ if(optarg != NULL)\r
+ root_path = optarg;\r
+ break;\r
+ case 'n':\r
+ if(optarg != NULL){\r
+ n_virtual_interfaces = atoi(optarg);\r
+ if(n_virtual_interfaces < 1 || n_virtual_interfaces > 128){\r
+ fprintf(stderr,"Warning, %d virtual interfaces specified, valid values are between 1 and 128 inclusive\n",n_virtual_interfaces);\r
+ n_virtual_interfaces = 1;\r
+ }\r
+ }\r
+ break;\r
+ case '?':\r
+ fprintf(stderr,"Error, argument %c not recognized.\n",optopt);\r
+ fprintf(stderr,"%s\n", usage_str);\r
+ exit(1);\r
+ default:\r
+ fprintf(stderr, "Argument was %c\n", optopt);\r
+ fprintf(stderr,"Invalid arguments\n");\r
+ fprintf(stderr,"%s\n", usage_str);\r
+ exit(1);\r
+ }\r
+ }\r
+ argc -= optind;\r
+ argv += optind;\r
+ for (int i = 0; i < argc; ++i) {\r
+ if((schema_file_name == "") && !hfta_only){\r
+ schema_file_name = argv[i];\r
+ }else{\r
+ input_file_names.push_back(argv[i]);\r
+ }\r
+ }\r
+\r
+ if(input_file_names.size() == 0){\r
+ fprintf(stderr,"%s\n", usage_str);\r
+ exit(1);\r
+ }\r
+\r
+ if(clean_make){\r
+ string clean_cmd = "rm Makefile hfta_*.cc";\r
+ int clean_ret = system(clean_cmd.c_str());\r
+ if(clean_ret){\r
+ fprintf(stderr,"Warning, return value %d when trying to clean out old source files.\n", clean_ret);\r
+ }\r
+ }\r
+\r
+\r
+ nic_prop_db *npdb = new nic_prop_db(config_dir_path);\r
+\r
+// Open globally used file names.\r
+\r
+ // prepend config directory to schema file\r
+ schema_file_name = config_dir_path + schema_file_name;\r
+ external_fcns_path = config_dir_path + string("external_fcns.def");\r
+ string ifx_fname = config_dir_path + string("ifres.xml");\r
+\r
+// Find interface query file(s).\r
+ if(hostname == ""){\r
+ gethostname(tmpstr,TMPSTRLEN);\r
+ hostname = tmpstr;\r
+ }\r
+ hostname_len = strlen(tmpstr);\r
+ string ifq_fname = config_dir_path + (distributed_mode ? "cluster" : hostname) + string(".ifq");\r
+ vector<string> ifq_fls;\r
+\r
+ ifq_fls.push_back(ifq_fname);\r
+\r
+\r
+// Get the field list, if it exists\r
+ string flist_fl = config_dir_path + "field_list.xml";\r
+ FILE *flf_in = NULL;\r
+ if((flf_in = fopen(flist_fl.c_str(), "r")) != NULL) {\r
+ fprintf(stderr,"Found field list file %s\n",flist_fl.c_str());\r
+ xml_leaves = new xml_t();\r
+ xmlParser_setfileinput(flf_in);\r
+ if(xmlParserparse()){\r
+ fprintf(stderr,"WARNING, could not parse field list file %s\n",flist_fl.c_str());\r
+ }else{\r
+ field_verifier = new field_list(xml_leaves);\r
+ }\r
+ }\r
+\r
+ if(!hfta_only){\r
+ if((table_schemas_in = fopen(schema_file_name.c_str(), "r")) == NULL) {\r
+ fprintf(stderr,"Can't open schema file %s\n%s\n",schema_file_name.c_str(),strerror(errno));\r
+ exit(1);\r
+ }\r
+ }\r
+\r
+/*\r
+ if(!(debug_only || hfta_only)){\r
+ if((lfta_out = fopen("lfta.c","w")) == NULL){\r
+ fprintf(stderr,"Can't open output file %s\n%s\n","lfta.c",strerror(errno));\r
+ exit(1);\r
+ }\r
+ }\r
+*/\r
+\r
+// Get the output specification file.\r
+// format is\r
+// query, operator, operator_param, directory, bucketwidth, partitioning_se, n_partitions\r
+ string ospec_fl = "output_spec.cfg";\r
+ FILE *osp_in = NULL;\r
+ vector<ospec_str *> output_specs;\r
+ multimap<string, int> qname_to_ospec;\r
+ if((osp_in = fopen(ospec_fl.c_str(), "r")) != NULL) {\r
+ char *flds[MAXFLDS];\r
+ int o_lineno = 0;\r
+ while(fgets(tmpstr,TMPSTRLEN,osp_in)){\r
+ o_lineno++;\r
+ int nflds = split_string(tmpstr,',',flds,MAXFLDS);\r
+ if(nflds == 7){\r
+// make operator type lowercase\r
+ char *tmpc;\r
+ for(tmpc=flds[1];*tmpc!='\0';++tmpc)\r
+ *tmpc = tolower(*tmpc);\r
+\r
+ ospec_str *tmp_ospec = new ospec_str();\r
+ tmp_ospec->query = flds[0];\r
+ tmp_ospec->operator_type = flds[1];\r
+ tmp_ospec->operator_param = flds[2];\r
+ tmp_ospec->output_directory = flds[3];\r
+ tmp_ospec->bucketwidth = atoi(flds[4]);\r
+ tmp_ospec->partitioning_flds = flds[5];\r
+ tmp_ospec->n_partitions = atoi(flds[6]);\r
+ qname_to_ospec.insert(pair<string,int>(tmp_ospec->query,output_specs.size()));\r
+ output_specs.push_back(tmp_ospec);\r
+ }else{\r
+ fprintf(stderr,"Warning, line %d corrupted in output_spec.cfg, has %d fields.\n",o_lineno,nflds);\r
+ }\r
+ }\r
+ fclose(osp_in);\r
+ }else{\r
+ fprintf(stderr,"output_spec.cfg not found. The query set has no output. exiting.\n");\r
+ exit(1);\r
+ }\r
+\r
+// hfta parallelism\r
+ string pspec_fl = "hfta_parallelism.cfg";\r
+ FILE *psp_in = NULL;\r
+ map<string, int> hfta_parallelism;\r
+ if((psp_in = fopen(pspec_fl.c_str(), "r")) != NULL){\r
+ char *flds[MAXFLDS];\r
+ int o_lineno = 0;\r
+ while(fgets(tmpstr,TMPSTRLEN,psp_in)){\r
+ bool good_entry = true;\r
+ o_lineno++;\r
+ int nflds = split_string(tmpstr,',',flds,MAXFLDS);\r
+ if(nflds == 2){\r
+ string hname = flds[0];\r
+ int par = atoi(flds[1]);\r
+ if(par <= 0 || par > n_virtual_interfaces){\r
+ fprintf(stderr,"Warning, line %d of %s is incorrectly formatted, parallelism is %d, must be between 1 and %d\n",o_lineno,pspec_fl.c_str(),par,n_virtual_interfaces);\r
+ good_entry = false;\r
+ }\r
+ if(good_entry && n_virtual_interfaces % par != 0){\r
+ fprintf(stderr,"Warning, line %d of %s is incorrectly formatted, parallelism is %d, must divide the number of virtual interfaces (%d), ignoring.\n",o_lineno,pspec_fl.c_str(),par,n_virtual_interfaces);\r
+ good_entry = false;\r
+ }\r
+ if(good_entry)\r
+ hfta_parallelism[hname] = par;\r
+ }\r
+ }\r
+ }else{\r
+ fprintf(stderr,"WARNING, no file %s found, using single copies of hftas.\n",pspec_fl.c_str());\r
+ }\r
+\r
+\r
+// LFTA hash table sizes\r
+ string htspec_fl = "lfta_htsize.cfg";\r
+ FILE *htsp_in = NULL;\r
+ map<string, int> lfta_htsize;\r
+ if((htsp_in = fopen(htspec_fl.c_str(), "r")) != NULL){\r
+ char *flds[MAXFLDS];\r
+ int o_lineno = 0;\r
+ while(fgets(tmpstr,TMPSTRLEN,htsp_in)){\r
+ bool good_entry = true;\r
+ o_lineno++;\r
+ int nflds = split_string(tmpstr,',',flds,MAXFLDS);\r
+ if(nflds == 2){\r
+ string lfta_name = flds[0];\r
+ int htsz = atoi(flds[1]);\r
+ if(htsz>0){\r
+ lfta_htsize[lfta_name] = htsz;\r
+ }else{\r
+ fprintf(stderr,"Warning, line %d of %s is incorrectly formatted, htsize is %d, must be larger than 0.\n",o_lineno,htspec_fl.c_str(),htsz);\r
+ }\r
+ }\r
+ }\r
+ }else{\r
+ fprintf(stderr,"WARNING, no file %s found, using default LFTA hash table sizes.\n",htspec_fl.c_str());\r
+ }\r
+\r
+// LFTA vitual interface hash split\r
+ string rtlspec_fl = "rts_load.cfg";\r
+ FILE *rtl_in = NULL;\r
+ map<string, vector<int> > rts_hload;\r
+ if((rtl_in = fopen(rtlspec_fl.c_str(), "r")) != NULL){\r
+ char *flds[MAXFLDS];\r
+ int r_lineno = 0;\r
+ string iface_name;\r
+ vector<int> hload;\r
+ while(fgets(tmpstr,TMPSTRLEN,rtl_in)){\r
+ bool good_entry = true;\r
+ r_lineno++;\r
+ iface_name = "";\r
+ hload.clear();\r
+ int nflds = split_string(tmpstr,',',flds,MAXFLDS);\r
+ if(nflds >1){\r
+ iface_name = flds[0];\r
+ int cumm_h = 0;\r
+ int j;\r
+ for(j=1;j<nflds;++j){\r
+ int h = atoi(flds[j]);\r
+ if(h<=0)\r
+ good_entry = false;\r
+ cumm_h += h;\r
+ hload.push_back(cumm_h);\r
+ }\r
+ }else{\r
+ good_entry = false;\r
+ }\r
+ if(good_entry){\r
+ rts_hload[iface_name] = hload;\r
+ }else{\r
+ fprintf(stderr,"Warning, line %d of %s is malformed, skipping.\n",r_lineno, rtlspec_fl.c_str());\r
+ }\r
+ }\r
+ }\r
+\r
+\r
+\r
+ if(output_query_names){\r
+ if((query_name_output = fopen("query_names.txt","w")) == NULL){\r
+ fprintf(stderr,"Can't open output file %s\n%s\n","query_names.txt",strerror(errno));\r
+ exit(1);\r
+ }\r
+ }\r
+\r
+ if(output_schema_summary){\r
+ if((schema_summary_output = fopen("schema_summary.txt","w")) == NULL){\r
+ fprintf(stderr,"Can't open output file %s\n%s\n","schema_summary.txt",strerror(errno));\r
+ exit(1);\r
+ }\r
+ }\r
+\r
+ if((qtree_output = fopen("qtree.xml","w")) == NULL){\r
+ fprintf(stderr,"Can't open output file %s\n%s\n","qtree.xml",strerror(errno));\r
+ exit(1);\r
+ }\r
+ fprintf(qtree_output,"<?xml version='1.0' encoding='ISO-8859-1'?>\n");\r
+ fprintf(qtree_output,"<?xml-stylesheet type='text/xsl' href='qtree.xsl'?>\n");\r
+ fprintf(qtree_output,"<QueryNodes>\n");\r
+\r
+\r
+// Get an initial Schema\r
+ table_list *Schema;\r
+ if(!hfta_only){\r
+// Parse the table schema definitions.\r
+ fta_parse_result = new fta_parse_t();\r
+ FtaParser_setfileinput(table_schemas_in);\r
+ if(FtaParserparse()){\r
+ fprintf(stderr,"Table schema parse failed.\n");\r
+ exit(1);\r
+ }\r
+ if(fta_parse_result->parse_type != TABLE_PARSE){\r
+ fprintf(stderr,"ERROR, file %s is not a table definition file.\n",schema_file_name.c_str());\r
+ exit(1);\r
+ }\r
+ Schema = fta_parse_result->tables;\r
+\r
+// Process schema field inheritance\r
+ int retval;\r
+ retval = Schema->unroll_tables(err_str);\r
+ if(retval){\r
+ fprintf(stderr,"Error processing schema filed inheritance:\n %s\n", err_str.c_str() );\r
+ exit(1);\r
+ }\r
+ }else{\r
+// hfta only => we will try to fetch schemas from the registry.\r
+// therefore, start off with an empty schema.\r
+ Schema = new table_list();\r
+ }\r
+\r
+\r
+// Open and parse the external functions file.\r
+ Ext_fcnsParserin = fopen(external_fcns_path.c_str(),"r");\r
+ if(Ext_fcnsParserin == NULL){\r
+ fprintf(stderr,"Warning, can't find external functions definition file (external_fcns.def).\n");\r
+ Ext_fcns = new ext_fcn_list();\r
+ }else{\r
+ if(Ext_fcnsParserparse()){\r
+ fprintf(stderr,"Warning, error parsing external functions definition file (external_fcns.def).\n");\r
+ Ext_fcns = new ext_fcn_list();\r
+ }\r
+ }\r
+ if(Ext_fcns->validate_fcns(err_str)){\r
+ fprintf(stderr,"Error processing external functions definition file (external_fcns.def):\n%s\n",err_str.c_str());\r
+ exit(1);\r
+ }\r
+\r
+// Open and parse the interface resources file.\r
+// ifq_t *ifaces_db = new ifq_t();\r
+// string ierr;\r
+// if(ifaces_db->load_ifaces(ifx_fname,use_live_hosts_file,distributed_mode,ierr)){\r
+// fprintf(stderr,"ERROR, can't load interface resource file %s :\n%s",\r
+// ifx_fname.c_str(), ierr.c_str());\r
+// exit(1);\r
+// }\r
+// if(ifaces_db->load_ifqs(ifq_fname, ierr)){\r
+// fprintf(stderr,"ERROR, can't load interface query file %s :\n%s",\r
+// ifq_fname.c_str(), ierr.c_str());\r
+// exit(1);\r
+// }\r
+\r
+\r
+// The LFTA code string.\r
+// Put the standard preamble here.\r
+// NOTE: the hash macros, fcns should go into the run time\r
+ map<string, string> lfta_val;\r
+ map<string, string> lfta_prefilter_val;\r
+\r
+ string lfta_header =\r
+"#include <limits.h>\n\n"\r
+"#include \"rts.h\"\n"\r
+"#include \"fta.h\"\n"\r
+"#include \"lapp.h\"\n"\r
+"#include \"rts_udaf.h\"\n\n"\r
+;\r
+// Get any locally defined parsing headers\r
+ glob_t glob_result;\r
+ memset(&glob_result, 0, sizeof(glob_result));\r
+\r
+ // do the glob operation\r
+ int return_value = glob("../../include/lfta/local/*h", GLOB_TILDE, NULL, &glob_result);\r
+ if(return_value == 0){\r
+ for(size_t i = 0; i < glob_result.gl_pathc; ++i) {\r
+ char *flds[1000];\r
+ int nflds = split_string(glob_result.gl_pathv[i],'/',flds,1000);\r
+ lfta_header += "#include \"local/"+string(flds[nflds-1])+"\"\n\n";\r
+ }\r
+ }else{\r
+ fprintf(stderr,"Warning, glob on ../../include/lfta/local/*h failed.\n");\r
+ }\r
+\r
+/*\r
+"#define IS_FILLED(bitmap,bucket) (bitmap[bucket >> 4] & (0x80000000 >> ((bucket & 15)<<1)))\n"\r
+"#define IS_NEW(bitmap,bucket) (bitmap[bucket >> 4] & (0x80000000 >> (((bucket & 15) << 1) + 1)))\n"\r
+"#define SET_EMPTY(bitmap,bucket) (bitmap[bucket >> 4] &= (~(0x80000000 >> ((bucket & 15)<<1))))\n"\r
+"#define SET_FILLED_AND_NEW(bitmap,bucket) (bitmap[bucket >> 4] |= (0xC0000000 >> ((bucket & 15)<<1)))\n"\r
+*/\r
+\r
+ lfta_header += \r
+"\n"\r
+"gs_uint64_t (*lfta_prefilter)(void *pkt) = NULL; // set at fta_init\n"\r
+"\n"\r
+"#define SLOT_FILLED 0x04\n"\r
+"#define SLOT_GEN_BITS 0x03\n"\r
+"#define SLOT_HASH_BITS 0xfffffff8\n"\r
+"#define SET_BF_BIT(bf_table, bf_num, bf_index, bucket) (bf_table[(((bucket) * (bf_num)) + (bf_index)) >> 3] |= (0x80 >> ((((bucket) * (bf_num)) + (bf_index)) & 7)))\n"\r
+"#define IS_BF_SET(bf_table, bf_num, bf_index, bucket) (bf_table[(((bucket) * (bf_num))+(bf_index)) >> 3] & (0x80 >> ((((bucket) * (bf_num))+(bf_index)) & 7)))\n"\r
+"#define SET_BF_EMPTY(bf_table, bf_num, bf_index, bucket) (bf_table[((bucket * bf_num)+bf_index) >> 3] &= (~0x80 >> (((bucket * bf_num)+bf_index) & 7)))\n"\r
+"\n\n"\r
+\r
+"#define lfta_BOOL_to_hash(x) (x)\n"\r
+"#define lfta_USHORT_to_hash(x) (x)\n"\r
+"#define lfta_UINT_to_hash(x) (x)\n"\r
+"#define lfta_IP_to_hash(x) (x)\n"\r
+"#define lfta_IPV6_to_hash(x) ( (x.v[0]) ^ (x.v[1]) ^ (x.v[2]) ^ (x.v[3]))\n"\r
+"#define lfta_INT_to_hash(x) (gs_uint32_t)(x)\n"\r
+"#define lfta_ULLONG_to_hash(x) ( (( (x) >>32)&0xffffffff) ^ ((x)&0xffffffff) )\n"\r
+"#define lfta_LLONG_to_hash(x) ( (( (x) >>32)&0xffffffff) ^ ((x)&0xffffffff) )\n"\r
+"#define lfta_FLOAT_to_hash(x) ( (( ((gs_uint64_t)(x)) >>32)&0xffffffff) ^ (((gs_uint64_t)(x))&0xffffffff) )\n"\r
+"static gs_uint32_t lfta_V_STR_to_hash(struct gs_string x){\n"\r
+" gs_uint32_t i,ret=0,tmp_sum = 0;\n"\r
+" for(i=0;i<x.length;++i){\n"\r
+" tmp_sum |= (x.data[i]) << (8*(i%4));\n"\r
+" if((i%4) == 3){\n"\r
+" ret ^= tmp_sum;\n"\r
+" tmp_sum = 0;\n"\r
+" }\n"\r
+" }\n"\r
+" if((i%4)!=0) ret ^=tmp_sum;\n"\r
+" return(ret);\n"\r
+"}\n\n\n";\r
+\r
+\r
+\r
+//////////////////////////////////////////////////////////////////\r
+///// Get all of the query parse trees\r
+\r
+\r
+ int i,p;\r
+ int hfta_count = 0; // for numeric suffixes to hfta .cc files\r
+\r
+//---------------------------\r
+// Global info needed for post processing.\r
+\r
+// Set of operator views ref'd in the query set.\r
+ opview_set opviews;\r
+// lftas on a per-machine basis.\r
+ map<string, vector<stream_query *> > lfta_mach_lists;\r
+ int nfiles = input_file_names.size();\r
+ vector<stream_query *> hfta_list; // list of hftas.\r
+ map<string, stream_query *> sq_map; // map from query name to stream query.\r
+\r
+\r
+//////////////////////////////////////////\r
+\r
+// Open and parse the interface resources file.\r
+ ifq_t *ifaces_db = new ifq_t();\r
+ string ierr;\r
+ if(ifaces_db->load_ifaces(ifx_fname,use_live_hosts_file,distributed_mode,ierr)){\r
+ fprintf(stderr,"ERROR, can't load interface resource file %s :\n%s",\r
+ ifx_fname.c_str(), ierr.c_str());\r
+ exit(1);\r
+ }\r
+ if(ifaces_db->load_ifqs(ifq_fls[0], ierr)){\r
+ fprintf(stderr,"ERROR, can't load interface query file %s :\n%s",\r
+ ifq_fls[0].c_str(), ierr.c_str());\r
+ exit(1);\r
+ }\r
+\r
+ map<string, string> qname_to_flname; // for detecting duplicate query names\r
+\r
+\r
+\r
+// Parse the files to create a vector of parse trees.\r
+// Load qnodes with information to perform a topo sort\r
+// based on query dependencies.\r
+ vector<query_node *> qnodes; // for topo sort.\r
+ map<string,int> name_node_map; // map query name to qnodes entry\r
+ for(i=0;i<input_file_names.size();i++){\r
+\r
+ if((fta_in = fopen(input_file_names[i].c_str(), "r")) == NULL) {\r
+ fprintf(stderr,"Can't open input file %s\n%s",input_file_names[i].c_str(),strerror(errno));\r
+ continue;\r
+ }\r
+fprintf(stderr,"Parsing file %s\n",input_file_names[i].c_str());\r
+\r
+// Parse the FTA query\r
+ fta_parse_result = new fta_parse_t();\r
+ FtaParser_setfileinput(fta_in);\r
+ if(FtaParserparse()){\r
+ fprintf(stderr,"FTA parse failed.\n");\r
+ exit(1);\r
+ }\r
+ if(fta_parse_result->parse_type != QUERY_PARSE){\r
+ fprintf(stderr,"ERROR, file %s is not a query file.\n",input_file_names[i].c_str());\r
+ exit(1);\r
+ }\r
+\r
+// returns a list of parse trees\r
+ vector<table_exp_t *> qlist = fta_parse_result->parse_tree_list->qlist;\r
+ for(p=0;p<qlist.size();++p){\r
+ table_exp_t *fta_parse_tree = qlist[p];\r
+// query_parse_trees.push_back(fta_parse_tree);\r
+\r
+// compute the default name -- extract from query name\r
+ strcpy(tmpstr,input_file_names[i].c_str());\r
+ char *qname = strrchr(tmpstr,PATH_DELIM);\r
+ if(qname == NULL)\r
+ qname = tmpstr;\r
+ else\r
+ qname++;\r
+ char *qname_end = strchr(qname,'.');\r
+ if(qname_end != NULL) *qname_end = '\0';\r
+ string qname_str = qname;\r
+ string imputed_qname = impute_query_name(fta_parse_tree, qname_str);\r
+\r
+// Deternmine visibility. Should I be attaching all of the output methods?\r
+ if(qname_to_ospec.count(imputed_qname)>0)\r
+ fta_parse_tree->set_visible(true);\r
+ else\r
+ fta_parse_tree->set_visible(false);\r
+\r
+\r
+// Create a manipulable repesentation of the parse tree.\r
+// the qnode inherits the visibility assigned to the parse tree.\r
+ int pos = qnodes.size();\r
+ qnodes.push_back( new query_node(pos, imputed_qname, input_file_names[i], fta_parse_tree ));\r
+ name_node_map[ qnodes[pos]->name ] = pos;\r
+//printf("name_node_map[ %s ] = %d\n",qnodes[pos]->name.c_str(),pos);\r
+// qnames.push_back(impute_query_name(fta_parse_tree, qname_str));\r
+// qfiles.push_back(i);\r
+\r
+// Check for duplicate query names\r
+// NOTE : in hfta-only generation, I should\r
+// also check with the names of the registered queries.\r
+ if(qname_to_flname.count(qnodes[pos]->name) > 0){\r
+ fprintf(stderr,"ERROR duplicate query name %s in files %s and %s.\n",\r
+ qnodes[pos]->name.c_str(), (qname_to_flname[qnodes[pos]->name]).c_str(), input_file_names[i].c_str());\r
+ exit(1);\r
+ }\r
+ if(Schema->find_tbl(qnodes[pos]->name) >= 0){\r
+ fprintf(stderr,"ERROR duplicate query name %s (file %s) was already defined as a PROTOCOL.\n",\r
+ qnodes[pos]->name.c_str(), input_file_names[i].c_str());\r
+ exit(1);\r
+ }\r
+ qname_to_flname[qnodes[pos]->name] = input_file_names[i].c_str();\r
+\r
+\r
+ }\r
+ }\r
+\r
+// Add the library queries\r
+\r
+ int pos;\r
+ for(pos=0;pos<qnodes.size();++pos){\r
+ int fi;\r
+ for(fi = 0;fi<qnodes[pos]->refd_tbls.size();++fi){\r
+ string src_tbl = qnodes[pos]->refd_tbls[fi];\r
+ if(qname_to_flname.count(src_tbl) == 0){\r
+ int last_sep = src_tbl.find_last_of('/');\r
+ if(last_sep != string::npos){\r
+fprintf(stderr,"Finding library query %s for query %s\n",src_tbl.c_str(),qnodes[pos]->name.c_str());\r
+ string target_qname = src_tbl.substr(last_sep+1);\r
+ string qpathname = library_path + src_tbl + ".gsql";\r
+ if((fta_in = fopen(qpathname.c_str(), "r")) == NULL) {\r
+ fprintf(stderr,"Can't open library file %s, referenced by query %s in file %s\n\t%s\n",qpathname.c_str(),qnodes[pos]->name.c_str(), qname_to_flname[qnodes[pos]->name].c_str(), strerror(errno));\r
+ exit(1);\r
+ fprintf(stderr,"After exit\n");\r
+ }\r
+fprintf(stderr,"Parsing file %s\n",qpathname.c_str());\r
+// Parse the FTA query\r
+ fta_parse_result = new fta_parse_t();\r
+ FtaParser_setfileinput(fta_in);\r
+ if(FtaParserparse()){\r
+ fprintf(stderr,"FTA parse failed.\n");\r
+ exit(1);\r
+ }\r
+ if(fta_parse_result->parse_type != QUERY_PARSE){\r
+ fprintf(stderr,"ERROR, file %s is not a query file.\n",input_file_names[i].c_str());\r
+ exit(1);\r
+ }\r
+\r
+ map<string, int> local_query_map;\r
+ vector<string> local_query_names;\r
+ vector<table_exp_t *> qlist = fta_parse_result->parse_tree_list->qlist;\r
+ for(p=0;p<qlist.size();++p){\r
+ table_exp_t *fta_parse_tree = qlist[p];\r
+ fta_parse_tree->set_visible(false); // assumed to not produce output\r
+ string imputed_qname = impute_query_name(fta_parse_tree, target_qname);\r
+ if(imputed_qname == target_qname)\r
+ imputed_qname = src_tbl;\r
+ if(local_query_map.count(imputed_qname)>0){\r
+ fprintf(stderr,"ERROR, multiple queries named %s in library file %s\n",imputed_qname.c_str(), qpathname.c_str());\r
+ exit(1);\r
+ }\r
+ local_query_map[ imputed_qname ] = p;\r
+ local_query_names.push_back(imputed_qname);\r
+ }\r
+\r
+ if(local_query_map.count(src_tbl)==0){\r
+ fprintf(stderr,"ERROR, library query file %s has no query named %s\n",qpathname.c_str(),target_qname.c_str());\r
+ exit(1);\r
+ }\r
+\r
+ vector<int> worklist;\r
+ set<int> added_queries;\r
+ vector<query_node *> new_qnodes;\r
+ worklist.push_back(local_query_map[target_qname]);\r
+ added_queries.insert(local_query_map[target_qname]);\r
+ int qq;\r
+ int qpos = qnodes.size();\r
+ for(qq=0;qq<worklist.size();++qq){\r
+ int q_id = worklist[qq];\r
+ query_node *new_qnode = new query_node(qpos+qq, local_query_names[q_id], qpathname, qlist[q_id] );\r
+ new_qnodes.push_back( new_qnode);\r
+ vector<string> refd_tbls = new_qnode->refd_tbls;\r
+ int ff;\r
+ for(ff = 0;ff<refd_tbls.size();++ff){\r
+ if(local_query_map.count(refd_tbls[ff])>0 && added_queries.count(local_query_map[refd_tbls[ff]])==0){\r
+\r
+ if(name_node_map.count(refd_tbls[ff])>0){\r
+ fprintf(stderr,"ERROR, query %s occurs both in the regular query set, file %s, and in library file %s\n",refd_tbls[ff].c_str(), qname_to_flname[refd_tbls[ff]].c_str(), qpathname.c_str() );\r
+ exit(1);\r
+ }else{\r
+ worklist.push_back(local_query_map[refd_tbls[ff]]);\r
+ }\r
+ }\r
+ }\r
+ }\r
+\r
+ for(qq=0;qq<new_qnodes.size();++qq){\r
+ int qpos = qnodes.size();\r
+ qnodes.push_back(new_qnodes[qq]);\r
+ name_node_map[qnodes[qpos]->name ] = qpos;\r
+ qname_to_flname[qnodes[qpos]->name ] = qpathname;\r
+ }\r
+ }\r
+ }\r
+ }\r
+ }\r
+\r
+\r
+\r
+\r
+\r
+\r
+\r
+\r
+//---------------------------------------\r
+\r
+\r
+// Add the UDOPS.\r
+\r
+ string udop_missing_sources;\r
+ for(i=0;i<qnodes.size();++i){\r
+ int fi;\r
+ for(fi = 0;fi<qnodes[i]->refd_tbls.size();++fi){\r
+ int sid = Schema->find_tbl(qnodes[i]->refd_tbls[fi]);\r
+ if(sid >= 0){\r
+ if(Schema->get_schema_type(sid) == OPERATOR_VIEW_SCHEMA){\r
+ if(name_node_map.count(qnodes[i]->refd_tbls[fi]) == 0){\r
+ int pos = qnodes.size();\r
+ qnodes.push_back( new query_node(pos, qnodes[i]->refd_tbls[fi], Schema));\r
+ name_node_map[ qnodes[pos]->name ] = pos;\r
+ qnodes[pos]->is_externally_visible = false; // its visible\r
+ // Need to mark the source queries as visible.\r
+ int si;\r
+ string missing_sources = "";\r
+ for(si=0;si<qnodes[pos]->refd_tbls.size();++si){\r
+ string src_tbl = qnodes[pos]->refd_tbls[si];\r
+ if(name_node_map.count(src_tbl)==0){\r
+ missing_sources += src_tbl + " ";\r
+ }\r
+ }\r
+ if(missing_sources != ""){\r
+ udop_missing_sources += "\tUDOP "+qnodes[pos]->name+" references undefined tables "+missing_sources+"\n";\r
+ }\r
+ }\r
+ }\r
+ }\r
+ }\r
+ }\r
+ if(udop_missing_sources != ""){\r
+ fprintf(stderr,"ERROR, User Defined OPerators reference source tables that are not part of the query set:\n%s",udop_missing_sources.c_str());\r
+ exit(1);\r
+ }\r
+\r
+\r
+\r
+////////////////////////////////////////////////////////////////////\r
+/// Check parse trees to verify that some\r
+/// global properties are met :\r
+/// if q1 reads from q2, then\r
+/// q2 is processed before q1\r
+/// q1 can supply q2's parameters\r
+/// Verify there is no cycle in the reads-from graph.\r
+\r
+// Compute an order in which to process the\r
+// queries.\r
+\r
+// Start by building the reads-from lists.\r
+//\r
+\r
+ for(i=0;i<qnodes.size();++i){\r
+ int qi, fi;\r
+ vector<string> refd_tbls = qnodes[i]->refd_tbls;\r
+ for(fi = 0;fi<refd_tbls.size();++fi){\r
+ if(name_node_map.count(refd_tbls[fi])>0){\r
+//printf("query %d (%s) reads from %s (%d)\n", i, qnodes[i]->name.c_str(),refd_tbls[fi].c_str(),name_node_map[refd_tbls[fi]]);\r
+ (qnodes[i]->reads_from).insert(name_node_map[refd_tbls[fi]]);\r
+ }\r
+ }\r
+ }\r
+\r
+\r
+// If one query reads the result of another,\r
+// check for parameter compatibility. Currently it must\r
+// be an exact match. I will move to requiring\r
+// containment after re-ordering, but will require\r
+// some analysis for code generation which is not\r
+// yet in place.\r
+//printf("There are %d query nodes.\n",qnodes.size());\r
+\r
+\r
+ for(i=0;i<qnodes.size();++i){\r
+ vector<var_pair_t *> target_params = qnodes[i]->params;\r
+ for(si=qnodes[i]->reads_from.begin();si!=qnodes[i]->reads_from.end();++si){\r
+ vector<var_pair_t *> source_params = qnodes[(*si)]->params;\r
+ if(target_params.size() != source_params.size()){\r
+ fprintf(stderr,"ERROR, query %s (in file %s) reads from %s (in file %s), but they do not have identical parameters.\n",qnodes[i]->name.c_str(), qnodes[i]->file.c_str(), qnodes[(*si)]->name.c_str(), qnodes[(*si)]->file.c_str());\r
+ exit(1);\r
+ }\r
+ int p;\r
+ for(p=0;p<target_params.size();++p){\r
+ if(! (target_params[p]->name == source_params[p]->name &&\r
+ target_params[p]->val == source_params[p]->val ) ){\r
+ fprintf(stderr,"ERROR, query %s (in file %s) reads from %s (in file %s), but they do not have identical parameters.\n",qnodes[i]->name.c_str(), qnodes[i]->file.c_str(), qnodes[(*si)]->name.c_str(), qnodes[(*si)]->file.c_str());\r
+ exit(1);\r
+ }\r
+ }\r
+ }\r
+ }\r
+\r
+\r
+// Find the roots.\r
+// Start by counting inedges.\r
+ for(i=0;i<qnodes.size();++i){\r
+ for(si=qnodes[i]->reads_from.begin();si!=qnodes[i]->reads_from.end();++si){\r
+ qnodes[(*si)]->n_consumers++;\r
+ }\r
+ }\r
+\r
+// The roots are the nodes with indegree zero.\r
+ set<int> roots;\r
+ for(i=0;i<qnodes.size();++i){\r
+ if(qnodes[i]->n_consumers == 0){\r
+ if(qnodes[i]->is_externally_visible == false){\r
+ fprintf(stderr,"WARNING: query %s (file %s) is a root query but it isn't visible. Ignored.\n",qnodes[i]->name.c_str(), qnodes[i]->file.c_str());\r
+ }\r
+ roots.insert(i);\r
+ }\r
+ }\r
+\r
+// Remove the parts of the subtree that produce no output.\r
+ set<int> valid_roots;\r
+ set<int> discarded_nodes;\r
+ set<int> candidates;\r
+ while(roots.size() >0){\r
+ for(si=roots.begin();si!=roots.end();++si){\r
+ if(qnodes[(*si)]->is_externally_visible){\r
+ valid_roots.insert((*si));\r
+ }else{\r
+ discarded_nodes.insert((*si));\r
+ set<int>::iterator sir;\r
+ for(sir=qnodes[(*si)]->reads_from.begin(); sir!=qnodes[(*si)]->reads_from.end();++sir){\r
+ qnodes[(*sir)]->n_consumers--;\r
+ if(qnodes[(*sir)]->n_consumers == 0)\r
+ candidates.insert( (*sir));\r
+ }\r
+ }\r
+ }\r
+ roots = candidates;\r
+ candidates.clear();\r
+ }\r
+ roots = valid_roots;\r
+ if(discarded_nodes.size()>0){\r
+ fprintf(stderr,"Warning, the following queries were discarded because they produce no output:\n");\r
+ int di = 0;\r
+ for(si=discarded_nodes.begin();si!=discarded_nodes.end();++si){\r
+ if(di>0 && (di%8)==0) fprintf(stderr,"\n");\r
+ di++;\r
+ fprintf(stderr," %s",qnodes[(*si)]->name.c_str());\r
+ }\r
+ fprintf(stderr,"\n");\r
+ }\r
+\r
+// Compute the sources_to set, ignoring discarded nodes.\r
+ for(i=0;i<qnodes.size();++i){\r
+ if(discarded_nodes.count(i)==0)\r
+ for(si=qnodes[i]->reads_from.begin();si!=qnodes[i]->reads_from.end();++si){\r
+ qnodes[(*si)]->sources_to.insert(i);\r
+ }\r
+ }\r
+\r
+\r
+// Find the nodes that are shared by multiple visible subtrees.\r
+// THe roots become inferred visible nodes.\r
+\r
+// Find the visible nodes.\r
+ vector<int> visible_nodes;\r
+ for(i=0;i<qnodes.size();i++){\r
+ if(qnodes[i]->is_externally_visible){\r
+ visible_nodes.push_back(i);\r
+ }\r
+ }\r
+\r
+// Find UDOPs referenced by visible nodes.\r
+ list<int> workq;\r
+ for(i=0;i<visible_nodes.size();++i){\r
+ workq.push_back(visible_nodes[i]);\r
+ }\r
+ while(!workq.empty()){\r
+ int node = workq.front();\r
+ workq.pop_front();\r
+ set<int>::iterator children;\r
+ if(qnodes[node]->is_udop && qnodes[node]->is_externally_visible == false){\r
+ qnodes[node]->is_externally_visible = true;\r
+ visible_nodes.push_back(node);\r
+ for(children=qnodes[node]->reads_from.begin();children!=qnodes[node]->reads_from.end();++children){\r
+ if(qnodes[(*children)]->is_externally_visible == false){\r
+ qnodes[(*children)]->is_externally_visible = true;\r
+ visible_nodes.push_back((*children));\r
+ }\r
+ }\r
+ }\r
+ for(children=qnodes[node]->reads_from.begin();children!=qnodes[node]->reads_from.end();++children){\r
+ workq.push_back((*children));\r
+ }\r
+ }\r
+\r
+ bool done = false;\r
+ while(!done){\r
+// reset the nodes\r
+ for(i=0;i<qnodes.size();i++){\r
+ qnodes[i]->subtree_roots.clear();\r
+ }\r
+\r
+// Walk the tree defined by a visible node, not descending into\r
+// subtrees rooted by a visible node. Mark the node visited with\r
+// the visible node ID.\r
+ for(i=0;i<visible_nodes.size();++i){\r
+ set<int> vroots;\r
+ vroots.insert(visible_nodes[i]);\r
+ while(vroots.size()>0){\r
+ for(si=vroots.begin();si!=vroots.end();++si){\r
+ qnodes[(*si)]->subtree_roots.insert(visible_nodes[i]);\r
+\r
+ set<int>::iterator sir;\r
+ for(sir=qnodes[(*si)]->reads_from.begin(); sir!=qnodes[(*si)]->reads_from.end();++sir){\r
+ if(! qnodes[(*sir)]->is_externally_visible){\r
+ candidates.insert( (*sir));\r
+ }\r
+ }\r
+ }\r
+ vroots = candidates;\r
+ candidates.clear();\r
+ }\r
+ }\r
+// Find the nodes in multiple visible node subtrees, but with no parent\r
+// that has is in multile visible node subtrees. Mark these as inferred visible nodes.\r
+ done = true; // until proven otherwise\r
+ for(i=0;i<qnodes.size();i++){\r
+ if(qnodes[i]->subtree_roots.size()>1){\r
+ bool is_new_root = true;\r
+ set<int>::iterator sir;\r
+ for(sir=qnodes[i]->sources_to.begin(); sir!=qnodes[i]->sources_to.end();++sir){\r
+ if(qnodes[(*sir)]->subtree_roots.size()>1)\r
+ is_new_root = false;\r
+ }\r
+ if(is_new_root){\r
+ qnodes[i]->is_externally_visible = true;\r
+ qnodes[i]->inferred_visible_node = true;\r
+ visible_nodes.push_back(i);\r
+ done = false;\r
+ }\r
+ }\r
+ }\r
+ }\r
+\r
+\r
+\r
+\r
+\r
+// get visible nodes in topo ordering.\r
+// for(i=0;i<qnodes.size();i++){\r
+// qnodes[i]->n_consumers = qnodes[i]->sources_to.size();\r
+// }\r
+ vector<int> process_order;\r
+ while(roots.size() >0){\r
+ for(si=roots.begin();si!=roots.end();++si){\r
+ if(discarded_nodes.count((*si))==0){\r
+ process_order.push_back( (*si) );\r
+ }\r
+ set<int>::iterator sir;\r
+ for(sir=qnodes[(*si)]->reads_from.begin(); sir!=qnodes[(*si)]->reads_from.end();++sir){\r
+ qnodes[(*sir)]->n_consumers--;\r
+ if(qnodes[(*sir)]->n_consumers == 0)\r
+ candidates.insert( (*sir));\r
+ }\r
+ }\r
+ roots = candidates;\r
+ candidates.clear();\r
+ }\r
+\r
+\r
+//printf("process_order.size() =%d\n",process_order.size());\r
+\r
+// Search for cyclic dependencies\r
+ string found_dep;\r
+ for(i=0;i<qnodes.size();++i){\r
+ if(discarded_nodes.count(i)==0 && qnodes[i]->n_consumers > 0){\r
+ if(found_dep.size() != 0) found_dep += ", ";\r
+ found_dep += "query "+qnodes[i]->name+" (file "+qnodes[i]->file+")";\r
+ }\r
+ }\r
+ if(found_dep.size()>0){\r
+ fprintf(stderr,"ERROR, the following queries contain a cyclic reads-from dependency:\n%s\n",found_dep.c_str());\r
+ exit(1);\r
+ }\r
+\r
+// Get a list of query sets, in the order to be processed.\r
+// Start at visible root and do bfs.\r
+// The query set includes queries referenced indirectly,\r
+// as sources for user-defined operators. These are needed\r
+// to ensure that they are added to the schema, but are not part\r
+// of the query tree.\r
+\r
+// stream_node_sets contains queries reachable only through the\r
+// FROM clause, so I can tell which queries to add to the stream\r
+// query. (DISABLED, UDOPS are integrated, does this cause problems?)\r
+\r
+// NOTE: this code works because in order for data to be\r
+// read by multiple hftas, the node must be externally visible.\r
+// But visible nodes define roots of process sets.\r
+// internally visible nodes can feed data only\r
+// to other nodes in the same query file.\r
+// Therefore, any access can be restricted to a file,\r
+// hfta output sharing is done only on roots\r
+// never on interior nodes.\r
+\r
+\r
+\r
+\r
+// Conpute the base collection of hftas.\r
+ vector<hfta_node *> hfta_sets;\r
+ map<string, int> hfta_name_map;\r
+// vector< vector<int> > process_sets;\r
+// vector< set<int> > stream_node_sets;\r
+ reverse(process_order.begin(), process_order.end()); // get listing in reverse order.\r
+ // i.e. process leaves 1st.\r
+ for(i=0;i<process_order.size();++i){\r
+ if(qnodes[process_order[i]]->is_externally_visible == true){\r
+//printf("Visible.\n");\r
+ int root = process_order[i];\r
+ hfta_node *hnode = new hfta_node();\r
+ hnode->name = qnodes[root]-> name;\r
+ hnode->source_name = qnodes[root]-> name;\r
+ hnode->is_udop = qnodes[root]->is_udop;\r
+ hnode->inferred_visible_node = qnodes[root]->inferred_visible_node;\r
+\r
+ vector<int> proc_list; proc_list.push_back(root);\r
+// Ensure that nodes are added only once.\r
+ set<int> proc_set; proc_set.insert(root);\r
+ roots.clear(); roots.insert(root);\r
+ candidates.clear();\r
+ while(roots.size()>0){\r
+ for(si=roots.begin();si!=roots.end();++si){\r
+//printf("Processing root %d\n",(*si));\r
+ set<int>::iterator sir;\r
+ for(sir=qnodes[(*si)]->reads_from.begin(); sir!=qnodes[(*si)]->reads_from.end();++sir){\r
+//printf("reads fom %d\n",(*sir));\r
+ if(qnodes[(*sir)]->is_externally_visible==false){\r
+ candidates.insert( (*sir) );\r
+ if(proc_set.count( (*sir) )==0){\r
+ proc_set.insert( (*sir) );\r
+ proc_list.push_back( (*sir) );\r
+ }\r
+ }\r
+ }\r
+ }\r
+ roots = candidates;\r
+ candidates.clear();\r
+ }\r
+\r
+ reverse(proc_list.begin(), proc_list.end());\r
+ hnode->query_node_indices = proc_list;\r
+ hfta_name_map[hnode->name] = hfta_sets.size();\r
+ hfta_sets.push_back(hnode);\r
+ }\r
+ }\r
+\r
+// Compute the reads_from / sources_to graphs for the hftas.\r
+\r
+ for(i=0;i<hfta_sets.size();++i){\r
+ hfta_node *hnode = hfta_sets[i];\r
+ for(q=0;q<hnode->query_node_indices.size();q++){\r
+ query_node *qnode = qnodes[hnode->query_node_indices[q]];\r
+ for(s=0;s<qnode->refd_tbls.size();++s){\r
+ if(hfta_name_map.count(qnode->refd_tbls[s])){\r
+ int other_hfta = hfta_name_map[qnode->refd_tbls[s]];\r
+ hnode->reads_from.insert(other_hfta);\r
+ hfta_sets[other_hfta]->sources_to.insert(i);\r
+ }\r
+ }\r
+ }\r
+ }\r
+\r
+// Compute a topological sort of the hfta_sets.\r
+\r
+ vector<int> hfta_topsort;\r
+ workq.clear();\r
+ int hnode_srcs[hfta_sets.size()];\r
+ for(i=0;i<hfta_sets.size();++i){\r
+ hnode_srcs[i] = 0;\r
+ if(hfta_sets[i]->sources_to.size() == 0)\r
+ workq.push_back(i);\r
+ }\r
+\r
+ while(! workq.empty()){\r
+ int node = workq.front();\r
+ workq.pop_front();\r
+ hfta_topsort.push_back(node);\r
+ set<int>::iterator stsi;\r
+ for(stsi=hfta_sets[node]->reads_from.begin();stsi!=hfta_sets[node]->reads_from.end();++stsi){\r
+ int parent = (*stsi);\r
+ hnode_srcs[parent]++;\r
+ if(hnode_srcs[parent] == hfta_sets[parent]->sources_to.size()){\r
+ workq.push_back(parent);\r
+ }\r
+ }\r
+ }\r
+\r
+// Decorate hfta nodes with the level of parallelism given as input.\r
+\r
+ map<string, int>::iterator msii;\r
+ for(msii=hfta_parallelism.begin();msii!=hfta_parallelism.end();++msii){\r
+ string hfta_name = (*msii).first;\r
+ int par = (*msii).second;\r
+ if(hfta_name_map.count(hfta_name) > 0){\r
+ hfta_sets[ hfta_name_map[hfta_name] ]->n_parallel = par;\r
+ }else{\r
+ fprintf(stderr,"Warning, hfta_parallelism.cfg has an entry for %s, but its not a hfta.\n",hfta_name.c_str());\r
+ }\r
+ }\r
+\r
+// Propagate levels of parallelism: children should have a level of parallelism\r
+// as large as any of its parents. Adjust children upwards to compensate.\r
+// Start at parents and adjust children, auto-propagation will occur.\r
+\r
+ for(i=hfta_sets.size()-1;i>=0;i--){\r
+ set<int>::iterator stsi;\r
+ for(stsi=hfta_sets[i]->reads_from.begin();stsi!=hfta_sets[i]->reads_from.end();++stsi){\r
+ if(hfta_sets[i]->n_parallel > hfta_sets[ (*stsi) ]->n_parallel){\r
+ hfta_sets[ (*stsi) ]->n_parallel = hfta_sets[i]->n_parallel;\r
+ }\r
+ }\r
+ }\r
+\r
+// Before all the name mangling, check if therey are any output_spec.cfg\r
+// or hfta_parallelism.cfg entries that do not have a matching query.\r
+\r
+ string dangling_ospecs = "";\r
+ for(msii=qname_to_ospec.begin();msii!=qname_to_ospec.end();++msii){\r
+ string oq = (*msii).first;\r
+ if(hfta_name_map.count(oq) == 0){\r
+ dangling_ospecs += " "+(*msii).first;\r
+ }\r
+ }\r
+ if(dangling_ospecs!=""){\r
+ fprintf(stderr,"WARNING, the following entries in output_spec.cfg don't have a matching query: %s\n",dangling_ospecs.c_str());\r
+ }\r
+\r
+ string dangling_par = "";\r
+ for(msii=hfta_parallelism.begin();msii!=hfta_parallelism.end();++msii){\r
+ string oq = (*msii).first;\r
+ if(hfta_name_map.count(oq) == 0){\r
+ dangling_par += " "+(*msii).first;\r
+ }\r
+ }\r
+ if(dangling_par!=""){\r
+ fprintf(stderr,"WARNING, the following entries in hfta_parallelism.cfg don't have a matching query: %s\n",dangling_par.c_str());\r
+ }\r
+\r
+\r
+\r
+// Replicate parallelized hftas. Do __copyX name mangling. Adjust\r
+// FROM clauses: retarget any name which is an internal node, and\r
+// any which is in hfta_sets (and which is parallelized). Add Merge nodes\r
+// when the source hfta has more parallelism than the target node.\r
+// Add new nodes to qnodes and hfta_nodes wth the appropriate linkages.\r
+\r
+\r
+ int n_original_hfta_sets = hfta_sets.size();\r
+ for(i=0;i<n_original_hfta_sets;++i){\r
+ if(hfta_sets[i]->n_parallel > 1){\r
+ hfta_sets[i]->do_generation =false; // set the deletion flag for this entry.\r
+ set<string> local_nodes; // names of query nodes in the hfta.\r
+ for(h=0;h<hfta_sets[i]->query_node_indices.size();++h){\r
+ local_nodes.insert(qnodes[ hfta_sets[i]->query_node_indices[h] ]->name);\r
+ }\r
+\r
+ for(p=0;p<hfta_sets[i]->n_parallel;++p){\r
+ string mangler = "__copy"+int_to_string(p);\r
+ hfta_node *par_hfta = new hfta_node();\r
+ par_hfta->name = hfta_sets[i]->name + mangler;\r
+ par_hfta->source_name = hfta_sets[i]->name;\r
+ par_hfta->is_udop = hfta_sets[i]->is_udop;\r
+ par_hfta->inferred_visible_node = hfta_sets[i]->inferred_visible_node;\r
+ par_hfta->n_parallel = hfta_sets[i]->n_parallel;\r
+ par_hfta->parallel_idx = p;\r
+\r
+ map<string, int> par_qnode_map; // qnode name-to-idx, aids dependency tracking.\r
+\r
+// Is it a UDOP?\r
+ if(hfta_sets[i]->is_udop){\r
+ int root = hfta_sets[i]->query_node_indices[0];\r
+\r
+ string unequal_par_sources;\r
+ set<int>::iterator rfsii;\r
+ for(rfsii=hfta_sets[i]->reads_from.begin();rfsii!=hfta_sets[i]->reads_from.end();++rfsii){\r
+ if(hfta_sets[(*rfsii)]->n_parallel != par_hfta->n_parallel){\r
+ unequal_par_sources = hfta_sets[(*rfsii)]->name+" ("+int_to_string(hfta_sets[(*rfsii)]->n_parallel)+") ";\r
+ }\r
+ }\r
+ if(unequal_par_sources != ""){\r
+ fprintf(stderr,"ERROR, UDOP %s has parallelism %d, but some of its sources have a different parallelism: %s\n",hfta_sets[i]->name.c_str(), hfta_sets[i]->n_parallel, unequal_par_sources.c_str());\r
+ exit(1);\r
+ }\r
+\r
+ int rti;\r
+ vector<string> new_sources;\r
+ for(rti=0;rti<qnodes[root]->refd_tbls.size();++rti){\r
+ new_sources.push_back(qnodes[root]->refd_tbls[rti]+mangler);\r
+ }\r
+\r
+ query_node *new_qn = new query_node(qnodes.size(), qnodes[root]->name, Schema);\r
+ new_qn->name += mangler;\r
+ new_qn->mangler = mangler;\r
+ new_qn->refd_tbls = new_sources;\r
+ par_hfta->query_node_indices.push_back(qnodes.size());\r
+ par_qnode_map[new_qn->name] = qnodes.size();\r
+ name_node_map[ new_qn->name ] = qnodes.size();\r
+ qnodes.push_back(new_qn);\r
+ }else{\r
+// regular query node\r
+ for(h=0;h<hfta_sets[i]->query_node_indices.size();++h){\r
+ int hqn_idx = hfta_sets[i]->query_node_indices[h];\r
+ table_exp_t *dup_pt = dup_table_exp(qnodes[hqn_idx]->parse_tree);\r
+// rehome the from clause on mangled names.\r
+// create merge nodes as needed for external sources.\r
+ for(f=0;f<dup_pt->fm->tlist.size();++f){\r
+ if(local_nodes.count(dup_pt->fm->tlist[f]->schema_name)>0){\r
+ dup_pt->fm->tlist[f]->schema_name += mangler;\r
+ }else if(hfta_name_map.count(dup_pt->fm->tlist[f]->schema_name)>0){\r
+// Ref's an external HFTA. No parallelism => leave as is. Else, if level of parallelism of the two hftas is the same, rename by the mangler. Else, there mnust be more sources, so create a merge node.\r
+ int other_hidx = hfta_name_map[dup_pt->fm->tlist[f]->schema_name];\r
+ if(par_hfta->n_parallel == hfta_sets[other_hidx]->n_parallel){\r
+ dup_pt->fm->tlist[f]->schema_name += mangler;\r
+ }else{\r
+ vector<string> src_tbls;\r
+ int stride = hfta_sets[other_hidx]->n_parallel / par_hfta->n_parallel;\r
+ if(stride == 0){\r
+ fprintf(stderr,"INTERNAL ERROR, parent hfta %s has a larger parallelism than its child %s\n",par_hfta->name.c_str(), hfta_sets[other_hidx]->name.c_str());\r
+ exit(1);\r
+ }\r
+ for(s=0;s<stride;++s){\r
+ string ext_src_name = hfta_sets[other_hidx]->name+"__copy"+int_to_string(s+stride*p);\r
+ src_tbls.push_back(ext_src_name);\r
+ }\r
+ table_exp_t *merge_pt = table_exp_t::make_deferred_merge(src_tbls);\r
+ string merge_node_name = qnodes[hqn_idx]->name+"__copy"+int_to_string(p)+"__merge"+int_to_string(f);\r
+ dup_pt->fm->tlist[f]->schema_name = merge_node_name;\r
+// Make a qnode to represent the new merge node\r
+ query_node *qn_pt = new query_node(qnodes.size(),merge_node_name, qnodes[hqn_idx]->file,merge_pt);\r
+ qn_pt->refd_tbls = src_tbls;\r
+ qn_pt->is_udop = false;\r
+ qn_pt->is_externally_visible = false;\r
+ qn_pt->inferred_visible_node = false;\r
+ par_hfta->query_node_indices.push_back(qnodes.size());\r
+ par_qnode_map[merge_node_name] = qnodes.size();\r
+ name_node_map[ merge_node_name ] = qnodes.size();\r
+ qnodes.push_back(qn_pt);\r
+ }\r
+ }\r
+ }\r
+ query_node *new_qn = new query_node(qnodes.size(),qnodes[hqn_idx]->name+"__copy"+int_to_string(p),qnodes[hqn_idx]->file,dup_pt);\r
+ for(f=0;f<dup_pt->fm->tlist.size();++f){\r
+ new_qn->refd_tbls.push_back(dup_pt->fm->tlist[f]->schema_name);\r
+ }\r
+ new_qn->params = qnodes[hqn_idx]->params;\r
+ new_qn->is_udop = false;\r
+ new_qn->is_externally_visible = qnodes[hqn_idx]->is_externally_visible;\r
+ new_qn->inferred_visible_node = qnodes[hqn_idx]->inferred_visible_node;\r
+ par_hfta->query_node_indices.insert(par_hfta->query_node_indices.begin(),qnodes.size());\r
+ par_qnode_map[new_qn->name] = qnodes.size();\r
+ name_node_map[ new_qn->name ] = qnodes.size();\r
+ qnodes.push_back(new_qn);\r
+ }\r
+ }\r
+ hfta_name_map[par_hfta->name] = hfta_sets.size();\r
+ hfta_sets.push_back(par_hfta);\r
+ }\r
+ }else{\r
+// This hfta isn't being parallelized, but add merge nodes for any parallelized\r
+// hfta sources.\r
+ if(!hfta_sets[i]->is_udop){\r
+ for(h=0;h<hfta_sets[i]->query_node_indices.size();++h){\r
+ int hqn_idx = hfta_sets[i]->query_node_indices[h];\r
+ for(f=0;f<qnodes[hqn_idx]->parse_tree->fm->tlist.size();++f){\r
+ if(hfta_name_map.count(qnodes[hqn_idx]->parse_tree->fm->tlist[f]->schema_name)>0){\r
+// Ref's an external HFTA. No parallelism => leave as is. Else, if level of parallelism of the two hftas is the same, rename by the mangler. Else, there mnust be more sources, so create a merge node.\r
+ int other_hidx = hfta_name_map[qnodes[hqn_idx]->parse_tree->fm->tlist[f]->schema_name];\r
+ if(hfta_sets[i]->n_parallel != hfta_sets[other_hidx]->n_parallel){\r
+ vector<string> src_tbls;\r
+ for(s=0;s<hfta_sets[other_hidx]->n_parallel;++s){\r
+ string ext_src_name = hfta_sets[other_hidx]->name+"__copy"+int_to_string(s);\r
+ src_tbls.push_back(ext_src_name);\r
+ }\r
+ table_exp_t *merge_pt = table_exp_t::make_deferred_merge(src_tbls);\r
+ string merge_node_name = qnodes[hqn_idx]->name+"__merge"+int_to_string(f);\r
+ qnodes[hqn_idx]->parse_tree->fm->tlist[f]->schema_name = merge_node_name;\r
+// Make a qnode to represent the new merge node\r
+ query_node *qn_pt = new query_node(qnodes.size(),merge_node_name, qnodes[hqn_idx]->file,merge_pt);\r
+ qn_pt->refd_tbls = src_tbls;\r
+ qn_pt->is_udop = false;\r
+ qn_pt->is_externally_visible = false;\r
+ qn_pt->inferred_visible_node = false;\r
+ hfta_sets[i]->query_node_indices.insert(hfta_sets[i]->query_node_indices.begin(),qnodes.size());\r
+ name_node_map[ merge_node_name ] = qnodes.size();\r
+ qnodes.push_back(qn_pt);\r
+ }\r
+ }\r
+ }\r
+ }\r
+ }\r
+ }\r
+ }\r
+\r
+// Rebuild the reads_from / sources_to lists in the qnodes\r
+ for(q=0;q<qnodes.size();++q){\r
+ qnodes[q]->reads_from.clear();\r
+ qnodes[q]->sources_to.clear();\r
+ }\r
+ for(q=0;q<qnodes.size();++q){\r
+ for(s=0;s<qnodes[q]->refd_tbls.size();++s){\r
+ if(name_node_map.count(qnodes[q]->refd_tbls[s])>0){\r
+ int rf = name_node_map[qnodes[q]->refd_tbls[s]];\r
+ qnodes[q]->reads_from.insert(rf);\r
+ qnodes[rf]->sources_to.insert(q);\r
+ }\r
+ }\r
+ }\r
+\r
+// Rebuild the reads_from / sources_to lists in hfta_sets\r
+ for(q=0;q<hfta_sets.size();++q){\r
+ hfta_sets[q]->reads_from.clear();\r
+ hfta_sets[q]->sources_to.clear();\r
+ }\r
+ for(q=0;q<hfta_sets.size();++q){\r
+ for(s=0;s<hfta_sets[q]->query_node_indices.size();++s){\r
+ int node = hfta_sets[q]->query_node_indices[s];\r
+ set<int>::iterator rfsii;\r
+ for(rfsii=qnodes[node]->reads_from.begin();rfsii!=qnodes[node]->reads_from.end();++rfsii){\r
+ if(hfta_name_map.count(qnodes[(*rfsii)]->name)>0){\r
+ hfta_sets[q]->reads_from.insert(hfta_name_map[qnodes[(*rfsii)]->name]);\r
+ hfta_sets[hfta_name_map[qnodes[(*rfsii)]->name]]->sources_to.insert(q);\r
+ }\r
+ }\r
+ }\r
+ }\r
+\r
+/*\r
+for(q=0;q<qnodes.size();++q){\r
+ printf("qnode %d reads-from %d:",q,qnodes[q]->reads_from.size());\r
+ set<int>::iterator rsii;\r
+ for(rsii=qnodes[q]->reads_from.begin();rsii!=qnodes[q]->reads_from.end();++rsii)\r
+ printf(" %d",(*rsii));\r
+ printf(", and sources-to %d:",qnodes[q]->sources_to.size());\r
+ for(rsii=qnodes[q]->sources_to.begin();rsii!=qnodes[q]->sources_to.end();++rsii)\r
+ printf(" %d",(*rsii));\r
+ printf("\n");\r
+}\r
+\r
+for(q=0;q<hfta_sets.size();++q){\r
+ if(hfta_sets[q]->do_generation==false)\r
+ continue;\r
+ printf("hfta %d (%s) reads-from %d:",q,hfta_sets[q]->name.c_str(),hfta_sets[q]->reads_from.size());\r
+ set<int>::iterator rsii;\r
+ for(rsii=hfta_sets[q]->reads_from.begin();rsii!=hfta_sets[q]->reads_from.end();++rsii)\r
+ printf(" %d",(*rsii));\r
+ printf(", and sources-to %d:",hfta_sets[q]->sources_to.size());\r
+ for(rsii=hfta_sets[q]->sources_to.begin();rsii!=hfta_sets[q]->sources_to.end();++rsii)\r
+ printf(" %d",(*rsii));\r
+ printf("\n");\r
+}\r
+*/\r
+\r
+\r
+\r
+// Re-topo sort the hftas\r
+ hfta_topsort.clear();\r
+ workq.clear();\r
+ int hnode_srcs_2[hfta_sets.size()];\r
+ for(i=0;i<hfta_sets.size();++i){\r
+ hnode_srcs_2[i] = 0;\r
+ if(hfta_sets[i]->sources_to.size() == 0 && hfta_sets[i]->do_generation){\r
+ workq.push_back(i);\r
+ }\r
+ }\r
+\r
+ while(workq.empty() == false){\r
+ int node = workq.front();\r
+ workq.pop_front();\r
+ hfta_topsort.push_back(node);\r
+ set<int>::iterator stsii;\r
+ for(stsii=hfta_sets[node]->reads_from.begin();stsii!=hfta_sets[node]->reads_from.end();++stsii){\r
+ int child = (*stsii);\r
+ hnode_srcs_2[child]++;\r
+ if(hnode_srcs_2[child] == hfta_sets[child]->sources_to.size()){\r
+ workq.push_back(child);\r
+ }\r
+ }\r
+ }\r
+\r
+// Ensure that all of the query_node_indices in hfta_sets are topologically\r
+// sorted, don't rely on assumptions that all transforms maintain some kind of order.\r
+ for(i=0;i<hfta_sets.size();++i){\r
+ if(hfta_sets[i]->do_generation){\r
+ map<int,int> n_accounted;\r
+ vector<int> new_order;\r
+ workq.clear();\r
+ vector<int>::iterator vii;\r
+ for(vii=hfta_sets[i]->query_node_indices.begin();vii!=hfta_sets[i]->query_node_indices.end();++vii){\r
+ n_accounted[(*vii)]= 0;\r
+ }\r
+ for(vii=hfta_sets[i]->query_node_indices.begin();vii!=hfta_sets[i]->query_node_indices.end();++vii){\r
+ set<int>::iterator rfsii;\r
+ for(rfsii=qnodes[(*vii)]->reads_from.begin();rfsii!=qnodes[(*vii)]->reads_from.end();++rfsii){\r
+ if(n_accounted.count((*rfsii)) == 0){\r
+ n_accounted[(*vii)]++;\r
+ }\r
+ }\r
+ if(n_accounted[(*vii)] == qnodes[(*vii)]->reads_from.size()){\r
+ workq.push_back((*vii));\r
+ }\r
+ }\r
+\r
+ while(workq.empty() == false){\r
+ int node = workq.front();\r
+ workq.pop_front();\r
+ new_order.push_back(node);\r
+ set<int>::iterator stsii;\r
+ for(stsii=qnodes[node]->sources_to.begin();stsii!=qnodes[node]->sources_to.end();++stsii){\r
+ if(n_accounted.count((*stsii))){\r
+ n_accounted[(*stsii)]++;\r
+ if(n_accounted[(*stsii)] == qnodes[(*stsii)]->reads_from.size()){\r
+ workq.push_back((*stsii));\r
+ }\r
+ }\r
+ }\r
+ }\r
+ hfta_sets[i]->query_node_indices = new_order;\r
+ }\r
+ }\r
+\r
+\r
+\r
+\r
+\r
+/// Global checkng is done, start the analysis and translation\r
+/// of the query parse tree in the order specified by process_order\r
+\r
+\r
+// Get a list of the LFTAs for global lfta optimization\r
+// TODO: separate building operators from spliting lftas,\r
+// that will make optimizations such as predicate pushing easier.\r
+ vector<stream_query *> lfta_list;\r
+\r
+ stream_query *rootq;\r
+\r
+ int qi,qj;\r
+\r
+ for(qi=hfta_topsort.size()-1;qi>=0;--qi){\r
+\r
+ int hfta_id = hfta_topsort[qi];\r
+ vector<int> curr_list = hfta_sets[hfta_id]->query_node_indices;\r
+\r
+\r
+\r
+// Two possibilities, either its a UDOP, or its a collection of queries.\r
+// if(qnodes[curr_list.back()]->is_udop)\r
+ if(hfta_sets[hfta_id]->is_udop){\r
+ int node_id = curr_list.back();\r
+ int udop_schref = Schema->find_tbl(qnodes[node_id]->file);\r
+ opview_entry *opv = new opview_entry();\r
+\r
+// Many of the UDOP properties aren't currently used.\r
+ opv->parent_qname = "no_parent";\r
+ opv->root_name = qnodes[node_id]->name;\r
+ opv->view_name = qnodes[node_id]->file;\r
+ opv->pos = qi;\r
+ sprintf(tmpstr,"%s_UDOP%d_%s",qnodes[node_id]->name.c_str(),qi,opv->view_name.c_str());\r
+ opv->udop_alias = tmpstr;\r
+ opv->mangler = qnodes[node_id]->mangler;\r
+\r
+ if(opv->mangler != ""){\r
+ int new_udop_schref = Schema->add_duplicate_table(opv->view_name,opv->root_name);\r
+ Schema->mangle_subq_names(new_udop_schref,opv->mangler);\r
+ }\r
+\r
+// This piece of code makes each hfta which referes to the same udop\r
+// reference a distinct running udop. Do this at query optimization time?\r
+// fmtbl->set_udop_alias(opv->udop_alias);\r
+\r
+ opv->exec_fl = Schema->get_op_prop(udop_schref, string("file"));\r
+ opv->liveness_timeout = atoi(Schema->get_op_prop(udop_schref, string("liveness_timeout")).c_str());\r
+\r
+ vector<subquery_spec *> subq = Schema->get_subqueryspecs(udop_schref);\r
+ int s,f,q;\r
+ for(s=0;s<subq.size();++s){\r
+// Validate that the fields match.\r
+ subquery_spec *sqs = subq[s];\r
+ string subq_name = sqs->name + opv->mangler;\r
+ vector<field_entry *> flds = Schema->get_fields(subq_name);\r
+ if(flds.size() == 0){\r
+ fprintf(stderr,"INTERNAL ERROR: the subquery %s of view %s not found in Schema.\n",subq_name.c_str(), opv->view_name.c_str());\r
+ return(1);\r
+ }\r
+ if(flds.size() < sqs->types.size()){\r
+ fprintf(stderr,"ERROR: subquery %s of view %s does not have enough fields (%lu found, %lu expected).\n",subq_name.c_str(), opv->view_name.c_str(),flds.size(), sqs->types.size());\r
+ return(1);\r
+ }\r
+ bool failed = false;\r
+ for(f=0;f<sqs->types.size();++f){\r
+ data_type dte(sqs->types[f],sqs->modifiers[f]);\r
+ data_type dtf(flds[f]->get_type(),flds[f]->get_modifier_list());\r
+ if(! dte.subsumes_type(&dtf) ){\r
+ fprintf(stderr,"ERROR: subquery %s of view %s does not have the correct type for field %d (%s found, %s expected).\n",subq_name.c_str(), opv->view_name.c_str(),f,dtf.to_string().c_str(), dte.to_string().c_str());\r
+ failed = true;\r
+ }\r
+/*\r
+ if(dte.is_temporal() && (dte.get_temporal() != dtf.get_temporal()) ){\r
+ string pstr = dte.get_temporal_string();\r
+ fprintf(stderr,"ERROR: subquery %s of view %s does not have the expected temporal value %s of field %d.\n",sqs->name.c_str(), opv->view_name.c_str(),pstr.c_str(),f);\r
+ failed = true;\r
+ }\r
+*/\r
+ }\r
+ if(failed)\r
+ return(1);\r
+/// Validation done, find the subquery, make a copy of the\r
+/// parse tree, and add it to the return list.\r
+ for(q=0;q<qnodes.size();++q)\r
+ if(qnodes[q]->name == subq_name)\r
+ break;\r
+ if(q==qnodes.size()){\r
+ fprintf(stderr,"INTERNAL ERROR: subquery %s of view %s not found in list of query names.\n",subq_name.c_str(), opv->view_name.c_str());\r
+ return(1);\r
+ }\r
+\r
+ }\r
+\r
+// Cross-link to from entry(s) in all sourced-to tables.\r
+ set<int>::iterator sii;\r
+ for(sii=qnodes[curr_list.back()]->sources_to.begin() ;sii!=qnodes[curr_list.back()]->sources_to.end();++sii){\r
+//printf("\tUDOP %s sources_to %d (%s)\n",hfta_sets[hfta_id]->name.c_str(),(*sii),hfta_sets[(*sii)]->name.c_str());\r
+ vector<tablevar_t *> tblvars = qnodes[(*sii)]->parse_tree->get_from()->get_table_list();\r
+ int ii;\r
+ for(ii=0;ii<tblvars.size();++ii){\r
+ if(tblvars[ii]->schema_name == opv->root_name){\r
+ tblvars[ii]->set_opview_idx(opviews.size());\r
+ }\r
+\r
+ }\r
+ }\r
+\r
+ opviews.append(opv);\r
+ }else{\r
+\r
+// Analyze the parse trees in this query,\r
+// put them in rootq\r
+// vector<int> curr_list = process_sets[qi];\r
+\r
+\r
+////////////////////////////////////////\r
+\r
+ rootq = NULL;\r
+//printf("Process set %d, has %d queries\n",qi,curr_list.size());\r
+ for(qj=0;qj<curr_list.size();++qj){\r
+ i = curr_list[qj];\r
+ fprintf(stderr,"Processing query %s (file %s) is_udop = %d\n",qnodes[i]->name.c_str(), qnodes[i]->file.c_str(),qnodes[i]->is_udop);\r
+\r
+// Select the current query parse tree\r
+ table_exp_t *fta_parse_tree = qnodes[i]->parse_tree;\r
+\r
+// if hfta only, try to fetch any missing schemas\r
+// from the registry (using the print_schema program).\r
+// Here I use a hack to avoid analyzing the query -- all referenced\r
+// tables must be in the from clause\r
+// If there is a problem loading any table, just issue a warning,\r
+//\r
+ tablevar_list_t *fm = fta_parse_tree->get_from();\r
+ vector<string> refd_tbls = fm->get_src_tbls(Schema);\r
+// iterate over all referenced tables\r
+ int t;\r
+ for(t=0;t<refd_tbls.size();++t){\r
+ int tbl_ref = Schema->get_table_ref(refd_tbls[t]);\r
+\r
+ if(tbl_ref < 0){ // if this table is not in the Schema\r
+\r
+ if(hfta_only){\r
+ string cmd="print_schema "+refd_tbls[t];\r
+ FILE *schema_in = popen(cmd.c_str(), "r");\r
+ if(schema_in == NULL){\r
+ fprintf(stderr,"ERROR: cannot execute command %s\n",cmd.c_str());\r
+ }else{\r
+ string schema_instr;\r
+ while(fgets(tmpstr,TMPSTRLEN,schema_in)){\r
+ schema_instr += tmpstr;\r
+ }\r
+ fta_parse_result = new fta_parse_t();\r
+ strcpy(tmp_schema_str,schema_instr.c_str());\r
+ FtaParser_setstringinput(tmp_schema_str);\r
+ if(FtaParserparse()){\r
+ fprintf(stderr,"FTA parse failed on result from command %s.\n",cmd.c_str());\r
+ }else{\r
+ if( fta_parse_result->tables != NULL){\r
+ int tl;\r
+ for(tl=0;tl<fta_parse_result->tables->size();++tl){\r
+ Schema->add_table(fta_parse_result->tables->get_table(tl));\r
+ }\r
+ }else{\r
+ fprintf(stderr,"ERROR command %s returned no tables.\n",cmd.c_str());\r
+ }\r
+ }\r
+ }\r
+ }else{\r
+ fprintf(stderr,"ERROR, query %s (file %s) references stream %s, which is neither a PROTOCOL nor an externally visible query stream.\n", qnodes[i]->name.c_str(), qnodes[i]->file.c_str(), refd_tbls[t].c_str());\r
+ exit(1);\r
+ }\r
+\r
+ }\r
+ }\r
+\r
+\r
+// Analyze the query.\r
+ query_summary_class *qs = analyze_fta(fta_parse_tree, Schema, Ext_fcns, qnodes[i]->name);\r
+ if(qs == NULL){\r
+ fprintf(stderr,"analyze_fta failed on query %s (file %s) \n",qnodes[i]->name.c_str(),qnodes[i]->file.c_str());\r
+ exit(1);\r
+ }\r
+\r
+ stream_query new_sq(qs, Schema);\r
+ if(new_sq.error_code){\r
+ fprintf(stderr,"ERROR, can't create query node for query %s (file %s):\n%s\n",qnodes[i]->name.c_str(),qnodes[i]->file.c_str(),new_sq.err_str.c_str());\r
+ exit(1);\r
+ }\r
+\r
+// Add it to the Schema\r
+ table_def *output_td = new_sq.get_output_tabledef();\r
+ Schema->add_table(output_td);\r
+\r
+// Create a query plan from the analyzed parse tree.\r
+// If its a query referneced via FROM, add it to the stream query.\r
+ if(rootq){\r
+ rootq->add_query(new_sq);\r
+ }else{\r
+ rootq = new stream_query(new_sq);\r
+// have the stream query object inherit properties form the analyzed\r
+// hfta_node object.\r
+ rootq->set_nparallel(hfta_sets[hfta_id]->n_parallel,hfta_sets[hfta_id]->parallel_idx);\r
+ rootq->n_successors = hfta_sets[hfta_id]->sources_to.size();\r
+ }\r
+\r
+\r
+ }\r
+\r
+// This stream query has all its parts\r
+// Build and optimize it.\r
+//printf("translate_fta: generating plan.\n");\r
+ if(rootq->generate_plan(Schema)){\r
+ fprintf(stderr,"INTERNAL ERROR: Can't generate query plan for query %s.\n",rootq->query_name.c_str());\r
+ continue;\r
+ }\r
+\r
+// If we've found the query plan head, so now add the output operators\r
+ if(qname_to_ospec.count(hfta_sets[hfta_id]->source_name)){\r
+ pair< multimap<string, int>::iterator, multimap<string, int>::iterator > oset;\r
+ multimap<string, int>::iterator mmsi;\r
+ oset = qname_to_ospec.equal_range(hfta_sets[hfta_id]->source_name);\r
+ for(mmsi=oset.first; mmsi!=oset.second; ++mmsi){\r
+ rootq->add_output_operator(output_specs[(*mmsi).second]);\r
+ }\r
+ }\r
+\r
+\r
+\r
+// Perform query splitting if necessary.\r
+ bool hfta_returned;\r
+ vector<stream_query *> split_queries = rootq->split_query(Ext_fcns, Schema, hfta_returned, ifaces_db, n_virtual_interfaces, hfta_sets[hfta_id]->n_parallel, hfta_sets[hfta_id]->parallel_idx);\r
+\r
+ int l;\r
+//for(l=0;l<split_queries.size();++l){\r
+//printf("split query %d is %s\n",l,split_queries[l]->q`uery_name.c_str());\r
+//}\r
+\r
+\r
+\r
+\r
+ if(split_queries.size() > 0){ // should be at least one component.\r
+\r
+// Compute the number of LFTAs.\r
+ int n_lfta = split_queries.size();\r
+ if(hfta_returned) n_lfta--;\r
+\r
+\r
+// Process the LFTA components.\r
+ for(l=0;l<n_lfta;++l){\r
+ if(lfta_names.count(split_queries[l]->query_name) == 0){\r
+// Grab the lfta for global optimization.\r
+ vector<tablevar_t *> tvec = split_queries[l]->query_plan[0]->get_input_tbls();\r
+ string liface = tvec[0]->get_interface();\r
+ string lmach = tvec[0]->get_machine();\r
+ if (lmach == "")\r
+ lmach = hostname;\r
+ interface_names.push_back(liface);\r
+ machine_names.push_back(lmach);\r
+//printf("Machine is %s\n",lmach.c_str());\r
+\r
+// Set the ht size from the recommendation, if there is one in the rec file\r
+ if(lfta_htsize.count(split_queries[l]->query_name)>0){\r
+ split_queries[l]->query_plan[0]->set_definition("aggregate_slots",int_to_string(lfta_htsize[split_queries[l]->query_name]));\r
+ }\r
+\r
+\r
+ lfta_names[split_queries[l]->query_name] = lfta_list.size();\r
+ split_queries[l]->set_gid(lfta_list.size()); // set lfta global id\r
+ lfta_list.push_back(split_queries[l]);\r
+ lfta_mach_lists[lmach].push_back(split_queries[l]);\r
+\r
+// THe following is a hack,\r
+// as I should be generating LFTA code through\r
+// the stream_query object.\r
+ split_queries[l]->query_plan[0]->bind_to_schema(Schema);\r
+// split_queries[l]->query_plan[0]->definitions = split_queries[l]->defines;\r
+\r
+/*\r
+// Create query description to embed in lfta.c\r
+ string lfta_schema_str = split_queries[l]->make_schema();\r
+ string lfta_schema_embed = make_C_embedded_string(lfta_schema_str);\r
+\r
+// get NIC capabilities.\r
+ int erri;\r
+ nic_property *nicprop = NULL;\r
+ vector<string> iface_codegen_type = ifaces_db->get_iface_vals(lmach,liface,"iface_codegen_type",erri,err_str);\r
+ if(iface_codegen_type.size()){\r
+ nicprop = npdb->get_nic_property(iface_codegen_type[0],erri);\r
+ if(!nicprop){\r
+ fprintf(stderr,"ERROR cannot load nic properties %s for interface %s of machine %s for lfta codegen\n",iface_codegen_type[0].c_str(), liface.c_str(), lmach.c_str());\r
+ exit(1);\r
+ }\r
+ }\r
+\r
+ lfta_val[lmach] += generate_lfta_block(split_queries[l]->query_plan[0], Schema, split_queries[l]->get_gid(), Ext_fcns, lfta_schema_embed, ifaces_db,nicprop);\r
+*/\r
+\r
+ snap_lengths.push_back(compute_snap_len(split_queries[l]->query_plan[0], Schema));\r
+ query_names.push_back(split_queries[l]->query_name);\r
+ mach_query_names[lmach].push_back(query_names.size()-1);\r
+// NOTE: I will assume a 1-1 correspondance between\r
+// mach_query_names[lmach] and lfta_mach_lists[lmach]\r
+// where mach_query_names[lmach][i] contains the index into\r
+// query_names, which names the lfta, and\r
+// mach_query_names[lmach][i] is the stream_query * of the\r
+// corresponding lfta.\r
+// Later, lfta_iface_qnames are the query names matching lfta_iface_lists\r
+\r
+\r
+\r
+ // check if lfta is reusable\r
+ // FTA is reusable if reusable option is set explicitly or it doesn't have any parameters\r
+\r
+ bool lfta_reusable = false;\r
+ if (split_queries[l]->query_plan[0]->get_val_of_def("reusable") == "yes" ||\r
+ split_queries[l]->query_plan[0]->get_param_tbl()->size() == 0) {\r
+ lfta_reusable = true;\r
+ }\r
+ lfta_reuse_options.push_back(lfta_reusable);\r
+\r
+ // LFTA will inherit the liveness timeout specification from the containing query\r
+ // it is too conservative as lfta are expected to spend less time per tuple\r
+ // then full query\r
+\r
+ // extract liveness timeout from query definition\r
+ int liveness_timeout = atoi(split_queries[l]->query_plan[0]->get_val_of_def("liveness_timeout").c_str());\r
+ if (!liveness_timeout) {\r
+// fprintf(stderr, "WARNING: No liveness timeout specified for lfta %s, using default value of %d\n",\r
+// split_queries[l]->query_name.c_str(), DEFAULT_UDOP_LIVENESS_TIMEOUT);\r
+ liveness_timeout = DEFAULT_UDOP_LIVENESS_TIMEOUT;\r
+ }\r
+ lfta_liveness_timeouts.push_back(liveness_timeout);\r
+\r
+// Add it to the schema\r
+ table_def *td = split_queries[l]->get_output_tabledef();\r
+ Schema->append_table(td);\r
+//printf("added lfta %d (%s)\n",l,split_queries[l]->query_name.c_str());\r
+\r
+ }\r
+ }\r
+\r
+// If the output is lfta-only, dump out the query name.\r
+ if(split_queries.size() == 1 && !hfta_returned){\r
+ if(output_query_names ){\r
+ fprintf(query_name_output,"%s L\n",split_queries[0]->query_name.c_str());\r
+ }\r
+/*\r
+else{\r
+ fprintf(stderr,"query name is %s\n",split_queries[0]->query_name.c_str());\r
+ }\r
+*/\r
+\r
+/*\r
+// output schema summary\r
+ if(output_schema_summary){\r
+ dump_summary(split_queries[0]);\r
+ }\r
+*/\r
+ }\r
+\r
+\r
+ if(hfta_returned){ // query also has an HFTA component\r
+ int hfta_nbr = split_queries.size()-1;\r
+\r
+ hfta_list.push_back(split_queries[hfta_nbr]);\r
+\r
+// report on generated query names\r
+ if(output_query_names){\r
+ string hfta_name =split_queries[hfta_nbr]->query_name;\r
+ fprintf(query_name_output,"%s H\n",hfta_name.c_str());\r
+ for(l=0;l<hfta_nbr;++l){\r
+ string lfta_name =split_queries[l]->query_name;\r
+ fprintf(query_name_output,"%s L\n",lfta_name.c_str());\r
+ }\r
+ }\r
+// else{\r
+// fprintf(stderr,"query names are ");\r
+// for(l=0;l<hfta_nbr;++l){\r
+// if(l>0) fprintf(stderr,",");\r
+// string fta_name =split_queries[l]->query_name;\r
+// fprintf(stderr," %s",fta_name.c_str());\r
+// }\r
+// fprintf(stderr,"\n");\r
+// }\r
+ }\r
+\r
+ }else{\r
+ fprintf(stderr,"ERROR, query in file %s has a bug.\n",qnodes[i]->file.c_str());\r
+ fprintf(stderr,"%s\n",rootq->get_error_str().c_str());\r
+ exit(1);\r
+ }\r
+ }\r
+}\r
+\r
+\r
+//-----------------------------------------------------------------\r
+// Compute and propagate the SE in PROTOCOL fields compute a field.\r
+//-----------------------------------------------------------------\r
+\r
+for(i=0;i<lfta_list.size();i++){\r
+ lfta_list[i]->generate_protocol_se(sq_map, Schema);\r
+ sq_map[lfta_list[i]->get_sq_name()] = lfta_list[i];\r
+}\r
+for(i=0;i<hfta_list.size();i++){\r
+ hfta_list[i]->generate_protocol_se(sq_map, Schema);\r
+ sq_map[hfta_list[i]->get_sq_name()] = hfta_list[i];\r
+}\r
+\r
+\r
+\r
+//------------------------------------------------------------------------\r
+// Perform individual FTA optimizations\r
+//-----------------------------------------------------------------------\r
+\r
+if (partitioned_mode) {\r
+\r
+ // open partition definition file\r
+ string part_fname = config_dir_path + "partition.txt";\r
+\r
+ FILE* partfd = fopen(part_fname.c_str(), "r");\r
+ if (!partfd) {\r
+ fprintf(stderr,"ERROR, unable to open partition definition file %s\n",part_fname.c_str());\r
+ exit(1);\r
+ }\r
+ PartnParser_setfileinput(partfd);\r
+ if (PartnParserparse()) {\r
+ fprintf(stderr,"ERROR, unable to parse partition definition file %s\n",part_fname.c_str());\r
+ exit(1);\r
+ }\r
+ fclose(partfd);\r
+}\r
+\r
+\r
+print_hftas("preopt_hfta_info.txt", hfta_list, lfta_names, query_names, interface_names);\r
+\r
+int num_hfta = hfta_list.size();\r
+for(i=0; i < hfta_list.size(); ++i){\r
+ hfta_list[i]->optimize(hfta_list, lfta_names, interface_names, machine_names, Ext_fcns, Schema, ifaces_db, partn_parse_result);\r
+}\r
+\r
+// Add all new hftas to schema\r
+for(i=num_hfta; i < hfta_list.size(); ++i){\r
+ table_def *td = hfta_list[i]->get_output_tabledef();\r
+ Schema->append_table(td);\r
+}\r
+\r
+print_hftas("postopt_hfta_info.txt", hfta_list, lfta_names, query_names, interface_names);\r
+\r
+\r
+\r
+//------------------------------------------------------------------------\r
+// Do global (cross-fta) optimization\r
+//-----------------------------------------------------------------------\r
+\r
+\r
+\r
+\r
+\r
+\r
+set<string> extra_external_libs;\r
+\r
+for(i=0;i<hfta_list.size();++i){ // query also has an HFTA component\r
+\r
+ if(! debug_only){\r
+// build hfta file name, create output\r
+ if(numeric_hfta_flname){\r
+ sprintf(tmpstr,"hfta_%d",hfta_count);\r
+ hfta_names.push_back(tmpstr);\r
+ sprintf(tmpstr,"hfta_%d.cc",hfta_count);\r
+ }else{\r
+ sprintf(tmpstr,"hfta_%s",hfta_list[i]->query_name.c_str());\r
+ hfta_names.push_back(tmpstr);\r
+ sprintf(tmpstr,"hfta_%s.cc",hfta_list[i]->query_name.c_str());\r
+ }\r
+ FILE *hfta_fl = fopen(tmpstr,"w");\r
+ if(hfta_fl == NULL){\r
+ fprintf(stderr,"ERROR can't open fta output file %s\n",tmpstr);\r
+ exit(1);\r
+ }\r
+ fprintf(hfta_fl,"%s\n\n",hfta_list[i]->generate_hfta(Schema, Ext_fcns, opviews, distributed_mode).c_str());\r
+\r
+// If there is a field verifier, warn about\r
+// lack of compatability\r
+// NOTE : this code assumes that visible non-lfta queries\r
+// are those at the root of a stream query.\r
+ string hfta_comment;\r
+ string hfta_title;\r
+ string hfta_namespace;\r
+ if(hfta_list[i]->defines.count("comment")>0)\r
+ hfta_comment = hfta_list[i]->defines["comment"];\r
+ if(hfta_list[i]->defines.count("Comment")>0)\r
+ hfta_comment = hfta_list[i]->defines["Comment"];\r
+ if(hfta_list[i]->defines.count("COMMENT")>0)\r
+ hfta_comment = hfta_list[i]->defines["COMMENT"];\r
+ if(hfta_list[i]->defines.count("title")>0)\r
+ hfta_title = hfta_list[i]->defines["title"];\r
+ if(hfta_list[i]->defines.count("Title")>0)\r
+ hfta_title = hfta_list[i]->defines["Title"];\r
+ if(hfta_list[i]->defines.count("TITLE")>0)\r
+ hfta_title = hfta_list[i]->defines["TITLE"];\r
+ if(hfta_list[i]->defines.count("namespace")>0)\r
+ hfta_namespace = hfta_list[i]->defines["namespace"];\r
+ if(hfta_list[i]->defines.count("Namespace")>0)\r
+ hfta_namespace = hfta_list[i]->defines["Namespace"];\r
+ if(hfta_list[i]->defines.count("Namespace")>0)\r
+ hfta_namespace = hfta_list[i]->defines["Namespace"];\r
+\r
+ if(field_verifier != NULL){\r
+ string warning_str;\r
+ if(hfta_comment == "")\r
+ warning_str += "\tcomment not found.\n";\r
+ if(hfta_title == "")\r
+ warning_str += "\ttitle not found.\n";\r
+ if(hfta_namespace == "")\r
+ warning_str += "\tnamespace not found.\n";\r
+\r
+ vector<field_entry *> flds = hfta_list[i]->get_output_tabledef()->get_fields();\r
+ int fi;\r
+ for(fi=0;fi<flds.size();fi++){\r
+ field_verifier->verify_field(flds[fi]->get_name(),flds[fi]->get_type(),warning_str);\r
+ }\r
+ if(warning_str != "")\r
+ fprintf(stderr,"Warning, in HFTA stream %s:\n%s",\r
+ hfta_list[i]->get_output_tabledef()->get_tbl_name().c_str(),warning_str.c_str());\r
+ }\r
+\r
+ fprintf(qtree_output,"\t<HFTA name='%s'>\n",hfta_list[i]->get_output_tabledef()->get_tbl_name().c_str());\r
+ if(hfta_comment != "")\r
+ fprintf(qtree_output,"\t\t<Description value='%s' />\n",hfta_comment.c_str());\r
+ if(hfta_title != "")\r
+ fprintf(qtree_output,"\t\t<Title value='%s' />\n",hfta_title.c_str());\r
+ if(hfta_namespace != "")\r
+ fprintf(qtree_output,"\t\t<Namespace value='%s' />\n",hfta_namespace.c_str());\r
+ fprintf(qtree_output,"\t\t<FileName value='%s' />\n",tmpstr);\r
+ fprintf(qtree_output,"\t\t<Rate value='100' />\n");\r
+\r
+// write info about fields to qtree.xml\r
+ vector<field_entry *> flds = hfta_list[i]->get_output_tabledef()->get_fields();\r
+ int fi;\r
+ for(fi=0;fi<flds.size();fi++){\r
+ fprintf(qtree_output,"\t\t<Field name='%s' pos='%d' type='%s' ",flds[fi]->get_name().c_str(), fi, flds[fi]->get_type().c_str());\r
+ if(flds[fi]->get_modifier_list()->size()){\r
+ fprintf(qtree_output,"mods='%s' ", flds[fi]->get_modifier_list()->to_string().c_str());\r
+ }\r
+ fprintf(qtree_output," />\n");\r
+ }\r
+\r
+ // extract liveness timeout from query definition\r
+ int liveness_timeout = atoi(hfta_list[i]->query_plan[hfta_list[i]->qhead]->get_val_of_def("liveness_timeout").c_str());\r
+ if (!liveness_timeout) {\r
+// fprintf(stderr, "WARNING: No liveness timeout specified for hfta %s, using default value of %d\n",\r
+// hfta_list[i]->get_output_tabledef()->get_tbl_name().c_str(), DEFAULT_UDOP_LIVENESS_TIMEOUT);\r
+ liveness_timeout = DEFAULT_UDOP_LIVENESS_TIMEOUT;\r
+ }\r
+ fprintf(qtree_output,"\t\t<LivenessTimeout value='%d' />\n", liveness_timeout);\r
+\r
+ vector<tablevar_t *> tmp_tv = hfta_list[i]->get_input_tables();\r
+ int itv;\r
+ for(itv=0;itv<tmp_tv.size();++itv){\r
+ fprintf(qtree_output,"\t\t<ReadsFrom value='%s' />\n",tmp_tv[itv]->get_schema_name().c_str());\r
+ }\r
+ string ifrs = hfta_list[i]->collect_refd_ifaces();\r
+ if(ifrs != ""){\r
+ fprintf(qtree_output,"\t\t<Ifaces_referenced value='%s'/>\n",ifrs.c_str());\r
+ }\r
+ fprintf(qtree_output,"\t</HFTA>\n");\r
+\r
+ fclose(hfta_fl);\r
+ }else{\r
+// debug only -- do code generation to catch generation-time errors.\r
+ hfta_list[i]->generate_hfta(Schema, Ext_fcns,opviews, distributed_mode);\r
+ }\r
+\r
+ hfta_count++; // for hfta file names with numeric suffixes\r
+\r
+ hfta_list[i]->get_external_libs(extra_external_libs);\r
+\r
+ }\r
+\r
+string ext_lib_string;\r
+set<string>::iterator ssi_el;\r
+for(ssi_el=extra_external_libs.begin();ssi_el!=extra_external_libs.end();++ssi_el)\r
+ ext_lib_string += (*ssi_el)+" ";\r
+\r
+\r
+\r
+// Report on the set of operator views\r
+ for(i=0;i<opviews.size();++i){\r
+ opview_entry *opve = opviews.get_entry(i);\r
+ fprintf(qtree_output,"\t<UDOP name='%s' >\n",opve->view_name.c_str());\r
+ fprintf(qtree_output,"\t\t<FileName value='%s' />\n",opve->exec_fl.c_str());\r
+ fprintf(qtree_output,"\t\t<Parent value='%s' />\n",opve->root_name.c_str());\r
+ fprintf(qtree_output,"\t\t<Alias value='%s' />\n",opve->udop_alias.c_str());\r
+ fprintf(qtree_output,"\t\t<Rate value='100' />\n");\r
+\r
+ if (!opve->liveness_timeout) {\r
+// fprintf(stderr, "WARNING: No liveness timeout specified for view %s, using default value of %d\n",\r
+// opve->view_name.c_str(), DEFAULT_UDOP_LIVENESS_TIMEOUT);\r
+ opve->liveness_timeout = DEFAULT_UDOP_LIVENESS_TIMEOUT;\r
+ }\r
+ fprintf(qtree_output,"\t\t<LivenessTimeout value='%d' />\n", opve->liveness_timeout);\r
+ int j;\r
+ for(j=0;j<opve->subq_names.size();j++)\r
+ fprintf(qtree_output,"\t\t<ReadsFrom value='%s' />\n",opve->subq_names[j].c_str());\r
+ fprintf(qtree_output,"\t</UDOP>\n");\r
+ }\r
+\r
+\r
+//-----------------------------------------------------------------\r
+\r
+// Create interface-specific meta code files.\r
+// first, open and parse the interface resources file.\r
+ ifaces_db = new ifq_t();\r
+ ierr = "";\r
+ if(ifaces_db->load_ifaces(ifx_fname,use_live_hosts_file,distributed_mode,ierr)){\r
+ fprintf(stderr,"ERROR, can't load interface resource file %s :\n%s",\r
+ ifx_fname.c_str(), ierr.c_str());\r
+ exit(1);\r
+ }\r
+\r
+ map<string, vector<stream_query *> >::iterator svsi;\r
+ for(svsi=lfta_mach_lists.begin(); svsi!=lfta_mach_lists.end(); ++svsi){\r
+ string lmach = (*svsi).first;\r
+\r
+ // For this machine, create a set of lftas per interface.\r
+ vector<stream_query *> mach_lftas = (*svsi).second;\r
+ map<string, vector<stream_query *> > lfta_iface_lists;\r
+ int li;\r
+ for(li=0;li<mach_lftas.size();++li){\r
+ vector<tablevar_t *> tvec = mach_lftas[li]->query_plan[0]->get_input_tbls();\r
+ string lfta_iface = tvec[0]->get_interface();\r
+ lfta_iface_lists[lfta_iface].push_back(mach_lftas[li]);\r
+ }\r
+\r
+ map<string, vector<stream_query *> >::iterator lsvsi;\r
+ for(lsvsi=lfta_iface_lists.begin(); lsvsi!=lfta_iface_lists.end(); ++lsvsi){\r
+ int erri;\r
+ string liface = (*lsvsi).first;\r
+ vector<stream_query *> iface_lftas = (*lsvsi).second;\r
+ vector<string> iface_codegen_type = ifaces_db->get_iface_vals(lmach,liface,"iface_codegen_type",erri,err_str);\r
+ if(iface_codegen_type.size()){\r
+ nic_property *nicprop = npdb->get_nic_property(iface_codegen_type[0],erri);\r
+ if(!nicprop){\r
+ fprintf(stderr,"ERROR cannot load nic properties %s for interface %s of machine %s\n",iface_codegen_type[0].c_str(), liface.c_str(), lmach.c_str());\r
+ exit(1);\r
+ }\r
+ string mcs = generate_nic_code(iface_lftas, nicprop);\r
+ string mcf_flnm;\r
+ if(lmach != "")\r
+ mcf_flnm = lmach + "_"+liface+".mcf";\r
+ else\r
+ mcf_flnm = hostname + "_"+liface+".mcf";\r
+ FILE *mcf_fl ;\r
+ if((mcf_fl = fopen(mcf_flnm.c_str(),"w")) == NULL){\r
+ fprintf(stderr,"Can't open output file %s\n%s\n",mcf_flnm.c_str(),strerror(errno));\r
+ exit(1);\r
+ }\r
+ fprintf(mcf_fl,"%s",mcs.c_str());\r
+ fclose(mcf_fl);\r
+//printf("mcs of machine %s, iface %s of type %s is \n%s\n",\r
+//lmach.c_str(), liface.c_str(),iface_codegen_type[0].c_str(), mcs.c_str());\r
+ }\r
+ }\r
+\r
+\r
+ }\r
+\r
+\r
+\r
+//-----------------------------------------------------------------\r
+\r
+\r
+// Find common filter predicates in the LFTAs.\r
+// in addition generate structs to store the temporal attributes unpacked by prefilter\r
+ \r
+ map<string, vector<stream_query *> >::iterator ssqi;\r
+ for(ssqi=lfta_mach_lists.begin(); ssqi!=lfta_mach_lists.end(); ++ssqi){\r
+\r
+ string lmach = (*ssqi).first;\r
+ bool packed_return = false;\r
+ int li, erri;\r
+\r
+\r
+// The LFTAs of this machine.\r
+ vector<stream_query *> mach_lftas = (*ssqi).second;\r
+// break up on a per-interface basis.\r
+ map<string, vector<stream_query *> > lfta_iface_lists;\r
+ map<string, vector<int> > lfta_iface_qname_ix; // need the query name\r
+ // for fta_init\r
+ for(li=0;li<mach_lftas.size();++li){\r
+ vector<tablevar_t *> tvec = mach_lftas[li]->query_plan[0]->get_input_tbls();\r
+ string lfta_iface = tvec[0]->get_interface();\r
+ lfta_iface_lists[lfta_iface].push_back(mach_lftas[li]);\r
+ lfta_iface_qname_ix[lfta_iface].push_back(mach_query_names[lmach][li]);\r
+ }\r
+\r
+\r
+// Are the return values "packed"?\r
+// This should be done on a per-interface basis.\r
+// But this is defunct code for gs-lite\r
+ for(li=0;li<mach_lftas.size();++li){\r
+ vector<tablevar_t *> tvec = mach_lftas[li]->query_plan[0]->get_input_tbls();\r
+ string liface = tvec[0]->get_interface();\r
+ vector<string> iface_codegen_type = ifaces_db->get_iface_vals(lmach,liface,"iface_codegen_type",erri,err_str);\r
+ if(iface_codegen_type.size()){\r
+ if(npdb->get_nic_prop_val(iface_codegen_type[0],"Return",erri) == "Packed"){\r
+ packed_return = true;\r
+ }\r
+ }\r
+ }\r
+\r
+\r
+// Separate lftas by interface, collect results on a per-interface basis.\r
+\r
+ vector<cnf_set *> no_preds; // fallback if there is no prefilter\r
+ map<string, vector<cnf_set *> > prefilter_preds;\r
+ set<unsigned int> pred_ids; // this can be global for all interfaces\r
+ for(auto mvsi=lfta_iface_lists.begin(); mvsi!=lfta_iface_lists.end(); ++mvsi){\r
+ string liface = (*mvsi).first;\r
+ vector<cnf_set *> empty_list;\r
+ prefilter_preds[liface] = empty_list;\r
+ if(! packed_return){\r
+ get_common_lfta_filter(lfta_iface_lists[liface], Schema,Ext_fcns, prefilter_preds[liface], pred_ids);\r
+ }\r
+\r
+// get NIC capabilities. (Is this needed?)\r
+ nic_property *nicprop = NULL;\r
+ vector<string> iface_codegen_type = ifaces_db->get_iface_vals(lmach,liface,"iface_codegen_type",erri,err_str);\r
+ if(iface_codegen_type.size()){\r
+ nicprop = npdb->get_nic_property(iface_codegen_type[0],erri);\r
+ if(!nicprop){\r
+ fprintf(stderr,"ERROR cannot load nic properties %s for interface %s of machine %s for lfta codegen\n",iface_codegen_type[0].c_str(), liface.c_str(), lmach.c_str());\r
+ exit(1);\r
+ }\r
+ }\r
+ }\r
+\r
+\r
+// Now that we know the prefilter preds, generate the lfta code.\r
+// Do this for all lftas in this machine.\r
+ for(li=0;li<mach_lftas.size();++li){\r
+ set<unsigned int> subsumed_preds;\r
+ set<unsigned int>::iterator sii;\r
+#ifdef PREFILTER_OK\r
+ for(sii=pred_ids.begin();sii!=pred_ids.end();++sii){\r
+ int pid = (*sii);\r
+ if((pid>>16) == li){\r
+ subsumed_preds.insert(pid & 0xffff);\r
+ }\r
+ }\r
+#endif\r
+ string lfta_schema_str = mach_lftas[li]->make_schema();\r
+ string lfta_schema_embed = make_C_embedded_string(lfta_schema_str);\r
+ nic_property *nicprop = NULL; // no NIC properties?\r
+ lfta_val[lmach] += generate_lfta_block(mach_lftas[li]->query_plan[0], Schema, mach_lftas[li]->get_gid(), Ext_fcns, lfta_schema_embed, ifaces_db,nicprop,subsumed_preds);\r
+ }\r
+\r
+\r
+// generate structs to store the temporal attributes\r
+// unpacked by prefilter\r
+ col_id_set temp_cids;\r
+ get_prefilter_temporal_cids(lfta_mach_lists[lmach], temp_cids);\r
+ lfta_prefilter_val[lmach] = generate_lfta_prefilter_struct(temp_cids, Schema);\r
+\r
+// Compute the lfta bit signatures and the lfta colrefs\r
+// do this on a per-interface basis\r
+#ifdef PREFILTER_OK\r
+ lfta_prefilter_val[lmach] += "#define PREFILTER_DEFINED 1;\n\n";\r
+#endif\r
+ map<string, vector<long long int> > lfta_sigs; // used again later\r
+ for(auto mvsi=lfta_iface_lists.begin(); mvsi!=lfta_iface_lists.end(); ++mvsi){\r
+ string liface = (*mvsi).first;\r
+ vector<long long int> empty_list;\r
+ lfta_sigs[liface] = empty_list;\r
+\r
+ vector<col_id_set> lfta_cols;\r
+ vector<int> lfta_snap_length;\r
+ for(li=0;li<lfta_iface_lists[liface].size();++li){\r
+ unsigned long long int mask=0, bpos=1;\r
+ int f_pos;\r
+ for(f_pos=0;f_pos<prefilter_preds[liface].size();++f_pos){\r
+ if(prefilter_preds[liface][f_pos]->lfta_id.count(li))\r
+ mask |= bpos;\r
+ bpos = bpos << 1;\r
+ }\r
+ lfta_sigs[liface].push_back(mask);\r
+ lfta_cols.push_back(lfta_iface_lists[liface][li]->query_plan[0]->get_colrefs(true,Schema));\r
+ lfta_snap_length.push_back(compute_snap_len(lfta_iface_lists[liface][li]->query_plan[0], Schema));\r
+ }\r
+\r
+//for(li=0;li<mach_lftas.size();++li){\r
+//printf("lfta %d, msak is %llu\n",li,lfta_sigs[li]);\r
+//col_id_set::iterator tcisi;\r
+//for(tcisi=lfta_cols[li].begin(); tcisi!=lfta_cols[li].end();++tcisi){\r
+//printf("\t%s %d (%d)\n",(*tcisi).field.c_str(),(*tcisi).schema_ref, (*tcisi).tblvar_ref);\r
+//}\r
+//}\r
+\r
+\r
+// generate the prefilter\r
+// Do this on a per-interface basis, except for the #define\r
+#ifdef PREFILTER_OK\r
+// lfta_prefilter_val[lmach] += "#define PREFILTER_DEFINED 1;\n\n";\r
+ lfta_val[lmach] += generate_lfta_prefilter(prefilter_preds[liface], temp_cids, Schema, Ext_fcns, lfta_cols, lfta_sigs[liface], lfta_snap_length, liface);\r
+#else\r
+ lfta_val[lmach] += generate_lfta_prefilter(no_preds, temp_cids, Schema,Ext_fcns, lfta_cols, lfta_sigs[liface], lfta_snap_length, liface);\r
+\r
+#endif\r
+ }\r
+\r
+// Generate interface parameter lookup function\r
+ lfta_val[lmach] += "// lookup interface properties by name\n";\r
+ lfta_val[lmach] += "// mulitple values of given property returned in the same string separated by commas\n";\r
+ lfta_val[lmach] += "// returns NULL if given property does not exist\n";\r
+ lfta_val[lmach] += "gs_sp_t get_iface_properties (const gs_sp_t iface_name, const gs_sp_t property_name) {\n";\r
+\r
+// collect a lit of interface names used by queries running on this host\r
+ set<std::string> iface_names;\r
+ for(i=0;i<mach_query_names[lmach].size();i++){\r
+ int mi = mach_query_names[lmach][i];\r
+ stream_query *lfta_sq = lfta_mach_lists[lmach][i];\r
+\r
+ if(interface_names[mi]=="")\r
+ iface_names.insert("DEFAULTDEV");\r
+ else\r
+ iface_names.insert(interface_names[mi]);\r
+ }\r
+\r
+// generate interface property lookup code for every interface\r
+ set<std::string>::iterator sir;\r
+ for (sir = iface_names.begin(); sir != iface_names.end(); sir++) {\r
+ if (sir == iface_names.begin())\r
+ lfta_val[lmach] += "\tif (!strcmp(iface_name, \"" + *sir + "\")) {\n";\r
+ else\r
+ lfta_val[lmach] += "\t} else if (!strcmp(iface_name, \"" + *sir + "\")) {\n";\r
+\r
+ // iterate through interface properties\r
+ vector<string> iface_properties = ifaces_db->get_iface_properties(lmach,*sir,erri,err_str);\r
+ if (erri) {\r
+ fprintf(stderr,"ERROR cannot retrieve interface properties for %s.%s, %s\n",lmach.c_str(), sir->c_str(), err_str.c_str());\r
+ exit(1);\r
+ }\r
+ if (iface_properties.empty())\r
+ lfta_val[lmach] += "\t\treturn NULL;\n";\r
+ else {\r
+ for (int i = 0; i < iface_properties.size(); ++i) {\r
+ if (i == 0)\r
+ lfta_val[lmach] += "\t\tif (!strcmp(property_name, \"" + iface_properties[i] + "\")) {\n";\r
+ else\r
+ lfta_val[lmach] += "\t\t} else if (!strcmp(property_name, \"" + iface_properties[i] + "\")) {\n";\r
+\r
+ // combine all values for the interface property using comma separator\r
+ vector<string> vals = ifaces_db->get_iface_vals(lmach,*sir,iface_properties[i],erri,err_str);\r
+ for (int j = 0; j < vals.size(); ++j) {\r
+ lfta_val[lmach] += "\t\t\treturn \"" + vals[j];\r
+ if (j != vals.size()-1)\r
+ lfta_val[lmach] += ",";\r
+ }\r
+ lfta_val[lmach] += "\";\n";\r
+ }\r
+ lfta_val[lmach] += "\t\t} else\n";\r
+ lfta_val[lmach] += "\t\t\treturn NULL;\n";\r
+ }\r
+ }\r
+ lfta_val[lmach] += "\t} else\n";\r
+ lfta_val[lmach] += "\t\treturn NULL;\n";\r
+ lfta_val[lmach] += "}\n\n";\r
+\r
+\r
+// Generate a full list of FTAs for clearinghouse reference\r
+ lfta_val[lmach] += "// list of FTAs clearinghouse expects to register themselves\n";\r
+ lfta_val[lmach] += "const gs_sp_t fta_names[] = {";\r
+\r
+ for (i = 0; i < query_names.size(); ++i) {\r
+ if (i)\r
+ lfta_val[lmach] += ", ";\r
+ lfta_val[lmach] += "\"" + query_names[i] + "\"";\r
+ }\r
+ for (i = 0; i < hfta_list.size(); ++i) {\r
+ lfta_val[lmach] += ", \"" + hfta_list[i]->query_name + "\"";\r
+ }\r
+ lfta_val[lmach] += ", NULL};\n\n";\r
+\r
+\r
+// Add the initialization function to lfta.c\r
+// Change to accept the interface name, and \r
+// set the prefilter function accordingly.\r
+// see the example in demo/err2\r
+ lfta_val[lmach] += "void fta_init(gs_sp_t device){\n";\r
+\r
+// for(i=0;i<mach_query_names[lmach].size();i++)\r
+// int mi = mach_query_names[lmach][i];\r
+// stream_query *lfta_sq = lfta_mach_lists[lmach][i];\r
+\r
+ for(auto mvsi=lfta_iface_lists.begin(); mvsi!=lfta_iface_lists.end(); ++mvsi){\r
+ string liface = (*mvsi).first;\r
+ vector<stream_query *> lfta_list = (*mvsi).second;\r
+ for(i=0;i<lfta_list.size();i++){\r
+ stream_query *lfta_sq = lfta_list[i];\r
+ int mi = lfta_iface_qname_ix[liface][i];\r
+ \r
+ fprintf(stderr,"interface %s, LFTA %d, snap length is %d\n",liface.c_str(),mi,snap_lengths[mi]);\r
+\r
+ string this_iface = "DEFAULTDEV";\r
+ if(interface_names[mi]!="")\r
+ this_iface = '"'+interface_names[mi]+'"';\r
+ lfta_val[lmach] += "\tif(!strcmp(device,"+this_iface+"))\n";\r
+ lfta_val[lmach] += "\t\tfta_register(\""+query_names[mi]+"\", " + (lfta_reuse_options[mi]?"1":"0") + ", ";\r
+// if(interface_names[mi]=="")\r
+// lfta_val[lmach]+="DEFAULTDEV";\r
+// else\r
+// lfta_val[lmach]+='"'+interface_names[mi]+'"';\r
+ lfta_val[lmach] += this_iface;\r
+\r
+\r
+ lfta_val[lmach] += ", "+generate_alloc_name(query_names[mi])\r
+ +"\n#ifndef LFTA_IN_NIC\n\t,"+generate_schema_string_name(query_names[mi])\r
+ +"\n#endif\n";\r
+ sprintf(tmpstr,",%d",snap_lengths[mi]);\r
+ lfta_val[lmach] += tmpstr;\r
+\r
+// unsigned long long int mask=0, bpos=1;\r
+// int f_pos;\r
+// for(f_pos=0;f_pos<prefilter_preds.size();++f_pos){\r
+// if(prefilter_preds[f_pos]->lfta_id.count(i))\r
+// mask |= bpos;\r
+// bpos = bpos << 1;\r
+// }\r
+\r
+#ifdef PREFILTER_OK\r
+// sprintf(tmpstr,",%lluull",mask);\r
+ sprintf(tmpstr,",%lluull",lfta_sigs[liface][i]);\r
+ lfta_val[lmach]+=tmpstr;\r
+#else\r
+ lfta_val[lmach] += ",0ull";\r
+#endif\r
+\r
+ lfta_val[lmach] += ");\n";\r
+\r
+\r
+\r
+// End of lfta prefilter stuff\r
+// --------------------------------------------------\r
+\r
+// If there is a field verifier, warn about\r
+// lack of compatability\r
+ string lfta_comment;\r
+ string lfta_title;\r
+ string lfta_namespace;\r
+ map<string,string> ldefs = lfta_sq->query_plan[0]->get_definitions();\r
+ if(ldefs.count("comment")>0)\r
+ lfta_comment = lfta_sq->defines["comment"];\r
+ if(ldefs.count("Comment")>0)\r
+ lfta_comment = lfta_sq->defines["Comment"];\r
+ if(ldefs.count("COMMENT")>0)\r
+ lfta_comment = lfta_sq->defines["COMMENT"];\r
+ if(ldefs.count("title")>0)\r
+ lfta_title = lfta_sq->defines["title"];\r
+ if(ldefs.count("Title")>0)\r
+ lfta_title = lfta_sq->defines["Title"];\r
+ if(ldefs.count("TITLE")>0)\r
+ lfta_title = lfta_sq->defines["TITLE"];\r
+ if(ldefs.count("NAMESPACE")>0)\r
+ lfta_namespace = lfta_sq->defines["NAMESPACE"];\r
+ if(ldefs.count("Namespace")>0)\r
+ lfta_namespace = lfta_sq->defines["Namespace"];\r
+ if(ldefs.count("namespace")>0)\r
+ lfta_namespace = lfta_sq->defines["namespace"];\r
+\r
+ string lfta_ht_size;\r
+ if(lfta_sq->query_plan[0]->node_type()== "sgah_qpn" || lfta_sq->query_plan[0]->node_type()== "rsgah_qpn")\r
+ lfta_ht_size = int_to_string(DEFAULT_LFTA_HASH_TABLE_SIZE);\r
+ if(ldefs.count("aggregate_slots")>0){\r
+ lfta_ht_size = ldefs["aggregate_slots"];\r
+ }\r
+\r
+// NOTE : I'm assuming that visible lftas do not start with _fta.\r
+// -- will fail for non-visible simple selection queries.\r
+ if(field_verifier != NULL && query_names[mi].substr(0,1) != "_"){\r
+ string warning_str;\r
+ if(lfta_comment == "")\r
+ warning_str += "\tcomment not found.\n";\r
+ if(lfta_title == "")\r
+ warning_str += "\ttitle not found.\n";\r
+ if(lfta_namespace == "")\r
+ warning_str += "\tnamespace not found.\n";\r
+\r
+ vector<field_entry *> flds = lfta_sq->get_output_tabledef()->get_fields();\r
+ int fi;\r
+ for(fi=0;fi<flds.size();fi++){\r
+ field_verifier->verify_field(flds[fi]->get_name(),flds[fi]->get_type(),warning_str);\r
+ }\r
+ if(warning_str != "")\r
+ fprintf(stderr,"Warning, in LFTA stream %s:\n%s",\r
+ query_names[mi].c_str(),warning_str.c_str());\r
+ }\r
+\r
+\r
+// Create qtree output\r
+ fprintf(qtree_output,"\t<LFTA name='%s' >\n",query_names[mi].c_str());\r
+ if(lfta_comment != "")\r
+ fprintf(qtree_output,"\t\t<Description value='%s' />\n",lfta_comment.c_str());\r
+ if(lfta_title != "")\r
+ fprintf(qtree_output,"\t\t<Title value='%s' />\n",lfta_title.c_str());\r
+ if(lfta_namespace != "")\r
+ fprintf(qtree_output,"\t\t<Namespace value='%s' />\n",lfta_namespace.c_str());\r
+ if(lfta_ht_size != "")\r
+ fprintf(qtree_output,"\t\t<HtSize value='%s' />\n",lfta_ht_size.c_str());\r
+ if(lmach != "")\r
+ fprintf(qtree_output,"\t\t<Host value='%s' />\n",lmach.c_str());\r
+ else\r
+ fprintf(qtree_output,"\t\t<Host value='%s' />\n",hostname.c_str());\r
+ fprintf(qtree_output,"\t\t<Interface value='%s' />\n",interface_names[mi].c_str());\r
+ fprintf(qtree_output,"\t\t<ReadsFrom value='%s' />\n",interface_names[mi].c_str());\r
+ fprintf(qtree_output,"\t\t<Rate value='100' />\n");\r
+ fprintf(qtree_output,"\t\t<LivenessTimeout value='%d' />\n", lfta_liveness_timeouts[mi]);\r
+// write info about fields to qtree.xml\r
+ vector<field_entry *> flds = lfta_sq->get_output_tabledef()->get_fields();\r
+ int fi;\r
+ for(fi=0;fi<flds.size();fi++){\r
+ fprintf(qtree_output,"\t\t<Field name='%s' pos='%d' type='%s' ",flds[fi]->get_name().c_str(), fi, flds[fi]->get_type().c_str());\r
+ if(flds[fi]->get_modifier_list()->size()){\r
+ fprintf(qtree_output,"mods='%s' ", flds[fi]->get_modifier_list()->to_string().c_str());\r
+ }\r
+ fprintf(qtree_output," />\n");\r
+ }\r
+ fprintf(qtree_output,"\t</LFTA>\n");\r
+\r
+\r
+ }\r
+ }\r
+\r
+ for(auto mvsi=lfta_iface_lists.begin(); mvsi!=lfta_iface_lists.end(); ++mvsi){\r
+ string liface = (*mvsi).first;\r
+ lfta_val[lmach] += \r
+" if (!strcmp(device, \""+liface+"\")) \n"\r
+" lfta_prefilter = &lfta_prefilter_"+liface+"; \n"\r
+;\r
+ }\r
+ lfta_val[lmach] += \r
+" if(lfta_prefilter == NULL){\n"\r
+" fprintf(stderr, \"Error, interface %s not found in fta_init.\\n\",device);\n"\r
+" exit(1);\n"\r
+" }\n"\r
+;\r
+\r
+\r
+\r
+ lfta_val[lmach] += "}\n\n";\r
+\r
+ if(!(debug_only || hfta_only) ){\r
+ string lfta_flnm;\r
+ if(lmach != "")\r
+ lfta_flnm = lmach + "_lfta.c";\r
+ else\r
+ lfta_flnm = hostname + "_lfta.c";\r
+ if((lfta_out = fopen(lfta_flnm.c_str(),"w")) == NULL){\r
+ fprintf(stderr,"Can't open output file %s\n%s\n","lfta.c",strerror(errno));\r
+ exit(1);\r
+ }\r
+ fprintf(lfta_out,"%s",lfta_header.c_str());\r
+ fprintf(lfta_out,"%s",lfta_prefilter_val[lmach].c_str());\r
+ fprintf(lfta_out,"%s",lfta_val[lmach].c_str());\r
+ fclose(lfta_out);\r
+ }\r
+ }\r
+\r
+// Say what are the operators which must execute\r
+ if(opviews.size()>0)\r
+ fprintf(stderr,"The queries use the following external operators:\n");\r
+ for(i=0;i<opviews.size();++i){\r
+ opview_entry *opv = opviews.get_entry(i);\r
+ fprintf(stderr,"\t%s\n",opv->view_name.c_str());\r
+ }\r
+\r
+ if(create_makefile)\r
+ generate_makefile(input_file_names, nfiles, hfta_names, opviews,\r
+ machine_names, schema_file_name,\r
+ interface_names,\r
+ ifaces_db, config_dir_path,use_pads,ext_lib_string, rts_hload);\r
+\r
+\r
+ fprintf(qtree_output,"</QueryNodes>\n");\r
+\r
+ return(0);\r
+}\r
+\r
+////////////////////////////////////////////////////////////\r
+\r
+void generate_makefile(vector<string> &input_file_names, int nfiles,\r
+ vector<string> &hfta_names, opview_set &opviews,\r
+ vector<string> &machine_names,\r
+ string schema_file_name,\r
+ vector<string> &interface_names,\r
+ ifq_t *ifdb, string &config_dir_path,\r
+ bool use_pads,\r
+ string extra_libs,\r
+ map<string, vector<int> > &rts_hload\r
+ ){\r
+ int i,j;\r
+\r
+ if(config_dir_path != ""){\r
+ config_dir_path = "-C "+config_dir_path;\r
+ }\r
+\r
+ struct stat sb;\r
+ bool libz_exists = stat((root_path+"/lib/libz.a").c_str(),&sb) >= 0;\r
+ bool libast_exists = stat((root_path+"/lib/libast.a").c_str(),&sb) >= 0;\r
+\r
+// if(libz_exists && !libast_exists){\r
+// fprintf(stderr,"Configuration ERROR: ../../lib/libz.a exists but ../../lib/libast.a dows not.\n");\r
+// exit(1);\r
+// }\r
+\r
+// Get set of operator executable files to run\r
+ set<string> op_fls;\r
+ set<string>::iterator ssi;\r
+ for(i=0;i<opviews.size();++i){\r
+ opview_entry *opv = opviews.get_entry(i);\r
+ if(opv->exec_fl != "") op_fls.insert(opv->exec_fl);\r
+ }\r
+\r
+ FILE *outfl = fopen("Makefile", "w");\r
+ if(outfl==NULL){\r
+ fprintf(stderr,"Can't open Makefile for write, exiting.\n");\r
+ exit(0);\r
+ }\r
+\r
+ fputs(\r
+("CPP= g++ -O3 -g -I "+root_path+"/include -I "+root_path+"/include/hfta\n"\r
+"CC= gcc -O3 -g -I . -I "+root_path+"/include -I "+root_path+"/include/lfta"\r
+).c_str(), outfl\r
+);\r
+ if(generate_stats)\r
+ fprintf(outfl," -DLFTA_STATS");\r
+\r
+// Gather the set of interfaces\r
+// Also, gather "base interface names" for use in computing\r
+// the hash splitting to virtual interfaces.\r
+// TODO : must update to hanndle machines\r
+ set<string> ifaces;\r
+ set<string> base_vifaces; // base interfaces of virtual interfaces\r
+ map<string, string> ifmachines;\r
+ map<string, string> ifattrs;\r
+ for(i=0;i<interface_names.size();++i){\r
+ ifaces.insert(interface_names[i]);\r
+ ifmachines[interface_names[i]] = machine_names[i];\r
+\r
+ size_t Xpos = interface_names[i].find_last_of("X");\r
+ if(Xpos!=string::npos){\r
+ string iface = interface_names[i].substr(0,Xpos);\r
+ base_vifaces.insert(iface);\r
+ }\r
+ // get interface attributes and add them to the list\r
+ }\r
+\r
+// Do we need to include protobuf libraries?\r
+ bool use_proto = false;\r
+ int erri;\r
+ string err_str;\r
+ for(ssi=ifaces.begin();ssi!=ifaces.end();++ssi){\r
+ string ifnm = (*ssi);\r
+ vector<string> ift = ifdb->get_iface_vals(ifmachines[ifnm],ifnm, "InterfaceType", erri, err_str);\r
+ for(int ift_i=0;ift_i<ift.size();ift_i++){\r
+ if(ift[ift_i]=="PROTO"){\r
+ use_proto = true;\r
+ }\r
+ }\r
+ }\r
+\r
+ fprintf(outfl,\r
+"\n"\r
+"\n"\r
+"all: rts");\r
+ for(i=0;i<hfta_names.size();++i)\r
+ fprintf(outfl," %s",hfta_names[i].c_str());\r
+ fputs(\r
+("\n"\r
+"\n"\r
+"rts: lfta.o "+root_path+"/lib/libgscphost.a "+root_path+"/lib/libgscplftaaux.a "+root_path+"/lib/libgscprts.a "+root_path+"/lib/libclearinghouse.a\n"\r
+"\tg++ -O3 -g -o rts lfta.o ").c_str(), outfl);\r
+ if(use_pads)\r
+ fprintf(outfl,"-L. ");\r
+ fputs(\r
+("-L"+root_path+"/lib -lgscplftaaux ").c_str(), outfl);\r
+ if(use_pads)\r
+ fprintf(outfl,"-lgscppads -lpads ");\r
+ fprintf(outfl,\r
+"-lgscprts -lgscphost -lm -lgscpaux -lgscplftaaux -lclearinghouse -lresolv -lpthread -lgscpinterface -lz");\r
+ if(use_pads)\r
+ fprintf(outfl, " -lpz -lz -lbz ");\r
+ if(libz_exists && libast_exists)\r
+ fprintf(outfl," -last ");\r
+ if(use_pads)\r
+ fprintf(outfl, " -ldll -ldl ");\r
+ if(use_proto)\r
+ fprintf(outfl, " -L/usr/local/lib/ -lprotobuf-c ");\r
+ fprintf(outfl," -lgscpaux");\r
+#ifdef GCOV\r
+ fprintf(outfl," -fprofile-arcs");\r
+#endif\r
+ fprintf(outfl,\r
+"\n"\r
+"\n"\r
+"lfta.o: %s_lfta.c\n"\r
+"\t$(CC) -o lfta.o -c %s_lfta.c\n"\r
+"\n"\r
+"%s_lfta.c: external_fcns.def %s ",hostname.c_str(), hostname.c_str(), hostname.c_str(),schema_file_name.c_str());\r
+ for(i=0;i<nfiles;++i)\r
+ fprintf(outfl," %s",input_file_names[i].c_str());\r
+ if(hostname == ""){\r
+ fprintf(outfl,"\n\t%s/bin/translate_fta %s %s ",root_path.c_str(), config_dir_path.c_str(),schema_file_name.c_str());\r
+ }else{\r
+ fprintf(outfl,"\n\t%s/bin/translate_fta -h %s %s %s ", root_path.c_str(), hostname.c_str(), config_dir_path.c_str(),schema_file_name.c_str());\r
+ }\r
+ for(i=0;i<nfiles;++i)\r
+ fprintf(outfl," %s",input_file_names[i].c_str());\r
+ fprintf(outfl,"\n");\r
+\r
+ for(i=0;i<hfta_names.size();++i)\r
+ fprintf(outfl,\r
+("%s: %s.o\n"\r
+"\t$(CPP) -o %s %s.o -L"+root_path+"/lib -lgscpapp -lgscphostaux -lgscphost -lgscpinterface -lgscphftaaux -lgscphostaux -lm -lgscpaux -lclearinghouse -lresolv -lpthread -lgscpaux -lgscphftaaux -lgscpaux %s\n"\r
+"\n"\r
+"%s.o: %s.cc\n"\r
+"\t$(CPP) -o %s.o -c %s.cc\n"\r
+"\n"\r
+"\n").c_str(),\r
+ hfta_names[i].c_str(), hfta_names[i].c_str(),\r
+ hfta_names[i].c_str(), hfta_names[i].c_str(), extra_libs.c_str(),\r
+ hfta_names[i].c_str(), hfta_names[i].c_str(),\r
+ hfta_names[i].c_str(), hfta_names[i].c_str()\r
+ );\r
+\r
+ fprintf(outfl,\r
+("\n"\r
+"packet_schema.txt:\n"\r
+"\tln -s "+root_path+"/cfg/packet_schema.txt .\n"\r
+"\n"\r
+"external_fcns.def:\n"\r
+"\tln -s "+root_path+"/cfg/external_fcns.def .\n"\r
+"\n"\r
+"clean:\n"\r
+"\trm -rf core rts *.o %s_lfta.c external_fcns.def packet_schema.txt").c_str(),hostname.c_str());\r
+ for(i=0;i<hfta_names.size();++i)\r
+ fprintf(outfl," %s %s.cc",hfta_names[i].c_str(),hfta_names[i].c_str());\r
+ fprintf(outfl,"\n");\r
+\r
+ fclose(outfl);\r
+\r
+\r
+\r
+// Gather the set of interfaces\r
+// TODO : must update to hanndle machines\r
+// TODO : lookup interface attributes and add them as a parameter to rts process\r
+ outfl = fopen("runit", "w");\r
+ if(outfl==NULL){\r
+ fprintf(stderr,"Can't open runit for write, exiting.\n");\r
+ exit(0);\r
+ }\r
+\r
+\r
+ fputs(\r
+("#!/bin/sh\n"\r
+"./stopit\n"\r
++root_path+"/bin/gshub.py> /dev/null 2>&1 &\n"\r
+"sleep 5\n"\r
+"if [ ! -f gshub.log ]\n"\r
+"then\n"\r
+"\techo \"Failed to start bin/gshub.py\"\n"\r
+"\texit -1\n"\r
+"fi\n"\r
+"ADDR=`cat gshub.log`\n"\r
+"ps opgid= $! >> gs.pids\n"\r
+"./rts $ADDR default ").c_str(), outfl);\r
+// int erri;\r
+// string err_str;\r
+ for(ssi=ifaces.begin();ssi!=ifaces.end();++ssi){\r
+ string ifnm = (*ssi);\r
+ fprintf(outfl, "%s ",ifnm.c_str());\r
+ vector<string> ifv = ifdb->get_iface_vals(ifmachines[ifnm],ifnm, "Command", erri, err_str);\r
+ for(j=0;j<ifv.size();++j)\r
+ fprintf(outfl, "%s ",ifv[j].c_str());\r
+ }\r
+ fprintf(outfl, " &\n");\r
+ fprintf(outfl, "echo $! >> gs.pids\n");\r
+ for(i=0;i<hfta_names.size();++i)\r
+ fprintf(outfl,"./%s $ADDR default &\n",hfta_names[i].c_str());\r
+\r
+ for(j=0;j<opviews.opview_list.size();++j){\r
+ fprintf(outfl,"%s/views/%s %s &\n",root_path.c_str(),opviews.opview_list[j]->exec_fl.c_str(), opviews.opview_list[j]->mangler.c_str());\r
+ }\r
+\r
+ fclose(outfl);\r
+ system("chmod +x runit");\r
+\r
+ outfl = fopen("stopit", "w");\r
+ if(outfl==NULL){\r
+ fprintf(stderr,"Can't open stopit for write, exiting.\n");\r
+ exit(0);\r
+ }\r
+\r
+ fprintf(outfl,"#!/bin/sh\n"\r
+"rm -f gshub.log\n"\r
+"if [ ! -f gs.pids ]\n"\r
+"then\n"\r
+"exit\n"\r
+"fi\n"\r
+"for pgid in `cat gs.pids`\n"\r
+"do\n"\r
+"kill -TERM -$pgid\n"\r
+"done\n"\r
+"sleep 1\n"\r
+"for pgid in `cat gs.pids`\n"\r
+"do\n"\r
+"kill -9 -$pgid\n"\r
+"done\n"\r
+"rm gs.pids\n");\r
+\r
+ fclose(outfl);\r
+ system("chmod +x stopit");\r
+\r
+//-----------------------------------------------\r
+\r
+/* For now disable support for virtual interfaces\r
+ outfl = fopen("set_vinterface_hash.bat", "w");\r
+ if(outfl==NULL){\r
+ fprintf(stderr,"Can't open set_vinterface_hash.bat for write, exiting.\n");\r
+ exit(0);\r
+ }\r
+\r
+// The format should be determined by an entry in the ifres.xml file,\r
+// but for now hardcode the only example I have.\r
+ for(ssi=base_vifaces.begin();ssi!=base_vifaces.end();++ssi){\r
+ if(rts_hload.count((*ssi))){\r
+ string iface_name = (*ssi);\r
+ string iface_number = "";\r
+ for(j=iface_name.size()-1; j>0 && iface_number == ""; j--){\r
+ if(isdigit(iface_name[j])){\r
+ iface_number = iface_name[j];\r
+ if(j>0 && isdigit(iface_name[j-1]))\r
+ iface_number = iface_name[j-1] + iface_number;\r
+ }\r
+ }\r
+\r
+ fprintf(outfl,"dagconfig -d%s -S hat_range=",iface_number.c_str());\r
+ vector<int> halloc = rts_hload[iface_name];\r
+ int prev_limit = 0;\r
+ for(j=0;j<halloc.size();++j){\r
+ if(j>0)\r
+ fprintf(outfl,":");\r
+ fprintf(outfl,"%d-%d",prev_limit,halloc[j]);\r
+ prev_limit = halloc[j];\r
+ }\r
+ fprintf(outfl,"\n");\r
+ }\r
+ }\r
+ fclose(outfl);\r
+ system("chmod +x set_vinterface_hash.bat");\r
+*/\r
+}\r
+\r
+// Code for implementing a local schema\r
+/*\r
+ table_list qpSchema;\r
+\r
+// Load the schemas of any LFTAs.\r
+ int l;\r
+ for(l=0;l<hfta_nbr;++l){\r
+ stream_query *sq0 = split_queries[l];\r
+ table_def *td = sq0->get_output_tabledef();\r
+ qpSchema.append_table(td);\r
+ }\r
+// load the schemas of any other ref'd tables.\r
+// (e.g., hftas)\r
+ vector<tablevar_t *> input_tbl_names = split_queries[hfta_nbr]->get_input_tables();\r
+ int ti;\r
+ for(ti=0;ti<input_tbl_names.size();++ti){\r
+ int tbl_ref = qpSchema.get_table_ref(input_tbl_names[ti]->get_schema_name());\r
+ if(tbl_ref < 0){\r
+ tbl_ref = Schema->get_table_ref(input_tbl_names[ti]->get_schema_name());\r
+ if(tbl_ref < 0){\r
+ fprintf(stderr,"ERROR file %s references table %s, which is not in the schema.\n",input_file_names[i].c_str(), (input_tbl_names[ti]->get_schema_name()).c_str());\r
+ exit(1);\r
+ }\r
+ qpSchema.append_table(Schema->get_table(tbl_ref));\r
+ }\r
+ }\r
+*/\r
+\r
+// Functions related to parsing.\r
+\r
+/*\r
+static int split_string(char *instr,char sep, char **words,int max_words){\r
+ char *loc;\r
+ char *str;\r
+ int nwords = 0;\r
+\r
+ str = instr;\r
+ words[nwords++] = str;\r
+ while( (loc = strchr(str,sep)) != NULL){\r
+ *loc = '\0';\r
+ str = loc+1;\r
+ if(nwords >= max_words){\r
+ fprintf(stderr,"Error in split_string, too many words discovered (max is %d)\n",max_words);\r
+ nwords = max_words-1;\r
+ }\r
+ words[nwords++] = str;\r
+ }\r
+\r
+ return(nwords);\r
+}\r
+\r
+*/\r
+\r
Code Review / com / gs-lite.git / blobdiff