ConsistencyMngr.h

Go to the documentation of this file.

#ifndef BOREALIS_CONSISTENCYMNGR_H
#define BOREALIS_CONSISTENCYMNGR_H

#include "common.h"
#include "Query.h"
#include "StreamEvent.h"
#include "Timestamp.h"
#include "Tuple.h"

#include <NMSTL/tqueuerpc>
#include <NMSTL/tcprpc>
#include <NMSTL/xmlrpc>
#include <fstream>

BOREALIS_NAMESPACE_BEGIN;

class QueryProcessor;

// Type of microbenchmark that we are running
class MicroBenchmarkType {

 public:

    enum Type {
        // Micro benchmarking the operator (one item at the time)
        NONE = 0,     // SJoin without anything else (like an ordinary join)
        TUPLE_TYPES,  // SJoin that looks at the message types before processing them
        UNDO_LOG,     // SJoin that looks keeps an undo log of variable size
        CHECKPOINT,   // SJoin interrupted by checkpoints

        // Micro benchmarking the sunion/soutput operators
        SUNION,       // Just add an sunion in front of an sjoin
        SUNION_PUNCTUATION,    // Just sunion and punctuation
        SOUTPUT,      // Just add an soutput at the end

        BLOCK,  // Never process any unstable bucket

        // Micro benchmarking the approach as a whole but don't move subscriptions
        ALL_TOGETHER_UNDO_REDO_NO_MOVE,
        ALL_TOGETHER_CHECKPOINT_REDO_NO_MOVE,

        // Micro benchmarking the approach as a whole
        ALL_TOGETHER_UNDO_REDO,
        ALL_TOGETHER_CHECKPOINT_REDO,

        // Finally, the complete adaptive algorithm
        ADAPTIVE


    };
    
    bool doKeepUndoLog() 
    { return ((_type == UNDO_LOG) 
              || (_type == ALL_TOGETHER_UNDO_REDO_NO_MOVE) 
              || (_type == ALL_TOGETHER_UNDO_REDO)
              || (_type == ADAPTIVE));
    }

    bool doSerialize() 
    { return ((_type == SUNION) || (_type == SUNION_PUNCTUATION) || (_type == BLOCK) 
              || (_type == ALL_TOGETHER_UNDO_REDO_NO_MOVE) || (_type == ALL_TOGETHER_CHECKPOINT_REDO_NO_MOVE)        
              || (_type == ALL_TOGETHER_UNDO_REDO) || (_type == ALL_TOGETHER_CHECKPOINT_REDO)
              || (_type == ADAPTIVE));        
    }

    bool doPunctuation() 
    { return ((_type == SUNION_PUNCTUATION) || (_type == BLOCK)
              || (_type == ALL_TOGETHER_UNDO_REDO_NO_MOVE) || (_type == ALL_TOGETHER_CHECKPOINT_REDO_NO_MOVE)
              || (_type == ALL_TOGETHER_UNDO_REDO) || (_type == ALL_TOGETHER_CHECKPOINT_REDO)
              || (_type == ADAPTIVE));        
              
    }

    bool doCheckpoint() 
    { return ((_type == CHECKPOINT) 
              || (_type == ALL_TOGETHER_CHECKPOINT_REDO_NO_MOVE) 
              || (_type == ALL_TOGETHER_CHECKPOINT_REDO)
              || (_type == ADAPTIVE)
              );
    }

    bool doBlock()
    { return (_type == BLOCK);
    }

    bool doForceMoveSubscriptions()
    { return ( (_type == ALL_TOGETHER_UNDO_REDO) || (_type == ALL_TOGETHER_CHECKPOINT_REDO));
    }

    bool doUndoReconciliation()
    { return ((_type == ALL_TOGETHER_UNDO_REDO_NO_MOVE) || (_type == ALL_TOGETHER_UNDO_REDO));
    } 

    bool doAdaptive()
    { return ( _type == ADAPTIVE);
    }

    MicroBenchmarkType(Type t = ALL_TOGETHER_UNDO_REDO) 
        : _type(t) 
    {
    }

    MicroBenchmarkType(string type) 
    {
       if ( type == "none" || type == "NONE" )
           _type = NONE;

       else if ( type == "tuple_types" || type == "TUPLE_TYPES" )
           _type = TUPLE_TYPES;

       else if ( type == "undo_log" || type == "UNDO_LOG" )
           _type = UNDO_LOG;

       else if ( type == "checkpoint" || type == "CHECKPOINT" )
           _type = CHECKPOINT;

       else if ( type == "sunion" || type == "SUNION" )
           _type = SUNION;

       else if ( type == "sunion_punctuation" || type == "SUNION_PUNCTUATION" )
           _type = SUNION_PUNCTUATION;

       else if ( type == "soutput" || type == "SOUTPUT" )
           _type = SOUTPUT;

       else if ( type == "block" || type == "BLOCK" )
           _type = BLOCK;

       else if ( type == "all_together_checkpoint_redo" || type == "ALL_TOGETHER_CHECKPOINT_REDO" )
           _type = ALL_TOGETHER_CHECKPOINT_REDO;

       else if ( type == "all_together_checkpoint_redo_no_move" || type == "ALL_TOGETHER_CHECKPOINT_REDO_NO_MOVE" )
           _type = ALL_TOGETHER_CHECKPOINT_REDO_NO_MOVE;

       else if ( type == "all_together_undo_redo" || type == "ALL_TOGETHER_UNDO_REDO" )
           _type = ALL_TOGETHER_UNDO_REDO;

       else if ( type == "all_together_undo_redo_no_move" || type == "ALL_TOGETHER_UNDO_REDO_NO_MOVE" )
           _type = ALL_TOGETHER_UNDO_REDO_NO_MOVE;

       else _type = ADAPTIVE;

   }

    string as_string( ) const {
        switch (_type) {
        case NONE:         return string("none"); 
        case TUPLE_TYPES : return string("tuple_types");
        case UNDO_LOG:     return string("undo_log");
        case CHECKPOINT:   return string("checkpoint");
        case SUNION:       return string("sunion");
        case SUNION_PUNCTUATION:           return string("sunion_punctuation");
        case SOUTPUT:      return string("soutput");
        case BLOCK:        return string("block");
        case ALL_TOGETHER_CHECKPOINT_REDO_NO_MOVE: return string("all_together_checkpoint_redo_no_move");
        case ALL_TOGETHER_CHECKPOINT_REDO: return string("all_together_checkpoint_redo");
        case ALL_TOGETHER_UNDO_REDO_NO_MOVE  : return string("all_together_undo_redo_no_move");
        case ALL_TOGETHER_UNDO_REDO  : return string("all_together_undo_redo");
        case ADAPTIVE :    return string("adaptive");
        default:           return string("adaptive");
        }
    }

    Type _type;
    
};

NMSTL_TO_STRING(MicroBenchmarkType);


class ConsistencyMngr {
 public:

    ConsistencyMngr(QueryProcessor& qp);

    ~ConsistencyMngr();

    void checkForControlStreams(Query& q);
    
    void set_replicas(vector<MedusaID> replicas);
    void add_subscription(Subscription sub);
    void remove_subscription(Subscription sub);

 private:

    Name findControlStreamName(const Box::StreamBindings& outputs);
    
    void monitorControlStream(Name control_stream_name);

    Status handleControlTuples(ptr<StreamEvent> event);

    void sendControlTuple(Name output_stream_name, TupleType type, Timestamp since);

    void startCheckpointing();

    void checkpoint();

    void launchReconciliation(Name control_stream, Timestamp time);
    void reconcile(Name control_stream, Timestamp time);
    void finishReconciliation();

    void updateSubHistory();
    void moveSubscriptionsIfNeeded();
    void releaseDataPathIfNeeded();
    void moveSubscriptionsAway();
    void moveSubscriptionsAway2(MedusaID replica, RPC< vector<Subscription> > result);
                                
    void moveSubscriptionsBack();
    void moveSubscriptionsBack2(RPC< vector<Subscription> > result);

    QueryProcessor& _qp;

    vector<MedusaID> _replicas;
    MedusaID _sub_temp_location;

    vector<Subscription> _subs;

    typedef map< Name, pair<Name, int> > IOControl;
    IOControl _io_control;
   
    bool _reconciliation_in_progress;

    static const int CHECKPOINT_INTERVAL = 1000;
    int _checkpoint_interval;
    Timestamp _last_checkpoint_time;
    bool _checkpointing_started;
    bool _suspend_checkpoints;


    static const int MAX_FAILURE = 30000;
    static const int MAX_NO_SEND_DELAY = 10000;
    static const int MAX_DELAY_DEFAULT = 1000; // In case we make the parameter optional
    static const double ALPHA_MAX = 0.6; // We never block for max_delay but a fraction of it    
    int _max_delay; // How much we can delay processing for
    double _actual_delay;
    Timestamp _failure_start_time;

    // All the stuff below is used to evaluate the approach
    MicroBenchmarkType _benchmark;
    ofstream _log_file;
    vector<long long> _measurements;
    int _state_size; 

};

BOREALIS_NAMESPACE_END;

#endif

---