diff --git a/cmd/plugins/topology-aware/policy/dra.go b/cmd/plugins/topology-aware/policy/dra.go index a7f4db627..f7b415584 100644 --- a/cmd/plugins/topology-aware/policy/dra.go +++ b/cmd/plugins/topology-aware/policy/dra.go @@ -14,18 +14,19 @@ package topologyaware -// We publish each NUMA node as a DRA device. Its capacity is the number of -// CPUs we can give out from that node. A claim requests a number of CPUs -// from one device. +// We publish each group of CPUs of the topology as a DRA device: the machine, +// each socket and each NUMA node. Its capacity is the number of CPUs we can +// give out from that group. A claim requests a number of CPUs from one device. // // We allocate the CPUs of a claim as a grant, the same way as for a // container. The grant has only exclusive CPUs: no shared part and no // memory. Like other grants, it is accounted, kept across a reconfiguration // and saved to the cache. The claimed CPUs leave the free supply, so the -// policy does not hand them out again. A claim on a socket device is not -// bound to its CPUs: when a claim on a NUMA node of the socket needs them, -// it moves to other CPUs of the socket, and the containers holding it move -// with it. +// policy does not hand them out again. A claim is not bound to its CPUs: +// when a claim on a smaller device inside its device needs them, it moves to +// other CPUs of its device, and the containers holding it move with it. +// Claims on larger devices holding these other CPUs move out of its device +// first. // A container holding claims runs on their CPUs, and also on its own CPUs // if it has a CPU request. A container naming only some requests of a claim // runs on the CPUs of those requests. @@ -55,15 +56,18 @@ import ( "github.com/containers/nri-plugins/pkg/resmgr/cache" "github.com/containers/nri-plugins/pkg/resmgr/dra" libmem "github.com/containers/nri-plugins/pkg/resmgr/lib/memory" + system "github.com/containers/nri-plugins/pkg/sysfs" "github.com/containers/nri-plugins/pkg/utils/cpuset" idset "github.com/intel/goresctrl/pkg/utils" ) const ( - // draNUMADeviceFormat names a device after the NUMA node it publishes. - draNUMADeviceFormat = "cpudevnuma%03d" + // draMachineDevice is the name of the device which publishes the machine. + draMachineDevice = "cpudevmachine" // draSocketDeviceFormat names a device after the socket it publishes. draSocketDeviceFormat = "cpudevsocket%03d" + // draNUMADeviceFormat names a device after the NUMA node it publishes. + draNUMADeviceFormat = "cpudevnuma%03d" // draCapacityCPU is the number of CPUs a claim consumes of a device. draCapacityCPU resourceapi.QualifiedName = "dra.cpu/cpu" // draAttrSocketID is the socket the CPUs of a device belong to. @@ -74,9 +78,10 @@ const ( draAttrLevel resourceapi.QualifiedName = "dra.cpu/level" // draAttrNumCPUs is the number of CPUs of a device, used or not. draAttrNumCPUs resourceapi.QualifiedName = "dra.cpu/numCPUs" - // draLevelNUMANode and draLevelSocket are the levels of our devices. - draLevelNUMANode = "numanode" + // The levels of our devices. + draLevelMachine = "machine" draLevelSocket = "socket" + draLevelNUMANode = "numanode" // draConfigAPIVersion and draConfigKind identify our claim configs. draConfigAPIVersion = "config.nri/v1alpha1" draConfigKind = "TopologyAwareClaimConfig" @@ -98,65 +103,102 @@ type claim struct { moves map[string]claimMove } +// cpus returns the CPUs of the claim's results. +func (c *claim) cpus() cpuset.CPUSet { + cpus := cpuset.New() + for _, held := range c.results { + cpus = cpus.Union(held) + } + return cpus +} + // claimMove is a claim moved to other CPUs, as it was before and after the // move. type claimMove struct { from, to *claim } -// draDevices returns one device per NUMA node and one per socket which has -// CPUs for claims. -func (p *policy) draDevices() []resourceapi.Device { - var devices []resourceapi.Device +// draCPUGroup is a group of CPUs of the topology which we publish as a DRA +// device: its device name, level and CPUs, and the ID attributes of its level. +type draCPUGroup struct { + name string + level string + cpus cpuset.CPUSet + ids map[resourceapi.QualifiedName]int64 +} - for _, id := range p.sys.NodeIDs() { - node := p.sys.Node(id) - attrs := map[resourceapi.QualifiedName]resourceapi.DeviceAttribute{ - deviceattribute.StandardDeviceAttributeNUMANode: { - IntValue: ptr.To(int64(id)), +// draCPUGroups returns the NUMA nodes, the sockets and the machine. +func draCPUGroups(sys system.System) []draCPUGroup { + var groups []draCPUGroup + for _, id := range sys.NodeIDs() { + groups = append(groups, draCPUGroup{ + name: fmt.Sprintf(draNUMADeviceFormat, id), + level: draLevelNUMANode, + cpus: sys.Node(id).CPUSet(), + ids: map[resourceapi.QualifiedName]int64{ + deviceattribute.StandardDeviceAttributeNUMANode: int64(id), + draAttrSocketID: int64(sys.Node(id).PackageID()), }, - } - name := fmt.Sprintf(draNUMADeviceFormat, id) - if d, ok := p.draDevice(name, draLevelNUMANode, node.PackageID(), attrs); ok { - devices = append(devices, d) - } + }) } - for _, id := range p.sys.PackageIDs() { - attrs := map[resourceapi.QualifiedName]resourceapi.DeviceAttribute{} - name := fmt.Sprintf(draSocketDeviceFormat, id) - if d, ok := p.draDevice(name, draLevelSocket, id, attrs); ok { + for _, id := range sys.PackageIDs() { + groups = append(groups, draCPUGroup{ + name: fmt.Sprintf(draSocketDeviceFormat, id), + level: draLevelSocket, + cpus: sys.Package(id).CPUSet(), + ids: map[resourceapi.QualifiedName]int64{draAttrSocketID: int64(id)}, + }) + } + return append(groups, draCPUGroup{name: draMachineDevice, level: draLevelMachine, cpus: sys.CPUSet()}) +} + +// draDevices returns a device for each CPU group which has CPUs for claims. +func (p *policy) draDevices() []resourceapi.Device { + var devices []resourceapi.Device + for _, group := range p.groups { + if d, ok := p.draDevice(group); ok { devices = append(devices, d) } } - return devices } -// draDevice returns the named device with the attributes of its level and -// socket added to attrs, or false if it has no CPUs for claims. -func (p *policy) draDevice(name, level string, socket idset.ID, attrs map[resourceapi.QualifiedName]resourceapi.DeviceAttribute) (resourceapi.Device, bool) { - cpus, _ := p.draDeviceCPUs(name) +// draDevice returns the device of the CPU group, or false if it has no CPUs +// for claims. +func (p *policy) draDevice(group draCPUGroup) (resourceapi.Device, bool) { + cpus, _ := p.draDeviceCPUs(group.name) if cpus.IsEmpty() { return resourceapi.Device{}, false } - capacity := p.draCapacity(name, cpus) + capacity := p.draCapacity(group.name, cpus) if capacity == 0 { return resourceapi.Device{}, false } - attrs[draAttrLevel] = resourceapi.DeviceAttribute{StringValue: ptr.To(level)} - attrs[draAttrSocketID] = resourceapi.DeviceAttribute{IntValue: ptr.To(int64(socket))} - attrs[draAttrSMTEnabled] = resourceapi.DeviceAttribute{BoolValue: ptr.To(p.sys.MaxThreadCount() > 1)} - attrs[draAttrNumCPUs] = resourceapi.DeviceAttribute{IntValue: ptr.To(int64(cpus.Size()))} - roots := p.sys.Package(socket).PCIeRoots() + attrs := map[resourceapi.QualifiedName]resourceapi.DeviceAttribute{ + draAttrLevel: {StringValue: ptr.To(group.level)}, + draAttrSMTEnabled: {BoolValue: ptr.To(p.sys.MaxThreadCount() > 1)}, + draAttrNumCPUs: {IntValue: ptr.To(int64(cpus.Size()))}, + } + for attr, id := range group.ids { + attrs[attr] = resourceapi.DeviceAttribute{IntValue: ptr.To(id)} + } + var roots []string + for _, id := range p.sys.PackageIDs() { + if pkg := p.sys.Package(id); !pkg.CPUSet().Intersection(cpus).IsEmpty() { + roots = append(roots, pkg.PCIeRoots()...) + } + } if len(roots) > 0 && p.options.Owner.DRAListAttributesSupported() { + // A root with CPUs of several sockets is a root of each of them. + slices.Sort(roots) attrs[deviceattribute.StandardDeviceAttributePCIeRoot] = resourceapi.DeviceAttribute{ - StringValues: roots, + StringValues: slices.Compact(roots), } } return resourceapi.Device{ - Name: name, + Name: group.name, AllowMultipleAllocations: ptr.To(true), Attributes: attrs, Capacity: map[resourceapi.QualifiedName]resourceapi.DeviceCapacity{ @@ -340,20 +382,62 @@ func (p *policy) ReleaseClaim(uid types.UID, undo bool) error { // moveClaimsBack moves the claims allocating c moved back to the CPUs they // had, unless they moved again since, and pins their holders to those CPUs. func (p *policy) moveClaimsBack(c *claim) { - var back []string - for _, uid := range slices.Sorted(maps.Keys(c.moves)) { - m := c.moves[uid] + left := map[string]claimMove{} + for uid, m := range c.moves { if p.allocations.claims[uid] != m.to { log.Warnf("DRA claim %s moved again or was released, not moving it back", uid) continue } - if p.swapClaim(uid, m.to, m.from) { + left[uid] = m + } + + // A claim may have moved to CPUs another claim had, which then moves back + // only after it. + var back []string + for len(left) > 0 { + uid, ok := nextMoveBack(left) + if ok && p.swapClaim(uid, left[uid].to, left[uid].from) { back = append(back, uid) + delete(left, uid) + continue + } + // Claims holding each other's CPUs move back together. So do claims + // which need the CPUs another one frees for the shared containers. + from := map[string]*claim{} + for uid, m := range left { + from[uid] = m.from + } + if p.swapClaims(from) { + back = append(back, slices.Sorted(maps.Keys(from))...) + break } + if !ok { + break + } + delete(left, uid) } p.repinHolders(back) } +// nextMoveBack returns the first claim of moves, by UID, whose CPUs before +// its move no other claim of moves holds, or false if there is none. +func nextMoveBack(moves map[string]claimMove) (string, bool) { + for _, uid := range slices.Sorted(maps.Keys(moves)) { + had := moves[uid].from.cpus() + free := true + for other, m := range moves { + if other != uid && !m.to.ExclusiveCPUs().Intersection(had).IsEmpty() { + free = false + break + } + } + if free { + return uid, true + } + } + return "", false +} + // pruneMoves removes the released claims from the moves of the claims, and // from those of the claims as they were before and after a move. func (p *policy) pruneMoves() { @@ -563,23 +647,12 @@ func (p *policy) draRequests(results []resourceapi.DeviceRequestAllocationResult // draDeviceCPUs returns the grantable CPUs of the device, or false for an // unknown device. func (p *policy) draDeviceCPUs(device string) (cpuset.CPUSet, bool) { - var cpus cpuset.CPUSet - if id, ok := draDeviceID(device, draNUMADeviceFormat); ok { - node := p.sys.Node(id) - if node == nil { - return cpuset.New(), false + for _, group := range p.groups { + if group.name == device { + return group.cpus.Intersection(grantableCPUs(p.root.GetSupply())), true } - cpus = node.CPUSet() - } else if id, ok := draDeviceID(device, draSocketDeviceFormat); ok { - pkg := p.sys.Package(id) - if pkg == nil { - return cpuset.New(), false - } - cpus = pkg.CPUSet() - } else { - return cpuset.New(), false } - return cpus.Intersection(grantableCPUs(p.root.GetSupply())), true + return cpuset.New(), false } // numaDevices returns the NUMA device of the first CPU of each result, or nil if @@ -601,16 +674,6 @@ func (p *policy) numaDevices(results []cpuset.CPUSet) []string { return devices } -// draDeviceID returns the ID in the device name of the given format, or false -// if the name is not of the format. -func draDeviceID(device, format string) (idset.ID, bool) { - var id int - if _, err := fmt.Sscanf(device, format, &id); err != nil { - return 0, false - } - return idset.ID(id), fmt.Sprintf(format, id) == device -} - // allocateClaim returns a claim with the requested CPUs reserved for it, each // result with CPUs of its own. If move is true, it moves claims on larger // devices off the CPUs a request needs, and records the moves in the returned @@ -634,11 +697,12 @@ func (p *policy) allocateClaim(reqs []draRequest, move bool) (*claim, error) { req := reqs[i] picked, err := p.pickCPUs(req, cpus) if err != nil && move { - p.moveClaimsOff(req, cpus, moved) + short := req.cpus - p.availableCPUs(req.deviceCPUs, cpus).Size() + p.moveClaimsOff(req.deviceCPUs, req.deviceCPUs, short, cpus, moved) picked, err = p.pickCPUs(req, cpus) } if err != nil { - p.undoMoves(moved) + p.swapClaims(moved) return nil, err } cpus = cpus.Union(picked) @@ -647,7 +711,7 @@ func (p *policy) allocateClaim(reqs []draRequest, move bool) (*claim, error) { } if err := p.reserveClaim(c); err != nil { - p.undoMoves(moved) + p.swapClaims(moved) return nil, err } for uid, from := range moved { @@ -662,64 +726,69 @@ func (p *policy) allocateClaim(reqs []draRequest, move bool) (*claim, error) { // reserveClaim reserves a grant of all CPUs of the claim's results for it, // from the deepest pool that has all of them. func (p *policy) reserveClaim(c *claim) error { - cpus := cpuset.New() - for _, held := range c.results { - cpus = cpus.Union(held) - } + cpus := c.cpus() pool := p.poolForCPUs(cpus) c.Grant = newGrant(pool, nil, cpuNormal, "", cpus, 0, 0, nil, 0) _, err := pool.FreeSupply().Reserve(c.Grant, nil) return err } -// moveClaimsOff moves claims on devices larger than the device of req to -// other CPUs of their devices, none of them in taken, until req has the CPUs -// it asks for or no claim can move. moved gets the claims as they were before -// their first move. -func (p *policy) moveClaimsOff(req draRequest, taken cpuset.CPUSet, moved map[string]*claim) { - short := req.cpus - p.availableCPUs(req.deviceCPUs, taken).Size() +// moveClaimsOff moves claims on devices larger than box off the CPUs in +// target, to other CPUs of their devices outside box, none of them in taken, +// until n more CPUs of target are free or no claim can move. target is part of +// box. moved gets the claims as they were before their first move. +func (p *policy) moveClaimsOff(box, target cpuset.CPUSet, n int, taken cpuset.CPUSet, moved map[string]*claim) { for _, uid := range slices.Sorted(maps.Keys(p.allocations.claims)) { - if short <= 0 { - return - } - c := p.allocations.claims[uid] - // The claim keeps its own moves, which a failed prepare still undoes. - next := &claim{results: slices.Clone(c.results), devices: c.devices, mpols: c.mpols, moves: c.moves} - gave, picked := 0, cpuset.New() - for i, held := range c.results { - if i >= len(c.devices) || !p.largerDevice(c.devices[i], req.deviceCPUs) { + for i := range p.allocations.claims[uid].results { + if n <= 0 { + return + } + c := p.allocations.claims[uid] + if i >= len(c.devices) || !p.largerDevice(c.devices[i], box) { continue } - // A result gives up only as many CPUs as req still needs. - give, replace := p.moveResult(c.devices[i], held, req.deviceCPUs, short-gave, taken.Union(picked)) + // A result gives up only as many CPUs as are still needed. + give, replace := p.moveResult(c.devices[i], c.results[i], box, target, n, taken, moved) if give.IsEmpty() { continue } - next.results[i] = held.Difference(give).Union(replace) - gave += give.Size() - picked = picked.Union(replace) - } - if gave > 0 && p.swapClaim(uid, c, next) { - if _, ok := moved[uid]; !ok { - moved[uid] = c + // Making room for the result may have moved other results of the + // claim. + c = p.allocations.claims[uid] + // The claim keeps its own moves, which a failed prepare still undoes. + next := &claim{results: slices.Clone(c.results), devices: c.devices, mpols: c.mpols, moves: c.moves} + next.results[i] = c.results[i].Difference(give).Union(replace) + if p.swapClaim(uid, c, next) { + if _, ok := moved[uid]; !ok { + moved[uid] = c + } + n -= give.Size() } - short -= gave } } } // moveResult returns up to n CPUs of a result held on device which are in -// cpus, and as many other CPUs of the device to replace them with, none of -// them in taken. The replacing CPUs leave the free CPUs of cpus free. It -// returns no CPUs if the device has no others free. -func (p *policy) moveResult(device string, held, cpus cpuset.CPUSet, n int, taken cpuset.CPUSet) (cpuset.CPUSet, cpuset.CPUSet) { - give := held.Intersection(cpus) +// target, and as many other CPUs of the device outside box to replace them +// with, none of them in taken. If too few of those are free, it first moves +// claims on larger devices off them. The replacing CPUs leave the free CPUs of +// target free. It returns no CPUs if none are free. +func (p *policy) moveResult(device string, held, box, target cpuset.CPUSet, n int, + taken cpuset.CPUSet, moved map[string]*claim) (cpuset.CPUSet, cpuset.CPUSet) { + give := held.Intersection(target) + if give.IsEmpty() { + return cpuset.New(), cpuset.New() + } deviceCPUs, _ := p.draDeviceCPUs(device) - other := draRequest{device: device, deviceCPUs: deviceCPUs.Difference(cpus)} - // Taking the free CPUs of cpus here keeps the CPUs of other from using - // up the spare CPUs of shared containers which cpus needs. - taken = taken.Union(p.availableCPUs(cpus, taken)) - other.cpus = min(n, give.Size(), p.availableCPUs(other.deviceCPUs, taken).Size()) + other := draRequest{device: device, deviceCPUs: deviceCPUs.Difference(box)} + // Taking the free CPUs of target here keeps the CPUs of other from using + // up the spare CPUs of shared containers which target needs. + taken = taken.Union(p.availableCPUs(target, taken)) + want := min(n, give.Size()) + if short := want - p.availableCPUs(other.deviceCPUs, taken).Size(); short > 0 { + p.moveClaimsOff(deviceCPUs, other.deviceCPUs, short, taken, moved) + } + other.cpus = min(want, p.availableCPUs(other.deviceCPUs, taken).Size()) if other.cpus == 0 { return cpuset.New(), cpuset.New() } @@ -727,7 +796,7 @@ func (p *policy) moveResult(device string, held, cpus cpuset.CPUSet, n int, take if other.cpus < give.Size() { // The CPUs the result keeps stay together. if _, err := p.cpuAllocator.ReleaseCpus(&give, other.cpus, defaultPrio.Option()); err != nil { - log.Errorf("device %s: failed to pick CPUs to move off %s: %v", device, cpus, err) + log.Errorf("device %s: failed to pick CPUs to move off %s: %v", device, target, err) return cpuset.New(), cpuset.New() } } @@ -743,39 +812,41 @@ func (p *policy) moveResult(device string, held, cpus cpuset.CPUSet, n int, take // whether it did. It keeps c if next holds the same CPUs or cannot have // them. func (p *policy) swapClaim(uid string, c, next *claim) bool { - from, to := c.ExclusiveCPUs(), cpuset.New() - for _, held := range next.results { - to = to.Union(held) - } - if to.Equals(from) { - return false - } - - c.Release() - if err := p.reserveClaim(next); err != nil { - log.Errorf("claim %s: failed to move from CPUs %s to %s: %v", uid, from, to, err) - if _, err := c.GetCPUNode().FreeSupply().Reserve(c.Grant, nil); err != nil { - log.Errorf("claim %s: failed to get CPUs %s back: %v", uid, from, err) - } + if next.cpus().Equals(c.ExclusiveCPUs()) { return false } - p.allocations.claims[uid] = next - log.Infof("moved DRA claim %s from CPUs %s to %s", uid, from, to) - return true + return p.swapClaims(map[string]*claim{uid: next}) } -// undoMoves gives the moved claims back the CPUs they had. -func (p *policy) undoMoves(moved map[string]*claim) { - for uid := range moved { +// swapClaims replaces the claims by the ones in next, with the CPUs those +// hold, and tells whether it did. It keeps all claims if one of next cannot +// have its CPUs. +func (p *policy) swapClaims(next map[string]*claim) bool { + uids := slices.Sorted(maps.Keys(next)) + for _, uid := range uids { p.allocations.claims[uid].Release() } - for uid, c := range moved { - if _, err := c.GetCPUNode().FreeSupply().Reserve(c.Grant, nil); err != nil { - log.Errorf("claim %s: failed to get CPUs %s back: %v", uid, c.ExclusiveCPUs(), err) + for i, uid := range uids { + if err := p.reserveClaim(next[uid]); err != nil { + log.Errorf("claim %s: failed to move from CPUs %s to %s: %v", + uid, p.allocations.claims[uid].ExclusiveCPUs(), next[uid].cpus(), err) + for _, uid := range uids[:i] { + next[uid].Release() + } + for _, uid := range uids { + c := p.allocations.claims[uid] + if _, err := c.GetCPUNode().FreeSupply().Reserve(c.Grant, nil); err != nil { + log.Errorf("claim %s: failed to get CPUs %s back: %v", uid, c.ExclusiveCPUs(), err) + } + } + return false } - p.allocations.claims[uid] = c - log.Infof("moved DRA claim %s back to CPUs %s", uid, c.ExclusiveCPUs()) } + for _, uid := range uids { + log.Infof("moved DRA claim %s from CPUs %s to %s", uid, p.allocations.claims[uid].ExclusiveCPUs(), next[uid].ExclusiveCPUs()) + p.allocations.claims[uid] = next[uid] + } + return true } // repinHolders pins the containers holding the given claims to the CPUs the diff --git a/cmd/plugins/topology-aware/policy/dra_test.go b/cmd/plugins/topology-aware/policy/dra_test.go index 3d850aaae..3e43a6af4 100644 --- a/cmd/plugins/topology-aware/policy/dra_test.go +++ b/cmd/plugins/topology-aware/policy/dra_test.go @@ -145,7 +145,7 @@ func TestDRADevices(t *testing.T) { }{ { // The two nodes with no CPUs of their own publish nothing. - name: "one device per NUMA node and socket with CPUs", + name: "one device per CPU group with CPUs", reserved: "cpuset:0,1", expected: map[string]draDevice{ "cpudevnuma000": {level: "numanode", numaNode: 0, socketID: 0, smt: true, cpus: 27}, @@ -154,6 +154,7 @@ func TestDRADevices(t *testing.T) { "cpudevnuma003": {level: "numanode", numaNode: 3, socketID: 1, smt: true, cpus: 28}, "cpudevsocket000": {level: "socket", numaNode: -1, socketID: 0, smt: true, cpus: 55}, "cpudevsocket001": {level: "socket", numaNode: -1, socketID: 1, smt: true, cpus: 55}, + "cpudevmachine": {level: "machine", numaNode: -1, socketID: -1, smt: true, cpus: 110}, }, }, { @@ -166,6 +167,7 @@ func TestDRADevices(t *testing.T) { "cpudevnuma003": {level: "numanode", numaNode: 3, socketID: 1, smt: true, cpus: 28}, "cpudevsocket000": {level: "socket", numaNode: -1, socketID: 0, smt: true, cpus: 28}, "cpudevsocket001": {level: "socket", numaNode: -1, socketID: 1, smt: true, cpus: 55}, + "cpudevmachine": {level: "machine", numaNode: -1, socketID: -1, smt: true, cpus: 83}, }, }, { @@ -175,6 +177,7 @@ func TestDRADevices(t *testing.T) { expected: map[string]draDevice{ "cpudevnuma000": {level: "numanode", numaNode: 0, socketID: 0, smt: true, cpus: 1}, "cpudevsocket000": {level: "socket", numaNode: -1, socketID: 0, smt: true, cpus: 1}, + "cpudevmachine": {level: "machine", numaNode: -1, socketID: -1, smt: true, cpus: 1}, }, }, { @@ -188,6 +191,7 @@ func TestDRADevices(t *testing.T) { "cpudevnuma003": {level: "numanode", numaNode: 3, socketID: 1, smt: true, cpus: 28}, "cpudevsocket000": {level: "socket", numaNode: -1, socketID: 0, smt: true, cpus: 55}, "cpudevsocket001": {level: "socket", numaNode: -1, socketID: 1, smt: true, cpus: 56}, + "cpudevmachine": {level: "machine", numaNode: -1, socketID: -1, smt: true, cpus: 111}, }, }, } { @@ -227,15 +231,24 @@ func TestDRADevicesFollowFreeCPUs(t *testing.T) { sys := testServerSystem(t) p, owner := draTestPolicy(t, sys, draTestConfig("", draTestReserved(sys))) - expected := map[string]int64{ - "cpudevnuma000": 28, - "cpudevnuma001": 28, - "cpudevnuma002": 28, - "cpudevnuma003": 26, - // Sockets 0 and 1 have nodes 0 and 2, and 1 and 3. + // Sockets 0 and 1 have nodes 0 and 2, and 1 and 3. + idle := map[string]int64{ + "cpudevnuma000": 28, + "cpudevnuma001": 28, + "cpudevnuma002": 28, + "cpudevnuma003": 26, "cpudevsocket000": 56, "cpudevsocket001": 54, + "cpudevmachine": 110, } + expected := maps.Clone(idle) + lower := func(n int64, devices ...string) { + for _, d := range devices { + expected[d] -= n + } + } + socket0 := []string{"cpudevsocket000", "cpudevmachine"} + socket1 := []string{"cpudevsocket001", "cpudevmachine"} check := func(what string) { t.Helper() got := map[string]int64{} @@ -252,26 +265,26 @@ func TestDRADevicesFollowFreeCPUs(t *testing.T) { if err := p.allocateResources(exclusive, "NUMA node #1"); err != nil { t.Fatalf("failed to allocate: %v", err) } - expected["cpudevnuma001"] -= 2 - expected["cpudevsocket001"] -= 2 + lower(2, append(socket1, "cpudevnuma001")...) check("after an exclusive allocation") - // A claim on a node cannot leave it, so the socket device leaves its - // CPUs out. + // A claim on a node cannot leave it, so the devices of its socket leave + // its CPUs out. allocateClaim(t, p, "a", draTestResult(0, "3")) - expected["cpudevsocket000"] -= 3 + lower(3, socket0...) check("after a NUMA claim") - // A claim on a socket lowers no node. + // A claim on a socket lowers no node. It cannot leave the socket, so it + // lowers the machine. allocateClaim(t, p, "b", draTestSocketResult(1, "4")) + lower(4, "cpudevmachine") check("after a socket claim") shared := draTestContainer("shared", v1.PodQOSBurstable, "1500m") if err := p.allocateResources(shared, "NUMA node #2"); err != nil { t.Fatalf("failed to allocate: %v", err) } - expected["cpudevnuma002"] -= 2 - expected["cpudevsocket000"] -= 2 + lower(2, append(socket0, "cpudevnuma002")...) check("after a shared allocation") // A shared container in a socket pool needs CPUs of the socket, but of @@ -280,7 +293,7 @@ func TestDRADevicesFollowFreeCPUs(t *testing.T) { if err := p.allocateResources(socketShared, "socket #1"); err != nil { t.Fatalf("failed to allocate: %v", err) } - expected["cpudevsocket001"]-- + lower(1, socket1...) check("after a shared allocation in a socket pool") for _, c := range []cache.Container{exclusive, shared, socketShared} { @@ -293,10 +306,7 @@ func TestDRADevicesFollowFreeCPUs(t *testing.T) { t.Fatalf("failed to release claim: %v", err) } } - expected["cpudevnuma001"] += 2 - expected["cpudevnuma002"] += 2 - expected["cpudevsocket000"] += 5 - expected["cpudevsocket001"] += 3 + expected = maps.Clone(idle) check("after releasing everything") } @@ -305,7 +315,7 @@ func TestDRADevicesFollowFreeCPUs(t *testing.T) { // left out. func TestDRADevicesNearlyFull(t *testing.T) { // Node 0 and socket 0 have CPUs 8 and 12 for claims, node 1 and socket 1 - // have CPU 1. + // have CPU 1, and the machine has all three. p, owner := draTestPolicy(t, testServerSystem(t), draTestConfig("cpuset:0,1,8,12", "cpuset:0")) if err := p.allocateResources(draTestContainer("c0", v1.PodQOSGuaranteed, "1"), "NUMA node #0"); err != nil { @@ -325,7 +335,8 @@ func TestDRADevicesNearlyFull(t *testing.T) { d.Name, capacity.Value.String(), r.Min.String(), r.Step) } } - if expected := []string{"cpudevnuma000", "cpudevsocket000"}; !slices.Equal(names, expected) { + expected := []string{"cpudevnuma000", "cpudevsocket000", "cpudevmachine"} + if !slices.Equal(names, expected) { t.Errorf("published devices %v, expected %v", names, expected) } } @@ -382,13 +393,44 @@ func TestDRADeviceShape(t *testing.T) { } numa := device("cpudevnuma000", "numanode") numa.Attributes[deviceattribute.StandardDeviceAttributeNUMANode] = resourceapi.DeviceAttribute{IntValue: ptr.To(int64(0))} - expected := []resourceapi.Device{numa, device("cpudevsocket000", "socket")} + // The machine has two sockets, so its device has no IDs. + machine := device("cpudevmachine", "machine") + delete(machine.Attributes, "dra.cpu/socketID") + expected := []resourceapi.Device{numa, device("cpudevsocket000", "socket"), machine} if !apiequality.Semantic.DeepEqual(owner.devices, expected) { t.Errorf("published devices\n%+v\nexpected\n%+v", owner.devices, expected) } } +// TestDRADeviceIDsByLevel checks that the ID attributes of a device follow its +// level, also when groups of different levels have the same CPUs. +func TestDRADeviceIDsByLevel(t *testing.T) { + sys, err := system.DiscoverSystemAt(path.Join(testSysfs(t), "sysfs", "desktop", "sys")) + if err != nil { + t.Fatalf("failed to discover system: %v", err) + } + if len(sys.PackageIDs()) != 1 || len(sys.NodeIDs()) != 1 { + t.Fatalf("expected one socket with one NUMA node, got sockets %v, nodes %v", sys.PackageIDs(), sys.NodeIDs()) + } + _, owner := draTestPolicy(t, sys, draTestConfig("", "cpuset:0")) + + expected := map[string][2]int64{ + "cpudevnuma000": {0, 0}, + "cpudevsocket000": {-1, 0}, + "cpudevmachine": {-1, -1}, + } + summary := draDeviceSummary(owner.devices) + if len(summary) != len(expected) { + t.Errorf("published devices %v, expected %v", summary, expected) + } + for name, ids := range expected { + if got := [2]int64{summary[name].numaNode, summary[name].socketID}; got != ids { + t.Errorf("device %s has numaNode/socketID %v, expected %v", name, got, ids) + } + } +} + // pcieSystem adds PCIe roots to the packages of a test system. type pcieSystem struct { system.System @@ -411,7 +453,8 @@ func (p *pciePackage) PCIeRoots() []string { return p.roots } func TestDRAPCIeRoots(t *testing.T) { sys := &pcieSystem{ System: testServerSystem(t), - roots: map[idset.ID][]string{0: {"pci0000:00", "pci0000:5d"}}, + // Root pci0000:00 has CPUs of both sockets. + roots: map[idset.ID][]string{0: {"pci0000:00", "pci0000:5d"}, 1: {"pci0000:00", "pci0000:80"}}, } p, owner := draTestPolicy(t, sys, draTestConfig("", "cpuset:0,1")) @@ -437,10 +480,29 @@ func TestDRAPCIeRoots(t *testing.T) { "cpudevnuma000": {"pci0000:00", "pci0000:5d"}, "cpudevnuma002": {"pci0000:00", "pci0000:5d"}, "cpudevsocket000": {"pci0000:00", "pci0000:5d"}, + "cpudevnuma001": {"pci0000:00", "pci0000:80"}, + "cpudevnuma003": {"pci0000:00", "pci0000:80"}, + "cpudevsocket001": {"pci0000:00", "pci0000:80"}, + "cpudevmachine": {"pci0000:00", "pci0000:5d", "pci0000:80"}, } if got := published(); !reflect.DeepEqual(got, expected) { t.Errorf("published PCIe roots %v, expected %v", got, expected) } + + // With CPUs for claims on socket 0 only, the machine has the roots of + // socket 0 only. + if err := p.Reconfigure(draTestConfig("cpuset:"+sys.Package(0).CPUSet().String(), "cpuset:0")); err != nil { + t.Fatalf("failed to reconfigure policy: %v", err) + } + expected = map[string][]string{ + "cpudevnuma000": {"pci0000:00", "pci0000:5d"}, + "cpudevnuma002": {"pci0000:00", "pci0000:5d"}, + "cpudevsocket000": {"pci0000:00", "pci0000:5d"}, + "cpudevmachine": {"pci0000:00", "pci0000:5d"}, + } + if got := published(); !reflect.DeepEqual(got, expected) { + t.Errorf("published PCIe roots %v with CPUs of socket 0 only, expected %v", got, expected) + } } // TestDRAPublish checks that policy publishes our devices on startup @@ -487,8 +549,13 @@ func draTestResult(node int, cpus string) resourceapi.DeviceRequestAllocationRes // draTestSocketResult is an allocation result of one socket device. func draTestSocketResult(socket int, cpus string) resourceapi.DeviceRequestAllocationResult { + return draTestDeviceResult(fmt.Sprintf(draSocketDeviceFormat, socket), cpus) +} + +// draTestDeviceResult is an allocation result of the named device. +func draTestDeviceResult(device, cpus string) resourceapi.DeviceRequestAllocationResult { r := draTestResult(0, cpus) - r.Device = fmt.Sprintf(draSocketDeviceFormat, socket) + r.Device = device return r } @@ -2377,6 +2444,25 @@ func TestSocketClaimFitsMemorylessNodeOfDie(t *testing.T) { } } +// A machine claim gets the CPUs of a memoryless NUMA node which has enough +// free CPUs, rather than CPUs of two nodes of a socket with enough. +func TestMachineClaimFitsMemorylessNode(t *testing.T) { + sys := draTestMemorylessSystem(t, 3) + cpus := func(node idset.ID, n int) cpuset.CPUSet { + normal := sys.Node(node).CPUSet().Difference(sys.IsolatedCPUs()) + return cpuset.New(normal.List()[:n]...) + } + // Nodes 0 and 2 of socket 0 have 2 CPUs for claims each. Node 1 of + // socket 1 has 2, node 3 has 4. + reserved := cpus(1, 1) + available := cpus(0, 2).Union(cpus(2, 2), cpus(1, 3), cpus(3, 4)) + p, _ := draTestPolicy(t, sys, draTestConfig("cpuset:"+available.String(), "cpuset:"+reserved.String())) + + if a := allocateClaim(t, p, "a", draTestDeviceResult(draMachineDevice, "4")); !a.Equals(cpus(3, 4)) { + t.Errorf("claim a got CPUs %s, expected the CPUs of node 3, %s", a, cpus(3, 4)) + } +} + // A claim moving on a restart moves a socket claim restored before it, if // the socket claim holds the CPUs it needs. func TestRestoreClaimsMoveSocketClaim(t *testing.T) { @@ -2412,6 +2498,220 @@ func TestRestoreClaimsMoveSocketClaim(t *testing.T) { } } +// A machine claim on the CPUs a socket claim needs moves only as many CPUs +// as the socket claim needs to the other socket. +func TestMoveMachineClaim(t *testing.T) { + sys := testServerSystem(t) + cpus := func(node idset.ID, n int) cpuset.CPUSet { + normal := sys.Node(node).CPUSet().Difference(sys.IsolatedCPUs()) + return cpuset.New(normal.List()[:n]...) + } + // Socket 0 has 8 CPUs for claims, socket 1 has 4. + reserved := cpus(3, 1) + available := cpus(0, 4).Union(cpus(2, 4), cpus(1, 4), reserved) + p, _ := draTestPolicy(t, sys, draTestConfig("cpuset:"+available.String(), "cpuset:"+reserved.String())) + socket := func(id idset.ID) cpuset.CPUSet { return sys.Package(id).CPUSet() } + + // Only socket 0 has 6 CPUs. + m := allocateClaim(t, p, "m", draTestDeviceResult(draMachineDevice, "6")) + if !m.IsSubsetOf(socket(0)) { + t.Fatalf("claim m got CPUs %s, expected CPUs of socket 0", m) + } + + s := allocateClaim(t, p, "s", draTestSocketResult(0, "4")) + m = p.allocations.claims["m"].ExclusiveCPUs() + if s.Size() != 4 || !s.IsSubsetOf(socket(0)) { + t.Errorf("claim s got CPUs %s, expected 4 CPUs of socket 0", s) + } + if m.Size() != 6 || m.Intersection(socket(1)).Size() != 2 || !m.Intersection(s).IsEmpty() { + t.Errorf("claim m holds CPUs %s, expected 4 CPUs of socket 0 and 2 of socket 1, none of claim s", m) + } +} + +// A socket claim which moves for a NUMA claim first moves a machine claim out +// of its socket, if the machine claim holds the CPUs it moves to. If that +// fails, every claim keeps its CPUs. +func TestMoveClaimsInChain(t *testing.T) { + sys := testServerSystem(t) + cpus := func(node idset.ID, n int) cpuset.CPUSet { + normal := sys.Node(node).CPUSet().Difference(sys.IsolatedCPUs()) + return cpuset.New(normal.List()[:n]...) + } + // Nodes 0 and 2 of socket 0 have 4 CPUs for claims each, node 1 of socket + // 1 has 5. + reserved := cpus(3, 1) + available := cpus(0, 4).Union(cpus(2, 4), cpus(1, 5), reserved) + p, _ := draTestPolicy(t, sys, draTestConfig("cpuset:"+available.String(), "cpuset:"+reserved.String())) + nodeOf := func(cpus cpuset.CPUSet) idset.ID { return sys.CPU(idset.ID(cpus.List()[0])).NodeID() } + + // The socket claim fills a node and the machine claim the other node of + // the socket, which has fewer free CPUs than node 1. + s := allocateClaim(t, p, "s", draTestSocketResult(0, "4")) + busy := nodeOf(s) + if !s.Equals(cpus(busy, 4)) { + t.Fatalf("claim s got CPUs %s, expected the CPUs of node %d, %s", s, busy, cpus(busy, 4)) + } + other := 2 - busy + if m := allocateClaim(t, p, "m", draTestDeviceResult(draMachineDevice, "4")); !m.Equals(cpus(other, 4)) { + t.Fatalf("claim m got CPUs %s, expected the CPUs of node %d, %s", m, other, cpus(other, 4)) + } + capacity := func() int { + device := fmt.Sprintf(draNUMADeviceFormat, busy) + cpus, _ := p.draDeviceCPUs(device) + return p.draCapacity(device, cpus) + } + if got := capacity(); got != 4 { + t.Errorf("node %d has capacity %d, expected 4", busy, got) + } + + // Socket 1 has room for only 2 CPUs of the machine claim. + allocateClaim(t, p, "o", draTestResult(1, "3")) + if got := capacity(); got != 2 { + t.Errorf("node %d has capacity %d with room for 2 CPUs in socket 1, expected 2", busy, got) + } + if _, err := p.AllocateClaim(draTestClaim("n"), []resourceapi.DeviceRequestAllocationResult{ + draTestResult(int(busy), "4"), + }); err == nil { + t.Errorf("claim n got CPUs of node %d with too few CPUs to move to", busy) + } + for uid, expected := range map[string]cpuset.CPUSet{"s": cpus(busy, 4), "m": cpus(other, 4)} { + if got := p.allocations.claims[uid].ExclusiveCPUs(); !got.Equals(expected) { + t.Errorf("claim %s holds CPUs %s after a failed claim, expected %s", uid, got, expected) + } + } + + if err := p.ReleaseClaim("o", false); err != nil { + t.Fatalf("failed to release claim o: %v", err) + } + if n := allocateClaim(t, p, "n", draTestResult(int(busy), "4")); !n.Equals(cpus(busy, 4)) { + t.Errorf("claim n got CPUs %s, expected the CPUs of node %d, %s", n, busy, cpus(busy, 4)) + } + for uid, expected := range map[string]cpuset.CPUSet{"s": cpus(other, 4), "m": cpus(1, 5)} { + if got := p.allocations.claims[uid].ExclusiveCPUs(); got.Size() != 4 || !got.IsSubsetOf(expected) { + t.Errorf("claim %s holds CPUs %s, expected 4 CPUs of %s", uid, got, expected) + } + } + + // Undoing the NUMA claim moves both claims back. + if err := p.ReleaseClaim("n", true); err != nil { + t.Fatalf("failed to release claim n: %v", err) + } + for uid, expected := range map[string]cpuset.CPUSet{"s": cpus(busy, 4), "m": cpus(other, 4)} { + if got := p.allocations.claims[uid].ExclusiveCPUs(); !got.Equals(expected) { + t.Errorf("claim %s holds CPUs %s after undoing claim n, expected %s", uid, got, expected) + } + } +} + +// Two claims with socket and machine results can each move onto CPUs the +// other had. Undoing the claim that moved them moves both back together. +func TestUndoMovedClaimsCycle(t *testing.T) { + sys := testServerSystem(t) + p, _ := draTestPolicy(t, sys, draTestConfig("", draTestReserved(sys))) + + allocateClaim(t, p, "n", draTestResult(0, "2"), draTestResult(1, "2")) + allocateClaim(t, p, "a", draTestSocketResult(0, "2"), draTestDeviceResult(draMachineDevice, "2")) + allocateClaim(t, p, "b", draTestSocketResult(1, "2"), draTestDeviceResult(draMachineDevice, "2")) + n, a, b := p.allocations.claims["n"], p.allocations.claims["a"], p.allocations.claims["b"] + + // The socket result of each claim had the CPUs of n on its socket, and + // its machine result had the CPUs the socket result of the other claim + // holds now. + aFrom := &claim{results: []cpuset.CPUSet{n.results[0], b.results[0]}, devices: a.devices} + bFrom := &claim{results: []cpuset.CPUSet{n.results[1], a.results[0]}, devices: b.devices} + n.moves = map[string]claimMove{"a": {from: aFrom, to: a}, "b": {from: bFrom, to: b}} + + if err := p.ReleaseClaim("n", true); err != nil { + t.Fatalf("failed to release claim n: %v", err) + } + for uid, expected := range map[string]cpuset.CPUSet{"a": aFrom.cpus(), "b": bFrom.cpus()} { + if got := p.allocations.claims[uid].ExclusiveCPUs(); !got.Equals(expected) { + t.Errorf("claim %s holds CPUs %s after undoing claim n, expected %s", uid, got, expected) + } + } +} + +// A moved claim which can move back only after another one frees a CPU for +// the shared containers moves back together with it. +func TestUndoMovedClaimsSharedLoad(t *testing.T) { + sys := testServerSystem(t) + // Node 0 has isolated CPU 4 and normal CPUs 0 and 8, node 1 has CPU 9. + p, _ := draTestPolicy(t, sys, draTestConfig("cpuset:0-1,4,8-9", "cpuset:1")) + allocateShared(t, p, "shared", "900m", p.poolForCPUs(cpuset.New(0, 8)).Name()) + reserve := func(uid string, cpus cpuset.CPUSet, device string) *claim { + c := &claim{results: []cpuset.CPUSet{cpus}, devices: []string{device}} + if err := p.reserveClaim(c); err != nil { + t.Fatalf("failed to reserve CPUs %s for claim %s: %v", cpus, uid, err) + } + p.allocations.claims[uid] = c + return c + } + m := reserve("m", cpuset.New(9), draMachineDevice) + s := reserve("s", cpuset.New(8), fmt.Sprintf(draSocketDeviceFormat, 0)) + n := reserve("n", cpuset.New(4), fmt.Sprintf(draNUMADeviceFormat, 0)) + + // m had CPU 0, which the shared containers need while s holds CPU 8. s + // had the isolated CPU 4. + mFrom := &claim{results: []cpuset.CPUSet{cpuset.New(0)}, devices: m.devices} + sFrom := &claim{results: []cpuset.CPUSet{cpuset.New(4)}, devices: s.devices} + n.moves = map[string]claimMove{"m": {from: mFrom, to: m}, "s": {from: sFrom, to: s}} + + if err := p.ReleaseClaim("n", true); err != nil { + t.Fatalf("failed to release claim n: %v", err) + } + for uid, expected := range map[string]cpuset.CPUSet{"m": mFrom.cpus(), "s": sFrom.cpus()} { + if got := p.allocations.claims[uid].ExclusiveCPUs(); !got.Equals(expected) { + t.Errorf("claim %s holds CPUs %s after undoing claim n, expected %s", uid, got, expected) + } + } +} + +// A socket claim which moves for a NUMA claim moves a machine claim out of its +// socket also when free CPUs of the socket are left, if the shared containers +// of the socket pool need them. +func TestMoveClaimsInChainSharedLoad(t *testing.T) { + sys := testServerSystem(t) + cpus := func(node idset.ID, n int) cpuset.CPUSet { + normal := sys.Node(node).CPUSet().Difference(sys.IsolatedCPUs()) + return cpuset.New(normal.List()[:n]...) + } + // Nodes 0 and 2 of socket 0 have 2 CPUs for claims each, node 1 of socket + // 1 has 4. + reserved := cpus(3, 1) + available := cpus(0, 2).Union(cpus(2, 2), cpus(1, 4), reserved) + p, _ := draTestPolicy(t, sys, draTestConfig("cpuset:"+available.String(), "cpuset:"+reserved.String())) + nodeOf := func(cpus cpuset.CPUSet) idset.ID { return sys.CPU(idset.ID(cpus.List()[0])).NodeID() } + + // Each node of socket 0 has one free CPU, the shared containers need one + // of them. + s := allocateClaim(t, p, "s", draTestSocketResult(0, "1")) + busy := nodeOf(s) + other := 2 - busy + allocateClaim(t, p, "full", draTestResult(int(busy), "1")) + if m := allocateClaim(t, p, "m", draTestDeviceResult(draMachineDevice, "1")); nodeOf(m) != other { + t.Fatalf("claim m got CPUs %s, expected a CPU of node %d", m, other) + } + if err := p.ReleaseClaim("full", false); err != nil { + t.Fatalf("failed to release claim full: %v", err) + } + allocateShared(t, p, "shared", "900m", "socket #0") + + device := fmt.Sprintf(draNUMADeviceFormat, busy) + deviceCPUs, _ := p.draDeviceCPUs(device) + if got := p.draCapacity(device, deviceCPUs); got != 2 { + t.Errorf("node %d has capacity %d, expected 2", busy, got) + } + if n := allocateClaim(t, p, "n", draTestResult(int(busy), "2")); !n.Equals(cpus(busy, 2)) { + t.Errorf("claim n got CPUs %s, expected the CPUs of node %d, %s", n, busy, cpus(busy, 2)) + } + if got := p.allocations.claims["s"].ExclusiveCPUs(); nodeOf(got) != other { + t.Errorf("claim s holds CPUs %s, expected a CPU of node %d", got, other) + } + if got := p.allocations.claims["m"].ExclusiveCPUs(); !got.IsSubsetOf(cpus(1, 4)) { + t.Errorf("claim m holds CPUs %s, expected a CPU of node 1, %s", got, cpus(1, 4)) + } +} + // treeSystem is a test system with two L3 caches per NUMA node, each shared // by half of the cores of the node. type treeSystem struct { @@ -2499,6 +2799,7 @@ func TestDRADevicesSharedPools(t *testing.T) { expected["cpudevnuma000"]-- expected["cpudevnuma002"]-- expected["cpudevsocket000"] -= 2 + expected["cpudevmachine"] -= 2 if got := capacities(); !reflect.DeepEqual(got, expected) { t.Errorf("published capacities %v, expected %v", got, expected) } @@ -2518,6 +2819,7 @@ func TestDRADeviceCPUs(t *testing.T) { {device: "cpudevnuma003", cpus: sys.Node(3).CPUSet(), ok: true}, {device: "cpudevsocket000", cpus: sys.Package(0).CPUSet().Difference(cpuset.New(0)), ok: true}, {device: "cpudevsocket001", cpus: sys.Package(1).CPUSet(), ok: true}, + {device: "cpudevmachine", cpus: sys.CPUSet().Difference(cpuset.New(0)), ok: true}, {device: "cpudevsocket0"}, {device: "cpudevsocket1000"}, {device: "cpudevsocket002"}, diff --git a/cmd/plugins/topology-aware/policy/topology-aware-policy.go b/cmd/plugins/topology-aware/policy/topology-aware-policy.go index 8abdbbd54..107250a45 100644 --- a/cmd/plugins/topology-aware/policy/topology-aware-policy.go +++ b/cmd/plugins/topology-aware/policy/topology-aware-policy.go @@ -59,6 +59,7 @@ type policy struct { cfg *cfgapi.Config cache cache.Cache // pod/container cache sys system.System // system/HW topology info + groups []draCPUGroup // groups of CPUs we publish as DRA devices allowed cpuset.CPUSet // bounding set of CPUs we're allowed to use reserved cpuset.CPUSet // system-/kube-reserved CPUs reserveCnt int // number of CPUs to reserve if given as resource.Quantity @@ -106,6 +107,7 @@ func (p *policy) Setup(opts *policyapi.BackendOptions) error { p.cfg = cfg p.cache = opts.Cache p.sys = opts.System + p.groups = draCPUGroups(opts.System) p.options = opts p.cpuAllocator = cpuallocator.NewCPUAllocator(opts.System) p.memAllocator, err = libmem.NewAllocator(libmem.WithSystemNodes(opts.System)) diff --git a/docs/resource-policy/policy/topology-aware.md b/docs/resource-policy/policy/topology-aware.md index 761eeaa9e..f1546d50a 100644 --- a/docs/resource-policy/policy/topology-aware.md +++ b/docs/resource-policy/policy/topology-aware.md @@ -1451,33 +1451,46 @@ capacity, and the policy runs the containers of the pod on the claimed CPUs. ### Devices -The policy publishes one device per NUMA node, named `cpudevnuma000`, -`cpudevnuma001` and so on, and one device per socket, named `cpudevsocket000`, -`cpudevsocket001` and so on. A socket device offers the CPUs of all NUMA nodes -of the socket. A socket with one NUMA node has a socket device with the same -CPUs as its NUMA device. +The policy publishes a device for each group of CPUs of the topology: + +| Device | Level | CPUs | +|---|---|---| +| `cpudevmachine` | `machine` | all CPUs | +| `cpudevsocket000`, `cpudevsocket001`, ... | `socket` | the CPUs of a socket | +| `cpudevnuma000`, `cpudevnuma001`, ... | `numanode` | the CPUs of a NUMA node | + +Every level is published, also when it adds no group of its own: a machine +with one socket has a socket device with the same CPUs as its machine device, +and a socket with one NUMA node has a NUMA device with the same CPUs as its +socket device. So a claim written for one level works on every machine. The capacity `dra.cpu/cpu` of a device is the number of its CPUs which claims -hold or can get: the CPUs of the node or socket, minus the reserved CPUs, -minus the CPUs containers hold exclusively, minus the whole CPUs the shared -containers need. The capacity of a socket device also leaves out the CPUs of -claims on its NUMA nodes. The scheduler subtracts what claims consume of a -device, so the CPUs of its own claims stay in its capacity. The capacity of a -NUMA device also has the CPUs socket claims hold on the node, as many as these -claims can move to other CPUs of the socket. The policy publishes the capacity -again whenever it changes. A node or socket with no CPUs left for claims has -no device until some are free again. A new capacity applies to claims -allocated after it: claims already allocated keep their CPUs. The devices have -these attributes: +hold or can get: the CPUs of the group, minus the reserved CPUs, minus the +CPUs containers hold exclusively, minus the whole CPUs the shared containers +need. The capacity of a device also leaves out the CPUs of claims on devices +which are not larger than it, such as the claims on the NUMA nodes of a socket +device. The scheduler subtracts what claims consume of a device, so the CPUs +of its own claims stay in its capacity. The capacity of a device also has the +CPUs which claims on larger devices hold in it, as many as these claims can +move to other CPUs of their device, also by moving claims on still larger +devices out of the way. The policy publishes the capacity again whenever it +changes. A group with no CPUs left for claims has no device until some are +free again. A new capacity applies to claims allocated after it: claims +already allocated keep their CPUs. The devices have these attributes: | Attribute | Type | Value | |---|---|---| -| `dra.cpu/level` | string | `numanode` or `socket` | -| `resource.kubernetes.io/numaNode` | int | the ID of the NUMA node, on NUMA devices only | -| `dra.cpu/socketID` | int | the ID of the socket | +| `dra.cpu/level` | string | `machine`, `socket` or `numanode` | +| `dra.cpu/socketID` | int | the ID of the socket, on NUMA and socket devices | +| `resource.kubernetes.io/numaNode` | int | the ID of the NUMA node, on NUMA devices | | `dra.cpu/smtEnabled` | bool | whether the CPUs have hyperthreading enabled | | `dra.cpu/numCPUs` | int | the number of CPUs of the device, without the reserved CPUs | -| `resource.kubernetes.io/pcieRoot` | list of strings | the PCIe roots of the socket, with `DRAListTypeAttributes` only | +| `resource.kubernetes.io/pcieRoot` | list of strings | the PCIe roots of the sockets of the CPUs, with `DRAListTypeAttributes` only | + +Which IDs a device has depends only on its level, so a selector works the +same way on every machine. A socket device has no `numaNode`, even when its +socket has one NUMA node, and the machine device has no IDs, even on a machine +with one socket. With sub-NUMA clustering, the kernel gives each PCIe root the CPUs of one NUMA node, but the root serves the whole socket. So every device of a socket @@ -1487,28 +1500,31 @@ selects the node with `numaNode` as well. A request consumes a number of whole CPUs of one device. A request without an amount gets one CPU, not the whole device. The policy picks the CPUs by topology: two CPUs are the two threads of one core when possible. A request -on a socket device gets CPUs of one NUMA node when a node has enough free -CPUs, otherwise CPUs of several nodes of the socket. - -A claim on a socket device is not bound to its CPUs. When a claim on a NUMA -node of the socket needs them, the claim on the socket device moves to other -CPUs of the socket. Its containers move with it, and find the new CPUs in the -`CLAIMED_CPUS` variable described below. Their memory stays on the nodes it -was on. +gets CPUs of the smallest group inside its device which has enough free CPUs. +For instance, a request on the machine device gets CPUs of one NUMA node when +a node has enough free CPUs, otherwise CPUs of one socket, otherwise CPUs of +several sockets. + +A claim is not bound to its CPUs. When a claim on a smaller device inside its +device needs them, for instance a claim on a NUMA node of a socket, the claim +moves to other CPUs of its device. If claims on larger devices hold these +CPUs, they move out of its device first. The containers of a claim move with +it, and find the new CPUs in the `CLAIMED_CPUS` variable described below. +Their memory stays on the nodes it was on. ### Selecting Devices -The `topology-aware.nri.io` device class matches the NUMA devices and the -socket devices. A selector which reads an attribute that a device does not +The `topology-aware.nri.io` device class matches the devices of every level. +A selector which reads an attribute that a device does not have fails the allocation of the claim on the node, instead of skipping the device. A selector on `numaNode` therefore first checks the level, for instance `device.attributes["dra.cpu"].level == "numanode" && device.attributes["resource.kubernetes.io"].numaNode == 1`. Claims written for earlier versions of the plugin, which had NUMA devices -only, select `numaNode` without this check, and need it now. A selector on -`socketID` alone, such as `device.attributes["dra.cpu"].socketID == 0`, -matches every device of the socket, and the scheduler picks a NUMA device or -the socket device. +only, select `numaNode` without this check, and need it now. The machine +device has no `socketID`, so a selector on `socketID` checks the level too: +`device.attributes["dra.cpu"].level != "machine" && device.attributes["dra.cpu"].socketID == 0` +matches every device inside socket 0, and the scheduler picks one of them. ### Claims @@ -1664,9 +1680,9 @@ why. Claims keep their CPUs when the plugin restarts. If the plugin restarts with a configuration which cannot give a claim its CPUs, for example one which reserves some of them, the claim moves to other CPUs of the same device, and -its containers move with it. If the claim needs CPUs of a socket claim, the -socket claim moves to other CPUs of its socket. If the device has too few CPUs -left for the claim, the plugin fails to start, and the status of the +its containers move with it. If the claim needs CPUs of a claim on a larger +device, that claim moves to other CPUs of its device. If the device has too +few CPUs left for the claim, the plugin fails to start, and the status of the configuration names the claim. Revert the configuration to start the plugin again. diff --git a/test/e2e/policies.test-suite/topology-aware/n4c16/test42-dra-devices/code.var.sh b/test/e2e/policies.test-suite/topology-aware/n4c16/test42-dra-devices/code.var.sh index 3c69dd305..2b9f7080a 100644 --- a/test/e2e/policies.test-suite/topology-aware/n4c16/test42-dra-devices/code.var.sh +++ b/test/e2e/policies.test-suite/topology-aware/n4c16/test42-dra-devices/code.var.sh @@ -131,10 +131,10 @@ vm-command "kubectl get resourceslices -o yaml" slices=$(vm-command-q "kubectl get resourceslices -o json") -# One device per NUMA node and one per socket. +# One device per NUMA node, one per socket and one for the machine. count=$(jq '[.items[] | select(.spec.driver == "topology-aware.nri.io") | .spec.devices[]] | length' <<< "$slices") -[ "$count" == "6" ] || - error "published $count devices, expected 6" +[ "$count" == "7" ] || + error "published $count devices, expected 7" for node in 0 1 2 3; do name=$(printf "cpudevnuma%03d" "$node") diff --git a/test/e2e/policies.test-suite/topology-aware/n4c16/test48-dra-socket-devices/code.var.sh b/test/e2e/policies.test-suite/topology-aware/n4c16/test48-dra-socket-devices/code.var.sh index 130e4e82d..88788c281 100644 --- a/test/e2e/policies.test-suite/topology-aware/n4c16/test48-dra-socket-devices/code.var.sh +++ b/test/e2e/policies.test-suite/topology-aware/n4c16/test48-dra-socket-devices/code.var.sh @@ -3,7 +3,8 @@ # CPUs of the socket, and CPUs of one NUMA node when it fits in one. The # capacity of the devices follows the CPUs which containers and claims use. # A socket claim moves to other CPUs of its socket when a NUMA node claim -# needs its CPUs. +# needs its CPUs, and a machine claim moves out of the socket when the socket +# claim needs its CPUs. # # n4c16 has two sockets of two NUMA nodes with four CPUs each: socket 0 has # CPUs 0-7, socket 1 CPUs 8-15. @@ -56,6 +57,22 @@ exported() { sed -n "s/^$2=\"\(.*\)\"$/\1/p" } +lowered() { + # Usage: lowered N DEVICE... + # + # Print the idle capacities with those of DEVICEs lowered by N. + local n=$1 entry name cpus expected="" + shift + for entry in $idle; do + name=${entry%=*} cpus=${entry#*=} + if [[ " $* " == *" $name "* ]]; then + cpus=$((cpus - n)) + fi + expected+="${expected:+ }$name=$cpus" + done + echo "$expected" +} + cpu-ids() { # Usage: cpu-ids CPUSET # @@ -94,6 +111,15 @@ numa-claim() { NAME=$1 create claim-pod } +machine-claim() { + # Usage: machine-claim POD CPUS + # + # Create POD with a claim for CPUS CPUs of the machine, and wait for it to + # start. + NAME=$1 CPUS=$2 wait="" create machine-claim + NAME=$1 create claim-pod +} + cleanup # Reserve CPU 15, so that socket 1 has one CPU less to give than socket 0. @@ -114,7 +140,8 @@ fi vm-command "kubectl get resourceslices -o yaml" -idle="cpudevnuma000=4 cpudevnuma001=4 cpudevnuma002=4 cpudevnuma003=3 cpudevsocket000=8 cpudevsocket001=7" +# Sorted by name, as capacities prints them. +idle="cpudevmachine=15 cpudevnuma000=4 cpudevnuma001=4 cpudevnuma002=4 cpudevnuma003=3 cpudevsocket000=8 cpudevsocket001=7" wait-capacities "$idle" for socket in 0 1; do @@ -138,6 +165,7 @@ done # subtracts the claim from the socket device. A NUMA device keeps the CPUs of # the claim in its capacity only as far as the claim can move off them, to the # one free CPU of the socket, so both NUMA devices of socket 1 have 1 CPU. +# The claim cannot leave the socket, so the machine device leaves its CPUs out. socket-claim pod0 1 6 cpus0=$(exported pod0 CLAIMED_CPUS) [ "$(cpu-ids "$cpus0" | wc -l)" == "6" ] && [ "$(cpu-nodes "$cpus0")" == "2 3 " ] || @@ -145,7 +173,8 @@ cpus0=$(exported pod0 CLAIMED_CPUS) pinned=$(vm-command-q "kubectl exec pod0 -- cat /sys/fs/cgroup/cpuset.cpus.effective") [ "$pinned" == "$cpus0" ] || error "pod0 runs on CPUs $pinned, expected its claimed CPUs $cpus0" -wait-capacities "${idle/cpudevnuma002=4 cpudevnuma003=3/cpudevnuma002=1 cpudevnuma003=1}" +expected=$(lowered 6 cpudevmachine) +wait-capacities "${expected/cpudevnuma002=4 cpudevnuma003=3/cpudevnuma002=1 cpudevnuma003=1}" # Deleting pod0 gives the CPUs of its claim back: all 7 CPUs of socket 1 can # be claimed again. @@ -165,8 +194,8 @@ node2=$(cpu-nodes "$cpus2") node2=${node2% } vm-command "kubectl delete pod pod2 --now --wait" -# A container with 2 exclusive CPUs lowers the capacity of the NUMA device and -# of the socket device of its CPUs by 2. +# A container with 2 exclusive CPUs lowers the capacity of every device of its +# CPUs by 2. NAME=pod3 CPU=2 create guaranteed cpus3=$(exported pod3 EXCLUSIVE_CPUS) [ -n "$cpus3" ] || cpus3=$(exported pod3 ISOLATED_CPUS) @@ -174,16 +203,8 @@ node3=$(cpu-nodes "$cpus3") [ "$(cpu-ids "$cpus3" | wc -l)" == "2" ] && [ "$(wc -w <<< "$node3")" == "1" ] || error "pod3 got exclusive CPUs \"$cpus3\", expected 2 CPUs of one NUMA node" node3=${node3% } -expected="" -for entry in $idle; do - name=${entry%=*} cpus=${entry#*=} - if [ "$name" == "$(printf "cpudevnuma%03d" "$node3")" ] || - [ "$name" == "$(printf "cpudevsocket%03d" $((node3 / 2)))" ]; then - cpus=$((cpus - 2)) - fi - expected+="${expected:+ }$name=$cpus" -done -wait-capacities "$expected" +wait-capacities "$(lowered 2 "$(printf "cpudevnuma%03d" "$node3")" "$(printf "cpudevsocket%03d" $((node3 / 2)))" \ + cpudevmachine)" vm-command "kubectl delete pod pod3 --now --wait" wait-capacities "$idle" @@ -208,6 +229,47 @@ cpus4=$(exported pod4 CLAIMED_CPUS) pinned=$(vm-command-q "kubectl exec pod4 -- cat /sys/fs/cgroup/cpuset.cpus.effective") [ "$pinned" == "$cpus4" ] || error "pod4 runs on CPUs $pinned, expected its claimed CPUs $cpus4" +vm-command "kubectl delete pods pod4 pod5 --now --wait" +wait-capacities "$idle" + +# A claim on the machine device gets CPUs of both sockets when it does not fit +# in one. +machine-claim pod6 10 +cpus6=$(exported pod6 CLAIMED_CPUS) +nodes6=$(cpu-nodes "$cpus6") +[ "$(cpu-ids "$cpus6" | wc -l)" == "10" ] && [[ "$nodes6" =~ [01] && "$nodes6" =~ [23] ]] || + error "pod6 got claimed CPUs \"$cpus6\", expected 10 CPUs of both sockets" +vm-command "kubectl delete pod pod6 --now --wait" + +# pod7 takes all CPUs of socket 1 and pod8 4 CPUs of socket 0, so a machine +# claim of pod9 gets the other NUMA node of socket 0. When pod7 is gone and +# pod10 claims the NUMA node of pod8, the claim of pod9 moves to socket 1 and +# the claim of pod8 to the CPUs pod9 left. +socket-claim pod7 1 7 +socket-claim pod8 0 4 +node8=$(cpu-nodes "$(exported pod8 CLAIMED_CPUS)") +node8=${node8% } +node9=$((1 - node8)) +machine-claim pod9 4 +[ "$(cpu-nodes "$(exported pod9 CLAIMED_CPUS)")" == "$node9 " ] || + error "pod9 got claimed CPUs \"$(exported pod9 CLAIMED_CPUS)\", expected the CPUs of NUMA node $node9" +vm-command "kubectl delete pod pod7 --now --wait" +numa-claim pod10 "$node8" 4 +[ "$(cpu-nodes "$(exported pod10 CLAIMED_CPUS)")" == "$node8 " ] || + error "pod10 got claimed CPUs \"$(exported pod10 CLAIMED_CPUS)\", expected the CPUs of NUMA node $node8" +retry-until --timeout 30 --message "the claim of pod8 to move to NUMA node $node9" \ + '[ "$(cpu-nodes "$(exported pod8 CLAIMED_CPUS)")" == "$node9 " ]' || + error "pod8 has claimed CPUs \"$(exported pod8 CLAIMED_CPUS)\", expected the CPUs of NUMA node $node9" +socket1='^(2 |3 |2 3 )$' +retry-until --timeout 30 --message "the claim of pod9 to move to socket 1" \ + '[[ "$(cpu-nodes "$(exported pod9 CLAIMED_CPUS)")" =~ $socket1 ]]' || + error "pod9 has claimed CPUs \"$(exported pod9 CLAIMED_CPUS)\", expected CPUs of socket 1" +for pod in pod8 pod9; do + cpus=$(exported $pod CLAIMED_CPUS) + pinned=$(vm-command-q "kubectl exec $pod -- cat /sys/fs/cgroup/cpuset.cpus.effective") + [ "$pinned" == "$cpus" ] || + error "$pod runs on CPUs $pinned, expected its claimed CPUs $cpus" +done cleanup -echo "OK: socket devices next to the NUMA devices, socket claims within their socket, a 3-CPU claim on NUMA node $node2, a socket claim moved to NUMA node $other" +echo "OK: socket devices next to the NUMA devices, socket claims within their socket, a 3-CPU claim on NUMA node $node2, a socket claim moved to NUMA node $other, a machine claim on both sockets, a machine claim moved out of the way of a moving socket claim" diff --git a/test/e2e/policies.test-suite/topology-aware/n4c16/test48-dra-socket-devices/machine-claim.yaml.in b/test/e2e/policies.test-suite/topology-aware/n4c16/test48-dra-socket-devices/machine-claim.yaml.in new file mode 100644 index 000000000..7fa3ff9f6 --- /dev/null +++ b/test/e2e/policies.test-suite/topology-aware/n4c16/test48-dra-socket-devices/machine-claim.yaml.in @@ -0,0 +1,17 @@ +apiVersion: resource.k8s.io/v1 +kind: ResourceClaimTemplate +metadata: + name: ${NAME} +spec: + spec: + devices: + requests: + - name: cpus + exactly: + deviceClassName: topology-aware.nri.io + selectors: + - cel: + expression: device.attributes[\"dra.cpu\"].level == \"machine\" + capacity: + requests: + dra.cpu/cpu: '${CPUS}'