diff --git a/schedulers/bpswClassMapWattsProacCC.go b/schedulers/bpswClassMapWattsProacCC.go
new file mode 100644
index 0000000..1b62a66
--- /dev/null
+++ b/schedulers/bpswClassMapWattsProacCC.go
@@ -0,0 +1,428 @@
+package schedulers
+
+import (
+ "bitbucket.org/sunybingcloud/electron/constants"
+ "bitbucket.org/sunybingcloud/electron/def"
+ "bitbucket.org/sunybingcloud/electron/pcp"
+ "bitbucket.org/sunybingcloud/electron/rapl"
+ "fmt"
+ "github.com/golang/protobuf/proto"
+ mesos "github.com/mesos/mesos-go/mesosproto"
+ "github.com/mesos/mesos-go/mesosutil"
+ sched "github.com/mesos/mesos-go/scheduler"
+ "log"
+ "math"
+ "strings"
+ "sync"
+ "time"
+ "sort"
+)
+
+// Decides if to take an offer or not
+func (*BPSWClassMapWattsProacCC) takeOffer(offer *mesos.Offer, task def.Task) bool {
+ cpus, mem, watts := OfferAgg(offer)
+
+ // TODO: Insert watts calculation here instead of taking them as parameter
+
+ if cpus >= task.CPU && mem >= task.RAM && watts >= task.Watts {
+ return true
+ }
+
+ return false
+}
+
+type BPSWClassMapWattsProacCC struct {
+ tasksCreated int
+ tasksRunning int
+ tasks []def.Task
+ metrics map[string]def.Metric
+ running map[string]map[string]bool
+ taskMonitor map[string][]def.Task
+ availablePower map[string]float64
+ totalPower map[string]float64
+ ignoreWatts bool
+ capper *pcp.ClusterwideCapper
+ ticker *time.Ticker
+ recapTicker *time.Ticker
+ isCapping bool // indicate whether we are currently performing cluster-wide capping.
+ isRecapping bool // indicate whether we are currently performing cluster-wide recapping.
+
+ // First set of PCP values are garbage values, signal to logger to start recording when we're
+ // about to schedule a new task
+ RecordPCP bool
+
+ // This channel is closed when the program receives an interrupt,
+ // signalling that the program should shut down
+ Shutdown chan struct{}
+ // This channel is closed after shutdown is closed, and only when all
+ // outstanding tasks have been cleaned up
+ Done chan struct{}
+
+ // Controls when to shutdown pcp logging
+ PCPLog chan struct{}
+}
+
+// New electron scheduler
+func NewBPSWClassMapWattsProacCC(tasks []def.Task, ignoreWatts bool) *BPSWClassMapWattsProacCC {
+ sort.Sort(def.WattsSorter(tasks))
+
+ s := &BPSWClassMapWattsProacCC{
+ tasks: tasks,
+ ignoreWatts: ignoreWatts,
+ Shutdown: make(chan struct{}),
+ Done: make(chan struct{}),
+ PCPLog: make(chan struct{}),
+ running: make(map[string]map[string]bool),
+ taskMonitor: make(map[string][]def.Task),
+ availablePower: make(map[string]float64),
+ totalPower: make(map[string]float64),
+ RecordPCP: false,
+ capper: pcp.GetClusterwideCapperInstance(),
+ ticker: time.NewTicker(10 * time.Second),
+ recapTicker: time.NewTicker(20 * time.Second),
+ isCapping: false,
+ isRecapping: false,
+ }
+ return s
+}
+
+// mutex
+var bpswClassMapWattsProacCCMutex sync.Mutex
+
+func (s *BPSWClassMapWattsProacCC) newTask(offer *mesos.Offer, task def.Task, newTaskClass string) *mesos.TaskInfo {
+ taskName := fmt.Sprintf("%s-%d", task.Name, *task.Instances)
+ s.tasksCreated++
+
+ if !s.RecordPCP {
+ // Turn on logging.
+ s.RecordPCP = true
+ time.Sleep(1 * time.Second) // Make sure we're recording by the time the first task starts
+ }
+
+ // If this is our first time running into this Agent
+ if _, ok := s.running[offer.GetSlaveId().GoString()]; !ok {
+ s.running[offer.GetSlaveId().GoString()] = make(map[string]bool)
+ }
+
+ // Setting the task ID to the task. This is done so that we can consider each task to be different,
+ // even though they have the same parameters.
+ task.SetTaskID(*proto.String("electron-" + taskName))
+ // Add task to the list of tasks running on the node.
+ s.running[offer.GetSlaveId().GoString()][taskName] = true
+ if len(s.taskMonitor[*offer.Hostname]) == 0 {
+ s.taskMonitor[*offer.Hostname] = []def.Task{task}
+ } else {
+ s.taskMonitor[*offer.Hostname] = append(s.taskMonitor[*offer.Hostname], task)
+ }
+
+ resources := []*mesos.Resource{
+ mesosutil.NewScalarResource("cpus", task.CPU),
+ mesosutil.NewScalarResource("mem", task.RAM),
+ }
+
+ if !s.ignoreWatts {
+ resources = append(resources, mesosutil.NewScalarResource("watts", task.ClassToWatts[newTaskClass]))
+ }
+
+ return &mesos.TaskInfo{
+ Name: proto.String(taskName),
+ TaskId: &mesos.TaskID{
+ Value: proto.String("electron-" + taskName),
+ },
+ SlaveId: offer.SlaveId,
+ Resources: resources,
+ Command: &mesos.CommandInfo{
+ Value: proto.String(task.CMD),
+ },
+ Container: &mesos.ContainerInfo{
+ Type: mesos.ContainerInfo_DOCKER.Enum(),
+ Docker: &mesos.ContainerInfo_DockerInfo{
+ Image: proto.String(task.Image),
+ Network: mesos.ContainerInfo_DockerInfo_BRIDGE.Enum(), // Run everything isolated
+ },
+ },
+ }
+}
+
+func (s *BPSWClassMapWattsProacCC) Registered(
+ _ sched.SchedulerDriver,
+ frameworkID *mesos.FrameworkID,
+ masterInfo *mesos.MasterInfo) {
+ log.Printf("Framework %s registered with master %s", frameworkID, masterInfo)
+}
+
+func (s *BPSWClassMapWattsProacCC) Reregistered(_ sched.SchedulerDriver, masterInfo *mesos.MasterInfo) {
+ log.Printf("Framework re-registered with master %s", masterInfo)
+}
+
+func (s *BPSWClassMapWattsProacCC) Disconnected(sched.SchedulerDriver) {
+ // Need to stop the capping process
+ s.ticker.Stop()
+ s.recapTicker.Stop()
+ bpswClassMapWattsProacCCMutex.Lock()
+ s.isCapping = false
+ bpswClassMapWattsProacCCMutex.Unlock()
+ log.Println("Framework disconnected with master")
+}
+
+// go routine to cap the entire cluster in regular intervals of time.
+var bpswClassMapWattsCapValue = 0.0 // initial value to indicate that we haven't capped the cluster yet.
+func (s *BPSWClassMapWattsProacCC) startCapping() {
+ go func() {
+ for {
+ select {
+ case <-s.ticker.C:
+ // Need to cap the cluster to the bpswClassMapWattsCapValue.
+ bpswClassMapWattsProacCCMutex.Lock()
+ if bpswClassMapWattsCapValue > 0.0 {
+ for _, host := range constants.Hosts {
+ // Rounding capValue to nearest int.
+ if err := rapl.Cap(host, "rapl", int(math.Floor(bpswClassMapWattsCapValue+0.5))); err != nil {
+ log.Println(err)
+ }
+ }
+ log.Printf("Capped the cluster to %d", int(math.Floor(bpswClassMapWattsCapValue+0.5)))
+ }
+ bpswClassMapWattsProacCCMutex.Unlock()
+ }
+ }
+ }()
+}
+
+// go routine to recap the entire cluster in regular intervals of time.
+var bpswClassMapWattsRecapValue = 0.0 // The cluster-wide cap value when recapping
+func (s *BPSWClassMapWattsProacCC) startRecapping() {
+ go func() {
+ for {
+ select {
+ case <-s.recapTicker.C:
+ bpswClassMapWattsProacCCMutex.Lock()
+ // If stopped performing cluster wide capping, then we need to recap
+ if s.isRecapping && bpswClassMapWattsRecapValue > 0.0 {
+ for _, host := range constants.Hosts {
+ // Rounding capValue to the nearest int
+ if err := rapl.Cap(host, "rapl", int(math.Floor(bpswClassMapWattsRecapValue+0.5))); err != nil {
+ log.Println(err)
+ }
+ }
+ log.Printf("Recapping the cluster to %d", int(math.Floor(bpswClassMapWattsRecapValue+0.5)))
+ }
+ // Setting recapping to false
+ s.isRecapping = false
+ bpswClassMapWattsProacCCMutex.Unlock()
+ }
+ }
+ }()
+}
+
+// Stop cluster wide capping
+func (s *BPSWClassMapWattsProacCC) stopCapping() {
+ if s.isCapping {
+ log.Println("Stopping the cluster-wide capping.")
+ s.ticker.Stop()
+ bpswClassMapWattsProacCCMutex.Lock()
+ s.isCapping = false
+ s.isRecapping = true
+ bpswClassMapWattsProacCCMutex.Unlock()
+ }
+}
+
+// Stop the cluster wide recapping
+func (s *BPSWClassMapWattsProacCC) stopRecapping() {
+ // If not capping, then definitely recapping.
+ if !s.isCapping && s.isRecapping {
+ log.Println("Stopping the cluster-wide re-capping.")
+ s.recapTicker.Stop()
+ bpswClassMapWattsProacCCMutex.Lock()
+ s.isRecapping = false
+ bpswClassMapWattsProacCCMutex.Unlock()
+ }
+}
+
+func (s *BPSWClassMapWattsProacCC) ResourceOffers(driver sched.SchedulerDriver, offers []*mesos.Offer) {
+ log.Printf("Received %d resource offers", len(offers))
+
+ // retrieving the available power for all the hosts in the offers.
+ for _, offer := range offers {
+ _, _, offerWatts := OfferAgg(offer)
+ s.availablePower[*offer.Hostname] = offerWatts
+ // setting total power if the first time
+ if _, ok := s.totalPower[*offer.Hostname]; !ok {
+ s.totalPower[*offer.Hostname] = offerWatts
+ }
+ }
+
+ for host, tpower := range s.totalPower {
+ log.Printf("TotalPower[%s] = %f", host, tpower)
+ }
+
+ for _, offer := range offers {
+ select {
+ case <-s.Shutdown:
+ log.Println("Done scheduling tasks: declining offer on [", offer.GetHostname(), "]")
+ driver.DeclineOffer(offer.Id, longFilter)
+
+ log.Println("Number of tasks still running: ", s.tasksRunning)
+ continue
+ default:
+ }
+
+ tasks := []*mesos.TaskInfo{}
+
+ offerCPU, offerRAM, offerWatts := OfferAgg(offer)
+
+ taken := false
+ totalWatts := 0.0
+ totalCPU := 0.0
+ totalRAM := 0.0
+ for i, task := range s.tasks {
+ // Check host if it exists
+ if task.Host != "" {
+ // Don't take offer it it doesn't match our task's host requirement.
+ if strings.HasPrefix(*offer.Hostname, task.Host) {
+ continue
+ }
+ }
+
+ for *task.Instances > 0 {
+ var nodeClass string
+ for _, attr := range offer.GetAttributes() {
+ if attr.GetName() == "class" {
+ nodeClass = attr.GetText().GetValue()
+ }
+ }
+ // Does the task fit
+ // OR Lazy evaluation. If ignore watts is set to true, second statement won't
+ // be evaluated.
+ if (s.ignoreWatts || (offerWatts >= (totalWatts + task.ClassToWatts[nodeClass]))) &&
+ (offerCPU >= (totalCPU + task.CPU)) &&
+ (offerRAM >= (totalRAM + task.RAM)) {
+
+ // Capping the cluster if haven't yet started
+ if !s.isCapping {
+ bpswClassMapWattsProacCCMutex.Lock()
+ s.isCapping = true
+ bpswClassMapWattsProacCCMutex.Unlock()
+ s.startCapping()
+ }
+
+ fmt.Println("Watts being used: ", task.ClassToWatts[nodeClass])
+ tempCap, err := s.capper.FCFSDeterminedCap(s.totalPower, &task)
+ if err == nil {
+ bpswClassMapWattsProacCCMutex.Lock()
+ bpswClassMapWattsCapValue = tempCap
+ bpswClassMapWattsProacCCMutex.Unlock()
+ } else {
+ log.Println("Failed to determine new cluster-wide cap:")
+ log.Println(err)
+ }
+ taken = true
+ totalWatts += task.ClassToWatts[nodeClass]
+ totalCPU += task.CPU
+ totalRAM += task.RAM
+ log.Println("Co-Located with: ")
+ coLocated(s.running[offer.GetSlaveId().GoString()])
+ tasks = append(tasks, s.newTask(offer, task, nodeClass))
+
+ fmt.Println("Inst: ", *task.Instances)
+ *task.Instances--
+
+ if *task.Instances <= 0 {
+ // All instances of task have been scheduled, remove it
+ s.tasks = append(s.tasks[:i], s.tasks[i+1:]...)
+
+ if len(s.tasks) == 0 {
+ log.Println("Done scheduling all tasks.")
+ // Need to stop the cluster wide capping as there aren't any more tasks to schedule.
+ s.stopCapping()
+ s.startRecapping() // Load changes after every task finishes and hence, we need to change the capping of the cluster.
+ close(s.Shutdown)
+ }
+ }
+ } else {
+ break // Continue on to the next task
+ }
+ }
+ }
+
+ if taken {
+ log.Printf("Starting on [%s]\n", offer.GetHostname())
+ driver.LaunchTasks([]*mesos.OfferID{offer.Id}, tasks, defaultFilter)
+ } else {
+ // If there was no match for the task
+ fmt.Println("There is not enough resources to launch a task:")
+ cpus, mem, watts := OfferAgg(offer)
+
+ log.Printf("<CPU: %f, RAM: %f, Watts: %f>\n", cpus, mem, watts)
+ driver.DeclineOffer(offer.Id, defaultFilter)
+ }
+ }
+}
+
+func (s *BPSWClassMapWattsProacCC) StatusUpdate(driver sched.SchedulerDriver, status *mesos.TaskStatus) {
+ log.Printf("Received task status [%s] for task [%s]", NameFor(status.State), *status.TaskId.Value)
+
+ if *status.State == mesos.TaskState_TASK_RUNNING {
+ s.tasksRunning++
+ } else if IsTerminal(status.State) {
+ delete(s.running[status.GetSlaveId().GoString()], *status.TaskId.Value)
+ // Need to remove the task from the window
+ s.capper.TaskFinished(*status.TaskId.Value)
+ // Determining the new cluster wide recap value
+ tempCap, err := s.capper.Recap(s.totalPower, s.taskMonitor, *status.TaskId.Value)
+ //tempCap, err := s.capper.CleverRecap(s.totalPower, s.taskMonitor, *status.TaskId.Value)
+ if err == nil {
+ // If new determined cap value is different from the current recap value, then we need to recap
+ if int(math.Floor(tempCap+0.5)) != int(math.Floor(bpswClassMapWattsRecapValue+0.5)) {
+ bpswClassMapWattsRecapValue = tempCap
+ bpswClassMapWattsProacCCMutex.Lock()
+ s.isRecapping = true
+ bpswClassMapWattsProacCCMutex.Unlock()
+ log.Printf("Determined re-cap value: %f\n", bpswClassMapWattsRecapValue)
+ } else {
+ bpswClassMapWattsProacCCMutex.Lock()
+ s.isRecapping = false
+ bpswClassMapWattsProacCCMutex.Unlock()
+ }
+ } else {
+ log.Println(err)
+ }
+
+ s.tasksRunning--
+ if s.tasksRunning == 0 {
+ select {
+ case <-s.Shutdown:
+ // Need to stop the cluster-wide recapping
+ s.stopRecapping()
+ close(s.Done)
+ default:
+ }
+ }
+ }
+ log.Printf("DONE: Task status [%s] for task[%s]", NameFor(status.State), *status.TaskId.Value)
+}
+
+func (s *BPSWClassMapWattsProacCC) FrameworkMessage(
+ driver sched.SchedulerDriver,
+ executorID *mesos.ExecutorID,
+ slaveID *mesos.SlaveID,
+ message string) {
+ log.Println("Getting a framework message: ", message)
+ log.Printf("Received framework message from some unknown source: %s", *executorID.Value)
+}
+
+func (s *BPSWClassMapWattsProacCC) OfferRescinded(_ sched.SchedulerDriver, offerID *mesos.OfferID) {
+ log.Printf("Offer %s rescinded", offerID)
+}
+
+func (s *BPSWClassMapWattsProacCC) SlaveLost(_ sched.SchedulerDriver, slaveID *mesos.SlaveID) {
+ log.Printf("Slave %s lost", slaveID)
+}
+
+func (s *BPSWClassMapWattsProacCC) ExecutorLost(_ sched.SchedulerDriver, executorID *mesos.ExecutorID, slaveID *mesos.SlaveID, status int) {
+ log.Printf("Executor %s on slave %s was lost", executorID, slaveID)
+}
+
+func (s *BPSWClassMapWattsProacCC) Error(_ sched.SchedulerDriver, err string) {
+ log.Printf("Receiving an error: %s", err)
+}