PseudoScheduler.h

Go to the documentation of this file.

#ifndef PSEUDO_SCHEDULER_H
#define PSEUDO_SCHEDULER_H

#include "TupleQueue.h"
#include "Scheduler.h"

BOREALIS_NAMESPACE_BEGIN;

class QBox;

class PseudoScheduler : public Scheduler {
  public:
    PseudoScheduler();

    void start();
    void shutdown();
    void scheduleExclusiveTask(ptr<SchedulerTask>);
    void topologyChanged();
    void drain();

    // Make sure box is not in any data structure!!!
    void removeFromSchedule(QBox* b);

    // Boxes that are disabled at the moment (names)
    vector<string> _disabled_boxes;

  private:
    struct MyBoxTupleQueueListener : public TupleQueueListener {
        void notify(const TupleQueue&);

        PseudoScheduler *_ps;
    };
    friend struct MyBoxTupleQueueListener;

    struct MyInputTupleQueueListener : public TupleQueueListener {
        void notify(const TupleQueue&);

        PseudoScheduler *_ps;
    };
    friend struct MyInputTupleQueueListener;

    bool _die;
    bool _draining;

    // condition/lock used for sleeping purposes when draining
    pthread_t _ps_thread;
    pthread_mutex_t     *_wait_mutex;
    pthread_cond_t              *_wait_cond;

    // which map has the latest result (note that there is no atomicity,
    // so this flag is updated *after* new value is inserted into shadow map).
    bool _use_shadow_boxmap;
    bool _use_shadow_qlength;
    unsigned long _last_boxcost_update;
    unsigned long _last_qlength_update;
    unsigned long _last_irate_update;

    // Boxes we've already added listeners to
    set<QBox*> _setup_boxes;

    //typedef vector<ptr<TupleQueueListener> > TQListeners;
    typedef vector<TupleQueueListener*> TQListeners;
    TQListeners _tqlisteners;

    PtMutex _ps_lock;
    PtCondition _ps_condition;
    set<QBox*> _ext_boxes_to_run;
    bool _perf_stats;

    set<QBox*> _boxes_to_run;

    //virtual void removeBoxes( vector<string> boxNames );
    
    void invalidateBoxes( vector<string> boxNames );

    void updateListeners( vector<string> boxNames, bool add );

    void chokeSubNetwork( vector<string> boxNames );

    void suspendSubNetwork( vector<string> boxNames );

    void resumeSubNetwork( vector<string> boxNames );

    void drainSubNetwork( vector<string> boxNames );

    //virtual void activateBoxes( );

    // busywait on CPU
    double use_cpu_time( double time_to_use );

    // windowsSize unit -- ms, historySize unit -- windows
    // this is being phased out to use Ying's structures.
    virtual void init_sched_stats( int windowSize, int historySize );

    virtual string to_string();

    typedef vector<ptr<SchedulerTask> > Tasks;
    Tasks _tasks;

    void setupListeners();

    static void *launch(void *pthis) { ((PseudoScheduler*)pthis)->run(); return 0; }
    void run();

    AURORA_DECLARE_SCHEDULER(PseudoScheduler);
};

BOREALIS_NAMESPACE_END;

#endif

---