SUnionQBox.h

Go to the documentation of this file.

#ifndef SUNION_QBOX_H
#define SUNION_QBOX_H

#include "UnionQBox.h"
#include "JoinQBox.h"
#include <vector>
#include "SControlQBox.h"

BOREALIS_NAMESPACE_BEGIN;

// --------------------------------------------------
// Just a simple buffer for tuples. Grows automatically. XXX should add iterators.
class TuplesBuffer {
 public:
    static const int DEFAULT_SIZE = 1000;
    TuplesBuffer(size_t tuple_size);
    void insertTuple(const Tuple& data);           
    int nbTuples() { return _cur_nb_tuples; }
    const Tuple tupleAt(int i);
    // Deletes all tuples from this buffer
    void clear();
    // Deletes all tuples following the one with given tuple id
    void clearAfter(int tuple_id);
    // Buffer is unstable if earliest tuple is unstable. It is stable otherwise
    bool isStableBuffer();
 private:
    size_t _tuple_size;
    int _cur_nb_tuples;
    int _max_nb_tuples;
    ptr<dynbuf> _tuples;
};



// --------------------------------------------------
class SUnionQBox : public SControlQBox
{

    friend class SUninBoxState;

 public:

    static TupleDescription createControlOutputDescription();

 protected:
    virtual void setupImpl() throw (AuroraException);
    virtual void initImpl() throw (AuroraException);
    virtual void runImpl(QBoxInvocation&) throw (AuroraException);
    void pureUnionRunImpl(QBoxInvocation&) throw (AuroraException);

    // packing and unpacking states
    void setPendingBoxState(ptr<AbstractBoxState> packed_box);
    bool isStableForCheckpoint(); 
    ptr<AbstractBoxState> packState();
    void unpackState(ptr<AbstractBoxState> box_state);

 private:

    // XXX No good reason why this specific value, we just want to avoid hogging the CPU too long
    static const int MAX_BUCKETS_PER_ITERATION = 5;
    
    // We block for _max_delay = ALPHA_BLOCK * max-delay 
    static const double MAX_ALPHA = 0.9; // For regular blocking
    //static const double MAX_ALPHA = 1.0; // For benefits1 benchmark 

    // Once we are in Unstable state, then we only block for a very short time
    static const double MIN_ALPHA = 0.1;
    //static const double MIN_ALPHA = 1.0; // For benefits1b benchmark

    // There is a race condition between checkpoints and info about unstable state, so we
    // erase buckets only after some slack
    static const int SLACK = 10; 

    class Bucket;

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

    void applyUndo(const Tuple& tuple, int stream_id, long long interval_start);

    void undoRedo(long long interval_start);

        //void produceRedoTuples(long long interval_start);

    //void completedRecovery();

    void insertTupleIntoBucket(const Tuple& tuple, int stream_id, long long interval_start);

    bool forwardWithoutWaiting();

    void produceOutput();
    bool processOutputStableBucket(Bucket& bucket, long long interval_nb);
    bool processOutputUnStableBucket(Bucket& bucket, long long interval_nb);

    void producePunctuation(Timestamp punctuation_time);

    void checkpoint();

        //void notifyOutputs();

    void emitTuples(Bucket& bucket, TupleType tuple_type);

        //void emitTuple(const Tuple& tuple);

        //void emitControlTuple(int stream_id, TupleType type, Timestamp data);

    void emitUndoTuple(const Tuple& tuple);

    
    long long timestampToIntervalNb(Timestamp timestamp);

    Timestamp intervalNbToTimestamp(long long interval_nb);

    static const long MILLION = 1000000;
    static const long THOUSAND = 1000;

    unsigned int _numberOfInputs;

    int _interval;  // In msecs (size of buckets)

    int _max_delay; // In msecs (max time we can wait for input tuples
    double _actual_delay; // In msecs (time we are actually going to wait for)
    bool _stagger; // Should we accelerate

    // The operator is a state machine
    struct State {
        public:
        State () : _state_value(STABLE_STATE), _unstable_start(Timestamp::now()),
                   _unstable_interval_nb(0) {}

        bool isStable() { return _state_value == STABLE_STATE; }

        bool isUnstable() { return _state_value == UNSTABLE_STATE; }

        bool canProcessUnstable();

        void processingUnstableBucket(Timestamp unstable_start, long long interval_nb); 

        void undoing(); 

        void newStableBucket();

        bool somethingToCorrect();
        Timestamp correctSinceWhen(); // Time of beginning of unstability
        long long firstUnstableInterval(); // First unstable bucket

        void reconciling();
        bool isReconciling();
        void reconcilingRollforward();
        bool isReconcilingForward();

        void reconciled();

        void noReconciliation();
        
        private:
        static const int STABLE_STATE = 0;      // Produce tuples in their final form
        static const int UNSTABLE_STATE = 1;    // Produce tuples to avoid delaying
        static const int UNDONE_STATE = 2;     // In the process of undoing/redoing buffer content
        static const int CONVERGED_STATE = 3;   // Produce correct tuple but left-over incorrect tuples remain 
        static const int RECONCILING_STATE_ROLLBACK = 4; // Correcting previous state (rolling back) 
        static const int RECONCILING_STATE_ROLLFORWARD = 5; // Correcting previous state (rolling forward)
        
        int _state_value;
        Timestamp _unstable_start;
        long long _unstable_interval_nb;
  
    };    
    State _state;    

    // A bucket associates windows of tuples received with their input stream (identified by port nb)
    typedef map<int, ptr<TuplesBuffer> > TuplesBufferSet;
    struct Bucket {

        Bucket(int nb_buffers, int tuple_size);

        TuplesBufferSet _buffers;

        bool _contains_unstable_tuples;
        map<int, bool> _punctuations;
        Timestamp _creation_time;

        Timestamp getCreationTime();
        void clear();
        bool punctuated();
        // Exhaustively checks all buffers for unstable tuples
        void resetStability();
    };

    // There is one bucket per time period. Periods (or intervals) are in msecs, same unit as _interval
    typedef map<long long, ptr<Bucket> > Buckets;

    // Set of buckets 
    Buckets _buckets;
    long long _current_interval_nb;

    int32 _last_stable_tuple_id;

    // Checkpointed state
    long long _checkpointed_interval_nb;

    // Last time operator emmitted punctuation tuple
    Timestamp _last_punctuation_time;

    // The benchmark that we are running
    MicroBenchmarkType _benchmark;

    bool _one_iteration_after_queues_cleared;

    // If there's a single input stream, should we sort or not?
    bool _sort;

    bool _should_reconcile;

    AURORA_DECLARE_QBOX(SUnionQBox, "sunion");
};

class SUnionBoxState : public AbstractBoxState
{
 public:
    SUnionBoxState() : _interval_nb(INIT) {}
    SUnionBoxState(long long interval_nb) : _interval_nb(interval_nb) {}
    ~SUnionBoxState() {}

    long long getIntervalNb() { return _interval_nb; }
    NMSTL_SERIAL_SUBCLASS(SUnionBoxState, AbstractBoxState, << _interval_nb);

 private:
    long long _interval_nb;
};

BOREALIS_NAMESPACE_END;

#endif

---