spdf/elektron
Archived
1
0
Fork 0
Code Issues 4 Pull requests 1 Projects Releases Packages Wiki Activity

Made sure that capping is done only if a new cap value is computed. Otherwise we don't unnecessarily cap the cluster. This reduces overhead.

Browse source
This commit is contained in:
Pradyumna Kaushik 2017-01-08 19:08:46 -08:00
parent 59e08f0a1a
commit 95c419785f
1 changed files with 22 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

37
schedulers/bpMaxMinProacCC.go
View file

@ -1,22 +1,22 @@
package schedulers package schedulers
import ( import (
"bitbucket.org/sunybingcloud/electron/constants"
"bitbucket.org/sunybingcloud/electron/def" "bitbucket.org/sunybingcloud/electron/def"
"bitbucket.org/sunybingcloud/electron/pcp" "bitbucket.org/sunybingcloud/electron/pcp"
"bitbucket.org/sunybingcloud/electron/rapl"
"fmt" "fmt"
"github.com/golang/protobuf/proto" "github.com/golang/protobuf/proto"
mesos "github.com/mesos/mesos-go/mesosproto" mesos "github.com/mesos/mesos-go/mesosproto"
"github.com/mesos/mesos-go/mesosutil" "github.com/mesos/mesos-go/mesosutil"
sched "github.com/mesos/mesos-go/scheduler" sched "github.com/mesos/mesos-go/scheduler"
"log" "log"
"math"
"os" "os"
"sort" "sort"
"strings" "strings"
"sync" "sync"
"time" "time"
"bitbucket.org/sunybingcloud/electron/constants"
"bitbucket.org/sunybingcloud/electron/rapl"
"math"
) )
// Decides if to take an offer or not // Decides if to take an offer or not
@ -156,21 +156,29 @@ func (s *BPMaxMinProacCC) newTask(offer *mesos.Offer, task def.Task) *mesos.Task
// go routine to cap the entire cluster in regular intervals of time. // go routine to cap the entire cluster in regular intervals of time.
var bpMaxMinProacCCCapValue = 0.0 // initial value to indicate that we haven't capped the cluster yet. var bpMaxMinProacCCCapValue = 0.0 // initial value to indicate that we haven't capped the cluster yet.
var bpMaxMinProacCCNewCapValue = 0.0 // newly computed cap value
func (s *BPMaxMinProacCC) startCapping() { func (s *BPMaxMinProacCC) startCapping() {
go func() { go func() {
for { for {
select { select {
case <-s.ticker.C: case <-s.ticker.C:
// Need to cap the cluster to the bpMaxMinProacCCCapValue // Need to cap the cluster only if new cap value different from old cap value.
// This way we don't unnecessarily cap the cluster.
bpMaxMinProacCCMutex.Lock() bpMaxMinProacCCMutex.Lock()
if bpMaxMinProacCCCapValue > 0.0 { if s.isCapping {
for _, host := range constants.Hosts { if int(math.Floor(bpMaxMinProacCCNewCapValue+0.5)) != int(math.Floor(bpMaxMinProacCCCapValue+0.5)) {
// Rounding the cap value to the nearest int // updating cap value
if err := rapl.Cap(host, "rapl", int(math.Floor(bpMaxMinProacCCCapValue+0.5))); err != nil { bpMaxMinProacCCCapValue = bpMaxMinProacCCNewCapValue
log.Println(err) if bpMaxMinProacCCCapValue > 0.0 {
for _, host := range constants.Hosts {
// Rounding cap value to nearest int
if err := rapl.Cap(host, "rapl", int(math.Floor(bpMaxMinProacCCCapValue+0.5))); err != nil {
log.Println(err)
}
}
log.Printf("Capped the cluster to %d", int(math.Floor(bpMaxMinProacCCCapValue+0.5)))
} }
} }
log.Printf("Capped the cluster to %d", int(math.Floor(bpMaxMinProacCCCapValue+0.5)))
} }
bpMaxMinProacCCMutex.Unlock() bpMaxMinProacCCMutex.Unlock()
} }
@ -256,7 +264,7 @@ func (s *BPMaxMinProacCC) CheckFit(i int,
tempCap, err := s.capper.FCFSDeterminedCap(s.totalPower, &task) tempCap, err := s.capper.FCFSDeterminedCap(s.totalPower, &task)
if err == nil { if err == nil {
bpMaxMinProacCCMutex.Lock() bpMaxMinProacCCMutex.Lock()
bpMaxMinProacCCCapValue = tempCap bpMaxMinProacCCNewCapValue = tempCap
bpMaxMinProacCCMutex.Unlock() bpMaxMinProacCCMutex.Unlock()
} else { } else {
log.Println("Failed to determine new cluster-wide cap:") log.Println("Failed to determine new cluster-wide cap:")
@ -334,7 +342,7 @@ func (s *BPMaxMinProacCC) ResourceOffers(driver sched.SchedulerDriver, offers []
// Attempt to schedule a single instance of the heaviest workload available first // Attempt to schedule a single instance of the heaviest workload available first
// Start from the back until one fits // Start from the back until one fits
for i:= len(s.tasks)-1; i >= 0; i-- { for i := len(s.tasks) - 1; i >= 0; i-- {
task := s.tasks[i] task := s.tasks[i]
// Check host if it exists // Check host if it exists
@ -379,7 +387,6 @@ func (s *BPMaxMinProacCC) ResourceOffers(driver sched.SchedulerDriver, offers []
} }
} }
if offerTaken { if offerTaken {
log.Printf("Starting on [%s]\n", offer.GetHostname()) log.Printf("Starting on [%s]\n", offer.GetHostname())
driver.LaunchTasks([]*mesos.OfferID{offer.Id}, tasks, defaultFilter) driver.LaunchTasks([]*mesos.OfferID{offer.Id}, tasks, defaultFilter)
@ -405,8 +412,8 @@ func (s *BPMaxMinProacCC) StatusUpdate(driver sched.SchedulerDriver, status *mes
// Need to remove the task from the window // Need to remove the task from the window
s.capper.TaskFinished(*status.TaskId.Value) s.capper.TaskFinished(*status.TaskId.Value)
// Determining the new cluster wide recap value // Determining the new cluster wide recap value
//tempCap, err := s.capper.Recap(s.totalPower, s.taskMonitor, *status.TaskId.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) //tempCap, err := s.capper.CleverRecap(s.totalPower, s.taskMonitor, *status.TaskId.Value)
if err == nil { if err == nil {
// If new determined recap value is different from the current recap value, then we need to recap. // If new determined recap value is different from the current recap value, then we need to recap.
if int(math.Floor(tempCap+0.5)) != int(math.Floor(bpMaxMinProacCCRecapValue+0.5)) { if int(math.Floor(tempCap+0.5)) != int(math.Floor(bpMaxMinProacCCRecapValue+0.5)) {