Pop from the backoffQ when the activeQ is empty

Author: macsko

pkg/scheduler/backend/queue/active_queue.go

@@ -113,14 +113,22 @@ func (uaq *unlockedActiveQueue) has(pInfo *framework.QueuedPodInfo) bool {
 	return uaq.queue.Has(pInfo)
 }
 
+// backoffQPopper defines method that is used to pop from the backoffQ when the activeQ is empty.
+type backoffQPopper interface {
+	// popBackoff pops the pInfo from the podBackoffQ.
+	popBackoff() (*framework.QueuedPodInfo, error)
+	// len returns length of the podBackoffQ queue.
+	lenBackoff() int
+}
+
 // activeQueue implements activeQueuer. All of the fields have to be protected using the lock.
 type activeQueue struct {
 	// lock synchronizes all operations related to activeQ.
 	// It protects activeQ, inFlightPods, inFlightEvents, schedulingCycle and closed fields.
 	// Caution: DO NOT take "SchedulingQueue.lock" after taking "lock".
 	// You should always take "SchedulingQueue.lock" first, otherwise the queue could end up in deadlock.
-	// "lock" should not be taken after taking "nLock".
-	// Correct locking order is: SchedulingQueue.lock > lock > nominator.nLock.
+	// "lock" should not be taken after taking "backoffQueue.lock" or "nominator.nLock".
+	// Correct locking order is: SchedulingQueue.lock > lock > backoffQueue.lock > nominator.nLock.
 	lock sync.RWMutex
 
 	// activeQ is heap structure that scheduler actively looks at to find pods to
@@ -132,6 +140,8 @@ type activeQueue struct {
 	unlockedQueue *unlockedActiveQueue
 
 	// cond is a condition that is notified when the pod is added to activeQ.
+	// When SchedulerPopFromBackoffQ feature is enabled,
+	// condition is also notified when the pod is added to backoffQ.
 	// It is used with lock.
 	cond sync.Cond
 
@@ -171,16 +181,21 @@ type activeQueue struct {
 	isSchedulingQueueHintEnabled bool
 
 	metricsRecorder metrics.MetricAsyncRecorder
+
+	// backoffQPopper is used to pop from backoffQ when activeQ is empty.
+	// It is non-nil only when SchedulerPopFromBackoffQ feature is enabled.
+	backoffQPopper backoffQPopper
 }
 
-func newActiveQueue(queue *heap.Heap[*framework.QueuedPodInfo], isSchedulingQueueHintEnabled bool, metricRecorder metrics.MetricAsyncRecorder) *activeQueue {
+func newActiveQueue(queue *heap.Heap[*framework.QueuedPodInfo], isSchedulingQueueHintEnabled bool, metricRecorder metrics.MetricAsyncRecorder, backoffQPopper backoffQPopper) *activeQueue {
 	aq := &activeQueue{
 		queue:                        queue,
 		inFlightPods:                 make(map[types.UID]*list.Element),
 		inFlightEvents:               list.New(),
 		isSchedulingQueueHintEnabled: isSchedulingQueueHintEnabled,
 		metricsRecorder:              metricRecorder,
 		unlockedQueue:                newUnlockedActiveQueue(queue),
+		backoffQPopper:               backoffQPopper,
 	}
 	aq.cond.L = &aq.lock
 
@@ -238,7 +253,13 @@ func (aq *activeQueue) pop(logger klog.Logger) (*framework.QueuedPodInfo, error)
 }
 
 func (aq *activeQueue) unlockedPop(logger klog.Logger) (*framework.QueuedPodInfo, error) {
+	var pInfo *framework.QueuedPodInfo
 	for aq.queue.Len() == 0 {
+		// backoffQPopper is non-nil only if SchedulerPopFromBackoffQ feature is enabled.
+		// In case of non-empty backoffQ, try popping from there.
+		if aq.backoffQPopper != nil && aq.backoffQPopper.lenBackoff() != 0 {
+			break
+		}
 		// When the queue is empty, invocation of Pop() is blocked until new item is enqueued.
 		// When Close() is called, the p.closed is set and the condition is broadcast,
 		// which causes this loop to continue and return from the Pop().
@@ -250,7 +271,15 @@ func (aq *activeQueue) unlockedPop(logger klog.Logger) (*framework.QueuedPodInfo
 	}
 	pInfo, err := aq.queue.Pop()
 	if err != nil {
-		return nil, err
+		if aq.backoffQPopper == nil {
+			return nil, err
+		}
+		// Try to pop from backoffQ when activeQ is empty.
+		pInfo, err = aq.backoffQPopper.popBackoff()
+		if err != nil {
+			return nil, err
+		}
+		metrics.SchedulerQueueIncomingPods.WithLabelValues("active", framework.PopFromBackoffQ).Inc()
 	}
 	pInfo.Attempts++
 	pInfo.BackoffExpiration = time.Time{}

pkg/scheduler/backend/queue/active_queue_test.go

@@ -30,7 +30,7 @@ import (
 func TestClose(t *testing.T) {
 	logger, ctx := ktesting.NewTestContext(t)
 	rr := metrics.NewMetricsAsyncRecorder(10, time.Second, ctx.Done())
-	aq := newActiveQueue(heap.NewWithRecorder(podInfoKeyFunc, heap.LessFunc[*framework.QueuedPodInfo](newDefaultQueueSort()), metrics.NewActivePodsRecorder()), true, *rr)
+	aq := newActiveQueue(heap.NewWithRecorder(podInfoKeyFunc, heap.LessFunc[*framework.QueuedPodInfo](newDefaultQueueSort()), metrics.NewActivePodsRecorder()), true, *rr, nil)
 
 	aq.underLock(func(unlockedActiveQ unlockedActiveQueuer) {
 		unlockedActiveQ.add(&framework.QueuedPodInfo{PodInfo: &framework.PodInfo{Pod: st.MakePod().Namespace("foo").Name("p1").UID("p1").Obj()}}, framework.EventUnscheduledPodAdd.Label())

pkg/scheduler/backend/queue/backoff_queue.go

@@ -17,6 +17,7 @@ limitations under the License.
 package queue
 
 import (
+	"sync"
 	"time"
 
 	v1 "k8s.io/api/core/v1"
@@ -35,13 +36,14 @@ import (
 const backoffQOrderingWindowDuration = time.Second
 
 // backoffQueuer is a wrapper for backoffQ related operations.
+// Its methods that relies on the queues, take the lock inside.
 type backoffQueuer interface {
 	// isPodBackingoff returns true if a pod is still waiting for its backoff timer.
 	// If this returns true, the pod should not be re-tried.
 	// If the pod backoff time is in the actual ordering window, it should still be backing off.
 	isPodBackingoff(podInfo *framework.QueuedPodInfo) bool
-	// popEachBackoffCompleted run fn for all pods from podBackoffQ and podErrorBackoffQ that completed backoff while popping them.
-	popEachBackoffCompleted(logger klog.Logger, fn func(pInfo *framework.QueuedPodInfo))
+	// popAllBackoffCompleted pops all pods from podBackoffQ and podErrorBackoffQ that completed backoff.
+	popAllBackoffCompleted(logger klog.Logger) []*framework.QueuedPodInfo
 
 	// podInitialBackoffDuration returns initial backoff duration that pod can get.
 	podInitialBackoffDuration() time.Duration
@@ -61,7 +63,8 @@ type backoffQueuer interface {
 	// It returns new pod info if updated, nil otherwise.
 	update(newPod *v1.Pod, oldPodInfo *framework.QueuedPodInfo) *framework.QueuedPodInfo
 	// delete deletes the pInfo from backoffQueue.
-	delete(pInfo *framework.QueuedPodInfo)
+	// It returns true if the pod was deleted.
+	delete(pInfo *framework.QueuedPodInfo) bool
 	// get returns the pInfo matching given pInfoLookup, if exists.
 	get(pInfoLookup *framework.QueuedPodInfo) (*framework.QueuedPodInfo, bool)
 	// has inform if pInfo exists in the queue.
@@ -75,6 +78,14 @@ type backoffQueuer interface {
 // backoffQueue implements backoffQueuer and wraps two queues inside,
 // providing seamless access as if it were one queue.
 type backoffQueue struct {
+	// lock synchronizes all operations related to backoffQ.
+	// It protects both podBackoffQ and podErrorBackoffQ.
+	// Caution: DO NOT take "SchedulingQueue.lock" or "activeQueue.lock" after taking "lock".
+	// You should always take "SchedulingQueue.lock" and "activeQueue.lock" first, otherwise the queue could end up in deadlock.
+	// "lock" should not be taken after taking "nominator.nLock".
+	// Correct locking order is: SchedulingQueue.lock > activeQueue.lock > lock > nominator.nLock.
+	lock sync.RWMutex
+
 	clock clock.WithTicker
 
 	// podBackoffQ is a heap ordered by backoff expiry. Pods which have completed backoff
@@ -239,7 +250,8 @@ func (bq *backoffQueue) calculateBackoffDuration(podInfo *framework.QueuedPodInf
 	return duration
 }
 
-func (bq *backoffQueue) popEachBackoffCompletedWithQueue(logger klog.Logger, fn func(pInfo *framework.QueuedPodInfo), queue *heap.Heap[*framework.QueuedPodInfo]) {
+func (bq *backoffQueue) popAllBackoffCompletedWithQueue(logger klog.Logger, queue *heap.Heap[*framework.QueuedPodInfo]) []*framework.QueuedPodInfo {
+	var poppedPods []*framework.QueuedPodInfo
 	for {
 		pInfo, ok := queue.Peek()
 		if !ok || pInfo == nil {
@@ -254,23 +266,27 @@ func (bq *backoffQueue) popEachBackoffCompletedWithQueue(logger klog.Logger, fn
 			logger.Error(err, "Unable to pop pod from backoff queue despite backoff completion", "pod", klog.KObj(pod))
 			break
 		}
-		if fn != nil {
-			fn(pInfo)
-		}
+		poppedPods = append(poppedPods, pInfo)
 	}
+	return poppedPods
 }
 
-// popEachBackoffCompleted run fn for all pods from podBackoffQ and podErrorBackoffQ that completed backoff while popping them.
-func (bq *backoffQueue) popEachBackoffCompleted(logger klog.Logger, fn func(pInfo *framework.QueuedPodInfo)) {
+// popAllBackoffCompleted pops all pods from podBackoffQ and podErrorBackoffQ that completed backoff.
+func (bq *backoffQueue) popAllBackoffCompleted(logger klog.Logger) []*framework.QueuedPodInfo {
+	bq.lock.Lock()
+	defer bq.lock.Unlock()
+
 	// Ensure both queues are called
-	bq.popEachBackoffCompletedWithQueue(logger, fn, bq.podBackoffQ)
-	bq.popEachBackoffCompletedWithQueue(logger, fn, bq.podErrorBackoffQ)
+	return append(bq.popAllBackoffCompletedWithQueue(logger, bq.podBackoffQ), bq.popAllBackoffCompletedWithQueue(logger, bq.podErrorBackoffQ)...)
 }
 
 // add adds the pInfo to backoffQueue.
 // The event should show which event triggered this addition and is used for the metric recording.
 // It also ensures that pInfo is not in both queues.
 func (bq *backoffQueue) add(logger klog.Logger, pInfo *framework.QueuedPodInfo, event string) {
+	bq.lock.Lock()
+	defer bq.lock.Unlock()
+
 	// If pod has empty both unschedulable plugins and pending plugins,
 	// it means that it failed because of error and should be moved to podErrorBackoffQ.
 	if pInfo.UnschedulablePlugins.Len() == 0 && pInfo.PendingPlugins.Len() == 0 {
@@ -297,6 +313,9 @@ func (bq *backoffQueue) add(logger klog.Logger, pInfo *framework.QueuedPodInfo,
 // update updates the pod in backoffQueue if oldPodInfo is already in the queue.
 // It returns new pod info if updated, nil otherwise.
 func (bq *backoffQueue) update(newPod *v1.Pod, oldPodInfo *framework.QueuedPodInfo) *framework.QueuedPodInfo {
+	bq.lock.Lock()
+	defer bq.lock.Unlock()
+
 	// If the pod is in the backoff queue, update it there.
 	if pInfo, exists := bq.podBackoffQ.Get(oldPodInfo); exists {
 		_ = pInfo.Update(newPod)
@@ -313,13 +332,32 @@ func (bq *backoffQueue) update(newPod *v1.Pod, oldPodInfo *framework.QueuedPodIn
 }
 
 // delete deletes the pInfo from backoffQueue.
-func (bq *backoffQueue) delete(pInfo *framework.QueuedPodInfo) {
-	_ = bq.podBackoffQ.Delete(pInfo)
-	_ = bq.podErrorBackoffQ.Delete(pInfo)
+// It returns true if the pod was deleted.
+func (bq *backoffQueue) delete(pInfo *framework.QueuedPodInfo) bool {
+	bq.lock.Lock()
+	defer bq.lock.Unlock()
+
+	if bq.podBackoffQ.Delete(pInfo) == nil {
+		return true
+	}
+	return bq.podErrorBackoffQ.Delete(pInfo) == nil
+}
+
+// popBackoff pops the pInfo from the podBackoffQ.
+// It returns error if the queue is empty.
+// This doesn't pop the pods from the podErrorBackoffQ.
+func (bq *backoffQueue) popBackoff() (*framework.QueuedPodInfo, error) {
+	bq.lock.Lock()
+	defer bq.lock.Unlock()
+
+	return bq.podBackoffQ.Pop()
 }
 
 // get returns the pInfo matching given pInfoLookup, if exists.
 func (bq *backoffQueue) get(pInfoLookup *framework.QueuedPodInfo) (*framework.QueuedPodInfo, bool) {
+	bq.lock.RLock()
+	defer bq.lock.RUnlock()
+
 	pInfo, exists := bq.podBackoffQ.Get(pInfoLookup)
 	if exists {
 		return pInfo, true
@@ -329,11 +367,17 @@ func (bq *backoffQueue) get(pInfoLookup *framework.QueuedPodInfo) (*framework.Qu
 
 // has inform if pInfo exists in the queue.
 func (bq *backoffQueue) has(pInfo *framework.QueuedPodInfo) bool {
+	bq.lock.RLock()
+	defer bq.lock.RUnlock()
+
 	return bq.podBackoffQ.Has(pInfo) || bq.podErrorBackoffQ.Has(pInfo)
 }
 
 // list returns all pods that are in the queue.
 func (bq *backoffQueue) list() []*v1.Pod {
+	bq.lock.RLock()
+	defer bq.lock.RUnlock()
+
 	var result []*v1.Pod
 	for _, pInfo := range bq.podBackoffQ.List() {
 		result = append(result, pInfo.Pod)
@@ -346,5 +390,16 @@ func (bq *backoffQueue) list() []*v1.Pod {
 
 // len returns length of the queue.
 func (bq *backoffQueue) len() int {
+	bq.lock.RLock()
+	defer bq.lock.RUnlock()
+
 	return bq.podBackoffQ.Len() + bq.podErrorBackoffQ.Len()
 }
+
+// lenBackoff returns length of the podBackoffQ.
+func (bq *backoffQueue) lenBackoff() int {
+	bq.lock.RLock()
+	defer bq.lock.RUnlock()
+
+	return bq.podBackoffQ.Len()
+}

pkg/scheduler/backend/queue/backoff_queue_test.go

@@ -78,7 +78,7 @@ func TestBackoffQueue_calculateBackoffDuration(t *testing.T) {
 	}
 }
 
-func TestBackoffQueue_popEachBackoffCompleted(t *testing.T) {
+func TestBackoffQueue_popAllBackoffCompleted(t *testing.T) {
 	fakeClock := testingclock.NewFakeClock(time.Now())
 	podInfos := map[string]*framework.QueuedPodInfo{
 		"pod0": {
@@ -156,10 +156,11 @@ func TestBackoffQueue_popEachBackoffCompleted(t *testing.T) {
 				for _, podName := range tt.podsInBackoff {
 					bq.add(logger, podInfos[podName], framework.EventUnscheduledPodAdd.Label())
 				}
+				gotPodInfos := bq.popAllBackoffCompleted(logger)
 				var gotPods []string
-				bq.popEachBackoffCompleted(logger, func(pInfo *framework.QueuedPodInfo) {
+				for _, pInfo := range gotPodInfos {
 					gotPods = append(gotPods, pInfo.Pod.Name)
-				})
+				}
 				if diff := cmp.Diff(tt.wantPods, gotPods); diff != "" {
 					t.Errorf("Unexpected pods moved (-want, +got):\n%s", diff)
 				}
@@ -248,10 +249,11 @@ func TestBackoffQueueOrdering(t *testing.T) {
 			for _, podInfo := range podInfos {
 				bq.add(logger, podInfo, framework.EventUnscheduledPodAdd.Label())
 			}
+			gotPodInfos := bq.popAllBackoffCompleted(logger)
 			var gotPods []string
-			bq.popEachBackoffCompleted(logger, func(pInfo *framework.QueuedPodInfo) {
+			for _, pInfo := range gotPodInfos {
 				gotPods = append(gotPods, pInfo.Pod.Name)
-			})
+			}
 			if diff := cmp.Diff(tt.wantPods, gotPods); diff != "" {
 				t.Errorf("Unexpected pods moved (-want, +got):\n%s", diff)
 			}

pkg/scheduler/backend/queue/nominator.go

@@ -35,10 +35,10 @@ import (
 type nominator struct {
 	// nLock synchronizes all operations related to nominator.
 	// It should not be used anywhere else.
-	// Caution: DO NOT take ("SchedulingQueue.lock" or "activeQueue.lock") after taking "nLock".
-	// You should always take "SchedulingQueue.lock" and "activeQueue.lock" first,
+	// Caution: DO NOT take ("SchedulingQueue.lock" or "activeQueue.lock" or "backoffQueue.lock") after taking "nLock".
+	// You should always take "SchedulingQueue.lock" and "activeQueue.lock" and "backoffQueue.lock" first,
 	// otherwise the nominator could end up in deadlock.
-	// Correct locking order is: SchedulingQueue.lock > activeQueue.lock > nLock.
+	// Correct locking order is: SchedulingQueue.lock > activeQueue.lock = backoffQueue.lock > nLock.
 	nLock sync.RWMutex
 
 	// podLister is used to verify if the given pod is alive.