AuroraNode.h

Go to the documentation of this file.

#ifndef AURORANODE_H
#define AURORANODE_H

#include "Exceptions.h"

#include "ConfFile.h"
#include "CpuStats.h"
#include "QBox.h"
#include "Stream.h"
#include "TupleDescription.h"
#include "Scheduler.h"
#include "StreamProperties.h"
#include "LockSet.h"

#include "StreamDef.h"

#include "FixLenTimeSeries.h"
#include "StatsMgr.h"
#include "TSStats.h"
#include "VersionRWLock.h"

#ifdef USE_STREAMMON
    #include "StreamMon.h"
#endif

#include "BoxPackage.h"
#include "NodeTable.h"

BOREALIS_NAMESPACE_BEGIN;

class SchedulerTask;
class ConverterBoxRunner;

class CP;
class CPView;
class CPViewDescription;


class AuroraNode
{
    struct Subscription
    {   ptr<TupleQueue> _q;
        
    };


    struct DequeueContext
    {   TuplesAvailableMailbox      _mailbox;
        map<Stream*, Subscription>  _subscriptions;
        list<ptr<TupleQueue> >      _deleted_queues;

        // Called by UnsubscribeAllTask
        void  unsubscribeAll();        
    };
    

    friend struct UnsubscribeAllTask;
    friend struct ModifyNetworkTask;
    friend struct GenericCheckpointTask;
    friend struct CheckpointTask;
    friend struct RecoveryTask;

  public:
    enum EntityType
    {
        STREAM = 1,

        SCHEMA = 2,

        BOX = 3,

        INPUT_STREAMS = 100,

        OUTPUT_STREAMS = 101
    };


    double  selval;

    class DequeueHandle
    {
      public:
        DequeueHandle() : _node(0), _context() {}


        void  subscribe( string  name ) throw( AuroraException );


        void  unsubscribe( string  name ) throw( AuroraException );


        size_t  dequeue( string  &stream_name,
               TupleDescription  &td,
                           void  *buffer,
                         size_t   buffer_size
                       )
                  throw( AuroraException );

        void close();

      private:
        AuroraNode *_node;
        ptr<DequeueContext> _context;

        friend class AuroraNode;
        friend struct UnsubscribeAllTask;
    };


    friend class DequeueHandle;

    typedef ::StreamProperties StreamProperties;


    AuroraNode( ptr<ConfFile>  conf = ptr<ConfFile>() )
         throw( AuroraException );


    ~AuroraNode();


    void  modifyNetwork( string   data,
                     BoxPackage  &boxes
                       )
                  throw( AuroraException );

    void load_box( string  file_path )
            throw( AuroraException );


    void typecheck( string  data, map<string, TupleDescription>  &streams )
             throw( AuroraException );


    void enqueue( const char  *stream_name,
                  const void  *data,
                      size_t   tuple_length,
                      size_t   num_tuples,
                  const void  *schema_uuid = 0
                )
           throw( AuroraException );


    DequeueHandle  getDequeueHandle();


    void  start() throw( AuroraException );


    void  shutdown() throw( AuroraException );


    void  typecheck( string  data, vector<StreamProperties>  &streams )
              throw( AuroraException );


    void  validate( string  data ) throw( AuroraException );


    StreamProperties  getStreamProperties( string  entity )
                                    throw( AuroraException );


    string  describe( string  entity ) throw( AuroraException );


    void  listEntities( EntityType  type, vector<string>  &names )
                 throw( AuroraException );


    void  drain() throw( AuroraException );


    void  chokeBoxes( vector<string>  names );

    void  suspendBoxes( vector<string>  names );

    void  resumeBoxes( vector<string>  names );


    void  packBoxes( string  data, BoxPackage  &boxes )
              throw( AuroraException );

    void checkpoint(vector<long long>& measurements);
    void recover();

    void schedule_checkpoint_task(ptr<GenericCheckpointTask> task);

    string  repr() const;


    Statistics  &getStatsObject()
    {   return( sm.getStats() );
    }


    const Statistics  &getStatsObject() const
    {   return( sm.getStats() );
    }


    set<ptr<QBox> > getBoxes() const;


    ptr<QBox> getBox( string  name ) const
    {   return( lookup( _boxes, name ));
    }


    ptr<Stream>  getStream( string  name ) const
    {   return( lookup( _streams, name ));
    }


    ptr<LockSet>  getLockSet( string  name );


    TupleDescription  getTupleDescription( string  name ) const
    {   return( lookup( _schemas, name ));
    }


    TupleDescription  getTupleDescriptionByHash( string  hash ) const
    {   return( lookup( _schemas_by_hash, hash ));
    }


    ptr<Stream>  createStream( string name, TupleDescription desc )
                        throw( AuroraException );


    void create_cpview( CPViewDescription  view_desc, StreamDef  streamdef )
                 throw( AuroraException );


    ptr<ConfFile> getProperties() const throw( AuroraException )
    {   return( _conf );
    }


    unsigned int  xGetInputQueueLengthSum() const;


    string getNetworkAsDot() const;
    string getNetworkAsDotSubgraph(string prefix, string attrs) const;
    string getNetworkAsDotStatements(string prefix) const;


    PagePool  &getPagePool()
    {   return( *_page_pool );
    }


    const CpuStats  &getCpuStats() const
    {   return( _cpu_stats );
    }


    StatsMgr sm;

  private:
    typedef map<string, ptr<Stream> > StreamMap;
    typedef map<string, ptr<QBox> > BoxMap;
    typedef map<string, TupleDescription> SchemaMap;
    typedef map<string, TupleDescription> SchemaByHashMap;

    //    typedef map<string, ptr<CPoint> > CPointMap;

    struct Topology
    {
        StreamMap      _streams;
        BoxMap         _boxes;
        SchemaMap      _schemas;
        TableMap       _tables;
      //CPointMap      _cpoints;

        bool containsEntity( string name ) const;

        void copyFrom( const Topology  &other );

        string repr() const;
    };


    void parseAddBoxXml( const DOMElement  *addNode,
                                 Topology  &new_topology,
                               BoxPackage  &packedBoxes
                       )
                  throw( AuroraException );


    void parseAddXml( const DOMElement  *addNode,
                              Topology  &new_topology,
                            BoxPackage  &packedBoxes
                    )
               throw( AuroraException );


    void parseRemoveBoxXml( const DOMElement  *removeBoxNode,
                                    Topology  &top
                          )
                     throw( AuroraException );


    void parseRemoveXml( const DOMElement  *removeNode,
                                 Topology  &top,
                               BoxPackage  &box_package
                       )
                  throw( AuroraException );



    void handleModifyXml( string   xml,
                            bool   commit,
                      map<string, TupleDescription>&,
                      BoxPackage  &boxes
                        )
                   throw( AuroraException );

    void schedule( ptr<SchedulerTask>  task );

    void loadLibrary(string file) throw (AuroraException);

    bool loadLibrariesInDirectory(string directory) throw (AuroraException);

    void loadLibraries(const ConfFile& conf);

    enum State
    {   STATE_INIT,
        STATE_RUNNING,
        STATE_STOPPED
    };

    State  _state;

    // Note that order is important here; the streams/boxes may depend
    // on the PagePool, so they must be destroyed *before* the
    // PagePool.
    //
    // Note that these are not currently protected by a mutex (which
    // is fine because the network cannot be modified after it has
    // started).

    ptr<PagePool>          _page_pool;

    Topology    _topology;

    NodeTable   _node_table;

    StreamMap  &_streams;
    BoxMap     &_boxes;
    SchemaMap  &_schemas;
    //    CPointMap  &_cpoints;

    SchemaByHashMap  _schemas_by_hash;

    map<string, ptr<LockSet> >  _lock_sets;

    // please ignore this hack. it is implemented until Ying fixes the
    // stats manager and implements a removeTSStat
    bool addStats;

    PtMutex  _lock;


    typedef set< ptr<DequeueContext> >  DeqContexts;
    DeqContexts  _deqs;


    ptr<ConfFile>   _conf;
    ptr<Scheduler>  _scheduler;


    ptr<ConverterBoxRunner>  _cbox_runner;

    
    typedef map<Name, ptr<CP> > CPs;
    CPs  m_cps;

    
    vector<ptr<CPView> >  m_cp_views;


    CpuStats  _cpu_stats;

    BoxPackage _checkpointed_state;


#ifdef USE_STREAMMON

    ptr<StreamMon> _stream_mon;
#endif
};

typedef AuroraNode::DequeueHandle DequeueHandle;

BOREALIS_NAMESPACE_END;

#endif                      // AURORANODE_H

---