From 9073c3a3898bdfcf81f1198e18d1b28c553ede29 Mon Sep 17 00:00:00 2001 From: Your Name Date: Thu, 30 Apr 2026 23:54:06 +0800 Subject: [PATCH] feat: raise per-model image queue concurrency --- .../settings/admin-settings-client.tsx | 25 +-- lib/canvas/image-task-config.ts | 5 +- lib/canvas/image-task-queue.ts | 183 +++++++++++++----- 3 files changed, 155 insertions(+), 58 deletions(-) diff --git a/app/(admin)/settings/admin-settings-client.tsx b/app/(admin)/settings/admin-settings-client.tsx index 4ec56c61a..8f20e57a8 100644 --- a/app/(admin)/settings/admin-settings-client.tsx +++ b/app/(admin)/settings/admin-settings-client.tsx @@ -79,6 +79,7 @@ import { import { CANVAS_IMAGE_TASK_QUEUE_CONFIG_KEY, DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY, + MIN_CANVAS_IMAGE_TASK_CONCURRENCY, MAX_CANVAS_IMAGE_TASK_CONCURRENCY, normalizeCanvasImageTaskQueueConfig, type CanvasImageTaskQueueConfig, @@ -5109,10 +5110,10 @@ export function AdminSettingsClient() { normalizeCanvasImageTaskQueueConfig(payload?.setting?.value || normalized) ); await mutateCanvasImageTaskQueueConfig(); - toast.success("生图任务队列并发配置保存成功"); + toast.success("生图任务单模型并发配置保存成功"); } catch (error) { toast.error( - error instanceof Error ? error.message : "保存生图任务队列并发配置失败" + error instanceof Error ? error.message : "保存生图任务单模型并发配置失败" ); } }; @@ -20148,16 +20149,16 @@ export function AdminSettingsClient() { - {/* 生图任务队列并发 */} + {/* 生图任务单模型并发 */}
- 生图任务队列并发 + 生图任务单模型并发 - 控制风格化/画布异步生图队列允许同时运行的任务数。配置保存到数据库，运行队列会自动读取，无需重启服务。 + 控制风格化/画布异步生图队列中，同一个模型允许同时运行的任务数。配置保存到数据库，运行队列会自动读取，无需重启服务。
生效范围

• 影响通过异步队列提交的风格化/画布生图任务。

-

• 多用户共享这个全局并发池，不按用户单独拆分。

+

• 多用户共享队列，但会按模型分别计数，不同模型之间不再互相阻塞。

+

• 当前配置的是单模型并发，不是全站总并发。

• 数值过高可能触发上游模型 429、超时或 504，建议配合模型 Key 池继续细分。

@@ -20200,16 +20202,17 @@ export function AdminSettingsClient() { } />
- 可配置范围 {DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY}- + 可配置范围 {MIN_CANVAS_IMAGE_TASK_CONCURRENCY}- {MAX_CANVAS_IMAGE_TASK_CONCURRENCY}，当前为{" "} - {canvasImageTaskQueueConfig.concurrency} 并发。 + {canvasImageTaskQueueConfig.concurrency} 并发，默认值为{" "} + {DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY}。
diff --git a/lib/canvas/image-task-config.ts b/lib/canvas/image-task-config.ts index ce87e0320..ed3a031fe 100644 --- a/lib/canvas/image-task-config.ts +++ b/lib/canvas/image-task-config.ts @@ -1,6 +1,7 @@ export const CANVAS_IMAGE_TASK_QUEUE_CONFIG_KEY = "canvas_image_task_queue_config"; -export const DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY = 1; +export const MIN_CANVAS_IMAGE_TASK_CONCURRENCY = 1; +export const DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY = 8; export const MAX_CANVAS_IMAGE_TASK_CONCURRENCY = 100; export interface CanvasImageTaskQueueConfig { @@ -10,7 +11,7 @@ export interface CanvasImageTaskQueueConfig { function clampConcurrency(value: number) { return Math.min( MAX_CANVAS_IMAGE_TASK_CONCURRENCY, - Math.max(DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY, Math.floor(value)) + Math.max(MIN_CANVAS_IMAGE_TASK_CONCURRENCY, Math.floor(value)) ); } diff --git a/lib/canvas/image-task-queue.ts b/lib/canvas/image-task-queue.ts index 4fcedb73d..63381d229 100644 --- a/lib/canvas/image-task-queue.ts +++ b/lib/canvas/image-task-queue.ts @@ -17,15 +17,22 @@ import { type CanvasImageTaskStatus = "pending" | "running" | "succeeded" | "failed"; +interface PendingTaskEntry { + taskId: string; + modelKey: string; +} + interface QueueState { - pending: string[]; + pending: PendingTaskEntry[]; pendingSet: Set; - running: Set; + running: Map; + runningByModel: Map>; draining: boolean; } const CONCURRENCY_CONFIG_CACHE_MS = 5 * 1000; const STALE_RUNNING_MS = 5 * 60 * 1000; +const PENDING_LOOKAHEAD_LIMIT = 100; const queueStateKey = "__canvas_image_task_queue_state__"; let cachedConcurrency: | { @@ -75,53 +82,138 @@ function getQueueState(): QueueState { globalObj[queueStateKey] = { pending: [], pendingSet: new Set(), - running: new Set(), + running: new Map(), + runningByModel: new Map>(), draining: false, }; } return globalObj[queueStateKey] as QueueState; } -function enqueueLocalTask(taskId: string) { +function normalizeTaskModelKey(request: unknown) { + if (!request || typeof request !== "object" || Array.isArray(request)) { + return "__default__"; + } + + const rawModel = (request as Record).model; + if (typeof rawModel !== "string") { + return "__default__"; + } + + const normalizedModel = rawModel.trim(); + return normalizedModel || "__default__"; +} + +async function getTaskModelKey(taskId: string) { + const rows = await db + .select({ request: canvasImageTask.request }) + .from(canvasImageTask) + .where(eq(canvasImageTask.id, taskId)) + .limit(1); + + return normalizeTaskModelKey(rows[0]?.request); +} + +function getRunningCountForModel(modelKey: string) { + const state = getQueueState(); + return state.runningByModel.get(modelKey)?.size || 0; +} + +function markRunningTask(taskId: string, modelKey: string) { + const state = getQueueState(); + state.running.set(taskId, modelKey); + const tasks = state.runningByModel.get(modelKey) || new Set(); + tasks.add(taskId); + state.runningByModel.set(modelKey, tasks); +} + +function releaseRunningTask(taskId: string) { + const state = getQueueState(); + const modelKey = state.running.get(taskId); + if (!modelKey) { + return; + } + + state.running.delete(taskId); + const tasks = state.runningByModel.get(modelKey); + if (!tasks) { + return; + } + + tasks.delete(taskId); + if (tasks.size === 0) { + state.runningByModel.delete(modelKey); + } +} + +async function enqueueLocalTask(taskId: string, modelKey?: string) { const state = getQueueState(); if (!taskId || state.running.has(taskId) || state.pendingSet.has(taskId)) { return; } - state.pending.push(taskId); + + const resolvedModelKey = modelKey || (await getTaskModelKey(taskId)); + state.pending.push({ + taskId, + modelKey: resolvedModelKey, + }); state.pendingSet.add(taskId); } -async function claimNextPendingTask(): Promise { +function hasModelCapacity(modelKey: string, maxConcurrentPerModel: number) { const state = getQueueState(); - while (state.pending.length > 0) { - const taskId = state.pending.shift(); - if (!taskId) continue; - state.pendingSet.delete(taskId); - if (!state.running.has(taskId)) { - return taskId; - } - } - - const runningIds = Array.from(state.running); - const whereCondition = - runningIds.length > 0 - ? and(eq(canvasImageTask.status, "pending"), notInArray(canvasImageTask.id, runningIds)) - : eq(canvasImageTask.status, "pending"); - - const rows = await db - .select({ id: canvasImageTask.id }) - .from(canvasImageTask) - .where(whereCondition) - .orderBy(asc(canvasImageTask.createdAt)) - .limit(1); - - return rows[0]?.id ?? null; + return getRunningCountForModel(modelKey) < maxConcurrentPerModel; } -async function runSingleTask(taskId: string) { +async function claimNextPendingTask( + maxConcurrentPerModel: number +): Promise { const state = getQueueState(); - state.running.add(taskId); + for (let index = 0; index < state.pending.length; index += 1) { + const entry = state.pending[index]; + if (!entry) continue; + if (state.running.has(entry.taskId)) continue; + if (!hasModelCapacity(entry.modelKey, maxConcurrentPerModel)) continue; + state.pending.splice(index, 1); + state.pendingSet.delete(entry.taskId); + return entry; + } + + const runningIds = Array.from(state.running.keys()); + const rows = await db + .select({ + id: canvasImageTask.id, + request: canvasImageTask.request, + }) + .from(canvasImageTask) + .where( + runningIds.length > 0 + ? and(eq(canvasImageTask.status, "pending"), notInArray(canvasImageTask.id, runningIds)) + : eq(canvasImageTask.status, "pending") + ) + .orderBy(asc(canvasImageTask.createdAt)) + .limit(PENDING_LOOKAHEAD_LIMIT); + + for (const row of rows) { + if (!row?.id) continue; + if (state.pendingSet.has(row.id) || state.running.has(row.id)) continue; + + const modelKey = normalizeTaskModelKey(row.request); + if (!hasModelCapacity(modelKey, maxConcurrentPerModel)) { + continue; + } + + return { + taskId: row.id, + modelKey, + }; + } + + return null; +} + +async function runSingleTask(entry: PendingTaskEntry) { try { const [task] = await db .update(canvasImageTask) @@ -133,7 +225,7 @@ async function runSingleTask(taskId: string) { error: null, attempts: sql`${canvasImageTask.attempts} + 1`, }) - .where(and(eq(canvasImageTask.id, taskId), eq(canvasImageTask.status, "pending"))) + .where(and(eq(canvasImageTask.id, entry.taskId), eq(canvasImageTask.status, "pending"))) .returning({ id: canvasImageTask.id, request: canvasImageTask.request, @@ -153,7 +245,7 @@ async function runSingleTask(taskId: string) { completedAt: new Date(), updatedAt: new Date(), }) - .where(eq(canvasImageTask.id, taskId)); + .where(eq(canvasImageTask.id, entry.taskId)); } catch (error) { await db .update(canvasImageTask) @@ -163,9 +255,9 @@ async function runSingleTask(taskId: string) { completedAt: new Date(), updatedAt: new Date(), }) - .where(eq(canvasImageTask.id, taskId)); + .where(eq(canvasImageTask.id, entry.taskId)); } finally { - state.running.delete(taskId); + releaseRunningTask(entry.taskId); void drainCanvasImageTaskQueue(); } } @@ -176,12 +268,13 @@ export async function drainCanvasImageTaskQueue() { state.draining = true; try { - const maxConcurrent = await getQueueConcurrency(); - while (state.running.size < maxConcurrent) { - const taskId = await claimNextPendingTask(); - if (!taskId) break; - if (state.running.has(taskId)) continue; - void runSingleTask(taskId); + const maxConcurrentPerModel = await getQueueConcurrency(); + while (true) { + const nextTask = await claimNextPendingTask(maxConcurrentPerModel); + if (!nextTask) break; + if (state.running.has(nextTask.taskId)) continue; + markRunningTask(nextTask.taskId, nextTask.modelKey); + void runSingleTask(nextTask); } } finally { state.draining = false; @@ -189,7 +282,7 @@ export async function drainCanvasImageTaskQueue() { } export async function enqueueCanvasImageTask(taskId: string) { - enqueueLocalTask(taskId); + await enqueueLocalTask(taskId); await drainCanvasImageTaskQueue(); } @@ -256,9 +349,9 @@ export async function getCanvasImageTaskForUser(params: { eq(canvasImageTask.id, task.id), eq(canvasImageTask.status, "running"), lt(canvasImageTask.startedAt, staleBefore) - ) + ) ); - enqueueLocalTask(task.id); + await enqueueLocalTask(task.id, normalizeTaskModelKey(task.request)); await drainCanvasImageTaskQueue(); task.status = "pending" as CanvasImageTaskStatus; task.startedAt = null; @@ -267,7 +360,7 @@ export async function getCanvasImageTaskForUser(params: { } if (task.status === "pending") { - enqueueLocalTask(task.id); + await enqueueLocalTask(task.id, normalizeTaskModelKey(task.request)); await drainCanvasImageTaskQueue(); }