/*
 Copyright 2019 The Kubernetes Authors.
 Copyright 2025 The Volcano Authors.

 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at

     http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.

This file includes code adapted from the Kubernetes scheduler cache
implementation (https://github.com/kubernetes/kubernetes/blob/master/pkg/scheduler/framework/types.go),
with modifications made by the Volcano Authors.
*/

package cache

import (
	"context"
	"fmt"
	"os"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	v1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/sets"
	"k8s.io/apimachinery/pkg/util/wait"
	utilfeature "k8s.io/apiserver/pkg/util/feature"
	"k8s.io/client-go/informers"
	infov1 "k8s.io/client-go/informers/core/v1"
	"k8s.io/client-go/kubernetes"
	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/record"
	"k8s.io/client-go/util/workqueue"
	resourceslicetracker "k8s.io/dynamic-resource-allocation/resourceslice/tracker"
	"k8s.io/klog/v2"
	fwk "k8s.io/kube-scheduler/framework"
	podutil "k8s.io/kubernetes/pkg/api/v1/pod"
	kubefeatures "k8s.io/kubernetes/pkg/features"
	k8sframework "k8s.io/kubernetes/pkg/scheduler/framework"
	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/dynamicresources"
	k8smetrics "k8s.io/kubernetes/pkg/scheduler/metrics"
	"k8s.io/kubernetes/pkg/scheduler/util/assumecache"

	vcclient "volcano.sh/apis/pkg/client/clientset/versioned"
	"volcano.sh/apis/pkg/client/clientset/versioned/scheme"
	vcinformer "volcano.sh/apis/pkg/client/informers/externalversions"
	shardinformerv1alpha1 "volcano.sh/apis/pkg/client/informers/externalversions/shard/v1alpha1"
	shardv1alpha1 "volcano.sh/apis/pkg/client/listers/shard/v1alpha1"
	"volcano.sh/volcano/cmd/agent-scheduler/app/options"
	agentapi "volcano.sh/volcano/pkg/agentscheduler/api"
	"volcano.sh/volcano/pkg/features"
	schedulingapi "volcano.sh/volcano/pkg/scheduler/api"
	"volcano.sh/volcano/pkg/scheduler/metrics"
	k8sutil "volcano.sh/volcano/pkg/scheduler/plugins/util/k8s"
	schedulercache "volcano.sh/volcano/pkg/schedulercommon/cache"
	"volcano.sh/volcano/pkg/util"
	k8sschedulingqueue "volcano.sh/volcano/third_party/kubernetes/pkg/scheduler/backend/queue"
)

const (
	handlerSyncPollPeriod = 100 * time.Millisecond
)

func init() {
	schemeBuilder := runtime.SchemeBuilder{
		v1.AddToScheme,
	}

	utilruntime.Must(schemeBuilder.AddToScheme(scheme.Scheme))
}

var _ Cache = &SchedulerCache{}

var generation int64

// New returns a Cache implementation.
func New(config *rest.Config, opt *options.ServerOption) Cache {
	return newSchedulerCache(config, opt)
}

//TODO: abstract common field and functions for volcano and agent scheduler, so each scheduler only depend on common part and customized part

// SchedulerCache cache for the kube batch
type SchedulerCache struct {
	Mutex       sync.RWMutex
	BinderMutex sync.RWMutex

	kubeClient kubernetes.Interface
	restConfig *rest.Config
	vcClient   vcclient.Interface
	// schedulerName is the name for volcano scheduler
	schedulerName      string
	nodeSelectorLabels map[string]sets.Empty
	metricsConf        map[string]string

	resyncPeriod      time.Duration
	podInformer       infov1.PodInformer
	nodeInformer      infov1.NodeInformer
	nodeShardInformer shardinformerv1alpha1.NodeShardInformer
	nodeShardLister   shardv1alpha1.NodeShardLister
	Binder            Binder
	StatusUpdater     StatusUpdater

	Recorder record.EventRecorder

	Nodes         map[string]*nodeInfoListItem // TODO: do we need to also add a separate lock for Nodes cache?
	headNode      *nodeInfoListItem
	NodeList      []string
	NodeShards    map[string]*schedulingapi.NodeShardInfo
	NodesInBinder map[string]int //Candidate nodes wait to be checked in binder

	taskCache *TaskCache

	nodeQueue               workqueue.TypedRateLimitingInterface[schedulercache.QueueObjectWrapper]
	nodeInitialEventTracker *schedulercache.InitialEventAsyncHandlerTracker

	informerFactory   informers.SharedInformerFactory
	vcInformerFactory vcinformer.SharedInformerFactory

	BindFlowChannel chan *agentapi.BindContext
	bindCache       []*agentapi.BindContext
	batchNum        int

	// A map from image name to its imageState.
	imageStates map[string]*imageState

	nodeWorkers uint32

	binderRegistry *BinderRegistry

	// sharedDRAManager is used in DRA plugin, contains resourceClaimTracker, resourceSliceLister and deviceClassLister
	sharedDRAManager fwk.SharedDRAManager

	// ConflictAwareBinder resolve confilct caused by multi workers parallel allocation
	ConflictAwareBinder *ConflictAwareBinder

	//ShardCoordinator keep sync of nodes that should be used based shards allocation, maintain the nodes in cache and in nodeshard cr
	ShardCoordinator *ShardCoordinator

	// schedulingQueue is used to store pods waiting to be scheduled
	schedulingQueue k8sschedulingqueue.SchedulingQueue

	// cancel is used to stop all goroutines started by scheduler cache,
	// currently is only needed to cancel the scheduling queues' metrics async recorder
	cancel context.CancelFunc

	shardingMode string

	registeredHandlers map[string]cache.ResourceEventHandlerRegistration

	// timeout on waiting for handlers handle initial resource synchronization before starting scheduling, 0 will skip waiting
	resourceSyncTimeout time.Duration
}

// TaskCache encapsulates the task map with a separate lock
type TaskCache struct {
	sync.RWMutex
	tasks map[schedulingapi.TaskID]*schedulingapi.TaskInfo
}

func NewTaskCache() *TaskCache {
	return &TaskCache{
		tasks: make(map[schedulingapi.TaskID]*schedulingapi.TaskInfo),
	}
}

func (tc *TaskCache) Get(id schedulingapi.TaskID) (*schedulingapi.TaskInfo, bool) {
	tc.RLock()
	defer tc.RUnlock()
	task, ok := tc.tasks[id]
	return task, ok
}

func (tc *TaskCache) Add(task *schedulingapi.TaskInfo) {
	tc.Lock()
	defer tc.Unlock()
	tc.tasks[task.UID] = task
}

func (tc *TaskCache) Update(task *schedulingapi.TaskInfo) {
	tc.Lock()
	defer tc.Unlock()
	tc.tasks[task.UID] = task
}

func (tc *TaskCache) Delete(id schedulingapi.TaskID) {
	tc.Lock()
	defer tc.Unlock()
	delete(tc.tasks, id)
}

type imageState struct {
	// Size of the image
	size int64
	// A set of node names for nodes having this image present
	nodes sets.Set[string]
}

// DefaultBinder with kube client and event recorder
type DefaultBinder struct {
	kubeclient kubernetes.Interface
	recorder   record.EventRecorder
}

// Bind will send bind request to api server
func (db *DefaultBinder) Bind(kubeClient kubernetes.Interface, tasks []*schedulingapi.TaskInfo) map[schedulingapi.TaskID]string {
	errMsg := make(map[schedulingapi.TaskID]string)
	for _, task := range tasks {
		p := task.Pod
		if err := db.kubeclient.CoreV1().Pods(p.Namespace).Bind(context.TODO(),
			&v1.Binding{
				ObjectMeta: metav1.ObjectMeta{Namespace: p.Namespace, Name: p.Name, UID: p.UID, Annotations: p.Annotations},
				Target: v1.ObjectReference{
					Kind: "Node",
					Name: task.NodeName,
				},
			},
			metav1.CreateOptions{}); err != nil {
			klog.Errorf("Failed to bind pod <%v/%v> to node %s : %#v", p.Namespace, p.Name, task.NodeName, err)
			errMsg[task.UID] = err.Error()
		}
	}

	return errMsg
}

// NewDefaultBinder create binder with kube client and event recorder, support fake binder if passed fake client and fake event recorder
func NewDefaultBinder(kbclient kubernetes.Interface, record record.EventRecorder) *DefaultBinder {
	return &DefaultBinder{
		kubeclient: kbclient,
		recorder:   record,
	}
}

// defaultStatusUpdater is the default implementation of the StatusUpdater interface
type defaultStatusUpdater struct {
	kubeclient kubernetes.Interface
	vcclient   vcclient.Interface
}

// following the same logic as podutil.UpdatePodCondition
func podConditionHaveUpdate(status *v1.PodStatus, condition *v1.PodCondition) bool {
	lastTransitionTime := metav1.Now()
	// Try to find this pod condition.
	_, oldCondition := podutil.GetPodCondition(status, condition.Type)

	if oldCondition == nil {
		// We are adding new pod condition.
		return true
	}
	// We are updating an existing condition, so we need to check if it has changed.
	if condition.Status == oldCondition.Status {
		lastTransitionTime = oldCondition.LastTransitionTime
	}

	isEqual := condition.Status == oldCondition.Status &&
		condition.Reason == oldCondition.Reason &&
		condition.Message == oldCondition.Message &&
		condition.LastProbeTime.Equal(&oldCondition.LastProbeTime) &&
		lastTransitionTime.Equal(&oldCondition.LastTransitionTime)

	// Return true if one of the fields have changed.
	return !isEqual
}

// UpdatePodStatus will Update pod status
func (su *defaultStatusUpdater) UpdatePodStatus(pod *v1.Pod) (*v1.Pod, error) {
	return su.kubeclient.CoreV1().Pods(pod.Namespace).UpdateStatus(context.TODO(), pod, metav1.UpdateOptions{})
}

// updateNodeSelectors parse and update node selector key value pairs to schedule cache
func (sc *SchedulerCache) updateNodeSelectors(nodeSelectors []string) {
	for _, nodeSelectorLabel := range nodeSelectors {
		nodeSelectorLabelLen := len(nodeSelectorLabel)
		if nodeSelectorLabelLen <= 0 {
			continue
		}
		// check input
		index := strings.Index(nodeSelectorLabel, ":")
		if index < 0 || index >= (nodeSelectorLabelLen-1) {
			continue
		}
		nodeSelectorLabelName := strings.TrimSpace(nodeSelectorLabel[:index])
		nodeSelectorLabelValue := strings.TrimSpace(nodeSelectorLabel[index+1:])
		key := nodeSelectorLabelName + ":" + nodeSelectorLabelValue
		sc.nodeSelectorLabels[key] = sets.Empty{}
	}
}

// setBatchBindParallel configure the parallel when binding tasks to apiserver
func (sc *SchedulerCache) setBatchBindParallel() {
	sc.BindFlowChannel = make(chan *agentapi.BindContext, 5000)
	var batchNum int
	batchNum, err := strconv.Atoi(os.Getenv("BATCH_BIND_NUM"))
	if err == nil && batchNum > 0 {
		sc.batchNum = batchNum
	} else {
		sc.batchNum = 1
	}
}

func newSchedulerCache(config *rest.Config, opt *options.ServerOption) *SchedulerCache {
	kubeClient, err := kubernetes.NewForConfig(config)
	if err != nil {
		panic(fmt.Sprintf("failed init kubeClient, with err: %v", err))
	}
	vcClient, err := vcclient.NewForConfig(config)
	if err != nil {
		panic(fmt.Sprintf("failed init vcClient, with err: %v", err))
	}
	eventClient, err := kubernetes.NewForConfig(config)
	if err != nil {
		panic(fmt.Sprintf("failed init eventClient, with err: %v", err))
	}

	sc := &SchedulerCache{
		Nodes:               make(map[string]*nodeInfoListItem),
		nodeQueue:           workqueue.NewTypedRateLimitingQueue(workqueue.DefaultTypedControllerRateLimiter[schedulercache.QueueObjectWrapper]()),
		kubeClient:          kubeClient,
		vcClient:            vcClient,
		restConfig:          config,
		schedulerName:       opt.SchedulerName,
		nodeSelectorLabels:  make(map[string]sets.Empty),
		imageStates:         make(map[string]*imageState),
		NodesInBinder:       make(map[string]int),
		NodeList:            []string{},
		NodeShards:          make(map[string]*schedulingapi.NodeShardInfo),
		nodeWorkers:         opt.NodeWorkerThreads,
		taskCache:           NewTaskCache(),
		shardingMode:        opt.ShardingMode,
		resourceSyncTimeout: opt.ResourceSyncTimeout,
	}

	sc.resyncPeriod = opt.ResyncPeriod

	if len(opt.NodeSelector) > 0 {
		sc.updateNodeSelectors(opt.NodeSelector)
	}
	// Prepare event clients.
	broadcaster := record.NewBroadcaster()
	broadcaster.StartRecordingToSink(&corev1.EventSinkImpl{Interface: eventClient.CoreV1().Events("")})
	sc.Recorder = broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: util.GenerateComponentName([]string{sc.schedulerName})})

	// set concurrency configuration when binding
	sc.setBatchBindParallel()
	if binder == nil {
		klog.V(3).Info("no registered bind method, new a default one")
		binder = NewDefaultBinder(sc.kubeClient, sc.Recorder)
	}
	sc.Binder = GetBindMethod()

	sc.StatusUpdater = &defaultStatusUpdater{
		kubeclient: sc.kubeClient,
		vcclient:   sc.vcClient,
	}

	sc.binderRegistry = NewBinderRegistry()

	// add all events handlers
	sc.addEventHandler()

	// init scheduling queue
	ctx, cancel := context.WithCancel(context.Background())
	metricsRecorder := k8smetrics.NewMetricsAsyncRecorder(1000, time.Second, ctx.Done())
	sc.cancel = cancel

	queueingHintMapPerProfile := make(k8sschedulingqueue.QueueingHintMapPerProfile)
	queueingHintMapPerProfile[sc.schedulerName] = buildQueueingHintMap()
	sc.schedulingQueue = k8sschedulingqueue.NewSchedulingQueue(
		Less,
		sc.informerFactory,
		k8sschedulingqueue.WithClock(defaultSchedulerOptions.clock),
		k8sschedulingqueue.WithPodInitialBackoffDuration(time.Duration(defaultSchedulerOptions.podInitialBackoffSeconds)*time.Second),
		k8sschedulingqueue.WithPodMaxBackoffDuration(time.Duration(defaultSchedulerOptions.podMaxBackoffSeconds)*time.Second),
		k8sschedulingqueue.WithPodMaxInUnschedulablePodsDuration(defaultSchedulerOptions.podMaxInUnschedulablePodsDuration),
		k8sschedulingqueue.WithMetricsRecorder(metricsRecorder),
		k8sschedulingqueue.WithQueueingHintMapPerProfile(queueingHintMapPerProfile),
	)

	sc.ConflictAwareBinder = NewConflictAwareBinder(sc, sc.schedulingQueue)
	if options.ServerOpts.ShardingMode != util.NoneShardingMode {
		sc.ShardCoordinator = NewShardCoordinator(sc, int(opt.ScheduleWorkerCount), opt.ShardName, opt.ShardingMode)
	}
	return sc
}

// defaultQueueingHintFn is the default queueing hint function.
// It always returns Queue as the queueing hint.
var defaultQueueingHintFn = func(_ klog.Logger, _ *v1.Pod, _, _ interface{}) (fwk.QueueingHint, error) {
	return fwk.Queue, nil
}

// buildQueueingHintMap builds the queueing hint map for the scheduling queue.
func buildQueueingHintMap() k8sschedulingqueue.QueueingHintMap {
	queueingHintMap := make(k8sschedulingqueue.QueueingHintMap)

	// Currently, we only register a wild card event with the default queueing hint function.
	// TODO: Support more specific events and queueing hint functions.
	wildCardEvent := fwk.ClusterEvent{Resource: fwk.WildCard, ActionType: fwk.All}
	queueingHintMap[wildCardEvent] = append(queueingHintMap[wildCardEvent], &k8sschedulingqueue.QueueingHintFunction{
		QueueingHintFn: defaultQueueingHintFn,
	})

	return queueingHintMap
}

func (sc *SchedulerCache) addEventHandler() {
	handlers := make(map[string]cache.ResourceEventHandlerRegistration, 10)
	var handlerRegistration cache.ResourceEventHandlerRegistration
	informerFactory := informers.NewSharedInformerFactory(sc.kubeClient, sc.resyncPeriod)
	sc.informerFactory = informerFactory

	// explicitly register informers to the factory, otherwise resources listers cannot get anything
	// even with no error returned.
	// `Namespace` informer is used by `InterPodAffinity` plugin,
	// `SelectorSpread` and `PodTopologySpread` plugins uses the following four so far.
	informerFactory.Core().V1().Namespaces().Informer()

	// `PodDisruptionBudgets` informer is used by `Pdb` plugin
	if utilfeature.DefaultFeatureGate.Enabled(features.PodDisruptionBudgetsSupport) {
		informerFactory.Policy().V1().PodDisruptionBudgets().Informer()
	}

	// Register storage releated informers
	informerFactory.Core().V1().PersistentVolumeClaims().Informer()
	informerFactory.Core().V1().PersistentVolumes().Informer()
	informerFactory.Storage().V1().StorageClasses().Informer()
	informerFactory.Storage().V1().CSINodes().Informer()
	if options.ServerOpts != nil && options.ServerOpts.EnableCSIStorage {
		informerFactory.Storage().V1().CSIDrivers().Informer()
		informerFactory.Storage().V1().CSIStorageCapacities().Informer()
	}

	// create informer for node information
	sc.nodeInformer = informerFactory.Core().V1().Nodes()
	handlerRegistration, _ = sc.nodeInformer.Informer().AddEventHandler(
		cache.FilteringResourceEventHandler{
			FilterFunc: func(obj interface{}) bool {
				switch t := obj.(type) {
				case *v1.Node:
					return true
				case cache.DeletedFinalStateUnknown:
					var ok bool
					_, ok = t.Obj.(*v1.Node)
					if !ok {
						klog.Errorf("Cannot convert to *v1.Node: %v", t.Obj)
						return false
					}
					return true
				default:
					return false
				}
			},
			Handler: cache.ResourceEventHandlerDetailedFuncs{
				AddFunc:    sc.AddNode,
				UpdateFunc: sc.UpdateNode,
				DeleteFunc: sc.DeleteNode,
			},
		},
	)
	//real node sync is handled in queue instead of event handler, use tracker to track the handling status in node queue
	sc.nodeInitialEventTracker = schedulercache.NewQueueHandlerTracker(handlerRegistration)
	handlers["node"] = schedulercache.NewInitialEventHandlerRegistration(handlerRegistration, sc.nodeInitialEventTracker)

	sc.podInformer = informerFactory.Core().V1().Pods()
	// 1. Pods already scheduled, refresh its state in cache
	handlerRegistration, _ = sc.podInformer.Informer().AddEventHandler(
		cache.FilteringResourceEventHandler{
			FilterFunc: func(obj interface{}) bool {
				switch v := obj.(type) {
				case *v1.Pod:
					if len(v.Spec.NodeName) != 0 {
						return true
					}
					return false
				case cache.DeletedFinalStateUnknown:
					if _, ok := v.Obj.(*v1.Pod); ok {
						// The carried object may be stale, always pass to clean up stale obj in event handlers.
						return true
					}
					klog.Errorf("Cannot convert object %T to *v1.Pod", v.Obj)
					return false
				default:
					klog.ErrorS(nil, "Unable to handle object", "objType", fmt.Sprintf("%T", obj), "obj", obj)
					return false
				}
			},
			Handler: cache.ResourceEventHandlerFuncs{
				AddFunc:    sc.AddPodToCache,
				UpdateFunc: sc.UpdatePodInCache,
				DeleteFunc: sc.DeletePodFromCache,
			},
		})
	handlers["pod-cache"] = handlerRegistration

	// 2. Pods not scheduled yet, and needed to be scheduled by agent scheduler, add them to scheduling queue
	handlerRegistration, _ = sc.podInformer.Informer().AddEventHandler(
		cache.FilteringResourceEventHandler{
			FilterFunc: func(obj interface{}) bool {
				switch v := obj.(type) {
				case *v1.Pod:
					// if the pod is not scheduled and scheduled by agent scheduler
					if len(v.Spec.NodeName) == 0 && sc.schedulerName == v.Spec.SchedulerName {
						return true
					}
					return false
				case cache.DeletedFinalStateUnknown:
					if _, ok := v.Obj.(*v1.Pod); ok {
						// The carried object may be stale, always pass to clean up stale obj in event handlers.
						return true
					}
					klog.Errorf("Cannot convert object %T to *v1.Pod", v.Obj)
					return false
				default:
					klog.ErrorS(nil, "Unable to handle object", "objType", fmt.Sprintf("%T", obj), "obj", obj)
					return false
				}
			},
			Handler: cache.ResourceEventHandlerFuncs{
				AddFunc:    sc.AddPodToSchedulingQueue,
				UpdateFunc: sc.UpdatePodInSchedulingQueue,
				DeleteFunc: sc.DeletePodFromSchedulingQueue,
			},
		})
	handlers["pod-queue"] = handlerRegistration

	vcinformers := vcinformer.NewSharedInformerFactory(sc.vcClient, sc.resyncPeriod)
	sc.vcInformerFactory = vcinformers
	if sc.shardingMode == util.HardShardingMode || sc.shardingMode == util.SoftShardingMode {
		sc.nodeShardInformer = sc.vcInformerFactory.Shard().V1alpha1().NodeShards()
		handlerRegistration, _ = sc.nodeShardInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
			AddFunc:    sc.AddNodeShard,
			UpdateFunc: sc.UpdateNodeShard,
			DeleteFunc: sc.DeleteNodeShard,
		})
		sc.nodeShardLister = sc.vcInformerFactory.Shard().V1alpha1().NodeShards().Lister()
		handlers["nodeShard"] = handlerRegistration
	}

	if utilfeature.DefaultFeatureGate.Enabled(kubefeatures.DynamicResourceAllocation) {
		ctx := context.TODO()
		logger := klog.FromContext(ctx)
		resourceClaimInformer := informerFactory.Resource().V1().ResourceClaims().Informer()
		resourceClaimCache := assumecache.NewAssumeCache(logger, resourceClaimInformer, "ResourceClaim", "", nil)
		resourceSliceTrackerOpts := resourceslicetracker.Options{
			EnableDeviceTaintRules: utilfeature.DefaultFeatureGate.Enabled(kubefeatures.DRADeviceTaintRules),
			SliceInformer:          informerFactory.Resource().V1().ResourceSlices(),
			KubeClient:             sc.kubeClient,
		}
		// If device taints are disabled, the additional informers are not needed and
		// the tracker turns into a simple wrapper around the slice informer.
		if resourceSliceTrackerOpts.EnableDeviceTaintRules {
			resourceSliceTrackerOpts.TaintInformer = informerFactory.Resource().V1beta2().DeviceTaintRules()
			resourceSliceTrackerOpts.ClassInformer = informerFactory.Resource().V1().DeviceClasses()
		}
		resourceSliceTracker, err := resourceslicetracker.StartTracker(ctx, resourceSliceTrackerOpts)
		if err != nil {
			klog.V(3).Infof("couldn't start resource slice tracker: %v", err)
		}
		sc.sharedDRAManager = dynamicresources.NewDRAManager(ctx, resourceClaimCache, resourceSliceTracker, informerFactory)
	}
	sc.registeredHandlers = handlers
}

// Run  starts the schedulerCache
func (sc *SchedulerCache) Run(stopCh <-chan struct{}) {
	sc.informerFactory.Start(stopCh)
	sc.vcInformerFactory.Start(stopCh)
	sc.WaitForCacheSync(stopCh)
	for i := 0; i < int(sc.nodeWorkers); i++ {
		go wait.Until(sc.runNodeWorker, 0, stopCh)
	}

	go wait.Until(sc.processBindTask, time.Millisecond*20, stopCh)

	ctx := context.TODO()
	logger := klog.FromContext(ctx)
	sc.schedulingQueue.Run(logger)
	if options.ServerOpts.ShardingMode != util.NoneShardingMode {
		sc.ShardCoordinator.Run(stopCh)
	}
	sc.ConflictAwareBinder.Run(stopCh)

	go func() {
		<-stopCh
		sc.cancel() // cancel other goroutines such as metricsRecorder
		sc.schedulingQueue.Close()
	}()
	if sc.resourceSyncTimeout > 0 {
		klog.V(3).Info("scheduler wait for handlers sync")
		sc.WaitForHandlerSync(stopCh)
		klog.V(3).Info("scheduler finished handlers sync")
	} else {
		klog.V(3).Info("skip waiting for handlers sync")
	}
}

// WaitForCacheSync sync the cache with the api server
func (sc *SchedulerCache) WaitForCacheSync(stopCh <-chan struct{}) {
	sc.informerFactory.WaitForCacheSync(stopCh)
	sc.vcInformerFactory.WaitForCacheSync(stopCh)
}

func (sc *SchedulerCache) WaitForHandlerSync(stopCh <-chan struct{}) {
	err := wait.PollUntilContextTimeout(wait.ContextForChannel(stopCh), handlerSyncPollPeriod, sc.resourceSyncTimeout, true, func(c context.Context) (done bool, err error) {
		for _, handler := range sc.registeredHandlers {
			if !handler.HasSynced() {
				return false, nil
			}
		}
		return true, nil
	})
	if err != nil {
		klog.Errorf("Error in waiting event handlers syncing: %v", err)
		for name, handler := range sc.registeredHandlers {
			if !handler.HasSynced() {
				klog.Errorf("%s handler synchronization is not completed", name)
			}
		}
	}
}

// Bind binds task to the target host.
func (sc *SchedulerCache) Bind(ctx context.Context, bindContexts []*agentapi.BindContext, preBinders map[string]PreBinder) {
	readyToBindTasks := make([]*schedulingapi.TaskInfo, len(bindContexts))
	for index := range readyToBindTasks {
		readyToBindTasks[index] = bindContexts[index].SchedCtx.Task
	}
	tmp := time.Now()
	errMsg := sc.Binder.Bind(sc.kubeClient, readyToBindTasks)
	if len(errMsg) == 0 {
		klog.V(3).Infof("bind ok, latency %v", time.Since(tmp))
	} else {
		klog.V(3).Infof("There are %d tasks in total and %d binds failed, latency %v", len(readyToBindTasks), len(errMsg), time.Since(tmp))
	}

	for _, bindContext := range bindContexts {
		task := bindContext.SchedCtx.Task
		if reason, ok := errMsg[task.UID]; !ok {
			sc.Recorder.Eventf(task.Pod, v1.EventTypeNormal, "Scheduled", "Successfully assigned %v/%v to %v", task.Namespace, task.Name, task.NodeName)
			metrics.UpdateTaskScheduleDuration(metrics.TaskStageBound, metrics.Duration(task.Pod.CreationTimestamp.Time))
		} else {
			unschedulableMsg := fmt.Sprintf("failed to bind to node %s: %s", task.NodeName, reason)
			if err := sc.TaskUnschedulable(task, schedulingapi.PodReasonSchedulerError, unschedulableMsg); err != nil {
				klog.ErrorS(err, "Failed to update pod status when bind task error", "task", task.Name)
			}

			for _, preBinder := range preBinders {
				if preBinder != nil {
					preBinder.PreBindRollBack(ctx, bindContext)
				}
			}

			klog.V(2).Infof("resyncTask task %s/%s", task.Namespace, task.Name)
			sc.resyncTask(bindContext.SchedCtx)
		}
	}
}

// Client returns the kubernetes clientSet
func (sc *SchedulerCache) Client() kubernetes.Interface {
	return sc.kubeClient
}

// VCClient returns the volcano clientSet
func (sc *SchedulerCache) VCClient() vcclient.Interface {
	return sc.vcClient
}

// ClientConfig returns the rest config
func (sc *SchedulerCache) ClientConfig() *rest.Config {
	return sc.restConfig
}

// SharedInformerFactory returns the scheduler SharedInformerFactory
func (sc *SchedulerCache) SharedInformerFactory() informers.SharedInformerFactory {
	return sc.informerFactory
}

// VCSharedInformerFactory returns the scheduler VC SharedInformerFactory
func (sc *SchedulerCache) VCSharedInformerFactory() vcinformer.SharedInformerFactory {
	return sc.vcInformerFactory
}

// SchedulingQueue returns the scheduling queue instance in the cache
func (sc *SchedulerCache) SchedulingQueue() k8sschedulingqueue.SchedulingQueue {
	return sc.schedulingQueue
}

// SetSharedInformerFactory sets the scheduler SharedInformerFactory for unit test
func (sc *SchedulerCache) SetSharedInformerFactory(factory informers.SharedInformerFactory) {
	sc.informerFactory = factory
}

// UpdateSchedulerNumaInfo used to update scheduler node cache NumaSchedulerInfo
func (sc *SchedulerCache) UpdateSchedulerNumaInfo(AllocatedSets map[string]schedulingapi.ResNumaSets) error {
	sc.Mutex.Lock()
	defer sc.Mutex.Unlock()

	for nodeName, sets := range AllocatedSets {
		if _, found := sc.Nodes[nodeName]; !found {
			continue
		}

		numaInfo := sc.Nodes[nodeName].info.NumaSchedulerInfo
		if numaInfo == nil {
			continue
		}

		numaInfo.Allocate(sets)
	}
	return nil
}

// EventRecorder returns the Event Recorder
func (sc *SchedulerCache) EventRecorder() record.EventRecorder {
	return sc.Recorder
}

// taskUnschedulable updates pod status of pending task
func (sc *SchedulerCache) TaskUnschedulable(task *schedulingapi.TaskInfo, reason, message string) error {
	pod := task.Pod

	condition := &v1.PodCondition{
		Type:    v1.PodScheduled,
		Status:  v1.ConditionFalse,
		Reason:  reason, // Add more reasons in order to distinguish more specific scenario of pending tasks
		Message: message,
	}

	updateCond := podConditionHaveUpdate(&pod.Status, condition)

	if updateCond {
		pod = pod.DeepCopy()

		if updateCond && podutil.UpdatePodCondition(&pod.Status, condition) {
			klog.V(3).Infof("Updating pod condition for %s/%s to (%s==%s)", pod.Namespace, pod.Name, condition.Type, condition.Status)
		}

		// The reason field in 'Events' should be "FailedScheduling", there is not constants defined for this in
		// k8s core, so using the same string here.
		// The reason field in PodCondition can be "Unschedulable"
		sc.Recorder.Eventf(pod, v1.EventTypeWarning, "FailedScheduling", "%s", message)
		if _, err := sc.StatusUpdater.UpdatePodStatus(pod); err != nil {
			return err
		}
	} else {
		klog.V(4).Infof("task unscheduleable %s/%s, message: %s, skip by no condition update", pod.Namespace, pod.Name, message)
	}

	return nil
}

// GetTaskInfo retrieves a task by its ID.
func (sc *SchedulerCache) GetTaskInfo(taskID schedulingapi.TaskID) (*schedulingapi.TaskInfo, bool) {
	return sc.taskCache.Get(taskID)
}

// AddTaskInfo adds a new task in the cache
func (sc *SchedulerCache) AddTaskInfo(task *schedulingapi.TaskInfo) {
	sc.taskCache.Add(task)
}

// UpdateTaskInfo updates a task in the cache.
func (sc *SchedulerCache) UpdateTaskInfo(task *schedulingapi.TaskInfo) {
	sc.taskCache.Update(task)
}

// DeleteTaskInfo removes a task from the cache by its ID.
func (sc *SchedulerCache) DeleteTaskInfo(taskID schedulingapi.TaskID) {
	sc.taskCache.Delete(taskID)
}

// resyncTask handles binding failures (prebind/bind errors) by releasing resources and re-queuing the pod.
func (sc *SchedulerCache) resyncTask(schedCtx *agentapi.SchedulingContext) {
	task := schedCtx.Task
	// 1. Remove task from node to release the occupied resource
	// This ensures other pending tasks can use these resources
	sc.Mutex.Lock()
	node, ok := sc.Nodes[task.NodeName]
	if !ok {
		klog.Warningf("Node %s not found for task %s/%s during resync", task.NodeName, task.Namespace, task.Name)
	} else {
		node.info.RemoveTask(task)
	}
	sc.Mutex.Unlock()

	// 2. Wake up other pending tasks that might be able to schedule now
	sc.schedulingQueue.MoveAllToActiveOrBackoffQueue(klog.Background(), k8sframework.EventAssignedPodDelete, task.Pod, nil, func(pod *v1.Pod) bool {
		return task.Pod.UID != pod.UID
	},
	)

	// 3. Re-queue the task using AddUnschedulableIfNotPresent
	if err := sc.schedulingQueue.AddUnschedulableIfNotPresent(klog.Background(), schedCtx.QueuedPodInfo, sc.schedulingQueue.SchedulingCycle()); err != nil {
		klog.ErrorS(err, "Failed to re-queue task during resync", "task", klog.KRef(task.Namespace, task.Name))
	}
}

func (sc *SchedulerCache) runNodeWorker() {
	for sc.processSyncNode() {
	}
}

func (sc *SchedulerCache) processSyncNode() bool {
	nodeWrapper, shutdown := sc.nodeQueue.Get()
	if shutdown {
		return false
	}
	defer sc.nodeQueue.Done(nodeWrapper)

	klog.V(5).Infof("started sync node %s", nodeWrapper.Object)
	err := sc.SyncNode(nodeWrapper.Object)
	if nodeWrapper.IsInInitialList {
		sc.nodeInitialEventTracker.Done(nodeWrapper.Object)
	}
	if err == nil {
		sc.nodeQueue.Forget(nodeWrapper)
		return true
	}

	klog.Errorf("Failed to sync node <%s>, retry it.", nodeWrapper.Object)
	sc.nodeQueue.AddRateLimited(nodeWrapper)
	return true
}

// AddBindTask add task to be bind to a cache which consumes by go runtime
func (sc *SchedulerCache) AddBindTask(bindContext *agentapi.BindContext) error {
	task := bindContext.SchedCtx.Task

	klog.V(5).Infof("add bind task %v/%v", task.Namespace, task.Name)
	sc.Mutex.Lock()
	defer sc.Mutex.Unlock()

	node, found := sc.Nodes[task.NodeName]
	if !found {
		return fmt.Errorf("failed to bind Task %v to host %v, host does not exist",
			task.UID, task.NodeName)
	}

	originalStatus := task.Status
	if err := sc.UpdateTaskStatus(task, schedulingapi.Binding); err != nil {
		return err
	}

	// Add task to the node.
	if err := node.info.AddTask(task); err != nil {
		// After failing to update task to a node we need to revert task status from Releasing,
		// otherwise task might be stuck in the Releasing state indefinitely.
		if err := sc.UpdateTaskStatus(task, originalStatus); err != nil {
			klog.Errorf("Task <%s/%s> will be resynchronized after failing to revert status "+
				"from %s to %s after failing to update Task on Node <%s>: %v",
				task.Namespace, task.Name, task.Status, originalStatus, node.info.Name, err)
		}
		return err
	}
	// bind generation after task is added to node, so next allocation on this node in newer generation must aware of this task
	node.info.NextBindGeneration()
	sc.Nodes[node.info.Name].info.Generation = nextGeneration()
	sc.moveNodeToHead(node.info.Name)

	sc.BindFlowChannel <- bindContext

	return nil
}

func (sc *SchedulerCache) processBindTask() {
	for {
		select {
		case bindContext, ok := <-sc.BindFlowChannel:
			if !ok {
				return
			}

			sc.bindCache = append(sc.bindCache, bindContext)
			if len(sc.bindCache) == sc.batchNum {
				sc.BindTask()
			}
		default:
		}

		if len(sc.BindFlowChannel) == 0 {
			break
		}
	}

	if len(sc.bindCache) == 0 {
		return
	}
	sc.BindTask()
}

// executePreBind executes PreBind for one bindContext
func (sc *SchedulerCache) executePreBind(ctx context.Context, bindContext *agentapi.BindContext, preBinders map[string]PreBinder) error {
	executedPreBinders := make([]PreBinder, 0, len(preBinders))
	for _, preBinder := range preBinders {
		if preBinder == nil {
			continue
		}

		if err := preBinder.PreBind(ctx, bindContext); err != nil {
			// If PreBind fails, rollback the executed PreBinders
			for i := len(executedPreBinders) - 1; i >= 0; i-- {
				if executedPreBinders[i] != nil {
					executedPreBinders[i].PreBindRollBack(ctx, bindContext)
				}
			}
			return err
		}
		if bindContext.SchedCtx.Task != nil && bindContext.SchedCtx.Task.Pod != nil {
			metrics.UpdateTaskScheduleDuration(metrics.TaskStagePreBound, metrics.Duration(bindContext.SchedCtx.Task.Pod.CreationTimestamp.Time))
		}
		executedPreBinders = append(executedPreBinders, preBinder)
	}

	return nil
}

// executePreBinds executes PreBind for a list of bindContexts
func (sc *SchedulerCache) executePreBinds(ctx context.Context, bindContexts []*agentapi.BindContext, preBinders map[string]PreBinder) []*agentapi.BindContext {
	logger := klog.FromContext(ctx)
	successfulBindContexts := make([]*agentapi.BindContext, 0, len(bindContexts))

	for _, bindContext := range bindContexts {
		if err := sc.executePreBind(ctx, bindContext, preBinders); err != nil {
			task := bindContext.SchedCtx.Task
			reason := fmt.Sprintf("execute preBind for pod %s failed: %v, resync the task", klog.KObj(task.Pod), err)
			klog.Error(reason)
			sc.resyncTask(bindContext.SchedCtx)
			if updateErr := sc.TaskUnschedulable(task, schedulingapi.PodReasonSchedulerError, reason); updateErr != nil {
				logger.Error(updateErr, "Failed to update pod status", "pod", klog.KObj(task.Pod))
			}
			continue
		}
		successfulBindContexts = append(successfulBindContexts, bindContext)
	}

	return successfulBindContexts
}

// BindTask do k8s binding with a goroutine
func (sc *SchedulerCache) BindTask() {
	klog.V(5).Infof("batch bind task count %d", sc.batchNum)
	tmpBindCache := make([]*agentapi.BindContext, len(sc.bindCache))
	copy(tmpBindCache, sc.bindCache)

	// Currently, bindContexts only contain 1 element.
	go func(bindContexts []*agentapi.BindContext) {
		logger := klog.Background()
		ctx := klog.NewContext(context.Background(), logger)
		cancelCtx, cancel := context.WithCancel(ctx)
		defer cancel()

		preBinders := sc.binderRegistry.getRegisteredPreBinders()
		successfulPreBindContexts := sc.executePreBinds(cancelCtx, bindContexts, preBinders)
		sc.Bind(ctx, successfulPreBindContexts, preBinders)
	}(tmpBindCache)

	// The slice here needs to point to a new underlying array, otherwise bindCache may not be able to trigger garbage collection immediately
	// if it is not expanded, causing memory leaks.
	sc.bindCache = make([]*agentapi.BindContext, 0)
}

// Snapshot returns the complete snapshot of the cluster from cache, used for dump purpose only
func (sc *SchedulerCache) Snapshot() *schedulingapi.ClusterInfo {
	sc.Mutex.Lock()
	defer sc.Mutex.Unlock()

	snapshot := &schedulingapi.ClusterInfo{
		Nodes:        make(map[string]*schedulingapi.NodeInfo),
		RealNodesSet: make(map[string]sets.Set[string]),
		NodeList:     make([]string, len(sc.NodeList)),
	}

	// TODO add agent scheduler cache

	copy(snapshot.NodeList, sc.NodeList)
	for _, value := range sc.Nodes {
		value.info.RefreshNumaSchedulerInfoByCrd()
	}

	for _, value := range sc.Nodes {
		if !value.info.Ready() {
			continue
		}

		snapshot.Nodes[value.info.Name] = value.info.Clone()
	}
	klog.V(3).InfoS("SnapShot for scheduling", "NodeNum", len(snapshot.Nodes))
	return snapshot
}

func (sc *SchedulerCache) UpdateSnapshot(snapshot *k8sutil.Snapshot) error {
	sc.BinderMutex.RLock()
	snapshot.CloneNodesInBinder(sc.NodesInBinder)
	sc.BinderMutex.RUnlock()

	sc.Mutex.RLock()
	defer sc.Mutex.RUnlock()

	klog.V(5).Infof("begin to update the snapshot ...")
	klog.V(5).Infof("the snapshot is %v", snapshot)

	snapshotGeneration := snapshot.GetGeneration()
	// currentNodeNames record the names of nodes that exist in the cache for later deletion of nodes that do not exist in the snapshot.
	currentNodeNames := make(map[string]bool)
	var nodesToUpdate []*schedulingapi.NodeInfo

	for _, node := range sc.Nodes {
		currentNodeNames[node.info.Name] = true
	}

	// Traverse the doubly linked list starting from the head, only processing nodes whose Generation is greater than the previous snapshot's Generation.
	for node := sc.headNode; node != nil; node = node.next {
		if node.info.Generation <= snapshotGeneration {
			klog.V(2).Infof("Stop traversal at node %s, generation %d <= snapshot generation %d",
				node.info.Name, node.info.Generation, snapshotGeneration)
			break
		}
		klog.V(5).Infof("current node name need to update in cache is %s", node.info.Name)
		nodesToUpdate = append(nodesToUpdate, node.info)
	}

	if len(nodesToUpdate) > 0 {
		snapshot.AddOrUpdateNodes(nodesToUpdate)
	}

	if sc.headNode != nil {
		snapshot.SetGeneration(sc.headNode.info.Generation)
	}
	// Remove deleted nodes and rebuild node lists in place
	snapshot.RemoveDeletedNodesFromSnapshot(currentNodeNames)
	klog.V(2).Infof("Snapshot updated: generation=%d, total nodes num=%d, updated nodes num=%d",
		snapshot.GetGeneration(), len(snapshot.GetFwkNodeInfoMap()), len(nodesToUpdate))
	return nil
}

func (sc *SchedulerCache) SharedDRAManager() fwk.SharedDRAManager {
	return sc.sharedDRAManager
}

// String returns information about the cache in a string format
func (sc *SchedulerCache) String() string {
	sc.Mutex.RLock()
	defer sc.Mutex.RUnlock()

	str := "Cache:\n"

	if len(sc.Nodes) != 0 {
		str += "Nodes:\n"
		for _, n := range sc.Nodes {
			str += fmt.Sprintf("\t %s: idle(%v) used(%v) allocatable(%v) pods(%d)\n",
				n.info.Name, n.info.Idle, n.info.Used, n.info.Allocatable, len(n.info.Tasks))

			i := 0
			for _, p := range n.info.Tasks {
				str += fmt.Sprintf("\t\t %d: %v\n", i, p)
				i++
			}
		}
	}

	if len(sc.NodeList) != 0 {
		str += fmt.Sprintf("NodeList: %v\n", sc.NodeList)
	}

	return str
}

func (sc *SchedulerCache) SetMetricsConf(conf map[string]string) {
	sc.metricsConf = conf
}

// createImageStateSummary returns a summarizing snapshot of the given image's state.
func (sc *SchedulerCache) createImageStateSummary(state *imageState) *fwk.ImageStateSummary {
	return &fwk.ImageStateSummary{
		Size:     state.size,
		NumNodes: len(state.nodes),
	}
}

func (sc *SchedulerCache) RegisterBinder(name string, binder interface{}) {
	if sc.binderRegistry == nil {
		sc.binderRegistry = NewBinderRegistry()
	}
	sc.binderRegistry.Register(name, binder)
}

// UpdateTaskStatus TODO: refer to update task status
func (sc *SchedulerCache) UpdateTaskStatus(task *schedulingapi.TaskInfo, status schedulingapi.TaskStatus) error {
	task.Status = status
	return nil
}

func (sc *SchedulerCache) EnqueueScheduleResult(scheduleResult *agentapi.PodScheduleResult) {
	sc.ConflictAwareBinder.EnqueueScheduleResult(scheduleResult)
}

func (sc *SchedulerCache) RecordCandidateNodesInBinder(nodes []*schedulingapi.NodeInfo) {
	sc.BinderMutex.Lock()
	defer sc.BinderMutex.Unlock()
	for _, nodeInfo := range nodes {
		if nodeInfo == nil {
			continue
		}
		sc.NodesInBinder[nodeInfo.Name] += 1
	}
}

func (sc *SchedulerCache) RemoveCandidateNodesFromBinder(nodes []*schedulingapi.NodeInfo) {
	sc.BinderMutex.Lock()
	defer sc.BinderMutex.Unlock()
	for _, nodeInfo := range nodes {
		if nodeInfo == nil {
			continue
		}
		sc.NodesInBinder[nodeInfo.Name] -= 1
		if sc.NodesInBinder[nodeInfo.Name] <= 0 {
			delete(sc.NodesInBinder, nodeInfo.Name)
		}
	}
}

// nodeInfoListItem holds a NodeInfo pointer and acts as an item in a doubly
// linked list. When a NodeInfo is updated, it goes to the head of the list.
// The items closer to the head are the most recently updated items.
type nodeInfoListItem struct {
	info *schedulingapi.NodeInfo
	next *nodeInfoListItem
	prev *nodeInfoListItem
}

func nextGeneration() int64 {
	return atomic.AddInt64(&generation, 1)
}

// moveNodeInfoToHead moves a NodeInfo to the head of "cache.nodes" doubly
// linked list. The head is the most recently updated NodeInfo.
// We assume cache lock is already acquired.
func (sc *SchedulerCache) moveNodeToHead(name string) {
	ni, ok := sc.Nodes[name]
	if !ok {
		klog.Errorf("No node info with given node name <%s> found in the cache", name)
		return
	}
	// if the node info list item is already at the head, we are done.
	if ni == sc.headNode {
		return
	}

	if ni.prev != nil {
		ni.prev.next = ni.next
	}
	if ni.next != nil {
		ni.next.prev = ni.prev
	}
	if sc.headNode != nil {
		sc.headNode.prev = ni
	}
	ni.next = sc.headNode
	ni.prev = nil
	sc.headNode = ni
}

// newNodeInfoListItem initializes a new nodeInfoListItem.
func newNodeInfoListItem(node *schedulingapi.NodeInfo) *nodeInfoListItem {
	return &nodeInfoListItem{
		info: node,
	}
}

func (sc *SchedulerCache) UpdateNodeShardStatus(shardName string, usedNodeInCache sets.Set[string]) error {
	klog.V(3).Infof("Update NodeShard %s status...", shardName)
	nodeShard, err := sc.nodeShardLister.Get(shardName)
	if err != nil {
		// Check if the error happens because the HyperNode is deleted
		if errors.IsNotFound(err) {
			klog.Infof("NodeShard %s has been deleted, no status update needed", shardName)
			return nil
		}
		klog.Error(err, "Failed to get NodeShard", "name", shardName)
		return err
	}

	oldNodesInUse := sets.New(nodeShard.Status.NodesInUse...)
	oldNodesToRemove := sets.New(nodeShard.Status.NodesToRemove...)
	oldNodesToAdd := sets.New(nodeShard.Status.NodesToAdd...)

	// Create a deep copy to avoid modifying cache objects
	nodeShardCopy := nodeShard.DeepCopy()
	desiredNodes := sets.New(nodeShard.Spec.NodesDesired...)

	nodesToRemove := usedNodeInCache.Difference(desiredNodes)
	nodesToAdd := desiredNodes.Difference(usedNodeInCache)
	if usedNodeInCache.Equal(oldNodesInUse) && nodesToRemove.Equal(oldNodesToRemove) && nodesToAdd.Equal(oldNodesToAdd) {
		klog.V(3).Infof("Skip update NodeShard %s status, no change", nodeShard.Name)
		return nil
	}

	nodeShardCopy.Status.NodesInUse = usedNodeInCache.UnsortedList()
	nodeShardCopy.Status.NodesToRemove = nodesToRemove.UnsortedList()
	nodeShardCopy.Status.NodesToAdd = nodesToAdd.UnsortedList()

	_, err = sc.vcClient.ShardV1alpha1().NodeShards().UpdateStatus(context.Background(), nodeShardCopy, metav1.UpdateOptions{})
	if err != nil {
		klog.Errorf("Failed to update NodeShard %s status %v", shardName, err)
		return err
	}
	klog.V(3).Infof("Updated NodeShard %s status", shardName)
	return nil
}

// OnWorkerStartSchedulingCycle is called when scheduler worker start a new scheduling cycle
func (sc *SchedulerCache) OnWorkerStartSchedulingCycle(index int, schedCtx *agentapi.SchedulingContext) {
	if options.ServerOpts.ShardingMode != util.NoneShardingMode {
		sc.ShardCoordinator.OnWorkerStartSchedulingCycle(index, schedCtx)
	}
}

// OnWorkerEndSchedulingCycle is called when scheduler worker end a scheduling cycle
func (sc *SchedulerCache) OnWorkerEndSchedulingCycle(index int) {
	if options.ServerOpts.ShardingMode != util.NoneShardingMode {
		sc.ShardCoordinator.OnWorkerEndSchedulingCycle(index)
	}
}
