fix(routing): 额度饱和时解除任务候选固定
已取得并发租约的任务继续保持候选绑定;仅当已绑定平台的 RPM 或 TPM 已满且存在可用候选时,允许任务在执行前原子迁移到下一平台,避免等待窗口重置。\n\n验证:\n- go test ./internal/runner -count=1\n- go vet ./internal/runner
This commit is contained in:
@@ -293,6 +293,10 @@ func pinCandidatesToTaskAdmission(
|
|||||||
if pinnedIndex <= 0 {
|
if pinnedIndex <= 0 {
|
||||||
return candidates, false
|
return candidates, false
|
||||||
}
|
}
|
||||||
|
pinnedCandidate := candidates[pinnedIndex]
|
||||||
|
if providerRateQuotaFull(pinnedCandidate) && !providerRateQuotaFull(candidates[0]) {
|
||||||
|
return candidates, false
|
||||||
|
}
|
||||||
out := append([]store.RuntimeModelCandidate(nil), candidates...)
|
out := append([]store.RuntimeModelCandidate(nil), candidates...)
|
||||||
pinned := out[pinnedIndex]
|
pinned := out[pinnedIndex]
|
||||||
copy(out[1:pinnedIndex+1], out[:pinnedIndex])
|
copy(out[1:pinnedIndex+1], out[:pinnedIndex])
|
||||||
@@ -300,6 +304,12 @@ func pinCandidatesToTaskAdmission(
|
|||||||
return out, true
|
return out, true
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func providerRateQuotaFull(candidate store.RuntimeModelCandidate) bool {
|
||||||
|
metrics := candidate.LoadMetrics
|
||||||
|
return (metrics.RPMLimit > 0 && metrics.RPMCurrent >= metrics.RPMLimit) ||
|
||||||
|
(metrics.TPMLimit > 0 && metrics.TPMCurrent >= metrics.TPMLimit)
|
||||||
|
}
|
||||||
|
|
||||||
func (s *Service) tryTaskAdmission(ctx context.Context, task store.GatewayTask, plan taskAdmissionPlan, waiterID string) (store.TaskAdmissionResult, error) {
|
func (s *Service) tryTaskAdmission(ctx context.Context, task store.GatewayTask, plan taskAdmissionPlan, waiterID string) (store.TaskAdmissionResult, error) {
|
||||||
return s.tryTaskAdmissionWithAdmittedHook(ctx, task, plan, waiterID, nil)
|
return s.tryTaskAdmissionWithAdmittedHook(ctx, task, plan, waiterID, nil)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -227,6 +227,70 @@ func TestPinCandidatesToTaskAdmissionAllowsRequestedReselection(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestPinCandidatesToTaskAdmissionAllowsProviderRateQuotaMigration(t *testing.T) {
|
||||||
|
input := []store.RuntimeModelCandidate{
|
||||||
|
{
|
||||||
|
PlatformID: "platform-b",
|
||||||
|
PlatformModelID: "model-b",
|
||||||
|
PlatformPriority: 200,
|
||||||
|
LoadMetrics: store.RuntimeCandidateLoadMetrics{
|
||||||
|
RPMLimit: 12,
|
||||||
|
RPMCurrent: 0,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
PlatformID: "platform-a",
|
||||||
|
PlatformModelID: "model-a",
|
||||||
|
PlatformPriority: 100,
|
||||||
|
LoadAvoided: true,
|
||||||
|
LoadMetrics: store.RuntimeCandidateLoadMetrics{
|
||||||
|
RPMLimit: 4,
|
||||||
|
RPMCurrent: 4,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
admission := &store.TaskAdmission{
|
||||||
|
Status: "admitted",
|
||||||
|
PlatformID: "platform-a",
|
||||||
|
PlatformModelID: "model-a",
|
||||||
|
}
|
||||||
|
|
||||||
|
got, pinned := pinCandidatesToTaskAdmission(input, admission)
|
||||||
|
|
||||||
|
if pinned {
|
||||||
|
t.Fatalf("rate-full provider remained pinned: %+v", got)
|
||||||
|
}
|
||||||
|
if got[0].PlatformModelID != "model-b" {
|
||||||
|
t.Fatalf("available fallback was not preserved: %+v", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPinCandidatesToTaskAdmissionKeepsConcurrencyLeaseBinding(t *testing.T) {
|
||||||
|
input := []store.RuntimeModelCandidate{
|
||||||
|
{PlatformID: "platform-b", PlatformModelID: "model-b"},
|
||||||
|
{
|
||||||
|
PlatformID: "platform-a",
|
||||||
|
PlatformModelID: "model-a",
|
||||||
|
LoadAvoided: true,
|
||||||
|
LoadMetrics: store.RuntimeCandidateLoadMetrics{
|
||||||
|
ConcurrentLimit: 2,
|
||||||
|
ConcurrentCurrent: 2,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
admission := &store.TaskAdmission{
|
||||||
|
Status: "admitted",
|
||||||
|
PlatformID: "platform-a",
|
||||||
|
PlatformModelID: "model-a",
|
||||||
|
}
|
||||||
|
|
||||||
|
got, pinned := pinCandidatesToTaskAdmission(input, admission)
|
||||||
|
|
||||||
|
if !pinned || got[0].PlatformModelID != "model-a" {
|
||||||
|
t.Fatalf("valid concurrency lease binding was not preserved: pinned=%v candidates=%+v", pinned, got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestPinCandidatesToTaskAdmissionIgnoresMissingCandidate(t *testing.T) {
|
func TestPinCandidatesToTaskAdmissionIgnoresMissingCandidate(t *testing.T) {
|
||||||
input := []store.RuntimeModelCandidate{
|
input := []store.RuntimeModelCandidate{
|
||||||
{PlatformID: "platform-a", PlatformModelID: "model-a"},
|
{PlatformID: "platform-a", PlatformModelID: "model-a"},
|
||||||
|
|||||||
Reference in New Issue
Block a user