perf(runs): build and serialize run detail off the event loop
GET /runs/{run_id} built its response synchronously inside an async
handler and returned the model, so both the build and FastAPI's JSON
encoding ran on the event loop. A full payload for a multi-symbol
backtest is ~70MB and stalled every other request for about a second.
Build and encode inside run_in_threadpool and return the rendered
bytes. The chart_payload/chart_symbol variants move into the same
call. Response bytes are unchanged: sha256 matches the previous
implementation on six real runs, full and summary variants.
The encoder itself is a single native call that holds the GIL, so a
~200ms stall per full payload remains; shrinking that needs a smaller
default payload and is left out of this change.
Signed-off-by: alex <alexwan02@163.com>
This commit is contained in:
1 parent
d7f1b9e22b
commit
f1d4ebb5c1
3 files changed
+143
-17
No files matched your search
@@ -19,7 +19,7 @@ from typing import Any, Awaitable, Callable, Deque, Dict, Generator, List, Optio
|
||||
|
||||
from fastapi import Depends, FastAPI, HTTPException, Query, status
|
||||
from fastapi.concurrency import run_in_threadpool
|
||||
from fastapi.responses import JSONResponse
|
||||
from fastapi.responses import JSONResponse, Response
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -648,6 +648,46 @@ def _run_response_payload(response: Any) -> Dict[str, Any]:
|
||||
return response.model_dump(mode="json")
|
||||
|
||||
|
||||
def _render_run_result(
|
||||
run_dir: Path,
|
||||
chart_symbol: Optional[str],
|
||||
chart_payload: Optional[str],
|
||||
) -> Response:
|
||||
"""Build and serialize a run detail response.
|
||||
|
||||
Full responses reach tens of megabytes, so both the build and the JSON
|
||||
encoding run here, off the event loop; returning the model instead would
|
||||
leave FastAPI to encode it on the loop and stall every other request.
|
||||
|
||||
Args:
|
||||
run_dir: Existing run directory.
|
||||
chart_symbol: Optional single chart symbol.
|
||||
chart_payload: Optional chart payload mode.
|
||||
|
||||
Returns:
|
||||
A rendered JSON response.
|
||||
"""
|
||||
wants_chart_meta = bool(chart_payload or chart_symbol)
|
||||
chart_symbols: List[str] = []
|
||||
response = _build_response_from_run_dir(
|
||||
run_dir,
|
||||
elapsed=0.0,
|
||||
include_analysis=True,
|
||||
chart_symbol=chart_symbol,
|
||||
chart_payload=chart_payload or "full",
|
||||
chart_symbols_out=chart_symbols if wants_chart_meta else None,
|
||||
)
|
||||
|
||||
if wants_chart_meta:
|
||||
payload = _run_response_payload(response)
|
||||
payload["chart_symbols"] = chart_symbols
|
||||
return JSONResponse(payload)
|
||||
|
||||
return Response(
|
||||
content=response.model_dump_json(by_alias=True), media_type="application/json"
|
||||
)
|
||||
|
||||
|
||||
def _build_response_from_run_dir(
|
||||
run_dir: Path,
|
||||
elapsed: float,
|
||||
@@ -963,24 +1003,10 @@ def register_runs_routes(
|
||||
detail=f"Run {run_id} not found"
|
||||
)
|
||||
|
||||
wants_chart_meta = bool(chart_payload or chart_symbol)
|
||||
chart_symbols: List[str] = []
|
||||
response = _build_response_from_run_dir(
|
||||
run_dir,
|
||||
elapsed=0.0,
|
||||
include_analysis=True,
|
||||
chart_symbol=chart_symbol,
|
||||
chart_payload=chart_payload or "full",
|
||||
chart_symbols_out=chart_symbols if wants_chart_meta else None,
|
||||
return await run_in_threadpool(
|
||||
_render_run_result, run_dir, chart_symbol, chart_payload
|
||||
)
|
||||
|
||||
if wants_chart_meta:
|
||||
payload = _run_response_payload(response)
|
||||
payload["chart_symbols"] = chart_symbols
|
||||
return JSONResponse(payload)
|
||||
|
||||
return response
|
||||
|
||||
@app.get("/runs/{run_id}/positions/sectors", dependencies=[Depends(require_auth)])
|
||||
async def get_run_positions_sectors(
|
||||
run_id: str,
|
||||
|
||||
@@ -0,0 +1,22 @@
|
||||
"""Shared fixtures for route tests under ``tests/api``."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Callable, List
|
||||
|
||||
import pytest
|
||||
|
||||
from src.api import runs_routes
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def offloaded(monkeypatch: pytest.MonkeyPatch) -> List[Callable[..., object]]:
|
||||
"""Record callables the runs routes hand to ``run_in_threadpool``."""
|
||||
calls: List[Callable[..., object]] = []
|
||||
|
||||
async def fake_run_in_threadpool(func, *args, **kwargs):
|
||||
calls.append(func)
|
||||
return func(*args, **kwargs)
|
||||
|
||||
monkeypatch.setattr(runs_routes, "run_in_threadpool", fake_run_in_threadpool)
|
||||
return calls
|
||||
@@ -0,0 +1,78 @@
|
||||
"""GET /runs/{run_id} builds and encodes its payload off the event loop."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
import api_server
|
||||
from src.api import runs_routes
|
||||
|
||||
RUN_ID = "run_20250101_000000"
|
||||
|
||||
|
||||
def _client(tmp_path: Path, monkeypatch) -> TestClient:
|
||||
monkeypatch.setattr(api_server, "RUNS_DIR", tmp_path / "runs")
|
||||
return TestClient(api_server.app, client=("127.0.0.1", 50000))
|
||||
|
||||
|
||||
def _make_run(tmp_path: Path) -> Path:
|
||||
run_dir = tmp_path / "runs" / RUN_ID
|
||||
artifacts = run_dir / "artifacts"
|
||||
artifacts.mkdir(parents=True)
|
||||
(run_dir / "state.json").write_text('{"status": "success"}\n', encoding="utf-8")
|
||||
(run_dir / "req.json").write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"context": {
|
||||
"codes": ["AAPL"],
|
||||
"start_date": "2025-01-01",
|
||||
"end_date": "2025-01-02",
|
||||
}
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
(artifacts / "price_series.csv").write_text(
|
||||
"timestamp,code,open,high,low,close,volume\n2025-01-01,AAPL,1,2,1,2,100\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
(artifacts / "equity.csv").write_text(
|
||||
"timestamp,equity,drawdown\n2025-01-01,100000,0\n2025-01-02,101000,0\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
return run_dir
|
||||
|
||||
|
||||
def test_full_run_detail_is_rendered_in_threadpool(
|
||||
tmp_path: Path, monkeypatch, offloaded: list
|
||||
) -> None:
|
||||
client = _client(tmp_path, monkeypatch)
|
||||
run_dir = _make_run(tmp_path)
|
||||
|
||||
response = client.get(f"/runs/{RUN_ID}")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.headers["content-type"] == "application/json"
|
||||
assert runs_routes._render_run_result in offloaded
|
||||
expected = api_server._build_response_from_run_dir(
|
||||
run_dir, elapsed=0.0, include_analysis=True
|
||||
)
|
||||
assert response.json() == expected.model_dump(mode="json")
|
||||
|
||||
|
||||
def test_chart_payload_variant_is_rendered_in_threadpool(
|
||||
tmp_path: Path, monkeypatch, offloaded: list
|
||||
) -> None:
|
||||
client = _client(tmp_path, monkeypatch)
|
||||
_make_run(tmp_path)
|
||||
|
||||
response = client.get(f"/runs/{RUN_ID}", params={"chart_payload": "summary"})
|
||||
|
||||
assert response.status_code == 200
|
||||
assert runs_routes._render_run_result in offloaded
|
||||
payload = response.json()
|
||||
assert payload["chart_symbols"] == ["AAPL"]
|
||||
assert payload["price_series"] == {}
|
||||
Reference in new issue
Block a user