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 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
5fd79a5649
commit
8d1cb4a53b
@@ -1,5 +1,10 @@
|
||||
# 印度占星 Skill 更新日志
|
||||
|
||||
## 2026-10-28 — 两人同时做校正或生成报告时不再互相拖慢
|
||||
|
||||
- 排盘服务把最重的三类计算(生时校正打分、24 小时分块扫描、长报告参考)放到两个独立的工作进程里。两个人同时算时,各自的等待时间和一个人单独算差不多(以前接近翻倍)。结果不变。
|
||||
- Skill 版本不变(BUG-1493,issue #133)。
|
||||
|
||||
## 2026-10-28 — 星盘页的大运日期和报告对齐
|
||||
|
||||
- 星盘页、对话旁星盘、今日星语、KP 月度报告里的大运和小运起止日期,改为和长报告一样从真实出生时刻起算。出生时间不在 0 点的盘,有些边界日期会挪一天(最多一天);大运顺序和年数不变。
|
||||
|
||||
@@ -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`。
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Process pool for the heaviest API computations (issue #133)."""
|
||||
@@ -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)
|
||||
@@ -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))}')
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user