QBox.h

Go to the documentation of this file.

#ifndef QBOX_H
#define QBOX_H

#include <vector>
#include <deque>
#include <sstream>

#include "Exceptions.h"
#include "TupleQueue.h"
#include <nmstl_util.h>
#include "TupleDescription.h"
#include "Stream.h"
#include "Registry.h"
#include "BoxStatsSample.h"
#include "PtMutex.h"
#include "LockHolder.h"
#include "Scheduler.h"
#include "Params.h"
#include "LockSet.h"
#include "Tuple.h"
#include "CPView.h"

#include "xercesDomFwd.h"
#include <sys/time.h>

#include "QBoxState.h"

BOREALIS_NAMESPACE_BEGIN;

class AuroraNode;
class Table;
class QBoxInvocation;

class QBox {
  public:
    static string getBoxType(const DOMElement*) throw (AuroraException);

    // Instantiates a QBox from a DOMElement.  Throws an
    // AuroraBadEntityException if the entity's type is unknown, or
    // throws other exceptions if instantiation fails for some other
    // reason.
        // 'node' is the AuroraNode whose query netowrk this QBox is part of.
    static QBox *instantiate(const DOMElement *) throw (AuroraException);

    virtual ~QBox();

    const DOMElement *getCatalogElement() const {
        ASSERT(_catalog_element);
        return _catalog_element;
    }

    Table *getTable() const {
        return _table;
    }

    void setTable(Table *table) {
        assert(_state == CONSTRUCTED);
        _table = table;
    }

    const TupleDescription& getInputDescription(unsigned int i) const {
        assert(i < _in_desc.size());
        return *(_in_desc[i]);
    }

    void setInputDescription(unsigned int i, ptr<TupleDescription> desc) {
        assert(_state == CONSTRUCTED);
        assert(i < 1000);
        if (_in_desc.size() <= i)
            _in_desc.resize(i + 1);
        _in_desc[i] = desc;
    }

    void setInputDescription(unsigned int i, const TupleDescription& desc) {
        // copies the tupledescription
        setInputDescription(i, ptr<TupleDescription>(new TupleDescription(desc)));
    }

    const TupleDescription& getOutputDescription(unsigned int i) const {
        assert(i < _out_desc.size());
        return *(_out_desc[i]);
    }

    void setOutputDescription(unsigned int i, ptr<TupleDescription> desc) {
        assert(_state == CONSTRUCTED);
        assert(i < 1000);
        if (_out_desc.size() <= i)
            _out_desc.resize(i + 1);
        _out_desc[i] = desc;
    }

    void setOutputDescription(unsigned int i, const TupleDescription& desc) {
        // copies the tupledescription
        setOutputDescription(i, ptr<TupleDescription>(new TupleDescription(desc)));
    }

    Stream& getInput(unsigned int n) const {
        assert(n < _inputs.size());
        return *(_inputs[n]);
    }

    void setInput(unsigned int i, Stream *desc) {
        assert(_state == CONSTRUCTED || _state == SETUP);
        assert(i < 1000);
        if (_inputs.size() <= i)
            _inputs.resize(i + 1);
        _inputs[i] = desc;
    }

    Stream& getOutput(unsigned int n) const {
        assert(n < _outputs.size());
        return *(_outputs[n]);
    }

    void setCPView(ptr<CPView> cpview) {
      _cpview_on_input = cpview;
    }

    void setOutput(unsigned int i, Stream *desc);

    unsigned int getNumInputs() { return _in_desc.size(); }

    unsigned int getNumOutputs() { return _out_desc.size(); }

    string getName() const { return _name; }

    string getType() const { return _type; }

    string as_string() const;

    void setup(AuroraNode&, const DOMElement* elt) throw (AuroraException);

    void init(PagePool& pool) throw (AuroraException);


    // Queue packing is common to all boxes.
    void setPendingQueueState(ptr<QueueState> packed_queues) { _pending_queue_state = packed_queues; }
    void unpackQueue(ptr<QueueState> packed_queue); // Added by Magda
    ptr<QueueState> packQueues() { return ptr<QueueState>(new QueueState(_inq)); };

    // State packing is box specific.
    virtual void setPendingBoxState(ptr<AbstractBoxState> packed_box) { _pending_box_state = packed_box; }
    virtual void unpackState(ptr<AbstractBoxState> box_state) {};
    virtual ptr<AbstractBoxState> packState() { return ptr<AbstractBoxState>(); };


#ifdef AURORA_RUNTIME_STATS

        const deque<BoxStatsSample> & lockBoxStatsHistory() {
                _boxStatsHistoryLock.lock();
                return _boxStatsHistory;
        }

        void unlockBoxStatsHistory() {
                _boxStatsHistoryLock.unlock();
        }

        static string dumpBoxStatsHistory(const deque<BoxStatsSample> & h);
    void run(QBoxInvocation& inv);

#else

    void run(QBoxInvocation& inv) { 
        runImpl(inv);
        }

#endif

    void scheduleRun() { getInput(0).notifyEnq(); }

    TupleQueue *getInputQueue(unsigned int input) { return _inq[input].get(); }


    static const Params::Req PARAM_NOT_REQUIRED = Params::NOT_REQUIRED;
    static const Params::Req PARAM_REQUIRED = Params::REQUIRED;
    static const Params::Req PARAM_NON_EMPTY = Params::NON_EMPTY;

    // Return a parameter.  See Params::param().
    string param(string name, Params::Req req = PARAM_NOT_REQUIRED) const
        throw (AuroraBadEntityException)
    { return _params.param(name, req); }

    // Return a parameter.  See Params::param().
    string param(string name, string def) const
    { return _params.param(name, def); }

    // Return a parameter.  See Params::typedParam().
    template <typename T>
    bool typedParam(string name, T& value, Params::Req req = PARAM_NOT_REQUIRED) const
        throw (AuroraBadEntityException)
    { return _params.typedParam(name, value, req); }

    const Params& getParams() { return _params; }

  protected:
    // We dequeue from TupleQueues, and enqueue to Streams.
    typedef TupleQueue::DeqIterator DeqIterator;
    typedef Stream::EnqIterator EnqIterator;
    typedef TupleQueue::SeekIterator SeekIterator; 

    QBox();

    // Called by runImpl(); returns a DeqIterator for the ith input port.
    DeqIterator deq(unsigned int index) { return _inq[index]->deq_iterator(); }

    // Called by runImpl(); returns a DeqIterator for the ith input port.
    SeekIterator seek(unsigned int index) { return _inq[index]->seek_iterator(); }

    // Called by runImpl(); returns an EnqIterator for the ith input port.
    EnqIterator enq(unsigned int index) { return _outputs[index]->enq_iterator(); }

    // METHODS TO IMPLEMENT IN SUBCLASSES:

    // Do preparing
    virtual void setupImpl() = 0;

    // Do whatever's necessary to init the box
    virtual void initImpl() {}

    // Run the thing
    virtual void runImpl(QBoxInvocation&) {}

    ptr<LockSet> getLockSet(string name);

    ptr<AbstractBoxState> _pending_box_state;

    AuroraNode *_node; //moved away from private for now  ...

    // CPView on the input TupleQueue - set when that cpview is created
    ptr<CPView> _cpview_on_input;


  private:

    void unpackQueue(Stream& stream, TupleQueue::EnqIterator enq);

    enum State {
        CONSTRUCTED,
        SETUP,
        INITED,
        DEAD
    };
    State _state;

    string _name;
    string _type;

    vector< ptr<TupleDescription> > _in_desc, _out_desc;
    vector<Stream*> _inputs, _outputs;

    Params _params;

    const DOMElement *_catalog_element;

    vector< ptr<TupleQueue> > _inq;

    ptr<QueueState> _pending_queue_state;

    Table *_table;

    ptr<Scheduler::BoxData> _sched_data;

#ifdef AURORA_RUNTIME_STATS
        // The only reason we declare this here rather than in the run() method
        // is so that we don't have to recreate the vector every time the box
        // is executed.
        //
        // The values in this vector are only meaningful to each individual invocation
        // of this box's run() method.
        vector<size_t> _tuplesEnqBeforeRun, _tuplesEnqDuringRun;

        // Contains the stats sample history for this box.  It will have the n-most
        // recent sample, where 'n' is _statsHistoryLength. No two elements in this deque
        //  will have the same _tv_sec value.  The deque is sorted by _tv_sec value such that the 
        // front() of the list has the most recent sample (the sample with the highest _tv_sec
        // value).
        deque<BoxStatsSample> _boxStatsHistory;

        // Any thread that's accessign _boxStatsHistory must hold this lock.
        PtMutex _boxStatsHistoryLock;

        // This is given by the config file value "server"/"StatsHistoryLength".  We cache the
        // value here to avoid extra function calls that would otherwise occur during each 
        // invocation of the run() method.
        size_t _statsHistoryLength;
#endif

    friend class QBoxInvocation;
    friend class Scheduler; // for _sched_data
};

#define AURORA_DECLARE_QBOX(ClassName, BoxName) AURORA_DECLARE_REG_CLASS(QBox, ClassName)
#define AURORA_DEFINE_QBOX(ClassName, BoxName) \
    AURORA_DEFINE_REG_CLASS_WITH_KEY(QBox, ClassName, BoxName)

typedef Registry<QBox> QBoxRegistry;




class QBoxInvocation {
  public:
    QBoxInvocation() : _tuples_dequeued( 0 ), _box(0), _end_time(0), _next_tick(1){}

    unsigned int getMaxTuplesToDequeue(unsigned int input_port) const {
        return numeric_limits<int>::max();
    }

    bool continueDequeueOn(QBox::DeqIterator deq, unsigned int input_port) const;

    // continueDequeueOn does not seem to do what it claims to do
    // i.e. it uses "processedsofar" from TupleQueue which is never reset
    // thus setting a _tuple_limit results in stalled execution once the
    // limit is reached. This is to be used to compare to _tuple_limit
    // instead of the current version.
    // Therefore the boxes should maintain this counter as they run.
    // It is also techincally wrong since there is a single counter, thus
    // single limit. but we migth want another controlling mechanism so
    // I am not going to make it too complicated now.
    mutable unsigned int _tuples_dequeued;

    // Public members for now
    QBox* _box;

    vector<unsigned int> _tuple_limit;
    unsigned long _end_time;

    mutable unsigned long _last_ticks;
    mutable unsigned int _next_tick;

    unsigned long ticks() const {
        if (!--_next_tick) {
            _next_tick = CALLS_BETWEEN_TICK;
            _last_ticks = Scheduler::ticks();
        }

        return _last_ticks;
    }

    static const int CALLS_BETWEEN_TICK = 1;
};

inline bool QBoxInvocation::continueDequeueOn(QBox::DeqIterator deq, unsigned int input_port) const {
    if (!deq.avail())
        return false;

    _tuples_dequeued++;

    //cout << " Box is here, t_sz " << _tuple_limit.size() << " num dequeued " << _box->getInputQueue(input_port) << " tuple limit " << ( _tuple_limit.size() ? _tuple_limit[input_port] : 0 ) << " already dequeued " << _tuples_dequeued << endl;

    if (_box && _tuple_limit.size() &&
        // the explanation of _tuples_dequeued for this commented line.
        //_box->getInputQueue(input_port)->getNumTuplesDequeued() >= _tuple_limit[input_port])
        _tuples_dequeued >= _tuple_limit[input_port] )
      {
        _tuples_dequeued = 0;
        //cout << endl << endl << endl << " RETURN FALSE " << endl << endl;
        return false;
      }
    if (_end_time && ticks() >= _end_time)
        return false;

    return true;
}

// Due to circular dependency in QBox, Scheduler (ick)
inline void Scheduler::setBoxData(QBox& box, ptr<BoxData> data) {
    box._sched_data = data;
}

inline Scheduler::BoxData* Scheduler::getBoxData(const QBox& box) const {
    return box._sched_data.get();
}

BOREALIS_NAMESPACE_END;

#endif

---