perf(判题): 判题机上同时跑的程序从最多 8 个压到 4 个,挤的时候不再把不超时的判成超时
原来四道闸各管各的(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 <noreply@anthropic.com>
This commit is contained in:
1 parent
64994b9499
commit
307b0725d0
4 files changed
+159
-41
No files matched your search
@@ -115,6 +115,8 @@ async function evaluateByRunning(
|
|||||||
memoryLimit: 256,
|
memoryLimit: 256,
|
||||||
cases: cases.map((item) => ({ input: item.input, output: item.expected })),
|
cases: cases.map((item) => ({ input: item.input, output: item.expected })),
|
||||||
fullOutput: true,
|
fullOutput: true,
|
||||||
|
// 后台评分,别和学生「运行例子」抢交互池的名额
|
||||||
|
pool: "judge",
|
||||||
})
|
})
|
||||||
|
|
||||||
let results: FlowchartCaseResult[]
|
let results: FlowchartCaseResult[]
|
||||||
|
|||||||
+19
-39
@@ -15,6 +15,7 @@ import type { JudgeJobData } from "./job"
|
|||||||
import { judgeConfigFor } from "./languages"
|
import { judgeConfigFor } from "./languages"
|
||||||
import { isAccepted, JudgeStatus, type JudgeStatusValue } from "./status"
|
import { isAccepted, JudgeStatus, type JudgeStatusValue } from "./status"
|
||||||
import { parseProblemTemplate, shiftCompileLines } from "./template"
|
import { parseProblemTemplate, shiftCompileLines } from "./template"
|
||||||
|
import { withJudgeSlot } from "./slots"
|
||||||
import { runSqlCase } from "./sql"
|
import { runSqlCase } from "./sql"
|
||||||
import {
|
import {
|
||||||
nativeRuntimeError,
|
nativeRuntimeError,
|
||||||
@@ -121,46 +122,22 @@ export async function postJudge(
|
|||||||
* 判题机只有 /judge 一个入口,得带一组测试点:给一组空输入,跑的那一下限 1 秒。
|
* 判题机只有 /judge 一个入口,得带一组测试点:给一组空输入,跑的那一下限 1 秒。
|
||||||
* 判题机连不上、超时,一律当编得过 —— 那就和原来一样照常格式化。
|
* 判题机连不上、超时,一律当编得过 —— 那就和原来一样照常格式化。
|
||||||
*
|
*
|
||||||
* 这一下不进判题队列,所以自己排一道:同时最多 COMPILE_CHECK_CONCURRENCY 份在判题机上编,
|
* 这一下不进判题队列,但和运行例子共用交互池的名额(judge/slots.ts):服务器只有两核,
|
||||||
* 其余等着。服务器只有两核,不排的话一个班 40 人一起交 C++ 能把判题机挤到每人 8 秒超时,
|
* 不排的话一个班 40 人一起交 C++ 能把判题机挤到每人 8 秒超时,还拖慢正式判题。
|
||||||
* 还拖慢正式判题。8 秒是连排队带编译的总预算,排不上也当编得过。
|
* 8 秒是连排队带编译的总预算,排不上也当编得过。
|
||||||
*/
|
*/
|
||||||
const COMPILE_CHECK_CONCURRENCY = 2
|
|
||||||
let compileChecks = 0
|
|
||||||
const compileCheckWaiting: (() => void)[] = []
|
|
||||||
|
|
||||||
export async function compileFails(language: string, source: string) {
|
export async function compileFails(language: string, source: string) {
|
||||||
const deadline = AbortSignal.timeout(8000)
|
const deadline = AbortSignal.timeout(8000)
|
||||||
if (compileChecks < COMPILE_CHECK_CONCURRENCY) {
|
|
||||||
compileChecks++
|
|
||||||
} else {
|
|
||||||
const admitted = await new Promise<boolean>((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 {
|
try {
|
||||||
const response = await postJudge(language, source, 1000, 256, [{ input: "", output: "" }], {
|
return await withJudgeSlot("interactive", { holdMs: 30_000, signal: deadline }, async () => {
|
||||||
signal: deadline,
|
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 {
|
} catch {
|
||||||
return false
|
return false
|
||||||
} finally {
|
|
||||||
const next = compileCheckWaiting.shift()
|
|
||||||
if (next) next()
|
|
||||||
else compileChecks--
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -210,13 +187,16 @@ async function requestJudge(
|
|||||||
) {
|
) {
|
||||||
const timeoutMs = judgeTimeoutMs(timeLimit)
|
const timeoutMs = judgeTimeoutMs(timeLimit)
|
||||||
try {
|
try {
|
||||||
const response = await postJudge(language, code, timeLimit, memoryLimit, testCase, {
|
// 超时从拿到名额才开始算:排队等名额不算判题机没回应
|
||||||
output,
|
return await withJudgeSlot("judge", { holdMs: timeoutMs + 30_000 }, async () => {
|
||||||
signal: AbortSignal.timeout(timeoutMs),
|
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) {
|
} catch (error) {
|
||||||
// 这句会写进 err_info,学生能看到,别把 "The operation timed out." 原样给他
|
// 这句会写进 err_info,学生能看到,别把 "The operation timed out." 原样给他
|
||||||
if (error instanceof DOMException && error.name === "TimeoutError") {
|
if (error instanceof DOMException && error.name === "TimeoutError") {
|
||||||
|
|||||||
@@ -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<JudgePool, number> = {
|
||||||
|
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<number>
|
||||||
|
}
|
||||||
|
|
||||||
|
function sleep(ms: number, signal?: AbortSignal) {
|
||||||
|
return new Promise<void>((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<T>(
|
||||||
|
pool: JudgePool,
|
||||||
|
options: { holdMs: number; signal?: AbortSignal },
|
||||||
|
run: () => Promise<T>,
|
||||||
|
): Promise<T> {
|
||||||
|
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)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -10,6 +10,7 @@ import {
|
|||||||
type JudgeResponse,
|
type JudgeResponse,
|
||||||
} from "./run"
|
} from "./run"
|
||||||
import { nativeRuntimeError, parsePythonTraceback } from "./runtime-diagnosis"
|
import { nativeRuntimeError, parsePythonTraceback } from "./runtime-diagnosis"
|
||||||
|
import { withJudgeSlot, type JudgePool } from "./slots"
|
||||||
import { JudgeStatus, type JudgeStatusValue } from "./status"
|
import { JudgeStatus, type JudgeStatusValue } from "./status"
|
||||||
import { shiftCompileLines } from "./template"
|
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
|
const MAX_WAITING = 60
|
||||||
@@ -91,6 +93,8 @@ export async function runTrial(options: {
|
|||||||
cases: { input: string; output?: string }[]
|
cases: { input: string; output?: string }[]
|
||||||
/** 管理员生成测试点要完整输出 */
|
/** 管理员生成测试点要完整输出 */
|
||||||
fullOutput: boolean
|
fullOutput: boolean
|
||||||
|
/** 判题机名额从哪个池拿,默认交互池;流程图评分是后台的,走 judge */
|
||||||
|
pool?: JudgePool
|
||||||
}): Promise<TrialRunResponse> {
|
}): Promise<TrialRunResponse> {
|
||||||
const inFlight = perUser.get(options.userId) ?? 0
|
const inFlight = perUser.get(options.userId) ?? 0
|
||||||
if (inFlight >= PER_USER_LIMIT) throw new TrialUserLimitError()
|
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<typeof runTrial>[0]): Promise<TrialRunResponse> {
|
async function judgeTrial(options: Parameters<typeof runTrial>[0]): Promise<TrialRunResponse> {
|
||||||
|
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<typeof runTrial>[0]): Promise<TrialRunResponse> {
|
||||||
const response = await postJudge(
|
const response = await postJudge(
|
||||||
options.language,
|
options.language,
|
||||||
options.source,
|
options.source,
|
||||||
|
|||||||
Reference in new issue
Block a user