From 3a7086fd8a0b3e307dc83bcb81b1dba1d9f044ab Mon Sep 17 00:00:00 2001 From: Daniel Smith Date: Thu, 14 May 2015 10:55:49 -0700 Subject: [PATCH] Make scheduler optimistic about its bindings --- plugin/pkg/scheduler/scheduler.go | 35 +++++++++++++++++--------- plugin/pkg/scheduler/scheduler_test.go | 11 +++++--- 2 files changed, 31 insertions(+), 15 deletions(-) diff --git a/plugin/pkg/scheduler/scheduler.go b/plugin/pkg/scheduler/scheduler.go index 2b059ed3474..d30a839439c 100644 --- a/plugin/pkg/scheduler/scheduler.go +++ b/plugin/pkg/scheduler/scheduler.go @@ -105,20 +105,21 @@ func (s *Scheduler) Run() { go util.Until(s.scheduleOne, 0, s.config.StopEverything) } -func (s *Scheduler) scheduleOne() { +func (s *Scheduler) schedule() (executeBinding func()) { pod := s.config.NextPod() glog.V(3).Infof("Attempting to schedule: %v", pod) start := time.Now() - defer func() { + recordTime := func() { metrics.E2eSchedulingLatency.Observe(metrics.SinceInMicroseconds(start)) - }() + } dest, err := s.config.Algorithm.Schedule(pod, s.config.MinionLister) metrics.SchedulingAlgorithmLatency.Observe(metrics.SinceInMicroseconds(start)) if err != nil { glog.V(1).Infof("Failed to schedule: %v", pod) s.config.Recorder.Eventf(pod, "failedScheduling", "Error scheduling: %v", err) s.config.Error(pod, err) - return + recordTime() + return func() {} } b := &api.Binding{ ObjectMeta: api.ObjectMeta{Namespace: pod.Namespace, Name: pod.Name}, @@ -128,22 +129,32 @@ func (s *Scheduler) scheduleOne() { }, } - // We want to add the pod to the model iff the bind succeeds, but we don't want to race - // with any deletions, which happen asyncronously. - s.config.Modeler.LockedAction(func() { + // Actually do the binding asynchronously with respect to the scheduling queue. + return func() { + defer recordTime() + defer util.HandleCrash() + + // Make an object representing our assumtion that the bind will succeed. + assumed := *pod + assumed.Spec.Host = dest + s.config.Modeler.AssumePod(&assumed) + bindingStart := time.Now() err := s.config.Binder.Bind(b) metrics.BindingLatency.Observe(metrics.SinceInMicroseconds(bindingStart)) if err != nil { + // Remove our (now invalid) assumption + s.config.Modeler.ForgetPod(&assumed) glog.V(1).Infof("Failed to bind pod: %v", err) s.config.Recorder.Eventf(pod, "failedScheduling", "Binding rejected: %v", err) s.config.Error(pod, err) return } s.config.Recorder.Eventf(pod, "scheduled", "Successfully assigned %v to %v", pod.Name, dest) - // tell the model to assume that this binding took effect. - assumed := *pod - assumed.Spec.Host = dest - s.config.Modeler.AssumePod(&assumed) - }) + } +} + +func (s *Scheduler) scheduleOne() { + bind := s.schedule() + go bind() } diff --git a/plugin/pkg/scheduler/scheduler_test.go b/plugin/pkg/scheduler/scheduler_test.go index 68a72e4dc2d..e39d582b128 100644 --- a/plugin/pkg/scheduler/scheduler_test.go +++ b/plugin/pkg/scheduler/scheduler_test.go @@ -112,6 +112,11 @@ func TestScheduler(t *testing.T) { AssumePodFunc: func(pod *api.Pod) { gotAssumedPod = pod }, + ForgetPodFunc: func(pod *api.Pod) { + if gotAssumedPod != nil && gotAssumedPod.Name == pod.Name && gotAssumedPod.Namespace == pod.Namespace { + gotAssumedPod = nil + } + }, }, MinionLister: scheduler.FakeMinionLister( api.NodeList{Items: []api.Node{{ObjectMeta: api.ObjectMeta{Name: "machine1"}}}}, @@ -138,7 +143,7 @@ func TestScheduler(t *testing.T) { } close(called) }) - s.scheduleOne() + s.schedule()() if e, a := item.expectAssumedPod, gotAssumedPod; !reflect.DeepEqual(e, a) { t.Errorf("%v: assumed pod: wanted %v, got %v", i, e, a) } @@ -228,7 +233,7 @@ func TestSchedulerForgetAssumedPodAfterDelete(t *testing.T) { // scheduledPodStore: [] // assumedPods: [] - s.scheduleOne() + s.schedule()() // queuedPodStore: [] // scheduledPodStore: [foo:8080] // assumedPods: [foo:8080] @@ -282,7 +287,7 @@ func TestSchedulerForgetAssumedPodAfterDelete(t *testing.T) { close(called) }) - s.scheduleOne() + s.schedule()() expectBind = &api.Binding{ ObjectMeta: api.ObjectMeta{Name: "bar"},