Skip to content

Commit 3452ead

Browse files
authored
fixing the awsneuron: derive core geometry per node (Project-HAMi#3029)
Signed-off-by: asadjan4611 <asadjan4611@gmail.com>
1 parent e07aea3 commit 3452ead

4 files changed

Lines changed: 367 additions & 92 deletions

File tree

‎pkg/device/awsneuron/device.go‎

Lines changed: 40 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ package awsneuron
1919
import (
2020
"cmp"
2121
"fmt"
22+
"maps"
2223
"math"
2324
"slices"
2425
"strconv"
@@ -37,8 +38,6 @@ import (
3738
type AWSNeuronDevices struct {
3839
resourceCountName string
3940
resourceCoreName string
40-
coresPerAWSNeuron uint
41-
coremask uint
4241
}
4342

4443
const (
@@ -54,6 +53,7 @@ const (
5453
AWSNeuronAllocated = "NEURON_ALLOCATED"
5554
AWSUsageInfo = "awsusageinfo"
5655
AWSNodeType = "AWSNodeType"
56+
AWSCoresPerNeuronDevice = "AWSCoresPerNeuronDevice"
5757
maxAWSNeuronDeviceCount = int64(math.MaxInt32)
5858
// maxCoresPerNeuronDevice caps per-device cores: addCoreUsage builds a
5959
// two-bit mask and PatchAnnotations only emits the first two core indexes.
@@ -74,8 +74,6 @@ func InitAWSNeuronDevice(config AWSNeuronConfig) *AWSNeuronDevices {
7474
return &AWSNeuronDevices{
7575
resourceCountName: config.ResourceCountName,
7676
resourceCoreName: config.ResourceCoreName,
77-
coresPerAWSNeuron: 0,
78-
coremask: 0,
7977
}
8078
}
8179

@@ -127,20 +125,10 @@ func validateResourceRequest(quantity resource.Quantity, resourceName string, ma
127125
return value, nil
128126
}
129127

130-
// coresPerDevice reports NeuronCores per device. It is zero until a node is
131-
// registered, so callers without one fall back to the cap.
132-
func (dev *AWSNeuronDevices) coresPerDevice() int64 {
133-
observed := int64(dev.coresPerAWSNeuron)
134-
if observed <= 0 {
135-
return maxCoresPerNeuronDevice
136-
}
137-
return min(observed, maxCoresPerNeuronDevice)
138-
}
139-
140128
// splitCoreRequest maps a NeuronCore count onto devices. One core count applies
141129
// to every device, so counts that neither fit nor fill devices are rejected.
142130
func (dev *AWSNeuronDevices) splitCoreRequest(cores int64) (int32, int32, error) {
143-
coresPerDevice := dev.coresPerDevice()
131+
coresPerDevice := maxCoresPerNeuronDevice
144132
if cores <= coresPerDevice {
145133
return 1, int32(cores), nil
146134
}
@@ -158,32 +146,41 @@ func (dev *AWSNeuronDevices) GetNodeDevices(n corev1.Node) ([]*device.DeviceInfo
158146
nodedevices := []*device.DeviceInfo{}
159147
i := 0
160148
counts, ok := n.Status.Capacity.Name(corev1.ResourceName(dev.resourceCountName), resource.DecimalSI).AsInt64()
161-
if !ok || counts == 0 {
149+
if !ok || counts <= 0 || counts > maxAWSNeuronDeviceCount {
162150
return []*device.DeviceInfo{}, fmt.Errorf("device not found %s", dev.resourceCountName)
163151
}
164-
coresTotal, _ := n.Status.Capacity.Name(corev1.ResourceName(dev.resourceCoreName), resource.DecimalSI).AsInt64()
165-
if dev.coresPerAWSNeuron == 0 {
166-
dev.coresPerAWSNeuron = uint(coresTotal) / uint(counts)
152+
coresTotal, ok := n.Status.Capacity.Name(corev1.ResourceName(dev.resourceCoreName), resource.DecimalSI).AsInt64()
153+
if !ok || coresTotal <= 0 {
154+
return []*device.DeviceInfo{}, fmt.Errorf("device not found %s", dev.resourceCoreName)
155+
}
156+
if coresTotal%counts != 0 {
157+
return []*device.DeviceInfo{}, fmt.Errorf("%s capacity %d is not divisible by %s capacity %d", dev.resourceCoreName, coresTotal, dev.resourceCountName, counts)
158+
}
159+
coresPerDevice := coresTotal / counts
160+
if coresPerDevice > math.MaxInt32 {
161+
return []*device.DeviceInfo{}, fmt.Errorf("cores per AWS Neuron device %d exceeds the maximum of %d", coresPerDevice, int64(math.MaxInt32))
167162
}
168163
// The mask must stay within the addressable cores addCoreUsage can track,
169164
// even when the hardware exposes more cores per device (inf1 has four).
170-
dev.coremask = 0
171-
for i < int(dev.coresPerDevice()) {
172-
dev.coremask *= 2
173-
dev.coremask++
165+
coremask := int32(0)
166+
for i < int(min(coresPerDevice, maxCoresPerNeuronDevice)) {
167+
coremask *= 2
168+
coremask++
174169
i++
175170
}
176171
i = 0
177-
customInfo := map[string]any{}
178-
customInfo[AWSNodeType] = n.Labels["node.kubernetes.io/instance-type"]
172+
customInfo := map[string]any{
173+
AWSNodeType: n.Labels["node.kubernetes.io/instance-type"],
174+
AWSCoresPerNeuronDevice: int32(coresPerDevice),
175+
}
179176

180177
for int64(i) < counts {
181178
nodedevices = append(nodedevices, &device.DeviceInfo{
182179
Index: uint(i),
183180
ID: n.Name + "-" + AWSNeuronDevice + "-" + fmt.Sprint(i),
184-
Count: int32(dev.coresPerAWSNeuron),
181+
Count: int32(coresPerDevice),
185182
Devmem: 0,
186-
Devcore: int32(dev.coremask),
183+
Devcore: coremask,
187184
Type: AWSNeuronDevice,
188185
Numa: 0,
189186
Health: true,
@@ -200,6 +197,13 @@ func (dev *AWSNeuronDevices) GetNodeDevices(n corev1.Node) ([]*device.DeviceInfo
200197
return nodedevices, nil
201198
}
202199

200+
func coresPerNeuronDevice(customInfo map[string]any) int32 {
201+
if value, ok := customInfo[AWSCoresPerNeuronDevice].(int32); ok && value > 0 {
202+
return value
203+
}
204+
return int32(maxCoresPerNeuronDevice)
205+
}
206+
203207
func (dev *AWSNeuronDevices) PatchAnnotations(pod *corev1.Pod, annoinput *map[string]string, pd device.PodDevices) map[string]string {
204208
devlist, ok := pd[AWSNeuronDevice]
205209
if ok && len(devlist) > 0 {
@@ -219,12 +223,14 @@ func (dev *AWSNeuronDevices) PatchAnnotations(pod *corev1.Pod, annoinput *map[st
219223
value = value + fmt.Sprint(val.Idx) + ","
220224
}
221225
} else {
226+
coresPerDevice := coresPerNeuronDevice(val.CustomInfo)
227+
coreOffset := int64(val.Idx) * int64(coresPerDevice)
222228
if (val.Usedcores & 1) != 0 {
223-
value = value + fmt.Sprint(dev.coresPerAWSNeuron*uint(val.Idx)) + ","
229+
value = value + fmt.Sprint(coreOffset) + ","
224230
(*annoinput)[AWSNeuronResourceType] = dev.resourceCoreName
225231
}
226232
if (val.Usedcores & 2) != 0 {
227-
value = value + fmt.Sprint(dev.coresPerAWSNeuron*uint(val.Idx)+1) + ","
233+
value = value + fmt.Sprint(coreOffset+1) + ","
228234
(*annoinput)[AWSNeuronResourceType] = dev.resourceCoreName
229235
}
230236
}
@@ -306,7 +312,7 @@ func (dev *AWSNeuronDevices) GenerateResourceRequests(ctr *corev1.Container) dev
306312
Type: AWSNeuronDevice,
307313
Memreq: 0,
308314
MemPercentagereq: 0,
309-
Coresreq: int32(dev.coresPerDevice()),
315+
Coresreq: int32(maxCoresPerNeuronDevice),
310316
}
311317
} else {
312318
core, ok := resourceQuantity(ctr, awsResourceCores)
@@ -368,7 +374,10 @@ func countMaskAvailable(mask int32) int32 {
368374
}
369375

370376
func addCoreUsage(prev map[string]any, require int) map[string]any {
371-
res := map[string]any{}
377+
res := maps.Clone(prev)
378+
if res == nil {
379+
res = map[string]any{}
380+
}
372381
count, ok := prev[AWSUsageInfo]
373382
if !ok {
374383
count = 0

0 commit comments

Comments
 (0)