TupleQueue.h

Go to the documentation of this file.

#ifndef TUPLEQUEUE_H
#define TUPLEQUEUE_H

#include "PagePool.h"
#include "TuplesAvailableMailbox.h"

BOREALIS_NAMESPACE_BEGIN;

class Stream;
class QBox;

class TupleQueueListener {
  public:
    virtual ~TupleQueueListener();

    virtual void notify(const TupleQueue& stream) = 0;

    QBox* getBox() { return _box_to_run; }
    void setBox(QBox* box) { _box_to_run = box; }

    void suspend() { _suspended = true; }
    void resume () { _suspended = false; }

  protected:
    TupleQueueListener();
    QBox* _box_to_run;
    bool _suspended;
};



class TupleQueue {
public:
    TupleQueue(PagePool& pool, unsigned int tuple_size);

    ~TupleQueue();

    void setStream(Stream *stream) { _stream = stream; }
    Stream *getStream() { return _stream; }


    /*
      Added for Connection point support
    */
    // empties the tuple queue by freeing all pages and resetting pointers
    void empty();

    // puts the vector of tuples in the tuple queue
    void enqueueMultiple(vector<char*> tuples);

    // returns all the tuples in the tuple queue as a vector of tuples
    vector<char*> dequeueAll();


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

        void *tuple() const {
            assert(_q);
            return _q->_enq_location;
        }

        EnqIterator& operator ++ ();

    private:
        explicit EnqIterator(TupleQueue &q) : _q(&q) {}

        EnqIterator operator ++ (int);

        TupleQueue *_q;

        friend class TupleQueue;
    };
    friend class EnqIterator;
    
    class DeqIterator {
    public:
        DeqIterator() : _q(0) {}

        bool avail() const {
            assert(_q);

            return _q->_deq_location != _q->_enq_location;
        }

        const void *tuple() const {
            assert(avail());
            return _q->_deq_location;
        }

        DeqIterator& operator ++ ();

    private:
        explicit DeqIterator(TupleQueue &q) : _q(&q) {}

        DeqIterator operator ++ (int);

        TupleQueue *_q;

        friend class TupleQueue;
    };
    friend class DeqIterator;

    class SeekIterator {
   public:
        SeekIterator() : _q(0) {}
     
        bool avail() const {
            assert(_q);
            return _seek_location != _q->_enq_location;
        }
 
      // in SeekIterator this would not work right. Not sure if I can remove
      // this function.
      //bool avail() const {
      //    assert(_q);

      //return true; //_q->_deq_location != _q->_enq_location;
      //}
      
      const void *tuple() const {
            return _seek_location;
        }
      
      SeekIterator& operator ++ ();
      SeekIterator& operator -- ();

    private:
        explicit SeekIterator(TupleQueue &q) : _q(&q) 
      { 
         _seek_location = _q->_deq_location; 
         _seek_page = _q->_deq_page; 

         size_t page_size = _q->_pool.page_size();
         size_t tuple_size = _q->_tuple_size;
         int tuples_per_page = (page_size - sizeof(PageHdr)) / tuple_size;
         _offset_last_tuple_on_page = sizeof(PageHdr) + tuple_size*(tuples_per_page-1);

      }

      SeekIterator operator ++ (int);

      SeekIterator operator --(int);
      
      TupleQueue *_q;

      // this is where our seek iterator is pointing
      void *_seek_location;
      void *_seek_page;

      size_t _offset_last_tuple_on_page;
      friend class TupleQueue;
    };

    class RSeekIterator {
    public:
        RSeekIterator() : _q(0) {}
     
        bool avail() const {
            assert(_q);
            return _seek_location != _q->_deq_location;
        }
        
        const void *tuple() const {
            return _seek_location;
        }
      
        RSeekIterator& operator ++ ();
        RSeekIterator& operator -- ();

    private:
        explicit RSeekIterator(TupleQueue &q) : _q(&q) {            
            
            size_t page_size = _q->_pool.page_size();
            size_t tuple_size = _q->_tuple_size;
            int tuples_per_page = (page_size - sizeof(PageHdr)) / tuple_size;
            _offset_last_tuple_on_page = sizeof(PageHdr) + tuple_size*(tuples_per_page-1);

            _seek_location = (char*)_q->_enq_location - tuple_size; 
            _seek_page = _q->_enq_page; 

            if ( (char*)_seek_location < (char*)_seek_page + sizeof(PageHdr)) {
                if (_seek_page != _q->_deq_page) {         
                    _seek_page = ((PageHdr*)_seek_page)->prev;
                    _seek_location = (char*)_seek_page + _offset_last_tuple_on_page; 
                } else { // Nothing there
                    _seek_page = _q->_deq_page; 
                    _seek_location = _q->_deq_location;
                }
            }
        }

        RSeekIterator operator ++ (int);

        RSeekIterator operator -- (int);
      
        TupleQueue *_q;

        // this is where our seek iterator is pointing
        void *_seek_location;
        void *_seek_page;

        size_t _offset_last_tuple_on_page;
        friend class TupleQueue;
    };

    RSeekIterator rseek_iterator() { return RSeekIterator(*this); }

    SeekIterator seek_iterator() { return SeekIterator(*this); }

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

    DeqIterator deq_iterator() { return DeqIterator(*this); }

    void dump() const;

    int size() const { return _enq_count - _deq_count; }

    size_t getNumTuplesDequeued() const { return _deq_count; }

    size_t getNumTuplesEnqueued() const { return _enq_count; }

    void addListener(TupleQueueListener* listener) {
        _listeners.push_back(listener);
    }

    void removeListener(TupleQueueListener* listener) {
        erase_in(_listeners, listener);
    }

    unsigned int getTupleSize() const {
        return _tuple_size;
    }

   void setEnqNotifyTarget(TuplesAvailableMailbox * pMailbox);

   void notifyEnq();

  private:
   TuplesAvailableMailbox * _pMailbox;

    struct PageHdr {
        PageHdr *next;
        PageHdr *prev;
    };

    PagePool &_pool;
    unsigned int _tuple_size;
    unsigned int _tuples_per_page;

    unsigned int _enq_count;
    unsigned int _deq_count;

    PageHdr *_enq_page;

    void *_enq_location;

    PageHdr *_deq_page;

    void *_deq_location;

    Stream *_stream;

    typedef vector<TupleQueueListener*> Listeners;
    Listeners _listeners;
};

inline TupleQueue::EnqIterator& TupleQueue::EnqIterator::operator ++ ()
{
    assert(_q);

    void *next_enq_location = (char*)_q->_enq_location + _q->_tuple_size;
    if ((char*)next_enq_location + _q->_tuple_size > (char*)_q->_enq_page + _q->_pool.page_size()) {
        // Allocate new page.  Note that the DeqIterator
        // cannot ever advance to the new page until we set
        // _enq_location (at the very end of this method)

        PageHdr *new_page = (PageHdr*)_q->_pool.alloc();
        new_page->next = 0;

        new_page->prev = _q->_enq_page;
        _q->_enq_page->next = new_page;
        _q->_enq_page = new_page;

        next_enq_location = (char*)new_page + sizeof(PageHdr);
    }

    _q->_enq_location = next_enq_location;
    ++(_q->_enq_count);
    return *this;
}

inline TupleQueue::DeqIterator& TupleQueue::DeqIterator::operator ++ ()
{
    assert(avail());

    void *next_deq_location = (char*)_q->_deq_location + _q->_tuple_size;
    if ((char*)next_deq_location + _q->_tuple_size > (char*)_q->_deq_page + _q->_pool.page_size()) {
        // Advance to next page; can free this page
        PageHdr *next_page = _q->_deq_page->next;
        _q->_pool.free(_q->_deq_page);
        _q->_deq_page = next_page;

        next_deq_location = (char*)next_page + sizeof(PageHdr);
    }

    _q->_deq_location = next_deq_location;
    ++(_q->_deq_count);
    return *this;
}

inline TupleQueue::SeekIterator& TupleQueue::SeekIterator::operator ++ ()
{
    void *next_seek_location = (char*)_seek_location + _q->_tuple_size;
    if ((char*)next_seek_location + _q->_tuple_size > (char*)_seek_page + _q->_pool.page_size()) {
        _seek_page = ((PageHdr *)_seek_page)->next;
        next_seek_location = (char*)_seek_page + sizeof(PageHdr);
    }

    _seek_location = next_seek_location;
    return *this;
}

inline TupleQueue::SeekIterator& TupleQueue::SeekIterator::operator -- ()
{
    void *next_seek_location = (char*)_seek_location - _q->_tuple_size;
    if ((char*)next_seek_location < (char*)_seek_page + sizeof(PageHdr) ) {
        _seek_page = ((PageHdr *)_seek_page)->prev;
        next_seek_location = (char*)_seek_page + _offset_last_tuple_on_page; 
    }

    _seek_location = next_seek_location;
    return *this;
}

inline TupleQueue::RSeekIterator& TupleQueue::RSeekIterator::operator ++ ()
{
    void *next_seek_location = (char*)_seek_location - _q->_tuple_size;
    if ((char*)next_seek_location < (char*)_seek_page + sizeof(PageHdr) ) {
        _seek_page = ((PageHdr *)_seek_page)->prev;
        next_seek_location = (char*)_seek_page + _offset_last_tuple_on_page; 
    }

    _seek_location = next_seek_location;
    return *this;
}

inline TupleQueue::RSeekIterator& TupleQueue::RSeekIterator::operator -- ()
{
    void *next_seek_location = (char*)_seek_location + _q->_tuple_size;
    if ((char*)next_seek_location + _q->_tuple_size > (char*)_seek_page + _q->_pool.page_size()) {
        _seek_page = ((PageHdr *)_seek_page)->next;
        next_seek_location = (char*)_seek_page + sizeof(PageHdr);
    }

    _seek_location = next_seek_location;
    return *this;
}

BOREALIS_NAMESPACE_END;

#endif

---