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

ignored error returned from LaunchTasks. Technically, this error shouldn't occur. Retrofitted scheduling policies to not receive an error when calling LaunchTasks(...)

Browse source
This commit is contained in:
Pradyumna Kaushik 2018-02-09 17:31:16 -05:00
parent f041e6668b
commit ec18b617da
5 changed files with 7 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

4
schedulers/MaxGreedyMins.go
View file

@ -158,9 +158,7 @@ func (s *MaxGreedyMins) ConsumeOffers(spc SchedPolicyContext, driver sched.Sched
if offerTaken {
baseSchedRef.LogTaskStarting(nil, offer)
if err := LaunchTasks([]*mesos.OfferID{offer.Id}, tasks, driver); err != nil {
baseSchedRef.LogElectronError(err)
}
LaunchTasks([]*mesos.OfferID{offer.Id}, tasks, driver)
} else {
// If there was no match for the task

4
schedulers/MaxMin.go
View file

@ -153,9 +153,7 @@ func (s *MaxMin) ConsumeOffers(spc SchedPolicyContext, driver sched.SchedulerDri
if offerTaken {
baseSchedRef.LogTaskStarting(nil, offer)
if err := LaunchTasks([]*mesos.OfferID{offer.Id}, tasks, driver); err != nil {
baseSchedRef.LogElectronError(err)
}
LaunchTasks([]*mesos.OfferID{offer.Id}, tasks, driver)
} else {
// If there was no match for the task
cpus, mem, watts := offerUtils.OfferAgg(offer)

4
schedulers/bin-packing.go
View file

@ -102,9 +102,7 @@ func (s *BinPackSortedWatts) ConsumeOffers(spc SchedPolicyContext, driver sched.
if offerTaken {
baseSchedRef.LogTaskStarting(nil, offer)
if err := LaunchTasks([]*mesos.OfferID{offer.Id}, tasks, driver); err != nil {
baseSchedRef.LogElectronError(err)
}
LaunchTasks([]*mesos.OfferID{offer.Id}, tasks, driver)
} else {
// If there was no match for the task

4
schedulers/first-fit.go
View file

@ -70,9 +70,7 @@ func (s *FirstFit) ConsumeOffers(spc SchedPolicyContext, driver sched.SchedulerD
tasks = append(tasks, taskToSchedule)
baseSchedRef.LogTaskStarting(&task, offer)
if err := LaunchTasks([]*mesos.OfferID{offer.Id}, tasks, driver); err != nil {
baseSchedRef.LogElectronError(err)
}
LaunchTasks([]*mesos.OfferID{offer.Id}, tasks, driver)
offerTaken = true
baseSchedRef.LogSchedTrace(taskToSchedule, offer)

12
schedulers/helpers.go
View file

@ -9,7 +9,6 @@ import (
mesos "github.com/mesos/mesos-go/api/v0/mesosproto"
sched "github.com/mesos/mesos-go/api/v0/scheduler"
"github.com/pkg/errors"
"log"
)
func coLocated(tasks map[string]bool, s BaseScheduler) {
@ -125,16 +124,11 @@ func WithSchedPolSwitchEnabled(enableSchedPolicySwitch bool) schedPolicyOption {
}
}
// Launch tasks and also update the resource availability for the corresponding host.
func LaunchTasks(offerIDs []*mesos.OfferID, tasksToLaunch []*mesos.TaskInfo, driver sched.SchedulerDriver) error {
// Launch tasks.
func LaunchTasks(offerIDs []*mesos.OfferID, tasksToLaunch []*mesos.TaskInfo, driver sched.SchedulerDriver) {
driver.LaunchTasks(offerIDs, tasksToLaunch, mesosUtils.DefaultFilter)
// Update resource availability
var err error
for _, task := range tasksToLaunch {
err = utilities.ResourceAvailabilityUpdate("ON_TASK_ACTIVE_STATE", *task.TaskId, *task.SlaveId)
if err != nil {
log.Println(err)
}
utilities.ResourceAvailabilityUpdate("ON_TASK_ACTIVE_STATE", *task.TaskId, *task.SlaveId)
}
return err
}