PriorityScheduler.h

Go to the documentation of this file.

#ifndef PRIORITY_SCHEDULER_H
#define PRIORITY_SCHEDULER_H

#include "TupleQueue.h"
#include "Scheduler.h"
#include <queue>

BOREALIS_NAMESPACE_BEGIN;

class QBox;

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

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

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

  private:


    struct MyBoxData : public BoxData {
        MyBoxData(QBox& box);

        QBox *_box;
        int _priority;
        int _weight;
        bool _scheduled;
        bool _in_pending;
        int _running_time;
    };
    struct PriorityLevel {
        PriorityLevel() : _scheduled(false) {}

        vector<MyBoxData*> _boxes_next;
        vector<MyBoxData*> _boxes_to_run;
        bool _scheduled;
    };
    map<int, PriorityLevel> _schedules;
    typedef priority_queue<int, vector<int>, greater<int> > PrioritiesToRun;
    PrioritiesToRun _priorities_to_run;

    void scheduleBox(MyBoxData&);
    void schedulePending();
    bool checkRep() const;

    MyBoxData& getMyBoxData(QBox& box) const {
        MyBoxData *data = static_cast<MyBoxData*>(Scheduler::getBoxData(box));
        ASSERT(data);
        return *data;
    }

    bool _die;
    bool _draining;

    pthread_t _ps_thread;

    virtual void invalidateBoxes( vector<string> boxNames );

    virtual void updateListeners( vector<string> boxNames, bool add );
    
    /*virtual void removeBoxes( vector<string> boxNames );
    
    virtual void addBoxes( );

    virtual void activateBoxes( );
    */

    // windowsSize unit -- ms, historySize unit -- windows
    virtual void init_sched_stats( int windowSize, int historySize );

    virtual string to_string();

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

        PriorityScheduler *_ps;
    };
    friend struct MyBoxTupleQueueListener;

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

        PriorityScheduler *_ps;
    };
    friend struct MyInputTupleQueueListener;

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

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

    PtMutex _ps_lock;
    PtCondition _ps_condition;

    vector<MyBoxData*> _ext_boxes_to_run;

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

    void setupListeners();

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

    AURORA_DECLARE_SCHEDULER(PriorityScheduler);
};

BOREALIS_NAMESPACE_END;

#endif

---