From 8d1cb4a53b98a1f45fd24615c3f595de8cde52b2 Mon Sep 17 00:00:00 2001 From: Jesse_Chen Date: Wed, 28 Oct 2026 14:27:10 +0800 Subject: [PATCH] Heavy routes run in worker processes, one per heavy slot (issue #133, BUG-1493) The single ThreadingHTTPServer ran pure-Python v5 score / block_scan / report reference under one GIL, so the gate's two slots shared a core. start_server now starts a spawn pool (workers = heavy concurrency, PYTHONHASHSEED=0, engine pre-imported); those three routes call the same compute method in a worker. Over HTTP: score alone 4.74 s, two at once 5.01 / 4.58 s (1.06x; in-process 1.56x). Results equal except per-process fingerprints; report markdown equal. ~165 MB extra. A crashed pool answers inline and restarts; off with JYOTISH_HEAVY_COMPUTE_PROCESSES=0; tests calling compute methods unchanged. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 5 + docs/BUG_HISTORY.md | 15 ++ scripts/api_process_pool/__init__.py | 1 + scripts/api_process_pool/pool.py | 137 ++++++++++++++++++ scripts/jyotish_api_server.py | 35 ++++- .../test_heavy_compute_processes_20261028.py | 81 +++++++++++ 6 files changed, 266 insertions(+), 8 deletions(-) create mode 100644 scripts/api_process_pool/__init__.py create mode 100644 scripts/api_process_pool/pool.py create mode 100644 tests/test_heavy_compute_processes_20261028.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 87d47a09..aa462351 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,10 @@ # 印度占星 Skill 更新日志 +## 2026-10-28 — 两人同时做校正或生成报告时不再互相拖慢 + +- 排盘服务把最重的三类计算（生时校正打分、24 小时分块扫描、长报告参考）放到两个独立的工作进程里。两个人同时算时，各自的等待时间和一个人单独算差不多（以前接近翻倍）。结果不变。 +- Skill 版本不变（BUG-1493，issue #133）。 + ## 2026-10-28 — 星盘页的大运日期和报告对齐 - 星盘页、对话旁星盘、今日星语、KP 月度报告里的大运和小运起止日期，改为和长报告一样从真实出生时刻起算。出生时间不在 0 点的盘，有些边界日期会挪一天（最多一天）；大运顺序和年数不变。 diff --git a/docs/BUG_HISTORY.md b/docs/BUG_HISTORY.md index 882da381..9901b963 100644 --- a/docs/BUG_HISTORY.md +++ b/docs/BUG_HISTORY.md @@ -20401,3 +20401,18 @@ - 关联：ERR-110（冻结文件）、BUG-1185。 - 复发自：无 - 修复版本：分支 `codex/chart-dasha-birth-moment-20261028`。 + +## BUG-1493 | 排盘服务只有一个进程，两人同时做重计算会抢同一个核 + +- 状态：resolved（代码 + 回归测试；待部署） +- 首次发现 / 最近更新：2026-10-28 / 2026-10-28（issue #118 代码体检第 9 条 → #133） +- 影响面：`/api/rectification/v5/score`、`/api/rectification/v5/block_scan`、`/api/professional_report_reference`（长报告参考）；新模块 `scripts/api_process_pool/pool.py`；`scripts/jyotish_api_server.py` 只在启动处和这三个分支各加一行。 +- 现象：排盘服务是单进程 `ThreadingHTTPServer`，重计算是纯 Python，受 GIL 限制。重计算闸门允许两路同时跑（`JYOTISH_HEAVY_COMPUTE_CONCURRENCY=2`），实际上两路在抢一个核：两个人同时做校正打分时，每人都要等将近一倍的时间。 +- 触发条件：两个以上用户同时做生时校正打分、24 小时分块扫描或生成长报告。 +- 根因：并发上限设成了 2，但执行仍在一个进程里。 +- 修复：服务启动时按重计算并发数起同样多个工作进程（spawn 方式；进程内固定 `PYTHONHASHSEED=0`，见 ERR-111），这三条路由在工作进程里用一个新的 `JyotishComputeService` 跑原来的同一个计算方法；谁先跑仍由原来的重计算闸门决定。工作进程启动时就把引擎载入好，第一个用户不用等。工作进程崩溃时这一次请求改在服务进程里算，进程池在后台重建。只有服务的 `start_server` 会启动进程池，测试里直接调计算方法的路径不变；`JYOTISH_HEAVY_COMPUTE_PROCESSES=0` 可关掉。 +- 验证：本机真起服务（`python scripts/jyotish_api_server.py`，2 个工作进程），经 HTTP 发校正打分（2 小时窗口、6 件事）：单跑 4.74 秒，两路同时 5.01 / 4.58 秒，是单跑的 1.06 倍（单子要求 ≤1.3 倍）；同样两路在单进程里是 1.56 倍。每个工作进程常驻约 75～77 MB，加一个 14 MB 的资源跟踪进程，合计多约 165 MB，4 GB 主机够用。Python 快速门全部 Python 步骤通过（pytest 434.6 s）。结果：工作进程与服务进程算出的打分结果，除了本来就随进程变的特征指纹和耗时字段之外逐项相同（指纹在改前也随每次重启变，不参与任何判断）；长报告参考经进程池生成的 markdown 与进程内逐字节相同（147 万字）。新测试 `tests/test_heavy_compute_processes_20261028.py`（6 条：未启动时在进程内跑；进程池结果与进程内相同；输入错误仍报 BadRequest；名单外的路由不进进程池；进程崩溃时这次在进程内算、随后进程池自动恢复；开关能关）。 +- 防复发：上述测试。CPU 密集的接口要么多进程，要么并发上限设成 1，不要用线程并发掩盖单核。 +- 关联：ERR-111（特征指纹随哈希种子变）、BUG-1482（#132 让模块能按名字被子进程找到）。 +- 复发自：无 +- 修复版本：分支 `codex/heavy-compute-processes-20261028`。 diff --git a/scripts/api_process_pool/__init__.py b/scripts/api_process_pool/__init__.py new file mode 100644 index 00000000..f666d665 --- /dev/null +++ b/scripts/api_process_pool/__init__.py @@ -0,0 +1 @@ +"""Process pool for the heaviest API computations (issue #133).""" diff --git a/scripts/api_process_pool/pool.py b/scripts/api_process_pool/pool.py new file mode 100644 index 00000000..a089aa3f --- /dev/null +++ b/scripts/api_process_pool/pool.py @@ -0,0 +1,137 @@ +"""Run the heaviest compute endpoints in worker processes (issue #133). + +The API server is one ``ThreadingHTTPServer`` process. Its heavy routes are +pure-Python number crunching, so two requests inside the gate's two slots +still shared one core under the GIL: one score alone 1.19 s, two at once +2.35-2.6 s each. Here the same compute methods run on a fresh +``JyotishComputeService`` in a spawn-started worker, one worker per heavy slot, +so two requests use two cores. The heavy-compute gate still decides who runs. + +- Enabled only by the server's ``main()`` (and ``JYOTISH_HEAVY_COMPUTE_PROCESSES`` + not ``0``); tests that call compute methods in process are unchanged. +- Workers start with ``PYTHONHASHSEED=0`` (ERR-111: Shadbala fingerprints vary + with the hash seed), and are started up front so the first user does not + pay the import. +- A crashed pool is replaced; the request that hit it runs in process. +""" + +from __future__ import annotations + +import multiprocessing +import os +import sys +import threading +from concurrent.futures import ProcessPoolExecutor +from concurrent.futures.process import BrokenProcessPool +from typing import Any + +ENV_PROCESSES = "JYOTISH_HEAVY_COMPUTE_PROCESSES" +POOLED_METHODS = frozenset({ + "_compute_rectification_v5_score", + "_compute_rectification_v5_block_scan", + "professional_report_reference", +}) + +_guard = threading.Lock() +_executor: ProcessPoolExecutor | None = None +_workers = 0 +_config: tuple[int, list[str]] | None = None +_service = None + + +def _worker_init(paths: list[str]) -> None: + for path in reversed(paths): + if path not in sys.path: + sys.path.insert(0, path) + + +def _worker_call(method: str, body: dict[str, Any]) -> Any: + global _service + from scripts.jyotish_api_server import BadRequest, JyotishComputeService + + if _service is None: + _service = JyotishComputeService() + if method == "professional_report_reference": + from scripts.professional_report_reference import ( + ProfessionalReportReferenceInputError, + build_professional_report_reference, + ) + try: + return build_professional_report_reference(_service, body) + except ProfessionalReportReferenceInputError as exc: + raise BadRequest(str(exc)) from exc + return getattr(_service, method)(body) + + +def _warm() -> int: + from scripts import jyotish_api_server # noqa: F401 - import the engine once per worker + + return os.getpid() + + +def enabled_by_env() -> bool: + return os.environ.get(ENV_PROCESSES, "1").strip().lower() not in {"0", "false", "no", "off"} + + +def start(workers: int, paths: list[str]) -> bool: + """Start the pool with ``workers`` processes; False when switched off.""" + global _executor, _workers, _config + if workers < 1 or not enabled_by_env(): + return False + _config = (workers, list(paths)) + with _guard: + if _executor is not None: + return True + previous = os.environ.get("PYTHONHASHSEED") + os.environ["PYTHONHASHSEED"] = "0" + try: + executor = ProcessPoolExecutor( + max_workers=workers, + mp_context=multiprocessing.get_context("spawn"), + initializer=_worker_init, + initargs=(paths,), + ) + # Start every worker now, while the hash seed is in the environment. + for future in [executor.submit(_warm) for _ in range(workers)]: + future.result(timeout=120) + finally: + if previous is None: + os.environ.pop("PYTHONHASHSEED", None) + else: + os.environ["PYTHONHASHSEED"] = previous + _executor, _workers = executor, workers + return True + + +def active() -> bool: + return _executor is not None + + +def run(method: str, body: dict[str, Any], inline) -> Any: + """``method`` in a worker when the pool is up, else ``inline(body)`` in this process.""" + executor = _executor + if executor is None or method not in POOLED_METHODS: + return inline(body) + try: + return executor.submit(_worker_call, method, body).result() + except BrokenProcessPool: + _replace_broken(executor) + return inline(body) + + +def _replace_broken(broken: ProcessPoolExecutor) -> None: + global _executor + with _guard: + if _executor is broken: + _executor = None + broken.shutdown(wait=False, cancel_futures=True) + if _config is not None: + threading.Thread(target=start, args=_config, name="heavy-pool-restart", daemon=True).start() + + +def stop() -> None: + global _executor + with _guard: + executor, _executor = _executor, None + if executor is not None: + executor.shutdown(wait=True, cancel_futures=True) diff --git a/scripts/jyotish_api_server.py b/scripts/jyotish_api_server.py index 91a7b1d1..0320b049 100644 --- a/scripts/jyotish_api_server.py +++ b/scripts/jyotish_api_server.py @@ -2536,6 +2536,24 @@ def _birth_moment_vimshottari(**kwargs): return compute_vimshottari_timeline(**kwargs) +def _professional_report_reference_inline(handler, body): + reference_module = _load_local_module('professional_report_reference') + try: + return reference_module.build_professional_report_reference(handler, body) + except reference_module.ProfessionalReportReferenceInputError as exc: + raise BadRequest(str(exc)) from exc + + +def _pooled(handler, method, body): + """A heavy route in a worker process when the server started the pool (issue #133).""" + from scripts.api_process_pool import pool as heavy_pool + if method == 'professional_report_reference': + inline = lambda payload: _professional_report_reference_inline(handler, payload) + else: + inline = getattr(handler, method) + return heavy_pool.run(method, body, inline) + + def _load_local_module(module_name): cached = _LOCAL_MODULE_CACHE.get(module_name) if cached: @@ -10613,12 +10631,7 @@ class JyotishAPIHandler(BaseHTTPRequestHandler, JyotishComputeService): result = self._compute_professional_reading(body) self._json(result) elif path == '/api/professional_report_reference': - reference_module = _load_local_module('professional_report_reference') - try: - result = reference_module.build_professional_report_reference(self, body) - except reference_module.ProfessionalReportReferenceInputError as exc: - raise BadRequest(str(exc)) from exc - self._json(result) + self._json(_pooled(self, 'professional_report_reference', body)) elif path == '/api/import_chart': result = self._import_chart_text(body) self._json(result) @@ -10703,7 +10716,7 @@ class JyotishAPIHandler(BaseHTTPRequestHandler, JyotishComputeService): elif path == '/api/rectification/v5/candidate-features': self._json(self._compute_rectification_v5_candidate_features(body)) elif path == '/api/rectification/v5/score': - self._json(self._compute_rectification_v5_score(body)) + self._json(_pooled(self, '_compute_rectification_v5_score', body)) elif path == '/api/rectification/v5/diagnostics': self._json(self._compute_rectification_v5_diagnostics(body)) elif path == '/api/rectification/v5/vedastro-validate': @@ -10711,7 +10724,7 @@ class JyotishAPIHandler(BaseHTTPRequestHandler, JyotishComputeService): elif path == '/api/rectification/v5/range_reading': self._json(self._compute_rectification_v5_range_reading(body)) elif path == '/api/rectification/v5/block_scan': - self._json(self._compute_rectification_v5_block_scan(body)) + self._json(_pooled(self, '_compute_rectification_v5_block_scan', body)) elif path == '/api/rectification/v5/versions': self._json(self._compute_rectification_v5_versions()) elif path == '/api/dynamic_rectification_opportunities': @@ -10927,6 +10940,12 @@ def start_server(port=5200, host='127.0.0.1', allowed_origins=None, allowed_host server.daemon_threads = True server.allowed_origins = allowed_origins or DEFAULT_ALLOWED_ORIGINS server.allowed_hosts = allowed_hosts or DEFAULT_ALLOWED_HOSTS + # issue #133: one worker process per heavy-compute slot, so two heavy requests use two cores. + from scripts import api_heavy_compute_gate as heavy_gate + from scripts.api_process_pool import pool as heavy_pool + heavy_workers = heavy_gate._parse_positive_int(os.environ.get(heavy_gate.ENV_CONCURRENCY), heavy_gate.DEFAULT_CONCURRENCY) + if heavy_pool.start(heavy_workers, [REPO_ROOT, SCRIPTS_DIR]): + print(f' Heavy compute: {heavy_workers} worker processes') print(f'Jyotish API v6.9.14 running on http://{host}:{port}') print(f" Async job cleanup: scanned={cleanup['scanned']}, removed={cleanup['removed']}") print(f' CORS origins: {", ".join(sorted(server.allowed_origins))}') diff --git a/tests/test_heavy_compute_processes_20261028.py b/tests/test_heavy_compute_processes_20261028.py new file mode 100644 index 00000000..0314488f --- /dev/null +++ b/tests/test_heavy_compute_processes_20261028.py @@ -0,0 +1,81 @@ +"""Issue #133: heavy routes run in worker processes when the server starts the pool.""" + +from __future__ import annotations + +import os +import time + +import pytest + +from scripts.api_process_pool import pool +from scripts.jyotish_api_server import REPO_ROOT, SCRIPTS_DIR, BadRequest, JyotishComputeService +from tests.test_rectification_v5_vedastro_validation import request as v5_request + +NOISE = ("fingerprint", "feature_hash", "_hash", "elapsed", "duration", "_ms", "timing") + + +def _body() -> dict: + body = v5_request() + body.pop("candidate_times") + return body + + +def _strip(value): + if isinstance(value, dict): + return {k: _strip(v) for k, v in value.items() if not any(n in k for n in NOISE)} + if isinstance(value, list): + return [_strip(v) for v in value] + return value + + +@pytest.fixture +def started_pool(): + assert pool.start(2, [REPO_ROOT, SCRIPTS_DIR]) + yield + pool.stop() + + +def test_without_the_pool_routes_run_in_process(): + assert not pool.active() + calls = [] + assert pool.run("_compute_rectification_v5_score", {"x": 1}, lambda body: calls.append(body) or "inline") == "inline" + assert calls == [{"x": 1}] + + +def test_a_pooled_score_equals_the_in_process_one(started_pool): + inline = JyotishComputeService()._compute_rectification_v5_score + pooled = pool.run("_compute_rectification_v5_score", _body(), lambda body: pytest.fail("ran inline")) + assert pooled["success"] is True + # Fingerprints already vary per process (ERR-111); everything else is identical. + assert _strip(pooled) == _strip(inline(_body())) + + +def test_bad_input_is_still_a_bad_request(started_pool): + body = _body() + body["start_time"] = "not-a-time" + with pytest.raises((BadRequest, ValueError, KeyError)): + pool.run("_compute_rectification_v5_score", body, lambda _body: pytest.fail("ran inline")) + + +def test_routes_outside_the_list_never_leave_the_process(started_pool): + assert pool.run("_compute_chart", {}, lambda body: "inline") == "inline" + + +def test_a_crashed_pool_answers_in_process_and_comes_back(started_pool): + victim = next(iter(pool._executor._processes.values())) + os.kill(victim.pid, 9) + time.sleep(0.5) + calls = [] + result = pool.run("_compute_rectification_v5_score", _body(), lambda body: calls.append(1) or "inline") + assert result == "inline" and calls == [1] + for _ in range(120): + if pool.active(): + break + time.sleep(0.5) + assert pool.active(), "the pool restarts in the background" + + +def test_the_switch_turns_the_pool_off(monkeypatch): + monkeypatch.setenv(pool.ENV_PROCESSES, "0") + assert pool.start(2, [REPO_ROOT, SCRIPTS_DIR]) is False + assert not pool.active()