refactor(cpu): enhance block handling with owner pool info and allocation reuse

- Introduce BlockInfo struct to track owner pool and entry details
- Split block allocation logic for dedicated and shared/reclaimed pools
- Reuse existing allocations for dedicated cores when possible
- Improve NUMA-aware CPU allocation strategy

Author: luomingmeng

pkg/agent/qrm-plugins/cpu/dynamicpolicy/cpuadvisor/helper.go

@@ -85,15 +85,25 @@ func (lwr *ListAndWatchResponse) GetSharedBindingNUMAs() (sets.Int, error) {
 	return sharedBindingNUMAs, nil
 }
 
-// GetBlocks parses ListAndWatchResponse and returns map[int][]*Block,
+type BlockInfo struct {
+	Block
+	OwnerPoolEntryMap map[string]BlockEntry
+}
+
+type BlockEntry struct {
+	EntryName    string
+	SubEntryName string
+}
+
+// GetBlocks parses ListAndWatchResponse and returns map[int][]*BlockInfo,
 // the map is keyed as numa id -> blocks slice (which has been sorted and deduped)
-func (lwr *ListAndWatchResponse) GetBlocks() (map[int][]*Block, error) {
+func (lwr *ListAndWatchResponse) GetBlocks() (map[int][]*BlockInfo, error) {
 	if lwr == nil {
 		return nil, fmt.Errorf("got nil ListAndWatchResponse")
 	}
 
-	numaToBlocks := make(map[int]map[string]*Block)
-	numaToSortedBlocks := make(map[int][]*Block)
+	numaToBlocks := make(map[int]map[string]*BlockInfo)
+	numaToSortedBlocks := make(map[int][]*BlockInfo)
 	blocksEntryNames := make(map[string][][]string)
 	visBlocksToNUMA := make(map[string]int)
 
@@ -119,7 +129,7 @@ func (lwr *ListAndWatchResponse) GetBlocks() (map[int][]*Block, error) {
 				}
 
 				if numaToBlocks[numaIdInt] == nil {
-					numaToBlocks[numaIdInt] = make(map[string]*Block)
+					numaToBlocks[numaIdInt] = make(map[string]*BlockInfo)
 				}
 
 				for _, block := range calculationResult.Blocks {
@@ -140,7 +150,19 @@ func (lwr *ListAndWatchResponse) GetBlocks() (map[int][]*Block, error) {
 						}
 					}
 
-					numaToBlocks[numaIdInt][blockId] = block
+					bi, ok := numaToBlocks[numaIdInt][blockId]
+					if !ok {
+						bi = &BlockInfo{
+							Block:             *block,
+							OwnerPoolEntryMap: make(map[string]BlockEntry),
+						}
+						numaToBlocks[numaIdInt][blockId] = bi
+					}
+
+					bi.OwnerPoolEntryMap[calculationInfo.OwnerPoolName] = BlockEntry{
+						EntryName:    entryName,
+						SubEntryName: subEntryName,
+					}
 					visBlocksToNUMA[blockId] = numaIdInt
 					blocksEntryNames[blockId] = append(blocksEntryNames[blockId], []string{entryName, subEntryName})
 				}
@@ -149,7 +171,7 @@ func (lwr *ListAndWatchResponse) GetBlocks() (map[int][]*Block, error) {
 	}
 
 	for numaId, blocksMap := range numaToBlocks {
-		blocks := make([]*Block, 0, len(blocksMap))
+		blocks := make([]*BlockInfo, 0, len(blocksMap))
 
 		for _, block := range blocksMap {
 			blocks = append(blocks, block)
@@ -170,7 +192,7 @@ func (lwr *ListAndWatchResponse) GetBlocks() (map[int][]*Block, error) {
 }
 
 // getBlocksLessFunc judges if a block is less than another by entryNames of them
-func getBlocksLessFunc(blocksEntryNames map[string][][]string, blocks []*Block) func(i, j int) bool {
+func getBlocksLessFunc(blocksEntryNames map[string][][]string, blocks []*BlockInfo) func(i, j int) bool {
 	return func(i, j int) bool {
 		blockId1 := blocks[i].BlockId
 		blockId2 := blocks[j].BlockId
@@ -231,10 +253,11 @@ func getBlocksLessFunc(blocksEntryNames map[string][][]string, blocks []*Block)
 	}
 }
 
-func logNUMAToBlocksSeq(numaToSortedBlocks map[int][]*Block, blocksEntryNames map[string][][]string) {
+func logNUMAToBlocksSeq(numaToSortedBlocks map[int][]*BlockInfo, blocksEntryNames map[string][][]string) {
 	for numaId, blocks := range numaToSortedBlocks {
 		for i, block := range blocks {
-			general.InfoS("logNUMAToBlocksSeq", "numaId", numaId, "seq", i, "blockId", block.BlockId, "entryNames", blocksEntryNames[block.BlockId])
+			general.InfoS("logNUMAToBlocksSeq", "numaId", numaId, "seq", i, "blockId",
+				block.BlockId, "OwnerPoolEntryMap", block.OwnerPoolEntryMap, "entryNames", blocksEntryNames[block.BlockId])
 		}
 	}
 }

pkg/agent/qrm-plugins/cpu/dynamicpolicy/policy_advisor_handler.go

@@ -548,13 +548,19 @@ func (p *DynamicPolicy) applyCgroupConfigs(resp *advisorapi.ListAndWatchResponse
 	return nil
 }
 
-// generateBlockCPUSet generates BlockCPUSet from cpu-advisor response
-// and the logic contains three main steps
-//  1. handle blocks for static pools
-//  2. handle blocks with specified NUMA ids (probably be blocks for
-//     numa_binding dedicated_cores containers and reclaimed_cores containers colocated with them)
-//  3. handle blocks without specified NUMA id (probably be blocks for
-//     not numa_binding dedicated_cores containers and pools of shared_cores and reclaimed_cores containers)
+// generateBlockCPUSet generates BlockCPUSet from cpu-advisor response.
+// The logic contains the following main steps:
+//  1. Handle blocks for static pools and forbidden pools
+//  2. Handle blocks with specified NUMA IDs (for NUMA-bound dedicated_cores containers
+//     and reclaimed_cores containers colocated with them)
+//  3. Handle blocks without specified NUMA ID (for non-NUMA-bound containers including
+//     dedicated_cores, shared_cores and reclaimed_cores containers)
+//
+// For each block, the function allocates CPU sets based on:
+//   - Already allocated CPUs for dedicated cores
+//   - Available CPUs considering already allocated static/forbidden pools
+//   - NUMA topology awareness for better performance
+//   - CPU allocation strategies that minimize CPU migrations
 func (p *DynamicPolicy) generateBlockCPUSet(resp *advisorapi.ListAndWatchResponse) (advisorapi.BlockCPUSet, error) {
 	if resp == nil {
 		return nil, fmt.Errorf("got nil resp")
@@ -569,8 +575,9 @@ func (p *DynamicPolicy) generateBlockCPUSet(resp *advisorapi.ListAndWatchRespons
 	topology := machineInfo.CPUTopology
 	availableCPUs := topology.CPUDetails.CPUs()
 
-	// walk through static pools to construct blockCPUSet (for static pool),
-	// and calculate availableCPUs after deducting static pools
+	// Walk through static pools to construct blockCPUSet (for static pool),
+	// and calculate availableCPUs after deducting static pools.
+	// Static pools are predefined pools that should not be changed during runtime.
 	blockCPUSet := advisorapi.NewBlockCPUSet()
 	for _, poolName := range state.StaticPools.List() {
 		allocationInfo := p.state.GetAllocationInfo(poolName, commonstate.FakedContainerName)
@@ -588,8 +595,8 @@ func (p *DynamicPolicy) generateBlockCPUSet(resp *advisorapi.ListAndWatchRespons
 		availableCPUs = availableCPUs.Difference(blockCPUSet[blockID])
 	}
 
-	// walk through forbidden pools to construct blockCPUSet (for forbidden pool),
-	// and calculate availableCPUs after deducting forbidden pools
+	// Walk through forbidden pools and deduct their CPUs from availableCPUs.
+	// Forbidden pools are reserved pools that should not be allocated to any containers.
 	for _, poolName := range state.ForbiddenPools.List() {
 		allocationInfo := p.state.GetAllocationInfo(poolName, commonstate.FakedContainerName)
 		if allocationInfo == nil {
@@ -599,20 +606,30 @@ func (p *DynamicPolicy) generateBlockCPUSet(resp *advisorapi.ListAndWatchRespons
 		availableCPUs = availableCPUs.Difference(allocationInfo.AllocationResult.Clone())
 	}
 
-	// walk through all blocks with specified NUMA ids
-	// for each block, add them into blockCPUSet (if not exist) and renew availableCPUs
+	// Process blocks with specified NUMA IDs (for NUMA-bound containers)
+	// These are typically dedicated_cores containers with NUMA binding and
+	// reclaimed_cores containers colocated with them
 	for numaID, blocks := range numaToBlocks {
 		if numaID == commonstate.FakedNUMAID {
 			continue
 		}
 
+		withNUMABindingShareOrDedicatedPod := false
 		numaAvailableCPUs := availableCPUs.Intersection(topology.CPUDetails.CPUsInNUMANodes(numaID))
+
+		// First handle blocks for NUMA-bound dedicated_cores containers
+		// Reuse already allocated CPU sets when possible to minimize CPU migration
 		for _, block := range blocks {
 			if block == nil {
 				general.Warningf("got nil block")
 				continue
 			}
 
+			entry, ok := block.OwnerPoolEntryMap[commonstate.PoolNameDedicated]
+			if !ok {
+				continue
+			}
+
 			blockID := block.BlockId
 
 			if _, found := blockCPUSet[blockID]; found {
@@ -626,6 +643,64 @@ func (p *DynamicPolicy) generateBlockCPUSet(resp *advisorapi.ListAndWatchRespons
 					blockID, err)
 			}
 
+			allocationInfo := p.state.GetAllocationInfo(entry.EntryName, entry.SubEntryName)
+			if allocationInfo == nil {
+				continue
+			}
+
+			alreadyAllocatedCPUs, ok := allocationInfo.TopologyAwareAssignments[numaID]
+			if !ok {
+				continue
+			}
+
+			var cpuset machine.CPUSet
+			alreadyAllocatedCPUs = alreadyAllocatedCPUs.Intersection(numaAvailableCPUs)
+			if alreadyAllocatedCPUs.Size() >= blockResult {
+				cpuset, err = calculator.TakeByTopology(machineInfo, alreadyAllocatedCPUs, blockResult, true)
+				if err != nil {
+					return nil, fmt.Errorf("allocate cpuset for NUMA Aware block: %s in NUMA: %d failed with error: %v, numaAvailableCPUs: %d(%s), blockResult: %d",
+						blockID, numaID, err, numaAvailableCPUs.Size(), numaAvailableCPUs.String(), blockResult)
+				}
+			} else {
+				cpuset, err = calculator.TakeByTopology(machineInfo, numaAvailableCPUs.Difference(alreadyAllocatedCPUs), blockResult-alreadyAllocatedCPUs.Size(), true)
+				if err != nil {
+					return nil, fmt.Errorf("allocate cpuset for NUMA Aware block: %s in NUMA: %d failed with error: %v, numaAvailableCPUs: %d(%s), blockResult: %d",
+						blockID, numaID, err, numaAvailableCPUs.Size(), numaAvailableCPUs.String(), blockResult)
+				}
+				cpuset = cpuset.Union(alreadyAllocatedCPUs)
+			}
+
+			blockCPUSet[blockID] = cpuset
+			numaAvailableCPUs = numaAvailableCPUs.Difference(cpuset)
+			availableCPUs = availableCPUs.Difference(cpuset)
+			withNUMABindingShareOrDedicatedPod = true
+		}
+
+		// Then handle blocks for NUMA-bound shared_cores containers and reclaimed_cores containers colocated with them
+		// These containers can share NUMA nodes with dedicated_cores containers
+		for _, block := range blocks {
+			if block == nil {
+				general.Warningf("got nil block")
+				continue
+			}
+
+			_, ok := block.OwnerPoolEntryMap[commonstate.PoolNameDedicated]
+			if ok {
+				continue
+			}
+
+			blockID := block.BlockId
+			if _, found := blockCPUSet[blockID]; found {
+				general.Warningf("block: %v already allocated", blockID)
+				continue
+			}
+
+			blockResult, err := general.CovertUInt64ToInt(block.Result)
+			if err != nil {
+				return nil, fmt.Errorf("parse block: %s result failed with error: %v",
+					blockID, err)
+			}
+
 			cpuset, err := calculator.TakeByTopology(machineInfo, numaAvailableCPUs, blockResult, false)
 			if err != nil {
 				return nil, fmt.Errorf("allocate cpuset for NUMA Aware block: %s in NUMA: %d failed with error: %v, numaAvailableCPUs: %d(%s), blockResult: %d",
@@ -635,11 +710,26 @@ func (p *DynamicPolicy) generateBlockCPUSet(resp *advisorapi.ListAndWatchRespons
 			blockCPUSet[blockID] = cpuset
 			numaAvailableCPUs = numaAvailableCPUs.Difference(cpuset)
 			availableCPUs = availableCPUs.Difference(cpuset)
+
+			_, ok = block.OwnerPoolEntryMap[commonstate.PoolNameShare]
+			if ok {
+				withNUMABindingShareOrDedicatedPod = true
+			}
+		}
+
+		// Finally, if there are NUMA-bound containers on this NUMA node,
+		// deduct all numaAvailableCPUs from availableCPUs to ensure that
+		// NUMA-bound pods don't share the same NUMA node with non-NUMA-bound pods
+		if withNUMABindingShareOrDedicatedPod {
+			// Because numaAvailableCPUs is a subset of availableCPUs,
+			// we need to deduct all numaAvailableCPUs from availableCPUs
+			availableCPUs = availableCPUs.Difference(numaAvailableCPUs)
 		}
 	}
 
-	// walk through all blocks without specified NUMA id
-	// for each block, add them into blockCPUSet (if not exist) and renew availableCPUs
+	// Walk through all blocks without specified NUMA ID (non-NUMA-bound containers)
+	// For each block, allocate CPUs using NUMA balance strategy to minimize
+	// memory access latency and CPU migrations
 	for _, block := range numaToBlocks[commonstate.FakedNUMAID] {
 		if block == nil {
 			general.Warningf("got nil block")
@@ -659,17 +749,16 @@ func (p *DynamicPolicy) generateBlockCPUSet(resp *advisorapi.ListAndWatchRespons
 				blockID, err)
 		}
 
-		// use NUMA balance strategy to aviod changing memset as much as possible
-		// for blocks with faked NUMA id
-		var cpuset machine.CPUSet
-		cpuset, availableCPUs, err = calculator.TakeByNUMABalance(machineInfo, availableCPUs, blockResult)
+		// Use NUMA balance strategy to avoid changing memory affinity (memset) as much as possible
+		// for blocks with faked NUMA ID (non-NUMA-bound containers)
+		resultCPUSet, _, err := calculator.TakeByNUMABalance(machineInfo, availableCPUs, blockResult)
 		if err != nil {
 			return nil, fmt.Errorf("allocate cpuset for non NUMA Aware block: %s failed with error: %v, availableCPUs: %d(%s), blockResult: %d",
 				blockID, err, availableCPUs.Size(), availableCPUs.String(), blockResult)
 		}
 
-		blockCPUSet[blockID] = cpuset
-		availableCPUs = availableCPUs.Difference(cpuset)
+		blockCPUSet[blockID] = resultCPUSet
+		availableCPUs = availableCPUs.Difference(resultCPUSet)
 	}
 
 	return blockCPUSet, nil