diff --git a/plugin/pkg/scheduler/scheduler.go b/plugin/pkg/scheduler/scheduler.go index 7d3830fcac2..6b51cd4e790 100644 --- a/plugin/pkg/scheduler/scheduler.go +++ b/plugin/pkg/scheduler/scheduler.go @@ -104,20 +104,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}, @@ -127,22 +128,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 39093a4c2e8..9ec4c7029fb 100644 --- a/plugin/pkg/scheduler/scheduler_test.go +++ b/plugin/pkg/scheduler/scheduler_test.go @@ -113,6 +113,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: algorithm.FakeMinionLister( api.NodeList{Items: []api.Node{{ObjectMeta: api.ObjectMeta{Name: "machine1"}}}}, @@ -139,7 +144,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) } @@ -229,7 +234,7 @@ func TestSchedulerForgetAssumedPodAfterDelete(t *testing.T) { // scheduledPodStore: [] // assumedPods: [] - s.scheduleOne() + s.schedule()() // queuedPodStore: [] // scheduledPodStore: [foo:8080] // assumedPods: [foo:8080] @@ -283,7 +288,7 @@ func TestSchedulerForgetAssumedPodAfterDelete(t *testing.T) { close(called) }) - s.scheduleOne() + s.schedule()() expectBind = &api.Binding{ ObjectMeta: api.ObjectMeta{Name: "bar"},