Skip to content

Commit

Permalink
move parent to hyperNodeInfo
Browse files Browse the repository at this point in the history 
Signed-off-by: Monokaix <[email protected]>
@Monokaix
Monokaix committed Jan 13, 2025
1 parent 3dbb30b commit 288f437
Showing 5 changed files with 347 additions and 294 deletions.
163 changes: 94 additions & 69 deletions pkg/scheduler/api/hyper_node_info.go
View file
Original file line number Diff line number Diff line change
@@ -19,12 +19,12 @@ package api
import (
"fmt"
"regexp"
"strconv"
"strings"
"sync"
"sync/atomic"

corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/util/sets"
listerv1 "k8s.io/client-go/listers/core/v1"
@@ -44,8 +44,6 @@ type HyperNodesInfo struct {
// realNodesSet stores the set of real nodes for each HyperNode, eg, s0 and s1 are members of s4,
// s0->node0, node1, s1->node2, node3, s4->node0, node1, s1->node2, node3
realNodesSet map[string]sets.Set[string]
// parentMap stores the parent of each HyperNode.
parentMap map[string]string
// nodeLister Lister to list Kubernetes nodes.
nodeLister listerv1.NodeLister

@@ -59,7 +57,6 @@ func NewHyperNodesInfo(lister listerv1.NodeLister) *HyperNodesInfo {
hyperNodes: make(map[string]*HyperNodeInfo),
hyperNodesSetByTier: make(map[int]sets.Set[string]),
realNodesSet: make(map[string]sets.Set[string]),
parentMap: make(map[string]string),
nodeLister: lister,
ready: new(atomic.Bool),
}
@@ -73,7 +70,6 @@ func NewHyperNodesInfoWithCache(hyperNodesSetByTier map[int]sets.Set[string], re
hyperNodes: make(map[string]*HyperNodeInfo),
hyperNodesSetByTier: hyperNodesSetByTier,
realNodesSet: realNodesSet,
parentMap: make(map[string]string),
ready: ready,
}
}
@@ -82,23 +78,28 @@ func NewHyperNodesInfoWithCache(hyperNodesSetByTier map[int]sets.Set[string], re
type HyperNodeInfo struct {
Name string
HyperNode *topologyv1alpha1.HyperNode
Tier int

tier int
parent string
isDeleting bool
}

// NewHyperNodeInfo creates a new HyperNodeInfo instance.
func NewHyperNodeInfo(hn *topologyv1alpha1.HyperNode, tier int) *HyperNodeInfo {
func NewHyperNodeInfo(hn *topologyv1alpha1.HyperNode) *HyperNodeInfo {
return &HyperNodeInfo{
Name: hn.Name,
HyperNode: hn,
Tier: tier,
tier: hn.Spec.Tier,
}
}

// String returns a string representation of the HyperNodeInfo.
func (hni *HyperNodeInfo) String() string {
return strings.Join([]string{hni.Name, strconv.Itoa(hni.Tier)}, "/")
return strings.Join([]string{
fmt.Sprintf("Name: %s", hni.Name),
fmt.Sprintf(" Tier: %d", hni.tier),
fmt.Sprintf(" Parent: %s", hni.parent)},
",")
}

// HyperNodes returns the map of all HyperNode information.
@@ -128,11 +129,6 @@ func (hni *HyperNodesInfo) RealNodesSet() map[string]sets.Set[string] {
return copiedRealNodesSet
}

// ParentMap returns the map of parent-child relationships between HyperNodes.
func (hni *HyperNodesInfo) ParentMap() map[string]string {
return hni.parentMap
}

// DeleteHyperNode deletes a HyperNode from the cache and update hyperNode tree.
func (hni *HyperNodesInfo) DeleteHyperNode(name string) error {
hni.markHyperNodeIsDeleting(name)
@@ -147,8 +143,19 @@ func (hni *HyperNodesInfo) DeleteHyperNode(name string) error {
}

func (hni *HyperNodesInfo) UpdateHyperNode(hn *topologyv1alpha1.HyperNode) error {
hni.updateHyperNode(hn)
return hni.updateAncestors(hn.Name)
hni.updateParent(hn)
hni.updateHyperNodesSetByTier(hn)

name := hn.Name
old, exists := hni.hyperNodes[name]
if exists {
old.HyperNode = hn
old.tier = hn.Spec.Tier
} else {
hni.hyperNodes[name] = NewHyperNodeInfo(hn)
}

return hni.updateAncestors(name)
}

func (hni *HyperNodesInfo) BuildHyperNodeCache(hn *HyperNodeInfo, processed sets.Set[string], ancestorsChain sets.Set[string], ancestors sets.Set[string], nodes []*corev1.Node) error {
@@ -168,7 +175,7 @@ func (hni *HyperNodesInfo) BuildHyperNodeCache(hn *HyperNodeInfo, processed sets
ancestorsChain.Insert(hn.Name)
defer ancestorsChain.Delete(hn.Name)

if hni.HyperNodeIsDeleting(hn.Name) {
if hni.hyperNodeIsDeleting(hn.Name) {
klog.InfoS("HyperNode is being deleted", "name", hn.Name)
return nil
}
@@ -190,7 +197,7 @@ func (hni *HyperNodesInfo) BuildHyperNodeCache(hn *HyperNodeInfo, processed sets
continue
}

if err := hni.setParent(hn, memberName); err != nil {
if err := hni.setParent(memberName, hn.Name); err != nil {
return err
}

@@ -231,19 +238,11 @@ func (hni *HyperNodesInfo) setReady(ready bool) {
hni.ready.Store(ready)
}

// UpdateParentMap updates the parent-child relationships in the parentMap.
func (hni *HyperNodesInfo) UpdateParentMap(hyperNode *topologyv1alpha1.HyperNode) {
oldMembers := sets.New[string]()
if oldHyperNodeInfo, ok := hni.hyperNodes[hyperNode.Name]; ok {
for _, member := range oldHyperNodeInfo.HyperNode.Spec.Members {
if member.Type == topologyv1alpha1.MemberTypeHyperNode && member.Selector.ExactMatch != nil {
oldMembers.Insert(member.Selector.ExactMatch.Name)
}
}
}

// updateParent updates the parent-child relationships if members are removed.
func (hni *HyperNodesInfo) updateParent(hn *topologyv1alpha1.HyperNode) {
oldMembers := hni.getChildren(hn.Name)
newMembers := sets.New[string]()
for _, member := range hyperNode.Spec.Members {
for _, member := range hn.Spec.Members {
if member.Type == topologyv1alpha1.MemberTypeHyperNode && member.Selector.ExactMatch != nil {
newMembers.Insert(member.Selector.ExactMatch.Name)
}
@@ -253,9 +252,8 @@ func (hni *HyperNodesInfo) UpdateParentMap(hyperNode *topologyv1alpha1.HyperNode
if removedMembers.Len() == 0 {
return
}

for member := range removedMembers {
delete(hni.parentMap, member)
hni.resetParent(member)
}
}

@@ -268,13 +266,19 @@ func (hni *HyperNodesInfo) GetAncestors(name string) sets.Set[string] {
current := queue[0]
queue = queue[1:]

parent, ok := hni.parentMap[current]
if !ok {
parent := ""
hn, ok := hni.hyperNodes[current]
if ok && hn.parent != "" {
parent = hn.parent
} else {
parent = hni.getParent(name)
}
if parent != "" && !ancestors.Has(parent) {
ancestors.Insert(parent)
queue = append(queue, parent)
if !ancestors.Has(parent) {
ancestors.Insert(parent)
queue = append(queue, parent)
}

}
}
return ancestors
@@ -308,7 +312,6 @@ func (hni *HyperNodesInfo) getChildren(hyperNodeName string) sets.Set[string] {
children := sets.New[string]()
hn, ok := hni.hyperNodes[hyperNodeName]
if !ok {
klog.ErrorS(nil, "HyperNode not found", "name", hyperNodeName)
return children
}

@@ -322,13 +325,27 @@ func (hni *HyperNodesInfo) getChildren(hyperNodeName string) sets.Set[string] {
}

// setParent sets the parent of a HyperNode member.
func (hni *HyperNodesInfo) setParent(hn *HyperNodeInfo, memberName string) error {
parent, ok := hni.parentMap[memberName]
if ok && parent != hn.Name {
return fmt.Errorf("HyperNode %s already has a parent %s, and cannot set another parent %s", memberName, parent, hn.Name)
} else {
hni.parentMap[memberName] = hn.Name
func (hni *HyperNodesInfo) setParent(member, parent string) error {
hn, ok := hni.hyperNodes[member]
if !ok {
klog.InfoS("HyperNode not exists in cache, maybe not created or not be watched, will set parent first", "name", member, "parent", parent)
hn = NewHyperNodeInfo(&topologyv1alpha1.HyperNode{ObjectMeta: metav1.ObjectMeta{
Name: member,
}})
hni.hyperNodes[member] = hn
hn.parent = parent
return nil
}
currentParent := hn.parent
if currentParent == "" {
hn.parent = parent
return nil
}

if currentParent != parent {
return fmt.Errorf("HyperNode %s already has a parent %s, and cannot set another parent %s", member, currentParent, parent)
}

return nil
}

@@ -337,10 +354,10 @@ func (hni *HyperNodesInfo) setParent(hn *HyperNodeInfo, memberName string) error
func (hni *HyperNodesInfo) getParent(name string) string {
tier := -1
if hn, ok := hni.hyperNodes[name]; ok {
tier = hn.Tier
tier = hn.tier
}
for _, hn := range hni.hyperNodes {
if hn.Tier <= tier {
if hn.tier <= tier {
continue
}
for _, member := range hn.HyperNode.Spec.Members {
@@ -394,10 +411,11 @@ func (hni *HyperNodesInfo) exactMatchMember(selector topologyv1alpha1.MemberSele

func (hni *HyperNodesInfo) updateAncestors(name string) error {
ancestors := hni.GetAncestors(name)
// Clear expectedRealNodesSet of hyperNode before rebuild hyperNode cache.
fmt.Println("acxx", ancestors)
// Clear realNodesSet of hyperNode before rebuild hyperNode cache.
for ancestor := range ancestors {
delete(hni.realNodesSet, ancestor)
delete(hni.parentMap, ancestor)
hni.resetParent(ancestor)
}

processed := sets.New[string]()
@@ -420,19 +438,6 @@ func (hni *HyperNodesInfo) updateAncestors(name string) error {
return nil
}

// UpdateHyperNode updates the information of a HyperNode.
func (hni *HyperNodesInfo) updateHyperNode(hyperNode *topologyv1alpha1.HyperNode) {
tier := hyperNode.Spec.Tier
old, exists := hni.hyperNodes[hyperNode.Name]
if exists {
if old.Tier != hyperNode.Spec.Tier {
hni.removeFromTierSet(hyperNode.Name, old.Tier)
}
}
hni.hyperNodes[hyperNode.Name] = NewHyperNodeInfo(hyperNode, tier)
hni.updateHyperNodeListByTier(hyperNode, tier)
}

// removeFromTierSet removes a HyperNode from the tier set.
func (hni *HyperNodesInfo) removeFromTierSet(hyperNodeName string, tier int) {
set, ok := hni.hyperNodesSetByTier[tier]
@@ -446,19 +451,31 @@ func (hni *HyperNodesInfo) removeFromTierSet(hyperNodeName string, tier int) {
}
}

// updateHyperNodeListByTier updates the list of HyperNodes by tier.
func (hni *HyperNodesInfo) updateHyperNodeListByTier(hyperNode *topologyv1alpha1.HyperNode, tier int) {
func (hni *HyperNodesInfo) updateHyperNodesSetByTier(hyperNode *topologyv1alpha1.HyperNode) {
tier := hyperNode.Spec.Tier
old, exists := hni.hyperNodes[hyperNode.Name]
if exists {
if old.tier != hyperNode.Spec.Tier {
hni.removeFromTierSet(hyperNode.Name, old.tier)
}
}
if _, ok := hni.hyperNodesSetByTier[tier]; !ok {
hni.hyperNodesSetByTier[tier] = sets.New[string]()
}
hni.hyperNodesSetByTier[tier].Insert(hyperNode.Name)
}

// removeParent removes the parent of a HyperNode.
func (hni *HyperNodesInfo) removeParent(hyperNodeName string) {
children := hni.getChildren(hyperNodeName)
func (hni *HyperNodesInfo) removeParent(name string) {
children := hni.getChildren(name)
for child := range children {
delete(hni.parentMap, child)
hni.resetParent(child)
}
}

func (hni *HyperNodesInfo) resetParent(name string) {
if hn, ok := hni.hyperNodes[name]; ok {
hn.parent = ""
}
}

@@ -469,8 +486,7 @@ func (hni *HyperNodesInfo) deleteHyperNode(name string) {
return
}
delete(hni.hyperNodes, name)
hni.removeFromTierSet(name, hn.Tier)
delete(hni.parentMap, name)
hni.removeFromTierSet(name, hn.tier)
}

// markHyperNodeIsDeleting marks a HyperNode as being deleted.
@@ -483,12 +499,21 @@ func (hni *HyperNodesInfo) markHyperNodeIsDeleting(name string) {
hn.isDeleting = true
}

// HyperNodeIsDeleting checks if a HyperNode is being deleted.
func (hni *HyperNodesInfo) HyperNodeIsDeleting(name string) bool {
// hyperNodeIsDeleting checks if a HyperNode is being deleted.
func (hni *HyperNodesInfo) hyperNodeIsDeleting(name string) bool {
hn, ok := hni.hyperNodes[name]
if !ok {
klog.ErrorS(nil, "HyperNode not exists in cache", "name", name)
return false
}
return hn.isDeleting
}

// HyperNodesInfo returns the hyperNodes info with a human-readable string.
func (hni *HyperNodesInfo) HyperNodesInfo() map[string]string {
actualHyperNodes := make(map[string]string)
for name, hn := range hni.hyperNodes {
actualHyperNodes[name] = hn.String()
}
return actualHyperNodes
}
Loading

0 comments on commit 288f437

Please sign in to comment.