fix(worker): 固定等待任务的准入候选
双 Worker 会按动态运行态排序重复规划同一批 waiting 任务,导致候选绑定来回迁移并把同步事务消耗在 rebind 上。\n\n调度 waiting 任务时优先使用首次登记的候选;仅当候选不再可用或已不在合法候选集时才按现有顺序迁移。执行阶段的 admitted 候选固定逻辑保持不变。\n\n验证:新增 waiting 候选固定回归测试;Go 全量测试、go vet、gofmt 通过。
This commit is contained in:
@@ -66,6 +66,15 @@ func distributedAdmissionModelType(modelType string) bool {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (s *Service) buildTaskAdmissionPlan(ctx context.Context, task store.GatewayTask, user *auth.User) (taskAdmissionPlan, error) {
|
func (s *Service) buildTaskAdmissionPlan(ctx context.Context, task store.GatewayTask, user *auth.User) (taskAdmissionPlan, error) {
|
||||||
|
return s.buildTaskAdmissionPlanForCurrentBinding(ctx, task, user, nil)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Service) buildTaskAdmissionPlanForCurrentBinding(
|
||||||
|
ctx context.Context,
|
||||||
|
task store.GatewayTask,
|
||||||
|
user *auth.User,
|
||||||
|
admission *store.TaskAdmission,
|
||||||
|
) (taskAdmissionPlan, error) {
|
||||||
restoredRequest, err := s.restoreTaskRequestReferences(ctx, task)
|
restoredRequest, err := s.restoreTaskRequestReferences(ctx, task)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return taskAdmissionPlan{}, err
|
return taskAdmissionPlan{}, err
|
||||||
@@ -107,6 +116,7 @@ func (s *Service) buildTaskAdmissionPlan(ctx context.Context, task store.Gateway
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return taskAdmissionPlan{}, err
|
return taskAdmissionPlan{}, err
|
||||||
}
|
}
|
||||||
|
candidates, _ = pinCandidatesToTaskAdmission(candidates, admission)
|
||||||
for _, candidate := range candidates {
|
for _, candidate := range candidates {
|
||||||
available, availabilityErr := s.store.RuntimeCandidateAvailable(ctx, candidate.PlatformID, candidate.PlatformModelID)
|
available, availabilityErr := s.store.RuntimeCandidateAvailable(ctx, candidate.PlatformID, candidate.PlatformModelID)
|
||||||
if availabilityErr != nil {
|
if availabilityErr != nil {
|
||||||
@@ -220,7 +230,9 @@ func pinCandidatesToTaskAdmission(
|
|||||||
candidates []store.RuntimeModelCandidate,
|
candidates []store.RuntimeModelCandidate,
|
||||||
admission *store.TaskAdmission,
|
admission *store.TaskAdmission,
|
||||||
) ([]store.RuntimeModelCandidate, bool) {
|
) ([]store.RuntimeModelCandidate, bool) {
|
||||||
if admission == nil || admission.Status != "admitted" || len(candidates) < 2 {
|
if admission == nil ||
|
||||||
|
(admission.Status != "waiting" && admission.Status != "admitted") ||
|
||||||
|
len(candidates) < 2 {
|
||||||
return candidates, false
|
return candidates, false
|
||||||
}
|
}
|
||||||
pinnedIndex := -1
|
pinnedIndex := -1
|
||||||
@@ -629,7 +641,14 @@ func (s *Service) dispatchWaitingAsyncTasks(ctx context.Context, tasks []store.G
|
|||||||
tasksByID := make(map[string]store.GatewayTask, len(tasks))
|
tasksByID := make(map[string]store.GatewayTask, len(tasks))
|
||||||
for _, task := range tasks {
|
for _, task := range tasks {
|
||||||
user := authUserFromTask(task)
|
user := authUserFromTask(task)
|
||||||
plan, err := s.buildTaskAdmissionPlan(ctx, task, user)
|
current, err := s.store.GetTaskAdmission(ctx, task.ID)
|
||||||
|
if err != nil {
|
||||||
|
if errors.Is(err, pgx.ErrNoRows) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
return false, err
|
||||||
|
}
|
||||||
|
plan, err := s.buildTaskAdmissionPlanForCurrentBinding(ctx, task, user, ¤t)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, err
|
return false, err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -87,31 +87,41 @@ func TestPinCandidatesToTaskAdmissionPreservesAdmittedCandidate(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestPinCandidatesToTaskAdmissionIgnoresWaitingOrMissingCandidate(t *testing.T) {
|
func TestPinCandidatesToTaskAdmissionPreservesWaitingCandidate(t *testing.T) {
|
||||||
input := []store.RuntimeModelCandidate{
|
input := []store.RuntimeModelCandidate{
|
||||||
{PlatformID: "platform-a", PlatformModelID: "model-a"},
|
{PlatformID: "platform-a", PlatformModelID: "model-a"},
|
||||||
{PlatformID: "platform-b", PlatformModelID: "model-b"},
|
{PlatformID: "platform-b", PlatformModelID: "model-b"},
|
||||||
}
|
}
|
||||||
for name, admission := range map[string]*store.TaskAdmission{
|
admission := &store.TaskAdmission{
|
||||||
"waiting": {
|
Status: "waiting",
|
||||||
Status: "waiting",
|
PlatformID: "platform-b",
|
||||||
PlatformID: "platform-b",
|
PlatformModelID: "model-b",
|
||||||
PlatformModelID: "model-b",
|
}
|
||||||
},
|
|
||||||
"missing": {
|
got, pinned := pinCandidatesToTaskAdmission(input, admission)
|
||||||
Status: "admitted",
|
|
||||||
PlatformID: "platform-c",
|
if !pinned || got[0].PlatformModelID != "model-b" {
|
||||||
PlatformModelID: "model-c",
|
t.Fatalf("waiting candidate was not pinned: pinned=%v candidates=%+v", pinned, got)
|
||||||
},
|
}
|
||||||
} {
|
}
|
||||||
t.Run(name, func(t *testing.T) {
|
|
||||||
got, pinned := pinCandidatesToTaskAdmission(input, admission)
|
func TestPinCandidatesToTaskAdmissionIgnoresMissingCandidate(t *testing.T) {
|
||||||
if pinned {
|
input := []store.RuntimeModelCandidate{
|
||||||
t.Fatalf("unexpected candidate pin for %s: %+v", name, got)
|
{PlatformID: "platform-a", PlatformModelID: "model-a"},
|
||||||
}
|
{PlatformID: "platform-b", PlatformModelID: "model-b"},
|
||||||
if got[0].PlatformModelID != "model-a" {
|
}
|
||||||
t.Fatalf("candidate order changed for %s: %+v", name, got)
|
admission := &store.TaskAdmission{
|
||||||
}
|
Status: "admitted",
|
||||||
})
|
PlatformID: "platform-c",
|
||||||
|
PlatformModelID: "model-c",
|
||||||
|
}
|
||||||
|
|
||||||
|
got, pinned := pinCandidatesToTaskAdmission(input, admission)
|
||||||
|
|
||||||
|
if pinned {
|
||||||
|
t.Fatalf("unexpected candidate pin for missing binding: %+v", got)
|
||||||
|
}
|
||||||
|
if got[0].PlatformModelID != "model-a" {
|
||||||
|
t.Fatalf("candidate order changed for missing binding: %+v", got)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user