snowy-yang
a9bc8238ab
为什么：snapshot() 同时压在 SSE 每 250ms 推送和 full=1 动作回执两条
高频路径上，而它的成本大头是重复求值——profile 显示单次快照
1,128,001 次函数调用里近半是属性链重复计算同一份配置数据。
- Plot.to_dict：grow_days 原来经 4 个属性各求值一遍（本体 + days_left
+ is_ripe + stage），16 块地就是 64 次配置表查询。改为算一次局部
复用；stage 的实现体拆到 _stage(grow)，规则本体仍只有一份。
- Animal.to_dict：info 配置表原来查 4 次、cycle int() 两遍，改取一次。
- GameState.to_dict：weather_info 3 键 2 次查询；granary_free 原来把
granary_capacity 和 storage_used 整个重算，改算一轮复用。
行为零变化（规则本体全在原属性里，532 项测试全绿）。
基准（tools/bench.py --repeat 5，最坏形态 16 地 + 8 牲畜）：
场景 优化前 优化后
engine.snapshot() 50.1 µs 44.2 µs -12%
session.snapshot() 51.0 µs 41.9 µs -18%
build_state_message() 50.9 µs 43.3 µs -15%
engine.available_actions 52.7 µs 48.7 µs -8%
单次 snapshot 函数调用数 1,128,001 → 588,001（-48%），本次数字已记入
bench.py docstring 作长期参考。重复计数器全 OK（to_dict/snapshot/
net_worth 每请求仍各 1 次，无回归）。
355 lines
14 KiB
Python
355 lines
14 KiB
Python
"""热路径基准：给「优化前后」提供可对比的数字。
|
||
|
||
为什么需要它
|
||
------------
|
||
这个服务同时扛着三条高频路径，而它们**都在重复序列化整局状态**：
|
||
|
||
* 每次 ``POST /api/actions``（TUI 客户端固定带 ``full=1``）
|
||
* ``Publisher`` 每 250ms 一次的 SSE 推送
|
||
* ``GET /api/world`` / ``/api/farms`` 的排行榜（遍历所有牧场）
|
||
|
||
改任何一处序列化代码，代价都会乘以上面这些倍数。所以先有数字，再动手。
|
||
|
||
跑法::
|
||
|
||
uv run python tools/bench.py
|
||
uv run python tools/bench.py --repeat 5 # 每项取最快一轮，抗系统噪声
|
||
|
||
输出两段：
|
||
|
||
1. **重复计数**——一次请求到底触发了几次 ``GameState.to_dict`` /
|
||
``net_worth`` / ``available_actions``。这是优化最直接的证据。
|
||
2. **耗时**——各场景的 µs/op 与 ops/s。
|
||
|
||
参考数字（16 块地 + 8 只牲畜的最坏形态，本机实测）
|
||
--------------------------------------------------
|
||
2026-09-19 那次优化（把 brief/summary 改成读标量、net_worth 只算一次）前后::
|
||
|
||
场景 优化前 优化后
|
||
brief(session, hub) 63.6 µs 15.4 µs -76%
|
||
session.summary() 60.2 µs 13.8 µs -77%
|
||
leaderboard(32 farms) 1981.7 µs 445.8 µs -78%
|
||
engine.snapshot() 57.3 µs 45.0 µs -21%
|
||
POST /api/actions full=1 464.7 µs 390.2 µs -16%
|
||
GET /api/actions 398.3 µs 333.2 µs -16%
|
||
GET /api/world 193.6 µs 133.5 µs -31%
|
||
|
||
一次 ``POST /api/actions?full=1`` 的重复工作量：``to_dict`` 2 → 1 次、
|
||
``snapshot`` 2 → 1 次、``net_worth`` 6 → 2 次。
|
||
|
||
2026-09-26 那次优化（to_dict 里重复求值的属性改局部复用：Plot 的
|
||
grow_days 4→1 次、Animal 的 info 查表 4→1 次、GameState 的仓储三兄弟
|
||
只算一轮）前后，单次 snapshot 的函数调用数 1,128,001 → 588,001（-48%）::
|
||
|
||
场景 优化前 优化后
|
||
engine.snapshot() 50.1 µs 44.2 µs -12%
|
||
session.snapshot() 51.0 µs 41.9 µs -18%
|
||
build_state_message() 50.9 µs 43.3 µs -15%
|
||
engine.available_actions 52.7 µs 48.7 µs -8%
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import json
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
from collections.abc import Callable
|
||
from typing import Any
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from farm import config, models
|
||
from farm.engine import FarmEngine
|
||
from farm.models import Animal, Plot
|
||
from farm.server.hub import ConnectionHub, build_state_message
|
||
from farm.server.session import SessionManager
|
||
|
||
try:
|
||
from farm.server.app import ServerConfig, brief, create_app
|
||
except ImportError as exc: # pragma: no cover - 环境没装服务器依赖时
|
||
raise SystemExit(f"本基准需要服务器依赖（fastapi 等）：{exc}") from None
|
||
|
||
SPEED_FARM = "bench"
|
||
ACTION_BODY = b'{"action":"pause","params":{}}'
|
||
|
||
|
||
# --------------------------------------------------------------------- 造数据
|
||
def fill_endgame(engine: FarmEngine) -> None:
|
||
"""把一个引擎填成"后期农场"：地种满一半、牲畜拉满。
|
||
|
||
挑后期形态是因为它才是序列化的最坏情况——16 块地 + 8 只牲畜。
|
||
"""
|
||
state = engine.state
|
||
state.level = 6 # 6 + 2*(6-1) = 16，正好顶到 MAX_PLOTS
|
||
state.gold = 50000
|
||
state.feed = 40
|
||
state.plots = [Plot(i) for i in range(config.MAX_PLOTS)]
|
||
for i, plot in enumerate(state.plots):
|
||
if i % 2 == 0: # 一半种着、一半空着，两条分支都走到
|
||
plot.crop = "corn"
|
||
plot.moisture = config.WATER_CAPACITY
|
||
plot.growth = 2.0
|
||
state.animals = [Animal("cow", aid=i) for i in range(config.MAX_ANIMALS)]
|
||
for i, animal in enumerate(state.animals):
|
||
animal.fed = i % 2 == 0
|
||
animal.pending = 1 if i % 3 == 0 else 0
|
||
|
||
|
||
def make_session(save_dir: str) -> Any:
|
||
manager = SessionManager(save_dir=save_dir, tick_seconds=config.DEFAULT_TICK_SECONDS)
|
||
session = manager.get(SPEED_FARM, create=True)
|
||
fill_endgame(session.engine)
|
||
return session
|
||
|
||
|
||
# --------------------------------------------------------------------- 计时
|
||
def measure(func: Callable[[], Any], repeat: int = 5, target: float = 0.15) -> float:
|
||
"""返回单次调用耗时（秒）。先估一次，再决定循环几轮让总时长接近 target。"""
|
||
start = time.perf_counter()
|
||
func()
|
||
single = max(time.perf_counter() - start, 1e-9)
|
||
number = max(1, min(20000, int(target / single)))
|
||
best = float("inf")
|
||
for _ in range(repeat):
|
||
start = time.perf_counter()
|
||
for _ in range(number):
|
||
func()
|
||
best = min(best, time.perf_counter() - start)
|
||
return best / number
|
||
|
||
|
||
class Scenario:
|
||
"""一条被测路径。"""
|
||
|
||
def __init__(self, name: str, note: str, func: Callable[[], Any]) -> None:
|
||
self.name = name
|
||
self.note = note
|
||
self.func = func
|
||
self.per_op = 0.0
|
||
|
||
def run(self, repeat: int) -> None:
|
||
self.per_op = measure(self.func, repeat=repeat)
|
||
|
||
|
||
# --------------------------------------------------------------------- ASGI 直连
|
||
async def asgi_call(
|
||
app: Any, method: str, path: str, query: str = "", body: bytes = b""
|
||
) -> tuple[int, dict[str, Any]]:
|
||
"""不经过任何 HTTP 客户端，直接把请求喂给 ASGI 应用。
|
||
|
||
这样基准里既能测到真实路由/中间件开销，又不需要 httpx 之类的依赖。
|
||
|
||
注意：这条路径**不会跑 lifespan**，所以它测不到时钟与推送线程的启动；
|
||
那部分由 ``tests/test_server.py`` 的真机用例覆盖。
|
||
"""
|
||
scope: dict[str, Any] = {
|
||
"type": "http",
|
||
"asgi": {"version": "3.0", "spec_version": "2.3"},
|
||
"http_version": "1.1",
|
||
"method": method,
|
||
"scheme": "http",
|
||
"path": path,
|
||
"raw_path": path.encode(),
|
||
"query_string": query.encode(),
|
||
"root_path": "",
|
||
"headers": [
|
||
(b"host", b"127.0.0.1"),
|
||
(b"content-type", b"application/json"),
|
||
(b"content-length", str(len(body)).encode()),
|
||
],
|
||
"client": ("127.0.0.1", 41234),
|
||
"server": ("127.0.0.1", 8765),
|
||
}
|
||
sent: list[dict[str, Any]] = []
|
||
|
||
async def receive() -> dict[str, Any]:
|
||
return {"type": "http.request", "body": body, "more_body": False}
|
||
|
||
async def send(message: dict[str, Any]) -> None:
|
||
sent.append(message)
|
||
|
||
await app(scope, receive, send)
|
||
status = next(m["status"] for m in sent if m["type"] == "http.response.start")
|
||
raw = b"".join(m.get("body", b"") for m in sent if m["type"] == "http.response.body")
|
||
return status, json.loads(raw) if raw else {}
|
||
|
||
|
||
def call_once(app: Any, method: str, path: str, query: str, body: bytes) -> None:
|
||
"""同步跑一次 ASGI 请求（给"只跑一次"的计数场景用）。"""
|
||
loop = asyncio.new_event_loop()
|
||
try:
|
||
status, _ = loop.run_until_complete(asgi_call(app, method, path, query, body))
|
||
if status >= 400:
|
||
raise RuntimeError(f"{method} {path} 返回 {status}")
|
||
finally:
|
||
loop.close()
|
||
|
||
|
||
def measure_http(
|
||
app: Any, method: str, path: str, query: str, body: bytes, repeat: int = 5, number: int = 300
|
||
) -> float:
|
||
"""批量跑 HTTP 调用；循环放在事件循环内部，避免把建循环的开销算进去。"""
|
||
|
||
async def burst() -> None:
|
||
for _ in range(number):
|
||
status, _ = await asgi_call(app, method, path, query, body)
|
||
if status >= 400:
|
||
raise RuntimeError(f"{method} {path} 返回 {status}")
|
||
|
||
loop = asyncio.new_event_loop()
|
||
try:
|
||
loop.run_until_complete(burst()) # 预热，顺带把牧场建出来
|
||
best = float("inf")
|
||
for _ in range(repeat):
|
||
start = time.perf_counter()
|
||
loop.run_until_complete(burst())
|
||
best = min(best, time.perf_counter() - start)
|
||
return best / number
|
||
finally:
|
||
loop.close()
|
||
|
||
|
||
# --------------------------------------------------------------------- 重复计数
|
||
def count_calls(func: Callable[[], Any], patches: list[tuple[Any, str, str]]) -> dict[str, int]:
|
||
"""给若干方法/属性装计数器，跑一次 ``func()``，返回各标签的命中次数。
|
||
|
||
``patches`` 每项是 ``(类, 属性名, 标签)``；属性会自动识破 property 与普通方法。
|
||
"""
|
||
counters: dict[str, int] = {label: 0 for _, _, label in patches}
|
||
originals: list[tuple[Any, str, Any]] = []
|
||
|
||
for cls, attr, label in patches:
|
||
original = getattr(cls, attr)
|
||
if isinstance(original, property):
|
||
getter = original.fget
|
||
|
||
def count_property(self: Any, _getter: Any = getter, _label: str = label) -> Any:
|
||
counters[_label] += 1
|
||
return _getter(self)
|
||
|
||
replacement: Any = property(count_property)
|
||
else:
|
||
|
||
def count_method(
|
||
self: Any, *args: Any, _orig: Any = original, _label: str = label, **kwargs: Any
|
||
) -> Any:
|
||
counters[_label] += 1
|
||
return _orig(self, *args, **kwargs)
|
||
|
||
replacement = count_method
|
||
|
||
originals.append((cls, attr, original))
|
||
setattr(cls, attr, replacement)
|
||
|
||
try:
|
||
func()
|
||
finally:
|
||
for cls, attr, original in originals:
|
||
setattr(cls, attr, original)
|
||
return counters
|
||
|
||
|
||
# --------------------------------------------------------------------- 主流程
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="SimpleFarm 热路径基准")
|
||
parser.add_argument("--repeat", type=int, default=5, help="每个场景取最快的一轮（默认 5）")
|
||
parser.add_argument("--farms", type=int, default=32, help="排行榜场景的牧场数（默认 32）")
|
||
args = parser.parse_args()
|
||
|
||
save_dir = tempfile.mkdtemp(prefix="farm-bench-")
|
||
session = make_session(save_dir)
|
||
hub = ConnectionHub()
|
||
engine = session.engine
|
||
|
||
cfg = ServerConfig(save_dir=save_dir, tick_seconds=config.DEFAULT_TICK_SECONDS)
|
||
app = create_app(cfg)
|
||
fill_endgame(app.state.manager.get(SPEED_FARM, create=True).engine)
|
||
|
||
# ---------------------------------------------------------- 重复计数
|
||
query = f"farm={SPEED_FARM}&full=1"
|
||
print("=" * 74)
|
||
print("重复工作量")
|
||
print("=" * 74)
|
||
per_action = count_calls(
|
||
lambda: call_once(app, "POST", "/api/actions", query, ACTION_BODY),
|
||
[
|
||
(models.GameState, "to_dict", "to_dict"),
|
||
(FarmEngine, "snapshot", "snapshot"),
|
||
(FarmEngine, "brief_source", "brief_source"),
|
||
],
|
||
)
|
||
expected_per_action = {"to_dict": 1, "snapshot": 1, "brief_source": 1}
|
||
print(" 一次 POST /api/actions?full=1:")
|
||
for label, value in per_action.items():
|
||
want = expected_per_action[label]
|
||
mark = "OK" if value == want else f"!! 期望 {want}"
|
||
print(f" {label:16} {value:3d} 次 {mark}")
|
||
|
||
# 动作回执与完整快照各需要一次 net_worth，所以单独量"一次 to_dict 里算几次"
|
||
per_to_dict = count_calls(engine.state.to_dict, [(models.GameState, "net_worth", "net_worth")])
|
||
count = per_to_dict["net_worth"]
|
||
print(
|
||
f" 一次 GameState.to_dict(): net_worth {count} 次 {'OK' if count == 1 else '!! 应为 1'}"
|
||
)
|
||
print()
|
||
|
||
# ---------------------------------------------------------- 耗时
|
||
scenarios = [
|
||
Scenario("engine.snapshot()", "完整序列化整局（16 地 + 8 牲畜）", engine.snapshot),
|
||
Scenario("state.net_worth", "净资产属性（每次访问都重算）", lambda: engine.state.net_worth),
|
||
Scenario("engine.available_actions()", "GET /api/actions 的主体", engine.available_actions),
|
||
Scenario(
|
||
"brief(session, hub)", "动作回执的精简状态（只读标量）", lambda: brief(session, hub)
|
||
),
|
||
Scenario("session.snapshot()", "full=1 时回执附带的完整快照", session.snapshot),
|
||
Scenario(
|
||
"build_state_message()",
|
||
"SSE 每 250ms 一次的推送",
|
||
lambda: build_state_message(session, 1),
|
||
),
|
||
Scenario("session.summary()", "排行榜单行摘要（只读标量）", session.summary),
|
||
]
|
||
|
||
# 排行榜：N 个牧场，每个都走一遍 summary()
|
||
league_dir = tempfile.mkdtemp(prefix="farm-league-")
|
||
league = SessionManager(save_dir=league_dir, tick_seconds=config.DEFAULT_TICK_SECONDS)
|
||
for i in range(args.farms):
|
||
fill_endgame(league.get(f"farm-{i:03d}", create=True).engine)
|
||
scenarios.append(
|
||
Scenario(f"leaderboard({args.farms} farms)", "GET /api/world 的排行榜", league.leaderboard)
|
||
)
|
||
|
||
print("=" * 74)
|
||
print(f"耗时（每项取 {args.repeat} 轮最快，越大越慢）")
|
||
print("=" * 74)
|
||
print(f" {'场景':<32} {'µs/op':>10} {'ops/s':>10} 说明")
|
||
for scenario in scenarios:
|
||
scenario.run(args.repeat)
|
||
us = scenario.per_op * 1e6
|
||
print(f" {scenario.name:<32} {us:10.1f} {1 / scenario.per_op:10.0f} {scenario.note}")
|
||
|
||
# ---------------------------------------------------------- HTTP 端到端
|
||
print()
|
||
print("=" * 74)
|
||
print("HTTP 端到端（真实路由 + 响应装配；不含 lifespan）")
|
||
print("=" * 74)
|
||
http_cases = [
|
||
("GET /api/actions", "GET", "/api/actions", f"farm={SPEED_FARM}", b""),
|
||
("POST /api/actions full=1", "POST", "/api/actions", query, ACTION_BODY),
|
||
("GET /api/state", "GET", "/api/state", f"farm={SPEED_FARM}", b""),
|
||
("GET /api/world", "GET", "/api/world", f"farm={SPEED_FARM}", b""),
|
||
]
|
||
print(f" {'场景':<32} {'µs/op':>10} {'ops/s':>10}")
|
||
for label, method, path, case_query, body in http_cases:
|
||
per_op = measure_http(app, method, path, case_query, body, repeat=args.repeat)
|
||
print(f" {label:<32} {per_op * 1e6:10.1f} {1 / per_op:10.0f}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|