From a13a322bf2ca62e37c4edf3ca4a2e5e5929b09d6 Mon Sep 17 00:00:00 2001 From: Zexi Li Date: Wed, 12 Aug 2026 20:17:22 +0800 Subject: [PATCH] feat(llm): sync isolated devices by sharing_mode on restart (#25307) Respect sharing_mode when attaching/restoring GPU devices in compute, and sync LLM guest devices with SKU config instead of only detaching when empty. --- pkg/compute/models/guest_actions.go | 33 +- pkg/compute/models/guest_queries.go | 6 +- pkg/compute/models/guests.go | 1 + pkg/compute/models/isolated_devices.go | 51 +++- .../isolated_devices_sharing_mode_test.go | 65 ++++ pkg/llm/models/llm_base.go | 281 ++++++++++++++++-- pkg/llm/models/llm_base_pod.go | 97 +++--- .../models/llm_isolated_device_sync_test.go | 119 ++++++++ pkg/llm/tasks/llm/llm_restart_task.go | 5 +- 9 files changed, 565 insertions(+), 93 deletions(-) create mode 100644 pkg/compute/models/isolated_devices_sharing_mode_test.go create mode 100644 pkg/llm/models/llm_isolated_device_sync_test.go diff --git a/pkg/compute/models/guest_actions.go b/pkg/compute/models/guest_actions.go index cb59daf040..99496ef1a5 100644 --- a/pkg/compute/models/guest_actions.go +++ b/pkg/compute/models/guest_actions.go @@ -1175,6 +1175,18 @@ func (self *SGuest) StartResumeTask(ctx context.Context, userCred mcclient.Token return driver.StartResumeTask(ctx, userCred, self, nil, parentTaskId) } +func isolatedDeviceRestoreKey(devType, sharingMode, model string) string { + return devType + "\x00" + sharingMode + "\x00" + model +} + +func parseIsolatedDeviceRestoreKey(key string) (devType, sharingMode, model string, ok bool) { + segs := strings.SplitN(key, "\x00", 3) + if len(segs) != 3 { + return "", "", "", false + } + return segs[0], segs[1], segs[2], true +} + func (self *SGuest) PerformRestoreVirtualIsolatedDevices(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) (jsonutils.JSONObject, error) { config := self.GetMetadataJson(ctx, api.VM_METADATA_VIRTUAL_ISOLATED_DEVICE_CONFIG, userCred) if config == nil { @@ -1191,15 +1203,12 @@ func (self *SGuest) PerformRestoreVirtualIsolatedDevices(ctx context.Context, us } devCount := map[string]int{} for i := range devConfigs { - key := devConfigs[i].DevType + "-" + devConfigs[i].Model - if cnt, ok := devCount[key]; ok { - devCount[key] = cnt + 1 - } else { - devCount[key] = 1 - } + key := isolatedDeviceRestoreKey(devConfigs[i].DevType, devConfigs[i].SharingMode, devConfigs[i].Model) + devCount[key] = devCount[key] + 1 } for i := range devs { - key := devConfigs[i].DevType + "-" + devConfigs[i].Model + isoDev := devs[i].GetIsolatedDevice() + key := isolatedDeviceRestoreKey(isoDev.DevType, isoDev.SharingMode, isoDev.Model) if cnt, ok := devCount[key]; ok { devCount[key] = cnt - 1 } @@ -1217,10 +1226,14 @@ func (self *SGuest) PerformRestoreVirtualIsolatedDevices(ctx context.Context, us if cnt <= 0 { continue } - segs := strings.SplitN(key, "-", 2) + devType, sharingMode, model, ok := parseIsolatedDeviceRestoreKey(key) + if !ok { + return nil, errors.Errorf("invalid restore key %q", key) + } devConfig := &api.IsolatedDeviceConfig{ - Model: segs[1], - DevType: segs[0], + Model: model, + DevType: devType, + SharingMode: sharingMode, } err := IsolatedDeviceManager.attachHostDeviceToGuestByModel(ctx, self, host, devConfig, userCred, usedDeviceMap, nil) if err != nil { diff --git a/pkg/compute/models/guest_queries.go b/pkg/compute/models/guest_queries.go index ae2b723cf0..ff5afdcd55 100644 --- a/pkg/compute/models/guest_queries.go +++ b/pkg/compute/models/guest_queries.go @@ -960,11 +960,15 @@ func fetchGuestIsolatedDevices(guestIds []string) map[string][]api.SIsolatedDevi } ret := make(map[string][]api.SIsolatedDevice) for i := range devs { - dev := api.SIsolatedDevice{} gdev := devs[i].GetIsolatedDevice() + if gdev == nil { + continue + } + dev := api.SIsolatedDevice{} dev.Id = gdev.Id dev.HostId = gdev.HostId dev.DevType = gdev.DevType + dev.SharingMode = gdev.SharingMode dev.Model = gdev.Model dev.Addr = gdev.Addr dev.VendorDeviceId = gdev.VendorDeviceId diff --git a/pkg/compute/models/guests.go b/pkg/compute/models/guests.go index a6bb5a7977..cf8013ca0e 100644 --- a/pkg/compute/models/guests.go +++ b/pkg/compute/models/guests.go @@ -7232,6 +7232,7 @@ func (self *SGuest) ToIsolatedDevicesConfig() []*api.IsolatedDeviceConfig { devConf.Model = dev.Model devConf.Vendor = dev.getVendor() devConf.DevType = dev.DevType + devConf.SharingMode = dev.SharingMode ret[idx] = devConf } return ret diff --git a/pkg/compute/models/isolated_devices.go b/pkg/compute/models/isolated_devices.go index 2ebd958eb9..76a51e0e52 100644 --- a/pkg/compute/models/isolated_devices.go +++ b/pkg/compute/models/isolated_devices.go @@ -674,10 +674,15 @@ func (manager *SIsolatedDeviceManager) attachHostDeviceToGuestByDevicePath(ctx c return fmt.Errorf("Model or DevicePath is empty: %#v", devConfig) } // if dev type is not nic, wire is empty string - devs, err := manager.findHostAvailableByDevAttr(devConfig.Model, "device_path", devConfig.DevicePath, host.Id, devConfig.WireId) + devs, err := manager.findHostAvailableByDevAttr(devConfig.Model, "device_path", devConfig.DevicePath, host.Id, devConfig.WireId, devConfig.SharingMode) if err != nil || len(devs) == 0 { return fmt.Errorf("Can't found model %s device_path %s on host %s", devConfig.Model, devConfig.DevicePath, host.Id) } + devs = filterDevicesBySharingMode(devs, devConfig.SharingMode) + if len(devs) == 0 { + return fmt.Errorf("Can't found model %s device_path %s sharing_mode %s on host %s", + devConfig.Model, devConfig.DevicePath, devConfig.SharingMode, host.Id) + } devs = filterDevicesByMemoryMb(devs, devConfig.MemoryMb) if len(devs) == 0 { return fmt.Errorf("device_path %s on host %s does not satisfy memory_mb=%d", @@ -722,6 +727,19 @@ func filterDevicesByMemoryMb(devs []SIsolatedDevice, minMemMb int) []SIsolatedDe return out } +func filterDevicesBySharingMode(devs []SIsolatedDevice, sharingMode string) []SIsolatedDevice { + if sharingMode == "" { + return devs + } + out := make([]SIsolatedDevice, 0, len(devs)) + for _, d := range devs { + if d.SharingMode == sharingMode { + out = append(out, d) + } + } + return out +} + type GroupDevs struct { DevPath string Devs []SIsolatedDevice @@ -760,7 +778,7 @@ type SNodeIsolateDevicesInfo struct { func (manager *SIsolatedDeviceManager) getDevNodesUsedRate( ctx context.Context, host *SHost, devConfig *api.IsolatedDeviceConfig, topo *hostapi.HostTopology, ) (map[string]SNodeIsolateDevicesInfo, error) { - devs, err := manager.findHostDevsByDevConfig(devConfig.Model, devConfig.DevType, host.Id, devConfig.WireId) + devs, err := manager.findHostDevsByDevConfig(devConfig.Model, devConfig.DevType, host.Id, devConfig.WireId, devConfig.SharingMode) if err != nil || len(devs) == 0 { return nil, fmt.Errorf("Can't found model %s on host %s", devConfig.Model, host.Id) } @@ -875,10 +893,15 @@ func (manager *SIsolatedDeviceManager) attachHostDeviceToGuestByModel( return fmt.Errorf("Not found model from info: %#v", devConfig) } // if dev type is not nic, wire is empty string - devs, err := manager.findHostAvailableByDevConfig(devConfig.Model, devConfig.DevType, host.Id, devConfig.WireId) + devs, err := manager.findHostAvailableByDevConfig(devConfig.Model, devConfig.DevType, host.Id, devConfig.WireId, devConfig.SharingMode) if err != nil || len(devs) == 0 { return fmt.Errorf("Can't found model %s on host %s", devConfig.Model, host.Id) } + devs = filterDevicesBySharingMode(devs, devConfig.SharingMode) + if len(devs) == 0 { + return fmt.Errorf("Can't found model %s sharing_mode %s on host %s", + devConfig.Model, devConfig.SharingMode, host.Id) + } // Honour the request's VRAM floor. Predicate already verified enough // fitting devices exist on the host; here we make sure attach picks one // of them rather than a same-Model but smaller-VRAM card. @@ -1138,19 +1161,22 @@ func (manager *SIsolatedDeviceManager) FindAvailableGpusOnHost(hostId string) ([ return devs, nil } -func (manager *SIsolatedDeviceManager) findHostAvailableByDevConfig(model, devType, hostId, wireId string) ([]SIsolatedDevice, error) { - return manager.findHostAvailableByDevAttr(model, "dev_type", devType, hostId, wireId) +func (manager *SIsolatedDeviceManager) findHostAvailableByDevConfig(model, devType, hostId, wireId, sharingMode string) ([]SIsolatedDevice, error) { + return manager.findHostAvailableByDevAttr(model, "dev_type", devType, hostId, wireId, sharingMode) } -func (manager *SIsolatedDeviceManager) findHostDevsByDevConfig(model, devType, hostId, wireId string) ([]SIsolatedDevice, error) { - return manager.findHostDevsByDevAttr(model, "dev_type", devType, hostId, wireId) +func (manager *SIsolatedDeviceManager) findHostDevsByDevConfig(model, devType, hostId, wireId, sharingMode string) ([]SIsolatedDevice, error) { + return manager.findHostDevsByDevAttr(model, "dev_type", devType, hostId, wireId, sharingMode) } -func (manager *SIsolatedDeviceManager) findHostDevsByDevAttr(model, attrKey, attrVal, hostId, wireId string) ([]SIsolatedDevice, error) { +func (manager *SIsolatedDeviceManager) findHostDevsByDevAttr(model, attrKey, attrVal, hostId, wireId, sharingMode string) ([]SIsolatedDevice, error) { devs := make([]SIsolatedDevice, 0) q := manager.Query() q = q.Equals("model", model).Equals("host_id", hostId) if attrVal != "" { - q.Equals(attrKey, attrVal) + q = q.Equals(attrKey, attrVal) + } + if sharingMode != "" { + q = q.Equals("sharing_mode", sharingMode) } if wireId != "" { wire := WireManager.FetchWireById(wireId) @@ -1165,12 +1191,15 @@ func (manager *SIsolatedDeviceManager) findHostDevsByDevAttr(model, attrKey, att return devs, nil } -func (manager *SIsolatedDeviceManager) findHostAvailableByDevAttr(model, attrKey, attrVal, hostId, wireId string) ([]SIsolatedDevice, error) { +func (manager *SIsolatedDeviceManager) findHostAvailableByDevAttr(model, attrKey, attrVal, hostId, wireId, sharingMode string) ([]SIsolatedDevice, error) { devs := make([]SIsolatedDevice, 0) q := manager.GetAvailableIsolatedDeviceQuery(nil) q = q.Equals("model", model).Equals("host_id", hostId) if attrVal != "" { - q.Equals(attrKey, attrVal) + q = q.Equals(attrKey, attrVal) + } + if sharingMode != "" { + q = q.Equals("sharing_mode", sharingMode) } if wireId != "" { wire := WireManager.FetchWireById(wireId) diff --git a/pkg/compute/models/isolated_devices_sharing_mode_test.go b/pkg/compute/models/isolated_devices_sharing_mode_test.go new file mode 100644 index 0000000000..acde466d18 --- /dev/null +++ b/pkg/compute/models/isolated_devices_sharing_mode_test.go @@ -0,0 +1,65 @@ +// Copyright 2019 Yunion +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package models + +import ( + "testing" + + "github.com/stretchr/testify/assert" + + api "yunion.io/x/onecloud/pkg/apis/compute" +) + +func TestFilterDevicesBySharingMode(t *testing.T) { + devs := []SIsolatedDevice{ + {Model: "GeForce RTX 4090", DevType: api.GPU_TYPE, SharingMode: api.DEVICE_SHARING_MODE_EXCLUSIVE}, + {Model: "GeForce RTX 4090", DevType: api.GPU_TYPE, SharingMode: api.DEVICE_SHARING_MODE_HAMI}, + {Model: "GeForce RTX 4090", DevType: api.GPU_TYPE, SharingMode: api.DEVICE_SHARING_MODE_MPS}, + } + + t.Run("empty sharing mode keeps all", func(t *testing.T) { + got := filterDevicesBySharingMode(devs, "") + assert.Equal(t, 3, len(got)) + }) + + t.Run("HAMI selects only HAMI", func(t *testing.T) { + got := filterDevicesBySharingMode(devs, api.DEVICE_SHARING_MODE_HAMI) + assert.Equal(t, 1, len(got)) + assert.Equal(t, api.DEVICE_SHARING_MODE_HAMI, got[0].SharingMode) + }) + + t.Run("EXCLUSIVE selects only EXCLUSIVE", func(t *testing.T) { + got := filterDevicesBySharingMode(devs, api.DEVICE_SHARING_MODE_EXCLUSIVE) + assert.Equal(t, 1, len(got)) + assert.Equal(t, api.DEVICE_SHARING_MODE_EXCLUSIVE, got[0].SharingMode) + }) + + t.Run("unmatched mode returns empty", func(t *testing.T) { + got := filterDevicesBySharingMode(devs, api.DEVICE_SHARING_MODE_UNLIMITED) + assert.Equal(t, 0, len(got)) + }) +} + +func TestIsolatedDeviceRestoreKey(t *testing.T) { + key := isolatedDeviceRestoreKey(api.GPU_TYPE, api.DEVICE_SHARING_MODE_HAMI, "GeForce RTX 4090") + devType, sharingMode, model, ok := parseIsolatedDeviceRestoreKey(key) + assert.True(t, ok) + assert.Equal(t, api.GPU_TYPE, devType) + assert.Equal(t, api.DEVICE_SHARING_MODE_HAMI, sharingMode) + assert.Equal(t, "GeForce RTX 4090", model) + + _, _, _, ok = parseIsolatedDeviceRestoreKey("bad-key") + assert.False(t, ok) +} diff --git a/pkg/llm/models/llm_base.go b/pkg/llm/models/llm_base.go index d910c7b314..03324315f3 100644 --- a/pkg/llm/models/llm_base.go +++ b/pkg/llm/models/llm_base.go @@ -99,36 +99,182 @@ func getEffectiveDevices(llmBase *SLLMBase, skuBase *SLLMSkuBase) *api.Devices { } // SyncDetachIsolatedDevicesIfEmpty detaches all guest isolated devices when -// effective devices (llm override or sku) are empty. Used on restart so stale -// GPU bindings do not survive after SKU devices are cleared. +// effective devices (llm override or sku) are empty. Kept for callers that only +// need the empty-SKU path; full sync (including sharing_mode changes) uses +// SyncIsolatedDevicesWithSku. func (llm *SLLM) SyncDetachIsolatedDevicesIfEmpty(ctx context.Context, userCred mcclient.TokenCredential, sku *SLLMSku) error { - eff := GetEffectiveDevices(llm, sku) - if eff != nil && !eff.IsZero() { - return nil + return llm.SyncIsolatedDevicesWithSku(ctx, userCred, sku) +} + +type isolatedDeviceBindKey struct { + Model string + SharingMode string +} + +// isolatedDevicesNeedSync reports whether bound guest devices differ from +// desired SKU configs in count, model, or sharing_mode. +func isolatedDevicesNeedSync(desired []*computeapi.IsolatedDeviceConfig, bound []computeapi.SIsolatedDevice) bool { + if len(desired) != len(bound) { + return true + } + want := map[isolatedDeviceBindKey]int{} + for _, d := range desired { + if d == nil { + continue + } + want[isolatedDeviceBindKey{Model: d.Model, SharingMode: d.SharingMode}]++ + } + for i := range bound { + k := isolatedDeviceBindKey{Model: bound[i].Model, SharingMode: bound[i].SharingMode} + if want[k] == 0 { + return true + } + want[k]-- + } + return false +} + +type isolatedDeviceAttachGroup struct { + Model string + SharingMode string + MemoryRequest int + Count int +} + +func groupIsolatedDeviceAttachConfigs(desired []*computeapi.IsolatedDeviceConfig) []isolatedDeviceAttachGroup { + type key struct { + Model string + SharingMode string + MemoryRequest int + } + order := make([]key, 0) + counts := map[key]int{} + for _, d := range desired { + if d == nil { + continue + } + k := key{Model: d.Model, SharingMode: d.SharingMode, MemoryRequest: d.MemoryRequest} + if _, ok := counts[k]; !ok { + order = append(order, k) + } + counts[k]++ + } + out := make([]isolatedDeviceAttachGroup, 0, len(order)) + for _, k := range order { + out = append(out, isolatedDeviceAttachGroup{ + Model: k.Model, + SharingMode: k.SharingMode, + MemoryRequest: k.MemoryRequest, + Count: counts[k], + }) + } + return out +} + +// SyncIsolatedDevicesWithSku reconciles guest isolated devices with SKU/LLM +// effective devices on restart: detach-all when desired differs from bound, +// then attach by model/sharing_mode/memory_request. +func (llm *SLLM) SyncIsolatedDevicesWithSku(ctx context.Context, userCred mcclient.TokenCredential, sku *SLLMSku) error { + vramClaimMb := 0 + if sku != nil { + vramClaimMb = sku.EstimateVramClaimMb() + } + desired, err := BuildIsolatedDeviceConfigs(GetEffectiveDevices(llm, sku), vramClaimMb) + if err != nil { + return errors.Wrap(err, "BuildIsolatedDeviceConfigs") } server, err := llm.GetServer(ctx) if err != nil { return errors.Wrap(err, "GetServer") } - if len(server.IsolatedDevices) == 0 { + if !isolatedDevicesNeedSync(desired, server.IsolatedDevices) { + return nil + } + if len(server.IsolatedDevices) > 0 { + if err := llm.detachAllIsolatedDevices(ctx, userCred); err != nil { + return errors.Wrap(err, "detachAllIsolatedDevices") + } + } + if len(desired) == 0 { return nil } + s := auth.GetSession(ctx, userCred, options.Options.Region) + for _, group := range groupIsolatedDeviceAttachConfigs(desired) { + if err := llm.waitServerReadyForIsolatedDeviceAction(ctx, userCred); err != nil { + return errors.Wrap(err, "wait ready before attach-isolated-device") + } + count := group.Count + input := &computeapi.ServerAttachIsolatedDeviceInput{ + Model: group.Model, + ServerAttachIsolatedDeviceBase: computeapi.ServerAttachIsolatedDeviceBase{ + SharingMode: group.SharingMode, + Count: &count, + AutoStart: false, + }, + } + if group.SharingMode == computeapi.DEVICE_SHARING_MODE_HAMI { + if group.MemoryRequest <= 0 { + return errors.Wrap(httperrors.ErrInputParameter, "HAMI attach requires memory_request > 0") + } + memReq := group.MemoryRequest + input.MemoryRequest = &memReq + } + _, err := compute.Servers.PerformAction(s, llm.CmpId, "attach-isolated-device", jsonutils.Marshal(input)) + if err != nil { + return errors.Wrapf(err, "attach-isolated-device model=%s sharing_mode=%s count=%d", + group.Model, group.SharingMode, group.Count) + } + // Wait through sync_config (if any) back to ready before next attach / final check. + if _, err := llm.waitAfterIsolatedDeviceAction(ctx, userCred, nil); err != nil { + return errors.Wrap(err, "waitAfterIsolatedDeviceAction after attach-isolated-device") + } + } + server, err = llm.waitAfterIsolatedDeviceAction(ctx, userCred, func(srv *computeapi.ServerDetails) bool { + return !isolatedDevicesNeedSync(desired, srv.IsolatedDevices) + }) + if err != nil { + return errors.Wrap(err, "waitAfterIsolatedDeviceAction for desired devices") + } + if isolatedDevicesNeedSync(desired, server.IsolatedDevices) { + return errors.Wrapf(errors.ErrInvalidStatus, + "isolated devices mismatch after sync: desired=%s bound=%s", + formatIsolatedDeviceBindKeys(desired), formatBoundIsolatedDeviceBindKeys(server.IsolatedDevices)) + } + return nil +} + +func formatIsolatedDeviceBindKeys(desired []*computeapi.IsolatedDeviceConfig) string { + parts := make([]string, 0, len(desired)) + for _, d := range desired { + if d == nil { + continue + } + parts = append(parts, fmt.Sprintf("%s/%s", d.Model, d.SharingMode)) + } + return fmt.Sprintf("%v", parts) +} + +func formatBoundIsolatedDeviceBindKeys(bound []computeapi.SIsolatedDevice) string { + parts := make([]string, 0, len(bound)) + for i := range bound { + parts = append(parts, fmt.Sprintf("%s/%s", bound[i].Model, bound[i].SharingMode)) + } + return fmt.Sprintf("%v", parts) +} + +func (llm *SLLM) detachAllIsolatedDevices(ctx context.Context, userCred mcclient.TokenCredential) error { s := auth.GetSession(ctx, userCred, options.Options.Region) params := jsonutils.NewDict() params.Set("detach_all", jsonutils.JSONTrue) - _, err = compute.Servers.PerformAction(s, llm.CmpId, "detach-isolated-device", params) + _, err := compute.Servers.PerformAction(s, llm.CmpId, "detach-isolated-device", params) if err != nil { return errors.Wrap(err, "detach-isolated-device") } - // detach-isolated-device schedules GuestIsolatedDeviceSyncTask asynchronously; - // status stays ready briefly then becomes sync_config. Waiting for ready - // immediately races and lets start run while still syncing. - if err := llm.waitServerLeaveReadyStatus(ctx, 120); err != nil { - return errors.Wrap(err, "waitServerLeaveReadyStatus after detach-isolated-device") - } - server, err = llm.WaitServerStatus(ctx, userCred, []string{computeapi.VM_READY}, 1800) + server, err := llm.waitAfterIsolatedDeviceAction(ctx, userCred, func(srv *computeapi.ServerDetails) bool { + return len(srv.IsolatedDevices) == 0 + }) if err != nil { - return errors.Wrap(err, "WaitServerStatus after detach-isolated-device") + return errors.Wrap(err, "waitAfterIsolatedDeviceAction after detach-isolated-device") } if len(server.IsolatedDevices) > 0 { return errors.Wrapf(errors.ErrInvalidStatus, "isolated devices still present after detach: %d", len(server.IsolatedDevices)) @@ -136,24 +282,107 @@ func (llm *SLLM) SyncDetachIsolatedDevicesIfEmpty(ctx context.Context, userCred return nil } -// waitServerLeaveReadyStatus polls until guest status is no longer ready -// (e.g. sync_config), or until timeoutSecs elapses while still ready. -func (llm *SLLM) waitServerLeaveReadyStatus(ctx context.Context, timeoutSecs int) error { - expire := time.Now().Add(time.Second * time.Duration(timeoutSecs)) - for time.Now().Before(expire) { - server, err := llm.GetServer(ctx) +const ( + // isolatedDeviceLeaveReadyProbeSecs: KVM sync briefly leaves ready; POD often + // never does. Keep the probe short so POD restart is not stuck for minutes. + isolatedDeviceLeaveReadyProbeSecs = 5 + // isolatedDeviceSettleTimeoutSecs: max wait for bound devices to match after + // status is back to ready. + isolatedDeviceSettleTimeoutSecs = 60 +) + +func (llm *SLLM) waitServerReadyForIsolatedDeviceAction(ctx context.Context, userCred mcclient.TokenCredential) error { + server, err := llm.GetServer(ctx) + if err != nil { + return errors.Wrap(err, "GetServer") + } + if server.Status == computeapi.VM_READY { + return nil + } + if strings.Contains(server.Status, "fail") { + return errors.Wrapf(errors.ErrInvalidStatus, "server status %s", server.Status) + } + if _, err := llm.WaitServerStatus(ctx, userCred, []string{computeapi.VM_READY}, 1800); err != nil { + return errors.Wrap(err, "WaitServerStatus ready") + } + return nil +} + +// waitAfterIsolatedDeviceAction waits for GuestIsolatedDeviceSyncTask to finish. +// It first probes for a leave-ready transition (sync_config) without treating +// "devices already match" as done — DB updates can land before status flips, +// and attaching while still sync_config fails. After back to ready, optionally +// wait until settled(). +func (llm *SLLM) waitAfterIsolatedDeviceAction( + ctx context.Context, + userCred mcclient.TokenCredential, + settled func(*computeapi.ServerDetails) bool, +) (*computeapi.ServerDetails, error) { + // Do not pass settled into the leave-ready probe: empty/desired bindings may + // appear while status is still about to become sync_config. + leftReady, server, err := llm.probeServerLeaveReadyStatus(ctx, isolatedDeviceLeaveReadyProbeSecs) + if err != nil { + return nil, err + } + if leftReady || (server != nil && server.Status != computeapi.VM_READY) { + server, err = llm.WaitServerStatus(ctx, userCred, []string{computeapi.VM_READY}, 1800) if err != nil { - return errors.Wrap(err, "GetServer") + return nil, errors.Wrap(err, "WaitServerStatus after isolated-device action") + } + } + if settled == nil { + return server, nil + } + if settled(server) { + return server, nil + } + expire := time.Now().Add(time.Second * time.Duration(isolatedDeviceSettleTimeoutSecs)) + for time.Now().Before(expire) { + server, err = llm.GetServer(ctx) + if err != nil { + return nil, errors.Wrap(err, "GetServer") } if server.Status != computeapi.VM_READY { if strings.Contains(server.Status, "fail") { - return errors.Wrapf(errors.ErrInvalidStatus, "server status %s", server.Status) + return nil, errors.Wrapf(errors.ErrInvalidStatus, "server status %s", server.Status) } - return nil + server, err = llm.WaitServerStatus(ctx, userCred, []string{computeapi.VM_READY}, 1800) + if err != nil { + return nil, errors.Wrap(err, "WaitServerStatus after late leave-ready") + } + continue } - time.Sleep(time.Second) + if settled(server) { + return server, nil + } + time.Sleep(500 * time.Millisecond) + } + return server, nil +} + +// probeServerLeaveReadyStatus polls until status leaves ready or timeout. +// Returns leftReady=true when status left ready. +func (llm *SLLM) probeServerLeaveReadyStatus( + ctx context.Context, + timeoutSecs int, +) (leftReady bool, server *computeapi.ServerDetails, err error) { + expire := time.Now().Add(time.Second * time.Duration(timeoutSecs)) + for { + server, err = llm.GetServer(ctx) + if err != nil { + return false, nil, errors.Wrap(err, "GetServer") + } + if server.Status != computeapi.VM_READY { + if strings.Contains(server.Status, "fail") { + return false, nil, errors.Wrapf(errors.ErrInvalidStatus, "server status %s", server.Status) + } + return true, server, nil + } + if !time.Now().Before(expire) { + return false, server, nil + } + time.Sleep(200 * time.Millisecond) } - return nil } // HasHygonDevices reports whether effective devices include Hygon DCU (exclusive or HAMI). diff --git a/pkg/llm/models/llm_base_pod.go b/pkg/llm/models/llm_base_pod.go index c8e6f3ad18..40827d6d3f 100644 --- a/pkg/llm/models/llm_base_pod.go +++ b/pkg/llm/models/llm_base_pod.go @@ -22,6 +22,56 @@ type PodPostStopCleanupConfig struct { Dirs []string `json:"dirs"` } +// BuildIsolatedDeviceConfigs builds pod IsolatedDeviceConfig list from SKU/LLM +// devices. Empty or nil devices yield nil. HAMI devices without per-device +// memory_mb require a positive vramClaimMb (evenly split across devices). +func BuildIsolatedDeviceConfigs(effectiveDevices *api.Devices, vramClaimMb int) ([]*computeapi.IsolatedDeviceConfig, error) { + if effectiveDevices == nil || effectiveDevices.IsZero() { + return nil, nil + } + devices := make(api.Devices, len(*effectiveDevices)) + copy(devices, *effectiveDevices) + for i := range devices { + normalizeLLMSkuDevice(&devices[i]) + } + hasHAMINeedingClaim := false + for i := range devices { + if devices[i].SharingMode == computeapi.DEVICE_SHARING_MODE_HAMI && devices[i].MemoryMb <= 0 { + hasHAMINeedingClaim = true + break + } + } + if hasHAMINeedingClaim && vramClaimMb <= 0 { + return nil, errors.Wrap(httperrors.ErrInputParameter, + "vram claim is 0 for HAMI devices: set devices[].memory_mb, mount InstantModel with weight_size_bytes, or use a non-HAMI sharing_mode") + } + // Evenly split estimated vram claim across requested devices when a + // device does not set memory_mb. Ceiling division so the sum is never + // less than the claim. + perDevFromClaim := 0 + if vramClaimMb > 0 && len(devices) > 0 { + perDevFromClaim = (vramClaimMb + len(devices) - 1) / len(devices) + } + out := make([]*computeapi.IsolatedDeviceConfig, 0, len(devices)) + for i := 0; i < len(devices); i++ { + memMb := devices[i].MemoryMb + if memMb <= 0 { + memMb = perDevFromClaim + } + out = append(out, &computeapi.IsolatedDeviceConfig{ + DevType: devices[i].DevType, + SharingMode: devices[i].SharingMode, + Vendor: devices[i].Vendor, + Model: devices[i].Model, + DevicePath: devices[i].DevicePath, + MemoryMb: memMb, + MemoryRequest: memMb, + SmUtilLimit: devices[i].SmUtilLimit, + }) + } + return out, nil +} + func GetLLMBasePodCreateInput( ctx context.Context, userCred mcclient.TokenCredential, @@ -65,50 +115,11 @@ func GetLLMBasePodCreateInput( } // isolated devices - effectiveDevices := getEffectiveDevices(llmBase, skuBase) - if effectiveDevices != nil && !effectiveDevices.IsZero() { - data.IsolatedDevices = make([]*computeapi.IsolatedDeviceConfig, 0) - devices := make(api.Devices, len(*effectiveDevices)) - copy(devices, *effectiveDevices) - for i := range devices { - normalizeLLMSkuDevice(&devices[i]) - } - hasHAMINeedingClaim := false - for i := range devices { - if devices[i].SharingMode == computeapi.DEVICE_SHARING_MODE_HAMI && devices[i].MemoryMb <= 0 { - hasHAMINeedingClaim = true - break - } - } - if hasHAMINeedingClaim && vramClaimMb <= 0 { - return nil, errors.Wrap(httperrors.ErrInputParameter, - "vram claim is 0 for HAMI devices: set devices[].memory_mb, mount InstantModel with weight_size_bytes, or use a non-HAMI sharing_mode") - } - // Evenly split estimated vram claim across requested devices when a - // device does not set memory_mb. Ceiling division so the sum is never - // less than the claim. - perDevFromClaim := 0 - if vramClaimMb > 0 && len(devices) > 0 { - perDevFromClaim = (vramClaimMb + len(devices) - 1) / len(devices) - } - for i := 0; i < len(devices); i++ { - memMb := devices[i].MemoryMb - if memMb <= 0 { - memMb = perDevFromClaim - } - isolatedDevice := &computeapi.IsolatedDeviceConfig{ - DevType: devices[i].DevType, - SharingMode: devices[i].SharingMode, - Vendor: devices[i].Vendor, - Model: devices[i].Model, - DevicePath: devices[i].DevicePath, - MemoryMb: memMb, - MemoryRequest: memMb, - SmUtilLimit: devices[i].SmUtilLimit, - } - data.IsolatedDevices = append(data.IsolatedDevices, isolatedDevice) - } + isolatedDevices, err := BuildIsolatedDeviceConfigs(getEffectiveDevices(llmBase, skuBase), vramClaimMb) + if err != nil { + return nil, err } + data.IsolatedDevices = isolatedDevices // port mappings // var portRange *computeapi.GuestPortMappingPortRange diff --git a/pkg/llm/models/llm_isolated_device_sync_test.go b/pkg/llm/models/llm_isolated_device_sync_test.go new file mode 100644 index 0000000000..9d70674ea2 --- /dev/null +++ b/pkg/llm/models/llm_isolated_device_sync_test.go @@ -0,0 +1,119 @@ +// Copyright 2019 Yunion +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package models + +import ( + "testing" + + computeapi "yunion.io/x/onecloud/pkg/apis/compute" + api "yunion.io/x/onecloud/pkg/apis/llm" +) + +func TestBuildIsolatedDeviceConfigsSharingModes(t *testing.T) { + devs := api.Devices{ + {Model: "A100", SharingMode: computeapi.DEVICE_SHARING_MODE_UNLIMITED}, + {Model: "A100", SharingMode: computeapi.DEVICE_SHARING_MODE_HAMI, MemoryMb: 8192}, + } + out, err := BuildIsolatedDeviceConfigs(&devs, 0) + if err != nil { + t.Fatalf("BuildIsolatedDeviceConfigs: %v", err) + } + if len(out) != 2 { + t.Fatalf("len = %d", len(out)) + } + if out[0].SharingMode != computeapi.DEVICE_SHARING_MODE_UNLIMITED { + t.Fatalf("dev0 SharingMode = %q", out[0].SharingMode) + } + if out[1].SharingMode != computeapi.DEVICE_SHARING_MODE_HAMI || out[1].MemoryRequest != 8192 { + t.Fatalf("dev1 = %#v", out[1]) + } +} + +func TestBuildIsolatedDeviceConfigsHAMISplitClaim(t *testing.T) { + devs := api.Devices{{Model: "A100"}, {Model: "A100"}} + claim := 40960 + out, err := BuildIsolatedDeviceConfigs(&devs, claim) + if err != nil { + t.Fatalf("BuildIsolatedDeviceConfigs: %v", err) + } + perDev := (claim + len(devs) - 1) / len(devs) + for i := range out { + if out[i].SharingMode != computeapi.DEVICE_SHARING_MODE_HAMI { + t.Fatalf("dev%d SharingMode = %q", i, out[i].SharingMode) + } + if out[i].MemoryRequest != perDev { + t.Fatalf("dev%d MemoryRequest = %d want %d", i, out[i].MemoryRequest, perDev) + } + } +} + +func TestBuildIsolatedDeviceConfigsEmpty(t *testing.T) { + out, err := BuildIsolatedDeviceConfigs(nil, 0) + if err != nil || out != nil { + t.Fatalf("nil devices: out=%v err=%v", out, err) + } + empty := api.Devices{} + out, err = BuildIsolatedDeviceConfigs(&empty, 0) + if err != nil || out != nil { + t.Fatalf("empty devices: out=%v err=%v", out, err) + } +} + +func TestIsolatedDevicesNeedSync(t *testing.T) { + desired := []*computeapi.IsolatedDeviceConfig{ + {Model: "A100", SharingMode: computeapi.DEVICE_SHARING_MODE_HAMI, MemoryRequest: 8192}, + } + boundSame := []computeapi.SIsolatedDevice{ + {Model: "A100", SharingMode: computeapi.DEVICE_SHARING_MODE_HAMI}, + } + if isolatedDevicesNeedSync(desired, boundSame) { + t.Fatal("same model/sharing_mode should not need sync") + } + + boundUnlimited := []computeapi.SIsolatedDevice{ + {Model: "A100", SharingMode: computeapi.DEVICE_SHARING_MODE_UNLIMITED}, + } + if !isolatedDevicesNeedSync(desired, boundUnlimited) { + t.Fatal("UNLIMITED -> HAMI should need sync") + } + + if !isolatedDevicesNeedSync(desired, nil) { + t.Fatal("empty bound should need sync when desired non-empty") + } + if !isolatedDevicesNeedSync(nil, boundUnlimited) { + t.Fatal("empty desired should need sync when bound non-empty") + } + if isolatedDevicesNeedSync(nil, nil) { + t.Fatal("both empty should not need sync") + } +} + +func TestGroupIsolatedDeviceAttachConfigs(t *testing.T) { + desired := []*computeapi.IsolatedDeviceConfig{ + {Model: "A100", SharingMode: computeapi.DEVICE_SHARING_MODE_HAMI, MemoryRequest: 8192}, + {Model: "A100", SharingMode: computeapi.DEVICE_SHARING_MODE_HAMI, MemoryRequest: 8192}, + {Model: "A100", SharingMode: computeapi.DEVICE_SHARING_MODE_UNLIMITED}, + } + groups := groupIsolatedDeviceAttachConfigs(desired) + if len(groups) != 2 { + t.Fatalf("groups len = %d want 2", len(groups)) + } + if groups[0].Count != 2 || groups[0].SharingMode != computeapi.DEVICE_SHARING_MODE_HAMI || groups[0].MemoryRequest != 8192 { + t.Fatalf("group0 = %#v", groups[0]) + } + if groups[1].Count != 1 || groups[1].SharingMode != computeapi.DEVICE_SHARING_MODE_UNLIMITED { + t.Fatalf("group1 = %#v", groups[1]) + } +} diff --git a/pkg/llm/tasks/llm/llm_restart_task.go b/pkg/llm/tasks/llm/llm_restart_task.go index 18146de881..6adb59e6d0 100644 --- a/pkg/llm/tasks/llm/llm_restart_task.go +++ b/pkg/llm/tasks/llm/llm_restart_task.go @@ -376,8 +376,8 @@ func (task *LLMRestartTask) OnResetDiskComplete(ctx context.Context, obj db.ISta return } - if err := llm.SyncDetachIsolatedDevicesIfEmpty(ctx, task.UserCred, sku); err != nil { - task.taskFailed(ctx, llm, errors.Wrap(err, "SyncDetachIsolatedDevicesIfEmpty").Error()) + if err := llm.SyncIsolatedDevicesWithSku(ctx, task.UserCred, sku); err != nil { + task.taskFailed(ctx, llm, errors.Wrap(err, "SyncIsolatedDevicesWithSku").Error()) return } @@ -398,6 +398,7 @@ func (task *LLMRestartTask) OnResetDiskComplete(ctx context.Context, obj db.ISta diskCaseInsensitive = true }*/ + // Refresh after GPU sync so container update uses the new bound device ids. srvDetails, err := llm.GetServer(ctx) if err != nil { task.taskFailed(ctx, llm, errors.Wrap(err, "GetServer").Error())