K8s源码分析(二)-K8s调度队列介绍

本文首发在个人博客上,欢迎来踩!
本次分析参考的K8s版本是

文章目录

  • 调度队列简介
  • 调度队列源代码分析
    • 队列初始化
    • QueuedPodInfo元素介绍
    • ActiveQ源代码介绍
    • UnschedulableQ源代码介绍
    • **BackoffQ**源代码介绍
    • 队列弹出待调度的Pod
    • 队列增加新的待调度的Pod
    • pod调度失败返回队列的处理
    • flushBackoffQCompleted
    • flushUnschedulablePodsLeftover
  • 调度队列总结

调度队列简介

这里是官方对于K8s中调度队列的介绍,很值得一看:Scheduling queue in kube-scheduler。整体的架构如下图所示。

请添加图片描述

简单来说K8s中的调度队列主要有3种:

  • ActiveQ(heap结构):在每个调度周期开始时都会从这里取出一个Pod尝试调度。一开始提交的所有没有指定.spec.nodeName的Pod都会发送到这里,也会接收来自unschedulableQ和BackoffQ刷新来的pod。默认的排序规则是按照优先级进行排列,高优先级的Pod在前面。
  • UnschedulableQ(Map结构):存储调度失败的Pod,以等待资源更新、其他相关Pod调度成功等事件,从而将其的Pod其进行重调度。
  • BackoffQ(heap结构):用来暂时退避的队列,默认的排列规则是按退避时间的长度进行排序,需要退避的时间短的Pod在前面。为了防止Pod频繁的重调度,每个Pod都会记录自己的重调度次数,退避时间随着每次失败的调度尝试呈指数增长,直到达到最大值,例如尝试失败 3 次的 Pod 的目标退避超时设置为 curTime + 2s^3 (8s)。注意有两种情况下Pod会进入到BackoffQ队列中:
    • unscheduleableQ会定时对其中的所有pod进行重调度,那么就需要计算各个pod是否退避了足够的时间,如果没有就放入到BackoffQ中再退避一段时间。
    • 如果一个Pod调度失败时,正好这时又异步地发生了资源变更事件(p.moveRequestCycle **>=** podSchedulingCycle )(schedulingCycle 是当前调度的周期,ActiveQ队列每pop一个pod,就加1,moveRequestCycle是事件发生时schedulingCycle 的值),那么就不会放入UnschedulableQ中,而是会直接放入到BackoffQ中。

调度队列机制有两个在后台运行的定期刷新 go协程,负责将 pod 移动到活动队列,后续也将详细介绍相关代码:

  • **flushUnschedulablePodsLeftover:**每 30 秒运行一次,将 Pod 从UnschedulableQ中移动,以允许未由任何事件移动的不可调度的 Pod 再次重试。
  • **flushBackoffQCompleted:**每1秒运行一次,将BackoffQ中已经回避了足够久的Pod移动到ActiveQ队列中

移动请求(move request)会触发一个事件,该事件负责将 Pod 从UnschedulableQ移动到ActiveQ或BackoffQ。集群中许多事件可以触发移动请求的发生,包括了 Pod、节点、服务、PV、PVC、存储类和 CSI 节点的更改。例如当某些pod被调度时,UnschedulableQ中与其具有亲和性要求而导致之前无法调度的pod就会被移动出去,或者当某个新node加入时,原本因为资源不够导致无法调度的Pod也会被移动出去。

调度队列源代码分析

队列初始化

Scheduler中的调度队列SchedulingQueueinternalqueue.SchedulingQueue类型,该类型的实现在pkg/scheduler/internal/queue/scheduling_queue.go:92,如下。

// SchedulingQueue is an interface for a queue to store pods waiting to be scheduled.
// The interface follows a pattern similar to cache.FIFO and cache.Heap and
// makes it easy to use those data structures as a SchedulingQueue.
type SchedulingQueue interface {framework.PodNominatorAdd(pod *v1.Pod) error// Activate moves the given pods to activeQ iff they're in unschedulablePods or backoffQ.// The passed-in pods are originally compiled from plugins that want to activate Pods,// by injecting the pods through a reserved CycleState struct (PodsToActivate).Activate(pods map[string]*v1.Pod)// AddUnschedulableIfNotPresent adds an unschedulable pod back to scheduling queue.// The podSchedulingCycle represents the current scheduling cycle number which can be// returned by calling SchedulingCycle().AddUnschedulableIfNotPresent(pod *framework.QueuedPodInfo, podSchedulingCycle int64) error// SchedulingCycle returns the current number of scheduling cycle which is// cached by scheduling queue. Normally, incrementing this number whenever// a pod is popped (e.g. called Pop()) is enough.SchedulingCycle() int64// Pop removes the head of the queue and returns it. It blocks if the// queue is empty and waits until a new item is added to the queue.Pop() (*framework.QueuedPodInfo, error)Update(oldPod, newPod *v1.Pod) errorDelete(pod *v1.Pod) errorMoveAllToActiveOrBackoffQueue(event framework.ClusterEvent, preCheck PreEnqueueCheck)AssignedPodAdded(pod *v1.Pod)AssignedPodUpdated(pod *v1.Pod)PendingPods() ([]*v1.Pod, string)// Close closes the SchedulingQueue so that the goroutine which is// waiting to pop items can exit gracefully.Close()// Run starts the goroutines managing the queue.Run()
}

上述代码定义了其需要的对队列中的元素添加、删除、更新、获取、运行等方法。而其标准实现PriorityQueue

pkg/scheduler/internal/queue/scheduling_queue.go:145 中,首先查看其需要的变量:

// PriorityQueue implements a scheduling queue.
// The head of PriorityQueue is the highest priority pending pod. This structure
// has two sub queues and a additional data structure, namely: activeQ,
// backoffQ and unschedulablePods.
//   - activeQ holds pods that are being considered for scheduling.
//   - backoffQ holds pods that moved from unschedulablePods and will move to
//     activeQ when their backoff periods complete.
//   - unschedulablePods holds pods that were already attempted for scheduling and
//     are currently determined to be unschedulable.
type PriorityQueue struct {*nominatorstop  chan struct{}clock clock.Clock// pod initial backoff duration.podInitialBackoffDuration time.Duration// pod maximum backoff duration.podMaxBackoffDuration time.Duration// the maximum time a pod can stay in the unschedulablePods.podMaxInUnschedulablePodsDuration time.Durationcond sync.Cond// activeQ is heap structure that scheduler actively looks at to find pods to// schedule. Head of heap is the highest priority pod.activeQ *heap.Heap// podBackoffQ is a heap ordered by backoff expiry. Pods which have completed backoff// are popped from this heap before the scheduler looks at activeQpodBackoffQ *heap.Heap// unschedulablePods holds pods that have been tried and determined unschedulable.unschedulablePods *UnschedulablePods// schedulingCycle represents sequence number of scheduling cycle and is incremented// when a pod is popped.schedulingCycle int64// moveRequestCycle caches the sequence number of scheduling cycle when we// received a move request. Unschedulable pods in and before this scheduling// cycle will be put back to activeQueue if we were trying to schedule them// when we received move request.moveRequestCycle int64clusterEventMap map[framework.ClusterEvent]sets.String// preEnqueuePluginMap is keyed with profile name, valued with registered preEnqueue plugins.preEnqueuePluginMap map[string][]framework.PreEnqueuePlugin// closed indicates that the queue is closed.// It is mainly used to let Pop() exit its control loop while waiting for an item.closed boolnsLister listersv1.NamespaceListermetricsRecorder metrics.MetricAsyncRecorder// pluginMetricsSamplePercent is the percentage of plugin metrics to be sampled.pluginMetricsSamplePercent int
}

pkg/scheduler/internal/queue/scheduling_queue.go:291 中给出了生成了该队列的初始化方法

// NewPriorityQueue creates a PriorityQueue object.
func NewPriorityQueue(lessFn framework.LessFunc,informerFactory informers.SharedInformerFactory,opts ...Option,
) *PriorityQueue {options := defaultPriorityQueueOptionsif options.podLister == nil {options.podLister = informerFactory.Core().V1().Pods().Lister()}for _, opt := range opts {opt(&options)}comp := func(podInfo1, podInfo2 interface{}) bool {pInfo1 := podInfo1.(*framework.QueuedPodInfo)pInfo2 := podInfo2.(*framework.QueuedPodInfo)return lessFn(pInfo1, pInfo2)}pq := &PriorityQueue{nominator:                         newPodNominator(options.podLister),clock:                             options.clock,stop:                              make(chan struct{}),podInitialBackoffDuration:         options.podInitialBackoffDuration,podMaxBackoffDuration:             options.podMaxBackoffDuration,podMaxInUnschedulablePodsDuration: options.podMaxInUnschedulablePodsDuration,activeQ:                           heap.NewWithRecorder(podInfoKeyFunc, comp, metrics.NewActivePodsRecorder()),unschedulablePods:                 newUnschedulablePods(metrics.NewUnschedulablePodsRecorder(), metrics.NewGatedPodsRecorder()),moveRequestCycle:                  -1,clusterEventMap:                   options.clusterEventMap,preEnqueuePluginMap:               options.preEnqueuePluginMap,metricsRecorder:                   options.metricsRecorder,pluginMetricsSamplePercent:        options.pluginMetricsSamplePercent,}pq.cond.L = &pq.lockpq.podBackoffQ = heap.NewWithRecorder(podInfoKeyFunc, pq.podsCompareBackoffCompleted, metrics.NewBackoffPodsRecorder())pq.nsLister = informerFactory.Core().V1().Namespaces().Lister()return pq
}

可以看到其包含了许多我们上面介绍的概念,包括activeQunschedulablePodspodBackoffQschedulingCyclemoveRequestCycle

QueuedPodInfo元素介绍

这里也多次出现了QueuedPodInfo这个关键的数据结构,它是Pod中的基础元素,在此进行介绍,其定义在pkg/scheduler/framework/types.go:98 中,包括了PodInfo、添加时间、尝试次数等

// QueuedPodInfo is a Pod wrapper with additional information related to
// the pod's status in the scheduling queue, such as the timestamp when
// it's added to the queue.
type QueuedPodInfo struct {*PodInfo// The time pod added to the scheduling queue.Timestamp time.Time// Number of schedule attempts before successfully scheduled.// It's used to record the # attempts metric.Attempts int// The time when the pod is added to the queue for the first time. The pod may be added// back to the queue multiple times before it's successfully scheduled.// It shouldn't be updated once initialized. It's used to record the e2e scheduling// latency for a pod.InitialAttemptTimestamp time.Time// If a Pod failed in a scheduling cycle, record the plugin names it failed by.UnschedulablePlugins sets.String// Whether the Pod is scheduling gated (by PreEnqueuePlugins) or not.Gated bool
}

PodInfo的定义在pkg/scheduler/framework/types.go:131

// PodInfo is a wrapper to a Pod with additional pre-computed information to
// accelerate processing. This information is typically immutable (e.g., pre-processed
// inter-pod affinity selectors).
type PodInfo struct {Pod                        *v1.PodRequiredAffinityTerms      []AffinityTermRequiredAntiAffinityTerms  []AffinityTermPreferredAffinityTerms     []WeightedAffinityTermPreferredAntiAffinityTerms []WeightedAffinityTerm
}

Pod的定义在staging/src/k8s.io/api/core/v1/types.go:4202

// Pod is a collection of containers that can run on a host. This resource is created
// by clients and scheduled onto hosts.
type Pod struct {metav1.TypeMeta `json:",inline"`// Standard object's metadata.// More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#metadata// +optionalmetav1.ObjectMeta `json:"metadata,omitempty" protobuf:"bytes,1,opt,name=metadata"`// Specification of the desired behavior of the pod.// More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#spec-and-status// +optionalSpec PodSpec `json:"spec,omitempty" protobuf:"bytes,2,opt,name=spec"`// Most recently observed status of the pod.// This data may not be up to date.// Populated by the system.// Read-only.// More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#spec-and-status// +optionalStatus PodStatus `json:"status,omitempty" protobuf:"bytes,3,opt,name=status"`
}

ActiveQ源代码介绍

从初始化代码中可以看到ActiveQ是一个heap,其相关定义在pkg/scheduler/internal/heap/heap.go

// Heap is a producer/consumer queue that implements a heap data structure.
// It can be used to implement priority queues and similar data structures.
type Heap struct {// data stores objects and has a queue that keeps their ordering according// to the heap invariant.data *data// metricRecorder updates the counter when elements of a heap get added or// removed, and it does nothing if it's nilmetricRecorder metrics.MetricRecorder
}
// data is an internal struct that implements the standard heap interface
// and keeps the data stored in the heap.
type data struct {// items is a map from key of the objects to the objects and their index.// We depend on the property that items in the map are in the queue and vice versa.items map[string]*heapItem// queue implements a heap data structure and keeps the order of elements// according to the heap invariant. The queue keeps the keys of objects stored// in "items".queue []string// keyFunc is used to make the key used for queued item insertion and retrieval, and// should be deterministic.keyFunc KeyFunc// lessFunc is used to compare two objects in the heap.lessFunc lessFunc
}

可以看到他是用queue实现了一个heap。

ActiveQ的默认排序代码在pkg/scheduler/framework/plugins/queuesort/priority_sort.go:42中,即按优先级进行排序,如果优先级相同就提交时间的早晚进行排序。

// Less is the function used by the activeQ heap algorithm to sort pods.
// It sorts pods based on their priority. When priorities are equal, it uses
// PodQueueInfo.timestamp.
func (pl *PrioritySort) Less(pInfo1, pInfo2 *framework.QueuedPodInfo) bool {p1 := corev1helpers.PodPriority(pInfo1.Pod)p2 := corev1helpers.PodPriority(pInfo2.Pod)return (p1 > p2) || (p1 == p2 && pInfo1.Timestamp.Before(pInfo2.Timestamp))
}

UnschedulableQ源代码介绍

UnschedulableQ进行初始化的具体代码在pkg/scheduler/internal/queue/scheduling_queue.go:998

// newUnschedulablePods initializes a new object of UnschedulablePods.
func newUnschedulablePods(unschedulableRecorder, gatedRecorder metrics.MetricRecorder) *UnschedulablePods {return &UnschedulablePods{podInfoMap:            make(map[string]*framework.QueuedPodInfo),keyFunc:               util.GetPodFullName,unschedulableRecorder: unschedulableRecorder,gatedRecorder:         gatedRecorder,}
}

其具体的定义代码在pkg/scheduler/internal/queue/scheduling_queue.go:939 ,

// UnschedulablePods holds pods that cannot be scheduled. This data structure
// is used to implement unschedulablePods.
type UnschedulablePods struct {// podInfoMap is a map key by a pod's full-name and the value is a pointer to the QueuedPodInfo.podInfoMap map[string]*framework.QueuedPodInfokeyFunc    func(*v1.Pod) string// unschedulableRecorder/gatedRecorder updates the counter when elements of an unschedulablePodsMap// get added or removed, and it does nothing if it's nil.unschedulableRecorder, gatedRecorder metrics.MetricRecorder
}

可以看到他没有进行heap的包装,而是直接采用Map结构进行保存。

BackoffQ源代码介绍

BackoffQ也是一个heap,与ActiveQ不同的一点在于排序函数不同,其排序函数的定义在pkg/scheduler/internal/queue/scheduling_queue.go:888

func (p *PriorityQueue) podsCompareBackoffCompleted(podInfo1, podInfo2 interface{}) bool {pInfo1 := podInfo1.(*framework.QueuedPodInfo)pInfo2 := podInfo2.(*framework.QueuedPodInfo)bo1 := p.getBackoffTime(pInfo1)bo2 := p.getBackoffTime(pInfo2)return bo1.Before(bo2)
}

getBackoffTime的定义在pkg/scheduler/internal/queue/scheduling_queue.go:911中,即计算完成避让的时间

// getBackoffTime returns the time that podInfo completes backoff
func (p *PriorityQueue) getBackoffTime(podInfo *framework.QueuedPodInfo) time.Time {duration := p.calculateBackoffDuration(podInfo)backoffTime := podInfo.Timestamp.Add(duration)return backoffTime
}

可以看到队列排序时会将完成避让最早的pod放在前面。

然后再看其是如何计算避让时间的,在pkg/scheduler/internal/queue/scheduling_queue.go

// calculateBackoffDuration is a helper function for calculating the backoffDuration
// based on the number of attempts the pod has made.
func (p *PriorityQueue) calculateBackoffDuration(podInfo *framework.QueuedPodInfo) time.Duration {duration := p.podInitialBackoffDurationfor i := 1; i < podInfo.Attempts; i++ {// Use subtraction instead of addition or multiplication to avoid overflow.if duration > p.podMaxBackoffDuration-duration {return p.podMaxBackoffDuration}duration += duration}return duration
}

其计算可以理解为初次为p.podInitialBackoffDuration,每次需要的避让时间都是前一次的两倍,如果计算得到的避让时间大于p.podMaxBackoffDuration/2 ,就将避让时间设置为p.podMaxBackoffDuration

队列弹出待调度的Pod

其代码在pkg/scheduler/internal/queue/scheduling_queue.go:593

// Pop removes the head of the active queue and returns it. It blocks if the
// activeQ is empty and waits until a new item is added to the queue. It
// increments scheduling cycle when a pod is popped.
func (p *PriorityQueue) Pop() (*framework.QueuedPodInfo, error) {p.lock.Lock()defer p.lock.Unlock()for p.activeQ.Len() == 0 {// 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().if p.closed {return nil, fmt.Errorf(queueClosed)}p.cond.Wait()}obj, err := p.activeQ.Pop()if err != nil {return nil, err}pInfo := obj.(*framework.QueuedPodInfo)pInfo.Attempts++p.schedulingCycle++return pInfo, nil
}

可以看到如果activeQ中没有需要调度的Pod了那么就会使用p.cond.Wait来进行等待,否则就冲activeQ中Pop一个元素QueuedPodInfo,同时这个QueuedPodInfo 的Attempts会+1,整个队列中的schedulingCycle也会增加。

队列增加新的待调度的Pod

其代码在pkg/scheduler/internal/queue/scheduling_queue.go:398

// Add adds a pod to the active queue. It should be called only when a new pod
// is added so there is no chance the pod is already in active/unschedulable/backoff queues
func (p *PriorityQueue) Add(pod *v1.Pod) error {p.lock.Lock()defer p.lock.Unlock()pInfo := p.newQueuedPodInfo(pod)gated := pInfo.Gatedif added, err := p.addToActiveQ(pInfo); !added {return err}if p.unschedulablePods.get(pod) != nil {klog.ErrorS(nil, "Error: pod is already in the unschedulable queue", "pod", klog.KObj(pod))p.unschedulablePods.delete(pod, gated)}// Delete pod from backoffQ if it is backing offif err := p.podBackoffQ.Delete(pInfo); err == nil {klog.ErrorS(nil, "Error: pod is already in the podBackoff queue", "pod", klog.KObj(pod))}klog.V(5).InfoS("Pod moved to an internal scheduling queue", "pod", klog.KObj(pod), "event", PodAdd, "queue", activeQName)metrics.SchedulerQueueIncomingPods.WithLabelValues("active", PodAdd).Inc()p.addNominatedPodUnlocked(pInfo.PodInfo, nil)p.cond.Broadcast()return nil
}

主要是将要加入的pod转化为QueuedPodInfo类型,然后添加到activeQ队列中,还需要检查其他队列中是否有这个pod,如果有就删除,同时做一些日志相关记录,然后还会调用p.cond.Broadcast()来解除上述提到的p.cond.Wait 的等待。

pod调度失败返回队列的处理

当Pod调度失败后,会调用来AddUnschedulableIfNotPresent函数来进行处理,其代码位置在pkg/scheduler/internal/queue/scheduling_queue.go 中。

// AddUnschedulableIfNotPresent inserts a pod that cannot be scheduled into
// the queue, unless it is already in the queue. Normally, PriorityQueue puts
// unschedulable pods in `unschedulablePods`. But if there has been a recent move
// request, then the pod is put in `podBackoffQ`.
func (p *PriorityQueue) AddUnschedulableIfNotPresent(pInfo *framework.QueuedPodInfo, podSchedulingCycle int64) error {p.lock.Lock()defer p.lock.Unlock()pod := pInfo.Podif p.unschedulablePods.get(pod) != nil {return fmt.Errorf("Pod %v is already present in unschedulable queue", klog.KObj(pod))}if _, exists, _ := p.activeQ.Get(pInfo); exists {return fmt.Errorf("Pod %v is already present in the active queue", klog.KObj(pod))}if _, exists, _ := p.podBackoffQ.Get(pInfo); exists {return fmt.Errorf("Pod %v is already present in the backoff queue", klog.KObj(pod))}// Refresh the timestamp since the pod is re-added.pInfo.Timestamp = p.clock.Now()// If a move request has been received, move it to the BackoffQ, otherwise move// it to unschedulablePods.for plugin := range pInfo.UnschedulablePlugins {metrics.UnschedulableReason(plugin, pInfo.Pod.Spec.SchedulerName).Inc()}if p.moveRequestCycle >= podSchedulingCycle {if err := p.podBackoffQ.Add(pInfo); err != nil {return fmt.Errorf("error adding pod %v to the backoff queue: %v", klog.KObj(pod), err)}klog.V(5).InfoS("Pod moved to an internal scheduling queue", "pod", klog.KObj(pod), "event", ScheduleAttemptFailure, "queue", backoffQName)metrics.SchedulerQueueIncomingPods.WithLabelValues("backoff", ScheduleAttemptFailure).Inc()} else {p.unschedulablePods.addOrUpdate(pInfo)klog.V(5).InfoS("Pod moved to an internal scheduling queue", "pod", klog.KObj(pod), "event", ScheduleAttemptFailure, "queue", unschedulablePods)metrics.SchedulerQueueIncomingPods.WithLabelValues("unschedulable", ScheduleAttemptFailure).Inc()}p.addNominatedPodUnlocked(pInfo.PodInfo, nil)return nil
}

这里首先检查了其他队列中是否含有该pod,如果有就返回错误,然后比较moveRequestCyclepodSchedulingCycle ,如果p.moveRequestCycle >= podSchedulingCycle 那就说明在刚刚调度这个pod的时候集群发生了变化,可能现在可以成功调度这个pod了,将其转入backoffQ中,不然就正常加入unschedulableQ中。

flushBackoffQCompleted

在队列运行时会初始化两个go协程,来分别不停检查backoffQunschedulableQ,以及时将相关的Pod移出。代码在pkg/scheduler/internal/queue/scheduling_queue.go:333

// Run starts the goroutine to pump from podBackoffQ to activeQ
func (p *PriorityQueue) Run() {go wait.Until(p.flushBackoffQCompleted, 1.0*time.Second, p.stop)go wait.Until(p.flushUnschedulablePodsLeftover, 30*time.Second, p.stop)
}

对于flushBackoffQCompleted即是每1s运行一次,直到接收到p.stop信息。对flushBackoffQCompleted 函数的具体定义在pkg/scheduler/internal/queue/scheduling_queue.go:537中,如下

// flushBackoffQCompleted Moves all pods from backoffQ which have completed backoff in to activeQ
func (p *PriorityQueue) flushBackoffQCompleted() {p.lock.Lock()defer p.lock.Unlock()activated := falsefor {rawPodInfo := p.podBackoffQ.Peek()if rawPodInfo == nil {break}pInfo := rawPodInfo.(*framework.QueuedPodInfo)pod := pInfo.Podif p.isPodBackingoff(pInfo) {break}_, err := p.podBackoffQ.Pop()if err != nil {klog.ErrorS(err, "Unable to pop pod from backoff queue despite backoff completion", "pod", klog.KObj(pod))break}if err := p.activeQ.Add(pInfo); err != nil {klog.ErrorS(err, "Error adding pod to the active queue", "pod", klog.KObj(pInfo.Pod))} else {klog.V(5).InfoS("Pod moved to an internal scheduling queue", "pod", klog.KObj(pod), "event", BackoffComplete, "queue", activeQName)metrics.SchedulerQueueIncomingPods.WithLabelValues("active", BackoffComplete).Inc()activated = true}}if activated {p.cond.Broadcast()}
}

其主要内容就是从backOffQ的首个元素开始查看,检查器是否已经过了避让时间,如果过了就将其放入到activeQ队列中,直到首个元素没有达到避让时间或者队列为空。

flushUnschedulablePodsLeftover

flushUnschedulablePodsLeftover每30s运行一次,这部分的代码在pkg/scheduler/internal/queue/scheduling_queue.go:572中,如下

// flushUnschedulablePodsLeftover moves pods which stay in unschedulablePods
// longer than podMaxInUnschedulablePodsDuration to backoffQ or activeQ.
func (p *PriorityQueue) flushUnschedulablePodsLeftover() {p.lock.Lock()defer p.lock.Unlock()var podsToMove []*framework.QueuedPodInfocurrentTime := p.clock.Now()for _, pInfo := range p.unschedulablePods.podInfoMap {lastScheduleTime := pInfo.Timestampif currentTime.Sub(lastScheduleTime) > p.podMaxInUnschedulablePodsDuration {podsToMove = append(podsToMove, pInfo)}}if len(podsToMove) > 0 {p.movePodsToActiveOrBackoffQueue(podsToMove, UnschedulableTimeout)}
}

可以看到其主要作用是遍历所有的pod,如果其在unschedulableQ中呆的时间如果超过了最大的p.podMaxInUnschedulablePodsDuration时间,就会将其移出去,至于是移动到activeQ中还是移动到backoffQ中,取决于movePodsToActiveOrBackoffQueue函数,在pkg/scheduler/internal/queue/scheduling_queue.go:771中,如下

// NOTE: this function assumes lock has been acquired in caller
func (p *PriorityQueue) movePodsToActiveOrBackoffQueue(podInfoList []*framework.QueuedPodInfo, event framework.ClusterEvent) {activated := falsefor _, pInfo := range podInfoList {// If the event doesn't help making the Pod schedulable, continue.// Note: we don't run the check if pInfo.UnschedulablePlugins is nil, which denotes// either there is some abnormal error, or scheduling the pod failed by plugins other than PreFilter, Filter and Permit.// In that case, it's desired to move it anyways.if len(pInfo.UnschedulablePlugins) != 0 && !p.podMatchesEvent(pInfo, event) {continue}pod := pInfo.Podif p.isPodBackingoff(pInfo) {if err := p.podBackoffQ.Add(pInfo); err != nil {klog.ErrorS(err, "Error adding pod to the backoff queue", "pod", klog.KObj(pod))} else {klog.V(5).InfoS("Pod moved to an internal scheduling queue", "pod", klog.KObj(pInfo.Pod), "event", event, "queue", backoffQName)metrics.SchedulerQueueIncomingPods.WithLabelValues("backoff", event.Label).Inc()p.unschedulablePods.delete(pod, pInfo.Gated)}} else {gated := pInfo.Gatedif added, _ := p.addToActiveQ(pInfo); added {klog.V(5).InfoS("Pod moved to an internal scheduling queue", "pod", klog.KObj(pInfo.Pod), "event", event, "queue", activeQName)activated = truemetrics.SchedulerQueueIncomingPods.WithLabelValues("active", event.Label).Inc()p.unschedulablePods.delete(pod, gated)}}}p.moveRequestCycle = p.schedulingCycleif activated {p.cond.Broadcast()}
}

注意这个函数不仅仅是在flushUnschedulablePodsLeftover中被调用,还会在处理其他移动请求时触发,只不过这里的移动请求是UnschedulableTimeout ,判断到底是如何移动也很容易从代码中看出,如果已经到达了避让时间,就加入到activeQ中,如果没有就加入到backoffQ中,注意到如果有移动进activeQ中,也是需要执行p.cond.Broadcast(),同时注意到这里更新了moveRequestCycleschedulingCycle,这也是其统一更新moveRequestCycle 的地方。

调度队列总结

考虑到调度队列的细节,我们可以用下图来对其进行归纳回顾。

请添加图片描述

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://xiahunao.cn/news/3029871.html

如若内容造成侵权/违法违规/事实不符,请联系瞎胡闹网进行投诉反馈,一经查实,立即删除!

相关文章

数据分析:基于sparcc的co-occurrence网络

介绍 Sparcc是基于16s或metagenomics数据等计算组成数据之间关联关系的算法。通常使用count matrix数据。 安装Sparcc软件 git clone gitgithub.com:JCSzamosi/SparCC3.git export PATH/path/SparCC3:$PATHwhich SparCC.py导入数据 注&#xff1a;使用rarefy抽平的count ma…

牛客小白月赛93

B交换数字 题目&#xff1a; 思路&#xff1a;我们可以知道&#xff0c;a*b% mod (a%mod) * (b%mod) 代码&#xff1a; void solve(){int n;cin >> n;string a, b;cin >> a >> b;for(int i 0;i < n;i )if(a[i] > b[i])swap(a[i], b[i]);int num1…

图片无损压缩工具-VIKY

一、前言 Viky v3.4是一款功能强大的图片压缩工具&#xff0c;它能够提供高效的图片无损压缩服务。通过使用独特的压缩算法&#xff0c;该软件在显著减小图片文件大小的同时&#xff0c;还保持了图像的清晰度和色彩饱和度&#xff0c;确保了图像质量的优异表现。 二、软件特点…

AS-VJ900实时视频拼接系统产品介绍:两画面视频拼接方法和操作

目录 一、实时视频拼接系统介绍 &#xff08;一&#xff09;实时视频拼接的定义 &#xff08;二&#xff09;无缝拼接 &#xff08;三&#xff09;AS-VJ900功能介绍 1、功能 2、拼接界面介绍 二、拼接前的准备 &#xff08;一&#xff09;摄像机选择 &#xff08;二&a…

区块链 | NFT 水印:Review on Watermarking Techniques(三)

&#x1f34d;原文&#xff1a;Review on Watermarking Techniques Aiming Authentication of Digital Image Artistic Works Minted as NFTs into Blockchains 一个 NFT 的水印认证协议 可以引入第三方实体来实现对交易的认证&#xff0c;即通过使用 R S A \mathsf{RSA} RSA…

力扣每日一题37:解数独

目录 题目 大致思路 方法一&#xff1a;回溯剪枝&#xff08;正常人能想出来的&#xff09; 方法二&#xff1a;位运算优化剪枝 需要使用的位运算技巧 代码 位运算怎么就优化了呢&#xff1f; 方法三&#xff1a;枚举优化。 官解代码 方法四&#xff1a;舞蹈链算法 题…

FreeRTOS任务调度器

目录 1、什么是任务调度器 2、FreeRTOS中的任务调度器 2.1 抢占式调度 2.2 时间片调度 2.3 协作式调度 3、任务调度案例分析 3.1 实验需求 3.2 CubeMX配置 3.3 代码实现 3.3.1 uart.c 重定向printf 3.3.2 打开freertos.c并添加代码 3.3.4 代码现象 1、什么是任务调度…

7 系列 FPGA 产品介绍及选型

目录 Spartan-7 FPGAsArtix-7 FPGAsKintex-7 FPGAsVirtex-7 FPGAsFPGA芯片命名规则DSP资源BRAM资源Transceivers 资源Transceivers 总带宽I/O 个数及带宽参考文档 Spartan-7 FPGAs Artix-7 FPGAs Kintex-7 FPGAs Virtex-7 FPGAs FPGA芯片命名规则 DSP资源 BRAM资源 Transceiver…

day5Qt作业

服务器端 #include "widget.h" #include "ui_widget.h"Widget::Widget(QWidget *parent): QWidget(parent), ui(new Ui::Widget) {ui->setupUi(this);//准备组件&#xff0c;初始化组件状态this->setFixedSize(800,600);chatwidget new QListWidge…

【Linux】Linux——Centos7安装Tomcat

1.下载Tomcat 安装包 官网地址&#xff1a;Apache Tomcat - Apache Tomcat 9 Software Downloadshttps://tomcat.apache.org/download-90.cgi 2.将下载的安装包上传到 Xftp 上&#xff0c;我是直接放到 usr 下了 3.将安装包解压到 /usr/local/ tar -zxvf apache-tomcat-9.0.8…

Java+SpringBoot+JSP实现在线心理评测与咨询系统

前言介绍 随着互联网技术的高速发展&#xff0c;人们生活的各方面都受到互联网技术的影响。现在人们可以通过互联网技术就能实现不出家门就可以通过网络进行系统管理&#xff0c;交易等&#xff0c;而且过程简单、快捷。同样的&#xff0c;在人们的工作生活中&#xff0c;也就…

249 基于matlab的MED、OMEDA、MOMEDA、MCKD信号处理方法

基于matlab的MED、OMEDA、MOMEDA、MCKD信号处理方法。最小熵反褶积(MED)&#xff0c;最优最小熵反卷积调整卷积 (OMEDA),多点最优最小熵解卷积调整&#xff08;Multipoint Optimal Minimum Entropy Deconvolution Adjusted&#xff0c;MOMEDA&#xff09;&#xff0c;最大相关峭…

Neuralink首个脑机接口患者:打游戏、搞研究两不误,重获自主能力

今年1月28日&#xff0c;Neuralink首次将侵入式脑机接口植入人类患者Noland Arbaugh的大脑。100天后&#xff0c;这家由埃隆马斯克创立的公司公布了最新的进展。从Neuralink的更新中我们可以看到&#xff0c;Arbaugh的恢复情况超出预期&#xff0c;他的用户体验也非常积极。 原…

Android 按钮Button点击音效

一、新建工程 编译运行&#xff0c;确保工程无误&#xff0c;这里不过多赘述。 二、UI布局 添加两个播放音效Button <?xml version"1.0" encoding"utf-8"?> <LinearLayout xmlns:android"http://schemas.android.com/apk/res/android"…

芋道系统springcloud模块启动报错,枚举类不能为空

问题描述&#xff1a; Error starting ApplicationContext. To display the conditions report re-run your application with debug enabled. 2024-05-10 15:50:15.756 | ERROR 9120 | main [TID: N/A] o.s.b.d.LoggingFailureAnalysisReporter | ************************…

腾讯云coding代码托管平台配置问题公钥拉取失败提示 Permission denied(publickey)

前言 最近在学校有个课设多人开发一个游戏&#xff0c;要团队协作&#xff0c;选用了腾讯云的coding作为代码管理仓库&#xff0c;但在配置的时候遇到了一些问题&#xff0c;相比于github&#xff0c;发现腾讯的coding更难用&#xff0c;&#xff0c;&#xff0c;这里记录一下…

【漫画版】指挥官的排序战术:快速排序算法解密

作者介绍&#xff1a;10年大厂数据\经营分析经验&#xff0c;现任字节跳动数据部门负责人。 会一些的技术&#xff1a;数据分析、算法、SQL、大数据相关、python&#xff0c;欢迎探讨交流 欢迎加入社区&#xff1a;码上找工作 作者专栏每日更新&#xff1a; LeetCode解锁1000题…

stm32单片机学习路线

第一步 编程及硬件基础知识 1.掌握C语言基础 作为STM32的主要编程语言&#xff0c;C语言的基础知识是必不可少的。建议通过书籍、在线课程或者教学视频系统地学习C语言的基础知识&#xff0c;包括语法、数据类型、函数、指针等。 2.了解电子电路基础 对于单片机开发来说&…

面向对象 03:类与对象的创建、初始化和使用,通过 new 关键字调用构造方法,以及创建对象过程的内存分析

一、前言 记录时间 [2024-05-10] 系列文章简摘&#xff1a; Java 笔记 01&#xff1a;Java 概述&#xff0c;MarkDown 常用语法整理 Java 笔记 11&#xff1a;Java 方法相关内容&#xff0c;方法的设计原则&#xff0c;以及方法的定义和调用 面向对象 01&#xff1a;Java 面向对…

ESP32引脚入门指南(四):从理论到实践(PWM)

引言 ESP32 作为物联网领域的明星微控制器&#xff0c;除了强大的Wi-Fi和蓝牙功能&#xff0c;还内置了丰富的外设资源&#xff0c;其中就包括高级的PWM&#xff08;脉冲宽度调制&#xff09;功能。本文将深入探讨ESP32的PWM引脚&#xff0c;解析其工作原理&#xff0c;并通过…