Admin.h

Go to the documentation of this file.

#ifndef BOREALIS_ADMIN_H
#define BOREALIS_ADMIN_H

#include "BasicComponent.h"
#include "Schema.h"
#include "Name.h"
#include "StreamDef.h"
#include "PendingQuery.h"
#include "Query.h"
#include "CPViewDescription.h"

BOREALIS_NAMESPACE_BEGIN

// create_query: the actual request from the clients.  Sends lookups
//     for input streams; lookup responses go to create_query_2.
//
// create_query_2: handles a lookup response.  When all responses
//     are received, calls distribute_query
//
// distribute_query: sends the query to the local QP for schema
//     resolution
//
// distribute_query_2: handles result of schema resolution. 
//     Sets-up query locally (no automatic distribution here)
//
//
class Admin : public BasicComponent {
  public:

    Admin(string id) : BasicComponent(id,"Admin") {}
    ~Admin() {}

    // RPC: #include "Schema.h"
    // RPC: #include "StreamDef.h"
    // RPC: #include "Query.h"
    // RPC: #include "Name.h"
    // RPC: #include "Stats.h"
    // RPC: #include "CPViewDescription.h"

    // Client interface
    AsyncRPC<void> create_schema(Schema schema);
    AsyncRPC<void> create_stream(StreamDef streamdef);
    AsyncRPC<void> create_cp(CPViewDescription view_desc, StreamDef streamdef);
    AsyncRPC<void> create_query(Query query);
    AsyncRPC<void> create_query_xml(string query);
    AsyncRPC<void> set_query_status(Name name, QueryStatus status);
    AsyncRPC<void> set_queries_status(vector<Name> name, QueryStatus status);

    // (Un)subscribe to receive events on a stream 
    AsyncRPC<void> subscribe(Subscription sub, unsigned int add_or_remove);
    AsyncRPC<void> subscribe_many(vector<Subscription> sub, unsigned int add_or_remove);

    // The two methods below are for batch processing
    AsyncRPC<void> create_and_start_queries(vector<Query> queries);
    AsyncRPC<void> queries_and_subscriptions(vector<Query> queries,vector<Subscription> subs,
                                             unsigned int add_or_remove);
    // Move load described by boxes to partner
    AsyncRPC<void> move_load(vector<Name> query_names, string partner); 

    // Hunk management for fine-grained box movement.
    AsyncRPC<void> split_query(Name query_hunk_name, map<Name, vector<Name> > boxes_per_hunk);
    AsyncRPC<void> merge_query(vector<Name> existing_query_hunk_names, Name new_query_hunk_name);


    MultiRPC<InputStreamInfo> add_stream_observer();

    MultiRPC<Query> add_query_observer();

    MultiRPC< vector<Subscription> > add_sub_observer();

    NMSTL_RPC_OBJECT(Admin);

 protected:
    void in_thread_init() {}
    
 private:

    void empty_cleanup();

    void create_stream_2(AsyncRPC<void> completion, ptr<StreamDef> sd, RPC<void> result);  

    ptr<StreamDef> Admin::best_match(vector< ptr<Object> > list, StreamDef& target);
    void create_query_2(ptr<PendingQuery> pending, vector< ptr<StreamDef> > incomplete_stream_defs,
                        RPC< vector< vector< ptr<Object> > > > streamdefs);
    void start_created_query_from_batch(ptr<PendingBatch> batch_queries, 
                                        Name name, RPC<void> result);
    void created_query_from_batch(ptr<PendingBatch> batch_queries, 
                                  Name name, RPC<void> result);
    void queries_and_subscriptions_2(AsyncRPC<void> completion, vector<Subscription> subs,
                                     unsigned int add_or_remove, RPC<void> result);

    void distribute_query(ptr<PendingQuery>);
    void distribute_query_2(ptr<PendingQuery> pending, RPC<Query> typed_query);
    void register_query_streams_2(ptr<PendingQuery> pending, RPC<void> result);
    void register_responsability_response(ptr<Query> query, RPC<void> result);
    void remove_responsability_response(Name item_name, RPC<void> result);
    void remove_responsabilities_response(vector<Name> item_names, RPC<void> result);
    void subscribe_remote(AsyncRPC<void> completion, Subscription sub,
                     unsigned int cmd,RPC< vector< ptr<Object> > > obj);
    void subscribe_2(AsyncRPC<void> completion, unsigned int cmd, int nb_subscriptions,
                     RPC<vector<Subscription> > successful);
    void subscribe_3(AsyncRPC<void> completion, int nb_subscriptions, RPC<void> res);
                
    struct MoveData {
        vector<Name> query_names;
        ptr< vector<Query> > queries;
        ptr< vector<Subscription> > subs;
        MedusaID dst;
        MoveData(vector<Name> qn, ptr< vector<Query> > q, ptr< vector<Subscription> > s, MedusaID d) 
            : query_names(qn), queries(q), subs(s), dst(d) {}

    };

    // Move a set of queries to another node
    void move_load_2(MoveData data, AsyncRPC<void> completion, RPC< vector<Query> > choked_queries);
    void move_load_3(MoveData data, AsyncRPC<void> completion, RPC< vector<Query> > updated_queries);
    void move_load_4(MoveData data, AsyncRPC<void> completion, RPC<void> result);
    void move_load_5(MoveData data, AsyncRPC<void> completion, RPC<void> result);
    void move_load_6(MoveData data, AsyncRPC<void> completion, RPC<void> result);

    // Split a query
    void split_query2(Name query_hunk_name, vector<ptr<Query> > new_hunk_ptrs, AsyncRPC<void> completion, RPC<void> result);

    // Merge a query
    void merge_query2(vector<Name> existing_query_hunk_names, ptr<Query> new_query_hunk, AsyncRPC<void> completion, RPC<void> result);

    void notify_stream_observers(ptr<StreamDef> sd, vector< ptr<Object> > alts = vector< ptr<Object> >());

    void notify_query_observers(ptr<Query> query);

    void notify_sub_observers(vector<Subscription> subs);

 public:
    typedef map<Name, ptr<PendingQuery> > PendingQueries;
    typedef map<Name, ptr<StreamDef> > StreamMap;
    typedef map<Name, vector< ptr<StreamDef> > > StreamVersionsMap;
    typedef map<Name, ptr<Query> > LocalQueries;
    
    // For each stream name, we hold a map of subscription names to subscriptions themselves
    typedef map<Name, ptr<Subscription> > SubscriptionMap;
    typedef map<Name, SubscriptionMap > Subscriptions;

    typedef map<Name, ptr<CPViewDescription> > CPViewDescriptionMap;

 private:
    // Queries that are (being) setup. Once, setup, we remove queries from this data structure
    PendingQueries pending_queries;
        
    // Once set-up, we only remember the hunks that are running locally
    LocalQueries m_local_queries;

    // Cache of locally owned streams
    StreamMap m_local_streams;

    // locally owned cps
    CPViewDescriptionMap m_local_cpviews;

    // As a first try we will keep info about stream versions in a separate structure
    StreamVersionsMap local_stream_versions;
    
    // Cache of responsabilities
    Subscriptions m_subscriptions;

    // List of Observers for new input streams
    typedef vector< MultiRPC<InputStreamInfo> > StreamObserversList;
    StreamObserversList m_stream_observers;

    // List of Observers for new queries
    typedef vector< MultiRPC<Query> > QueryObserversList;
    QueryObserversList m_query_observers;

    // List of Observers for new subscriptions
    typedef vector< MultiRPC< vector<Subscription> > > SubObserversList;
    SubObserversList m_sub_observers;

};


typedef Callback< void > CleanUpMethod;

class ParallelActions {
 public:
    ParallelActions(AsyncRPC<void> completion, 
                    int nb_parallel_tracks, 
                    string msg,
                    CleanUpMethod clean_up = CleanUpMethod()) 
        : m_completion(completion), 
        m_pending(nb_parallel_tracks), 
        m_msg(msg),
        m_already_failed(false),
        m_actions_results(nb_parallel_tracks, true),
        m_clean_up_method(clean_up) {}

    void clean_up() { 
        WARN << "Should invoke a callback to clean-up state for " 
             << m_msg << " and result vector " << m_actions_results; 
        if ( m_clean_up_method )
            m_clean_up_method();
    }
    
    string as_string() const {
        return m_msg;
    }

    AsyncRPC<void> m_completion;
    int m_pending;
    string m_msg;
    bool m_already_failed;        
    vector<bool> m_actions_results;
    CleanUpMethod m_clean_up_method;
};

NMSTL_TO_STRING(ParallelActions);


class ParallelActionsMgr {
 public:
    static void action_result(ptr<ParallelActions> p, int action_nb, RPC<void> result) { 
        
        WARN << "Number pending is " << p->m_pending;
        p->m_pending--;
        
        // Action has failed
        if (!result.valid() ) {
            
            // Make a note of the failure in the vector of results
            p->m_actions_results[action_nb] = false;
            
            // Return failure message to client
            if ( ! p->m_already_failed ) {
                p->m_already_failed = true;
                ERROR << "One of several actions failed in group " << p->m_msg;
                p->m_completion.post(RPCFault(p->m_msg + result.stat()));
            }
        }
        // If all actions have succeeded, we can return success message to client
        else if ( p->m_pending <= 0 ) {                
            p->m_completion.post(true);
        }
        
        // If we're all done and something failed, then we need to clean-up
        if ( (p->m_pending <= 0) && (p->m_already_failed) ) {
            p->clean_up();
        }
    }
};


BOREALIS_NAMESPACE_END

#endif

---