Stream.h

Go to the documentation of this file.

#ifndef STREAM_H
#define STREAM_H

#include <vector>
#include <deque>
#include <sstream>
#include <sys/time.h>
#include "common.h"

using namespace std;

#include "LockHolder.h"
#include "TupleQueue.h"
#include "TupleDescription.h"
#include "TuplesAvailableMailbox.h"
#include "QBoxInputPort.h"
#include "QBoxOutputPort.h"
#include "StreamStatsSample.h"
#include "Params.h"
#include "CpuStats.h"
#include "Expression.h"
#include "common.h"

BOREALIS_NAMESPACE_BEGIN;

class AuroraNode;
class QBox;
//class CPoint;
class CP;

class Stream
{
  public:
    Stream(string name, const TupleDescription& desc, AuroraNode& node,
           const Params& params = Params())
        throw (AuroraException);

    ~Stream();

    string getName() const { return _name; }

    size_t getNumTuplesEnqueued() const { return _numTuplesEnqueued; }

    //void setConnectionPoint(CPoint *cpoint) { _cpoint = cpoint; }

    const TupleDescription& getTupleDescription() const { return _desc; }

    void addQueue(TupleQueue* q);

    void removeQueue(TupleQueue*);

    const vector<TupleQueue*>& getQueues() { return _queues; }

    void addDestPort(QBoxInputPort port);

    void removeDestPort(QBoxInputPort port);

    const vector<QBoxInputPort>& getDestPorts() { return _dest_ports; }

    void setSourcePort(QBoxOutputPort port);

    QBoxOutputPort getSourcePort() { return _source_port; }

#ifdef AURORA_RUNTIME_STATS
    void updateStats()
    {
        if (_statsUnreportedEnq == 0)
        {
            return;
        }

        timeval tm;
        gettimeofday(& tm, NULL);

        LockHolder lh(_streamStatsHistoryLock);

        if ((_streamStatsHistory.size() == 0) || (_streamStatsHistory.front()._tv_sec != tm.tv_sec))
        {
            _streamStatsHistory.push_front(StreamStatsSample(tm.tv_sec, _statsUnreportedEnq));

            if (_streamStatsHistory.size() > _statsHistoryLength)
            {
                _streamStatsHistory.pop_back();
            }
        }
        else
        {
            _streamStatsHistory.front()._tuplesEnqueued += _statsUnreportedEnq;
        }

        _statsUnreportedEnq = 0;
    }

    const deque<StreamStatsSample> & lockStreamStatsHistory()
    {
        _streamStatsHistoryLock.lock();
        return _streamStatsHistory;
    }

    void unlockStreamStatsHistory()
    {
        _streamStatsHistoryLock.unlock();
    }

    static string dumpStreamStatsHistory(const deque<StreamStatsSample> & h)
    {
        ostringstream os;
        for (deque<StreamStatsSample>::const_iterator pos = h.begin();
             pos != h.end(); ++pos)
        {
            os << pos->to_string() << endl;
        }
        os << "---------------------------";
        return os.str();
    }

#endif

    class EnqIterator
    {
      public:
        EnqIterator() : _stream(0) {}

        void *tuple() const { ASSERT(_stream); return _stream->_buf; }

        inline EnqIterator& operator ++ ()
        {
            ASSERT(_stream);
            _stream->enqueueTuple();
            return *this;
        }
        
      private:
        EnqIterator(Stream& stream) : _stream(&stream) {}
        Stream* _stream;

        friend class Stream;
    };
    friend class EnqIterator;

    EnqIterator enq_iterator() { return EnqIterator(*this); }    

    string as_string() const;

    enum SchemaRepresentation
    {
        SCHEMA_NAME,

        FULL_SCHEMA_IF_ANONYMOUS,

        FULL_SCHEMA
    };

    string as_xml(SchemaRepresentation rep = FULL_SCHEMA_IF_ANONYMOUS) const;

    void notifyEnq();

    void beginBuffering();

    void stopBuffering();

    const Params& getParams() const { return _params; }

    // methods to access CP on this stream
    void setCP(ptr<CP> cp); 
    ptr<CP> getCP(); 

  private:
    void enqueueTuple();
    //void enqFromCPoint(TupleQueue* q);

    // The AuroraNode whose query network uses this Stream.
    AuroraNode & _node;
    const CpuStats& _cpu_stats;

    // Buffer returned by the iterator
    void *_buf;

    // this is just a expression elimination. The function would
    // send the _buf to all outputs.
    void pushTuple( const void *_buf );

    string _name;
    size_t _numTuplesEnqueued;
    TupleDescription _desc;
    vector<TupleQueue*> _queues;
    vector<QBoxInputPort> _dest_ports;
    QBoxOutputPort _source_port;

    bool _can_drop;
    double _drop_threshold;
    ptr<Expression> _drop_expr;
    ptr<EvalContext> _drop_ctxt;

    // this flag tells the stream to start buffering tuples (instead of
    // forwarding them to queues) until further notice.
    bool _buffering; 

    // tuple queue to keep buffered tuples. Will use the same page pool
    // so that storage manager has no surprises
    TupleQueue *_tuple_buffer;

    Params _params;
    //CPoint *_cpoint;

    // pointer to the CP on this stream
    ptr<CP> _cp;

#ifdef AURORA_RUNTIME_STATS
    // The number of Enqueues that have occured on this stream, but which haven't yet
    // been reported by the updateStats() method.
    size_t _statsUnreportedEnq;

    // Contains the stats sample history for this Stream.  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<StreamStatsSample> _streamStatsHistory;

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

    // 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

};

BOREALIS_NAMESPACE_END;

#endif                              // STREAM_H

---