From ebf541c1fedde65c7f6d580fcbabf903a5343164 Mon Sep 17 00:00:00 2001 From: yuetsh <517252939@qq.com> Date: Mon, 7 Sep 2026 04:45:49 -0600 Subject: [PATCH] =?UTF-8?q?fix:=20=E8=B0=83=E8=AF=95=E6=89=A7=E8=A1=8C?= =?UTF-8?q?=E7=A7=BB=E5=85=A5=E5=AD=90=E8=BF=9B=E7=A8=8B=E5=B9=B6=E5=8A=A0?= =?UTF-8?q?=E4=B8=8A=E8=B5=84=E6=BA=90=E9=99=90=E5=88=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pg_logger 原来直接在 API 进程内跑用户代码，有三个问题： - /debug 是 async def 却做同步的 CPU 密集跟踪，单个请求会堵死事件循环 - exec_script_str_local 传了 disable_security_checks=True，setrlimit 被跳过， MAX_EXECUTED_LINES 只能限制行事件数，挡不住 sum(range(10**9)) 这类单行重计算 - input_string_queue 和 sys.stdout 都是模块级全局，并发请求会互相干扰 改成在子进程里执行：子进程内设 CPU 5s / 内存 512MB（软硬限留差值， 让 SIGXCPU 先于 SIGKILL 到达以便区分原因），父进程再加 15s 墙钟超时兜住 time.sleep 这类不耗 CPU 的挂起。端点改同步 def 走线程池，异常统一转成 400 + 中文 detail。 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_019QGZaLUWzGPEMTk6A5RShC --- debug_runner.py | 129 ++++++++++++++++++++++++++++++++++++++++++++++++ main.py | 21 ++++---- 2 files changed, 139 insertions(+), 11 deletions(-) create mode 100644 debug_runner.py diff --git a/debug_runner.py b/debug_runner.py new file mode 100644 index 0000000..915092e --- /dev/null +++ b/debug_runner.py @@ -0,0 +1,129 @@ +"""在独立子进程中执行 Python Tutor 跟踪。 + +pg_logger 是在 API 进程内直接跑用户代码的：它会替换全局 sys.stdout、 +用模块级全局变量 input_string_queue 传输入，并且 MAX_EXECUTED_LINES 只能 +限制"行事件"数量，挡不住 sum(range(10**9)) 这种单行重计算。因此这里把 +执行整体挪进子进程： + +- 子进程内 setrlimit 限制 CPU 和内存（SIGXCPU 只会杀子进程）； +- 父进程再加一层墙钟超时，兜住 time.sleep 这类不耗 CPU 的挂起； +- 全局 stdout / input_string_queue 天然按请求隔离，可以安全并发。 + +本文件同时是父进程的调用入口和子进程的脚本入口： + python debug_runner.py <请求 json> <结果 json> +""" + +import json +import os +import subprocess +import sys +import tempfile + +# 子进程资源上限 +CPU_LIMIT_SECONDS = 5 +MEMORY_LIMIT_BYTES = 512 * 1024 * 1024 +# 父进程墙钟上限，要比 CPU 上限宽，留出解释器启动和跟踪开销 +WALL_TIMEOUT_SECONDS = 15 + +# SIGXCPU / SIGKILL 对应的返回码（subprocess 用负数表示被信号终止） +_SIGXCPU_RETURNCODE = -24 +_SIGKILL_RETURNCODE = -9 + + +class DebugError(Exception): + """调试执行失败，消息可直接展示给用户""" + + +# ==================== 父进程 ==================== +def run_debug(code: str, inputs: list[str]) -> dict: + """在子进程里跟踪执行 code，返回 {"code": ..., "trace": [...]}""" + with tempfile.TemporaryDirectory(prefix="debug-") as workdir: + request_path = os.path.join(workdir, "request.json") + result_path = os.path.join(workdir, "result.json") + + with open(request_path, "w", encoding="utf-8") as f: + json.dump({"code": code, "inputs": inputs}, f) + + try: + proc = subprocess.run( + [sys.executable, __file__, request_path, result_path], + cwd=os.path.dirname(os.path.abspath(__file__)), + capture_output=True, + timeout=WALL_TIMEOUT_SECONDS, + ) + except subprocess.TimeoutExpired: + raise DebugError( + f"代码运行超过 {WALL_TIMEOUT_SECONDS} 秒，请检查是否有死循环或长时间等待" + ) + + if proc.returncode == _SIGXCPU_RETURNCODE: + raise DebugError( + f"代码占用 CPU 超过 {CPU_LIMIT_SECONDS} 秒，请减少计算量或循环次数" + ) + if proc.returncode == _SIGKILL_RETURNCODE: + raise DebugError("代码占用资源过多，已被终止") + + if not os.path.exists(result_path): + stderr = proc.stderr.decode("utf-8", "replace").strip() + raise DebugError(f"调试进程异常退出：{stderr[-500:] or '无输出'}") + + with open(result_path, encoding="utf-8") as f: + result = json.load(f) + + if "error" in result: + raise DebugError(result["error"]) + return result + + +# ==================== 子进程 ==================== +def _apply_limits() -> None: + try: + import resource + except ImportError: # 非 POSIX 平台，跳过 + return + + # 软硬限之间留出差值：软限先发 SIGXCPU（可识别成"CPU 超时"）， + # 若同时设成一样的值，内核会直接发 SIGKILL，就分辨不出原因了 + resource.setrlimit( + resource.RLIMIT_CPU, (CPU_LIMIT_SECONDS, CPU_LIMIT_SECONDS + 2) + ) + resource.setrlimit(resource.RLIMIT_AS, (MEMORY_LIMIT_BYTES, MEMORY_LIMIT_BYTES)) + + +def _main() -> None: + request_path, result_path = sys.argv[1], sys.argv[2] + + with open(request_path, encoding="utf-8") as f: + request = json.load(f) + + # 先把结果文件打开，避免用户代码耗尽资源后写不出结果 + result_file = open(result_path, "w", encoding="utf-8") + + # 在设限之前完成导入，用户代码执行期间不再需要读文件系统 + from pg_logger import exec_script_str_local + + data: dict = {} + + def dump(input_code, output_trace): + data.update(code=input_code, trace=output_trace) + + _apply_limits() + + try: + exec_script_str_local(request["code"], request["inputs"], False, False, dump) + except MemoryError: + data = {"error": "代码占用内存过多，请减少数据规模"} + except RecursionError: + data = {"error": "递归层数过深，请检查递归的终止条件"} + except BaseException as e: # 跟踪器自身出错，不能让它变成 500 + data = {"error": f"调试执行失败：{type(e).__name__}: {e}"} + + if not data: + data = {"error": "调试执行失败：未生成跟踪数据"} + + json.dump(data, result_file) + result_file.close() + + +if __name__ == "__main__": + _main() diff --git a/main.py b/main.py index 3eb77fa..299bd94 100644 --- a/main.py +++ b/main.py @@ -13,7 +13,7 @@ from schemas import ( FormatResponse, ) from database import DatabaseService -from pg_logger import exec_script_str_local +from debug_runner import DebugError, run_debug from formatter import format_code, FormatError from dotenv import load_dotenv @@ -132,17 +132,16 @@ async def ai_analysis(request: AIAnalysisRequest): @app.post("/debug") -async def debug(request: DebugRequest): - """调试端点""" - code = request.code - inputs = request.inputs +def debug(request: DebugRequest) -> dict: + """调试端点 - data = {} - - def dump(input_code, output_trace): - data.update(dict(code=input_code, trace=output_trace)) - - exec_script_str_local(code, inputs, False, False, dump) + 用同步 def 而不是 async def：跟踪执行是 CPU 密集的阻塞调用， + 交给 FastAPI 的线程池，避免堵死事件循环。 + """ + try: + data = run_debug(request.code, request.inputs) + except DebugError as e: + raise HTTPException(status_code=400, detail=str(e)) return {"data": data}