feat(host): kvm vcpus bind cpuset on numa nodes and cpu dies (#15732)

- alloc cpuset on guest startup, try to alloc on one numa or cpu die
- fix guest startup task not clean
- fix cgrup tasks not clean

pkg/hostman/guestman/desc/desc.go

@@ -37,6 +37,11 @@ type SGuestCpu struct {
 	// CpuCacheMode string
 }
 
+type CpuPin struct {
+	Vcpus string
+	Pcpus string
+}
+
 type SMemObject struct {
 	*Object
 	SizeMB int64
@@ -67,6 +72,7 @@ type SGuestMem struct {
 type SGuestHardwareDesc struct {
 	Cpu     int64
 	CpuDesc *SGuestCpu `json:",omitempty"`
+	VcpuPin []CpuPin   `json:",omitempty"`
 	// Clock   *SGuestClock `json:",omitempty"`
 
 	Mem     int64

pkg/hostman/guestman/guesthelper.go

@@ -15,6 +15,9 @@
 package guestman
 
 import (
+	"container/heap"
+	"sync"
+
 	"yunion.io/x/cloudmux/pkg/multicloud/esxi/vcenter"
 	"yunion.io/x/jsonutils"
 
@@ -22,6 +25,7 @@ import (
 	"yunion.io/x/onecloud/pkg/hostman/guestman/desc"
 	"yunion.io/x/onecloud/pkg/hostman/storageman"
 	"yunion.io/x/onecloud/pkg/mcclient"
+	"yunion.io/x/onecloud/pkg/util/cgrouputils/cpuset"
 )
 
 type SBaseParms struct {
@@ -169,3 +173,231 @@ type SQgaGuestSetPassword struct {
 	*hostapi.GuestSetPasswordRequest
 	Sid string
 }
+
+type CpuSetCounter struct {
+	Nodes []*NumaNode
+	Lock  sync.Mutex
+}
+
+func NewGuestCpuSetCounter(info *hostapi.HostTopology, reservedCpus *cpuset.CPUSet) *CpuSetCounter {
+	cpuSetCounter := new(CpuSetCounter)
+	cpuSetCounter.Nodes = make([]*NumaNode, len(info.Nodes))
+	for i := 0; i < len(info.Nodes); i++ {
+		node := new(NumaNode)
+		node.LogicalProcessors = cpuset.NewCPUSet()
+		node.NodeId = info.Nodes[i].ID
+		cpuDies := make([]*CPUDie, 0)
+		for j := 0; j < len(info.Nodes[i].Caches); j++ {
+			if info.Nodes[i].Caches[j].Level != 3 {
+				continue
+			}
+			cpuDie := new(CPUDie)
+			dieBuilder := cpuset.NewBuilder()
+			for k := 0; k < len(info.Nodes[i].Caches[j].LogicalProcessors); k++ {
+				if reservedCpus != nil && reservedCpus.Contains(int(info.Nodes[i].Caches[j].LogicalProcessors[k])) {
+					continue
+				}
+				dieBuilder.Add(int(info.Nodes[i].Caches[j].LogicalProcessors[k]))
+			}
+			cpuDie.LogicalProcessors = dieBuilder.Result()
+			node.CpuCount += cpuDie.LogicalProcessors.Size()
+			node.LogicalProcessors = node.LogicalProcessors.Union(cpuDie.LogicalProcessors)
+			cpuDies = append(cpuDies, cpuDie)
+		}
+		node.CpuDies = cpuDies
+		cpuSetCounter.Nodes[i] = node
+	}
+	heap.Init(cpuSetCounter)
+	return cpuSetCounter
+}
+
+func (pq *CpuSetCounter) AllocCpuset(vcpuCount int) map[int][]int {
+	res := map[int][]int{}
+	sourceVcpuCount := vcpuCount
+	pq.Lock.Lock()
+	defer pq.Lock.Unlock()
+	for vcpuCount > 0 {
+		count := vcpuCount
+		if vcpuCount > pq.Nodes[0].CpuCount {
+			count = vcpuCount/2 + vcpuCount%2
+		}
+		res[pq.Nodes[0].NodeId] = pq.Nodes[0].AllocCpuset(count)
+		pq.Nodes[0].VcpuCount += sourceVcpuCount
+		heap.Fix(pq, 0)
+		vcpuCount -= count
+	}
+	return res
+}
+
+func (pq *CpuSetCounter) ReleaseCpus(cpus []int, vcpuCount int) {
+	pq.Lock.Lock()
+	defer pq.Lock.Unlock()
+	var numaCpuCount = map[int][]int{}
+	for i := 0; i < len(cpus); i++ {
+		for j := 0; j < len(pq.Nodes); j++ {
+			if pq.Nodes[j].LogicalProcessors.Contains(cpus[i]) {
+				if numaCpus, ok := numaCpuCount[pq.Nodes[j].NodeId]; !ok {
+					numaCpuCount[pq.Nodes[j].NodeId] = []int{cpus[i]}
+				} else {
+					numaCpuCount[pq.Nodes[j].NodeId] = append(numaCpus, cpus[i])
+				}
+				break
+			}
+		}
+	}
+	for i := 0; i < len(pq.Nodes); i++ {
+		if numaCpus, ok := numaCpuCount[pq.Nodes[i].NodeId]; ok {
+			pq.Nodes[i].CpuDies.ReleaseCpus(numaCpus, vcpuCount)
+			pq.Nodes[i].VcpuCount -= vcpuCount
+			heap.Fix(pq, i)
+		}
+	}
+}
+
+func (pq *CpuSetCounter) LoadCpus(cpus []int, vcpuCpunt int) {
+	pq.Lock.Lock()
+	defer pq.Lock.Unlock()
+	var numaCpuCount = map[int][]int{}
+	for i := 0; i < len(cpus); i++ {
+		for j := 0; j < len(pq.Nodes); j++ {
+			if pq.Nodes[j].LogicalProcessors.Contains(cpus[i]) {
+				if numaCpus, ok := numaCpuCount[pq.Nodes[j].NodeId]; !ok {
+					numaCpuCount[pq.Nodes[j].NodeId] = []int{cpus[i]}
+				} else {
+					numaCpuCount[pq.Nodes[j].NodeId] = append(numaCpus, cpus[i])
+				}
+				break
+			}
+		}
+	}
+	for i := 0; i < len(pq.Nodes); i++ {
+		if numaCpus, ok := numaCpuCount[pq.Nodes[i].NodeId]; ok {
+			pq.Nodes[i].CpuDies.LoadCpus(numaCpus, vcpuCpunt)
+			pq.Nodes[i].VcpuCount += vcpuCpunt
+			heap.Fix(pq, i)
+		}
+	}
+}
+
+func (pq CpuSetCounter) Len() int { return len(pq.Nodes) }
+
+func (pq CpuSetCounter) Less(i, j int) bool {
+	return pq.Nodes[i].VcpuCount < pq.Nodes[j].VcpuCount
+}
+
+func (pq CpuSetCounter) Swap(i, j int) {
+	pq.Nodes[i], pq.Nodes[j] = pq.Nodes[j], pq.Nodes[i]
+}
+
+func (pq *CpuSetCounter) Push(item interface{}) {
+	(*pq).Nodes = append((*pq).Nodes, item.(*NumaNode))
+}
+
+func (pq *CpuSetCounter) Pop() interface{} {
+	old := *pq
+	n := len(old.Nodes)
+	item := old.Nodes[n-1]
+	old.Nodes[n-1] = nil // avoid memory leak
+	(*pq).Nodes = old.Nodes[0 : n-1]
+	return item
+}
+
+type NumaNode struct {
+	CpuDies           SorttedCPUDie
+	LogicalProcessors cpuset.CPUSet
+	VcpuCount         int
+	CpuCount          int
+	NodeId            int
+}
+
+func (n *NumaNode) AllocCpuset(vcpuCount int) []int {
+	cpus := make([]int, 0)
+	for vcpuCount > 0 {
+		dies := n.CpuDies
+		count := vcpuCount
+		if vcpuCount > dies[0].LogicalProcessors.Size() {
+			count = dies[0].LogicalProcessors.Size()
+		}
+		dies[0].VcpuCount += count
+		heap.Fix(&n.CpuDies, 0)
+		vcpuCount -= count
+		cpus = append(cpus, dies[0].LogicalProcessors.ToSliceNoSort()...)
+	}
+	return cpus
+}
+
+type CPUDie struct {
+	LogicalProcessors cpuset.CPUSet
+	VcpuCount         int
+}
+
+type SorttedCPUDie []*CPUDie
+
+func (pq SorttedCPUDie) Len() int { return len(pq) }
+
+func (pq SorttedCPUDie) Less(i, j int) bool {
+	return pq[i].VcpuCount < pq[j].VcpuCount
+}
+
+func (pq SorttedCPUDie) Swap(i, j int) {
+	pq[i], pq[j] = pq[j], pq[i]
+}
+
+func (pq *SorttedCPUDie) Push(item interface{}) {
+	*pq = append(*pq, item.(*CPUDie))
+}
+
+func (pq *SorttedCPUDie) Pop() interface{} {
+	old := *pq
+	n := len(old)
+	item := old[n-1]
+	old[n-1] = nil // avoid memory leak
+	*pq = old[0 : n-1]
+	return item
+}
+
+func (pq *SorttedCPUDie) ReleaseCpus(cpus []int, vcpuCount int) {
+	var cpuDies = map[int][]int{}
+	for i := 0; i < len(cpus); i++ {
+		for j := 0; j < len(*pq); j++ {
+			if (*pq)[j].LogicalProcessors.Contains(cpus[i]) {
+				if cpuDie, ok := cpuDies[j]; !ok {
+					cpuDies[j] = []int{cpus[i]}
+				} else {
+					cpuDies[j] = append(cpuDie, cpus[i])
+				}
+				break
+			}
+		}
+	}
+
+	for i := 0; i < len(*pq); i++ {
+		if _, ok := cpuDies[i]; ok {
+			(*pq)[i].VcpuCount -= vcpuCount
+			heap.Fix(pq, i)
+		}
+	}
+}
+
+func (pq *SorttedCPUDie) LoadCpus(cpus []int, vcpuCount int) {
+	var cpuDies = map[int][]int{}
+	for i := 0; i < len(cpus); i++ {
+		for j := 0; j < len(*pq); j++ {
+			if (*pq)[j].LogicalProcessors.Contains(cpus[i]) {
+				if cpuDie, ok := cpuDies[j]; !ok {
+					cpuDies[j] = []int{cpus[i]}
+				} else {
+					cpuDies[j] = append(cpuDie, cpus[i])
+				}
+				break
+			}
+		}
+	}
+
+	for i := 0; i < len(*pq); i++ {
+		if _, ok := cpuDies[i]; ok {
+			(*pq)[i].VcpuCount += vcpuCount
+			heap.Fix(pq, i)
+		}
+	}
+}

pkg/hostman/guestman/guestman.go

@@ -52,6 +52,7 @@ import (
 	"yunion.io/x/onecloud/pkg/mcclient"
 	modules "yunion.io/x/onecloud/pkg/mcclient/modules/compute"
 	"yunion.io/x/onecloud/pkg/util/cgrouputils"
+	"yunion.io/x/onecloud/pkg/util/cgrouputils/cpuset"
 	"yunion.io/x/onecloud/pkg/util/fileutils2"
 	"yunion.io/x/onecloud/pkg/util/netutils2"
 	"yunion.io/x/onecloud/pkg/util/procutils"
@@ -94,6 +95,8 @@ type SGuestManager struct {
 
 	qemuMachineCpuMax map[string]uint
 	qemuMaxMem        int
+
+	cpuSet *CpuSetCounter
 }
 
 func NewGuestManager(host hostutils.IHost, serversPath string) *SGuestManager {
@@ -106,6 +109,7 @@ func NewGuestManager(host hostutils.IHost, serversPath string) *SGuestManager {
 	manager.UnknownServers = new(sync.Map)
 	manager.ServersLock = &sync.Mutex{}
 	manager.GuestStartWorker = appsrv.NewWorkerManager("GuestStart", 1, appsrv.DEFAULT_BACKLOG, false)
+	manager.cpuSet = NewGuestCpuSetCounter(host.GetHostTopology(), host.GetReservedCpusInfo())
 	// manager.StartCpusetBalancer()
 	manager.LoadExistingGuests()
 	manager.host.StartDHCPServer()
@@ -326,7 +330,7 @@ func (m *SGuestManager) CPUSet(ctx context.Context, sid string, req *compute.Ser
 	if !ok {
 		return nil, httperrors.NewNotFoundError("Not found")
 	}
-	return guest.CPUSet(ctx, req)
+	return guest.CPUSet(ctx, req.CPUS)
 }
 
 func (m *SGuestManager) CPUSetRemove(ctx context.Context, sid string) error {
@@ -384,6 +388,25 @@ func (m *SGuestManager) LoadServer(sid string) {
 		go guest.sendStreamDisksComplete(context.Background())
 	}
 	m.CandidateServers[sid] = guest
+	m.loadGuestCpuset(guest)
+}
+
+func (m *SGuestManager) loadGuestCpuset(guest *SKVMGuestInstance) {
+	if guest.GetPid() > 0 {
+		for _, vcpuPin := range guest.Desc.VcpuPin {
+			pcpuSet, err := cpuset.Parse(vcpuPin.Pcpus)
+			if err != nil {
+				log.Errorf("failed parse %s pcpus: %s", guest.GetName(), vcpuPin.Pcpus)
+				continue
+			}
+			vcpuSet, err := cpuset.Parse(vcpuPin.Vcpus)
+			if err != nil {
+				log.Errorf("failed parse %s vcpus: %s", guest.GetName(), vcpuPin.Vcpus)
+				continue
+			}
+			m.cpuSet.LoadCpus(pcpuSet.ToSlice(), vcpuSet.Size())
+		}
+	}
 }
 
 func (m *SGuestManager) ShutdownServers() {

pkg/hostman/guestman/guesttasks.go

@@ -80,7 +80,7 @@ func (s *SGuestStopTask) Start() {
 }
 
 func (s *SGuestStopTask) onPowerdownGuest(results string) {
-	s.ExitCleanup(true)
+	//s.ExitCleanup(true)
 	s.startPowerdown = time.Now()
 	s.checkGuestRunning()
 }
@@ -1461,6 +1461,7 @@ func (s *SGuestResumeTask) onStartRunning() {
 			func() { s.startStreamDisks(nil) })
 	} else {
 		s.SyncStatus("")
+		s.detachStartupTask()
 	}
 }
 
@@ -2323,6 +2324,10 @@ func (task *SGuestHotplugCpuMemTask) updateGuestDesc() {
 		}
 		task.Desc.MemDesc.MemSlots = append(task.Desc.MemDesc.MemSlots, task.memSlot)
 	}
+	if task.addedCpuCount > 0 && len(task.Desc.VcpuPin) == 1 {
+		task.Desc.VcpuPin[0].Vcpus = fmt.Sprintf("0-%d", task.Desc.Cpu-1)
+	}
+
 	if task.addedCpuCount > 0 || task.addedMemSize > 0 {
 		task.SaveLiveDesc(task.Desc)
 	}

pkg/hostman/guestman/qemu-kvm.go

@@ -62,6 +62,7 @@ import (
 	modules "yunion.io/x/onecloud/pkg/mcclient/modules/compute"
 	identity_modules "yunion.io/x/onecloud/pkg/mcclient/modules/identity"
 	"yunion.io/x/onecloud/pkg/util/cgrouputils"
+	"yunion.io/x/onecloud/pkg/util/cgrouputils/cpuset"
 	"yunion.io/x/onecloud/pkg/util/fileutils2"
 	"yunion.io/x/onecloud/pkg/util/fuseutils"
 	"yunion.io/x/onecloud/pkg/util/netutils2"
@@ -1217,7 +1218,26 @@ func (s *SKVMGuestInstance) SlaveDisksBlockStream() error {
 	return nil
 }
 
+func (s *SKVMGuestInstance) releaseGuestCpuset() {
+	for _, vcpuPin := range s.Desc.VcpuPin {
+		pcpuSet, err := cpuset.Parse(vcpuPin.Pcpus)
+		if err != nil {
+			log.Errorf("failed parse %s pcpus: %s", s.GetName(), vcpuPin.Pcpus)
+			continue
+		}
+		vcpuSet, err := cpuset.Parse(vcpuPin.Vcpus)
+		if err != nil {
+			log.Errorf("failed parse %s vcpus: %s", s.GetName(), vcpuPin.Vcpus)
+			continue
+		}
+		s.manager.cpuSet.ReleaseCpus(pcpuSet.ToSlice(), vcpuSet.Size())
+	}
+	s.Desc.VcpuPin = nil
+	s.SaveLiveDesc(s.Desc)
+}
+
 func (s *SKVMGuestInstance) clearCgroup(pid int) {
+	s.releaseGuestCpuset()
 	if pid == 0 && s.cgroupPid > 0 {
 		pid = s.cgroupPid
 	}
@@ -1542,6 +1562,8 @@ func (s *SKVMGuestInstance) ExitCleanup(clear bool) {
 		pid := s.GetPid()
 		if pid > 0 {
 			s.clearCgroup(pid)
+		} else {
+			s.clearCgroup(0)
 		}
 	}
 	if s.Monitor != nil {
@@ -2102,7 +2124,7 @@ func (s *SKVMGuestInstance) GetCgroupName() string {
 	return ""
 }
 
-func (s *SKVMGuestInstance) SetCgroup() {
+func (s *SKVMGuestInstance) GuestPrelaunchSetCgroup() {
 	s.cgroupPid = s.GetPid()
 	s.setCgroupIo()
 	s.setCgroupCpu()
@@ -2142,24 +2164,43 @@ func (s *SKVMGuestInstance) setCgroupCpu() {
 }
 
 func (s *SKVMGuestInstance) setCgroupCPUSet() {
-	var input *api.ServerCPUSetInput
+	var cpus []int
 	if cpuset, ok := s.Desc.Metadata[api.VM_METADATA_CGROUP_CPUSET]; ok {
 		cpusetJson, err := jsonutils.ParseString(cpuset)
 		if err != nil {
 			log.Errorf("failed parse server %s cpuset %s: %s", s.Id, cpuset, err)
 			return
 		}
-		input = new(api.ServerCPUSetInput)
+		input := new(api.ServerCPUSetInput)
 		err = cpusetJson.Unmarshal(input)
 		if err != nil {
 			log.Errorf("failed unmarshal server %s cpuset %s", s.Id, err)
 			return
 		}
+		cpus = input.CPUS
+	} else {
+		cpus = s.allocGuestCpuset()
 	}
-	if _, err := s.CPUSet(context.Background(), input); err != nil {
+	if _, err := s.CPUSet(context.Background(), cpus); err != nil {
 		log.Errorf("Do CPUSet error: %v", err)
 		return
 	}
+	s.Desc.VcpuPin = []desc.CpuPin{
+		{
+			Vcpus: fmt.Sprintf("0-%d", s.Desc.Cpu-1),
+			Pcpus: cpuset.NewCPUSet(cpus...).String(),
+		},
+	}
+	s.SaveLiveDesc(s.Desc)
+}
+
+func (s *SKVMGuestInstance) allocGuestCpuset() []int {
+	var cpuset = []int{}
+	numaCpus := s.manager.cpuSet.AllocCpuset(int(s.Desc.Cpu))
+	for _, cpus := range numaCpus {
+		cpuset = append(cpuset, cpus...)
+	}
+	return cpuset
 }
 
 func (s *SKVMGuestInstance) CreateFromDesc(desc *desc.SGuestDesc) error {
@@ -2236,6 +2277,10 @@ func (s *SKVMGuestInstance) optimizeOom() error {
 }
 
 func (s *SKVMGuestInstance) SyncMetadata(meta *jsonutils.JSONDict) error {
+	metaMap, _ := meta.GetMap()
+	for k, v := range metaMap {
+		s.Desc.Metadata[k] = v.String()
+	}
 	_, err := modules.Servers.SetMetadata(hostutils.GetComputeSession(context.Background()),
 		s.Id, meta)
 	if err != nil {
@@ -2328,7 +2373,7 @@ func (s *SKVMGuestInstance) onGuestPrelaunch() error {
 		}
 	}
 	s.OnResumeSyncMetadataInfo()
-	s.SetCgroup()
+	s.GuestPrelaunchSetCgroup()
 	s.optimizeOom()
 	s.doBlockIoThrottle()
 	return nil
@@ -2745,7 +2790,7 @@ func (s *SKVMGuestInstance) getQemuCmdlineFromContent(content string) (string, e
 	return cmdStr, nil
 }
 
-func (s *SKVMGuestInstance) CPUSet(ctx context.Context, input *api.ServerCPUSetInput) (*api.ServerCPUSetResp, error) {
+func (s *SKVMGuestInstance) CPUSet(ctx context.Context, input []int) (*api.ServerCPUSetResp, error) {
 	if !s.IsRunning() {
 		return nil, nil
 	}
@@ -2753,7 +2798,7 @@ func (s *SKVMGuestInstance) CPUSet(ctx context.Context, input *api.ServerCPUSetI
 	var cpusetStr string
 	if input != nil {
 		cpus := []string{}
-		for _, id := range input.CPUS {
+		for _, id := range input {
 			cpus = append(cpus, fmt.Sprintf("%d", id))
 		}
 		cpusetStr = strings.Join(cpus, ",")

pkg/hostman/hostinfo/hostinfo.go

@@ -39,6 +39,7 @@ import (
 
 	"yunion.io/x/onecloud/pkg/apis"
 	api "yunion.io/x/onecloud/pkg/apis/compute"
+	hostapi "yunion.io/x/onecloud/pkg/apis/host"
 	identityapi "yunion.io/x/onecloud/pkg/apis/identity"
 	napi "yunion.io/x/onecloud/pkg/apis/notify"
 	"yunion.io/x/onecloud/pkg/cloudcommon/consts"
@@ -2190,6 +2191,18 @@ func (h *SHostInfo) GetKubeletConfig() kubelet.KubeletConfig {
 	return h.kubeletConfig
 }
 
+func (h *SHostInfo) GetHostTopology() *hostapi.HostTopology {
+	return h.sysinfo.Topology
+}
+
+func (h *SHostInfo) GetReservedCpusInfo() *cpuset.CPUSet {
+	if h.reservedCpusInfo == nil {
+		return nil
+	}
+	cpus, _ := cpuset.Parse(h.reservedCpusInfo.Cpus)
+	return &cpus
+}
+
 func NewHostInfo() (*SHostInfo, error) {
 	var res = new(SHostInfo)
 	res.sysinfo = &SSysInfo{}

pkg/hostman/hostutils/hostutils.go

@@ -24,6 +24,7 @@ import (
 	"yunion.io/x/pkg/appctx"
 
 	"yunion.io/x/onecloud/pkg/apis"
+	hostapi "yunion.io/x/onecloud/pkg/apis/host"
 	"yunion.io/x/onecloud/pkg/appsrv"
 	"yunion.io/x/onecloud/pkg/cloudcommon/consts"
 	"yunion.io/x/onecloud/pkg/cloudcommon/workmanager"
@@ -36,6 +37,7 @@ import (
 	"yunion.io/x/onecloud/pkg/mcclient/auth"
 	modules "yunion.io/x/onecloud/pkg/mcclient/modules/compute"
 	"yunion.io/x/onecloud/pkg/mcclient/modules/k8s"
+	"yunion.io/x/onecloud/pkg/util/cgrouputils/cpuset"
 )
 
 type IHost interface {
@@ -45,6 +47,8 @@ type IHost interface {
 	GetCpuArchitecture() string
 	GetKernelVersion() string
 	IsAarch64() bool
+	GetHostTopology() *hostapi.HostTopology
+	GetReservedCpusInfo() *cpuset.CPUSet
 
 	IsHugepagesEnabled() bool
 	HugepageSizeKb() int

pkg/hostman/options/options.go

@@ -114,7 +114,7 @@ type SHostOptions struct {
 	SetVncPassword         bool `default:"true" help:"Auto set vnc password after monitor connected"`
 	UseBootVga             bool `default:"false" help:"Use boot VGA GPU for guest"`
 
-	EnableCpuBinding         bool `default:"false" help:"Enable cpu binding and rebalance"`
+	EnableCpuBinding         bool `default:"true" help:"Enable cpu binding and rebalance"`
 	EnableOpenflowController bool `default:"false"`
 
 	PingRegionInterval int      `default:"60" help:"interval to ping region, deefault is 1 minute"`