/home/sebastian/gpudbms/trunk/hype-library/include/query_processing/binary_operator.hpp

Go to the documentation of this file.

#pragma once

//#include <core/scheduling_decision.hpp>
//#include <boost/shared_ptr.hpp>

#include <config/global_definitions.hpp>
#include <config/configuration.hpp>
#include "operator.hpp"
#include "typed_operator.hpp"


//#ifndef HYPE_ENABLE_PARALLEL_QUERY_PLAN_EVALUATION
//      #define HYPE_ENABLE_PARALLEL_QUERY_PLAN_EVALUATION
//#endif


#ifdef HYPE_ENABLE_PARALLEL_QUERY_PLAN_EVALUATION
    #include <boost/thread.hpp>
    #include <boost/bind.hpp>
#endif

namespace hype {
    namespace queryprocessing {

        template <typename OperatorInputTypeLeftChild, typename OperatorInputTypeRightChild, typename OperatorOutputType>
        class BinaryOperator : public TypedOperator<OperatorOutputType> { //uses 2D Learning Method
        public:
            //typedef typename OperatorMapper_Helper_Template<OperatorOutputType>::TypedOperator TypedOperator;
            typedef typename OperatorMapper_Helper_Template<OperatorOutputType>::TypedOperatorPtr TypedOperatorPtr;
            typedef typename OperatorMapper_Helper_Template<OperatorOutputType>::TypedNodePtr TypedNodePtr;

            BinaryOperator(const hype::SchedulingDecision& sched_dec,
                    boost::shared_ptr<TypedOperator<OperatorInputTypeLeftChild> > left_child,
                    boost::shared_ptr<TypedOperator<OperatorInputTypeRightChild> > right_child)
            : TypedOperator<OperatorOutputType>(sched_dec), left_child_(left_child), right_child_(right_child) {
            }

            virtual ~BinaryOperator() {

            }

            OperatorInputTypeLeftChild getInputDataLeftChild() {
                return this->left_child_->getResult();
            }

            OperatorInputTypeRightChild getInputDataRightChild() {
                return this->right_child_->getResult();
            }

            virtual bool run() {
                if (!hype::core::Runtime_Configuration::instance().isQueryChoppingEnabled()) {
#ifdef HYPE_ENABLE_PARALLEL_QUERY_PLAN_EVALUATION
                //std::cout << "Binary Operator: launching both childs in parallel" << std::endl;
                boost::thread_group threads;
                if (left_child_) {
                    //left_child_->run();
                    threads.add_thread(new boost::thread(boost::bind(&TypedOperator<OperatorInputTypeLeftChild>::run, left_child_.get())));
                }
                if (right_child_) {
                    //right_child_->run();
                    threads.add_thread(new boost::thread(boost::bind(&TypedOperator<OperatorInputTypeLeftChild>::run, right_child_.get())));
                }
                threads.join_all();
#else                                                
                if (left_child_) {
                    left_child_->run();
                }
                if (right_child_) {
                    right_child_->run();
                }
#endif
                }
                //execute this operator
                double begin = double(hype::core::getTimestamp());
                timeEstimated = double(this->getEstimatedExecutionTime().getTimeinNanoseconds());
                bool retVal = (*this)();
                timeNeeded = double( hype::core::getTimestamp()) - begin;
                return retVal;
            }

            virtual void print(unsigned int tree_level) const {
                for (unsigned int i = 0; i < tree_level; ++i) {
                    std::cout << "\t";
                }
                assert(!this->getFeatureValues().empty());
                std::cout << this->getAlgorithmName() << " ET: " << this->getEstimatedExecutionTime().getTimeinNanoseconds()
                        << "\tEC: " << this->getFeatureValues()[0]
                        << std::endl; // << " Features: " << this->getFeatureValues() << std::endl;
                if (left_child_) left_child_->print(tree_level + 1);
                if (right_child_) right_child_->print(tree_level + 1);
            }

            virtual void printResult(unsigned int tree_level) const {
                for (unsigned int i = 0; i < tree_level; ++i) {
                    std::cout << "\t";
                }
                assert(!this->getFeatureValues().empty());
                std::cout << this->getAlgorithmName() << " ET: " << timeEstimated
                        << "\tMT: " << timeNeeded
                        << "\tEE: " << 1 - (timeEstimated / timeNeeded)
                        << "\tEC: " << this->getFeatureValues()[0];
                if (this->getResultSize() != -1) {
                    std::cout << "\tRR: " << this->getResultSize();
                }
                std::cout << std::endl;
                if (left_child_) left_child_->printResult(tree_level + 1);
                if (right_child_) right_child_->printResult(tree_level + 1);
            }

            virtual double getRecursiveExecutionTimeInNanoseconds() {
                double executiontime = this->getEstimatedExecutionTime().getTimeinNanoseconds();
                if (executiontime < 0) {
                    executiontime = 0;
                }
                if (!hype::core::Runtime_Configuration::instance().isQueryChoppingEnabled()) {
                if (left_child_) {
                    executiontime += left_child_->getRecursiveExecutionTimeInNanoseconds();
                }
                if (right_child_) {
                    executiontime += right_child_->getRecursiveExecutionTimeInNanoseconds();
                }
                } else {
                if (left_child_ && right_child_) {
                    if (left_child_->getRecursiveExecutionTimeInNanoseconds() > right_child_->getRecursiveExecutionTimeInNanoseconds()) {
                        executiontime += left_child_->getRecursiveExecutionTimeInNanoseconds();
                    } else {
                        executiontime += right_child_->getRecursiveExecutionTimeInNanoseconds();
                    }
                } else {
                if (left_child_) {
                    executiontime += left_child_->getRecursiveExecutionTimeInNanoseconds();
                }
                if (right_child_) {
                    executiontime += right_child_->getRecursiveExecutionTimeInNanoseconds();
                    }
                    }
                }

                return executiontime;
            }
            /*
            virtual std::list<TypedOperator<OperatorOutputType>& > getOperatorQueue(){

                    std::list<TypedOperator<OperatorOutputType>& > left_list = left_child_->getOperatorQueue();
                    left_list.push_front(*this);

                    std::list<TypedOperator<OperatorOutputType>& > right_list = right_child_->getOperatorQueue();

                    left_list.insert(left_list.end(),right_list.begin(),right_list.end());
                    return left_list;
                    //this->left_child_->
            }*/

        private:
            double timeNeeded;
            double timeEstimated;

            boost::shared_ptr<TypedOperator<OperatorInputTypeLeftChild> > left_child_;
            boost::shared_ptr<TypedOperator<OperatorInputTypeRightChild> > right_child_;

            //boost::shared_ptr<TypedOperator<OperatorOutputType> > succsessor_;
        };


    }; //end namespace queryprocessing
}; //end namespace hype