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

added utility to compute the scheduling window. Right now there's only criteria on which this is determined -- fillNextOfferCycle. So, the schedWindow is the max number of tasks, that aren't yet scheduled, whose aggregate resource requirement is as close as possible to the resource available in the next round of resource offers. To be able to make the most use of the next offer cycle, one would need to perform a non-polynomial search of the TaskQueue and as this is computationally expensive, a linear search is performed on the TaskQueue. Retrofitted scheduling policies to also call utilities.schedUtils#schedWindowResizingStrategy#Apply before switching to a new scheduling policy.

Browse source
This commit is contained in:
Pradyumna Kaushik 2018-01-31 19:00:31 -05:00
parent 6f0f3788b9
commit 3ebd7b0c7e
12 changed files with 115 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

6
schedulers/MaxGreedyMins.go
View file

@ -4,6 +4,7 @@ import (
"bitbucket.org/sunybingcloud/elektron/def"
"bitbucket.org/sunybingcloud/elektron/utilities/mesosUtils"
"bitbucket.org/sunybingcloud/elektron/utilities/offerUtils"
"bitbucket.org/sunybingcloud/elektron/utilities/schedUtils"
"fmt"
mesos "github.com/mesos/mesos-go/api/v0/mesosproto"
sched "github.com/mesos/mesos-go/api/v0/scheduler"
@ -174,6 +175,11 @@ func (s *MaxGreedyMins) ConsumeOffers(spc SchedPolicyContext, driver sched.Sched
// Switch scheduling policy only if feature enabled from CLI
if baseSchedRef.schedPolSwitchEnabled {
// Need to recompute the schedWindow for the next offer cycle.
// The next scheduling policy will schedule at max schedWindow number of tasks.
baseSchedRef.schedWindow = schedUtils.
SchedWindowResizingCritToStrategy[baseSchedRef.schedWindowResizeCrit].Apply(
func() interface{} { return baseSchedRef.tasks })
// Switching to a random scheduling policy.
// TODO: Switch based on some criteria.
index := rand.Intn(len(SchedPolicies))

6
schedulers/MaxMin.go
View file

@ -4,6 +4,7 @@ import (
"bitbucket.org/sunybingcloud/elektron/def"
"bitbucket.org/sunybingcloud/elektron/utilities/mesosUtils"
"bitbucket.org/sunybingcloud/elektron/utilities/offerUtils"
"bitbucket.org/sunybingcloud/elektron/utilities/schedUtils"
"fmt"
mesos "github.com/mesos/mesos-go/api/v0/mesosproto"
sched "github.com/mesos/mesos-go/api/v0/scheduler"
@ -168,6 +169,11 @@ func (s *MaxMin) ConsumeOffers(spc SchedPolicyContext, driver sched.SchedulerDri
// Switch scheduling policy only if feature enabled from CLI
if baseSchedRef.schedPolSwitchEnabled {
// Need to recompute the schedWindow for the next offer cycle.
// The next scheduling policy will schedule at max schedWindow number of tasks.
baseSchedRef.schedWindow = schedUtils.
SchedWindowResizingCritToStrategy[baseSchedRef.schedWindowResizeCrit].Apply(
func() interface{} { return baseSchedRef.tasks })
// Switching to a random scheduling policy.
// TODO: Switch based on some criteria.
index := rand.Intn(len(SchedPolicies))

13
schedulers/base.go
View file

@ -4,6 +4,7 @@ import (
"bitbucket.org/sunybingcloud/elektron/def"
elecLogDef "bitbucket.org/sunybingcloud/elektron/logging/def"
"bitbucket.org/sunybingcloud/elektron/utilities"
"bitbucket.org/sunybingcloud/elektron/utilities/schedUtils"
"bytes"
"fmt"
"github.com/golang/protobuf/proto"
@ -55,6 +56,17 @@ type BaseScheduler struct {
// Whether switching of scheduling policies at runtime has been enabled
schedPolSwitchEnabled bool
// Size of window of tasks that can be scheduled in the next offer cycle.
// The window size can be adjusted to make the most use of every resource offer.
// By default, the schedulingWindow would correspond to all the remaining tasks that haven't yet been scheduled.
schedulingWindow int
// Criteria to resize the schedulingWindow.
schedWindowResizeCrit schedUtils.SchedulingWindowResizingCriteria
// Window of tasks that the current scheduling policy has to schedule.
// Once #schedWindow tasks are scheduled, the current scheduling policy has to stop scheduling.
schedWindow int
}
func (s *BaseScheduler) init(opts ...schedPolicyOption) {
@ -66,6 +78,7 @@ func (s *BaseScheduler) init(opts ...schedPolicyOption) {
}
s.running = make(map[string]map[string]bool)
s.mutex = sync.Mutex{}
s.schedWindowResizeCrit = "fillNextOfferCycle"
}
func (s *BaseScheduler) SwitchSchedPol(newSchedPol SchedPolicyState) {

6
schedulers/bin-packing.go
View file

@ -4,6 +4,7 @@ import (
"bitbucket.org/sunybingcloud/elektron/def"
"bitbucket.org/sunybingcloud/elektron/utilities/mesosUtils"
"bitbucket.org/sunybingcloud/elektron/utilities/offerUtils"
"bitbucket.org/sunybingcloud/elektron/utilities/schedUtils"
"fmt"
mesos "github.com/mesos/mesos-go/api/v0/mesosproto"
sched "github.com/mesos/mesos-go/api/v0/scheduler"
@ -118,6 +119,11 @@ func (s *BinPackSortedWatts) ConsumeOffers(spc SchedPolicyContext, driver sched.
// Switch scheduling policy only if feature enabled from CLI
if baseSchedRef.schedPolSwitchEnabled {
// Need to recompute the schedWindow for the next offer cycle.
// The next scheduling policy will schedule at max schedWindow number of tasks.
baseSchedRef.schedWindow = schedUtils.
SchedWindowResizingCritToStrategy[baseSchedRef.schedWindowResizeCrit].Apply(
func() interface{} { return baseSchedRef.tasks })
// Switching to a random scheduling policy.
// TODO: Switch based on some criteria.
index := rand.Intn(len(SchedPolicies))

6
schedulers/first-fit.go
View file

@ -4,6 +4,7 @@ import (
"bitbucket.org/sunybingcloud/elektron/def"
"bitbucket.org/sunybingcloud/elektron/utilities/mesosUtils"
"bitbucket.org/sunybingcloud/elektron/utilities/offerUtils"
"bitbucket.org/sunybingcloud/elektron/utilities/schedUtils"
"fmt"
mesos "github.com/mesos/mesos-go/api/v0/mesosproto"
sched "github.com/mesos/mesos-go/api/v0/scheduler"
@ -103,6 +104,11 @@ func (s *FirstFit) ConsumeOffers(spc SchedPolicyContext, driver sched.SchedulerD
// Switch scheduling policy only if feature enabled from CLI
if baseSchedRef.schedPolSwitchEnabled {
// Need to recompute the schedWindow for the next offer cycle.
// The next scheduling policy will schedule at max schedWindow number of tasks.
baseSchedRef.schedWindow = schedUtils.
SchedWindowResizingCritToStrategy[baseSchedRef.schedWindowResizeCrit].Apply(
func() interface{} { return baseSchedRef.tasks })
// Switching to a random scheduling policy.
// TODO: Switch based on some criteria.
index := rand.Intn(len(SchedPolicies))

2
schedulers/helpers.go
View file

@ -125,7 +125,7 @@ func WithSchedPolSwitchEnabled(enableSchedPolicySwitch bool) schedPolicyOption {
}
}
// Launch tasks.
// Launch tasks and also update the resource availability for the corresponding host.
func LaunchTasks(offerIDs []*mesos.OfferID, tasksToLaunch []*mesos.TaskInfo, driver sched.SchedulerDriver) error {
driver.LaunchTasks(offerIDs, tasksToLaunch, mesosUtils.DefaultFilter)
// Update resource availability

16
schedulers/store.go
View file

@ -6,18 +6,18 @@ import (
// Names of different scheduling policies.
const (
ff = "first-fit"
bp = "bin-packing"
mgm = "max-greedymins"
mm = "max-min"
ff = "first-fit"
bp = "bin-packing"
mgm = "max-greedymins"
mm = "max-min"
)
// Scheduling policy factory
var SchedPolicies map[string]SchedPolicyState = map[string]SchedPolicyState{
ff: &FirstFit{},
bp: &BinPackSortedWatts{},
mgm: &MaxGreedyMins{},
mm: &MaxMin{},
ff: &FirstFit{},
bp: &BinPackSortedWatts{},
mgm: &MaxGreedyMins{},
mm: &MaxMin{},
}
// build the scheduling policy with the options being applied