刷题系统的评测引擎架构：从单体判题到分布式评测集群

刷题系统的评测引擎架构从单体判题到分布式评测集群一、一个判题请求为什么能让整个系统卡住想象一个场景你写了一个在线刷题的平台用户提交代码后你的服务在单机上启动一个子进程编译运行用户代码比对输出结果后返回 AC 或 WA。这个方案在小规模下完全可行——直到有 100 个用户同时提交。单机判题面临三个核心问题。第一代码执行是 CPU 密集和 I/O 密集的混合任务一个用户的死循环就能让一个核心占满。第二用户代码是不可信的它可能读取系统文件、发起网络请求、甚至 fork 大量子进程耗尽系统资源。第三不同语言的编译、执行环境差异巨大Java 的 JVM 启动时间远大于 Python 的直解执行。这就是为什么在线判题系统的评测引擎必须从单体演进到分布式评测集群——不是面子工程是生存需求。flowchart TB A[用户提交代码] -- B[API 网关: 提交请求] B -- C[任务分发层: 消息队列] C -- D[评测调度器: 路由策略] D -- E1[评测节点1: Python沙箱] D -- E2[评测节点2: Java沙箱] D -- E3[评测节点3: C沙箱] E1 -- F[结果聚合器: 超时/内存/输出判断] E2 -- F E3 -- F F -- G[结果存储: AC/WA/TLE/MLE] G -- H[用户获取结果] subgraph 评测节点内部 I[接收任务] -- J[环境准备: 编译/加载用例] J -- K[沙箱执行: seccomp cgroup 隔离] K -- L[结果比对: 字符精确/浮点容差] L -- M[上报结果] end二、判题系统的核心挑战安全隔离与资源控制判题系统的第一优先级不是性能而是安全。用户提交的代码是任意代码不能直接在宿主机上执行。现代判题系统通常使用三层隔离第一层进程级隔离seccomp。Linux 的安全计算模式seccomp可以限制进程的系统调用。判题节点的子进程只被允许使用 read、write、exit 等基础系统调用禁止 fork、execve、network 等危险调用。第二层资源控制cgroup。cgroup 限制 CPU 时间防止死循环、内存上限防止 OOM、进程数防止 fork 炸弹。判题常设 CPU 2s、内存 256MB、进程数 1。第三层文件系统隔离chroot/overlay。判题节点使用独立的文件系统视图用户代码只能访问标准库和题目相关的输入输出文件。禁止访问/etc/passwd、/proc等系统文件。三、评测调度器的核心实现 分布式评测引擎的设计与实现 核心架构 1. 消息队列解耦提交与判题 2. 评测节点按语言池化 3. 基于优先级的调度策略 import json import time import hashlib from dataclasses import dataclass from typing import Optional, Dict from enum import Enum from collections import defaultdict import threading import queue class Verdict(Enum): 评测结果 AC Accepted WA Wrong Answer TLE Time Limit Exceeded MLE Memory Limit Exceeded RE Runtime Error CE Compilation Error SE System Error class Language(Enum): 支持的编程语言 PYTHON python JAVA java CPP cpp GO go dataclass class JudgeTask: 一次评测任务 task_id: str problem_id: int language: Language source_code: str time_limit_ms: int # CPU 时间限制 memory_limit_mb: int # 内存限制 test_cases: list # 测试用例列表 priority: int 0 # 优先级越大越优先 created_at: float 0.0 def __post_init__(self): if self.created_at 0.0: self.created_at time.time() dataclass class JudgeResult: 单次评判结果 task_id: str verdict: Verdict time_used_ms: int memory_used_mb: int test_case_index: int # 在第几个测试用例出错 error_message: str class JudgeScheduler: 评测调度器 使用多级优先级队列 为什么需要优先级调度比赛提交应优先于练习提交 被 CE 后的重提交应优先于首次提交。 def __init__(self, num_workers: int 4): self.num_workers num_workers # 按语言分组的评测节点池 self.worker_pools: Dict[Language, list] { lang: [] for lang in Language } # 任务队列——按优先级分组 self.high_priority_queue queue.PriorityQueue() self.normal_queue queue.Queue() # 统计信息 self.stats { total_submitted: 0, total_judged: 0, avg_wait_time_ms: 0 } # 结果缓存——相同代码不重复评测 # key hash(source_code) hash(test_cases) self.result_cache: Dict[str, JudgeResult] {} self.cache_hits 0 self._running False def start(self): 启动评测调度器 self._running True # 启动工作线程 for i in range(self.num_workers): t threading.Thread(targetself._worker_loop, namefJudgeWorker-{i}, daemonTrue) t.start() def submit(self, task: JudgeTask) - str: 提交评测任务 为什么先去重重复提交用户连点两次提交产生相同代码 直接用缓存结果减少评测负载。 self.stats[total_submitted] 1 # 去重检查相同源码 相同测试用例 相同结果 cache_key self._cache_key(task) if cache_key in self.result_cache: self.cache_hits 1 return self.result_cache[cache_key].verdict.value # 按优先级入队 # 负数使高优先级任务排在优先队列前面 if task.priority 0: self.high_priority_queue.put((-task.priority, time.time(), task)) else: self.normal_queue.put(task) return SUBMITTED def _worker_loop(self): 评测工作线程的主循环 while self._running: task self._get_next_task() if task is None: time.sleep(0.01) # 没任务时空转减少 CPU 占用 continue result self._judge(task) # 更新缓存 cache_key self._cache_key(task) self.result_cache[cache_key] result # 存储结果 self._store_result(task.task_id, result) self.stats[total_judged] 1 def _get_next_task(self) - Optional[JudgeTask]: 从队列中获取下一个任务 try: # 为什么先检查高优先级队列 # 比赛提交的超时更短必须优先处理 _, _, task self.high_priority_queue.get_nowait() return task except queue.Empty: try: return self.normal_queue.get_nowait() except queue.Empty: return None def _judge(self, task: JudgeTask) - JudgeResult: 执行实际评测 为什么逐个用例测试一旦发现 WA/TLE/MLE 就停止 节省评测资源和用户等待时间。 # 编译阶段 if task.language in (Language.JAVA, Language.CPP): compile_result self._compile(task) if not compile_result[success]: return JudgeResult( task_idtask.task_id, verdictVerdict.CE, time_used_ms0, memory_used_mb0, test_case_index-1, error_messagecompile_result[error] ) # 运行阶段——逐个测试用例 for idx, test_case in enumerate(task.test_cases): run_result self._run_in_sandbox( sourcetask.source_code, languagetask.language, input_datatest_case[input], expected_outputtest_case[output], time_limit_mstask.time_limit_ms, memory_limit_mbtask.memory_limit_mb ) if run_result[verdict] ! Verdict.AC: return JudgeResult( task_idtask.task_id, verdictrun_result[verdict], time_used_msrun_result[time_ms], memory_used_mbrun_result[memory_mb], test_case_indexidx, error_messagerun_result.get(error, ) ) # 全部用例通过 return JudgeResult( task_idtask.task_id, verdictVerdict.AC, time_used_msrun_result[time_ms], memory_used_mbrun_result[memory_mb], test_case_indexlen(task.test_cases) - 1 ) def _run_in_sandbox(self, source: str, language: Language, input_data: str, expected_output: str, time_limit_ms: int, memory_limit_mb: int) - dict: 在沙箱中执行用户代码 为什么使用独立进程 防止用户代码中的恶意操作影响判题系统本身。 # 生产环境应使用 seccomp cgroup 隔离 # 这里展示核心逻辑 import subprocess try: process subprocess.run( self._get_run_command(language), inputinput_data, capture_outputTrue, timeouttime_limit_ms / 1000, # 转为秒 textTrue, # 生产环境还应限制内存、进程数、文件访问 ) actual_output process.stdout.strip() expected expected_output.strip() # 输出比对——支持浮点数容差 if self._compare_output(actual_output, expected): return { verdict: Verdict.AC, time_ms: int((time.time() - time.time()) * 1000), memory_mb: 0 } else: return { verdict: Verdict.WA, time_ms: 0, memory_mb: 0, error: f期望: {expected}\n实际: {actual_output} } except subprocess.TimeoutExpired: return {verdict: Verdict.TLE, time_ms: time_limit_ms, memory_mb: 0} except Exception as e: return {verdict: Verdict.RE, time_ms: 0, memory_mb: 0, error: str(e)} def _compare_output(self, actual: str, expected: str, float_tolerance: float 1e-6) - bool: 输出比对——核心判题逻辑 为什么支持浮点容差 不同语言的浮点数精度不同 精确字符串比对会导致假 WA。 actual_lines actual.splitlines() expected_lines expected.splitlines() if len(actual_lines) ! len(expected_lines): return False for a, e in zip(actual_lines, expected_lines): a, e a.strip(), e.strip() if a e: continue # 尝试浮点数容差比对 try: fa, fe float(a), float(e) if abs(fa - fe) float_tolerance: return False except ValueError: return False return True def _cache_key(self, task: JudgeTask) - str: 生成缓存键 content task.source_code json.dumps(task.test_cases, sort_keysTrue) return hashlib.md5(content.encode()).hexdigest() def _compile(self, task: JudgeTask) - dict: # 编译逻辑——省略具体实现 pass def _get_run_command(self, language: Language) - list: # 获取运行命令——省略具体实现 pass def _store_result(self, task_id: str, result: JudgeResult): # 结果持久化——省略具体实现 pass四、分布式评测的边界与权衡评测节点异构性问题。不同机器的 CPU 型号、主频不同同样的代码在不同节点上的运行时间可能相差 10%。公平竞赛应该使用同构集群或对时间做归一化处理。沙箱逃逸风险。即使有 seccomp 和 cgroup仍有复杂的逃逸技术。生产级判题系统通常使用更强的隔离——虚拟机如 Firecracker或 Docker 容器代价是资源开销增大。评测结果的可重现性。分布式判题的一个隐含需求是同一份代码在不同节点上的结果应该一致。这要求所有节点使用完全相同的编译环境编译器版本、标准库版本、系统库版本。缓存策略的权衡。相同代码缓存可以大幅减少评测负载但缓存键的设计需要平衡冲突率和缓存命中率。仅用源码做哈希可能因为测试用例不同而失效。五、总结从单体到分布式的评测引擎架构演进本质上是在解决三个问题安全性隔离不可信代码、可扩展性支撑高并发提交、公平性保证评测结果的准确一致。核心架构组件包括消息队列做请求缓冲、语言池化做资源隔离、优先级调度做任务排序、沙箱执行做安全保证。每层的设计决策都不是孤立的——沙箱的选择影响评测速度评测速度影响调度策略调度策略影响用户体验。理解判题系统的全链路不仅是工程能力的体现更是在设计刷题平台时做出正确 trade-off 的前提。