perf(worker): 持续批量填满异步执行槽
异步准入通知在突发提交时会合并,原调度器每次只填充一个 8 任务批次,导致全局容量长期空闲并反向拖慢入口登记。\n\n收到通知或周期唤醒后连续拉取并提交批次,直到队列为空或全局执行容量真正饱和;保留每批 8 个的短事务边界。\n\n验证:env -u AI_GATEWAY_TEST_DATABASE_URL go test ./... -count=1;go vet ./...;gofmt -l 无输出。
This commit is contained in:
@@ -704,42 +704,55 @@ func (s *Service) dispatchWaitingAsyncAdmissions(ctx context.Context) {
|
||||
case <-s.asyncAdmissionWake:
|
||||
case <-ticker.C:
|
||||
}
|
||||
batchLimit := s.cfg.AsyncWorkerHardLimit
|
||||
if batchLimit <= 0 {
|
||||
batchLimit = 64
|
||||
}
|
||||
if batchLimit > asyncAdmissionBatchMax {
|
||||
batchLimit = asyncAdmissionBatchMax
|
||||
}
|
||||
taskIDs, err := s.store.ListWaitingAsyncAdmissionTaskIDs(ctx, batchLimit)
|
||||
if err != nil {
|
||||
s.logger.Warn("list waiting async admissions failed", "error", err)
|
||||
continue
|
||||
}
|
||||
tasks := make([]store.GatewayTask, 0, len(taskIDs))
|
||||
for _, taskID := range taskIDs {
|
||||
task, err := s.store.GetTask(ctx, taskID)
|
||||
for {
|
||||
batchLimit := s.cfg.AsyncWorkerHardLimit
|
||||
if batchLimit <= 0 {
|
||||
batchLimit = 64
|
||||
}
|
||||
if batchLimit > asyncAdmissionBatchMax {
|
||||
batchLimit = asyncAdmissionBatchMax
|
||||
}
|
||||
taskIDs, err := s.store.ListWaitingAsyncAdmissionTaskIDs(ctx, batchLimit)
|
||||
if err != nil {
|
||||
if !errors.Is(err, pgx.ErrNoRows) {
|
||||
s.logger.Warn("load waiting async task failed", "taskID", taskID, "error", err)
|
||||
s.logger.Warn("list waiting async admissions failed", "error", err)
|
||||
break
|
||||
}
|
||||
if len(taskIDs) == 0 {
|
||||
break
|
||||
}
|
||||
tasks := make([]store.GatewayTask, 0, len(taskIDs))
|
||||
for _, taskID := range taskIDs {
|
||||
task, err := s.store.GetTask(ctx, taskID)
|
||||
if err != nil {
|
||||
if !errors.Is(err, pgx.ErrNoRows) {
|
||||
s.logger.Warn("load waiting async task failed", "taskID", taskID, "error", err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
continue
|
||||
if task.Status != "queued" {
|
||||
continue
|
||||
}
|
||||
tasks = append(tasks, task)
|
||||
}
|
||||
if task.Status != "queued" {
|
||||
continue
|
||||
if len(tasks) == 0 {
|
||||
break
|
||||
}
|
||||
tasks = append(tasks, task)
|
||||
}
|
||||
saturated, err := s.dispatchWaitingAsyncTasks(ctx, tasks)
|
||||
if err != nil {
|
||||
s.logger.Warn("dispatch waiting async admission batch failed", "tasks", len(tasks), "error", err)
|
||||
continue
|
||||
}
|
||||
if saturated {
|
||||
// Every asynchronous task shares the global worker-capacity FIFO
|
||||
// scope. Once its current head cannot be admitted, later tasks
|
||||
// cannot make progress until a lease is released.
|
||||
s.observeTaskAdmission("dispatch_saturated")
|
||||
saturated, err := s.dispatchWaitingAsyncTasks(ctx, tasks)
|
||||
if err != nil {
|
||||
s.logger.Warn("dispatch waiting async admission batch failed", "tasks", len(tasks), "error", err)
|
||||
break
|
||||
}
|
||||
if saturated {
|
||||
// Every asynchronous task shares the global worker-capacity FIFO
|
||||
// scope. Once its current head cannot be admitted, later tasks
|
||||
// cannot make progress until a lease is released.
|
||||
s.observeTaskAdmission("dispatch_saturated")
|
||||
break
|
||||
}
|
||||
// PostgreSQL notifications are wake-up hints and collapse while a
|
||||
// batch is being admitted. Keep filling consecutive batches until
|
||||
// the global execution scope is actually saturated so a large burst
|
||||
// cannot strand most Worker slots behind the one-second fallback.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user