fix(worker): 统一异步执行准入范围
Worker 执行阶段重建准入范围时遗漏 worker_capacity,导致已预留 River 槽的任务再次同步等待准入并占住执行槽。 调度、执行和候选切换统一复用 taskAdmissionScopes,确保异步任务始终携带同一全局 Worker 容量范围。 验证:Go 全量测试、runner race、go vet、迁移安全检查通过。
This commit is contained in:
@@ -113,14 +113,10 @@ func (s *Service) buildTaskAdmissionPlan(ctx context.Context, task store.Gateway
|
|||||||
if !available {
|
if !available {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
scopes, groupID := s.admissionScopes(ctx, user, candidate)
|
scopes, groupID, scopeErr := s.taskAdmissionScopes(ctx, task, user, candidate)
|
||||||
if task.AsyncMode {
|
if scopeErr != nil {
|
||||||
scopes, err = s.withAsyncWorkerCapacityScope(ctx, scopes)
|
return taskAdmissionPlan{}, scopeErr
|
||||||
if err != nil {
|
|
||||||
return taskAdmissionPlan{}, err
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
scopes = acceptanceAdmissionScopes(task, scopes)
|
|
||||||
hasConcurrentLimit := false
|
hasConcurrentLimit := false
|
||||||
for _, scope := range scopes {
|
for _, scope := range scopes {
|
||||||
if scope.ConcurrentLimit > 0 {
|
if scope.ConcurrentLimit > 0 {
|
||||||
@@ -172,6 +168,23 @@ func (s *Service) withAsyncWorkerCapacityScope(ctx context.Context, scopes []sto
|
|||||||
return out, nil
|
return out, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (s *Service) taskAdmissionScopes(
|
||||||
|
ctx context.Context,
|
||||||
|
task store.GatewayTask,
|
||||||
|
user *auth.User,
|
||||||
|
candidate store.RuntimeModelCandidate,
|
||||||
|
) ([]store.AdmissionScope, string, error) {
|
||||||
|
scopes, groupID := s.admissionScopes(ctx, user, candidate)
|
||||||
|
if task.AsyncMode {
|
||||||
|
var err error
|
||||||
|
scopes, err = s.withAsyncWorkerCapacityScope(ctx, scopes)
|
||||||
|
if err != nil {
|
||||||
|
return nil, "", err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return acceptanceAdmissionScopes(task, scopes), groupID, nil
|
||||||
|
}
|
||||||
|
|
||||||
func acceptanceAdmissionScopes(task store.GatewayTask, scopes []store.AdmissionScope) []store.AdmissionScope {
|
func acceptanceAdmissionScopes(task store.GatewayTask, scopes []store.AdmissionScope) []store.AdmissionScope {
|
||||||
if task.RunMode != "acceptance" && task.RunMode != "acceptance_canary" {
|
if task.RunMode != "acceptance" && task.RunMode != "acceptance_canary" {
|
||||||
return scopes
|
return scopes
|
||||||
@@ -357,15 +370,10 @@ func (s *Service) ensureCandidateAdmission(
|
|||||||
body map[string]any,
|
body map[string]any,
|
||||||
candidate store.RuntimeModelCandidate,
|
candidate store.RuntimeModelCandidate,
|
||||||
) (store.TaskAdmissionResult, bool, error) {
|
) (store.TaskAdmissionResult, bool, error) {
|
||||||
scopes, groupID := s.admissionScopes(ctx, user, candidate)
|
scopes, groupID, err := s.taskAdmissionScopes(ctx, task, user, candidate)
|
||||||
if task.AsyncMode {
|
if err != nil {
|
||||||
var err error
|
return store.TaskAdmissionResult{}, true, err
|
||||||
scopes, err = s.withAsyncWorkerCapacityScope(ctx, scopes)
|
|
||||||
if err != nil {
|
|
||||||
return store.TaskAdmissionResult{}, true, err
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
scopes = acceptanceAdmissionScopes(task, scopes)
|
|
||||||
hasConcurrentLimit := false
|
hasConcurrentLimit := false
|
||||||
for _, scope := range scopes {
|
for _, scope := range scopes {
|
||||||
if scope.ConcurrentLimit > 0 {
|
if scope.ConcurrentLimit > 0 {
|
||||||
|
|||||||
@@ -544,8 +544,10 @@ func (s *Service) executeWithToken(ctx context.Context, task store.GatewayTask,
|
|||||||
var admittedLeases []store.ConcurrencyLease
|
var admittedLeases []store.ConcurrencyLease
|
||||||
admittedPlatformModelID := ""
|
admittedPlatformModelID := ""
|
||||||
if distributedAdmission && len(candidates) > 0 {
|
if distributedAdmission && len(candidates) > 0 {
|
||||||
admissionScopes, groupID := s.admissionScopes(ctx, user, candidates[0])
|
admissionScopes, groupID, scopeErr := s.taskAdmissionScopes(ctx, task, user, candidates[0])
|
||||||
admissionScopes = acceptanceAdmissionScopes(task, admissionScopes)
|
if scopeErr != nil {
|
||||||
|
return Result{}, scopeErr
|
||||||
|
}
|
||||||
hasConcurrentLimit := false
|
hasConcurrentLimit := false
|
||||||
for _, scope := range admissionScopes {
|
for _, scope := range admissionScopes {
|
||||||
if scope.ConcurrentLimit > 0 {
|
if scope.ConcurrentLimit > 0 {
|
||||||
|
|||||||
Reference in New Issue
Block a user