From 307b0725d0afe17742813a429c7339c251c77312 Mon Sep 17 00:00:00 2001 From: yuetsh <517252939@qq.com> Date: Fri, 16 Oct 2026 14:47:24 +0800 Subject: [PATCH] =?UTF-8?q?perf(=E5=88=A4=E9=A2=98):=20=E5=88=A4=E9=A2=98?= =?UTF-8?q?=E6=9C=BA=E4=B8=8A=E5=90=8C=E6=97=B6=E8=B7=91=E7=9A=84=E7=A8=8B?= =?UTF-8?q?=E5=BA=8F=E4=BB=8E=E6=9C=80=E5=A4=9A=208=20=E4=B8=AA=E5=8E=8B?= =?UTF-8?q?=E5=88=B0=204=20=E4=B8=AA=EF=BC=8C=E6=8C=A4=E7=9A=84=E6=97=B6?= =?UTF-8?q?=E5=80=99=E4=B8=8D=E5=86=8D=E6=8A=8A=E4=B8=8D=E8=B6=85=E6=97=B6?= =?UTF-8?q?=E7=9A=84=E5=88=A4=E6=88=90=E8=B6=85=E6=97=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 原来四道闸各管各的（worker 判题 2、api 试运行 2、提交前编译检查 2、worker 里流程图 评分的试运行又 2），api 和 worker 两个进程互相看不见，峰值 8 个程序压在服务器两个核上； 判题机按墙钟判超时，挤到一定程度会冤枉学生。 judge/slots.ts 在 Redis 里按池数名额（每个站点各数各的）： - 后台池（正式判题、诊断重跑、流程图评分）上限 JUDGE_CONCURRENCY - 交互池（运行例子、自己输入、生成测试点、编译检查）上限 TRIAL_CONCURRENCY 两池不互相借，正式判题排再长也挤不掉学生的「运行例子」。名额是带到期时间的租约， 进程被杀了到期自动作废；Redis 出错直接放行。正式判题的超时从拿到名额才开始算。 Co-Authored-By: Claude Opus 5.5 --- apps/api/src/flowchart/run.ts | 2 + apps/api/src/judge/run.ts | 58 ++++++----------- apps/api/src/judge/slots.ts | 113 ++++++++++++++++++++++++++++++++++ apps/api/src/judge/trial.ts | 27 +++++++- 4 files changed, 159 insertions(+), 41 deletions(-) create mode 100644 apps/api/src/judge/slots.ts diff --git a/apps/api/src/flowchart/run.ts b/apps/api/src/flowchart/run.ts index f5c57e1..d279600 100644 --- a/apps/api/src/flowchart/run.ts +++ b/apps/api/src/flowchart/run.ts @@ -115,6 +115,8 @@ async function evaluateByRunning( memoryLimit: 256, cases: cases.map((item) => ({ input: item.input, output: item.expected })), fullOutput: true, + // 后台评分，别和学生「运行例子」抢交互池的名额 + pool: "judge", }) let results: FlowchartCaseResult[] diff --git a/apps/api/src/judge/run.ts b/apps/api/src/judge/run.ts index 416e8ff..0745bb7 100644 --- a/apps/api/src/judge/run.ts +++ b/apps/api/src/judge/run.ts @@ -15,6 +15,7 @@ import type { JudgeJobData } from "./job" import { judgeConfigFor } from "./languages" import { isAccepted, JudgeStatus, type JudgeStatusValue } from "./status" import { parseProblemTemplate, shiftCompileLines } from "./template" +import { withJudgeSlot } from "./slots" import { runSqlCase } from "./sql" import { nativeRuntimeError, @@ -121,46 +122,22 @@ export async function postJudge( * 判题机只有 /judge 一个入口，得带一组测试点：给一组空输入，跑的那一下限 1 秒。 * 判题机连不上、超时，一律当编得过 —— 那就和原来一样照常格式化。 * - * 这一下不进判题队列，所以自己排一道：同时最多 COMPILE_CHECK_CONCURRENCY 份在判题机上编， - * 其余等着。服务器只有两核，不排的话一个班 40 人一起交 C++ 能把判题机挤到每人 8 秒超时， - * 还拖慢正式判题。8 秒是连排队带编译的总预算，排不上也当编得过。 + * 这一下不进判题队列，但和运行例子共用交互池的名额（judge/slots.ts）：服务器只有两核， + * 不排的话一个班 40 人一起交 C++ 能把判题机挤到每人 8 秒超时，还拖慢正式判题。 + * 8 秒是连排队带编译的总预算，排不上也当编得过。 */ -const COMPILE_CHECK_CONCURRENCY = 2 -let compileChecks = 0 -const compileCheckWaiting: (() => void)[] = [] - export async function compileFails(language: string, source: string) { const deadline = AbortSignal.timeout(8000) - if (compileChecks < COMPILE_CHECK_CONCURRENCY) { - compileChecks++ - } else { - const admitted = await new Promise((resolve) => { - const giveUp = () => { - compileCheckWaiting.splice(compileCheckWaiting.indexOf(wake), 1) - resolve(false) - } - const wake = () => { - deadline.removeEventListener("abort", giveUp) - resolve(true) - } - deadline.addEventListener("abort", giveUp, { once: true }) - compileCheckWaiting.push(wake) - }) - // 被叫醒时名额是直接转交过来的，compileChecks 不用再加 - if (!admitted) return false - } try { - const response = await postJudge(language, source, 1000, 256, [{ input: "", output: "" }], { - signal: deadline, + return await withJudgeSlot("interactive", { holdMs: 30_000, signal: deadline }, async () => { + const response = await postJudge(language, source, 1000, 256, [{ input: "", output: "" }], { + signal: deadline, + }) + const body = (await response.json()) as JudgeResponse + return body.err === "CompileError" }) - const body = (await response.json()) as JudgeResponse - return body.err === "CompileError" } catch { return false - } finally { - const next = compileCheckWaiting.shift() - if (next) next() - else compileChecks-- } } @@ -210,13 +187,16 @@ async function requestJudge( ) { const timeoutMs = judgeTimeoutMs(timeLimit) try { - const response = await postJudge(language, code, timeLimit, memoryLimit, testCase, { - output, - signal: AbortSignal.timeout(timeoutMs), + // 超时从拿到名额才开始算：排队等名额不算判题机没回应 + return await withJudgeSlot("judge", { holdMs: timeoutMs + 30_000 }, async () => { + const response = await postJudge(language, code, timeLimit, memoryLimit, testCase, { + output, + signal: AbortSignal.timeout(timeoutMs), + }) + const body = await readCapped(response, MAX_RESPONSE_BYTES) + if (body === null) throw new Error("判题机返回的内容太大") + return JSON.parse(body) as JudgeResponse }) - const body = await readCapped(response, MAX_RESPONSE_BYTES) - if (body === null) throw new Error("判题机返回的内容太大") - return JSON.parse(body) as JudgeResponse } catch (error) { // 这句会写进 err_info，学生能看到，别把 "The operation timed out." 原样给他 if (error instanceof DOMException && error.name === "TimeoutError") { diff --git a/apps/api/src/judge/slots.ts b/apps/api/src/judge/slots.ts new file mode 100644 index 0000000..067bd23 --- /dev/null +++ b/apps/api/src/judge/slots.ts @@ -0,0 +1,113 @@ +import { randomUUID } from "node:crypto" + +import { config } from "../config" +import { redis } from "../redis" + +/** + * 判题沙箱的名额，按池在 Redis 里数。 + * + * 原来是四道各管各的闸：worker 判题 2、api 试运行 2、api 提交前编译检查 2、worker 里 + * 流程图评分又起一道试运行 2 —— 峰值 8 个程序同时压在服务器那两个核上，判题机按墙钟 + * 判超时（CPU 时限的 3 倍），挤到一定程度就会把本来不超时的程序判成超时。 + * api 和 worker 是两个进程，进程内的计数管不到对方，所以放 Redis。每个站点有自己的 + * Redis 和判题机，按站各数各的正好。 + * + * 两个池，名额不互相借： + * - `judge`：后台的 —— 正式判题、诊断重跑、流程图评分的试运行。上限 JUDGE_CONCURRENCY； + * - `interactive`：学生在页面上等着的 —— 运行例子、自己输入、生成测试点、提交前的编译检查。 + * 上限 TRIAL_CONCURRENCY。单独留着，正式判题排再长也不会把它挤掉，反过来也一样。 + * + * 每个名额是一份租约（有序集合里一个成员，分数是到期时刻）：拿着名额的进程被杀了， + * 到期后自动作废，不会永远占着。Redis 出错时直接放行 —— 判题比限流要紧， + * 而 Redis 真挂了，会话也读不出来，整站本来就不可用。 + */ +export type JudgePool = "judge" | "interactive" + +const LIMITS: Record = { + judge: config.judgeConcurrency, + interactive: config.trialConcurrency, +} + +// KEYS[1] = 池的有序集合 +// ARGV = now(ms), expireAt(ms), token, limit +const ACQUIRE_SCRIPT = ` +redis.call('ZREMRANGEBYSCORE', KEYS[1], '-inf', ARGV[1]) +if redis.call('ZCARD', KEYS[1]) >= tonumber(ARGV[4]) then return 0 end +redis.call('ZADD', KEYS[1], ARGV[2], ARGV[3]) +return 1 +` + +redis.defineCommand("judgeSlotAcquire", { numberOfKeys: 1, lua: ACQUIRE_SCRIPT }) + +type SlotRedis = typeof redis & { + judgeSlotAcquire( + key: string, + now: string, + expireAt: string, + token: string, + limit: string, + ): Promise +} + +function sleep(ms: number, signal?: AbortSignal) { + return new Promise((resolve, reject) => { + if (signal?.aborted) return reject(signal.reason) + const timer = setTimeout(() => { + signal?.removeEventListener("abort", onAbort) + resolve() + }, ms) + const onAbort = () => { + clearTimeout(timer) + reject(signal!.reason) + } + signal?.addEventListener("abort", onAbort, { once: true }) + }) +} + +/** + * 拿到一个名额再跑 `run`，跑完（不论成败）还回去。 + * + * - `holdMs`：租约多长。要比 `run` 最长可能跑的时间长，否则跑到一半名额被别人拿走； + * 调用方都给的是自己的请求超时再加 30 秒。 + * - `signal`：等名额等到它 abort 就放弃，抛它的 reason。不给就一直等（正式判题： + * job 在 BullMQ 里本来就是排着的，多等一会儿不要紧）。 + * + * 等的方式是隔一会儿问一次（50ms 起翻倍到 500ms）。同时在等的最多也就十来个 + * （两个池的上限加上两个进程里各自排着的），这点 Redis 往返不值得为它上阻塞队列。 + */ +export async function withJudgeSlot( + pool: JudgePool, + options: { holdMs: number; signal?: AbortSignal }, + run: () => Promise, +): Promise { + const key = `judge-slots:${pool}` + const token = randomUUID() + let acquired = false + for (let delay = 50; ; delay = Math.min(500, delay * 2)) { + options.signal?.throwIfAborted() + try { + const now = Date.now() + acquired = + (await (redis as SlotRedis).judgeSlotAcquire( + key, + String(now), + String(now + options.holdMs), + token, + String(LIMITS[pool]), + )) === 1 + } catch (error) { + console.warn(`[judge-slots] Redis 不可用，不排队直接判`, error) + return run() + } + if (acquired) break + await sleep(delay, options.signal) + } + try { + return await run() + } finally { + await redis.zrem(key, token).catch((error: unknown) => { + // 还不回去也不要紧，租约到期自己作废 + console.warn(`[judge-slots] 归还名额失败`, error) + }) + } +} diff --git a/apps/api/src/judge/trial.ts b/apps/api/src/judge/trial.ts index cb53665..883e124 100644 --- a/apps/api/src/judge/trial.ts +++ b/apps/api/src/judge/trial.ts @@ -10,6 +10,7 @@ import { type JudgeResponse, } from "./run" import { nativeRuntimeError, parsePythonTraceback } from "./runtime-diagnosis" +import { withJudgeSlot, type JudgePool } from "./slots" import { JudgeStatus, type JudgeStatusValue } from "./status" import { shiftCompileLines } from "./template" @@ -25,8 +26,9 @@ import { shiftCompileLines } from "./template" * - 一个人同时最多两批（「运行例子」和「自己输入」可以同时点），再多直接拒； * - 等的人太多、或者等得太久，也直接拒，让学生过一会儿再点，别无限挂着。 * - * 队列在进程内：api 只有一个进程，机房和服务器各有各的 api 和判题机， - * 各排各的正好。 + * 这道队在进程内，管的是排队上限、每人两批和先来后到；判题机上真正同时跑几批由 + * judge/slots.ts 的交互池管（和提交前的编译检查共用，跨 api / worker 两个进程）。 + * 机房和服务器各有各的 api、Redis 和判题机，各排各的正好。 */ const MAX_WAITING = 60 @@ -91,6 +93,8 @@ export async function runTrial(options: { cases: { input: string; output?: string }[] /** 管理员生成测试点要完整输出 */ fullOutput: boolean + /** 判题机名额从哪个池拿，默认交互池；流程图评分是后台的，走 judge */ + pool?: JudgePool }): Promise { const inFlight = perUser.get(options.userId) ?? 0 if (inFlight >= PER_USER_LIMIT) throw new TrialUserLimitError() @@ -109,7 +113,26 @@ export async function runTrial(options: { } } +/** + * 进程内那道队排到了，还要在判题机的名额池里拿到一个（judge/slots.ts）： + * 编译检查和这里共用交互池，流程图评分走后台池。等名额也算在 WAIT_TIMEOUT_MS 里， + * 等不到一样让学生过一会儿再点。 + */ async function judgeTrial(options: Parameters[0]): Promise { + const wait = AbortSignal.timeout(WAIT_TIMEOUT_MS) + try { + return await withJudgeSlot( + options.pool ?? "interactive", + { holdMs: JUDGE_TIMEOUT_MS + 30_000, signal: wait }, + () => judgeTrialNow(options), + ) + } catch (error) { + if (wait.aborted && error === wait.reason) throw new TrialBusyError() + throw error + } +} + +async function judgeTrialNow(options: Parameters[0]): Promise { const response = await postJudge( options.language, options.source,