# -*- coding: utf-8 -*- """M11b-4：端到端时延基准（adjudication → write → COMMIT → broadcast → consumer）。 链路分段（对应任务验收「P99 ≤ 1s」）:: t0 裁决触发（run_chain 进入，模拟裁决服务决定下发事件） t1 write_event_with_state 返回（内部已完成事件+状态单事务 COMMIT，M11b-2） t2 post-commit hook 广播发出完成（M11b-3 install_broadcast_hook 包装后的函数） t3 消费端回调收到广播消息（感知） e2e_latency = t3 - t0 本模块 **只组合调用** M11b-2/M11b-3 已交付的公开 API，不修改其核心逻辑： - 写入：``world_sync.write_event_with_state``（若宿主已 ``load_world_sync()`` 则 该引用已被 ``install_broadcast_hook`` 包成 COMMIT 后自动广播版本）； - 广播：``subscribe_broadcast`` / ``drain_broadcast`` / ``broadcast_stats``。 消费端感知判定：优先「广播回调到达」（真实链路）；若广播丢失，则由 :mod:`world_sync.pbl_runtime_polling` 的 3s 轮询兜底补漏到达（此时 e2e 必然 >1s，报告会把它归入 ``fallback_recovered`` 并单独统计，供瓶颈分析）。 """ import time from typing import Any, Callable, Dict, List, Optional from .pbl_runtime_broadcast import ( broadcast_stats, drain_broadcast, get_broadcaster, subscribe_broadcast, unsubscribe_broadcast, ) from .pbl_runtime_errors import TxConfigError from .pbl_runtime_polling import FALLBACK_SLO_MS, get_fallback from .pbl_runtime_sql import percentile #: 端到端时延 SLO（毫秒）——任务规定 P99 ≤ 1s E2E_LATENCY_SLO_MS = 1000.0 __all__ = [ "E2E_LATENCY_SLO_MS", "FALLBACK_SLO_MS", "measure_once", "run_chain", "summarize_latency", "latency_report", ] def _now_ms() -> float: return time.monotonic() * 1000.0 def measure_once(write_fn: Callable[..., Any], event_data: Dict[str, Any], state_update: Optional[Dict[str, Any]] = None, wait_ms: float = 1000.0, poll_fallback: bool = False) -> Dict[str, Any]: """跑一条事件的端到端测量。 :param write_fn: 写入函数（应为已安装 post-commit hook 的 ``write_event_with_state``；纯写入端时延也能测，广播段记 None）。 :param poll_fallback: True 时广播丢失不判 fail，改由轮询兜底补漏计时。 """ scene_id = str(event_data.get("scene_id") or event_data.get("world_id") or "") fb = get_fallback() holder: Dict[str, Any] = {} t_done = {"v": None} def _on_msg(msg: Any) -> None: if holder.get("t_recv_ms") is None: holder["t_recv_ms"] = _now_ms() holder["msg"] = msg # 广播正常路径：推进兜底水位线（消费端标准接线姿势） try: fb.notify_broadcast(msg) except Exception: pass holder["t_recv_ms"] = None sub = subscribe_broadcast(_on_msg) try: t0 = _now_ms() try: result = write_fn(event_data, state_update or {}) finally: t1 = _now_ms() # post-commit hook 在 write_fn 内部同步/异步发出；drain 确保已派发 try: drain_broadcast(timeout=0.5) except Exception: pass t2 = _now_ms() # 等消费端感知（广播回调） deadline = _now_ms() + wait_ms while holder.get("t_recv_ms") is None and _now_ms() < deadline: time.sleep(0.001) t3 = holder.get("t_recv_ms") via = "broadcast" if t3 is None and poll_fallback: # 广播丢失 → 3s 轮询兜底补漏（真实节奏：interval=3s） eid = None if isinstance(result, dict): eid = result.get("event_id") or result.get("id") w = fb.wait_event(scene_id, str(eid or ""), timeout=FALLBACK_SLO_MS / 1000.0) via = "polling_fallback" t3 = _now_ms() if w.get("found") else None e2e = (t3 - t0) if t3 is not None else None return { "scene_id": scene_id, "event_id": (result.get("event_id") if isinstance(result, dict) else None), "t_write_ms": round(t1 - t0, 3), "t_broadcast_emit_ms": round(t2 - t1, 3), "t_consume_ms": round(t3 - t1, 3) if t3 is not None else None, "e2e_ms": round(e2e, 3) if e2e is not None else None, "via": via, "perceived": t3 is not None, } finally: unsubscribe_broadcast(sub) def run_chain(write_fn: Callable[..., Any], events: List[Dict[str, Any]], state_builder: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None, wait_ms: float = 1000.0, poll_fallback: bool = False) -> List[Dict[str, Any]]: """批量端到端测量（裁决触发→…→消费端感知），返回逐条样本。""" samples: List[Dict[str, Any]] = [] for ev in events: st = state_builder(ev) if state_builder else {} samples.append(measure_once(write_fn, ev, st, wait_ms=wait_ms, poll_fallback=poll_fallback)) return samples def summarize_latency(samples: List[Dict[str, Any]], slo_ms: float = E2E_LATENCY_SLO_MS) -> Dict[str, Any]: """汇总：P50/P90/P99/max + SLO 判定 + 分段瓶颈定位。""" e2e = [s["e2e_ms"] for s in samples if s.get("e2e_ms") is not None] write_seg = [s["t_write_ms"] for s in samples if s.get("t_write_ms") is not None] emit_seg = [s["t_broadcast_emit_ms"] for s in samples if s.get("t_broadcast_emit_ms") is not None] consume_seg = [s["t_consume_ms"] for s in samples if s.get("t_consume_ms") is not None] p99 = round(percentile(e2e, 99), 3) if e2e else None via_broadcast = sum(1 for s in samples if s.get("via") == "broadcast") via_polling = sum(1 for s in samples if s.get("via") == "polling_fallback") lost = sum(1 for s in samples if not s.get("perceived")) # 瓶颈定位：P99 分段最大者 seg_p99 = { "write_commit": round(percentile(write_seg, 99), 3) if write_seg else 0.0, "broadcast_emit": round(percentile(emit_seg, 99), 3) if emit_seg else 0.0, "consume_wait": round(percentile(consume_seg, 99), 3) if consume_seg else 0.0, } bottleneck = max(seg_p99.items(), key=lambda kv: kv[1])[0] if seg_p99 else None return { "n": len(samples), "perceived": len(samples) - lost, "lost": lost, "via_broadcast": via_broadcast, "via_polling_fallback": via_polling, "p50_ms": round(percentile(e2e, 50), 3) if e2e else None, "p90_ms": round(percentile(e2e, 90), 3) if e2e else None, "p99_ms": p99, "max_ms": round(max(e2e), 3) if e2e else None, "segment_p99_ms": seg_p99, "bottleneck_segment": bottleneck, "slo_ms": slo_ms, "slo_passed": (p99 is not None and p99 <= slo_ms), "broadcast_stats": broadcast_stats(), } def latency_report(write_fn: Callable[..., Any], events: List[Dict[str, Any]], state_builder: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None, slo_ms: float = E2E_LATENCY_SLO_MS, wait_ms: float = 1000.0, poll_fallback: bool = False) -> Dict[str, Any]: """一键基准：跑链路 + 汇总 + 结论（供集成测试 / scripts/m11b4_e2e_bench.py 复用）。""" if write_fn is None or not callable(write_fn): raise TxConfigError("latency_report: write_fn must be a callable " "(world_sync.write_event_with_state)") samples = run_chain(write_fn, events, state_builder=state_builder, wait_ms=wait_ms, poll_fallback=poll_fallback) summary = summarize_latency(samples, slo_ms=slo_ms) summary["samples"] = samples summary["verdict"] = ("PASS" if summary["slo_passed"] else "FAIL: P99 %.1fms > SLO %.0fms, bottleneck=%s" % (summary["p99_ms"] or -1, slo_ms, summary["bottleneck_segment"])) return summary