From d1380874e46c6626adafeffd05e34fd19189792d Mon Sep 17 00:00:00 2001 From: greenthree <1395214327@qq.com> Date: Sun, 2 Aug 2026 17:30:16 +0800 Subject: [PATCH] Optimize Unix managed process startup --- .github/workflows/ci.yml | 2 + CHANGELOG.md | 2 + package.json | 1 + probhub/_unix_exec.py | 52 -------------- probhub/process_control.py | 63 +++++++++++++--- references/process-control.md | 4 +- scripts/benchmark_process_startup.py | 104 +++++++++++++++++++++++++++ scripts/check_release.py | 11 +-- tests/test_npm_packages.py | 25 +++++++ tests/test_process_control.py | 74 +++++++++++++++++++ 10 files changed, 272 insertions(+), 66 deletions(-) delete mode 100644 probhub/_unix_exec.py create mode 100644 scripts/benchmark_process_startup.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f6eb1e0..8a9a3ae 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -100,6 +100,8 @@ jobs: - run: python -m pip install --upgrade pip - run: python -m pip install -r requirements.txt - run: npm ci + - name: Record managed short-process startup benchmark + run: python scripts/benchmark_process_startup.py --rounds 40 --warmup 5 - run: npm run check - name: Verify both npm package contents run: npm run pack:check diff --git a/CHANGELOG.md b/CHANGELOG.md index 9da49d4..d00a401 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,8 @@ ## [Unreleased] +- Unix 内存限制 helper 改为隔离 Python 的内联 `-c` 启动，保留 `RLIMIT_AS`、状态管道、信号复位、session/process group 与 fail-closed 语义，同时避免 WSL 从 Windows 挂载目录逐次读取 helper 文件的冷启动开销。 +- 发布包清单检查同时兼容 npm 10 的单元素数组与 npm 12 的单包名对象 JSON 响应，畸形或多包响应仍 fail closed。 - 主包 README、兼容包 README、Agent Skill 与 Release 安装说明统一为 Node.js 18+ / Python 3.10+ 的系统 Python 显式授权流程；非虚拟环境依赖只写入用户目录并兼容 Ubuntu PEP 668，pip 子进程清除 Python 环境污染；临时 `npx`、Doctor 修复和 WebUI 检查不再遗漏 `PROBHUB_ALLOW_SYSTEM_PYTHON=1`，安装器报错不再引导用户创建虚拟环境。 - 空工作区锁文件只在取得 OS 文件锁后初始化，消除 Windows 并发 generation 首次启动时的写入、刷新与关闭竞态。 - Build Manifest 升至 schema v4、试卷 generation 升至 schema v3，统一记录 ProbHub/Core、Typst、pypdf、模板与固定字体的 `builder_fingerprint`；`status` 提供字段级 stale 原因，旧 schema 和不可探测工具链不再误报 `current`，build/seal/generation 在发布前以 `builder_changed` 阻断身份漂移。Noto Sans CJK SC 与许可证改为随 npm 包发布，Typst 正式编译只使用校验后的包内字体。 diff --git a/package.json b/package.json index aa8a8f1..a61e2a2 100644 --- a/package.json +++ b/package.json @@ -51,6 +51,7 @@ "!scripts/check_clean_install.py", "!scripts/check_release.py", "!scripts/audit_python_dependencies.py", + "!scripts/benchmark_process_startup.py", "probhub/**/*.py", "probhub/assets/fonts/**" ], diff --git a/probhub/_unix_exec.py b/probhub/_unix_exec.py deleted file mode 100644 index c205d62..0000000 --- a/probhub/_unix_exec.py +++ /dev/null @@ -1,52 +0,0 @@ -"""Apply Unix-only limits, then replace this helper with the target process.""" - -import os -import signal -import sys - - -READY = b"PROBHUB_UNIX_EXEC_READY_V1\n" - - -def _report_failure(status_fd, message): - try: - os.write(status_fd, ("Unix execution helper failed: " + message).encode("utf-8")) - except OSError: - pass - - -def main(argv=None): - argv = list(sys.argv[1:] if argv is None else argv) - status_fd = None - try: - if len(argv) < 7 or argv[0] != "--memory-limit-bytes" or argv[2] != "--status-fd": - raise ValueError("invalid helper arguments") - limit_bytes = int(argv[1]) - status_fd = int(argv[3]) - if argv[4] not in ("--restore-signals", "--keep-signals"): - raise ValueError("invalid signal restore mode") - restore_signals = argv[4] == "--restore-signals" - if argv[5] != "--" or not argv[6:]: - raise ValueError("target command is missing") - if limit_bytes <= 0: - raise ValueError("memory limit must be positive") - - import resource - - os.set_inheritable(status_fd, False) - resource.setrlimit(resource.RLIMIT_AS, (limit_bytes, limit_bytes)) - if restore_signals: - for name in ("SIGPIPE", "SIGXFZ", "SIGXFSZ"): - signum = getattr(signal, name, None) - if signum is not None: - signal.signal(signum, signal.SIG_DFL) - os.write(status_fd, READY) - os.execvpe(argv[6], argv[6:], os.environ) - except BaseException as exc: - if status_fd is not None: - _report_failure(status_fd, str(exc)) - return 127 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/probhub/process_control.py b/probhub/process_control.py index 486d340..0925f42 100644 --- a/probhub/process_control.py +++ b/probhub/process_control.py @@ -21,6 +21,56 @@ UNIX_EXEC_START_TIMEOUT_SECONDS = 10.0 UNIX_EXEC_READY = b"PROBHUB_UNIX_EXEC_READY_V1\n" UNIX_EXEC_STATUS_LIMIT = 64 * 1024 +UNIX_EXEC_HELPER_CODE = r''' +import os +import signal +import sys + +READY = b"PROBHUB_UNIX_EXEC_READY_V1\n" + + +def report_failure(status_fd, message): + try: + os.write(status_fd, ("Unix execution helper failed: " + message).encode("utf-8")) + except OSError: + pass + + +def main(): + argv = sys.argv[1:] + status_fd = None + try: + if len(argv) < 7 or argv[0] != "--memory-limit-bytes" or argv[2] != "--status-fd": + raise ValueError("invalid helper arguments") + limit_bytes = int(argv[1]) + status_fd = int(argv[3]) + if argv[4] not in ("--restore-signals", "--keep-signals"): + raise ValueError("invalid signal restore mode") + restore_signals = argv[4] == "--restore-signals" + if argv[5] != "--" or not argv[6:]: + raise ValueError("target command is missing") + if limit_bytes <= 0: + raise ValueError("memory limit must be positive") + + import resource + + os.set_inheritable(status_fd, False) + resource.setrlimit(resource.RLIMIT_AS, (limit_bytes, limit_bytes)) + if restore_signals: + for name in ("SIGPIPE", "SIGXFZ", "SIGXFSZ"): + signum = getattr(signal, name, None) + if signum is not None: + signal.signal(signum, signal.SIG_DFL) + os.write(status_fd, READY) + os.execvpe(argv[6], argv[6:], os.environ) + except BaseException as exc: + if status_fd is not None: + report_failure(status_fd, str(exc)) + return 127 + + +raise SystemExit(main()) +'''.lstrip() class ProcessCancelled(Exception): @@ -112,21 +162,19 @@ def _prepare_unix_memory_limited_command(command, memory_limit_mb): command = list(command) if not command: raise ValueError("memory-limited Unix target command is missing") - helper = Path(__file__).with_name("_unix_exec.py") - if not helper.is_file(): - raise OSError(f"Unix execution helper is missing: {helper}") limit_bytes = int(float(memory_limit_mb) * 1024 * 1024) if limit_bytes <= 0: raise ValueError("memory limit must be positive") - return command, helper, limit_bytes + return command, limit_bytes -def _unix_memory_limited_command(command, helper, limit_bytes, status_fd, restore_signals): +def _unix_memory_limited_command(command, limit_bytes, status_fd, restore_signals): return [ sys.executable, "-I", "-S", - str(helper), + "-c", + UNIX_EXEC_HELPER_CODE, "--memory-limit-bytes", str(limit_bytes), "--status-fd", @@ -676,7 +724,7 @@ def spawn_managed( if kwargs.get("executable") is not None: raise ValueError("memory-limited Unix commands do not support executable=") pass_fds = tuple(kwargs.get("pass_fds") or ()) - command, helper, limit_bytes = _prepare_unix_memory_limited_command( + command, limit_bytes = _prepare_unix_memory_limited_command( command, memory_limit_mb ) read_fd, write_fd = os.pipe() @@ -686,7 +734,6 @@ def spawn_managed( kwargs["close_fds"] = True command = _unix_memory_limited_command( command, - helper, limit_bytes, write_fd, kwargs.get("restore_signals", True), diff --git a/references/process-control.md b/references/process-control.md index 1c15bf8..2c5aea4 100644 --- a/references/process-control.md +++ b/references/process-control.md @@ -87,11 +87,11 @@ Windows 虚拟环境的 `python.exe` 可能是一个重定向启动器，尤其 Linux/Unix 使用： - 独立 session/process group； -- 由独立 exec helper 在子进程内设置 `RLIMIT_AS`，随后以 `exec` 原位替换为目标程序； +- 由隔离 Python 启动的内联 exec helper 在子进程内设置 `RLIMIT_AS`，随后以 `exec` 原位替换为目标程序； - Linux `/proc` 低频采样整棵进程树的 RSS 和进程数； - `killpg` 在结束时清理进程组。 -Flask 多线程请求和 CLI 都不会在父进程中使用 Python `preexec_fn`。helper 会通过仅在成功 `exec` 时关闭的状态管道报告参数、`setrlimit` 或 `exec` 失败；启动阶段超时同样 fail closed，不会退化成无内存限制执行。 +Flask 多线程请求和 CLI 都不会在父进程中使用 Python `preexec_fn`。helper 代码通过 `python -I -S -c` 传入，避免 WSL 在 Windows 挂载目录中为每个短进程重新打开 helper 脚本；它仍通过仅在成功 `exec` 时关闭的状态管道报告参数、`setrlimit` 或 `exec` 失败。启动阶段超时同样 fail closed，不会退化成无内存限制执行。 资源采样约每 `50 ms` 进行一次，时间和输出检查使用更短轮询，以降低大量短进程和 stress 场景的监控开销。Linux 的 `RLIMIT_AS` 是每进程地址空间限制，和 Windows Job 的整树共享内存配额并不完全等价；ProbHub 同时使用 `/proc` 聚合 RSS 做补充监控。无法使用 `/proc` 时，进程组清理与 `RLIMIT_AS` 仍然有效，但进程数和聚合内存遥测能力会受限。 diff --git a/scripts/benchmark_process_startup.py b/scripts/benchmark_process_startup.py new file mode 100644 index 0000000..4d3943f --- /dev/null +++ b/scripts/benchmark_process_startup.py @@ -0,0 +1,104 @@ +#!/usr/bin/env python3 +"""Measure managed short-process startup latency with and without a memory limit.""" + +import argparse +import json +import math +import platform +import statistics +import subprocess +import sys +import time +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from probhub.process_control import spawn_managed + + +def _target_command(): + if platform.system() == "Windows": + return [sys.executable, "-I", "-S", "-c", "pass"] + return ["/bin/true"] + + +def _percentile(values, percentile): + index = max(0, min(len(values) - 1, math.ceil(len(values) * percentile) - 1)) + return values[index] + + +def _measure(command, *, rounds, warmup, memory_limit_mb): + values = [] + for index in range(rounds + warmup): + started = time.perf_counter() + managed = spawn_managed( + command, + memory_limit_mb=memory_limit_mb, + process_limit=8, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + try: + returncode = managed.proc.wait(timeout=10) + finally: + try: + if managed.proc.poll() is None: + managed.terminate() + managed.proc.wait(timeout=2) + finally: + managed.close() + if returncode != 0: + raise RuntimeError(f"benchmark target exited with {returncode}") + elapsed_ms = (time.perf_counter() - started) * 1000 + if index >= warmup: + values.append(elapsed_ms) + values.sort() + return { + "rounds": len(values), + "mean_ms": statistics.mean(values), + "p50_ms": statistics.median(values), + "p95_ms": _percentile(values, 0.95), + "min_ms": values[0], + "max_ms": values[-1], + } + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--rounds", type=int, default=100) + parser.add_argument("--warmup", type=int, default=10) + parser.add_argument("--memory-limit-mb", type=int, default=256) + args = parser.parse_args(argv) + if args.rounds <= 0 or args.warmup < 0 or args.memory_limit_mb <= 0: + parser.error("rounds and memory limit must be positive; warmup must be non-negative") + + command = _target_command() + unlimited = _measure( + command, + rounds=args.rounds, + warmup=args.warmup, + memory_limit_mb=None, + ) + limited = _measure( + command, + rounds=args.rounds, + warmup=args.warmup, + memory_limit_mb=args.memory_limit_mb, + ) + print(json.dumps({ + "platform": platform.platform(), + "python": sys.version.split()[0], + "command": command, + "memory_limit_mb": args.memory_limit_mb, + "unlimited": unlimited, + "memory_limited": limited, + "incremental_p50_ms": limited["p50_ms"] - unlimited["p50_ms"], + }, ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/check_release.py b/scripts/check_release.py index 2b091df..fae1b52 100644 --- a/scripts/check_release.py +++ b/scripts/check_release.py @@ -145,9 +145,13 @@ def _parse_npm_json(stdout, command): payload = json.loads(stdout) except json.JSONDecodeError as exc: raise ReleaseCheckError(f"invalid npm JSON from {' '.join(command)}: {stdout[-2000:]}") from exc - if not isinstance(payload, list) or len(payload) != 1 or not isinstance(payload[0], dict): - raise ReleaseCheckError(f"unexpected npm pack response: {payload!r}") - return payload[0] + if isinstance(payload, list) and len(payload) == 1 and isinstance(payload[0], dict): + return payload[0] + if isinstance(payload, dict) and len(payload) == 1: + manifest = next(iter(payload.values())) + if isinstance(manifest, dict): + return manifest + raise ReleaseCheckError(f"unexpected npm pack response: {payload!r}") def npm_pack_manifest(target=None, *, dry_run=True, destination=None): @@ -185,7 +189,6 @@ def validate_pack_inventories(*, dry_run=True, destination=None): "requirements.txt", "bin/init.js", "bin/probhub.js", "bin/python.js", "probhub/__init__.py", "probhub/cli.py", "probhub/install_deps.py", "probhub/install_skill.py", "probhub/process_control.py", - "probhub/_unix_exec.py", "probhub/webui_runtime.py", "probhub/assets/fonts/NotoSansCJKsc-Regular.otf", "probhub/assets/fonts/OFL.txt", diff --git a/tests/test_npm_packages.py b/tests/test_npm_packages.py index 4d1a2c3..54a33ff 100644 --- a/tests/test_npm_packages.py +++ b/tests/test_npm_packages.py @@ -100,6 +100,31 @@ def test_release_metadata_gate_passes_for_the_source_tree(self): self.assertEqual(result.returncode, 0, result.stderr) self.assertTrue(json.loads(result.stdout)["ok"]) + def test_npm_pack_json_accepts_legacy_and_current_single_package_shapes(self): + from scripts import check_release + + manifest = {"name": "probhub", "files": [{"path": "package.json"}]} + command = ["npm", "pack", "--json"] + self.assertEqual( + check_release._parse_npm_json(json.dumps([manifest]), command), + manifest, + ) + self.assertEqual( + check_release._parse_npm_json( + json.dumps({"probhub": manifest}), + command, + ), + manifest, + ) + + for payload in ([], {}, {"probhub": []}, {"one": manifest, "two": manifest}): + with self.subTest(payload=payload): + with self.assertRaisesRegex( + check_release.ReleaseCheckError, + "unexpected npm pack response", + ): + check_release._parse_npm_json(json.dumps(payload), command) + def test_publish_release_gate_requires_matching_tag_and_clean_worktree(self): from scripts import check_release diff --git a/tests/test_process_control.py b/tests/test_process_control.py index c6d37c3..bbeebc7 100644 --- a/tests/test_process_control.py +++ b/tests/test_process_control.py @@ -279,6 +279,18 @@ def test_spawn_rejects_caller_preexec_fn(self): stderr=subprocess.DEVNULL, ) + def test_unix_memory_limit_helper_is_passed_inline(self): + command = process_control._unix_memory_limited_command( + ["/bin/true"], + 256 * 1024 * 1024, + 17, + True, + ) + self.assertEqual(command[:4], [sys.executable, "-I", "-S", "-c"]) + self.assertEqual(command[4], process_control.UNIX_EXEC_HELPER_CODE) + self.assertNotIn("_unix_exec.py", command) + self.assertEqual(command[-2:], ["--", "/bin/true"]) + @mock.patch.object(process_control.platform, "system", return_value="Linux") @mock.patch.object(process_control.os, "pipe") def test_invalid_unix_memory_limited_command_does_not_open_status_pipe( @@ -321,6 +333,20 @@ def test_unix_exec_helper_applies_address_space_limit(self): self.assertEqual(managed.proc.returncode, 0, stderr) self.assertEqual(int(stdout.strip()), limit_mb * 1024 * 1024) + @unittest.skipIf(platform.system() == "Windows", "Unix exec helper only") + def test_unix_exec_status_fd_closes_before_target_exits(self): + managed = process_control.spawn_managed( + [sys.executable, "-c", "import time; time.sleep(2)"], + memory_limit_mb=256, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + try: + self.assertIsNone(managed.proc.poll()) + finally: + managed.terminate() + managed.close() + @unittest.skipUnless(Path("/proc/self/status").is_file(), "Linux proc status only") def test_unix_exec_helper_restores_popen_signal_defaults(self): managed = process_control.spawn_managed( @@ -341,6 +367,54 @@ def test_unix_exec_helper_restores_popen_signal_defaults(self): if signum is not None: self.assertEqual(ignored & (1 << (signum - 1)), 0, name) + @unittest.skipUnless(Path("/proc/self/status").is_file(), "Linux proc status only") + def test_unix_exec_helper_can_preserve_ignored_signals(self): + managed = process_control.spawn_managed( + ["/bin/sh", "-c", "grep '^SigIgn:' /proc/self/status"], + memory_limit_mb=256, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + restore_signals=False, + ) + try: + stdout, stderr = managed.proc.communicate(timeout=10) + finally: + managed.close() + self.assertEqual(managed.proc.returncode, 0, stderr) + ignored = int(stdout.split()[-1], 16) + sigpipe = getattr(signal, "SIGPIPE", None) + if sigpipe is not None: + self.assertNotEqual(ignored & (1 << (sigpipe - 1)), 0) + + @unittest.skipIf(platform.system() == "Windows", "Unix pass_fds only") + def test_unix_exec_helper_preserves_caller_pass_fds(self): + read_fd, write_fd = os.pipe() + try: + managed = process_control.spawn_managed( + [ + sys.executable, + "-c", + f"import os; os.write({write_fd}, b'caller-fd')", + ], + memory_limit_mb=256, + pass_fds=(write_fd,), + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + ) + os.close(write_fd) + write_fd = None + try: + _, stderr = managed.proc.communicate(timeout=10) + finally: + managed.close() + self.assertEqual(managed.proc.returncode, 0, stderr) + self.assertEqual(os.read(read_fd, 64), b"caller-fd") + finally: + os.close(read_fd) + if write_fd is not None: + os.close(write_fd) + @unittest.skipIf(platform.system() == "Windows", "Unix selector only") def test_unix_exec_status_supports_high_numbered_fd(self): try: