QueryProcessor.h

Go to the documentation of this file.

#ifndef BOREALIS_QUERYPROCESSOR_H
#define BOREALIS_QUERYPROCESSOR_H

#include "AuroraNode.h"
#include "BasicComponent.h"
#include "DataPath.h"
#include "StreamState.h"
#include "QueryState.h"
#include "Recovery.h"
#include "ConsistencyMngr.h"

#include "QBoxState.h"

#include "CP.h"

BOREALIS_NAMESPACE_BEGIN;

class QueryProcessor : public BasicComponent {

  public:
  
    QueryProcessor(string id, InetAddress data_add, 
                   int ha_interval = HA_INTERVAL,
                   bool no_dups = false);
    virtual ~QueryProcessor();

    // RPC: #include "MTuple.h"
    // RPC: #include "Name.h"
    // RPC: #include "Query.h"
    // RPC: #include "Stats.h"
    // RPC: #include "Recovery.h"
    // RPC: #include "StreamEvent.h"
    // RPC: #include "BoxPackage.h"
    
    RPC<Query> typecheck(Query query);

    AsyncRPC<void> setup_query(Query query);

    RPC<Query> typecheck_and_setup(Query query);


    AsyncRPC<void> load_box(string file_path);

    RPC< vector<Query> > choke_queries(vector<Query> queries);

    RPC<void> resume_queries(vector<Query> queries);

    RPC<Query> pack_query(Query query);

    RPC< vector<Query> > pack_queries(vector<Query> queries);

    RPC<void> remove_query(Query query); 


    RPC<void> replace_query(vector<Name> old_queries, vector<Query> new_queries);

    RPC<void> set_query_status(Name name, QueryStatus status); 

    RPC<void> set_queries_status( vector<Name> name, QueryStatus status); 

    RPC<void> subscribe(Subscription sub, unsigned int add_or_remove);

    RPC< vector<Subscription> > subscribe_many(vector<Subscription> sub, unsigned int add_or_remove);


    RPC< vector<Subscription> > get_subscriptions(vector<Name> streams);

    RPC<vector<Stats> > get_stats();

    RPC<double> get_sel();

    RPC<void> create_stream(StreamDef stream);

    RPC<void> create_cpview(CPViewDescription view_desc, StreamDef streamdef);

    RPC<void> update_stream(StreamDef old_sd, StreamDef new_sd);

    RPC<void> ack(MedusaID node, StreamID id, string last_tuple);

    RPC<void> trim(MedusaID node, StreamID id, string last_tuple);

    RPC<void> set_recovery_method(int method);
    
    RPC<void> set_primary_status(bool status);

    RPC<void> set_secondaries(vector<MedusaID> secondaries);

    RPC<void> set_replicas(vector<MedusaID> replicas);

    RPC<void> checkpoint(vector<StreamEvent> tuples_to_checkpoint, map<StreamID,string> tuples_to_trim);


    NMSTL_RPC_OBJECT(QueryProcessor);

 private: // methods

    void init();
    void in_thread_init();
    Status do_subscribe(Subscription sub, unsigned int add_or_remove);
    Status local_subscribe(Name stream_name, unsigned int add_or_remove, MedusaID fast_dst = MedusaID());
    Status local_typecheck(Query& new_query);
    Status local_setup_query(Query& query);
    Status local_load_box(string file_path);
    Status local_choke_query(Query& query);
    Status local_resume_query(Query& query);
    Status local_pack_query(Query& query);
    Status local_remove_query(Query query);
    Status local_replace_query(vector<Name> old_queries, vector<Query> new_queries);
    Status local_set_query_status(Name name, QueryStatus status);
    Status local_start_query(QueryState& qs);
    Status check_dependencies_moved_query(Query query, MedusaID new_location);
    Status subscribe_self(MedusaID owner, ptr<StreamState> st, bool with_history);
    Status unsubscribe_self(ptr<StreamState> st);
    void update_stream_defs(Query& new_query, 
                            map<string,TupleDescription>& simple_stream_defs, 
                            vector<Name>& cp_ids);
    void update_stats(Time start_time);
    void do_ha_tasks();
    void send_checkpoints();
    void checkpoint_resp(MedusaID secondary, vector<StreamEvent> checkpointed_tuples, RPC<void> result);
    void send_downstream(vector<StreamEvent> tuples_to_send);

    void handle_subscription_response(ptr<StreamState>, 
                                      unsigned int add_or_remove,
                                      RPC<void>);

    Status delete_query_state(ptr<QueryState> qs);
    Status remove_from_engine(string as_xml);
    Status update_local_streams(Query& query);
    
    void send_acks();
    
    void send_ack(MedusaID node, StreamID stream_id, string tuple);
    
    void send_queue_trimming_message(MedusaID node, StreamID stream_id, string tuple);

    void handle_ack_response(MedusaID node, StreamID id, string last_tuple, RPC<void> result);
    void handle_trim_response(MedusaID node, StreamID id, string last_tuple, RPC<void> result);
    
 private: // fields

    AuroraNode m_aurora;
    bool m_aurora_started;

    InetAddress m_data_add;

    DataPath m_data_path;

    vector<Name> m_injected_streams;

    typedef map<Name, ptr<QueryState> > QueryStates;
    QueryStates m_queries;

    typedef map<Name, ptr<StreamState> > StreamStates;
    StreamStates m_streams;
  
    // For load management (measuring load)
    Time m_start_period;

    vector<MedusaID> m_secondary_ids;
    bool m_clear_secondary_endpoints;

    map<MedusaID,bool> m_pending_ack;
    map<MedusaID,bool> m_pending_trim;

    int m_ha_interval;

    Time m_next_time_stats_computed;

    typedef vector< pair< ptr<StreamState> ,string> > TrimPairs;

    friend class ConsistencyMngr;

    ConsistencyMngr m_constman;

};

BOREALIS_NAMESPACE_END;

#endif

---