scheduler: remove deepCopy of reservationInfo

pkg/scheduler/frameworkext/reservation_info.go

@@ -17,6 +17,8 @@ limitations under the License.
 package frameworkext
 
 import (
+	"sync/atomic"
+
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -34,6 +36,14 @@ import (
 	reservationutil "github.com/koordinator-sh/koordinator/pkg/util/reservation"
 )
 
+var generation int64
+
+// nextGeneration: Let's make sure history never forgets the name...
+// Increments the generation number monotonically ensuring that generation numbers never collide.
+func nextGeneration() int64 {
+	return atomic.AddInt64(&generation, 1)
+}
+
 type ReservationInfo struct {
 	Reservation      *schedulingv1alpha1.Reservation
 	Pod              *corev1.Pod
@@ -45,6 +55,7 @@ type ReservationInfo struct {
 	AssignedPods     map[types.UID]*PodRequirement
 	OwnerMatchers    []reservationutil.ReservationOwnerMatcher
 	ParseError       error
+	Generation       int64
 }
 
 type PodRequirement struct {
@@ -111,6 +122,7 @@ func NewReservationInfo(r *schedulingv1alpha1.Reservation) *ReservationInfo {
 		AssignedPods:     map[types.UID]*PodRequirement{},
 		OwnerMatchers:    ownerMatchers,
 		ParseError:       parseError,
+		Generation:       nextGeneration(),
 	}
 }
 
@@ -152,6 +164,7 @@ func NewReservationInfoFromPod(pod *corev1.Pod) *ReservationInfo {
 		AssignedPods:     map[types.UID]*PodRequirement{},
 		OwnerMatchers:    ownerMatchers,
 		ParseError:       parseError,
+		Generation:       nextGeneration(),
 	}
 }
 
@@ -304,6 +317,9 @@ func (ri *ReservationInfo) Clone() *ReservationInfo {
 		AllocatablePorts: util.CloneHostPorts(ri.AllocatablePorts),
 		AllocatedPorts:   util.CloneHostPorts(ri.AllocatedPorts),
 		AssignedPods:     assignedPods,
+		OwnerMatchers:    ri.OwnerMatchers,
+		ParseError:       ri.ParseError,
+		Generation:       ri.Generation,
 	}
 }
 
@@ -337,6 +353,7 @@ func (ri *ReservationInfo) UpdateReservation(r *schedulingv1alpha1.Reservation)
 		parseError = utilerrors.NewAggregate(parseErrors)
 	}
 	ri.ParseError = parseError
+	ri.Generation = nextGeneration()
 }
 
 func (ri *ReservationInfo) UpdatePod(pod *corev1.Pod) {
@@ -374,6 +391,7 @@ func (ri *ReservationInfo) UpdatePod(pod *corev1.Pod) {
 		parseError = utilerrors.NewAggregate(parseErrors)
 	}
 	ri.ParseError = parseError
+	ri.Generation = nextGeneration()
 }
 
 func (ri *ReservationInfo) AddAssignedPod(pod *corev1.Pod) {
@@ -385,6 +403,7 @@ func (ri *ReservationInfo) AddAssignedPod(pod *corev1.Pod) {
 	ri.Allocated = quotav1.Add(ri.Allocated, quotav1.Mask(requirement.Requests, ri.ResourceNames))
 	ri.AllocatedPorts = util.AppendHostPorts(ri.AllocatedPorts, requirement.Ports)
 	ri.AssignedPods[pod.UID] = requirement
+	ri.Generation = nextGeneration()
 }
 
 func (ri *ReservationInfo) RemoveAssignedPod(pod *corev1.Pod) {
@@ -398,4 +417,5 @@ func (ri *ReservationInfo) RemoveAssignedPod(pod *corev1.Pod) {
 
 		delete(ri.AssignedPods, pod.UID)
 	}
+	ri.Generation = nextGeneration()
 }

pkg/scheduler/frameworkext/reservation_info_test.go

@@ -500,6 +500,7 @@ func TestReservationInfoUpdateReservation(t *testing.T) {
 			})
 			tt.want.Reservation = tt.newReservation
 			tt.want.Pod = reservationutil.NewReservePod(tt.newReservation)
+			reservationInfo.Generation = 0
 			assert.Equal(t, tt.want, reservationInfo)
 		})
 	}
@@ -645,6 +646,7 @@ func TestReservationInfoUpdatePod(t *testing.T) {
 				return tt.want.ResourceNames[i] < tt.want.ResourceNames[j]
 			})
 			tt.want.Pod = tt.newPod
+			reservationInfo.Generation = 0
 			assert.Equal(t, tt.want, reservationInfo)
 		})
 	}

pkg/scheduler/plugins/reservation/cache.go

@@ -24,6 +24,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/klog/v2"
 	"k8s.io/kubernetes/pkg/scheduler/framework"
 
 	schedulingv1alpha1 "github.com/koordinator-sh/koordinator/apis/scheduling/v1alpha1"
@@ -38,6 +39,7 @@ var theReservationCache atomic.Value
 
 type ReservationCache interface {
 	DeleteReservation(r *schedulingv1alpha1.Reservation) *frameworkext.ReservationInfo
+	// this reservation info is only
 	GetReservationInfoByPod(pod *corev1.Pod, nodeName string) *frameworkext.ReservationInfo
 }
 
@@ -54,17 +56,81 @@ func SetReservationCache(cache ReservationCache) {
 }
 
 type reservationCache struct {
-	reservationLister  schedulinglister.ReservationLister
-	lock               sync.RWMutex
-	reservationInfos   map[types.UID]*frameworkext.ReservationInfo
-	reservationsOnNode map[string]map[types.UID]struct{}
+	reservationLister   schedulinglister.ReservationLister
+	lock                sync.RWMutex
+	reservationsOnNode  map[string]map[types.UID]struct{}
+	reservationInfos    map[types.UID]*reservationInfoListItem
+	headReservationInfo *reservationInfoListItem
+	snapshot            *Snapshot
+}
+
+type reservationInfoListItem struct {
+	info *frameworkext.ReservationInfo
+	next *reservationInfoListItem
+	prev *reservationInfoListItem
+}
+
+// newNodeInfoListItem initializes a new nodeInfoListItem.
+func newReservationInfoListItem(rInfo *frameworkext.ReservationInfo) *reservationInfoListItem {
+	return &reservationInfoListItem{
+		info: rInfo,
+	}
+}
+
+func (cache *reservationCache) moveReservationInfoToHead(uid types.UID) {
+	rInfoItem, ok := cache.reservationInfos[uid]
+	if !ok {
+		klog.ErrorS(nil, "No  info with given name found in the cache", "node", klog.KRef("", string(uid)))
+		return
+	}
+	// if the node info list item is already at the head, we are done.
+	if rInfoItem == cache.headReservationInfo {
+		return
+	}
+
+	if rInfoItem.prev != nil {
+		rInfoItem.prev.next = rInfoItem.next
+	}
+	if rInfoItem.next != nil {
+		rInfoItem.next.prev = rInfoItem.prev
+	}
+	if cache.headReservationInfo != nil {
+		cache.headReservationInfo.prev = rInfoItem
+	}
+	rInfoItem.next = cache.headReservationInfo
+	rInfoItem.prev = nil
+	cache.headReservationInfo = rInfoItem
+}
+
+// removeReservationInfoFromList removes a NodeInfo from the "cache.nodes" doubly
+// linked list.
+// We assume cache lock is already acquired.
+func (cache *reservationCache) removeReservationInfoFromList(uid types.UID) {
+	rInfo, ok := cache.reservationInfos[uid]
+	if !ok {
+		klog.ErrorS(nil, "No node info with given name found in the cache", "node", klog.KRef("", string(uid)))
+		return
+	}
+
+	if rInfo.prev != nil {
+		rInfo.prev.next = rInfo.next
+	}
+	if rInfo.next != nil {
+		rInfo.next.prev = rInfo.prev
+	}
+	// if the removed item was at the head, we must update the head.
+	if rInfo == cache.headReservationInfo {
+		cache.headReservationInfo = rInfo.next
+	}
+	delete(cache.reservationInfos, uid)
 }
 
 func newReservationCache(reservationLister schedulinglister.ReservationLister) *reservationCache {
 	cache := &reservationCache{
 		reservationLister:  reservationLister,
-		reservationInfos:   map[types.UID]*frameworkext.ReservationInfo{},
+		reservationInfos:   map[types.UID]*reservationInfoListItem{},
 		reservationsOnNode: map[string]map[types.UID]struct{}{},
+		snapshot:           NewEmptySnapshot(),
 	}
 	return cache
 }
@@ -106,11 +172,12 @@ func (cache *reservationCache) updateReservation(newR *schedulingv1alpha1.Reserv
 	defer cache.lock.Unlock()
 	rInfo := cache.reservationInfos[newR.UID]
 	if rInfo == nil {
-		rInfo = frameworkext.NewReservationInfo(newR)
+		rInfo = newReservationInfoListItem(frameworkext.NewReservationInfo(newR))
 		cache.reservationInfos[newR.UID] = rInfo
 	} else {
-		rInfo.UpdateReservation(newR)
+		rInfo.info.UpdateReservation(newR)
 	}
+	cache.moveReservationInfoToHead(newR.UID)
 	if newR.Status.NodeName != "" {
 		cache.updateReservationsOnNode(newR.Status.NodeName, newR.UID)
 	}
@@ -121,7 +188,8 @@ func (cache *reservationCache) updateReservationIfExists(newR *schedulingv1alpha
 	defer cache.lock.Unlock()
 	rInfo := cache.reservationInfos[newR.UID]
 	if rInfo != nil {
-		rInfo.UpdateReservation(newR)
+		rInfo.info.UpdateReservation(newR)
+		cache.moveReservationInfoToHead(newR.UID)
 		if newR.Status.NodeName != "" {
 			cache.updateReservationsOnNode(newR.Status.NodeName, newR.UID)
 		}
@@ -133,25 +201,26 @@ func (cache *reservationCache) DeleteReservation(r *schedulingv1alpha1.Reservati
 	defer cache.lock.Unlock()
 	rInfo := cache.reservationInfos[r.UID]
 	delete(cache.reservationInfos, r.UID)
+	cache.removeReservationInfoFromList(r.UID)
 	cache.deleteReservationOnNode(r.Status.NodeName, r.UID)
-	return rInfo
+	return rInfo.info
 }
 
 func (cache *reservationCache) updateReservationOperatingPod(newPod *corev1.Pod, currentOwner *corev1.ObjectReference) {
 	cache.lock.Lock()
 	defer cache.lock.Unlock()
 	rInfo := cache.reservationInfos[newPod.UID]
 	if rInfo == nil {
-		rInfo = frameworkext.NewReservationInfoFromPod(newPod)
+		rInfo = newReservationInfoListItem(frameworkext.NewReservationInfoFromPod(newPod))
 		cache.reservationInfos[newPod.UID] = rInfo
 	} else {
-		rInfo.UpdatePod(newPod)
+		rInfo.info.UpdatePod(newPod)
 	}
 	if newPod.Spec.NodeName != "" {
 		cache.updateReservationsOnNode(newPod.Spec.NodeName, newPod.UID)
 	}
 	if currentOwner != nil {
-		rInfo.AddAssignedPod(&corev1.Pod{
+		rInfo.info.AddAssignedPod(&corev1.Pod{
 			ObjectMeta: metav1.ObjectMeta{
 				Name:      currentOwner.Name,
 				Namespace: currentOwner.Namespace,
@@ -165,6 +234,7 @@ func (cache *reservationCache) deleteReservationOperatingPod(pod *corev1.Pod) {
 	cache.lock.Lock()
 	defer cache.lock.Unlock()
 	delete(cache.reservationInfos, pod.UID)
+	cache.removeReservationInfoFromList(pod.UID)
 	cache.deleteReservationOnNode(pod.Spec.NodeName, pod.UID)
 }
 
@@ -184,10 +254,11 @@ func (cache *reservationCache) addPod(reservationUID types.UID, pod *corev1.Pod)
 	if rInfo == nil {
 		return fmt.Errorf("cannot find target reservation")
 	}
-	if rInfo.IsTerminating() {
+	if rInfo.info.IsTerminating() {
 		return fmt.Errorf("target reservation is terminating")
 	}
-	rInfo.AddAssignedPod(pod)
+	rInfo.info.AddAssignedPod(pod)
+	cache.moveReservationInfoToHead(reservationUID)
 	return nil
 }
 
@@ -198,12 +269,13 @@ func (cache *reservationCache) updatePod(reservationUID types.UID, oldPod, newPo
 	rInfo := cache.reservationInfos[reservationUID]
 	if rInfo != nil {
 		if oldPod != nil {
-			rInfo.RemoveAssignedPod(oldPod)
+			rInfo.info.RemoveAssignedPod(oldPod)
 		}
 		if newPod != nil {
-			rInfo.AddAssignedPod(newPod)
+			rInfo.info.AddAssignedPod(newPod)
 		}
 	}
+	cache.moveReservationInfoToHead(reservationUID)
 }
 
 func (cache *reservationCache) deletePod(reservationUID types.UID, pod *corev1.Pod) {
@@ -212,7 +284,8 @@ func (cache *reservationCache) deletePod(reservationUID types.UID, pod *corev1.P
 
 	rInfo := cache.reservationInfos[reservationUID]
 	if rInfo != nil {
-		rInfo.RemoveAssignedPod(pod)
+		rInfo.info.RemoveAssignedPod(pod)
+		cache.moveReservationInfoToHead(reservationUID)
 	}
 }
 
@@ -229,14 +302,14 @@ func (cache *reservationCache) getReservationInfoByUID(uid types.UID) *framework
 	defer cache.lock.RUnlock()
 	rInfo := cache.reservationInfos[uid]
 	if rInfo != nil {
-		return rInfo.Clone()
+		return rInfo.info.Clone()
 	}
 	return nil
 }
 
 func (cache *reservationCache) GetReservationInfoByPod(pod *corev1.Pod, nodeName string) *frameworkext.ReservationInfo {
 	var target *frameworkext.ReservationInfo
-	cache.forEachAvailableReservationOnNode(nodeName, func(rInfo *frameworkext.ReservationInfo) (bool, *framework.Status) {
+	cache.snapshot.forEachAvailableReservationOnNode(nodeName, func(rInfo *frameworkext.ReservationInfo) (bool, *framework.Status) {
 		if _, ok := rInfo.AssignedPods[pod.UID]; ok {
 			target = rInfo
 			return false, nil
@@ -255,8 +328,8 @@ func (cache *reservationCache) forEachAvailableReservationOnNode(nodeName string
 	}
 	for uid := range rOnNode {
 		rInfo := cache.reservationInfos[uid]
-		if rInfo != nil && rInfo.IsAvailable() {
-			beContinue, status := fn(rInfo)
+		if rInfo != nil && rInfo.info.IsAvailable() {
+			beContinue, status := fn(rInfo.info)
 			if !status.IsSuccess() {
 				return status
 			}

pkg/scheduler/plugins/reservation/cache_test.go

@@ -86,6 +86,7 @@ func TestCacheUpdateReservation(t *testing.T) {
 		},
 		Allocated:    nil,
 		AssignedPods: map[types.UID]*frameworkext.PodRequirement{},
+		Generation:   1,
 	}
 	sort.Slice(rInfo.ResourceNames, func(i, j int) bool {
 		return rInfo.ResourceNames[i] < rInfo.ResourceNames[j]
@@ -97,6 +98,7 @@ func TestCacheUpdateReservation(t *testing.T) {
 	assert.Len(t, reservationInfos, 1)
 	rInfo = reservationInfos[0]
 	expectReservationInfo.Allocated = corev1.ResourceList{}
+	expectReservationInfo.Generation = 2
 	sort.Slice(rInfo.ResourceNames, func(i, j int) bool {
 		return rInfo.ResourceNames[i] < rInfo.ResourceNames[j]
 	})
@@ -160,6 +162,7 @@ func TestCacheDeleteReservation(t *testing.T) {
 	sort.Slice(rInfo.ResourceNames, func(i, j int) bool {
 		return rInfo.ResourceNames[i] < rInfo.ResourceNames[j]
 	})
+	rInfo.Generation = 0
 	assert.Equal(t, expectReservationInfo, rInfo)
 
 	cache.DeleteReservation(reservation)
@@ -260,6 +263,7 @@ func TestCacheAddOrUpdateOrDeletePod(t *testing.T) {
 			},
 		},
 	}
+	rInfo.Generation = 0
 	assert.Equal(t, expectReservationInfo, rInfo)
 
 	cache.updatePod(reservation.UID, pod, pod)
@@ -271,6 +275,7 @@ func TestCacheAddOrUpdateOrDeletePod(t *testing.T) {
 		corev1.ResourceCPU:    resource.MustParse("0"),
 		corev1.ResourceMemory: resource.MustParse("0"),
 	})
+	rInfo.Generation = 0
 	assert.Equal(t, expectReservationInfo, rInfo)
 
 	cache.deletePod(reservation.UID, pod)
@@ -282,6 +287,7 @@ func TestCacheAddOrUpdateOrDeletePod(t *testing.T) {
 		corev1.ResourceCPU:    resource.MustParse("0"),
 		corev1.ResourceMemory: resource.MustParse("0"),
 	}
+	rInfo.Generation = 0
 	expectReservationInfo.AssignedPods = map[types.UID]*frameworkext.PodRequirement{}
 	assert.Equal(t, expectReservationInfo, rInfo)
 }

pkg/scheduler/plugins/reservation/plugin.go

@@ -379,7 +379,7 @@ func (pl *Plugin) Filter(ctx context.Context, cycleState *framework.CycleState,
 			allocatePolicy = schedulingv1alpha1.ReservationAllocatePolicyAligned
 		}
 
-		status := pl.reservationCache.forEachAvailableReservationOnNode(node.Name, func(rInfo *frameworkext.ReservationInfo) (bool, *framework.Status) {
+		status := pl.reservationCache.snapshot.forEachAvailableReservationOnNode(node.Name, func(rInfo *frameworkext.ReservationInfo) (bool, *framework.Status) {
 			// ReservationAllocatePolicyDefault cannot coexist with other allocate policies
 			if (allocatePolicy == schedulingv1alpha1.ReservationAllocatePolicyDefault ||
 				rInfo.GetAllocatePolicy() == schedulingv1alpha1.ReservationAllocatePolicyDefault) &&