Scheduler.h

Go to the documentation of this file.

#ifndef SCHEDULER_H
#define SCHEDULER_H

#include <nmstl_util.h>
#include "Registry.h"
#include <queue>
#include <string>

BOREALIS_NAMESPACE_BEGIN;

class AuroraNode;
class QBox;

class SchedStats
{
 public:

    map<string, deque<double> > _selectivities;
    map<string, deque<double> > _box_costs;
    map<string, deque<double> > _avg_queue_increment;

    int cpu_load;
};

class SchedulerStats
{
 public:
    map<string, double > _total_cost;
    map<string, int > _execution_count;

    map<string, int> _total_input;
    map<string, int> _total_output;

    map<string, deque<double> > _selectivities;
    map<string, deque<double> > _avg_queue_increment;

    map<string, double > _box_costs, _shadow_box_costs;
    map<string, int > _box_executions, _shadow_box_executions;

    map<string, double > _queue_sizes, _shadow_queue_sizes;
    map<string, int> _queue_entries, _shadow_queue_entries;

    map<string, int> _last_enq_tuples, _curr_enq_tuples;
};  


class SchedulerTask {
  public:
    virtual ~SchedulerTask() {}

    virtual void run() = 0;
};

class Scheduler {
    friend class AuroraNode;

  public:
    typedef Registry<Scheduler> Registry;

    class BoxData {
      public:
        virtual ~BoxData() {}
    };
    
    virtual ~Scheduler();

    AuroraNode& node() { return *_node; }

    virtual void start() = 0;

    virtual void shutdown() = 0;

    virtual void scheduleExclusiveTask(ptr<SchedulerTask>) = 0;

    virtual void topologyChanged() { ASSERT(_dynamic_support); }

    bool hasDynamicSupport() { return _dynamic_support; }

    virtual void drain() { ASSERT(_drain_support); }

    bool hasDrain() { return _drain_support; }

    static unsigned long ticks();

    // On a 2.8GHz box.  Yee-haw!
    // 3932 on my Athlon machine. I think. 
    // TODO: Need to measure this properly
    static double secondsPerTick() { return (1<<21) / 3.9e9; }

    // 2.8GHz box; throwing away the low-order 21 bits of the
    // RTDSC register; so there are 2.8/(2^21) ticks per second
    static unsigned long ticksPerSecond() { return 1335; }

    // my stuff, put it here for now, methinks
    SchedulerStats _sched_stats;

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

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

    virtual void chokeSubNetwork ( vector<string> boxNames );

    virtual void suspendSubNetwork ( vector<string> boxNames );

    virtual void resumeSubNetwork( vector<string> boxNames );

    virtual void drainSubNetwork( vector<string> boxNames );

    virtual void removeFromSchedule(QBox* box) { }

    virtual string to_string();

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

  protected:
    unsigned int _init_time;

    void setBoxData(QBox& box, ptr<BoxData> data);
    BoxData *getBoxData(const QBox& box) const;

    Scheduler(bool dynamic_support = false, bool drain_support = false);

  private:
    AuroraNode *_node;

    bool _dynamic_support;
    bool _drain_support;
};

inline unsigned long Scheduler::ticks()
{
    unsigned int low, high;
    asm volatile ("rdtsc" : "=a" (low), "=d" (high));

    return (high<<11) + (low>>21);
}

#define AURORA_DECLARE_SCHEDULER(ClassName) AURORA_DECLARE_REG_CLASS(Scheduler, ClassName)
#define AURORA_DEFINE_SCHEDULER(ClassName) \
    AURORA_DEFINE_REG_CLASS(Scheduler, ClassName)

BOREALIS_NAMESPACE_END;

// For Scheduler::setBoxData, Scheduler::getBoxData
#include "QBox.h"

#endif

---