From 94a22f0cc6a7fb990b80a9118a8219c8d3b3edac Mon Sep 17 00:00:00 2001 From: Your Name Date: Fri, 1 May 2026 01:14:07 +0800 Subject: [PATCH] feat: add image batch dispatch pacing --- .../settings/admin-settings-client.tsx | 98 +++++++++++----- lib/canvas/image-concurrency.ts | 109 +++++++++++++++--- lib/canvas/image-task-config.ts | 84 +++++++++++++- lib/canvas/image-task-queue.ts | 81 ++++++++++++- 4 files changed, 317 insertions(+), 55 deletions(-) diff --git a/app/(admin)/settings/admin-settings-client.tsx b/app/(admin)/settings/admin-settings-client.tsx index aa55a06b7..4eff9106d 100644 --- a/app/(admin)/settings/admin-settings-client.tsx +++ b/app/(admin)/settings/admin-settings-client.tsx @@ -78,8 +78,11 @@ import { } from "@/lib/model3d/model-alias-config"; import { CANVAS_IMAGE_TASK_QUEUE_CONFIG_KEY, + DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS, DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY, + MAX_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS, MIN_CANVAS_IMAGE_TASK_CONCURRENCY, + MIN_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS, MAX_CANVAS_IMAGE_TASK_CONCURRENCY, normalizeCanvasImageTaskQueueConfig, type CanvasImageTaskQueueConfig, @@ -2087,6 +2090,7 @@ export function AdminSettingsClient() { useState(() => normalizeCanvasImageTaskQueueConfig({ concurrency: DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY, + batchDispatchIntervalMs: DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS, }) ); @@ -5110,10 +5114,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 : "保存全站生图并发与批次间隔配置失败" ); } }; @@ -20155,10 +20159,10 @@ export function AdminSettingsClient() {
- 全站生图单模型并发 + 全站生图单模型并发与批次间隔 - 控制全站生图链路中，同一个模型允许同时运行的任务数。配置保存到数据库，异步队列和直调生成都会自动读取，无需重启服务。 + 控制全站生图链路中，同一个模型允许同时运行的任务数，以及排队后下一批任务的发起间隔。配置保存到数据库，异步队列和直调生成都会自动读取，无需重启服务。
• 影响画布异步生图、智能体直调生图、视频分镜高清预览等全站生图入口。

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

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

+

• 批次间隔只控制排队补位的“下一批”发起节奏，不影响首批任务即时启动。

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

-
- - - setCanvasImageTaskQueueConfig( - normalizeCanvasImageTaskQueueConfig({ - concurrency: e.target.value, - }) - ) - } - /> -
- 可配置范围 {MIN_CANVAS_IMAGE_TASK_CONCURRENCY}- - {MAX_CANVAS_IMAGE_TASK_CONCURRENCY}，当前为{" "} - {canvasImageTaskQueueConfig.concurrency} 并发，默认值为{" "} - {DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY}。 +
+
+ + + setCanvasImageTaskQueueConfig( + normalizeCanvasImageTaskQueueConfig({ + ...canvasImageTaskQueueConfig, + concurrency: e.target.value, + }) + ) + } + /> +
+ 可配置范围 {MIN_CANVAS_IMAGE_TASK_CONCURRENCY}- + {MAX_CANVAS_IMAGE_TASK_CONCURRENCY}，当前为{" "} + {canvasImageTaskQueueConfig.concurrency} 并发，默认值为{" "} + {DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY}。 +
+
+ +
+ + + setCanvasImageTaskQueueConfig( + normalizeCanvasImageTaskQueueConfig({ + ...canvasImageTaskQueueConfig, + batchDispatchIntervalMs: Math.round( + Number.parseFloat(e.target.value || "0") * 1000 + ), + }) + ) + } + /> +
+ 可配置范围 {MIN_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS / 1000}- + {MAX_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS / 1000} 秒，当前为{" "} + {(canvasImageTaskQueueConfig.batchDispatchIntervalMs / 1000).toFixed(1)}{" "} + 秒，默认值为{" "} + {(DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS / 1000).toFixed(1)}{" "} + 秒。 +
diff --git a/lib/canvas/image-concurrency.ts b/lib/canvas/image-concurrency.ts index 97011002e..50f8af110 100644 --- a/lib/canvas/image-concurrency.ts +++ b/lib/canvas/image-concurrency.ts @@ -3,7 +3,9 @@ import "server-only"; import { getSystemSetting } from "@/lib/db/queries"; import { CANVAS_IMAGE_TASK_QUEUE_CONFIG_KEY, + DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS, DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY, + type CanvasImageTaskQueueConfig, normalizeCanvasImageTaskQueueConfig, } from "./image-task-config"; @@ -18,16 +20,18 @@ interface ModelWaiter { interface ModelConcurrencyState { running: Set; waiters: ModelWaiter[]; + dispatchTimer: ReturnType | null; + delayedUntil: number | null; } interface CanvasImageConcurrencyState { models: Map; } -let cachedConcurrency: +let cachedConfig: | { expiresAt: number; - value: number; + value: CanvasImageTaskQueueConfig; } | null = null; @@ -55,11 +59,22 @@ function getOrCreateModelState(modelKey: string) { const created: ModelConcurrencyState = { running: new Set(), waiters: [], + dispatchTimer: null, + delayedUntil: null, }; state.models.set(modelKey, created); return created; } +function clearModelDispatchTimer(modelState: ModelConcurrencyState) { + if (modelState.dispatchTimer) { + clearTimeout(modelState.dispatchTimer); + modelState.dispatchTimer = null; + } + + modelState.delayedUntil = null; +} + function cleanupModelState(modelKey: string) { const state = getGlobalConcurrencyState(); const modelState = state.models.get(modelKey); @@ -68,6 +83,7 @@ function cleanupModelState(modelKey: string) { } if (modelState.running.size === 0 && modelState.waiters.length === 0) { + clearModelDispatchTimer(modelState); state.models.delete(modelKey); } } @@ -81,25 +97,28 @@ export function normalizeCanvasImageModelKey(value: unknown) { return normalized || "__default__"; } -export async function getCanvasImageSingleModelConcurrency(): Promise { +export async function getCanvasImageTaskQueueConfig(): Promise { const now = Date.now(); - if (cachedConcurrency && cachedConcurrency.expiresAt > now) { - return cachedConcurrency.value; + if (cachedConfig && cachedConfig.expiresAt > now) { + return cachedConfig.value; } const envConcurrency = process.env.CANVAS_IMAGE_TASK_CONCURRENCY; - let value = normalizeCanvasImageTaskQueueConfig(undefined, { + const envBatchDispatchIntervalMs = process.env.CANVAS_IMAGE_TASK_BATCH_DISPATCH_INTERVAL_MS; + let config = normalizeCanvasImageTaskQueueConfig(undefined, { envConcurrency, - }).concurrency; + envBatchDispatchIntervalMs, + }); try { const setting = await getSystemSetting({ key: CANVAS_IMAGE_TASK_QUEUE_CONFIG_KEY, }); if (setting?.value !== undefined && setting.value !== null) { - value = normalizeCanvasImageTaskQueueConfig(setting.value, { + config = normalizeCanvasImageTaskQueueConfig(setting.value, { envConcurrency, - }).concurrency; + envBatchDispatchIntervalMs, + }); } } catch (error) { console.warn("[Canvas Image Concurrency] Failed to load DB config, using fallback", { @@ -107,12 +126,25 @@ export async function getCanvasImageSingleModelConcurrency(): Promise { }); } - cachedConcurrency = { + cachedConfig = { expiresAt: now + CONCURRENCY_CONFIG_CACHE_MS, - value: value || DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY, + value: { + concurrency: config.concurrency || DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY, + batchDispatchIntervalMs: Number.isFinite(config.batchDispatchIntervalMs) + ? config.batchDispatchIntervalMs + : DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS, + }, }; - return cachedConcurrency.value; + return cachedConfig.value; +} + +export async function getCanvasImageSingleModelConcurrency(): Promise { + return (await getCanvasImageTaskQueueConfig()).concurrency; +} + +export async function getCanvasImageBatchDispatchIntervalMs(): Promise { + return (await getCanvasImageTaskQueueConfig()).batchDispatchIntervalMs; } export function getCanvasImageRunningCountForModel(modelKey: string) { @@ -120,7 +152,10 @@ export function getCanvasImageRunningCountForModel(modelKey: string) { return getGlobalConcurrencyState().models.get(normalizedModelKey)?.running.size || 0; } -async function drainModelWaiters(modelKey: string) { +async function drainModelWaiters( + modelKey: string, + trigger: "arrival" | "slot-release" | "timer" +) { const normalizedModelKey = normalizeCanvasImageModelKey(modelKey); const modelState = getGlobalConcurrencyState().models.get(normalizedModelKey); if (!modelState || modelState.waiters.length === 0) { @@ -128,7 +163,31 @@ async function drainModelWaiters(modelKey: string) { return; } - const maxConcurrent = await getCanvasImageSingleModelConcurrency(); + const { concurrency: maxConcurrent, batchDispatchIntervalMs } = + await getCanvasImageTaskQueueConfig(); + if (modelState.running.size >= maxConcurrent) { + return; + } + + const now = Date.now(); + if (modelState.delayedUntil && modelState.delayedUntil > now) { + return; + } + + if (trigger === "slot-release" && batchDispatchIntervalMs > 0) { + if (!modelState.dispatchTimer) { + modelState.delayedUntil = now + batchDispatchIntervalMs; + modelState.dispatchTimer = setTimeout(() => { + const state = getGlobalConcurrencyState().models.get(normalizedModelKey); + if (!state) return; + clearModelDispatchTimer(state); + void drainModelWaiters(normalizedModelKey, "timer"); + }, batchDispatchIntervalMs); + } + return; + } + + clearModelDispatchTimer(modelState); while (modelState.running.size < maxConcurrent && modelState.waiters.length > 0) { const waiter = modelState.waiters.shift(); if (!waiter) { @@ -147,22 +206,32 @@ export async function acquireCanvasImageModelSlot(params: { debugMeta?: Record; }) { const normalizedModelKey = normalizeCanvasImageModelKey(params.modelKey); - const maxConcurrent = await getCanvasImageSingleModelConcurrency(); + const { + concurrency: maxConcurrent, + batchDispatchIntervalMs, + } = await getCanvasImageTaskQueueConfig(); const modelState = getOrCreateModelState(normalizedModelKey); const ticketId = `${normalizedModelKey}:${Date.now()}:${Math.random().toString(36).slice(2, 10)}`; const waitingSince = Date.now(); + const hasActiveDelay = + modelState.delayedUntil !== null && modelState.delayedUntil > waitingSince; - if (modelState.waiters.length === 0 && modelState.running.size < maxConcurrent) { + if ( + modelState.waiters.length === 0 && + !hasActiveDelay && + 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); + await drainModelWaiters(normalizedModelKey, "slot-release"); }, waitMs: 0, maxConcurrent, + batchDispatchIntervalMs, }; } @@ -171,7 +240,7 @@ export async function acquireCanvasImageModelSlot(params: { ticketId, resolve, }); - void drainModelWaiters(normalizedModelKey); + void drainModelWaiters(normalizedModelKey, "arrival"); }); const waitMs = Date.now() - waitingSince; @@ -180,6 +249,7 @@ export async function acquireCanvasImageModelSlot(params: { debugLabel: params.debugLabel || "canvas-image", modelKey: normalizedModelKey, waitMs, + batchDispatchIntervalMs, ...params.debugMeta, }); } @@ -189,9 +259,10 @@ export async function acquireCanvasImageModelSlot(params: { const state = getGlobalConcurrencyState().models.get(normalizedModelKey); if (!state) return; state.running.delete(ticketId); - await drainModelWaiters(normalizedModelKey); + await drainModelWaiters(normalizedModelKey, "slot-release"); }, waitMs, maxConcurrent, + batchDispatchIntervalMs, }; } diff --git a/lib/canvas/image-task-config.ts b/lib/canvas/image-task-config.ts index ed3a031fe..808ea903a 100644 --- a/lib/canvas/image-task-config.ts +++ b/lib/canvas/image-task-config.ts @@ -3,9 +3,13 @@ export const CANVAS_IMAGE_TASK_QUEUE_CONFIG_KEY = "canvas_image_task_queue_confi 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 const MIN_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS = 0; +export const DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS = 0; +export const MAX_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS = 60 * 1000; export interface CanvasImageTaskQueueConfig { concurrency: number; + batchDispatchIntervalMs: number; } function clampConcurrency(value: number) { @@ -15,6 +19,13 @@ function clampConcurrency(value: number) { ); } +function clampBatchDispatchIntervalMs(value: number) { + return Math.min( + MAX_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS, + Math.max(MIN_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS, Math.floor(value)) + ); +} + function parseConcurrencyCandidate(value: unknown): number | null { if (typeof value === "number" && Number.isFinite(value)) { return clampConcurrency(value); @@ -30,14 +41,38 @@ function parseConcurrencyCandidate(value: unknown): number | null { return null; } +function parseBatchDispatchIntervalCandidate( + value: unknown, + options?: { treatAsSeconds?: boolean } +): number | null { + const normalize = (candidate: number) => + clampBatchDispatchIntervalMs( + options?.treatAsSeconds ? candidate * 1000 : candidate + ); + + if (typeof value === "number" && Number.isFinite(value)) { + return normalize(value); + } + + if (typeof value === "string" && value.trim()) { + const parsed = Number.parseFloat(value.trim()); + if (Number.isFinite(parsed)) { + return normalize(parsed); + } + } + + return null; +} + export function normalizeCanvasImageTaskQueueConfig( value: unknown, - options?: { envConcurrency?: unknown } + options?: { envConcurrency?: unknown; envBatchDispatchIntervalMs?: unknown } ): CanvasImageTaskQueueConfig { const raw = value && typeof value === "object" && !Array.isArray(value) ? (value as Record) : {}; - const candidates = [ + + const concurrencyCandidates = [ raw.concurrency, raw.maxConcurrency, raw.maxConcurrent, @@ -45,12 +80,51 @@ export function normalizeCanvasImageTaskQueueConfig( options?.envConcurrency, ]; - for (const candidate of candidates) { + let concurrency = DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY; + for (const candidate of concurrencyCandidates) { const parsed = parseConcurrencyCandidate(candidate); if (parsed !== null) { - return { concurrency: parsed }; + concurrency = parsed; + break; } } - return { concurrency: DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY }; + const intervalCandidates = [ + raw.batchDispatchIntervalMs, + raw.dispatchIntervalMs, + raw.batchDispatchDelayMs, + raw.batchDispatchInterval, + raw.dispatchInterval, + options?.envBatchDispatchIntervalMs, + ]; + const intervalSecondsCandidates = [ + raw.batchDispatchIntervalSeconds, + raw.dispatchIntervalSeconds, + ]; + + let batchDispatchIntervalMs = DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS; + for (const candidate of intervalCandidates) { + const parsed = parseBatchDispatchIntervalCandidate(candidate); + if (parsed !== null) { + batchDispatchIntervalMs = parsed; + break; + } + } + + if (batchDispatchIntervalMs === DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS) { + for (const candidate of intervalSecondsCandidates) { + const parsed = parseBatchDispatchIntervalCandidate(candidate, { + treatAsSeconds: true, + }); + if (parsed !== null) { + batchDispatchIntervalMs = parsed; + break; + } + } + } + + return { + concurrency, + batchDispatchIntervalMs, + }; } diff --git a/lib/canvas/image-task-queue.ts b/lib/canvas/image-task-queue.ts index f80426b2d..34d1273ac 100644 --- a/lib/canvas/image-task-queue.ts +++ b/lib/canvas/image-task-queue.ts @@ -4,6 +4,7 @@ import { and, asc, eq, lt, notInArray, sql } from "drizzle-orm"; import { db } from "@/lib/db"; import { canvasImageTask } from "@/lib/db/schema"; import { + getCanvasImageBatchDispatchIntervalMs, getCanvasImageSingleModelConcurrency, normalizeCanvasImageModelKey, } from "./image-concurrency"; @@ -25,6 +26,9 @@ interface QueueState { pendingSet: Set; running: Map; runningByModel: Map>; + nextBatchEligibleAt: Map; + nextBatchTimers: Map>; + awaitingNextBatch: Set; draining: boolean; } @@ -42,6 +46,9 @@ function getQueueState(): QueueState { pendingSet: new Set(), running: new Map(), runningByModel: new Map>(), + nextBatchEligibleAt: new Map(), + nextBatchTimers: new Map>(), + awaitingNextBatch: new Set(), draining: false, }; } @@ -84,7 +91,7 @@ function releaseRunningTask(taskId: string) { const state = getQueueState(); const modelKey = state.running.get(taskId); if (!modelKey) { - return; + return null; } state.running.delete(taskId); @@ -97,6 +104,55 @@ function releaseRunningTask(taskId: string) { if (tasks.size === 0) { state.runningByModel.delete(modelKey); } + + return modelKey; +} + +function isModelBatchDispatchDelayed(modelKey: string) { + const state = getQueueState(); + const delayedUntil = state.nextBatchEligibleAt.get(modelKey); + if (!delayedUntil) { + return false; + } + + if (delayedUntil <= Date.now()) { + state.nextBatchEligibleAt.delete(modelKey); + const timer = state.nextBatchTimers.get(modelKey); + if (timer) { + clearTimeout(timer); + state.nextBatchTimers.delete(modelKey); + } + return false; + } + + return true; +} + +function markModelAwaitingNextBatch(modelKey: string) { + getQueueState().awaitingNextBatch.add(modelKey); +} + +function scheduleModelNextBatch(modelKey: string, delayMs: number) { + const state = getQueueState(); + if (delayMs <= 0) { + state.awaitingNextBatch.delete(modelKey); + return false; + } + + if (state.nextBatchTimers.has(modelKey)) { + return true; + } + + state.awaitingNextBatch.delete(modelKey); + state.nextBatchEligibleAt.set(modelKey, Date.now() + delayMs); + const timer = setTimeout(() => { + const latestState = getQueueState(); + latestState.nextBatchTimers.delete(modelKey); + latestState.nextBatchEligibleAt.delete(modelKey); + void drainCanvasImageTaskQueue(); + }, delayMs); + state.nextBatchTimers.set(modelKey, timer); + return true; } async function enqueueLocalTask(taskId: string, modelKey?: string) { @@ -111,6 +167,11 @@ async function enqueueLocalTask(taskId: string, modelKey?: string) { modelKey: resolvedModelKey, }); state.pendingSet.add(taskId); + + const maxConcurrentPerModel = await getCanvasImageSingleModelConcurrency(); + if (getRunningCountForModel(resolvedModelKey) >= maxConcurrentPerModel) { + markModelAwaitingNextBatch(resolvedModelKey); + } } function hasModelCapacity(modelKey: string, maxConcurrentPerModel: number) { @@ -126,7 +187,11 @@ async function claimNextPendingTask( const entry = state.pending[index]; if (!entry) continue; if (state.running.has(entry.taskId)) continue; - if (!hasModelCapacity(entry.modelKey, maxConcurrentPerModel)) continue; + if (isModelBatchDispatchDelayed(entry.modelKey)) continue; + if (!hasModelCapacity(entry.modelKey, maxConcurrentPerModel)) { + markModelAwaitingNextBatch(entry.modelKey); + continue; + } state.pending.splice(index, 1); state.pendingSet.delete(entry.taskId); @@ -153,7 +218,11 @@ async function claimNextPendingTask( if (state.pendingSet.has(row.id) || state.running.has(row.id)) continue; const modelKey = normalizeTaskModelKey(row.request); + if (isModelBatchDispatchDelayed(modelKey)) { + continue; + } if (!hasModelCapacity(modelKey, maxConcurrentPerModel)) { + markModelAwaitingNextBatch(modelKey); continue; } @@ -210,7 +279,13 @@ async function runSingleTask(entry: PendingTaskEntry) { }) .where(eq(canvasImageTask.id, entry.taskId)); } finally { - releaseRunningTask(entry.taskId); + const modelKey = releaseRunningTask(entry.taskId); + if (modelKey) { + const batchDispatchIntervalMs = await getCanvasImageBatchDispatchIntervalMs(); + if (getQueueState().awaitingNextBatch.has(modelKey)) { + scheduleModelNextBatch(modelKey, batchDispatchIntervalMs); + } + } void drainCanvasImageTaskQueue(); } }