From e7cf10bb281c4612bb378adb54a6b6dc2552b79a Mon Sep 17 00:00:00 2001 From: Your Name Date: Fri, 1 May 2026 00:17:25 +0800 Subject: [PATCH] feat: unify site-wide image concurrency --- .../settings/admin-settings-client.tsx | 16 +- lib/canvas/generate-image.ts | 23 ++ lib/canvas/image-concurrency.ts | 197 ++++++++++++++++++ lib/canvas/image-task-queue.ts | 57 +---- 4 files changed, 233 insertions(+), 60 deletions(-) create mode 100644 lib/canvas/image-concurrency.ts diff --git a/app/(admin)/settings/admin-settings-client.tsx b/app/(admin)/settings/admin-settings-client.tsx index 8f20e57a8..aa55a06b7 100644 --- a/app/(admin)/settings/admin-settings-client.tsx +++ b/app/(admin)/settings/admin-settings-client.tsx @@ -5110,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 : "保存全站生图单模型并发配置失败" ); } }; @@ -20149,16 +20149,16 @@ export function AdminSettingsClient() { - {/* 生图任务单模型并发 */} + {/* 全站生图单模型并发 */}
- 生图任务单模型并发 + 全站生图单模型并发 - 控制风格化/画布异步生图队列中，同一个模型允许同时运行的任务数。配置保存到数据库，运行队列会自动读取，无需重启服务。 + 控制全站生图链路中，同一个模型允许同时运行的任务数。配置保存到数据库，异步队列和直调生成都会自动读取，无需重启服务。
生效范围
-

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

-

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

-

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

+

• 影响画布异步生图、智能体直调生图、视频分镜高清预览等全站生图入口。

+

• 多用户共享同一套并发限制，但会按模型分别计数，不同模型之间不再互相阻塞。

+

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

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

diff --git a/lib/canvas/generate-image.ts b/lib/canvas/generate-image.ts index 339a70267..4ff80984c 100644 --- a/lib/canvas/generate-image.ts +++ b/lib/canvas/generate-image.ts @@ -12,6 +12,10 @@ import { createMidjourney } from "@/lib/ai/providers/midjourney"; import { createRunway } from "@/lib/ai/providers/runway"; import { createKrea } from "@/lib/ai/providers/krea"; import { filterReachableRemoteUrls } from "@/lib/net/url-access"; +import { + acquireCanvasImageModelSlot, + normalizeCanvasImageModelKey, +} from "./image-concurrency"; export interface CanvasGenerateImageRequest { prompt: string; @@ -445,6 +449,19 @@ export async function generateCanvasImage( const targetImageCount = Math.max(1, imageCount); const collectedImages: ParsedImage[] = []; const maxBatches = Math.max(1, Math.ceil(targetImageCount / providerBatchLimit)); + const resolvedModelKey = normalizeCanvasImageModelKey( + selectedModel.model.modelId || selectedModel.model.id || model + ); + const slot = await acquireCanvasImageModelSlot({ + modelKey: resolvedModelKey, + debugLabel, + debugMeta: { + selectedModelId: selectedModel.model.modelId, + selectedModelRef: selectedModel.model.id, + providerType: provider.type, + ...debugMeta, + }, + }); try { for (let batchIndex = 0; batchIndex < maxBatches; batchIndex += 1) { @@ -490,6 +507,8 @@ export async function generateCanvasImage( console.info("[Canvas Generate Image] Completed:", { debugLabel, elapsedMs: Date.now() - startedAt, + waitMs: slot.waitMs, + singleModelConcurrency: slot.maxConcurrent, finalImageCount: finalImages.length, providerType: provider.type, selectedModelRef: selectedModel.model.id, @@ -507,6 +526,8 @@ export async function generateCanvasImage( console.error("[Canvas Generate Image] Failed:", { debugLabel, elapsedMs: Date.now() - startedAt, + waitMs: slot.waitMs, + singleModelConcurrency: slot.maxConcurrent, providerType: provider.type, selectedModelRef: selectedModel.model.id, selectedModelId: selectedModel.model.modelId, @@ -514,5 +535,7 @@ export async function generateCanvasImage( ...debugMeta, }); throw error; + } finally { + await slot.release(); } } diff --git a/lib/canvas/image-concurrency.ts b/lib/canvas/image-concurrency.ts new file mode 100644 index 000000000..97011002e --- /dev/null +++ b/lib/canvas/image-concurrency.ts @@ -0,0 +1,197 @@ +import "server-only"; + +import { getSystemSetting } from "@/lib/db/queries"; +import { + CANVAS_IMAGE_TASK_QUEUE_CONFIG_KEY, + DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY, + normalizeCanvasImageTaskQueueConfig, +} from "./image-task-config"; + +const CONCURRENCY_CONFIG_CACHE_MS = 5 * 1000; +const concurrencyStateKey = "__canvas_image_model_concurrency__"; + +interface ModelWaiter { + ticketId: string; + resolve: () => void; +} + +interface ModelConcurrencyState { + running: Set; + waiters: ModelWaiter[]; +} + +interface CanvasImageConcurrencyState { + models: Map; +} + +let cachedConcurrency: + | { + expiresAt: number; + value: number; + } + | null = null; + +function getGlobalConcurrencyState(): CanvasImageConcurrencyState { + const globalState = globalThis as typeof globalThis & { + [concurrencyStateKey]?: CanvasImageConcurrencyState; + }; + + if (!globalState[concurrencyStateKey]) { + globalState[concurrencyStateKey] = { + models: new Map(), + }; + } + + return globalState[concurrencyStateKey] as CanvasImageConcurrencyState; +} + +function getOrCreateModelState(modelKey: string) { + const state = getGlobalConcurrencyState(); + const existing = state.models.get(modelKey); + if (existing) { + return existing; + } + + const created: ModelConcurrencyState = { + running: new Set(), + waiters: [], + }; + state.models.set(modelKey, created); + return created; +} + +function cleanupModelState(modelKey: string) { + const state = getGlobalConcurrencyState(); + const modelState = state.models.get(modelKey); + if (!modelState) { + return; + } + + if (modelState.running.size === 0 && modelState.waiters.length === 0) { + state.models.delete(modelKey); + } +} + +export function normalizeCanvasImageModelKey(value: unknown) { + if (typeof value !== "string") { + return "__default__"; + } + + const normalized = value.trim(); + return normalized || "__default__"; +} + +export async function getCanvasImageSingleModelConcurrency(): Promise { + const now = Date.now(); + if (cachedConcurrency && cachedConcurrency.expiresAt > now) { + return cachedConcurrency.value; + } + + const envConcurrency = process.env.CANVAS_IMAGE_TASK_CONCURRENCY; + let value = normalizeCanvasImageTaskQueueConfig(undefined, { + envConcurrency, + }).concurrency; + + try { + const setting = await getSystemSetting({ + key: CANVAS_IMAGE_TASK_QUEUE_CONFIG_KEY, + }); + if (setting?.value !== undefined && setting.value !== null) { + value = normalizeCanvasImageTaskQueueConfig(setting.value, { + envConcurrency, + }).concurrency; + } + } catch (error) { + console.warn("[Canvas Image Concurrency] Failed to load DB config, using fallback", { + error: error instanceof Error ? error.message : String(error), + }); + } + + cachedConcurrency = { + expiresAt: now + CONCURRENCY_CONFIG_CACHE_MS, + value: value || DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY, + }; + + return cachedConcurrency.value; +} + +export function getCanvasImageRunningCountForModel(modelKey: string) { + const normalizedModelKey = normalizeCanvasImageModelKey(modelKey); + return getGlobalConcurrencyState().models.get(normalizedModelKey)?.running.size || 0; +} + +async function drainModelWaiters(modelKey: string) { + const normalizedModelKey = normalizeCanvasImageModelKey(modelKey); + const modelState = getGlobalConcurrencyState().models.get(normalizedModelKey); + if (!modelState || modelState.waiters.length === 0) { + cleanupModelState(normalizedModelKey); + return; + } + + const maxConcurrent = await getCanvasImageSingleModelConcurrency(); + while (modelState.running.size < maxConcurrent && modelState.waiters.length > 0) { + const waiter = modelState.waiters.shift(); + if (!waiter) { + break; + } + modelState.running.add(waiter.ticketId); + waiter.resolve(); + } + + cleanupModelState(normalizedModelKey); +} + +export async function acquireCanvasImageModelSlot(params: { + modelKey: string; + debugLabel?: string; + debugMeta?: Record; +}) { + const normalizedModelKey = normalizeCanvasImageModelKey(params.modelKey); + const maxConcurrent = await getCanvasImageSingleModelConcurrency(); + const modelState = getOrCreateModelState(normalizedModelKey); + const ticketId = `${normalizedModelKey}:${Date.now()}:${Math.random().toString(36).slice(2, 10)}`; + const waitingSince = Date.now(); + + if (modelState.waiters.length === 0 && modelState.running.size < maxConcurrent) { + modelState.running.add(ticketId); + return { + release: async () => { + const state = getGlobalConcurrencyState().models.get(normalizedModelKey); + if (!state) return; + state.running.delete(ticketId); + await drainModelWaiters(normalizedModelKey); + }, + waitMs: 0, + maxConcurrent, + }; + } + + await new Promise((resolve) => { + modelState.waiters.push({ + ticketId, + resolve, + }); + void drainModelWaiters(normalizedModelKey); + }); + + const waitMs = Date.now() - waitingSince; + if (waitMs > 250) { + console.info("[Canvas Image Concurrency] Waited for model slot:", { + debugLabel: params.debugLabel || "canvas-image", + modelKey: normalizedModelKey, + waitMs, + ...params.debugMeta, + }); + } + + return { + release: async () => { + const state = getGlobalConcurrencyState().models.get(normalizedModelKey); + if (!state) return; + state.running.delete(ticketId); + await drainModelWaiters(normalizedModelKey); + }, + waitMs, + maxConcurrent, + }; +} diff --git a/lib/canvas/image-task-queue.ts b/lib/canvas/image-task-queue.ts index 63381d229..f80426b2d 100644 --- a/lib/canvas/image-task-queue.ts +++ b/lib/canvas/image-task-queue.ts @@ -2,13 +2,11 @@ import "server-only"; import { and, asc, eq, lt, notInArray, sql } from "drizzle-orm"; import { db } from "@/lib/db"; -import { getSystemSetting } from "@/lib/db/queries"; import { canvasImageTask } from "@/lib/db/schema"; import { - CANVAS_IMAGE_TASK_QUEUE_CONFIG_KEY, - DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY, - normalizeCanvasImageTaskQueueConfig, -} from "./image-task-config"; + getCanvasImageSingleModelConcurrency, + normalizeCanvasImageModelKey, +} from "./image-concurrency"; import { generateCanvasImage, type CanvasGenerateImageRequest, @@ -30,49 +28,9 @@ interface QueueState { 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: - | { - expiresAt: number; - value: number; - } - | null = null; - -async function getQueueConcurrency(): Promise { - const now = Date.now(); - if (cachedConcurrency && cachedConcurrency.expiresAt > now) { - return cachedConcurrency.value; - } - - const envConcurrency = process.env.CANVAS_IMAGE_TASK_CONCURRENCY; - let value = normalizeCanvasImageTaskQueueConfig(undefined, { - envConcurrency, - }).concurrency; - - try { - const setting = await getSystemSetting({ - key: CANVAS_IMAGE_TASK_QUEUE_CONFIG_KEY, - }); - if (setting?.value !== undefined && setting.value !== null) { - value = normalizeCanvasImageTaskQueueConfig(setting.value, { - envConcurrency, - }).concurrency; - } - } catch (error) { - console.warn("[Canvas Image Task Queue] Failed to load DB concurrency config, using fallback", { - error: error instanceof Error ? error.message : String(error), - }); - } - - cachedConcurrency = { - expiresAt: now + CONCURRENCY_CONFIG_CACHE_MS, - value: value || DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY, - }; - return cachedConcurrency.value; -} function getQueueState(): QueueState { const globalObj = globalThis as typeof globalThis & { @@ -96,12 +54,7 @@ function normalizeTaskModelKey(request: unknown) { } const rawModel = (request as Record).model; - if (typeof rawModel !== "string") { - return "__default__"; - } - - const normalizedModel = rawModel.trim(); - return normalizedModel || "__default__"; + return normalizeCanvasImageModelKey(rawModel); } async function getTaskModelKey(taskId: string) { @@ -268,7 +221,7 @@ export async function drainCanvasImageTaskQueue() { state.draining = true; try { - const maxConcurrentPerModel = await getQueueConcurrency(); + const maxConcurrentPerModel = await getCanvasImageSingleModelConcurrency(); while (true) { const nextTask = await claimNextPendingTask(maxConcurrentPerModel); if (!nextTask) break;