Merge pull request #65714 from resouer/fix-63784

pkg/scheduler/cache/cache.go

@@ -52,7 +52,7 @@ type schedulerCache struct {
 	period time.Duration
 
 	// This mutex guards all fields within this cache struct.
-	mu sync.Mutex
+	mu sync.RWMutex
 	// a set of assumed pod keys.
 	// The key could further be used to get an entry in podStates.
 	assumedPods map[string]bool
@@ -112,8 +112,8 @@ func newSchedulerCache(ttl, period time.Duration, stop <-chan struct{}) *schedul
 // Snapshot takes a snapshot of the current schedulerCache. The method has performance impact,
 // and should be only used in non-critical path.
 func (cache *schedulerCache) Snapshot() *Snapshot {
-	cache.mu.Lock()
-	defer cache.mu.Unlock()
+	cache.mu.RLock()
+	defer cache.mu.RUnlock()
 
 	nodes := make(map[string]*NodeInfo)
 	for k, v := range cache.nodes {
@@ -164,8 +164,8 @@ func (cache *schedulerCache) List(selector labels.Selector) ([]*v1.Pod, error) {
 }
 
 func (cache *schedulerCache) FilteredList(podFilter PodFilter, selector labels.Selector) ([]*v1.Pod, error) {
-	cache.mu.Lock()
-	defer cache.mu.Unlock()
+	cache.mu.RLock()
+	defer cache.mu.RUnlock()
 	// podFilter is expected to return true for most or all of the pods. We
 	// can avoid expensive array growth without wasting too much memory by
 	// pre-allocating capacity.
@@ -216,8 +216,8 @@ func (cache *schedulerCache) finishBinding(pod *v1.Pod, now time.Time) error {
 		return err
 	}
 
-	cache.mu.Lock()
-	defer cache.mu.Unlock()
+	cache.mu.RLock()
+	defer cache.mu.RUnlock()
 
 	glog.V(5).Infof("Finished binding for pod %v. Can be expired.", key)
 	currState, ok := cache.podStates[key]
@@ -387,8 +387,8 @@ func (cache *schedulerCache) IsAssumedPod(pod *v1.Pod) (bool, error) {
 		return false, err
 	}
 
-	cache.mu.Lock()
-	defer cache.mu.Unlock()
+	cache.mu.RLock()
+	defer cache.mu.RUnlock()
 
 	b, found := cache.assumedPods[key]
 	if !found {
@@ -403,8 +403,8 @@ func (cache *schedulerCache) GetPod(pod *v1.Pod) (*v1.Pod, error) {
 		return nil, err
 	}
 
-	cache.mu.Lock()
-	defer cache.mu.Unlock()
+	cache.mu.RLock()
+	defer cache.mu.RUnlock()
 
 	podState, ok := cache.podStates[key]
 	if !ok {
@@ -539,8 +539,8 @@ func (cache *schedulerCache) RemovePDB(pdb *policy.PodDisruptionBudget) error {
 }
 
 func (cache *schedulerCache) ListPDBs(selector labels.Selector) ([]*policy.PodDisruptionBudget, error) {
-	cache.mu.Lock()
-	defer cache.mu.Unlock()
+	cache.mu.RLock()
+	defer cache.mu.RUnlock()
 	var pdbs []*policy.PodDisruptionBudget
 	for _, pdb := range cache.pdbs {
 		if selector.Matches(labels.Set(pdb.Labels)) {
@@ -551,8 +551,8 @@ func (cache *schedulerCache) ListPDBs(selector labels.Selector) ([]*policy.PodDi
 }
 
 func (cache *schedulerCache) IsUpToDate(n *NodeInfo) bool {
-	cache.mu.Lock()
-	defer cache.mu.Unlock()
+	cache.mu.RLock()
+	defer cache.mu.RUnlock()
 	node, ok := cache.nodes[n.Node().Name]
 	return ok && n.generation == node.generation
 }

pkg/scheduler/core/equivalence/eqivalence_test.go

@@ -243,17 +243,21 @@ func TestRunPredicate(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			node := schedulercache.NewNodeInfo()
-			node.SetNode(&v1.Node{ObjectMeta: metav1.ObjectMeta{Name: "n1"}})
+			testNode := &v1.Node{ObjectMeta: metav1.ObjectMeta{Name: "n1"}}
+			node.SetNode(testNode)
 			pod := &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "p1"}}
 			meta := algorithm.EmptyPredicateMetadataProducer(nil, nil)
 
+			// Initialize and populate equivalence class cache.
 			ecache := NewCache()
+			nodeCache, _ := ecache.GetNodeCache(testNode.Name)
+
 			equivClass := NewClass(pod)
 			if test.expectCacheHit {
-				ecache.updateResult(pod.Name, "testPredicate", test.expectFit, test.expectedReasons, equivClass.hash, test.cache, node)
+				nodeCache.updateResult(pod.Name, "testPredicate", test.expectFit, test.expectedReasons, equivClass.hash, test.cache, node)
 			}
 
-			fit, reasons, err := ecache.RunPredicate(test.pred.predicate, "testPredicate", pod, meta, node, equivClass, test.cache)
+			fit, reasons, err := nodeCache.RunPredicate(test.pred.predicate, "testPredicate", pod, meta, node, equivClass, test.cache)
 
 			if err != nil {
 				if err.Error() != test.expectedError {
@@ -284,7 +288,7 @@ func TestRunPredicate(t *testing.T) {
 			if !test.expectCacheHit && test.pred.callCount == 0 {
 				t.Errorf("Predicate should be called")
 			}
-			_, ok := ecache.lookupResult(pod.Name, node.Node().Name, "testPredicate", equivClass.hash)
+			_, ok := nodeCache.lookupResult(pod.Name, node.Node().Name, "testPredicate", equivClass.hash)
 			if !ok && test.expectCacheWrite {
 				t.Errorf("Cache write should happen")
 			}
@@ -339,21 +343,25 @@ func TestUpdateResult(t *testing.T) {
 		},
 	}
 	for _, test := range tests {
+		node := schedulercache.NewNodeInfo()
+		testNode := &v1.Node{ObjectMeta: metav1.ObjectMeta{Name: test.nodeName}}
+		node.SetNode(testNode)
+
+		// Initialize and populate equivalence class cache.
 		ecache := NewCache()
+		nodeCache, _ := ecache.GetNodeCache(testNode.Name)
+
 		if test.expectPredicateMap {
-			ecache.cache[test.nodeName] = make(predicateMap)
 			predicateItem := predicateResult{
 				Fit: true,
 			}
-			ecache.cache[test.nodeName][test.predicateKey] =
+			nodeCache.cache[test.predicateKey] =
 				resultMap{
 					test.equivalenceHash: predicateItem,
 				}
 		}
 
-		node := schedulercache.NewNodeInfo()
-		node.SetNode(&v1.Node{ObjectMeta: metav1.ObjectMeta{Name: test.nodeName}})
-		ecache.updateResult(
+		nodeCache.updateResult(
 			test.pod,
 			test.predicateKey,
 			test.fit,
@@ -363,7 +371,7 @@ func TestUpdateResult(t *testing.T) {
 			node,
 		)
 
-		cachedMapItem, ok := ecache.cache[test.nodeName][test.predicateKey]
+		cachedMapItem, ok := nodeCache.cache[test.predicateKey]
 		if !ok {
 			t.Errorf("Failed: %s, can't find expected cache item: %v",
 				test.name, test.expectCacheItem)
@@ -473,11 +481,16 @@ func TestLookupResult(t *testing.T) {
 	}
 
 	for _, test := range tests {
+		testNode := &v1.Node{ObjectMeta: metav1.ObjectMeta{Name: test.nodeName}}
+
+		// Initialize and populate equivalence class cache.
 		ecache := NewCache()
+		nodeCache, _ := ecache.GetNodeCache(testNode.Name)
+
 		node := schedulercache.NewNodeInfo()
-		node.SetNode(&v1.Node{ObjectMeta: metav1.ObjectMeta{Name: test.nodeName}})
+		node.SetNode(testNode)
 		// set cached item to equivalence cache
-		ecache.updateResult(
+		nodeCache.updateResult(
 			test.podName,
 			test.predicateKey,
 			test.cachedItem.fit,
@@ -493,7 +506,7 @@ func TestLookupResult(t *testing.T) {
 			ecache.InvalidatePredicatesOnNode(test.nodeName, predicateKeys)
 		}
 		// calculate predicate with equivalence cache
-		result, ok := ecache.lookupResult(test.podName,
+		result, ok := nodeCache.lookupResult(test.podName,
 			test.nodeName,
 			test.predicateKey,
 			test.equivalenceHashForCalPredicate,
@@ -689,9 +702,12 @@ func TestInvalidateCachedPredicateItemOfAllNodes(t *testing.T) {
 
 	for _, test := range tests {
 		node := schedulercache.NewNodeInfo()
-		node.SetNode(&v1.Node{ObjectMeta: metav1.ObjectMeta{Name: test.nodeName}})
+		testNode := &v1.Node{ObjectMeta: metav1.ObjectMeta{Name: test.nodeName}}
+		node.SetNode(testNode)
+
+		nodeCache, _ := ecache.GetNodeCache(testNode.Name)
 		// set cached item to equivalence cache
-		ecache.updateResult(
+		nodeCache.updateResult(
 			test.podName,
 			testPredicate,
 			test.cachedItem.fit,
@@ -707,8 +723,8 @@ func TestInvalidateCachedPredicateItemOfAllNodes(t *testing.T) {
 
 	// there should be no cached predicate any more
 	for _, test := range tests {
-		if algorithmCache, exist := ecache.cache[test.nodeName]; exist {
-			if _, exist := algorithmCache[testPredicate]; exist {
+		if nodeCache, exist := ecache.nodeToCache[test.nodeName]; exist {
+			if _, exist := nodeCache.cache[testPredicate]; exist {
 				t.Errorf("Failed: cached item for predicate key: %v on node: %v should be invalidated",
 					testPredicate, test.nodeName)
 				break
@@ -761,9 +777,12 @@ func TestInvalidateAllCachedPredicateItemOfNode(t *testing.T) {
 
 	for _, test := range tests {
 		node := schedulercache.NewNodeInfo()
-		node.SetNode(&v1.Node{ObjectMeta: metav1.ObjectMeta{Name: test.nodeName}})
+		testNode := &v1.Node{ObjectMeta: metav1.ObjectMeta{Name: test.nodeName}}
+		node.SetNode(testNode)
+
+		nodeCache, _ := ecache.GetNodeCache(testNode.Name)
 		// set cached item to equivalence cache
-		ecache.updateResult(
+		nodeCache.updateResult(
 			test.podName,
 			testPredicate,
 			test.cachedItem.fit,
@@ -775,10 +794,10 @@ func TestInvalidateAllCachedPredicateItemOfNode(t *testing.T) {
 	}
 
 	for _, test := range tests {
-		// invalidate cached predicate for all nodes
+		// invalidate all cached predicate for node
 		ecache.InvalidateAllPredicatesOnNode(test.nodeName)
-		if _, exist := ecache.cache[test.nodeName]; exist {
-			t.Errorf("Failed: cached item for node: %v should be invalidated", test.nodeName)
+		if _, ok := ecache.GetNodeCache(test.nodeName); ok {
+			t.Errorf("Failed: node: %v should not be found in internal cache", test.nodeName)
 			break
 		}
 	}

pkg/scheduler/core/generic_scheduler.go

@@ -356,20 +356,25 @@ func (g *genericScheduler) findNodesThatFit(pod *v1.Pod, nodes []*v1.Node) ([]*v
 		meta := g.predicateMetaProducer(pod, g.cachedNodeInfoMap)
 
 		var equivClass *equivalence.Class
+
 		if g.equivalenceCache != nil {
 			// getEquivalenceClassInfo will return immediately if no equivalence pod found
 			equivClass = equivalence.NewClass(pod)
 		}
 
 		checkNode := func(i int) {
+			var nodeCache *equivalence.NodeCache
 			nodeName := nodes[i].Name
+			if g.equivalenceCache != nil {
+				nodeCache, _ = g.equivalenceCache.GetNodeCache(nodeName)
+			}
 			fits, failedPredicates, err := podFitsOnNode(
 				pod,
 				meta,
 				g.cachedNodeInfoMap[nodeName],
 				g.predicates,
 				g.cache,
-				g.equivalenceCache,
+				nodeCache,
 				g.schedulingQueue,
 				g.alwaysCheckAllPredicates,
 				equivClass,
@@ -472,7 +477,7 @@ func podFitsOnNode(
 	info *schedulercache.NodeInfo,
 	predicateFuncs map[string]algorithm.FitPredicate,
 	cache schedulercache.Cache,
-	ecache *equivalence.Cache,
+	nodeCache *equivalence.NodeCache,
 	queue SchedulingQueue,
 	alwaysCheckAllPredicates bool,
 	equivClass *equivalence.Class,
@@ -512,7 +517,7 @@ func podFitsOnNode(
 		// Bypass eCache if node has any nominated pods.
 		// TODO(bsalamat): consider using eCache and adding proper eCache invalidations
 		// when pods are nominated or their nominations change.
-		eCacheAvailable = equivClass != nil && !podsAdded
+		eCacheAvailable = equivClass != nil && nodeCache != nil && !podsAdded
 		for _, predicateKey := range predicates.Ordering() {
 			var (
 				fit     bool
@@ -522,7 +527,7 @@ func podFitsOnNode(
 			//TODO (yastij) : compute average predicate restrictiveness to export it as Prometheus metric
 			if predicate, exist := predicateFuncs[predicateKey]; exist {
 				if eCacheAvailable {
-					fit, reasons, err = ecache.RunPredicate(predicate, predicateKey, pod, metaToUse, nodeInfoToUse, equivClass, cache)
+					fit, reasons, err = nodeCache.RunPredicate(predicate, predicateKey, pod, metaToUse, nodeInfoToUse, equivClass, cache)
 				} else {
 					fit, reasons, err = predicate(pod, metaToUse, nodeInfoToUse)
 				}

pkg/scheduler/core/generic_scheduler_test.go

@@ -1405,7 +1405,8 @@ func TestCacheInvalidationRace(t *testing.T) {
 
 	// Set up the mock cache.
 	cache := schedulercache.New(time.Duration(0), wait.NeverStop)
-	cache.AddNode(&v1.Node{ObjectMeta: metav1.ObjectMeta{Name: "machine1"}})
+	testNode := &v1.Node{ObjectMeta: metav1.ObjectMeta{Name: "machine1"}}
+	cache.AddNode(testNode)
 	mockCache := &syncingMockCache{
 		Cache:            cache,
 		cycleStart:       make(chan struct{}),

pkg/scheduler/factory/factory.go

@@ -772,6 +772,11 @@ func (c *configFactory) addNodeToCache(obj interface{}) {
 		glog.Errorf("scheduler cache AddNode failed: %v", err)
 	}
 
+	if c.enableEquivalenceClassCache {
+		// GetNodeCache() will lazily create NodeCache for given node if it does not exist.
+		c.equivalencePodCache.GetNodeCache(node.GetName())
+	}
+
 	c.podQueue.MoveAllToActiveQueue()
 	// NOTE: add a new node does not affect existing predicates in equivalence cache
 }