text:
//---------------------------------------------------------------------------//
// Copyright (c) 2018-2021 Mikhail Komarov <nemo@nil.foundation>
//
// MIT License
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//---------------------------------------------------------------------------//
/*! \file
    \brief Some non-INET-specific socket address code

    Extracted from inet_address.cc.
 */
#include <ostream>
#include <arpa/inet.h>

#include <nil/actor/network/socket_defs.hh>
#include <nil/actor/network/inet_address.hh>
#include <nil/actor/network/ip.hh>

#include <nil/actor/core/print.hh>

#include <boost/functional/hash.hpp>

/*
 * Equality
 * NOTE: Some of kernel programming environment (for example, openbsd/sparc)
 * does not supply memcmp().  For userland memcmp() is preferred as it is
 * in ANSI standard.
 *
 * This modification is specific to FreeBSD's struct sin6_addr implementation.
 */
#if !defined(_POSIX_C_SOURCE) || defined(_DARWIN_C_SOURCE)
#define SIN6_ARE_ADDR_EQUAL(a, b) \
    (memcmp(&(a)->sin6_addr.s6_addr[0], &(b)->sin6_addr.s6_addr[0], sizeof(struct in6_addr)) == 0)
#endif /* (_POSIX_C_SOURCE && !_DARWIN_C_SOURCE) */

using namespace std::string_literals;

size_t std::hash<nil::actor::socket_address>::operator()(const nil::actor::socket_address &a) const {
    auto h = std::hash<nil::actor::net::inet_address>()(a.addr());
    boost::hash_combine(h, a.as_posix_sockaddr_in().sin_port);
    return h;
}

namespace nil {
    namespace actor {

        static_assert(std::is_nothrow_default_constructible<socket_address>::value);
        static_assert(std::is_nothrow_copy_constructible<socket_address>::value);
        static_assert(std::is_nothrow_move_constructible<socket_address>::value);

        socket_address::socket_address() noexcept
            // set max addr_length, as we (probably) want to use the constructed object
            // in accept() or get_address()
            :
            addr_length(sizeof(::sockaddr_storage)) {
            static_assert(AF_UNSPEC == 0, "just checking");
            memset(&u, 0, sizeof(u));
        }

        socket_address::socket_address(uint16_t p) noexcept : socket_address(ipv4_addr(p)) {
        }

        socket_address::socket_address(ipv4_addr addr) noexcept {
            addr_length = sizeof(::sockaddr_in);
            u.in.sin_family = AF_INET;
            u.in.sin_port = htons(addr.port);
            u.in.sin_addr.s_addr = htonl(addr.ip);
        }

        socket_address::socket_address(const ipv6_addr &addr, uint32_t scope) noexcept {
            addr_length = sizeof(::sockaddr_in6);
            u.in6.sin6_family = AF_INET6;
            u.in6.sin6_port = htons(addr.port);
            u.in6.sin6_flowinfo = 0;
            u.in6.sin6_scope_id = scope;
            std::copy(addr.ip.begin(), addr.ip.end(), u.in6.sin6_addr.s6_addr);
        }

        socket_address::socket_address(const ipv6_addr &addr) noexcept :
            socket_address(addr, net::inet_address::invalid_scope) {
        }

        socket_address::socket_address(uint32_t ipv4, uint16_t p) noexcept :
            socket_address(make_ipv4_address(ipv4, p)) {
        }

        socket_address::socket_address(const net::inet_address &a, uint16_t p) noexcept :
            socket_address(a.is_ipv6() ? socket_address(ipv6_addr(a, p), a.scope()) : socket_address(ipv4_addr(a, p))) {
        }

        socket_address::socket_address(const unix_domain_addr &s) noexcept {
            u.un.sun_family = AF_UNIX;
            memset(u.un.sun_path, '\0', sizeof(u.un.sun_path));
            auto path_length = std::min((int)sizeof(u.un.sun_path), s.path_length());
            memcpy(u.un.sun_path, s.path_bytes(), path_length);
            addr_length = path_length + offsetof(struct ::sockaddr_un, sun_path);
        }

        bool socket_address::is_unspecified() const noexcept {
            return u.sa.sa_family == AF_UNSPEC;
        }

        static int adjusted_path_length(const socket_address &a) noexcept {
            int l = std::max(0, (int)a.addr_length - (int)(offsetof(sockaddr_un, sun_path)));
            // "un-count" a trailing null in filesystem-namespace paths
            if (a.u.un.sun_path[0] != '\0' && (l > 1) && a.u.un.sun_path[l - 1] == '\0') {
                --l;
            }
            return l;
        }

        bool socket_address::operator==(const socket_address &a) const noexcept {
            if (u.sa.sa_family != a.u.sa.sa_family) {
                return false;
            }
            if (u.sa.sa_family == AF_UNIX) {
                // tolerate counting/not counting a terminating null in filesystem-namespace paths
                int adjusted_len = adjusted_path_length(*this);
                int a_adjusted_len = adjusted_path_length(a);
                if (adjusted_len != a_adjusted_len) {
                    return false;
                }
                return (memcmp(u.un.sun_path, a.u.un.sun_path, adjusted_len) == 0);
            }

            // an INET address
            if (u.in.sin_port != a.u.in.sin_port) {
                return false;
            }
            switch (u.sa.sa_family) {
                case AF_INET:
                    return u.in.sin_addr.s_addr == a.u.in.sin_addr.s_addr;
                case AF_UNSPEC:
                case AF_INET6:
                    // handled below
                    break;
                default:
                    return false;
            }

            auto &in1 = as_posix_sockaddr_in6();
            auto &in2 = a.as_posix_sockaddr_in6();

#if defined(__APPLE__) || defined(__FreeBSD__)
            return SIN6_ARE_ADDR_EQUAL(&in1, &in2);
#elif defined(__linux__)
            return IN6_ARE_ADDR_EQUAL(&in1, &in2);
#endif
        }

        std::string unix_domain_addr_text(const socket_address &sa) {
            if (sa.length() <= offsetof(sockaddr_un, sun_path)) {
                return "{unnamed}"s;
            }
            if (sa.u.un.sun_path[0]) {
                // regular (filesystem-namespace) path
                return std::string {sa.u.un.sun_path};
            }

            const size_t path_length {sa.length() - offsetof(sockaddr_un, sun_path)};
            std::string ud_path(path_length, 0);
            char *targ = &ud_path[0];
            *targ++ = '@';
            const char *src = sa.u.un.sun_path + 1;
            int k = (int)path_length;

            for (; --k > 0; src++) {
                *targ++ = std::isprint(*src) ? *src : '_';
            }
            return ud_path;
        }

        std::ostream &operator<<(std::ostream &os, const socket_address &a) {
            if (a.is_af_unix()) {
                return os << unix_domain_addr_text(a);
            }

            auto addr = a.addr();
            // CMH. maybe skip brackets for ipv4-mapped
            auto bracket = addr.in_family() == nil::actor::net::inet_address::family::INET6;

            if (bracket) {
                os << '[';
            }
            os << addr;
            if (bracket) {
                os << ']';
            }

            return os << ':' << ntohs(a.u.in.sin_port);
        }

    }    // namespace actor
}    // namespace nil

ext: cc

max_stars_repo_name: NilFoundation/actorio

max_issues_repo_name: NilFoundation/actorio

max_forks_repo_name: NilFoundation/actorio


text:
#ifndef WEBMOCK_UTIL_URI_PARSER_HPP
#define WEBMOCK_UTIL_URI_PARSER_HPP

#include <iostream>
#include <map>
#include <boost/operators.hpp>

namespace webmock { namespace util {
    // Simple URI Parser
    // see: RFC3986(http://www.ietf.org/rfc/rfc3986.txt)
    struct uri_parser: boost::equality_comparable<uri_parser, uri_parser> {
        std::string scheme;
        std::string userinfo;
        std::string host;
        std::string port;
        std::string path;
        std::string query;
        std::string fragment;

        uri_parser(std::map<std::string, std::string> components);
        uri_parser(std::string const & uri);
    };

    bool operator ==(uri_parser const & lop, uri_parser const & rop);
    std::ostream & operator <<(std::ostream & lop, uri_parser const & rop);
}}

#ifndef WEBMOCK_BUILD_LIBRARY
#include <webmock/util/uri_parser.ipp>
#endif

#endif

ext: hpp


text:
#include <cstdlib>
#include <cstdio>
#include <stdlib.h>
#include <stdint.h>
#include <iostream>
#include <cstring>
#include <sstream>
#include <string>
#include <vector>
#include <queue>
#include <deque>

#include <atomic>
#include <thread>
#include <mutex>
#include <condition_variable>

#include <ctime>
#include <chrono>
#include <unistd.h>

#include <sys/socket.h>
#include <sys/types.h>
#include <netdb.h>
#include <arpa/inet.h>

#include <opencv2/opencv.hpp>

#include <boost/property_tree/ptree.hpp>
#include <boost/property_tree/json_parser.hpp>

#include <boost/fusion/container/vector/convert.hpp>
#include <boost/fusion/include/as_vector.hpp>

#include <boost/asio.hpp>

#include <boost/beast/core.hpp>
#include <boost/beast/http.hpp>
#include <boost/beast/version.hpp>
#include <boost/asio/connect.hpp>
#include <boost/asio/ip/tcp.hpp>

#include <Timer.hpp>

extern "C"
{
#include <libavdevice/avdevice.h>
#include <libavformat/avformat.h>
#include <libswscale/swscale.h>
#include <libavcodec/avcodec.h>
#include <libavutil/avutil.h>
#include <libavutil/imgutils.h>
}

#include <utils.hpp>
#include <string_manip.hpp>

#include <loop_condition.hpp>

#include <timed_mat.hpp>

#include <rtconnection.hpp>

rtconnection::~rtconnection()
{
  { fprintf(stderr, "CALLING RTCONNECTION DESTRUCTOR METHOD!!!\n"); }
}

rtconnection::rtconnection()
{
  loop = NULL;
  calibrated=false;
  calibrating=false;
  zeroed_pts=0;
  fromfile=false;
  scenevid=false;
  eyetracking=false;
  audio=false;
  eyevid=false;
}

rtconnection::rtconnection(loop_condition* _loop)
{
  loop = _loop;
  calibrated=false;
  calibrating=false;
  zeroed_pts=0;
  fromfile=false;
  scenevid=false;
  eyetracking=false;
  audio=false;
  eyevid=false;
}

ext: cpp


text:
/*
 * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
 */

/*
 * This file will not contain actual query processing code but instead only
 * the code to
 * a) Interact with external interfaces like REDIS etc.
 * b) Parse JSON strings passed to populate the query structures
 */

#include "rapidjson/document.h"
#include "base/logging.h"
#include "query.h"
#include <boost/assign/list_of.hpp>
#include <cerrno>
#include "analytics/vizd_table_desc.h"
#include "stats_select.h"
#include "stats_query.h"
#include <base/connection_info.h>

using std::map;
using std::string;
using boost::assign::map_list_of;
using boost::system::error_code;
using process::ConnectionState;
using process::ConnectionType;
using process::ConnectionStatus;

GenDb::GenDbIf* query_result_unit_t::dbif = NULL;
int QueryEngine::max_slice_ = 100;

typedef  std::vector< std::pair<std::string, std::string> > spair_vector;
static spair_vector query_string_to_column_name(0);

std::string get_column_name(std::string query_string)
{
    spair_vector::iterator iter;

    for (iter = query_string_to_column_name.begin();
            iter != query_string_to_column_name.end();
            iter++)
    {
        if (iter->first == query_string)
            return iter->second;
    }

    return query_string;
}

std::string get_query_string(std::string column_name)
{
    spair_vector::iterator iter;

    for (iter = query_string_to_column_name.begin();
            iter != query_string_to_column_name.end();
            iter++)
    {
        if (iter->second == column_name)
            return iter->first;
    }

    return column_name;
}

QueryResultMetaData::~QueryResultMetaData() {
}

PostProcessingQuery::PostProcessingQuery(
    std::map<std::string, std::string>& json_api_data,
    QueryUnit *main_query) :  QueryUnit(main_query, main_query),
        sorted(false), limit(0) {
    AnalyticsQuery *m_query = (AnalyticsQuery *)main_query;
    std::map<std::string, std::string>::iterator iter;

    QE_TRACE(DEBUG, __func__ );

    json_string_ = "";

    for (iter = json_api_data.begin(); iter != json_api_data.end(); iter++)
    {
        if (iter->first == QUERY_SORT_OP)
        {
            sorted = true;
            int tmp;
            std::istringstream(iter->second) >> tmp;
            sorting_type = (sort_op)tmp;
            m_query->merge_needed = true;
            QE_TRACE(DEBUG, "sorting_type :" << sorting_type);
            json_string_ += iter->second;
            json_string_ += " ";
        }

        if (iter->first == QUERY_LIMIT)
        {
            std::istringstream(iter->second) >> limit;
            m_query->merge_needed = true;
            QE_TRACE(DEBUG, "limit :"<< limit);
            json_string_ += iter->second;
            json_string_ += " ";
        }

        if (iter->first == QUERY_SORT_FIELDS)
        {
            rapidjson::Document d;
            std::string json_string = "{ \"sort_fields\" : " +
                iter->second + " }";
            json_string_ += json_string;
            json_string_ += " ";

            d.Parse<0>(const_cast<char *>(json_string.c_str()));
            const rapidjson::Value& json_sort_fields =
                d["sort_fields"];
            QE_PARSE_ERROR(json_sort_fields.IsArray());
            QE_TRACE(DEBUG, "# of sort fields:"<< json_sort_fields.Size());
            for (rapidjson::SizeType i = 0; i<json_sort_fields.Size(); i++)
            {
                QE_PARSE_ERROR(json_sort_fields[i].IsString());
                std::string sort_str(json_sort_fields[i].GetString());
                QE_TRACE(DEBUG, sort_str);
                std::string datatype(m_query->get_column_field_datatype(sort_str));
                if (!m_query->is_stat_table_query()) {
                    QE_INVALIDARG_ERROR(datatype != std::string(""));
                } else if (m_query->stats().is_stat_table_static()) {
                    // This is a static StatTable. We can check the schema
                    QE_INVALIDARG_ERROR(datatype != std::string(""));
                }
                QE_INVALIDARG_ERROR(
                    m_query->is_valid_sort_field(sort_str) != false);
                sort_field_t sort_field(get_column_name(sort_str), datatype);
                sort_fields.push_back(sort_field);
            }
        }

        /*
         * old filter style is just list of expr ANDed
         * new filter are list of ANDs over OR
         * both modes are supported with the below code
         */
        if (iter->first == QUERY_FILTER) {
            rapidjson::Document d;
            std::string json_string = "{ \"filter\" : " +
                iter->second + " }";
            json_string_ += json_string;
            json_string_ += " ";

            d.Parse<0>(const_cast<char *>(json_string.c_str()));
            const rapidjson::Value& json_filters =
                d["filter"];
            QE_PARSE_ERROR(json_filters.IsArray());
            QE_TRACE(DEBUG, "# of filters:"<< json_filters.Size());
            bool single_list = false;
            if (json_filters.Size()) {
                rapidjson::SizeType zeroth = 0;
                const rapidjson::Value& json_filters_0 = json_filters[zeroth];
                if (!json_filters_0.IsArray()) {
                    single_list = true;
                }
            }

            if (single_list) {
                //parse the old format
                std::vector<filter_match_t> filter_and;
                for (rapidjson::SizeType j = 0; j<json_filters.Size(); j++)
                  {
                    filter_match_t filter;

                    const rapidjson::Value& name_value =
                        json_filters[j][WHERE_MATCH_NAME];
                    const rapidjson::Value&  value_value =
                        json_filters[j][WHERE_MATCH_VALUE];
                    const rapidjson::Value& op_value =
                        json_filters[j][WHERE_MATCH_OP];

                    // do some validation checks
                    QE_INVALIDARG_ERROR(name_value.IsString());
                    QE_INVALIDARG_ERROR
                        ((value_value.IsString() || value_value.IsNumber()));
                    QE_INVALIDARG_ERROR(op_value.IsNumber());

                    filter.name = name_value.GetString();
                    filter.op = (match_op)op_value.GetInt();

                    // extract value after type conversion
                      {
                        if (value_value.IsString())
                          {
                            filter.value = value_value.GetString();
                          } else if (value_value.IsInt()){
                              int int_value;
                              std::ostringstream convert;
                              int_value = value_value.GetInt();
                              convert << int_value;
                              filter.value = convert.str();
                          } else if (value_value.IsUint()) {
                              uint32_t uint_value;
                              std::ostringstream convert;
                              uint_value = value_value.GetUint();
                              convert << uint_value;
                              filter.value = convert.str();
                          }
                      }

                    if (filter.op == REGEX_MATCH)
                      {
                        // compile regex beforehand
                        filter.match_e = boost::regex(filter.value);
                      }

                    filter_and.push_back(filter);
                  }
                filter_list.push_back(filter_and);
            } else {
                //new OR of ANDs
                for (rapidjson::SizeType j = 0; j<json_filters.Size(); j++) {
                    std::vector<filter_match_t> filter_and;
                    const rapidjson::Value& json_filter_and = json_filters[j];
                    QE_PARSE_ERROR(json_filter_and.IsArray());

                    for (rapidjson::SizeType k = 0; k<json_filter_and.Size(); k++) {
                        filter_match_t filter;

                        const rapidjson::Value& name_value =
                            json_filter_and[k][WHERE_MATCH_NAME];
                        const rapidjson::Value&  value_value =
                            json_filter_and[k][WHERE_MATCH_VALUE];
                        const rapidjson::Value& op_value =
                            json_filter_and[k][WHERE_MATCH_OP];

                        // do some validation checks
                        QE_INVALIDARG_ERROR(name_value.IsString());
                        QE_INVALIDARG_ERROR
                            ((value_value.IsString() || value_value.IsNumber()));
                        QE_INVALIDARG_ERROR(op_value.IsNumber());

                        filter.name = name_value.GetString();
                        filter.op = (match_op)op_value.GetInt();

                        // extract value after type conversion
                        if (value_value.IsString()) {
                            filter.value = value_value.GetString();
                        } else if (value_value.IsInt()) {
                            int int_value;
                            std::ostringstream convert;
                            int_value = value_value.GetInt();
                            convert << int_value;
                            filter.value = convert.str();
                        } else if (value_value.IsUint()) {
                            uint32_t uint_value;
                            std::ostringstream convert;
                            uint_value = value_value.GetUint();
                            convert << uint_value;
                            filter.value = convert.str();
                        }

                        if (filter.op == REGEX_MATCH) {
                            // compile regex beforehand
                            filter.match_e = boost::regex(filter.value);
                        }

                        filter_and.push_back(filter);
                    }
                    filter_list.push_back(filter_and);
                }
            }
        }
    }

    // add filter to filter query engine logs if requested
    if ((((AnalyticsQuery *)main_query)->filter_qe_logs) &&
            (((AnalyticsQuery *)main_query)->table() ==
             g_viz_constants.COLLECTOR_GLOBAL_TABLE)) {
        QE_TRACE(DEBUG,  " Adding filter for QE logs");
        filter_match_t filter;
        filter.name = g_viz_constants.MODULE;
        filter.value =
            ((AnalyticsQuery *)main_query)->sandesh_moduleid;
        filter.op = NOT_EQUAL;
        filter.ignore_col_absence = true;
        if (!filter_list.size()) {
            std::vector<filter_match_t> filter_and;
            filter_and.push_back(filter);
            filter_list.push_back(filter_and);
        } else {
            for (unsigned int i = 0; i < filter_list.size(); i++) {
                filter_list[i].push_back(filter);
            }
        }
    }

    // If the user has specified the sorting field and not the sorting order,
    // then sort the result in ascending order.
    if (sort_fields.size() && sorted == false) {
        sorted = true;
        sorting_type = ASCENDING;
    }
}

bool AnalyticsQuery::merge_processing(
    const QEOpServerProxy::BufferT& input,
    QEOpServerProxy::BufferT& output) {

    if (status_details != 0)
    {
        QE_TRACE(DEBUG,
             "No need to process query, as there were errors previously");
        return false;
    }

    // Have the result ready and processing is done
    status_details = 0;
    return postprocess_->merge_processing(input, output);
}

bool AnalyticsQuery::final_merge_processing(
const std::vector<boost::shared_ptr<QEOpServerProxy::BufferT> >& inputs,
    QEOpServerProxy::BufferT& output) {

    if (status_details != 0)
    {
        QE_TRACE(DEBUG,
             "No need to process query, as there were errors previously");
        return false;
    }

    // Have the result ready and processing is done
    status_details = 0;
    return postprocess_->final_merge_processing(inputs, output);
}

// this is to get parallelization details once the query is parsed
void AnalyticsQuery::get_query_details(bool& is_merge_needed, bool& is_map_output,
        std::vector<uint64_t>& chunk_sizes,
        std::string& where,
        std::string& select,
        std::string& post,
        uint64_t& time_period,
        int& parse_status)
{
    QE_TRACE(DEBUG, "time_slice is " << time_slice);
    if (status_details == 0)
    {
        for (uint64_t chunk_start = original_from_time;
                chunk_start < original_end_time; chunk_start += time_slice)
        {
            if ((chunk_start+time_slice) <= original_end_time) {
                chunk_sizes.push_back(time_slice);
            } else {
                chunk_sizes.push_back((original_end_time - chunk_start));
            }
        }
    } else {
        chunk_sizes.push_back(0); // just return some dummy value
    }

    time_period = (end_time_ - from_time_) / 1000000;

    parse_status = status_details;
    if (parse_status != 0) return;

    if (is_stat_table_query()) {
        is_merge_needed = selectquery_->stats_->IsMergeNeeded();
    } else {
        is_merge_needed = merge_needed;
    }

    where = wherequery_->json_string_;
    select = selectquery_->json_string_;
    post = postprocess_->json_string_;
    is_map_output = is_stat_table_query();
}

bool AnalyticsQuery::can_parallelize_query() {
    parallelize_query_ = true;
    if (table_ == g_viz_constants.OBJECT_VALUE_TABLE) {
        parallelize_query_ = false;
    }
    return parallelize_query_;
}

void AnalyticsQuery::Init(GenDb::GenDbIf *db_if, std::string qid,
    std::map<std::string, std::string>& json_api_data,
    uint64_t analytics_start_time)
{
    std::map<std::string, std::string>::iterator iter;

    QE_TRACE(DEBUG, __func__);

    // populate fields
    query_id = qid;

    // Initialize database
    query_result_unit_t::dbif = db_if;
    dbif = db_if;
    QE_IO_ERROR(dbif != NULL)

    sandesh_moduleid =
        g_vns_constants.ModuleNames.find(Module::QUERY_ENGINE)->second;
    {
        std::stringstream json_string; json_string << " { ";
        for (std::map<std::string,
                std::string>::iterator it = json_api_data_.begin();
                it != json_api_data_.end(); it++) {
            json_string <<
                ((it != json_api_data_.begin())? " , " : "") <<
                it->first << ": " << it->second;
        }
        json_string << " } ";
        QE_LOG_GLOBAL(DEBUG, "json query is: " << json_string.str());
    }

    // parse JSON query
    // FROM field
    {
        iter = json_api_data.find(QUERY_TABLE);
        QE_PARSE_ERROR(iter != json_api_data.end());

        //strip " from the passed string
        table_ = iter->second.substr(1, iter->second.size()-2);

        // boost::to_upper(table);
        QE_TRACE(DEBUG,  " table is " << table_);
        if (StatsQuery::is_stat_table_query(table_)) {
            stats_.reset(new StatsQuery(table_));
        }
        QE_INVALIDARG_ERROR(is_valid_from_field(table_));
    }

    // Start time
    {
        iter = json_api_data.find(QUERY_START_TIME);
        QE_PARSE_ERROR(iter != json_api_data.end());
        req_from_time_ = parse_time(iter->second);
        QE_TRACE(DEBUG,  " from_time is " << req_from_time_);
        if (req_from_time_ < analytics_start_time)
        {
            from_time_ = analytics_start_time;
            QE_TRACE(DEBUG, "updated start_time to:" << from_time_);
        } else {
            from_time_ = req_from_time_;
        }
    }

    // End time
    {
        struct timeval curr_time;
        gettimeofday(&curr_time, NULL);

        iter = json_api_data.find(QUERY_END_TIME);
        QE_PARSE_ERROR(iter != json_api_data.end());
        req_end_time_ = parse_time(iter->second);
        QE_TRACE(DEBUG,  " end_time is " << req_end_time_);

        if (req_end_time_ < analytics_start_time) {
            end_time_ = analytics_start_time;
        } else if (req_end_time_ >
                (uint64_t)(curr_time.tv_sec*1000000+curr_time.tv_usec)) {
            end_time_ = curr_time.tv_sec*1000000+curr_time.tv_usec;
            QE_TRACE(DEBUG, "updated end_time to:" << end_time_);
        } else {
            end_time_ = req_end_time_;
        }
    }

    // Initialize SELECT/WHERE/Post-Processing components of query
    // for input validation

    // where processing initialization
    std::string where_json_string;
    {
        int direction = INGRESS;
        iter = json_api_data.find(QUERY_FLOW_DIR);
        if (iter != json_api_data.end()) {
            std::istringstream(iter->second) >> direction;
            QE_TRACE(DEBUG,  "set flow direction to:" << direction);
        }

        iter = json_api_data.find(QUERY_WHERE);
        if (iter == json_api_data.end())
        {
            QE_TRACE(DEBUG, "Where * query");
            where_json_string = std::string("");
        } else {
            where_json_string = iter->second;
        }

        QE_TRACE(DEBUG,  " Initializing Where Query");
        wherequery_ = new WhereQuery(where_json_string, direction, this);
        this->status_details = wherequery_->status_details;
        if (this->status_details != 0 )
        {
            QE_LOG_GLOBAL(DEBUG, "Error in WHERE parsing");
            return;
        }
    }

    // select processing initialization
    {
        QE_TRACE(DEBUG,  " Initializing Select Query");
        selectquery_ = new SelectQuery(this, json_api_data);
        this->status_details = selectquery_->status_details;
        if (this->status_details != 0 )
        {
            QE_LOG_GLOBAL(DEBUG, "Error in SELECT parsing");
            return;
        }
        /*
         * ObjectId queries are special, they are requested from Object* tables,
         * but the values are extrated from g_viz_constants.OBJECT_VALUE_TABLE
         */
        if (is_object_table_query()) {
            if (selectquery_->ObjectIdQuery()) {
                object_value_key = table_;
                table_ = g_viz_constants.OBJECT_VALUE_TABLE;
            }
        }
    }

    // post processing initialization
    QE_TRACE(DEBUG,  " Initializing PostProcessing Query");
    postprocess_ = new PostProcessingQuery(json_api_data, this);
    this->status_details = postprocess_->status_details;
    if (this->status_details != 0 )
    {
        QE_LOG_GLOBAL(DEBUG, "Error in PostProcess parsing");
        return;
    }

    if (this->is_stat_table_query()) {
        selectquery_->stats_->SetSortOrder(postprocess_->sort_fields);
    }

    // just to take care of issues with Analytics start time
         if (from_time_ > end_time_)
            from_time_ = end_time_ - 1;

    // Get the right job slice for parallelization
    original_from_time = from_time_;
    original_end_time = end_time_;

    if (can_parallelize_query()) {
        uint64_t smax = pow(2,g_viz_constants.RowTimeInBits) * \
              QueryEngine::max_slice_;

        time_slice = ((end_time_ - from_time_)/total_parallel_batches) + 1;

        if (time_slice < (uint64_t)pow(2,g_viz_constants.RowTimeInBits)) {
            time_slice = pow(2,g_viz_constants.RowTimeInBits);
        }
        if (time_slice > smax) {
            time_slice = smax;
        }
        // Adjust the time_slice for Flowseries query, if time granularity is
        // specified. Divide the query based on the time granularity.
        if (selectquery_->provide_timeseries && selectquery_->granularity) {
            if (selectquery_->granularity >= time_slice) {
                time_slice = selectquery_->granularity;
            } else {
                time_slice = ((time_slice/selectquery_->granularity)+1)*
                    selectquery_->granularity;
            }
        }

        uint8_t fs_query_type = selectquery_->flowseries_query_type();
        if ((table() == g_viz_constants.FLOW_TABLE) ||
            (table() == g_viz_constants.FLOW_SERIES_TABLE &&
             (fs_query_type == SelectQuery::FS_SELECT_STATS ||
              fs_query_type == SelectQuery::FS_SELECT_FLOW_TUPLE_STATS))) {
            merge_needed = true;
        }

        QE_TRACE(DEBUG, "time_slice:" << time_slice << " , # of parallel "
                "batches:" << total_parallel_batches);

    } else {
        // No parallelization
        QE_LOG_GLOBAL(DEBUG, "No parallelization for this query");
        merge_needed = false;
        parallelize_query_ = false;
        time_slice = end_time_ - from_time_;
    }

    from_time_ =
        original_from_time + time_slice*parallel_batch_num;
    end_time_ = from_time_ + time_slice;
    if (from_time_ >= original_end_time)
    {
        processing_needed = false;
    } else if (end_time_ > original_end_time) {
        end_time_ = original_end_time;
    }

    if (processing_needed)
    {
        // change it to trace later TBD
        QE_TRACE(DEBUG, "For batch:" << parallel_batch_num << " from_time:" << from_time_ << " end_time:" << end_time_ << " time slice:" << time_slice);
    } else {
        QE_TRACE(DEBUG, "No processing needed for batch:" << parallel_batch_num);
    }

}

/*
 * time could be in now +/-10m/h/s format.Parse that and return th
 * UTC corresponding to the parsed val
 */
uint64_t AnalyticsQuery::parse_time(const std::string& relative_time)
{
    uint64_t offset_usec = 0;
    std::string temp;
    if (!relative_time.compare("\"now\"")) {
        return UTCTimestampUsec();
    } else if (!(relative_time.substr(1,3)).compare("now")) {
        //Find the offset to be shifted
        int found = 0;
    //Extract any number after now
        if ((found = relative_time.find_last_of("h")) > 0) {
            std::istringstream(relative_time.substr(5,found-4)) >> offset_usec;
            offset_usec = offset_usec*3600*1000000;
        } else if ((found = relative_time.find_last_of("m")) > 0) {
            std::istringstream(relative_time.substr(5,found-4)) >> offset_usec;
            offset_usec = offset_usec*60*1000000;
        } else if ((found = relative_time.find_last_of("s")) > 0) {
            std::istringstream(relative_time.substr(5,found-4)) >> offset_usec;
            offset_usec = offset_usec*1000000;
        } else {
            QE_LOG_GLOBAL(DEBUG, "Error in time parsing.h/m/s expected");
            return 0;
        }

        //If now+ return UTC + offset else UTC - offset
        if (!(relative_time.substr(1,4)).compare("now+")) {
            return (UTCTimestampUsec() + offset_usec);
        } else if (!(relative_time.substr(1,4)).compare("now-")) {
            return (UTCTimestampUsec() - offset_usec);
        } else {
            QE_LOG_GLOBAL(DEBUG, "Error in time parsing. now+/-expected");
            return 0;
        }
    } else {
        //To handle old version of input where integer is parsed
        std::istringstream(relative_time) >> offset_usec;
        return offset_usec;
    }
}

QueryUnit::QueryUnit(QueryUnit *p_query, QueryUnit *m_query):
    parent_query(p_query), main_query(m_query), pending_subqueries(0),
    query_status(QUERY_PROCESSING_NOT_STARTED), status_details(0)
{
    if (p_query)
        p_query->sub_queries.push_back(this);
};

QueryUnit::~QueryUnit()
{
    int num_sub_queries = sub_queries.size();
    for(int i = 0; i<num_sub_queries; i++)
        delete sub_queries[i];
}

// Get UUID from the info field
void query_result_unit_t::get_uuid(boost::uuids::uuid& u)
{
    try {
        u = boost::get<boost::uuids::uuid>(info.at(0));
    } catch (boost::bad_get& ex) {
        QE_ASSERT(0);
    }
}

// Get UUID and stats
void query_result_unit_t::get_uuid_stats(boost::uuids::uuid& u,
        flow_stats& stats)
{
    try {
        stats.bytes = boost::get<uint64_t>(info.at(0));
    } catch (boost::bad_get& ex) {
        QE_ASSERT(0);
    }

    try {
        stats.pkts = boost::get<uint64_t>(info.at(1));
    } catch (boost::bad_get& ex) {
        QE_ASSERT(0);
    }

    try {
        stats.short_flow = (boost::get<uint8_t>(info.at(2)) == 1)? true : false;
    } catch (boost::bad_get& ex) {
        QE_ASSERT(0);
    }

    try {
        u = boost::get<boost::uuids::uuid>(info.at(3));
    } catch (boost::bad_get& ex) {
        QE_ASSERT(0);
    }

    return;
}

void query_result_unit_t::set_stattable_info(
        const std::string& attribstr,
        const boost::uuids::uuid& uuid) {
    info.push_back(attribstr);
    info.push_back(uuid);
}

void  query_result_unit_t::get_stattable_info(
            std::string& attribstr,
            boost::uuids::uuid& uuid) {

    int index = 0;

    try {
        attribstr = boost::get<std::string>(info.at(index++));
    } catch (boost::bad_get& ex) {
        QE_ASSERT(0);
    } catch (const std::out_of_range& oor) {
        QE_ASSERT(0);
    }

    try {
        uuid = boost::get<boost::uuids::uuid>(info.at(index++));
    } catch (boost::bad_get& ex) {
        QE_ASSERT(0);
    } catch (const std::out_of_range& oor) {
        QE_ASSERT(0);
    }

}

// Get UUID and stats and 8-tuple
void query_result_unit_t::get_uuid_stats_8tuple(boost::uuids::uuid& u,
       flow_stats& stats, flow_tuple& tuple)
{
    int index = 0;
    try {
        stats.bytes = boost::get<uint64_t>(info.at(index++));
    } catch (boost::bad_get& ex) {
        QE_ASSERT(0);
    } catch (const std::out_of_range& oor) {
        QE_ASSERT(0);
    }

    try {
        stats.pkts = boost::get<uint64_t>(info.at(index++));
    } catch (boost::bad_get& ex) {
        QE_ASSERT(0);
    } catch (const std::out_of_range& oor) {
        QE_ASSERT(0);
    }

    try {
        stats.short_flow = (boost::get<uint8_t>(info.at(index++)) == 1? true : false);
    } catch (boost::bad_get& ex) {
        QE_ASSERT(0);
    } catch (const std::out_of_range& oor) {
        QE_ASSERT(0);
    }

    try {
        u = boost::get<boost::uuids::uuid>(info.at(index++));
    } catch (boost::bad_get& ex) {
        QE_ASSERT(0);
    } catch (const std::out_of_range& oor) {
        QE_ASSERT(0);
    }

    try {
        tuple.vrouter = boost::get<std::string>(info.at(index++));
    } catch (boost::bad_get& ex) {
        QE_ASSERT(0);
    } catch (const std::out_of_range& oor) {
        QE_ASSERT(0);
    }

    try {
        tuple.source_vn = boost::get<std::string>(info.at(index++));
    } catch (boost::bad_get& ex) {
        QE_ASSERT(0);
    } catch (const std::out_of_range& oor) {
        QE_ASSERT(0);
    }

    try {
        tuple.dest_vn = boost::get<std::string>(info.at(index++));
    } catch (boost::bad_get& ex) {
        QE_ASSERT(0);
    } catch (const std::out_of_range& oor) {
        QE_ASSERT(0);
    }
    try {
        tuple.source_ip = boost::get<uint32_t>(info.at(index++));
    } catch (boost::bad_get& ex) {
        QE_ASSERT(0);
    } catch (const std::out_of_range& oor) {
        QE_ASSERT(0);
    }
    try {
        tuple.dest_ip = boost::get<uint32_t>(info.at(index++));
    } catch (boost::bad_get& ex) {
        QE_ASSERT(0);
    } catch (const std::out_of_range& oor) {
        QE_ASSERT(0);
    }
    try {
        tuple.protocol = boost::get<uint8_t>(info.at(index++));
    } catch (boost::bad_get& ex) {
        QE_ASSERT(0);
    } catch (const std::out_of_range& oor) {
        QE_ASSERT(0);
    }
    try {
        tuple.source_port = boost::get<uint16_t>(info.at(index++));
    } catch (boost::bad_get& ex) {
        QE_ASSERT(0);
    } catch (const std::out_of_range& oor) {
        QE_ASSERT(0);
    }
    try {
        tuple.dest_port = boost::get<uint16_t>(info.at(index++));
    } catch (boost::bad_get& ex) {
        QE_ASSERT(0);
    } catch (const std::out_of_range& oor) {
        QE_ASSERT(0);
    }
    return;
}

query_status_t AnalyticsQuery::process_query()
{
    if (status_details != 0)
    {
        QE_TRACE(DEBUG,
             "No need to process query, as there were errors previously");
        return QUERY_FAILURE;
    }

    QE_TRACE(DEBUG, "Start Where Query Processing");
    where_start_ = UTCTimestampUsec();
    query_status = wherequery_->process_query();
    qperf_.chunk_where_time =
            static_cast<uint32_t>((UTCTimestampUsec() - where_start_)/1000);

    status_details = wherequery_->status_details;
    if (query_status != QUERY_SUCCESS)
    {
        QE_LOG(DEBUG, "where processing failed with error:"<< query_status);
        return query_status;
    }
    QE_TRACE(DEBUG, "End Where Query Processing");

    QE_TRACE(DEBUG, "Start Select Processing");
    select_start_ = UTCTimestampUsec();
    query_status = selectquery_->process_query();
    status_details = selectquery_->status_details;
    qperf_.chunk_select_time =
            static_cast<uint32_t>((UTCTimestampUsec() - select_start_)/1000);

    if (query_status != QUERY_SUCCESS)
    {
        QE_LOG(DEBUG,
                "select processing failed with error:"<< query_status);
        return query_status;
    }
    QE_TRACE(DEBUG, "End Select Processing. row #s:" <<
            selectquery_->result_->size());

    QE_TRACE(DEBUG, "Start PostProcessing");
    postproc_start_ = UTCTimestampUsec();
    query_status = postprocess_->process_query();
    status_details = postprocess_->status_details;
    qperf_.chunk_postproc_time =
            static_cast<uint32_t>((UTCTimestampUsec() - postproc_start_)/1000);

    final_result = postprocess_->result_;
    final_mresult = postprocess_->mresult_;
    if (query_status != QUERY_SUCCESS)
    {
        QE_LOG(DEBUG,
                "post processing failed with error:"<< query_status);
        return query_status;
    }
    QE_TRACE(DEBUG, "End PostProcessing. row #s:" <<
            final_result->size());

    return QUERY_SUCCESS;
}

AnalyticsQuery::AnalyticsQuery(std::string qid, std::map<std::string,
        std::string>& json_api_data, uint64_t analytics_start_time,
        EventManager *evm, std::vector<std::string> cassandra_ips,
        std::vector<int> cassandra_ports, int batch,
        int total_batches):
        QueryUnit(NULL, this),
        dbif_(GenDb::GenDbIf::GenDbIfImpl(
            boost::bind(&AnalyticsQuery::db_err_handler, this),
            cassandra_ips, cassandra_ports, 0, "QueryEngine", true)),
        filter_qe_logs(true),
        json_api_data_(json_api_data),
        where_start_(0),
        select_start_(0),
        postproc_start_(0),
        merge_needed(false),
        parallel_batch_num(batch),
        total_parallel_batches(total_batches),
        processing_needed(true),
        stats_(NULL)
{
    // Need to do this for logging/tracing with query ids
    query_id = qid;

    QE_TRACE(DEBUG, __func__);

    // Initialize database connection
    QE_TRACE(DEBUG, "Initializing database");
    dbif = dbif_.get();

    boost::system::error_code ec;
    if (!dbif->Db_Init("qe::DbHandler", -1)) {
        QE_LOG(ERROR, "Database initialization failed");
        this->status_details = EIO;
    }

    if (!dbif->Db_SetTablespace(g_viz_constants.COLLECTOR_KEYSPACE)) {
        QE_LOG(ERROR,  ": Create/Set KEYSPACE: " <<
           g_viz_constants.COLLECTOR_KEYSPACE << " FAILED");
        this->status_details = EIO;
    }
    for (std::vector<GenDb::NewCf>::const_iterator it = vizd_tables.begin();
            it != vizd_tables.end(); it++) {
        if (!dbif_->Db_UseColumnfamily(*it)) {
            QE_LOG(ERROR, "Database initialization:Db_UseColumnfamily failed");
            this->status_details = EIO;
        }
    }
    for (std::vector<GenDb::NewCf>::const_iterator it = vizd_flow_tables.begin();
            it != vizd_flow_tables.end(); it++) {
        if (!dbif_->Db_UseColumnfamily(*it)) {
            QE_LOG(ERROR, "Database initialization:Db_UseColumnfamily failed");
            this->status_details = EIO;
        }
    }
    for (std::vector<GenDb::NewCf>::const_iterator it = vizd_stat_tables.begin();
            it != vizd_stat_tables.end(); it++) {
        if (!dbif_->Db_UseColumnfamily(*it)) {
            QE_LOG(ERROR, "Database initialization:Db_UseColumnfamily failed");
            this->status_details = EIO;
        }
    }
    boost::asio::ip::address db_addr(boost::asio::ip::address::from_string(
        dbif->Db_GetHost(), ec));
    boost::asio::ip::tcp::endpoint db_endpoint(db_addr, dbif->Db_GetPort());
    if (this->status_details != 0) {
        // Update connection info
        ConnectionState::GetInstance()->Update(ConnectionType::DATABASE,
            std::string(), ConnectionStatus::DOWN, db_endpoint, std::string());
    } else {
        // Update connection info
        ConnectionState::GetInstance()->Update(ConnectionType::DATABASE,
            std::string(), ConnectionStatus::UP, db_endpoint, std::string());
    }
    dbif->Db_SetInitDone(true);
    Init(dbif, qid, json_api_data, analytics_start_time);
}

AnalyticsQuery::AnalyticsQuery(std::string qid, GenDb::GenDbIf *dbif,
    std::map<std::string, std::string> json_api_data,
    uint64_t analytics_start_time, int batch, int total_batches) :
    QueryUnit(NULL, this),
    dbif_(dbif),
    query_id(qid),
    json_api_data_(json_api_data),
    where_start_(0),
    select_start_(0),
    postproc_start_(0),
    merge_needed(false),
    parallel_batch_num(batch),
    total_parallel_batches(total_batches),
    processing_needed(true),
    stats_(NULL) {
    Init(dbif, qid, json_api_data, analytics_start_time);
}

QueryEngine::QueryEngine(EventManager *evm,
            const std::string & redis_ip, unsigned short redis_port,
            int max_tasks, int max_slice, uint64_t anal_ttl) :
        qosp_(new QEOpServerProxy(evm,
            this, redis_ip, redis_port, max_tasks)),
        evm_(evm),
        cassandra_ports_(0)
{
    max_slice_ =  max_slice;
    init_vizd_tables();

    // Initialize database connection
    QE_LOG_NOQID(DEBUG, "Initializing QE without database!");

    uint64_t curr_time = UTCTimestampUsec();
    QE_LOG_NOQID(DEBUG, "Could not find analytics start time");
    uint64_t ttl = anal_ttl*60*60*1000000;
    stime = curr_time - ttl;
    QE_LOG_NOQID(DEBUG, "set stime to " << stime << "and AnalyticsTTL to " << g_viz_constants.AnalyticsTTL);
}

QueryEngine::QueryEngine(EventManager *evm,
            std::vector<std::string> cassandra_ips,
            std::vector<int> cassandra_ports,
            const std::string & redis_ip, unsigned short redis_port,
            int max_tasks, int max_slice, uint64_t anal_ttl,
            uint64_t start_time) :
        dbif_(GenDb::GenDbIf::GenDbIfImpl(
            boost::bind(&QueryEngine::db_err_handler, this),
            cassandra_ips, cassandra_ports, 0, "QueryEngine", true)),
        qosp_(new QEOpServerProxy(evm,
            this, redis_ip, redis_port, max_tasks)),
        evm_(evm),
        cassandra_ports_(cassandra_ports),
        cassandra_ips_(cassandra_ips)
{
    max_slice_ = max_slice;
    init_vizd_tables();

    // Initialize database connection
    QE_TRACE_NOQID(DEBUG, "Initializing database");
    GenDb::GenDbIf *db_if = dbif_.get();

    boost::system::error_code ec;
    int retries = 0;
    bool retry = true;
    while (retry == true) {
        retry = false;

        if (!db_if->Db_Init("qe::DbHandler", -1)) {
            QE_LOG_NOQID(ERROR, "Database initialization failed");
            retry = true;
        }

        if (!retry) {
            if (!db_if->Db_SetTablespace(g_viz_constants.COLLECTOR_KEYSPACE)) {
                QE_LOG_NOQID(ERROR,  ": Create/Set KEYSPACE: " <<
                        g_viz_constants.COLLECTOR_KEYSPACE << " FAILED");
                retry = true;
            }
        }

        if (!retry) {
            for (std::vector<GenDb::NewCf>::const_iterator it = vizd_tables.begin();
                    it != vizd_tables.end(); it++) {
                if (!dbif_->Db_UseColumnfamily(*it)) {
                    retry = true;
                    break;
                }
            }
        }

        if (!retry) {
            for (std::vector<GenDb::NewCf>::const_iterator it = vizd_flow_tables.begin();
                    it != vizd_flow_tables.end(); it++) {
                if (!dbif_->Db_UseColumnfamily(*it)) {
                    retry = true;
                    break;
                }
            }
        }
        if (retry) {
            std::stringstream ss;
            ss << "initialization of database failed. retrying " << retries++ << " time";
            // Update connection info
            boost::asio::ip::address db_addr(boost::asio::ip::address::from_string(
                db_if->Db_GetHost(), ec));
            boost::asio::ip::tcp::endpoint db_endpoint(db_addr, db_if->Db_GetPort());
            ConnectionState::GetInstance()->Update(ConnectionType::DATABASE,
                std::string(), ConnectionStatus::DOWN, db_endpoint, std::string());
            Q_E_LOG_LOG("QeInit", SandeshLevel::SYS_WARN, ss.str());
            db_if->Db_Uninit("qe::DbHandler", -1);
            sleep(5);
        }
    }
    if (start_time != 0) {
        stime = start_time;
    } else {
        bool init_done = false;
        retries = 0;
        while (!init_done && retries < 5) {
            GenDb::ColList col_list;
            std::string cfname = g_viz_constants.SYSTEM_OBJECT_TABLE;
            GenDb::DbDataValueVec key;
            key.push_back(g_viz_constants.SYSTEM_OBJECT_ANALYTICS);

            if (dbif_->Db_GetRow(col_list, cfname, key)) {
                for (GenDb::NewColVec::iterator it = col_list.columns_.begin();
                        it != col_list.columns_.end(); it++) {
                    std::string col_name;
                    try {
                        col_name = boost::get<std::string>(it->name->at(0));
                    } catch (boost::bad_get& ex) {
                        QE_LOG_NOQID(ERROR, __func__ << ": Exception on col_name get");
                    }

                    if (col_name == g_viz_constants.SYSTEM_OBJECT_START_TIME) {
                        try {
                            stime = boost::get<uint64_t>(it->value->at(0));
                            init_done = true;
                        } catch (boost::bad_get& ex) {
                            QE_LOG_NOQID(ERROR, __func__ << "Exception for boost::get, what=" << ex.what());
                            break;
                        }
                    }
                }
            }
            retries++;
            if (!init_done)
                sleep(5);
        }
        if (!init_done) {
            uint64_t ttl = anal_ttl*60*60*1000000;
            uint64_t curr_time = UTCTimestampUsec();
            stime = curr_time - ttl;
            QE_LOG_NOQID(ERROR, __func__ << "setting start_time manually to" << stime);
        }
    }
    dbif_->Db_SetInitDone(true);
    // Update connection info
    boost::asio::ip::address db_addr(boost::asio::ip::address::from_string(
        db_if->Db_GetHost(), ec));
    boost::asio::ip::tcp::endpoint db_endpoint(db_addr, db_if->Db_GetPort());
    ConnectionState::GetInstance()->Update(ConnectionType::DATABASE,
        std::string(), ConnectionStatus::UP, db_endpoint, std::string());
}

using std::vector;

int
QueryEngine::QueryPrepare(QueryParams qp,
        std::vector<uint64_t> &chunk_size,
        bool & need_merge, bool & map_output,
        std::string& where, std::string& select, std::string& post,
        uint64_t& time_period,
        std::string &table) {
    string& qid = qp.qid;
    QE_LOG_NOQID(INFO,
             " Got Query to prepare for QID " << qid);
    int ret_code;
    if (cassandra_ports_.size() == 1 && cassandra_ports_[0] == 0) {
        chunk_size.push_back(999);
        need_merge = false;
        map_output = false;
        ret_code = 0;
        table = string("ObjectCollectorInfo");
    } else {

        AnalyticsQuery *q = new AnalyticsQuery(qid, qp.terms, stime, evm_,
                cassandra_ips_, cassandra_ports_, 0, qp.maxChunks);
        chunk_size.clear();
        q->get_query_details(need_merge, map_output, chunk_size,
            where, select, post, time_period, ret_code);
        table = q->table();
        delete q;
    }
    return ret_code;
}

bool
QueryEngine::QueryAccumulate(QueryParams qp,
        const QEOpServerProxy::BufferT& input,
        QEOpServerProxy::BufferT& output) {

    QE_TRACE_NOQID(DEBUG, "Creating analytics query object for merge_processing");
    AnalyticsQuery *q = new AnalyticsQuery(qp.qid, qp.terms, stime, evm_,
        cassandra_ips_, cassandra_ports_, 1, qp.maxChunks);
    QE_TRACE_NOQID(DEBUG, "Calling merge_processing");
    bool ret = q->merge_processing(input, output);
    delete q;
    return ret;
}

bool
QueryEngine::QueryFinalMerge(QueryParams qp,
        const std::vector<boost::shared_ptr<QEOpServerProxy::BufferT> >& inputs,
        QEOpServerProxy::BufferT& output) {

    QE_TRACE_NOQID(DEBUG, "Creating analytics query object for final_merge_processing");
    AnalyticsQuery *q = new AnalyticsQuery(qp.qid, qp.terms, stime, evm_,
        cassandra_ips_, cassandra_ports_, 1, qp.maxChunks);
    QE_TRACE_NOQID(DEBUG, "Calling final_merge_processing");
    bool ret = q->final_merge_processing(inputs, output);
    delete q;
    return ret;
}

bool
QueryEngine::QueryFinalMerge(QueryParams qp,
        const std::vector<boost::shared_ptr<QEOpServerProxy::OutRowMultimapT> >& inputs,
        QEOpServerProxy::OutRowMultimapT& output) {
    QE_TRACE_NOQID(DEBUG, "Creating analytics query object for final_merge_processing");
    AnalyticsQuery *q = new AnalyticsQuery(qp.qid, qp.terms, stime, evm_,
        cassandra_ips_, cassandra_ports_, 1, qp.maxChunks);

    if (!q->is_stat_table_query()) {
        QE_TRACE_NOQID(DEBUG, "MultiMap merge_final is for Stats only");
        delete q;
        return false;
    }
    QE_TRACE_NOQID(DEBUG, "Calling final_merge_processing for Stats");

    q->selectquery_->stats_->MergeFinal(inputs, output);
    delete q;
    return true;
}

bool
QueryEngine::QueryExec(void * handle, QueryParams qp, uint32_t chunk)
{
    string& qid = qp.qid;
    QE_TRACE_NOQID(DEBUG,
             " Got Query to execute for QID " << qid << " chunk:"<< chunk);
    //GenDb::GenDbIf *db_if = dbif_.get();
    if (cassandra_ports_.size() == 1 && cassandra_ports_[0] == 0) {
        std::auto_ptr<QEOpServerProxy::BufferT> final_output(new QEOpServerProxy::BufferT);
        QEOpServerProxy::OutRowT outrow = boost::assign::map_list_of(
            "MessageTS", "1368037623434740")(
            "Messagetype", "IFMapString")(
            "ModuleId", "ControlNode")(
            "Source","b1s1")(
            "ObjectLog","\n<IFMapString type=\"sandesh\"><message type=\"string\" identifier=\"1\">Cancelling Response timer.</message><file type=\"string\" identifier=\"-32768\">src/ifmap/client/ifmap_state_machine.cc</file><line type=\"i32\" identifier=\"-32767\">578</line></IFMapString>");
        QEOpServerProxy::MetadataT metadata;
        std::auto_ptr<QEOpServerProxy::OutRowMultimapT> final_moutput(new QEOpServerProxy::OutRowMultimapT);
        for (int i = 0 ; i < 100; i++)
            final_output->push_back(std::make_pair(outrow, metadata));
        QE_TRACE_NOQID(DEBUG, " Finished query processing for QID " << qid << " chunk:" << chunk);
        QEOpServerProxy::QPerfInfo qperf(0,0,0);
        qperf.error = 0;
        qosp_->QueryResult(handle, qperf, final_output, final_moutput);
        return true;
    }

    AnalyticsQuery *q = new AnalyticsQuery(qid, qp.terms, stime, evm_,
            cassandra_ips_, cassandra_ports_, chunk, qp.maxChunks);

    QE_TRACE_NOQID(DEBUG, " Finished parsing and starting processing for QID " << qid << " chunk:" << chunk);
    q->process_query();

    QE_TRACE_NOQID(DEBUG, " Finished query processing for QID " << qid << " chunk:" << chunk);
    q->qperf_.error = q->status_details;
    qosp_->QueryResult(handle, q->qperf_, q->final_result, q->final_mresult);
    delete q;
    return true;
}

std::ostream &operator<<(std::ostream &out, query_result_unit_t& res)
{
    out << "T:" << res.timestamp << " : Need to extract other information";
#if 0
    out << "T:" << res.timestamp << " : ";

    if (res.info.length() < 48) {
        boost::uuids::uuid tmp_u;
        res.get_uuid(tmp_u);
        out << " UUID:" << tmp_u;
    } else if (res.info.length() == 48) {
        boost::uuids::uuid tmp_u; flow_stats tmp_stats;
        res.get_uuid_stats(tmp_u, tmp_stats);
        out << " UUID:" << tmp_u <<
            " Bytes: " << tmp_stats.bytes <<
            " Pkts: " << tmp_stats.pkts <<
            " Short-Flow: " << tmp_stats.short_flow;
    } else if (res.info.length() > 48) {
        boost::uuids::uuid tmp_u; flow_stats tmp_stats;
            flow_tuple tmp_tuple;
        res.get_uuid_stats_8tuple(tmp_u, tmp_stats, tmp_tuple);
        out << " UUID:" << tmp_u <<
            " SVN: " << tmp_tuple.source_vn <<
            " DVN: " << tmp_tuple.dest_vn<<
            " SIP: " << tmp_tuple.source_ip <<
            " DIP: " << tmp_tuple.dest_ip <<
            " PROTO: " << tmp_tuple.protocol <<
            " SPORT: " << tmp_tuple.source_port <<
            " DPORT: " << tmp_tuple.dest_port <<
            " DIR: " << tmp_tuple.direction <<
            " Bytes: " << tmp_stats.bytes <<
            " Pkts: " << tmp_stats.pkts <<
            " Short-Flow: " << tmp_stats.short_flow;
    }
#endif

    return out;
}

// validation functions
bool AnalyticsQuery::is_object_table_query()
{
    return (
        (this->table_ != g_viz_constants.COLLECTOR_GLOBAL_TABLE) &&
        (this->table_ != g_viz_constants.FLOW_TABLE) &&
        (this->table_ != g_viz_constants.FLOW_SERIES_TABLE) &&
        (this->table_ != g_viz_constants.OBJECT_VALUE_TABLE) &&
        !is_stat_table_query());
}

bool AnalyticsQuery::is_valid_from_field(const std::string& from_field)
{
    for(size_t i = 0; i < g_viz_constants._TABLES.size(); i++)
    {
        if (g_viz_constants._TABLES[i].name == from_field)
            return true;
    }

    for (std::map<std::string, objtable_info>::const_iterator it =
            g_viz_constants._OBJECT_TABLES.begin();
            it != g_viz_constants._OBJECT_TABLES.end(); it++) {
        if (it->first == from_field)
            return true;
    }
    if (is_stat_table_query())
        return true;

    return false;
}

bool AnalyticsQuery::is_valid_where_field(const std::string& where_field)
{
    for(size_t i = 0; i < g_viz_constants._TABLES.size(); i++)
    {
        if (g_viz_constants._TABLES[i].name == table_)
        {
            for (size_t j = 0;
                j < g_viz_constants._TABLES[i].schema.columns.size(); j++)
            {
                if ((g_viz_constants._TABLES[i].schema.columns[j].name ==
                        where_field) &&
                        g_viz_constants._TABLES[i].schema.columns[j].index)
                    return true;
            }
            return false;
        }
    }
    for (std::map<std::string, objtable_info>::const_iterator it =
            g_viz_constants._OBJECT_TABLES.begin();
            it != g_viz_constants._OBJECT_TABLES.end(); it++) {
        if (it->first == table_)
        {
            for (size_t j = 0; j < g_viz_constants._OBJECT_TABLE_SCHEMA.columns.size(); j++)
            {
                if ((g_viz_constants._OBJECT_TABLE_SCHEMA.columns[j].name == where_field) &&
                        g_viz_constants._OBJECT_TABLE_SCHEMA.columns[j].index)
                    return true;
            }
            return false;
        }
    }

    if (is_stat_table_query()) {
        AnalyticsQuery *m_query = (AnalyticsQuery *)main_query;
        if (m_query->stats().is_stat_table_static()) {
            StatsQuery::column_t cdesc = m_query->stats().get_column_desc(where_field);
            if (cdesc.index) return true;
        } else {
            // For dynamic Stat Table queries, allow anything in the where clause
            return true;
        }
    }
    return false;
}

bool AnalyticsQuery::is_valid_sort_field(const std::string& sort_field) {
    if (
        (sort_field == SELECT_PACKETS) ||
        (sort_field == SELECT_BYTES) ||
        (sort_field == SELECT_SUM_PACKETS) ||
        (sort_field == SELECT_SUM_BYTES)
        )
        return true;

    return selectquery_->is_present_in_select_column_fields(sort_field);
}

std::string AnalyticsQuery::get_column_field_datatype(
                                    const std::string& column_field) {
    for(size_t i = 0; i < g_viz_constants._TABLES.size(); i++) {
        if (g_viz_constants._TABLES[i].name == table_) {
            for (size_t j = 0;
                 j < g_viz_constants._TABLES[i].schema.columns.size(); j++) {
                if (g_viz_constants._TABLES[i].schema.columns[j].name ==
                        column_field) {
                    return g_viz_constants._TABLES[i].schema.columns[j].datatype;
                }
            }
            return std::string("");
        }
    }
    for (std::map<std::string, objtable_info>::const_iterator it =
            g_viz_constants._OBJECT_TABLES.begin();
            it != g_viz_constants._OBJECT_TABLES.end(); it++) {
        if (it->first == table_) {
            for (size_t j = 0; j < g_viz_constants._OBJECT_TABLE_SCHEMA.columns.size(); j++) {
                if (g_viz_constants._OBJECT_TABLE_SCHEMA.columns[j].name == column_field) {
                    return g_viz_constants._OBJECT_TABLE_SCHEMA.columns[j].datatype;
                }
            }
            return std::string("");
        }
    }
    if (stats_.get()) {
        StatsQuery::column_t vt = stats().get_column_desc(column_field);
        if (vt.datatype == QEOpServerProxy::STRING)
            return string("string");
        else if (vt.datatype == QEOpServerProxy::UINT64)
            return string("int");
        else if (vt.datatype == QEOpServerProxy::DOUBLE)
            return string("double");
        else
            return string("");
    }
    return std::string("");
}

bool AnalyticsQuery::is_flow_query()
{
    return ((this->table_ == g_viz_constants.FLOW_SERIES_TABLE) ||
        (this->table_ == g_viz_constants.FLOW_TABLE));
}

#define UNIT_TEST_MESSAGE_FILTERS

void QueryEngine::QueryEngine_Test()
{
#if 0
    GenDb::GenDbIf *db_if = dbif_.get();

    // Create the query first
    std::string qid("TEST-QUERY");
    std::map<std::string, std::string> json_api_data;

#ifdef UNIT_TEST_MESSAGES
    QE_TRACE_NOQID(DEBUG,  " Testing messages ");
    QE_TRACE_NOQID(DEBUG,  " Parsing sample query");
        // Flow-Series Query
        json_api_data.insert(std::pair<std::string, std::string>(
                    "table", "\"MessageTable\""
        ));

        json_api_data.insert(std::pair<std::string, std::string>(
        "start_time", "1363918451328671"
        ));
        json_api_data.insert(std::pair<std::string, std::string>(
        "end_time",   "1363918651330163"
        ));
        json_api_data.insert(std::pair<std::string, std::string>(
        "where", "[[{\"name\":\"Source\", \"value\":\"127.0.0.1\", \"op\":1} , {\"name\":\"Messagetype\", \"value\":\"UveVirtualMachineConfigTrace\", \"op\":1} ]]"
        ));
        json_api_data.insert(std::pair<std::string, std::string>(
        "select_fields", "[\"Module\", \"Source\", \"Messagetype\"]"
        ));

        AnalyticsQuery *q =
            new AnalyticsQuery(db_if, qid, json_api_data, "0");
    QE_TRACE_NOQID(DEBUG,  " Parsing of messages query done");

    QE_TRACE_NOQID(DEBUG,  " Invoking messages query");
        q->process_query();
    QE_TRACE_NOQID(DEBUG,  " Processed messages query");
#endif

#ifdef UNIT_TEST_MESSAGE_FILTERS
    QE_TRACE_NOQID(DEBUG,  " Testing messages ");
    QE_TRACE_NOQID(DEBUG,  " Parsing sample query");
        // Flow-Series Query
        json_api_data.insert(std::pair<std::string, std::string>(
                    "table", "\"MessageTable\""
        ));

        json_api_data.insert(std::pair<std::string, std::string>(
        "start_time", "1365021325382585"
        ));
        json_api_data.insert(std::pair<std::string, std::string>(
        "end_time",   "1365025325382585"
        ));
        json_api_data.insert(std::pair<std::string, std::string>(
        "select_fields", "[\"ModuleId\", \"Source\", \"Level\", \"Messagetype\"]"
        ));
        json_api_data.insert(std::pair<std::string, std::string>("filter", "[{\"name\":\"Level\", \"value\":\"6\", \"op\":5}]"));

#if 0
        AnalyticsQuery *q =
            new AnalyticsQuery(db_if, qid, json_api_data, "0");
#endif
    QE_TRACE_NOQID(DEBUG,  " Parsing of messages query done");

    QE_TRACE_NOQID(DEBUG,  " Invoking messages query");
        q->process_query();
    QE_TRACE_NOQID(DEBUG, "# of rows: " << q->final_result->size());
    QE_TRACE_NOQID(DEBUG,  " Processed messages query");
#endif

#ifdef UNIT_TEST_FLOW_SERIES
    QE_TRACE_NOQID(DEBUG,  " Testing flow series");
    QE_TRACE_NOQID(DEBUG,  " Parsing sample query");
        // Flow-Series Query
        json_api_data.insert(std::pair<std::string, std::string>(
                    "table", "\"FlowSeriesTable\""
        ));

        json_api_data.insert(std::pair<std::string, std::string>(
        "start_time", "1363816971234206"
        ));
        json_api_data.insert(std::pair<std::string, std::string>(
        "end_time",   "1363816971284356"
        ));
        json_api_data.insert(std::pair<std::string, std::string>(
        "where", "[[{\"name\":\"sourcevn\", \"value\":\"default-domain:admin:vn0\", \"op\":1},{\"name\":\"destvn\", \"value\":\"default-domain:admin:vn1\", \"op\":1}],[{\"name\":\"protocol\", \"value\":\"17\", \"op\":1},{\"name\":\"dport\", \"value\":\"80\", \"op\":1}]]"
        ));
        json_api_data.insert(std::pair<std::string, std::string>(
        "select_fields", "[\"T=5\", \"sourcevn\", \"destvn\", \"sum(packets)\"]"
        ));
        json_api_data.insert(std::pair<std::string, std::string>(
        "dir", "1"
        ));

        AnalyticsQuery *q =
            new AnalyticsQuery(db_if, qid, json_api_data, "0");
    QE_TRACE_NOQID(DEBUG,  " Parsing of flow series query done");

    QE_TRACE_NOQID(DEBUG,  " Invoking flow series query");
        q->process_query();
    QE_TRACE_NOQID(DEBUG,  " Processed flow series query");
#endif
#endif

}

std::map< std::string, int > trace_enable_map;
void TraceEnable::HandleRequest() const
{
    if (get_enable())
    {
        std::string trace_type = get_TraceType();
        trace_enable_map.insert(std::make_pair(trace_type, 0));
    } else {
        trace_enable_map.erase(get_TraceType());
    }
}

std::ostream& operator<<(std::ostream& out, const flow_tuple& ft) {
    out << ft.vrouter << ":" << ft.source_vn << ":"
        << ft.dest_vn << ":" << ft.source_ip << ":"
        << ft.dest_ip << ":" << ft.protocol << ":"
        << ft.source_port << ":" << ft.dest_port << ":"
        << ft.direction;
    return out;
}


text:
#include <iostream>
#include <fstream>
#include <string>
#include <algorithm>
#include <cstring>
#include <cstdlib>

#include <boost/filesystem.hpp>

#include "dgtypes.hpp"
#include "dgexcept.hpp"
#include "dgparsers.hpp"

LANG::Enum getFileType(const std::string &fileName) {
    std::string fileExt = "";

    // Determine file type.
    for (auto c = fileName.end() - 1;
         *c != '.'  &&
         *c != '/'  &&
         *c != '\\';
        --c) {
        fileExt += *c;
        if (c == fileName.begin()) {
            // This if statement is needed to allow
            // the first character of fileName
            // to be appended to fileExt.
            break;
        }
    }
    std::reverse(fileExt.begin(), fileExt.end());

    if (fileExt == "py") {
        return LANG::PYTHON;
    } else {
        return LANG::OTHER;
    }
}

// Python functions.
Arg parseFuncArg(std::string expr) {
    Arg parsedArg;
    size_t startIndex, endIndex;

    expr.erase(0, expr.find_first_of('#') + 1);

    startIndex = 0;
    endIndex = expr.find_first_of(')', startIndex) + 1;
    parsedArg.types = expr.substr(startIndex, endIndex);

    startIndex = endIndex;
    endIndex = expr.find_first_of(':', startIndex);
    parsedArg.name = expr.substr(startIndex, endIndex - startIndex);

    if (endIndex != expr.npos) {
        // If the description for the arg is not omitted.
        startIndex = endIndex + 1;
        parsedArg.desc = expr.substr(startIndex, expr.npos - startIndex);
    } else {
        parsedArg.desc = "";
    }

    // Trim leading whitespace after delimiters.
    parsedArg.types.erase(0, parsedArg.types.find_first_not_of(' '));
    parsedArg.name.erase(0, parsedArg.name.find_first_not_of(' '));
    parsedArg.desc.erase(0, parsedArg.desc.find_first_not_of(' '));

    return parsedArg;
}

std::string parseDesc(std::string desc) {
    desc.erase(0, desc.find_first_not_of(' '));
    desc = desc.substr(1, desc.length() - 1);
    desc.erase(0, desc.find_first_not_of(' '));

    return desc;
}

inline void trimKeyword(std::string &instruction, const std::string keyword) {
    // Trim a keyword from an instruction.
    // A helper function for parsing.
    instruction.erase(0, keyword.length());
    instruction.erase(0, instruction.find_first_not_of(' '));
}
std::vector<std::string> *parseInstruction(Instruction instruction) {
    // Return a vector of tokens to make executing instructions simpler.

    // parsedInstruction dynamically allocated so that nullptr can be returned
    // if the intruction is an argument or description.
    std::vector<std::string> *parsedInstruction;
    parsedInstruction = new std::vector<std::string>;

    if (instruction.instruction.substr(0, 5) == "START") {
        parsedInstruction->push_back("START");
        trimKeyword(instruction.instruction, "START");

        parsedInstruction->push_back(instruction.instruction);
    } else if (instruction.instruction.substr(0, 3) == "END") {
        parsedInstruction->push_back("END");
        trimKeyword(instruction.instruction, "END");

        parsedInstruction->push_back(instruction.instruction);
    } else if (instruction.instruction.substr(0, 7) == "RETURNS") {
        parsedInstruction->push_back("RETURNS");
        trimKeyword(instruction.instruction, "RETURNS");

        // Iterate through all provided possible return
        // values and append them to parsedInstruction.
        char instruction_cstr[instruction.instruction.length() + 1];
        strcpy(instruction_cstr, instruction.instruction.c_str());

        char* returnType;
        returnType = strtok(instruction_cstr, " ");

        while (returnType != NULL) {
            parsedInstruction->push_back(returnType);
            returnType = strtok(NULL, " ");
        }
    } else if (instruction.instruction.substr(0, 5) == "BRIEF") {
        parsedInstruction->push_back("BRIEF");
        trimKeyword(instruction.instruction, "BRIEF");

        parsedInstruction->push_back(instruction.instruction);
    } else {
        delete parsedInstruction;
        return nullptr;
    }

    return parsedInstruction;
}

inline std::string file_pton(std::string filePath) {
    // File path to file name.
    filePath.erase(0, (filePath.find_last_of('/') == filePath.npos) ? 0 : filePath.find_last_of('/') + 1);
    return filePath;
}

void genDoc_file(const std::string &fileName) {
    // Concatenate the function/class docs in a file to buffers.
    try {
        if (getFileType(fileName) == LANG::OTHER)
            throw unsupported_filetype_error(fileName);
        else if (getFileType(fileName) == LANG::PYTHON) {
            std::cout << "Generating documentation for " << file_pton(fileName) << std::endl;

            std::ifstream file(fileName);
            std::ofstream docFile;
            docFile.open("documentation.md", std::ios::out | std::ios::app);

            docFile << "# " << file_pton(fileName) << "\n";
            // Include file name when appending to project documentation.

            std::string line;

            Instruction instruction;
            std::vector<std::string> *parsedInstruction;

            ClassDoc *currClass = nullptr;
            FuncDoc  *currFunc = nullptr;

            bool isArg = false, isDesc = false;

            std::string prevLine = "";
            for (int lineNo = 1; std::getline(file, line); ++lineNo) {
                // Use a for loop to keep track of line numbers.
                line.erase(0, line.find_first_not_of(' ')); // Trim leading whitespace.
                if (line.substr(0, 3) == "#dg") {
                    line = line.substr(4, line.length());
                    instruction = Instruction(fileName, lineNo, line);

                    parsedInstruction = parseInstruction(instruction);

                    if (parsedInstruction != nullptr) {
                        if ((*parsedInstruction)[0] == "START") {
                            if ((*parsedInstruction)[1] == "CLASS") {
                                if (currFunc) {
                                    throw dg_syntax_error("Cannot document a CLASS inside a FUNC", fileName, lineNo);
                                } else if (currClass) {
                                    throw dg_syntax_error("Cannot document a CLASS inside a CLASS", fileName, lineNo);
                                }
                                currClass = new ClassDoc();

                                prevLine.erase(0, prevLine.find_first_not_of(' '));
                                prevLine.erase(prevLine.find_last_not_of(' ') + 1);
                                prevLine.insert(6, "__");

                                currClass->setName(prevLine.substr(0, prevLine.length() - 1) + "__");

                                docFile << currClass->name() << "\n";
                            } else if ((*parsedInstruction)[1] == "FUNC") {
                                if (currFunc) {
                                    throw dg_syntax_error("Cannot document a FUNC inside a FUNC", fileName, lineNo);
                                }

                                if (currClass) {
                                    currFunc = new FuncDoc(currClass);
                                } else {
                                    currFunc = new FuncDoc();
                                }

                                prevLine.erase(0, prevLine.find_first_not_of(' '));
                                prevLine.erase(prevLine.find_last_not_of(' ') + 1);
                                prevLine.erase(0, 4);

                                for (size_t i = 0; i < prevLine.length(); ++i) {
                                    // Escape underscores in function names because
                                    // underscores are part of markdown notation.
                                    if (prevLine[i] == '_') {
                                        prevLine.insert(i, "\\");
                                        ++i;
                                    }
                                }

                                std::string name = std::string("function __") + prevLine.substr(0, prevLine.length() - 1);
                                name.insert(name.find_first_of('('), "__");

                                currFunc->setName(name);
                            } else if ((*parsedInstruction)[1] == "ARGS") {
                                isArg = true;
                            } else if ((*parsedInstruction)[1] == "DESC") {
                                isDesc = true;
                            } else {
                                throw dg_syntax_error("Invalid START type \"" + (*parsedInstruction)[1] + "\"", fileName, lineNo);
                            }
                        } else if ((*parsedInstruction)[0] == "END") {
                            if ((*parsedInstruction)[1] == "CLASS") {
                                docFile << "---\n";
                                delete currClass;
                                currClass = nullptr;
                            } else if ((*parsedInstruction)[1] == "FUNC") {
                                docFile << currFunc->name() << "\n"
                                        << currFunc->args() << "\n"
                                        << currFunc->returnType() << "\n"
                                        << currFunc->desc() << "\n";

                                if (!currClass)
                                    docFile << "---\n";

                                delete currFunc;
                                currFunc = nullptr;
                            } else if ((*parsedInstruction)[1] == "ARGS") {
                                isArg = false;
                            } else if ((*parsedInstruction)[1] == "DESC") {
                                if (currClass && !currFunc) {
                                    docFile << currClass->desc() << "\n";
                                }
                                isDesc = false;
                            } else {
                                throw dg_syntax_error("Invalid END type \"" + (*parsedInstruction)[1] + "\"", fileName, lineNo);
                            }
                        } else if ((*parsedInstruction)[0] == "RETURNS") {
                            if (!currFunc) {
                                throw dg_syntax_error("Cannot use RETURNS outside of a FUNC", fileName, lineNo);
                            }
                            std::string returnTypes = (*parsedInstruction)[1];
                            for (size_t i = 2; i < parsedInstruction->size(); ++i) {
                                returnTypes += std::string("\n    ") + (*parsedInstruction)[i];
                            }
                            currFunc->setReturnType(returnTypes);
                        } else if ((*parsedInstruction)[0] == "BRIEF") {
                            if (currFunc) {
                                currFunc->addDescLine((*parsedInstruction)[1]);
                            } else if (currClass) {
                                currClass->addDescLine((*parsedInstruction)[1]);
                                docFile << currClass->desc() << "\n";
                            } else {
                                throw dg_syntax_error("Cannot use BRIEF outside of a FUNC or CLASS", fileName, lineNo);
                            }
                        }
                    } else {
                        // If an instruction is not parsed by parseInstruction and we are not
                        // in an argument/description block.
                        throw dg_syntax_error("Invalid instruction", fileName, lineNo);
                    }
                    delete parsedInstruction;
                } else if (isArg) {
                    if (currFunc == nullptr) {
                        throw dg_syntax_error("Args defined while not in function", fileName, lineNo);
                    }
                    currFunc->addArg(parseFuncArg(line));
                } else if (isDesc) {
                    if (currFunc) {
                        currFunc->addDescLine(parseDesc(line));
                    } else if (currClass) {
                        currClass->addDescLine(parseDesc(line));
                    } else {
                        throw dg_syntax_error("Desc defined while not in class or function", fileName, lineNo);
                    }
                }
                prevLine = line;
            }
        file.close();
        docFile.close();
        }
    } catch (const unsupported_filetype_error &err) {
        std::cerr << err.errmsg() << std::endl;
    } catch (const dg_syntax_error &err) {
        std::cerr << std::endl << err.errmsg() << std::endl;
        exit(EXIT_FAILURE);
    }
}

void genDoc_project(const std::string &projectName) {
    boost::filesystem::path projectPath(".");

    // Clear documentation.md and write the project name as the title
    std::ofstream docFile;
    docFile.open("documentation.md", std::ofstream::out | std::ofstream::trunc);
    docFile << "# __" << projectName << "__\n";
    docFile.close();

    boost::filesystem::directory_iterator endItr;
    for (boost::filesystem::directory_iterator file(projectPath); file != endItr; ++file) {
        genDoc_file(boost::filesystem::canonical(file->path()).string());
    }
}


text:
#include <boost/thread.hpp>

#include <ros/ros.h>
#include <pluginlib/class_list_macros.h>
#include <nodelet/nodelet.h>
#include "usb_cam_wrapper.h"

namespace usb_cam {

class UsbCamNodelet: public nodelet::Nodelet
{
 public:
  UsbCamNodelet() {}
  ~UsbCamNodelet() {
  ROS_INFO("shutting down driver thread");
  if (device_thread_ != nullptr && device_thread_->joinable()) {
    device_thread_->join();
  }
  ROS_INFO("driver thread stopped");
};

 private:
  virtual void onInit();
  boost::shared_ptr<UsbCamWrapper> usb_cam_ = nullptr;
  boost::shared_ptr<boost::thread> device_thread_ = nullptr;
};

void UsbCamNodelet::onInit()
{
  ROS_INFO("Usb cam nodelet init");
  usb_cam_.reset(new UsbCamWrapper(getNodeHandle(), getPrivateNodeHandle()));
  // spawn device poll thread
  device_thread_ = boost::shared_ptr<boost::thread>
        (new boost::thread(boost::bind(&UsbCamWrapper::spin, usb_cam_)));
}

}

// Register this plugin with pluginlib.  Names must match nodelets.xml.
//
// parameters: package, class name, class type, base class type
PLUGINLIB_DECLARE_CLASS(usb_cam, UsbCamNodelet,
                        usb_cam::UsbCamNodelet, nodelet::Nodelet);

max_stars_repo_name: BaiduXLab/apollo

max_issues_repo_name: BaiduXLab/apollo

max_forks_repo_name: BaiduXLab/apollo


text:
#include "modelloader.h"

#include "model.h"
#include "modelprocessor.h"
#include <boost/filesystem.hpp>
#include <stdexcept>
#include <iostream>

using namespace std;

void ModelLoader::register_processor(std::unique_ptr<ModelProcessor> &processor)
{
	m_processors.push_back(std::move(processor));
}

std::shared_ptr<Model> ModelLoader::load(const std::string &path, bool allow_cached_materials, size_t vertex_limit) const
{
	boost::filesystem::path model_path(path);
	std::string extension = model_path.extension().string();

	for(const auto &processor : m_processors) {
		for(const auto &handled_ext : processor->handled_extensions()) {
			if(handled_ext == extension) {
				auto model =  processor->load(path, vertex_limit, allow_cached_materials);
				const Vector3d min_bound = model->min_bound();
				const Vector3d max_bound = model->max_bound();
				cout << "Model bounds: [" << min_bound.x() << ", " << min_bound.y() << ", " << min_bound.z() << "] - [" << max_bound.x() << ", " << max_bound.y() << ", " << max_bound.z() << "]\n";
				return model;
			}
		}
	}

	throw runtime_error("No ModelProcessor exists for the given file.");
}

max_stars_repo_name: EmilNorden/candle

max_issues_repo_name: EmilNorden/candle

max_forks_repo_name: EmilNorden/candle


text:
#pragma once
#ifndef OPENGM_STRUCTSVM_HXX
#define OPENGM_STRUCTSVM_HXX

#include <cassert>
#include <ostream>
#include <vector>

#include <dlib/svm.h>
#include <dlib/optimization.h>
#include <dlib/matrix.h>

#include "opengm/opengm.hxx"
#include "pgmlink/ext_opengm/loss_hamming.hxx"
#include "pgmlink/ext_opengm/loglinearmodel.hxx"
#include "opengm/inference/lpcplex.hxx"
#include "opengm/operations/minimizer.hxx"

namespace opengm {

template<class LLM>
class StructSvmDlib
  : public dlib::structural_svm_problem<dlib::matrix<typename LLM::ValueType> > {
public:
  typedef LLM LoglinearModelType;

  StructSvmDlib(const std::vector<LoglinearModelType>& samples,
		const std::vector<std::vector<typename LoglinearModelType::LabelType> > labels)
    : samples_(samples), samples_and_loss_(samples), labels_(labels) {
    for(size_t i=0; i < samples.size(); ++i) {
      augmentWithLoss(samples_and_loss_[i], labels_[i]);
      LLM loss(samples_[i].space());
      augmentWithLoss(loss, labels_[i]);
      loss_.push_back(loss);
    }
  };

  void train(std::vector<typename LoglinearModelType::ValueType>& learned_weights);

  ///
  /// interface required by dlib
  ///
  typedef dlib::matrix<typename LLM::ValueType> matrix_type_dlib;
  typedef typename matrix_type_dlib::type scalar_type_dlib;
  typedef matrix_type_dlib feature_vector_type_dlib;

  virtual long get_num_dimensions () const;
  virtual long get_num_samples () const ;
  virtual void get_truth_joint_feature_vector ( long idx, feature_vector_type_dlib& psi ) const;
  virtual void separation_oracle (
				  const long idx,
				  const matrix_type_dlib& current_solution,
				  scalar_type_dlib& loss,
				  feature_vector_type_dlib& psi
				  ) const;

private:
  void augmentWithLoss( LoglinearModelType&, std::vector<typename LoglinearModelType::LabelType>& );

  mutable std::vector<LoglinearModelType> samples_;
  mutable std::vector<LoglinearModelType> samples_and_loss_;
  std::vector<LoglinearModelType> loss_;
  std::vector<std::vector<typename LoglinearModelType::LabelType> > labels_;
};

/******************/
/* implementation */
/******************/

template<class LLM>
void StructSvmDlib<LLM>::train(std::vector<typename LoglinearModelType::ValueType>& learned_weights) {
  learned_weights.clear();
  dlib::matrix<typename LLM::ValueType> weights;
  dlib::oca solver;
  solver.set_subproblem_epsilon(0.0000001);
  //solver.set_subproblem_max_iterations(100);
  std::cout << "\n";
  std::cout << "oca: subproblem epsilon: " << solver.get_subproblem_epsilon() << "\n";
  std::cout << "oca: subproblem max iterations: " << solver.get_subproblem_max_iterations() << "\n";

  this->be_verbose();
  solver( *this, weights);
  //assert(weights.size() == 2);
  assert(weights.nc() == 1);
  for(long i = 0; i< weights.nr(); ++i) {
    learned_weights.push_back(weights(0,i));
  }
}

template<class LLM>
long StructSvmDlib<LLM>::get_num_dimensions() const {
  return samples_.front().numberOfWeights();
}

template<class LLM>
long StructSvmDlib<LLM>::get_num_samples() const {
  return samples_.size();
}

template<class LLM>
void StructSvmDlib<LLM>::get_truth_joint_feature_vector( long idx, feature_vector_type_dlib& psi ) const {
  psi.set_size(this->get_num_dimensions(), 1);

  std::vector<typename LLM::ValueType> feats(samples_[idx].numberOfWeights());
  samples_[idx].weightedFeatureSums( labels_[idx], feats);
  for(long i = 0; i < psi.size(); ++i) {
    psi(i,0) = feats[i];
    std::cout << "true feature["<<i<<"]: " << psi(i,0) << "\n";
  }
}

template<class LLM>
void StructSvmDlib<LLM>::separation_oracle(const long idx,
					   const matrix_type_dlib& current_solution,
					   scalar_type_dlib& loss,
					   feature_vector_type_dlib& psi
					   ) const {
  std::cout << "\n";
  std::cout << "Entered separation oracle\n";
  std::vector<typename LLM::ValueType> weights;
  for(int i = 0; i < current_solution.nr(); ++i) {
    weights.push_back(-1*current_solution(i, 0));
    std::cout << "weights["<<i<<"]: " << current_solution(i,0) << "\n";
  }
  samples_and_loss_[idx].setWeights(weights);
  samples_[idx].setWeights(weights);

  typedef opengm::LPCplex<LLM, opengm::Minimizer> optimizer_type;
  typename optimizer_type::Parameter params;
  params.verbose_ = true;
  params.integerConstraint_ = true;
  params.epGap_ = 0.01;
  opengm::InferenceTermination status;
  optimizer_type inference = optimizer_type(samples_and_loss_[idx], params);
  status = inference.infer();
  if(status != opengm::NORMAL) {
    throw RuntimeError("GraphicalModel::infer(): optimizer terminated unnormally");
  }

  std::vector<typename LLM::LabelType> optimal;
  status = inference.arg(optimal);
  if(status != opengm::NORMAL) {
    throw RuntimeError("GraphicalModel::infer(): solution extraction terminated unnormally");
  }
  std::cout << "current[0]: " << optimal[0] << " label[0]: " << labels_[idx][0]  << "\n";
  std::cout << "current[1]: " << optimal[1] << " label[1]: " << labels_[idx][1]  << "\n";
  std::cout << "current[2]: " << optimal[2] << " label[2]: " << labels_[idx][2]  << "\n";
  std::cout << "current[3]: " << optimal[3] << " label[3]: " << labels_[idx][3]  << "\n";
  std::cout << "etc. etc.\n";

  std::vector<typename LLM::ValueType> feats(samples_and_loss_[idx].numberOfWeights());
  samples_and_loss_[idx].weightedFeatureSums( optimal, feats );
  psi.set_size(samples_[idx].numberOfWeights(), 1);
  for(long i = 0; i < psi.size(); ++i) {
    std::cout << "feature["<<i <<"]: " << feats[i] << "\n";
    psi(i,0) = feats[i];
  }

  loss = -1*loss_[idx].evaluate(optimal);
  std::cout << "ground truth energy for idx" << idx <<": "<< -1*samples_[idx].evaluate(labels_[idx])<<"\n";
  std::cout << "energy for idx" << idx <<": "<< -1*samples_[idx].evaluate(optimal)<<"\n";
  std::cout << "energy incl. loss for idx" << idx <<": "<< -1*samples_and_loss_[idx].evaluate(optimal)<<"\n";
  std::cout << "loss for idx "<< idx <<": " << loss << "\n";
  std::cout << "leaving oracle\n\n";
}

template<class LLM>
void StructSvmDlib<LLM>::augmentWithLoss( LLM& m, std::vector<typename LoglinearModelType::LabelType>& target_labels ) {
  typename LLM::IndexType lidx[] = {0};
  HammingFunction<typename LLM::ValueType, typename LLM::IndexType, typename LLM::LabelType> hf;
  for(typename LLM::IndexType i = 0; i < target_labels.size(); ++i) {
    lidx[0] = i;
    hf = HammingFunction<typename LLM::ValueType, typename LLM::IndexType, typename LLM::LabelType>(m.numberOfLabels(i), target_labels[i]);
    m.addFactor(m.addFunction(hf), lidx, lidx+1 );
  }
}

} // end namespace opengm

#endif

ext: hxx

max_stars_repo_name: chaubold/pgmlink

max_issues_repo_name: chaubold/pgmlink

max_forks_repo_name: chaubold/pgmlink


text:
#pragma once

#include "../core/datastream.hpp"
#include "../core/name.hpp"
#include "../core/database.h"
#include "../core/wasm_log.hpp"
#include "../core/strings.hpp"

#include <vector>
#include <memory>
#include <boost/hana.hpp>

using namespace std;
using namespace wasm;
namespace hana = boost::hana;

namespace wasm { namespace db {

    #define WASM_DB_TABLE_TYPE_DEFINITION(T) \
    typedef T table_type;

    constexpr static inline regid payer_reserved{};

    //for debug before gettable updated
    constexpr static uint64_t SECONDARY_KEY_PREFIX = 0x000000000000000FULL;
    // constexpr static uint64_t PRIMARY_KEY_PREFIX   = 0x00;
    // constexpr static uint64_t SECONDARY_KEY_PREFIX = 0x01;

    template<class Class, typename Type, Type (Class::*PtrToMemberFunction)()const>
    struct const_mem_fun
    {
        typedef typename std::remove_reference<Type>::type result_type;

        template<typename ChainedPtr>
        auto operator()(const ChainedPtr& x)const -> std::enable_if_t<!std::is_convertible<const ChainedPtr&, const Class&>::value, Type>
        {
            return operator()(*x);
        }

        Type operator()(const Class& x) const
        {
            return (x.*PtrToMemberFunction)();
        }

        Type operator()(const std::reference_wrapper<const Class>& x)const
        {
            return operator()(x.get());
        }

        Type operator()(const std::reference_wrapper<Class>& x)const
        {
            return operator()(x.get());
        }
    };

    template<name::raw IndexName, typename Extractor>
    struct indexed_by {
       enum constants { index_name   = static_cast<uint64_t>(IndexName) };
       typedef Extractor secondary_extractor_type;
    };

    template<name::raw TableName, typename T, typename PrimaryKeyType, typename... Indices>
    struct index_table {

    public:
        template<name::raw IndexName, typename Extractor, uint64_t Number, bool IsConst>
        struct index {
             public:
                typedef Extractor  secondary_extractor_type;
                typedef typename std::decay<decltype( Extractor()(nullptr) )>::type secondary_key_type;

                constexpr static bool validate_index_name( wasm::name n ) {
                   return n.value != 0 && n != wasm::name("primary"); // Primary is a reserve index name.
                }

                static_assert( validate_index_name( wasm::name(IndexName) ), "invalid index name used in index_table" );

                enum constants {
                   table_name   = static_cast<uint64_t>(TableName),
                   index_name   = static_cast<uint64_t>(IndexName),
                   index_number = Number,
                   index_table_name = (static_cast<uint64_t>(TableName) & 0xFFFFFFFFFFFFFFF0ULL)
                                        | (Number & 0x000000000000000FULL) // Assuming no more than 16 secondary indices are allowed
                };

                constexpr static uint64_t name()   { return index_table_name; }
                constexpr static uint64_t number() { return Number; }

                static auto extract_secondary_key(const T& obj) { return secondary_extractor_type()(obj); }
             private:
                friend class index_table;

                index( typename std::conditional<IsConst, const index_table*, index_table*>::type midx )
                :_idx_table(midx){}

                typename std::conditional<IsConst, const index_table*, index_table*>::type _idx_table;
        }; /// struct index_table::index

        template<uint64_t I>
        struct intc { enum e{ value = I }; operator uint64_t()const{ return I; }  };

        static constexpr auto transform_indices( ) {

            typedef decltype( hana::zip_shortest(
                             hana::make_tuple( intc<0>(), intc<1>(), intc<2>(), intc<3>(), intc<4>(), intc<5>(),
                                               intc<6>(), intc<7>(), intc<8>(), intc<9>(), intc<10>(), intc<11>(),
                                               intc<12>(), intc<13>(), intc<14>(), intc<15>() ),
                             hana::tuple<Indices...>() ) ) indices_input_type;

            return hana::transform( indices_input_type(), [&]( auto&& idx ){
                typedef typename std::decay<decltype(hana::at_c<0>(idx))>::type num_type;
                typedef typename std::decay<decltype(hana::at_c<1>(idx))>::type idx_type;
                return hana::make_tuple( hana::type_c<index<wasm::name::raw(static_cast<uint64_t>(idx_type::index_name)),
                                                         typename idx_type::secondary_extractor_type,
                                                         num_type::e::value, false> >,
                                         hana::type_c<index<wasm::name::raw(static_cast<uint64_t>(idx_type::index_name)),
                                                             typename idx_type::secondary_extractor_type,
                                                             num_type::e::value, true> > );

            });
        }

        typedef decltype( index_table::transform_indices() ) indices_type;
        indices_type _indices;

      public:
        typedef T object_type;
        typedef PrimaryKeyType primary_key_type;

      public:
         name::raw table_name() const {
         	return TableName;
         }
    };

    template<typename ObjectType>
    bool get(ObjectType& object) {

        typename ObjectType::table_type table;

        //vector<char> key = pack(std::tuple(PRIMARY_KEY_PREFIX, table.table_name(), object.primary_key()));
        vector<char> key = pack(std::tuple(table.table_name(), object.primary_key()));
		uint32_t key_len = key.size();

		auto value_len = db_get(key.data(), key_len, NULL, 0);
		if(value_len == 0){ return false; }

		auto v = (char*)malloc(sizeof(char) * value_len);
		db_get( key.data(), key_len, v, value_len);
		object = unpack<ObjectType>(v, value_len);
		free(v);
		return true;
    }

    template<typename ObjectType, typename PrimaryKeyType>
    bool get(ObjectType& object, const PrimaryKeyType& primary) {

        typename ObjectType::table_type table;

        //vector<char> key = pack(std::tuple(PRIMARY_KEY_PREFIX, table.table_name(), primary));
        vector<char> key = pack(std::tuple(table.table_name(), primary));
        uint32_t key_len = key.size();

        auto value_len = db_get(key.data(), key_len, NULL, 0);
        if(value_len == 0){ return false; }

        auto v = (char*)malloc(sizeof(char) * value_len);
        db_get( key.data(), key_len, v, value_len);
        object = unpack<ObjectType>(v, value_len);
        free(v);
        return true;
    }

    template<typename ObjectType>
    void set(const ObjectType& object) {

        typename ObjectType::table_type table;

        //vector<char> key   = pack(std::tuple(PRIMARY_KEY_PREFIX,table.table_name(), object.primary_key()));
        vector<char> key   = pack(std::tuple(table.table_name(), object.primary_key()));
		vector<char> value = pack(object);

        ObjectType old_object;
        bool has_old_object   = wasm::db::get(old_object, object.primary_key());

        hana::for_each( table._indices, [&]( auto& idx ) {
            typedef typename decltype(+hana::at_c<0>(idx))::type index_type;
            if(has_old_object && index_type::extract_secondary_key(old_object) != index_type::extract_secondary_key(object))
            {
                auto old_secondary = pack(std::tuple(SECONDARY_KEY_PREFIX, index_type::name(), index_type::extract_secondary_key(old_object)));
                db_remove(wasm::db::payer_reserved.value, old_secondary.data(), old_secondary.size());
            }

            auto secondary = pack(std::tuple(SECONDARY_KEY_PREFIX, index_type::name(), index_type::extract_secondary_key(object)));
            auto primary   = pack(object.primary_key());

            db_store(payer_reserved.value, secondary.data(), secondary.size(), primary.data(), primary.size());
        });

        db_store(payer_reserved.value, key.data(), key.size(), value.data(), value.size());

    }

    template<typename ObjectType>
    void del(const ObjectType& object) {

        typename ObjectType::table_type table;

        //vector<char> key = pack(std::tuple(PRIMARY_KEY_PREFIX, table.table_name(), object.primary_key()));
        vector<char> key = pack(std::tuple( table.table_name(), object.primary_key()));

        // bool has_object = wasm::db::get(object);
        // if(has_object){
            hana::for_each( table._indices, [&]( auto& idx ) {
                typedef typename decltype(+hana::at_c<0>(idx))::type index_type;
                vector<char> secondary = pack(std::tuple(SECONDARY_KEY_PREFIX, index_type::name(), index_type::extract_secondary_key(object)));
                db_remove(payer_reserved.value, secondary.data(), secondary.size());
            });
            db_remove(payer_reserved.value, key.data(), key.size());
        //}

    }

    template<typename ObjectType>
    bool get(uint64_t by_index, ObjectType& object, const char* error_msg = "unable to find key" ) {

         //WASM_LOG_FPRINT(true, "by_index:%", wasm::name(by_index))
        typedef typename ObjectType::table_type table_type;
        typename table_type::primary_key_type primary;
        table_type table;

        bool has_primary = false;
        hana::for_each( table._indices, [&]( auto& idx ) {
           typedef typename decltype(+hana::at_c<0>(idx))::type index_type;
           if(index_type::index_name == by_index)
           {
               //auto secondary   = pack(std::tuple(SECONDARY_KEY_PREFIX, table.table_name(), object.scope(), index_type::name(), index_type::extract_secondary_key(object)));
               auto secondary   = pack(std::tuple(SECONDARY_KEY_PREFIX, index_type::name(), index_type::extract_secondary_key(object)));
               auto primary_len = db_get(secondary.data(), secondary.size(), NULL, 0);

               //WASM_LOG_FPRINT(true, "by_index:% secondary_key:%", wasm::name(by_index),  index_type::extract_secondary_key(object))

               if(primary_len == 0) {return;}

               auto v = (char*)malloc(sizeof(char) * primary_len);
               db_get( secondary.data(), secondary.size(), v, primary_len);
               primary = unpack<typename table_type::primary_key_type>(v, primary_len);
               free(v);

               //WASM_LOG_FPRINT(true, "by_index:% secondary_key:% primary:%", wasm::name(by_index),  index_type::extract_secondary_key(object), primary)
               has_primary = true;
               return;
            }

        });

        if(!has_primary) return false;

        return get(object, primary);

    }

}}//db//wasm

max_stars_repo_name: uenctech/wasm-cdt

max_issues_repo_name: uenctech/wasm-cdt

max_forks_repo_name: uenctech/wasm-cdt