Skip to content

fix(scheduler): serialize Filter device selection to prevent double G… #2559

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
yxxhero wants to merge 2 commits into from
Closed

fix(scheduler): serialize Filter device selection to prevent double G… #2559

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
75e70ca
fix: replace klog.Fatalf with error returns to prevent process crash
yxxhero May 24, 2026
4ac00e1
fix(scheduler): serialize Filter device selection to prevent double G…
yxxhero Aug 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion cmd/scheduler/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,7 +119,9 @@ func start() error {
client.WithTimeout(config.Timeout),
)

config.InitDevices()
if err := config.InitDevices(); err != nil {

@mesutoezdil mesutoezdil Aug 11, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this initdevices error change is not about the filterlock race. this repo closed a past pr for bundling unrelated changes together. should this be its own pr?

return fmt.Errorf("failed to initialize devices: %w", err)
}

var err error
config.HostName, err = os.Hostname()
Expand Down
18 changes: 9 additions & 9 deletions pkg/scheduler/config/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -278,24 +278,25 @@ func validateConfig(config *Config) error {
return fmt.Errorf("all configurations are empty")
}

func InitDevices() {
func InitDevices() error {
if len(device.DevicesMap) > 0 {
klog.Info("Devices are already initialized, skipping initialization")
return
return nil
}
klog.Infof("Loading device configuration from file: %s", configFile)
config, err := LoadConfig(configFile)
if err != nil {
klog.Fatalf("Failed to load device config file %s: %v", configFile, err)
return fmt.Errorf("failed to load device config file %s: %w", configFile, err)
}
klog.Infof("Loaded config: %v", config)
err = InitDevicesWithConfig(config)
if err != nil {
klog.Fatalf("Failed to initialize devices: %v", err)
return fmt.Errorf("failed to initialize devices: %w", err)
}
Comment on lines +281 to 295

@coderabbitai coderabbitai Bot Aug 10, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift

Make failed device initialization atomic.

InitDevicesWithConfig adds successful backends to the global maps before it returns initialization errors. If a later backend fails, a subsequent InitDevices call can see the non-empty map and return nil. InitDefaultDevices can leave the same partial state after its nested call fails. Build the maps locally and publish them only after all initializers succeed, or clear both global maps on failure.

Also applies to: 461-464

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@pkg/scheduler/config/config.go` around lines 281 - 295, Make device
initialization atomic across InitDevicesWithConfig and InitDefaultDevices:
prevent partially initialized backends from remaining in the global device maps
when any initializer fails. Build temporary maps and publish them only after all
initialization succeeds, or clear both global maps on every failure path, so
subsequent InitDevices calls retry instead of incorrectly returning success.

return nil
}

func InitDefaultDevices() {
func InitDefaultDevices() error {
configMapdata := `
nvidia:
resourceCountName: "nvidia.com/gpu"
Expand Down Expand Up @@ -454,14 +455,13 @@ vnpus:
var yamlData Config
err := yaml.Unmarshal([]byte(configMapdata), &yamlData)
if err != nil {
klog.Fatalf("Failed to unmarshal default config: %v", err)
return
return fmt.Errorf("failed to unmarshal default config: %w", err)
}

// Initialize devices with configuration
if err := InitDevicesWithConfig(&yamlData); err != nil {
klog.Fatalf("Failed to initialize devices with default config: %v", err)
return fmt.Errorf("failed to initialize devices with default config: %w", err)
}
return nil
}

func GlobalFlagSet() *flag.FlagSet {
Expand Down
3 changes: 2 additions & 1 deletion pkg/scheduler/config/config_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -479,7 +479,8 @@ func Test_GetDevices(t *testing.T) {
}

func Test_InitDefaultDevices(t *testing.T) {
InitDefaultDevices()
err := InitDefaultDevices()
assert.NilError(t, err, "Expected InitDefaultDevices to succeed")
assert.Assert(t, len(device.DevicesMap) > 0, "Expected devicesMap to be populated")
assert.Assert(t, len(device.DevicesToHandle) > 0, "Expected DevicesToHandle to be populated")
}
Expand Down
2 changes: 1 addition & 1 deletion pkg/scheduler/nodes_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,7 @@ func Test_addNode_ListNodes(t *testing.T) {
err: nil,
},
}
config.InitDefaultDevices()
assert.NilError(t, config.InitDefaultDevices(), "Expected InitDefaultDevices to succeed")
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
m := nodeManager{
Expand Down
100 changes: 68 additions & 32 deletions pkg/scheduler/scheduler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,10 @@ type Scheduler struct {

lock sync.RWMutex
synced bool

// filterLock serializes device selection in Filter so concurrent requests
// cannot reserve the same device (issue #2232).
filterLock sync.Mutex
}

func NewScheduler() *Scheduler {
Expand Down Expand Up @@ -1030,30 +1034,75 @@ func (s *Scheduler) Filter(args extenderv1.ExtenderArgs) (*extenderv1.ExtenderFi
if args.Nodes != nil {
return s.filterSimulation(args, resourceReqs)
}
klog.V(2).InfoS("Choosing live filter path",
"pod", klog.KObj(args.Pod),
"reason", "request does not contain full nodes",
"nodeNamesLen", nodeNamesLen(args.NodeNames))
// selectAndCommitDevice holds filterLock during selection; the annotation
// patch below runs outside it (issue #2232).
selection, err := s.selectAndCommitDevice(args, resourceReqs)
if err != nil {
s.recordScheduleFilterResultEvent(args.Pod, EventReasonFilteringFailed, "", err)
return nil, err
}
if selection.chosen == nil {
s.recordScheduleFilterResultEvent(args.Pod, EventReasonFilteringFailed, "", fmt.Errorf("no available node, %d nodes do not meet", len(*args.NodeNames)))
return &extenderv1.ExtenderFilterResult{
FailedNodes: selection.failedNodes,
}, nil
}
m := selection.chosen
// Patch the annotation outside the lock; roll back the cache on failure.
if err = util.PatchPodAnnotations(args.Pod, selection.annotations); err != nil {
s.recordScheduleFilterResultEvent(args.Pod, EventReasonFilteringFailed, "", err)
if selection.added {
s.quotaManager.RmUsage(args.Pod, selection.effectiveDevices)
}
s.podManager.DelPod(args.Pod)
return nil, err
}
Comment on lines +1055 to +1063

@coderabbitai coderabbitai Bot Aug 10, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win

The rollback path treats the pod cache and the quota inconsistently, and no test binds them together. Filter deletes the pod-cache entry unconditionally but removes the quota usage only when selection.added is true, so the two states can diverge after a patch failure.

  • pkg/scheduler/scheduler.go#L974-L982: gate s.podManager.DelPod(args.Pod) on selection.added, matching the s.quotaManager.RmUsage call, so both reservation effects are undone together.
  • pkg/scheduler/scheduler_test.go#L1918-L1923: add a quota assertion after the inCache check, so the test covers both halves of the rollback.
📍 Affects 2 files
  • pkg/scheduler/scheduler.go#L974-L982 (this comment)
  • pkg/scheduler/scheduler_test.go#L1918-L1923
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@pkg/scheduler/scheduler.go` around lines 974 - 982, In
pkg/scheduler/scheduler.go lines 974-982, update the rollback in Filter’s
PatchPodAnnotations failure path so podManager.DelPod(args.Pod) is gated by
selection.added, matching quotaManager.RmUsage and keeping reservation cleanup
consistent. In pkg/scheduler/scheduler_test.go lines 1918-1923, extend the
existing inCache assertion with a quota assertion that verifies both reservation
effects are rolled back.

successMsg := genSuccessMsg(len(*args.NodeNames), m.NodeID, selection.nodeList)
s.recordScheduleFilterResultEvent(args.Pod, EventReasonFilteringSucceed, successMsg, nil)
res := extenderv1.ExtenderFilterResult{NodeNames: &[]string{m.NodeID}}
return &res, nil
}

// filterSelection holds a Filter reservation; chosen is nil when no node fits
// (failedNodes then carries the per-node reasons).
type filterSelection struct {
chosen *policy.NodeScore
annotations map[string]string
added bool
nodeList []*policy.NodeScore
failedNodes map[string]string
effectiveDevices device.PodDevices
}

// selectAndCommitDevice selects a device and commits it to the cache under
// filterLock. It does not publish the pod annotation; the caller patches it.
func (s *Scheduler) selectAndCommitDevice(args extenderv1.ExtenderArgs, resourceReqs device.PodDeviceRequests) (*filterSelection, error) {

@mesutoezdil mesutoezdil Aug 12, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

which k8s version, and where in kube-scheduler's source does that happen? without that this fix may be locking against a race that never occurs in real scheduling.

s.filterLock.Lock()

@mesutoezdil mesutoezdil Aug 11, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this lock is global, not per node or per device. a big concurrent deployment create now queues through one lock for every node. was a per node lock considered, not only per pod uid?

defer s.filterLock.Unlock()

selection := &filterSelection{}
if pi, ok := s.podManager.TakeAndDeletePod(args.Pod); ok {
s.quotaManager.RmUsage(args.Pod, pi.Devices)
}
nodeUsage, _, failedNodes, err := s.getNodesUsage(args.NodeNames, args.Pod)
if err != nil {
s.recordScheduleFilterResultEvent(args.Pod, EventReasonFilteringFailed, "", err)
return nil, err
}
if len(failedNodes) != 0 {
klog.V(5).InfoS("Nodes failed during usage retrieval", "nodes", failedNodes)
}
nodeScores, err := s.calcScore(nodeUsage, resourceReqs, args.Pod, failedNodes)
if err != nil {
err := fmt.Errorf("calcScore failed %v for pod %v", err, args.Pod.Name)
s.recordScheduleFilterResultEvent(args.Pod, EventReasonFilteringFailed, "", err)
return nil, err
return nil, fmt.Errorf("calcScore failed %v for pod %v", err, args.Pod.Name)
}
if len((*nodeScores).NodeList) == 0 {
klog.V(4).InfoS("No available nodes meet the required scores", "pod", args.Pod.Name)
s.recordScheduleFilterResultEvent(args.Pod, EventReasonFilteringFailed, "", fmt.Errorf("no available node, %d nodes do not meet", len(*args.NodeNames)))
return &extenderv1.ExtenderFilterResult{
FailedNodes: failedNodes,
}, nil
selection.failedNodes = failedNodes
return selection, nil
}
klog.V(4).Infoln("nodeScores_len=", len((*nodeScores).NodeList))
sort.Sort(nodeScores)
Expand All @@ -1066,33 +1115,20 @@ func (s *Scheduler) Filter(args extenderv1.ExtenderArgs) (*extenderv1.ExtenderFi
annotations := make(map[string]string)
annotations[util.AssignedNodeAnnotations] = m.NodeID
annotations[util.AssignedTimeAnnotations] = strconv.FormatInt(time.Now().Unix(), 10)

for _, val := range device.GetDevices() {
val.PatchAnnotations(args.Pod, &annotations, m.Devices)
}

rawDevices := m.Devices
effectiveDevices := device.CollapseInitContainerUsage(args.Pod, rawDevices)
if args.Nodes == nil {
added := s.podManager.AddPod(args.Pod, m.NodeID, effectiveDevices)
if added {
s.quotaManager.AddUsage(args.Pod, effectiveDevices) // use collapsed
}
err = util.PatchPodAnnotations(args.Pod, annotations)
if err != nil {
s.recordScheduleFilterResultEvent(args.Pod, EventReasonFilteringFailed, "", err)
if added {
s.quotaManager.RmUsage(args.Pod, effectiveDevices)
}
s.podManager.DelPod(args.Pod)
return nil, err
}
}

successMsg := genSuccessMsg(len(*args.NodeNames), m.NodeID, nodeScores.NodeList)
s.recordScheduleFilterResultEvent(args.Pod, EventReasonFilteringSucceed, successMsg, nil)
res := extenderv1.ExtenderFilterResult{NodeNames: &[]string{m.NodeID}}
return &res, nil
// Collapse init-container usage so the cache reflects the effective footprint.
effectiveDevices := device.CollapseInitContainerUsage(args.Pod, m.Devices)
selection.chosen = m
selection.annotations = annotations
selection.nodeList = nodeScores.NodeList
selection.effectiveDevices = effectiveDevices
selection.added = s.podManager.AddPod(args.Pod, m.NodeID, effectiveDevices)
if selection.added {
s.quotaManager.AddUsage(args.Pod, effectiveDevices)
}
return selection, nil
}

func (s *Scheduler) filterSimulation(args extenderv1.ExtenderArgs, resourceReqs device.PodDeviceRequests) (*extenderv1.ExtenderFilterResult, error) {
Expand Down
158 changes: 158 additions & 0 deletions pkg/scheduler/scheduler_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import (
"maps"
"slices"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
Expand Down Expand Up @@ -1584,6 +1585,163 @@ func Test_Filter_EvictsStaleEntry(t *testing.T) {
}
}

// Test_Filter_ConcurrentNoDoubleAllocation asserts that concurrent Filter
// calls do not reserve the same device (issue #2232). Each device holds one
// pod, so the 8 pods must each get a distinct device id.
func Test_Filter_ConcurrentNoDoubleAllocation(t *testing.T) {
const numPods = 8

require.NoError(t, config.InitDevicesWithConfig(&config.Config{
NvidiaConfig: nvidia.NvidiaConfig{
ResourceCountName: "hami.io/gpu",
ResourceMemoryName: "hami.io/gpumem",
ResourceMemoryPercentageName: "hami.io/gpumem-percentage",
ResourceCoreName: "hami.io/gpucores",
DefaultGPUNum: 1,
},
}))

client.KubeClient = fake.NewClientset()
t.Cleanup(func() { client.KubeClient = nil })
s := NewScheduler()
s.kubeClient = client.KubeClient
informerFactory := informers.NewSharedInformerFactoryWithOptions(client.KubeClient, time.Hour)
s.podLister = informerFactory.Core().V1().Pods().Lister()
s.nodeLister = informerFactory.Core().V1().Nodes().Lister()
informerFactory.Start(s.stopCh)
informerFactory.WaitForCacheSync(s.stopCh)

// One node with numPods devices; each device fits exactly one pod.
deviceInfos := make([]device.DeviceInfo, 0, numPods)
for i := range numPods {
deviceInfos = append(deviceInfos, device.DeviceInfo{
ID: fmt.Sprintf("device-%d", i),
Index: uint(i),
Count: 10,
Devmem: 8000,
Devcore: 100,
Mode: "hami",
Type: nvidia.NvidiaGPUDevice,
Health: true,
DeviceVendor: nvidia.NvidiaGPUDevice,
})
}
s.addNode("node1", &device.NodeInfo{
ID: "node1",
Node: &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node1"}},
Devices: map[string][]device.DeviceInfo{
nvidia.NvidiaGPUDevice: deviceInfos,
},
})

// Create the pods in the fake apiserver so PatchPodAnnotations succeeds.
pods := make([]*corev1.Pod, numPods)
for i := range numPods {
pod := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: fmt.Sprintf("pod-%d", i),
UID: types.UID(fmt.Sprintf("uid-%d", i)),
Annotations: map[string]string{
util.GPUSchedulerPolicyAnnotationKey: util.GPUSchedulerPolicyBinpack.String(),
util.NodeSchedulerPolicyAnnotationKey: util.NodeSchedulerPolicyBinpack.String(),
},
},
Spec: corev1.PodSpec{
Containers: []corev1.Container{{
Name: "gpu-burn",
Resources: corev1.ResourceRequirements{Limits: corev1.ResourceList{
"hami.io/gpu": *resource.NewQuantity(1, resource.BinarySI),
"hami.io/gpumem": *resource.NewQuantity(5000, resource.BinarySI),
"hami.io/gpucores": *resource.NewQuantity(10, resource.BinarySI),
}},
}},
},
}
pods[i] = pod
_, err := client.KubeClient.CoreV1().Pods(pod.Namespace).Create(t.Context(), pod, metav1.CreateOptions{})
require.NoError(t, err)
}

nodeNames := []string{"node1"}

// Barrier so all goroutines enter Filter simultaneously.
start := make(chan struct{})
var wg sync.WaitGroup
for i := range numPods {
wg.Add(1)
go func(pod *corev1.Pod) {
defer wg.Done()
<-start
_, _ = s.Filter(extenderv1.ExtenderArgs{Pod: pod, NodeNames: &nodeNames})
}(pods[i])
}
close(start)
wg.Wait()

// Every pod must have been annotated with a distinct device id.
seen := make(map[string]string, numPods) // deviceID -> podName
for _, pod := range pods {
refreshed, err := client.KubeClient.CoreV1().Pods(pod.Namespace).Get(t.Context(), pod.Name, metav1.GetOptions{})
require.NoError(t, err)
allocated, err := device.DecodePodDevices(device.InRequestDevices, refreshed.Annotations)
require.NoError(t, err, "decoding devices for pod %s", pod.Name)
single, ok := allocated[nvidia.NvidiaGPUDevice]
require.True(t, ok && len(single) > 0 && len(single[0]) > 0, "pod %s has no allocated device", pod.Name)
deviceID := single[0][0].UUID
require.NotEmpty(t, deviceID, "pod %s has empty device id", pod.Name)
if other, dup := seen[deviceID]; dup {
t.Fatalf("device %s double-allocated to pod %s and pod %s", deviceID, other, pod.Name)
}
seen[deviceID] = pod.Name
}
require.Len(t, seen, numPods, "expected each pod on a distinct device")
}

// Test_Filter_RollbackOnPatchFailure verifies that when the annotation patch
// fails, Filter rolls back the reservation it just committed to the cache.
func Test_Filter_RollbackOnPatchFailure(t *testing.T) {
require.NoError(t, config.InitDevicesWithConfig(&config.Config{
NvidiaConfig: nvidia.NvidiaConfig{
ResourceCountName: "hami.io/gpu",
ResourceMemoryName: "hami.io/gpumem",
ResourceCoreName: "hami.io/gpucores",
DefaultGPUNum: 1,
},
}))
// Empty clientset: the pod is unknown to the apiserver, so PatchPodAnnotations fails.
client.KubeClient = fake.NewClientset()
t.Cleanup(func() { client.KubeClient = nil })

s := NewScheduler()
s.addNode("node1", &device.NodeInfo{
ID: "node1",
Node: &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node1"}},
Devices: map[string][]device.DeviceInfo{
nvidia.NvidiaGPUDevice: {{
ID: "device1", Index: 0, Count: 10, Devmem: 8000, Devcore: 100,
Mode: "hami", Type: nvidia.NvidiaGPUDevice, Health: true, DeviceVendor: nvidia.NvidiaGPUDevice,
}},
},
})

pod := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{Name: "pod-fail", UID: "uid-fail"},
Spec: corev1.PodSpec{Containers: []corev1.Container{{
Name: "c",
Resources: corev1.ResourceRequirements{Limits: corev1.ResourceList{
"hami.io/gpu": *resource.NewQuantity(1, resource.BinarySI),
"hami.io/gpumem": *resource.NewQuantity(1000, resource.BinarySI),
}},
}}},
}

_, err := s.Filter(extenderv1.ExtenderArgs{Pod: pod, NodeNames: &[]string{"node1"}})
require.Error(t, err, "expected patch failure to surface as an error")

_, inCache := s.podManager.GetPod(pod)
require.False(t, inCache, "reservation must be rolled back when the annotation patch fails")
}

func TestFilterUsesTemplateNodesWithoutSideEffects(t *testing.T) {
s := NewScheduler()
client.KubeClient = fake.NewSimpleClientset()
Expand Down
Loading