From 231be1545977708fd41baa72a6fb3819d48323df Mon Sep 17 00:00:00 2001 From: 134 Date: Mon, 3 Aug 2026 14:43:21 +0800 Subject: [PATCH] fix: release image queue slots after dispatch --- app/(chat)/api/chat/_lib/stream.ts | 6 + app/(chat)/api/chat/route.ts | 8 ++ app/api/scene-fusion/route.ts | 16 ++- components/canvas/CanvasChatPanel.tsx | 10 ++ .../canvas/chat-panel/canvas-agent-stream.ts | 1 + components/image-generation-loading.tsx | 13 ++- components/scene-fusion-view.tsx | 21 ++-- components/style-pro/StyleProWorkspace.tsx | 16 +-- lib/canvas/generate-image.ts | 110 +++++++++++------- lib/canvas/image-concurrency.ts | 100 +++++++++++----- lib/canvas/image-queue-fairness.ts | 11 ++ lib/canvas/image-task-queue.ts | 107 ++++++----------- lib/canvas/layout-replace-task.ts | 2 + lib/style-pro/task.ts | 4 + .../canvas-image-queue-fairness.test.ts | 24 +++- 15 files changed, 281 insertions(+), 168 deletions(-) diff --git a/app/(chat)/api/chat/_lib/stream.ts b/app/(chat)/api/chat/_lib/stream.ts index 2b999ad7a..d3cb3335c 100644 --- a/app/(chat)/api/chat/_lib/stream.ts +++ b/app/(chat)/api/chat/_lib/stream.ts @@ -214,6 +214,7 @@ export function writeImageGenerationStatusChunk( sameModelPendingCount?: number | null; sameModelRunningCount?: number | null; concurrencyLimit?: number | null; + dispatchIntervalMs?: number | null; estimatedWaitMs?: number | null; }, label: string @@ -249,6 +250,11 @@ export function writeImageGenerationStatusChunk( Number(params.concurrencyLimit) > 0 ? Number(params.concurrencyLimit) : undefined, + dispatchIntervalMs: + Number.isFinite(Number(params.dispatchIntervalMs)) && + Number(params.dispatchIntervalMs) >= 0 + ? Number(params.dispatchIntervalMs) + : undefined, estimatedWaitMs: Number.isFinite(Number(params.estimatedWaitMs)) && Number(params.estimatedWaitMs) >= 0 diff --git a/app/(chat)/api/chat/route.ts b/app/(chat)/api/chat/route.ts index 26231236f..092c94584 100644 --- a/app/(chat)/api/chat/route.ts +++ b/app/(chat)/api/chat/route.ts @@ -6040,6 +6040,8 @@ export async function POST(request: Request) { sameModelRunningCount: queueStatus.sameModelRunningCount, concurrencyLimit: queueStatus.concurrencyLimit, + dispatchIntervalMs: + queueStatus.dispatchIntervalMs, estimatedWaitMs: queueStatus.estimatedWaitMs, }, `agent-mode:detail-replacement:queue:${reference.id}`, @@ -6940,6 +6942,8 @@ export async function POST(request: Request) { queueStatus.sameModelPendingCount, sameModelRunningCount: queueStatus.sameModelRunningCount, concurrencyLimit: queueStatus.concurrencyLimit, + dispatchIntervalMs: + queueStatus.dispatchIntervalMs, estimatedWaitMs: queueStatus.estimatedWaitMs, }, `agent-preset:replacement:queue:${reference.id}`, @@ -7267,6 +7271,8 @@ export async function POST(request: Request) { queueStatus.sameModelPendingCount, sameModelRunningCount: queueStatus.sameModelRunningCount, concurrencyLimit: queueStatus.concurrencyLimit, + dispatchIntervalMs: + queueStatus.dispatchIntervalMs, estimatedWaitMs: queueStatus.estimatedWaitMs, }, "agent-preset:image:scene-generation:queue", @@ -7496,6 +7502,8 @@ export async function POST(request: Request) { queueStatus.sameModelPendingCount, sameModelRunningCount: queueStatus.sameModelRunningCount, concurrencyLimit: queueStatus.concurrencyLimit, + dispatchIntervalMs: + queueStatus.dispatchIntervalMs, estimatedWaitMs: queueStatus.estimatedWaitMs, }, `agent-preset:image:slot-queue:${slot.slotIndex}`, diff --git a/app/api/scene-fusion/route.ts b/app/api/scene-fusion/route.ts index 063c6eb96..5e5aa4acf 100644 --- a/app/api/scene-fusion/route.ts +++ b/app/api/scene-fusion/route.ts @@ -2134,6 +2134,7 @@ async function processSceneWithDiversity( sameModelPendingCount: number; sameModelRunningCount: number; concurrencyLimit: number; + dispatchIntervalMs: number; estimatedWaitMs: number; source: "model_queue"; }) => void | Promise; @@ -2217,12 +2218,13 @@ async function processSceneWithDiversity( onQueued: params.onQueueStatus, }); let result: Awaited>; + const runtimeModelId = resolveModelRuntimeId(candidate.model, { + providerType: candidate.provider.type, + }); + await params.onProcessingStart?.(); + let execution: ReturnType | null = null; try { - const runtimeModelId = resolveModelRuntimeId(candidate.model, { - providerType: candidate.provider.type, - }); - await params.onProcessingStart?.(); - result = await processSceneWithProvider( + execution = processSceneWithProvider( params.productImageUrl, params.sceneImageUrl, params.productName, @@ -2246,6 +2248,10 @@ async function processSceneWithDiversity( } finally { await modelSlot.release(); } + if (!execution) { + throw new Error("场景裂变请求未能提交到模型服务"); + } + result = await execution; const echoCheck = await detectEchoedProductResult({ requestId: params.requestId, diff --git a/components/canvas/CanvasChatPanel.tsx b/components/canvas/CanvasChatPanel.tsx index 01dff07e9..38d8a10d6 100644 --- a/components/canvas/CanvasChatPanel.tsx +++ b/components/canvas/CanvasChatPanel.tsx @@ -7190,6 +7190,11 @@ export function CanvasChatPanel({ Number(payload.concurrencyLimit) > 0 ? Number(payload.concurrencyLimit) : undefined, + dispatchIntervalMs: + Number.isFinite(Number(payload.dispatchIntervalMs)) && + Number(payload.dispatchIntervalMs) >= 0 + ? Number(payload.dispatchIntervalMs) + : undefined, estimatedWaitMs: Number.isFinite(Number(payload.estimatedWaitMs)) && Number(payload.estimatedWaitMs) >= 0 @@ -13699,6 +13704,8 @@ export function CanvasChatPanel({ const loadingSameModelRunningCount = streamImageGenerationState?.sameModelRunningCount; const loadingConcurrencyLimit = streamImageGenerationState?.concurrencyLimit; + const loadingDispatchIntervalMs = + streamImageGenerationState?.dispatchIntervalMs; const loadingEstimatedWaitMs = streamImageGenerationState?.estimatedWaitMs; const canvasGenerationPlaceholderState = useMemo(() => { @@ -13714,10 +13721,12 @@ export function CanvasChatPanel({ sameModelPendingCount: loadingSameModelPendingCount, sameModelRunningCount: loadingSameModelRunningCount, concurrencyLimit: loadingConcurrencyLimit, + dispatchIntervalMs: loadingDispatchIntervalMs, estimatedWaitMs: loadingEstimatedWaitMs, }; }, [ loadingConcurrencyLimit, + loadingDispatchIntervalMs, loadingEstimatedWaitMs, loadingImageCount, loadingModelName, @@ -15451,6 +15460,7 @@ export function CanvasChatPanel({ sameModelPendingCount={loadingSameModelPendingCount} sameModelRunningCount={loadingSameModelRunningCount} concurrencyLimit={loadingConcurrencyLimit} + dispatchIntervalMs={loadingDispatchIntervalMs} estimatedWaitMs={loadingEstimatedWaitMs} /> )} diff --git a/components/canvas/chat-panel/canvas-agent-stream.ts b/components/canvas/chat-panel/canvas-agent-stream.ts index 192765f7a..137a6caf3 100644 --- a/components/canvas/chat-panel/canvas-agent-stream.ts +++ b/components/canvas/chat-panel/canvas-agent-stream.ts @@ -158,6 +158,7 @@ export type ImageGenerationStatusStreamPayload = { sameModelPendingCount?: number; sameModelRunningCount?: number; concurrencyLimit?: number; + dispatchIntervalMs?: number; estimatedWaitMs?: number; }; diff --git a/components/image-generation-loading.tsx b/components/image-generation-loading.tsx index 52b597454..5ee08e5ca 100644 --- a/components/image-generation-loading.tsx +++ b/components/image-generation-loading.tsx @@ -18,6 +18,7 @@ interface ImageGenerationLoadingProps { sameModelPendingCount?: number; sameModelRunningCount?: number; concurrencyLimit?: number; + dispatchIntervalMs?: number; estimatedWaitMs?: number; } @@ -110,7 +111,7 @@ export const ImageGenerationLoading = memo( queuePosition, sameModelPendingCount, sameModelRunningCount, - concurrencyLimit, + dispatchIntervalMs, estimatedWaitMs, }: ImageGenerationLoadingProps) => { const shouldTrackElapsed = @@ -164,10 +165,16 @@ export const ImageGenerationLoading = memo( ? `队列共 ${Math.max( 0, Number(sameModelPendingCount || 0) - )} 个待处理，当前模型运行 ${Math.max( + )} 个待提交，已提交 ${Math.max( 0, Number(sameModelRunningCount || 0) - )}/${Math.max(1, Number(concurrencyLimit || 0) || 1)}，${formatImageQueueWaitLabel( + )} 个生成任务，${ + Number(dispatchIntervalMs || 0) > 0 + ? `每 ${Math.max(0.1, Number(dispatchIntervalMs) / 1000).toFixed( + Number(dispatchIntervalMs) % 1000 === 0 ? 0 : 1 + )} 秒提交 1 个，` + : "" + }${formatImageQueueWaitLabel( Math.max( 0, Number(estimatedWaitMs || 0) - resolvedElapsedSeconds * 1000 diff --git a/components/scene-fusion-view.tsx b/components/scene-fusion-view.tsx index cbca236f9..27e705ef2 100644 --- a/components/scene-fusion-view.tsx +++ b/components/scene-fusion-view.tsx @@ -114,6 +114,7 @@ interface ProcessItem { sameModelPendingCount?: number; sameModelRunningCount?: number; concurrencyLimit?: number; + dispatchIntervalMs?: number; estimatedWaitMs?: number; source?: "model_queue" | "batch_queue"; }; @@ -3067,10 +3068,6 @@ export function SceneFusionView({ 0, Number(item.queueStatus?.sameModelRunningCount || 0) ); - const concurrencyLimit = Math.max( - 0, - Number(item.queueStatus?.concurrencyLimit || 0) - ); const serverQueuePosition = Math.max( 0, Number(item.queueStatus?.queuePosition || 0) @@ -3094,11 +3091,11 @@ export function SceneFusionView({ const queueDetail = item.status === "pending" ? serverQueuePosition > 0 - ? `当前模型运行 ${serverRunningCount}/${Math.max( - concurrencyLimit, - serverRunningCount, - 1 - )}，${formatQueueWaitLabel( + ? `前方任务将按 ${Math.max( + 0.1, + Number(item.queueStatus?.dispatchIntervalMs || 0) / + 1000 + )} 秒间隔提交，${formatQueueWaitLabel( item.queueStatus?.estimatedWaitMs )}` : localBatchQueueMeta @@ -3106,11 +3103,7 @@ export function SceneFusionView({ : "点击开始批量处理后进入队列" : item.status === "processing" && serverQueuePosition > 0 - ? `当前模型运行 ${serverRunningCount}/${Math.max( - concurrencyLimit, - serverRunningCount, - 1 - )}，${formatQueueWaitLabel( + ? `已提交 ${serverRunningCount} 个生成任务，${formatQueueWaitLabel( item.queueStatus?.estimatedWaitMs )}` : null; diff --git a/components/style-pro/StyleProWorkspace.tsx b/components/style-pro/StyleProWorkspace.tsx index 5fc2d065f..4b670fd6a 100644 --- a/components/style-pro/StyleProWorkspace.tsx +++ b/components/style-pro/StyleProWorkspace.tsx @@ -112,6 +112,7 @@ type StyleProGenerateResponse = { sameModelPendingCount?: number; sameModelRunningCount?: number; concurrencyLimit?: number; + dispatchIntervalMs?: number; estimatedWaitMs?: number; source?: "model_queue" | "batch_queue"; }; @@ -709,10 +710,6 @@ function GeneratingImagePlaceholder({ 0, Number(queueStatus?.sameModelRunningCount || 0) ); - const concurrencyLimit = Math.max( - 0, - Number(queueStatus?.concurrencyLimit || 0) - ); const isQueued = taskState?.status === "pending"; const isFailed = taskState?.status === "failed"; const title = isFailed @@ -725,9 +722,14 @@ function GeneratingImagePlaceholder({ const description = isFailed ? taskState?.error || "请调整参数后重试。" : isQueued - ? `当前模型运行 ${sameModelRunningCount}/${concurrencyLimit || "?"}，${formatStyleProEstimatedWait( - queueStatus?.estimatedWaitMs - )}` + ? `已提交 ${sameModelRunningCount} 个生成任务，${ + Number(queueStatus?.dispatchIntervalMs || 0) > 0 + ? `每 ${Math.max( + 0.1, + Number(queueStatus?.dispatchIntervalMs || 0) / 1000 + )} 秒提交 1 个，` + : "" + }${formatStyleProEstimatedWait(queueStatus?.estimatedWaitMs)}` : taskState?.generationMode === "realistic" ? "系统会先完成基础实景化，再合成最终风格结果。" : "模型已开始执行，结果完成后会自动回填。"; diff --git a/lib/canvas/generate-image.ts b/lib/canvas/generate-image.ts index 2be113087..c1af875c1 100644 --- a/lib/canvas/generate-image.ts +++ b/lib/canvas/generate-image.ts @@ -139,6 +139,7 @@ export interface CanvasGenerateImageRequest { modelRef?: string | null; } ) => void | Promise; + onRequestDispatched?: () => void | Promise; } export interface ParsedImage { @@ -4256,41 +4257,9 @@ export async function generateCanvasImage( const resolvedModelKey = normalizeCanvasImageModelKey( selectedModel.model.modelId || selectedModel.model.id || model ); - const slot = await acquireCanvasImageModelSlot({ - modelKey: resolvedModelKey, - ownerKey, - debugLabel, - debugMeta: { - selectedModelId: selectedModel.model.modelId, - selectedModelRef: selectedModel.model.id, - providerType: provider.type, - ...debugMeta, - }, - onQueued: request.onQueueStatusChange - ? async (queueStatus) => { - await request.onQueueStatusChange?.({ - ...queueStatus, - modelName: selectedModel.model.displayName || null, - modelRef: selectedModel.model.id || null, - }); - } - : undefined, - }); - - if (slot.waitMs > 0 && request.onQueueStatusChange) { - const runningStatus = await getCanvasImageModelQueueStatus({ - modelKey: resolvedModelKey, - }); - await request.onQueueStatusChange({ - ...runningStatus, - status: "running", - queuePosition: 0, - estimatedWaitMs: 0, - modelName: selectedModel.model.displayName || null, - modelRef: selectedModel.model.id || null, - }); - } - + let totalQueueWaitMs = 0; + let singleModelConcurrency = 1; + let requestDispatchNotified = false; let caughtError: unknown = null; try { @@ -4298,6 +4267,44 @@ export async function generateCanvasImage( const remaining = targetImageCount - collectedImages.length; if (remaining <= 0) break; const batchImageCount = Math.min(providerBatchLimit, remaining); + const slot = await acquireCanvasImageModelSlot({ + modelKey: resolvedModelKey, + ownerKey, + debugLabel, + debugMeta: { + selectedModelId: selectedModel.model.modelId, + selectedModelRef: selectedModel.model.id, + providerType: provider.type, + batchIndex: batchIndex + 1, + ...debugMeta, + }, + onQueued: request.onQueueStatusChange + ? async (queueStatus) => { + await request.onQueueStatusChange?.({ + ...queueStatus, + modelName: selectedModel.model.displayName || null, + modelRef: selectedModel.model.id || null, + }); + } + : undefined, + }); + totalQueueWaitMs += slot.waitMs; + singleModelConcurrency = slot.maxConcurrent; + + if (slot.waitMs > 0 && request.onQueueStatusChange) { + const runningStatus = await getCanvasImageModelQueueStatus({ + modelKey: resolvedModelKey, + }); + await request.onQueueStatusChange({ + ...runningStatus, + status: "running", + queuePosition: 0, + estimatedWaitMs: 0, + modelName: selectedModel.model.displayName || null, + modelRef: selectedModel.model.id || null, + }); + } + const batchStartedAt = Date.now(); console.info("[Canvas Generate Image] Batch started:", { debugLabel, @@ -4309,8 +4316,29 @@ export async function generateCanvasImage( selectedModelId: selectedModel.model.modelId, ...debugMeta, }); + let batchExecution: Promise; + try { + batchExecution = executeGenerateBatch(batchImageCount); + } finally { + await slot.release(); + } + if (!requestDispatchNotified) { + requestDispatchNotified = true; + await Promise.resolve(request.onRequestDispatched?.()).catch( + (dispatchCallbackError) => { + console.warn("[Canvas Generate Image] Dispatch callback failed:", { + debugLabel, + error: + dispatchCallbackError instanceof Error + ? dispatchCallbackError.message + : String(dispatchCallbackError), + ...debugMeta, + }); + } + ); + } const imageMarkdown = await withCanvasProviderBatchTimeout( - executeGenerateBatch(batchImageCount), + batchExecution, { debugLabel, modelId: selectedModel.model.modelId, @@ -4362,8 +4390,8 @@ export async function generateCanvasImage( console.info("[Canvas Generate Image] Completed:", { debugLabel, elapsedMs: Date.now() - startedAt, - waitMs: slot.waitMs, - singleModelConcurrency: slot.maxConcurrent, + waitMs: totalQueueWaitMs, + singleModelConcurrency, finalImageCount: finalImages.length, providerType: provider.type, selectedModelRef: selectedModel.model.id, @@ -4396,8 +4424,8 @@ export async function generateCanvasImage( console.error("[Canvas Generate Image] Failed:", { debugLabel, elapsedMs: Date.now() - startedAt, - waitMs: slot.waitMs, - singleModelConcurrency: slot.maxConcurrent, + waitMs: totalQueueWaitMs, + singleModelConcurrency, providerType: provider.type, selectedModelRef: selectedModel.model.id, selectedModelId: selectedModel.model.modelId, @@ -4405,8 +4433,6 @@ export async function generateCanvasImage( ...debugMeta, }); caughtError = error; - } finally { - await slot.release(); } const retryVariants = Array.isArray(request.retryVariants) diff --git a/lib/canvas/image-concurrency.ts b/lib/canvas/image-concurrency.ts index 3233adfe2..858274aca 100644 --- a/lib/canvas/image-concurrency.ts +++ b/lib/canvas/image-concurrency.ts @@ -8,7 +8,10 @@ import { type CanvasImageTaskQueueConfig, normalizeCanvasImageTaskQueueConfig, } from "./image-task-config"; -import { getAdaptiveImageQueueOwnerLimit } from "./image-queue-fairness"; +import { + estimateImageDispatchWaitMs, + getAdaptiveImageQueueOwnerLimit, +} from "./image-queue-fairness"; const CONCURRENCY_CONFIG_CACHE_MS = 5 * 1000; const RECENT_EXECUTION_SAMPLE_LIMIT = 24; @@ -95,7 +98,9 @@ function cleanupModelState(modelKey: string) { if ( modelState.running.size === 0 && modelState.waiters.length === 0 && - modelState.recentExecutionDurationsMs.length === 0 + modelState.recentExecutionDurationsMs.length === 0 && + !modelState.dispatchTimer && + (!modelState.delayedUntil || modelState.delayedUntil <= Date.now()) ) { clearModelDispatchTimer(modelState); state.models.delete(modelKey); @@ -201,6 +206,7 @@ export type CanvasImageModelQueueStatus = { sameModelPendingCount: number; sameModelRunningCount: number; concurrencyLimit: number; + dispatchIntervalMs: number; estimatedWaitMs: number; source: "model_queue"; }; @@ -222,16 +228,28 @@ function buildModelQueueSnapshot(params: { sameModelPendingCount: number; sameModelRunningCount: number; concurrencyLimit: number; + dispatchIntervalMs: number; }): CanvasImageModelQueueStatus { const averageExecutionMs = getRecentAverageExecutionMs(params.modelState); + const delayRemainingMs = Math.max( + 0, + Number(params.modelState.delayedUntil || 0) - Date.now() + ); const estimatedWaitMs = params.sameModelPendingCount <= 0 ? 0 - : Math.max( - averageExecutionMs, - Math.ceil((params.queuePosition + 1) / Math.max(1, params.concurrencyLimit)) * - averageExecutionMs - ); + : params.dispatchIntervalMs > 0 + ? estimateImageDispatchWaitMs({ + queuePosition: params.queuePosition, + dispatchIntervalMs: params.dispatchIntervalMs, + delayRemainingMs, + }) + : Math.max( + averageExecutionMs, + Math.ceil( + (params.queuePosition + 1) / Math.max(1, params.concurrencyLimit) + ) * averageExecutionMs + ); return { status: "pending", @@ -239,6 +257,7 @@ function buildModelQueueSnapshot(params: { sameModelPendingCount: Math.max(0, params.sameModelPendingCount), sameModelRunningCount: Math.max(0, params.sameModelRunningCount), concurrencyLimit: Math.max(1, params.concurrencyLimit), + dispatchIntervalMs: Math.max(0, params.dispatchIntervalMs), estimatedWaitMs, source: "model_queue", }; @@ -247,7 +266,8 @@ function buildModelQueueSnapshot(params: { function notifyModelWaiters( modelKey: string, modelState: ModelConcurrencyState, - concurrencyLimit: number + concurrencyLimit: number, + dispatchIntervalMs: number ) { const sameModelPendingCount = modelState.waiters.length; for (const [queuePosition, waiter] of modelState.waiters.entries()) { @@ -261,12 +281,14 @@ function notifyModelWaiters( sameModelPendingCount, sameModelRunningCount: modelState.running.size, concurrencyLimit, + dispatchIntervalMs, }); const statusSignature = [ queueStatus.queuePosition, queueStatus.sameModelPendingCount, queueStatus.sameModelRunningCount, queueStatus.concurrencyLimit, + queueStatus.dispatchIntervalMs, queueStatus.estimatedWaitMs, ].join(":"); if (waiter.lastStatusSignature === statusSignature) { @@ -289,7 +311,8 @@ export async function getCanvasImageModelQueueStatus(params: { modelKey: string; }): Promise { const normalizedModelKey = normalizeCanvasImageModelKey(params.modelKey); - const { concurrency: maxConcurrent } = await getCanvasImageTaskQueueConfig(); + const { concurrency: maxConcurrent, batchDispatchIntervalMs } = + await getCanvasImageTaskQueueConfig(); const modelState = getGlobalConcurrencyState().models.get(normalizedModelKey) || { running: new Map(), @@ -304,6 +327,7 @@ export async function getCanvasImageModelQueueStatus(params: { sameModelPendingCount: modelState.waiters.length, sameModelRunningCount: modelState.running.size, concurrencyLimit: maxConcurrent, + dispatchIntervalMs: batchDispatchIntervalMs, }); } @@ -328,7 +352,7 @@ export function recordCanvasImageModelExecutionDuration( async function drainModelWaiters( modelKey: string, - trigger: "arrival" | "slot-release" | "timer" + _trigger: "arrival" | "slot-release" | "timer" ) { const normalizedModelKey = normalizeCanvasImageModelKey(modelKey); const modelState = getGlobalConcurrencyState().models.get(normalizedModelKey); @@ -339,7 +363,12 @@ async function drainModelWaiters( const { concurrency: maxConcurrent, batchDispatchIntervalMs } = await getCanvasImageTaskQueueConfig(); - notifyModelWaiters(normalizedModelKey, modelState, maxConcurrent); + notifyModelWaiters( + normalizedModelKey, + modelState, + maxConcurrent, + batchDispatchIntervalMs + ); if (modelState.running.size >= maxConcurrent) { return; } @@ -349,19 +378,6 @@ async function drainModelWaiters( 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 waiterIndex = modelState.waiters.findIndex( @@ -383,9 +399,24 @@ async function drainModelWaiters( } modelState.running.set(waiter.ticketId, waiter.ownerKey); waiter.resolve(); + if (batchDispatchIntervalMs > 0) { + modelState.delayedUntil = Date.now() + batchDispatchIntervalMs; + modelState.dispatchTimer = setTimeout(() => { + const state = getGlobalConcurrencyState().models.get(normalizedModelKey); + if (!state) return; + clearModelDispatchTimer(state); + void drainModelWaiters(normalizedModelKey, "timer"); + }, batchDispatchIntervalMs); + break; + } } - notifyModelWaiters(normalizedModelKey, modelState, maxConcurrent); + notifyModelWaiters( + normalizedModelKey, + modelState, + maxConcurrent, + batchDispatchIntervalMs + ); cleanupModelState(normalizedModelKey); } @@ -408,6 +439,9 @@ export async function acquireCanvasImageModelSlot(params: { const waitingSince = Date.now(); const hasActiveDelay = modelState.delayedUntil !== null && modelState.delayedUntil > waitingSince; + if (!hasActiveDelay && modelState.dispatchTimer) { + clearModelDispatchTimer(modelState); + } const ownerLimit = getAdaptiveImageQueueOwnerLimit({ maxConcurrent, hasCompetingOwner: hasCompetingWaiter(modelState, normalizedOwnerKey), @@ -420,6 +454,15 @@ export async function acquireCanvasImageModelSlot(params: { getRunningCountForOwner(modelState, normalizedOwnerKey) < ownerLimit ) { modelState.running.set(ticketId, normalizedOwnerKey); + if (batchDispatchIntervalMs > 0) { + modelState.delayedUntil = Date.now() + batchDispatchIntervalMs; + modelState.dispatchTimer = setTimeout(() => { + const state = getGlobalConcurrencyState().models.get(normalizedModelKey); + if (!state) return; + clearModelDispatchTimer(state); + void drainModelWaiters(normalizedModelKey, "timer"); + }, batchDispatchIntervalMs); + } return { release: async () => { const state = getGlobalConcurrencyState().models.get(normalizedModelKey); @@ -442,7 +485,12 @@ export async function acquireCanvasImageModelSlot(params: { debugLabel: params.debugLabel, debugMeta: params.debugMeta, }); - notifyModelWaiters(normalizedModelKey, modelState, maxConcurrent); + notifyModelWaiters( + normalizedModelKey, + modelState, + maxConcurrent, + batchDispatchIntervalMs + ); void drainModelWaiters(normalizedModelKey, "arrival"); }); diff --git a/lib/canvas/image-queue-fairness.ts b/lib/canvas/image-queue-fairness.ts index 41e5b3930..65b9442ab 100644 --- a/lib/canvas/image-queue-fairness.ts +++ b/lib/canvas/image-queue-fairness.ts @@ -9,3 +9,14 @@ export function getAdaptiveImageQueueOwnerLimit(params: { return Math.max(1, Math.ceil(maxConcurrent / 2)); } + +export function estimateImageDispatchWaitMs(params: { + queuePosition: number; + dispatchIntervalMs: number; + delayRemainingMs?: number; +}) { + const queuePosition = Math.max(0, Math.floor(params.queuePosition || 0)); + const dispatchIntervalMs = Math.max(0, params.dispatchIntervalMs || 0); + const delayRemainingMs = Math.max(0, params.delayRemainingMs || 0); + return delayRemainingMs + queuePosition * dispatchIntervalMs; +} diff --git a/lib/canvas/image-task-queue.ts b/lib/canvas/image-task-queue.ts index a0966ddb4..77252e2ab 100644 --- a/lib/canvas/image-task-queue.ts +++ b/lib/canvas/image-task-queue.ts @@ -4,9 +4,9 @@ import { and, asc, desc, eq, lt, notInArray, sql } from "drizzle-orm"; import { db } from "@/lib/db"; import { canvasImageTask } from "@/lib/db/schema"; import { - getCanvasImageBatchDispatchIntervalMs, getCanvasImageRunningCountForModel, getCanvasImageSingleModelConcurrency, + getCanvasImageTaskQueueConfig, normalizeCanvasImageModelKey, } from "./image-concurrency"; import { @@ -25,7 +25,10 @@ import { isCanvasLayoutReplaceTaskRequest, type CanvasLayoutReplaceTaskRequest, } from "@/lib/canvas/layout-replace-task"; -import { getAdaptiveImageQueueOwnerLimit } from "./image-queue-fairness"; +import { + estimateImageDispatchWaitMs, + getAdaptiveImageQueueOwnerLimit, +} from "./image-queue-fairness"; type CanvasImageTaskStatus = "pending" | "running" | "succeeded" | "failed"; type CanvasImageTaskRequest = @@ -47,6 +50,7 @@ export type CanvasImageTaskQueueStatus = { sameModelPendingCount: number; sameModelRunningCount: number; concurrencyLimit: number; + dispatchIntervalMs: number; estimatedWaitMs: number; source: "model_queue" | "batch_queue"; }; @@ -98,8 +102,6 @@ const LOCAL_ACTIVE_RUNNING_HARD_TIMEOUT_MS = 30 * 60 * 1000; const STALE_RUNNING_CLEANUP_INTERVAL_MS = 60 * 1000; const PENDING_LOOKAHEAD_LIMIT = 100; const QUEUE_STATUS_PENDING_LOOKAHEAD_LIMIT = 500; -const QUEUE_STATUS_RECENT_SAMPLE_LIMIT = 120; -const DEFAULT_ESTIMATED_WAIT_MS = 45 * 1000; const queueStateKey = "__canvas_image_task_queue_state__"; function getQueueState(): QueueState { @@ -513,6 +515,16 @@ async function markStaleDatabaseRunningTasksFailed() { } async function runSingleTask(entry: PendingTaskEntry) { + let dispatchSlotReleased = false; + const releaseDispatchSlot = async () => { + if (dispatchSlotReleased) return; + dispatchSlotReleased = true; + releaseRunningTask(entry.taskId); + setTimeout(() => { + void drainCanvasImageTaskQueue(); + }, 0); + }; + try { const [task] = await db .update(canvasImageTask) @@ -541,15 +553,18 @@ async function runSingleTask(entry: PendingTaskEntry) { ? executeStyleProTask({ request, userId: String(task.userId || entry.userId), + onRequestDispatched: releaseDispatchSlot, }) : isCanvasLayoutReplaceTaskRequest(request) ? executeCanvasLayoutReplaceTask({ request, userId: String(task.userId || entry.userId), + onRequestDispatched: releaseDispatchSlot, }) : generateCanvasImage({ ...(request as CanvasGenerateImageRequest), ownerKey: task.userId, + onRequestDispatched: releaseDispatchSlot, }); const result = await withCanvasImageTaskExecutionTimeout( execution, @@ -581,14 +596,7 @@ async function runSingleTask(entry: PendingTaskEntry) { and(eq(canvasImageTask.id, entry.taskId), eq(canvasImageTask.status, "running")) ); } finally { - const modelKey = releaseRunningTask(entry.taskId); - if (modelKey) { - const batchDispatchIntervalMs = await getCanvasImageBatchDispatchIntervalMs(); - if (getQueueState().awaitingNextBatch.has(modelKey)) { - scheduleModelNextBatch(modelKey, batchDispatchIntervalMs); - } - } - void drainCanvasImageTaskQueue(); + await releaseDispatchSlot(); } } @@ -599,12 +607,14 @@ export async function drainCanvasImageTaskQueue() { state.draining = true; try { await markStaleDatabaseRunningTasksFailed(); - const maxConcurrentPerModel = await getCanvasImageSingleModelConcurrency(); + const { concurrency: maxConcurrentPerModel, batchDispatchIntervalMs } = + await getCanvasImageTaskQueueConfig(); while (true) { const nextTask = await claimNextPendingTask(maxConcurrentPerModel); if (!nextTask) break; if (state.running.has(nextTask.taskId)) continue; markRunningTask(nextTask.taskId, nextTask.modelKey, nextTask.userId); + scheduleModelNextBatch(nextTask.modelKey, batchDispatchIntervalMs); void runSingleTask(nextTask); } } finally { @@ -670,6 +680,7 @@ export async function generateCanvasImageWithPersistentQueue(params: { }): Promise { const { onQueueStatusChange, + onRequestDispatched: _requestDispatchCallback, ownerKey: _requestOwnerKey, ...serializableRequest } = params.request; @@ -700,6 +711,7 @@ export async function generateCanvasImageWithPersistentQueue(params: { queueStatus.sameModelPendingCount, queueStatus.sameModelRunningCount, queueStatus.concurrencyLimit, + queueStatus.dispatchIntervalMs, ].join(":"); if (statusSignature !== lastStatusSignature) { lastStatusSignature = statusSignature; @@ -739,58 +751,6 @@ export async function generateCanvasImageWithPersistentQueue(params: { throw new Error(`图片生成任务等待超时（${task.id}）`); } -async function estimateModelQueueWaitMs(params: { - modelKey: string; - sameModelRunningCount: number; - queuePosition: number; - concurrencyLimit: number; -}) { - const rows = await db - .select({ - request: canvasImageTask.request, - startedAt: canvasImageTask.startedAt, - completedAt: canvasImageTask.completedAt, - }) - .from(canvasImageTask) - .where(eq(canvasImageTask.status, "succeeded")) - .orderBy(desc(canvasImageTask.completedAt)) - .limit(QUEUE_STATUS_RECENT_SAMPLE_LIMIT); - - const durations = rows - .filter((row) => normalizeTaskModelKey(row.request) === params.modelKey) - .map((row) => { - if (!row.startedAt || !row.completedAt) return 0; - return row.completedAt.getTime() - row.startedAt.getTime(); - }) - .filter((value) => Number.isFinite(value) && value > 1000) - .slice(0, 12); - - const averageDurationMs = - durations.length > 0 - ? Math.round(durations.reduce((sum, value) => sum + value, 0) / durations.length) - : DEFAULT_ESTIMATED_WAIT_MS; - const immediatelyStartable = Math.max( - params.concurrencyLimit - params.sameModelRunningCount, - 0 - ); - - if (params.queuePosition < immediatelyStartable) { - return 0; - } - - const tasksAheadAfterImmediateSlots = Math.max( - 0, - params.queuePosition - immediatelyStartable - ); - const waveCount = - Math.floor( - tasksAheadAfterImmediateSlots / Math.max(1, params.concurrencyLimit) - ) + - (params.sameModelRunningCount > 0 || immediatelyStartable === 0 ? 1 : 0); - - return Math.max(0, waveCount * averageDurationMs); -} - async function buildCanvasImageTaskQueueStatus(task: { id: string; status: CanvasImageTaskStatus; @@ -798,7 +758,10 @@ async function buildCanvasImageTaskQueueStatus(task: { createdAt: Date; }) { const modelKey = normalizeTaskModelKey(task.request); - const concurrencyLimit = await getCanvasImageSingleModelConcurrency(); + const { + concurrency: concurrencyLimit, + batchDispatchIntervalMs: dispatchIntervalMs, + } = await getCanvasImageTaskQueueConfig(); const runningRows = await db .select({ request: canvasImageTask.request, @@ -817,6 +780,7 @@ async function buildCanvasImageTaskQueueStatus(task: { sameModelPendingCount: 0, sameModelRunningCount, concurrencyLimit, + dispatchIntervalMs, estimatedWaitMs: 0, source: "model_queue" as const, }; @@ -841,11 +805,13 @@ async function buildCanvasImageTaskQueueStatus(task: { 0 ); const sameModelPendingCount = sameModelPendingRows.length; - const estimatedWaitMs = await estimateModelQueueWaitMs({ - modelKey, - sameModelRunningCount, + const estimatedWaitMs = estimateImageDispatchWaitMs({ queuePosition, - concurrencyLimit, + dispatchIntervalMs, + delayRemainingMs: Math.max( + 0, + Number(getQueueState().nextBatchEligibleAt.get(modelKey) || 0) - Date.now() + ), }); return { @@ -854,6 +820,7 @@ async function buildCanvasImageTaskQueueStatus(task: { sameModelPendingCount, sameModelRunningCount, concurrencyLimit, + dispatchIntervalMs, estimatedWaitMs, source: "model_queue" as const, }; diff --git a/lib/canvas/layout-replace-task.ts b/lib/canvas/layout-replace-task.ts index 1df3a9757..e4397217c 100644 --- a/lib/canvas/layout-replace-task.ts +++ b/lib/canvas/layout-replace-task.ts @@ -466,6 +466,7 @@ async function buildLayoutLockedCompositeImage(params: { export async function executeCanvasLayoutReplaceTask(params: { request: CanvasLayoutReplaceTaskRequest; userId: string; + onRequestDispatched?: () => void | Promise; }): Promise { const sourceImageUrl = normalizeHttpUrl(params.request.sourceImageUrl); const referenceImageUrl = normalizeHttpUrl(params.request.referenceImageUrl); @@ -542,6 +543,7 @@ export async function executeCanvasLayoutReplaceTask(params: { displayPrompt: "排版替换结果", model: resolvedModelRef, ownerKey: params.userId, + onRequestDispatched: params.onRequestDispatched, aspectRatio: referenceGeometry?.aspectRatio || "auto", imageSize: params.request.imageSize || "2K", imageCount: 1, diff --git a/lib/style-pro/task.ts b/lib/style-pro/task.ts index 0282484c6..7fdbb1a23 100644 --- a/lib/style-pro/task.ts +++ b/lib/style-pro/task.ts @@ -264,6 +264,7 @@ async function saveStyleProFinalImageToGallery(params: { export async function executeStyleProTask(params: { request: StyleProTaskRequest; userId: string; + onRequestDispatched?: () => void | Promise; }) { const { request, userId } = params; const aspectRatio = request.prompts.aspectRatio || "auto"; @@ -289,6 +290,7 @@ export async function executeStyleProTask(params: { imageCount: 1, referenceImage: request.sourceImageUrl, ownerKey: userId, + onRequestDispatched: params.onRequestDispatched, allowFailover: false, debugLabel: "style-pro-realistic-stage1", debugMeta: { @@ -317,6 +319,7 @@ export async function executeStyleProTask(params: { imageCount: 1, referenceImage: firstPassImage.imageUrl, ownerKey: userId, + onRequestDispatched: params.onRequestDispatched, allowFailover: false, debugLabel: "style-pro-realistic-stage2", debugMeta: { @@ -342,6 +345,7 @@ export async function executeStyleProTask(params: { imageCount: 1, referenceImage: request.sourceImageUrl, ownerKey: userId, + onRequestDispatched: params.onRequestDispatched, allowFailover: false, debugLabel: "style-pro", debugMeta: { diff --git a/tests/routes/canvas-image-queue-fairness.test.ts b/tests/routes/canvas-image-queue-fairness.test.ts index c13bac471..4021dd1c7 100644 --- a/tests/routes/canvas-image-queue-fairness.test.ts +++ b/tests/routes/canvas-image-queue-fairness.test.ts @@ -1,5 +1,8 @@ import { expect, test } from "@playwright/test"; -import { getAdaptiveImageQueueOwnerLimit } from "@/lib/canvas/image-queue-fairness"; +import { + estimateImageDispatchWaitMs, + getAdaptiveImageQueueOwnerLimit, +} from "@/lib/canvas/image-queue-fairness"; test.describe("canvas image queue fairness", () => { test("lets one owner use the full model capacity without competition", () => { @@ -34,4 +37,23 @@ test.describe("canvas image queue fairness", () => { }) ).toBe(1); }); + + test("estimates queue wait from request dispatch spacing, not generation duration", () => { + expect( + estimateImageDispatchWaitMs({ + queuePosition: 3, + dispatchIntervalMs: 1000, + delayRemainingMs: 400, + }) + ).toBe(3400); + }); + + test("does not add wait after the request has no tasks ahead", () => { + expect( + estimateImageDispatchWaitMs({ + queuePosition: 0, + dispatchIntervalMs: 1000, + }) + ).toBe(0); + }); });