SJoinQBox.h

Go to the documentation of this file.

#ifndef SJOIN_QBOX_H
#define SJOIN_QBOX_H

#include "JoinQBox.h"
#include "../ConsistencyMngr.h" // For benchmark types
#include <fstream>

BOREALIS_NAMESPACE_BEGIN;

class SJoinQBox : public JoinQBox
{

 public:
    ~SJoinQBox(); // So we can log the stats in a file

 protected:
    void setupImpl() throw (AuroraException);
    void initImpl() throw (AuroraException);
    void runImpl(QBoxInvocation&) throw (AuroraException);
        
  private:
    
    // Using a struct instead of a pair so we can do tests like
    // if ( _state_bounds[tuple_id] )
    struct StateBound {        
        int32 _left;
        int32 _right;
        StateBound(int32 left = -1, int32 right = -1) 
            : _left(left), _right(right) {}
        operator const void *() const {
            return ((_left == -1) || (_right == -1)) ? 0 : this;
        }       
    };

    static const int MAX_HISTORY = 101001000; // In tuples
    static const int PAGE_SIZE = 16384;   // Arbitrary small value
    
    static const int TWO_INPUTS = 2; // There are 2 input streams JoinQBox::LEFT && RIGHT

    void handleInputTuple(const Tuple& tuple, int stream_id, bool& outputted); 

    // Punctuation
    void handlePunctuation(const Tuple& left_tuple, const Tuple& right_tuple);
    void passThrough(const Tuple& tuple);
    Timestamp findMinTimestamp(int stream_id);
    int32 findOldestTupleId(int stream_id);
    void emitPunctuationTuple(Timestamp min_timestamp);

    // Undo-redo recovery
    void handleUndo();
    void scanForUndo(int stream_id, bool& has_undo, int32& undo_id);
    void dequeueUntilUndo(int stream_id, int32 undo_tuple_id);
    void checkIterator(int stream_id);
    void undoState(bool has_undo[], int32 undo_tuple_id[]);
    StateBound findBound(TupleQueue::RSeekIterator ri[], bool has_undo[], int32 undo_tuple_ids[]);
    int rollback(TupleQueue::RSeekIterator ri[], int32 bound_ids[]);
    void clearState();
    void rollforward(TupleQueue::RSeekIterator ri[], int counter);

    void emitTuple(const Tuple& undo_tuple, TupleType type);

    void emitUndoRedoTuples(const Tuple& tuple);

    // Normal operation
    void storeTupleForUndo(const Tuple& tuple, int stream_id);
    void saveStateBound(const Tuple& tuple, int stream_id);
    void cutHistory(int stream_id);
    void processTuple(const Tuple& tuple, int stream_id, bool& outputted);

    int _output_tuple_size;
    int _input_tuple_size[TWO_INPUTS]; 
    ptr<TupleQueue> _input_history[TWO_INPUTS]; 
    int _max_history; // Maximum undo/redo size
    ptr<PagePool> _pool; // Small page pool because TupleQueues use page pools

    bool _punctuation[TWO_INPUTS]; // We wait to receive both punctuations before producing one ourselves
    Timestamp _punctuation_time[TWO_INPUTS]; // Timestamp received as punctuation

    // Information about the state of the join
    size_t _buffer_size[TWO_INPUTS];

    typedef map< int32, StateBound > TupleStateBound;
    TupleStateBound _state_bounds;
    //int32 _oldest_id[TWO_INPUTS];

    int _state_bound_interval;
    Timestamp _last_state_bound_time;
    MicroBenchmarkType _benchmark;

    AURORA_DECLARE_QBOX(SJoinQBox, "sjoin");

    // Temporary for experiments
    ofstream _log_file;
    vector<long long> _measurements; //Sorted list of measurements
};


BOREALIS_NAMESPACE_END;

#endif // SJOIN_QBOX_H

---