DataPath.h

Go to the documentation of this file.

#ifndef BOREALIS_DATAPATH_H
#define BOREALIS_DATAPATH_H

#include <NMSTL/callback>
#include <NMSTL/thread>
#include <NMSTL/tqueue>
#include <NMSTL/netioevent>
#include <NMSTL/serial>

#include "runtime/AuroraNode.h"
#include "Exceptions.h"

#include "MTuple.h"
#include "StreamState.h"
#include "DataHandler.h"
#include "Recovery.h"

#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>

BOREALIS_NAMESPACE_BEGIN;


class DataPath {


 public:
    DataPath(AuroraNode& node, InetAddress data_add, bool m_no_dups = false);
    ~DataPath();
    void init();
    void stop_enqueue();
    
    // So we don't set-up queries and enqueue at the same time
    Mutex m_aurora_lock; 

    vector< pair< ptr<StreamState> , string> > last_input_tuples();

    vector< pair< ptr<StreamState> , string> > last_output_tuples();

    vector<StreamEvent> checkpoint_tuples();

    void set_recovery_method(RecoveryMethod method); 

    RecoveryMethod get_recovery_method();

    void set_primary_status(bool status);
    
    bool is_primary();

    void suspend_dequeue();  

    void resume_dequeue();

    // Forward events on that stream to Aurora
    void add_input_path(ptr<StreamState> str);
    bool is_input_path(ptr<StreamState> str);
    void remove_input_path(ptr<StreamState> str);
    string update_input_path(Name stream_name, StreamDef& sd);

    // Forward events on that stream from Aurora to downstream nodes
    void add_output_path(ptr<StreamState> str);
    void remove_output_path(ptr<StreamState> str);   
    void set_data_path(Name stream_name, MedusaID fast_dst, bool with_history, string first_tuple); 
    void close_data_path(Name stream_name, MedusaID fast_dst); 
          
    Status send_downstream(vector<StreamEvent>& events_to_send);

    void update(StreamEvent& stream_event);

    void update(StreamID stream_id, string last_tuple);


    void subscribe_cb(Name stream_name, DataHandler::DHCallback cb);


    void enqueue(ptr<StreamEvent> event) {
        m_enq_thread.do_handle_input_event(event);
    }
    
  private:
 
    static const int MAX_PER_SENDER_OUTPUT_QUEUE = 10000000; // 10 MB

    // Is the thread that enqueues into Aurora running?
    bool m_enqueuing;

    // Stream processing engine
    AuroraNode& m_aurora;
    
    // Handle through which we dequeue tuples from Aurora
    DequeueHandle m_deq_handle;
    
    // Lists of all input and output streams
    typedef map< Name, ptr<StreamState> > IOPaths;
    IOPaths m_input_paths;
    Mutex m_input_paths_lock;
    IOPaths m_output_paths;
    Mutex m_output_paths_lock;

    // If query network produce control streams, callbacks handle their output
    map< Name, DataHandler::DHCallback > m_control_callbacks;

    // Are we the primary or secondary (the HA component has to tell us)
    bool m_is_primary;

    // What is the recovery method (the HA component has to tell us)
    RecoveryMethod m_recovery_method;

    // Should we eliminate duplicates on input streams?
    bool m_no_dups;

    // --------------------------------------------------
    friend class EnqueueThread : public Thread {
        DataPath& m_data_path;
        InetAddress m_data_add;  // Adress where we listen for new data connections
        IOEventLoop m_data_loop; // Event loop for this thread only
        DataHandlers m_data_handlers; // Handlers that receive events on input streams
     
    public:
        EnqueueThread(DataPath& data_path, InetAddress data_add) 
            : m_data_path(data_path), m_data_add(data_add) {}
        virtual ~EnqueueThread() { stop(); }
        // Pushes one event into Aurora 
        Status do_handle_input_event( ptr<StreamEvent> event);
        void stop() { m_data_loop.terminate(); }
        void run();
    };
    EnqueueThread m_enq_thread;
        
    // --------------------------------------------------
    friend class DequeueThread : public Thread {
        DataPath& m_data_path;
        bool m_done;
    public:
        DequeueThread(DataPath& data_path) : m_data_path(data_path) {}
        virtual ~DequeueThread() { stop(); }
        void stop() { m_done = true; }
        void run();
        StreamID buffer(string stream_name, char* buf, size_t total_read, size_t tuple_size);

    };
    DequeueThread m_deq_thread;
    
    // Should we suspend forwarding all events
    bool m_fwd_suspend;

    list< ptr<StreamEvent> > m_events_to_send;

    Mutex m_events_to_send_lock;
 
    // Allows dequeue thread to signal forward thread in a non-blocking manner
    IOHandle m_deq_pipe_write, m_deq_pipe_read;


    // --------------------------------------------------
    friend class FastDequeueForwarder : public Thread {
        DataPath& m_data_path;
    public:
        IOEventLoop m_data_fwd_loop;

        friend class ForwardHandler : public IOHandler {
        public:
            ForwardHandler(IOEventLoop& loop, DataPath::FastDequeueForwarder& fwd_thread);
            void ravail();
        private:
            DataPath& m_data_path;
            DataPath::FastDequeueForwarder& m_fast_dequeue_fwd;
        };
        ptr<ForwardHandler> m_fwd_handler;

        //list of data handlers, one list for each output stream
        typedef map< Name, list< pair<MedusaID, ptr<DataHandler> > > > DataHandlers;
        DataHandlers m_data_handlers;
        vector< ptr<DataHandler> > m_failed;
        
        //lock on data handlers. 
        Mutex m_data_handlers_lock;

        FastDequeueForwarder(DataPath& data_path);
        ~FastDequeueForwarder();

        Status set_data_path(Name stream_name, MedusaID dst, 
                             bool with_history, string first_tuple);

        // Stop forwarding a stream to a downstream node
        Status close_data_path(Name stream_name, MedusaID dst);

        void close_all_data_paths(Name stream_name);

        // Send one event on one stream (multiple destinations)
        Status send_data(Name stream_name, ptr<StreamEvent> event);

        Status send_downstream(vector<StreamEvent>& events_to_send);

        //bool send_one_event_persistently(ptr<StreamEvent> event, ptr<DataHandler> h);

        void clean_failed_handlers();

        void run();
        void stop();
    };
    FastDequeueForwarder m_fast_dequeue_fwd;



};

BOREALIS_NAMESPACE_END;

#endif

---