From 587b0b96e9d453358819b5fbc7e7b76ffa59d4be Mon Sep 17 00:00:00 2001 From: Alice Cheng Date: Mon, 20 Jul 2026 11:00:36 -0700 Subject: [PATCH 1/6] docs: reconcile design docs with current source (drift cleanup pass 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First pass of the doc-drift cleanup surfaced by the project review. Applies the mechanical rule-1 (source exists → fix doc) and rule-2 (source gone → remove stale section) fixes; non-trivial rewrites are left for a follow-up. - ENDPOINT_CLIENT.md: socket table now matches http.py (SO_KEEPALIVE 1→0 with the connection_lost/eof_received rationale, TCP_KEEPCNT 1→5 marked inert, SO_RCVBUF/SNDBUF 4MB→128KB sliding-window buffers); is_stale() select()→poll() in §8.3/§8.5; appendix loadgen_cores 2→5. - PERF_ARCHITECTURE.md + CLIENT_PERFORMANCE_TUNING.md: loadgen cores → 5 (DEFAULT_LOADGEN_CORES); diagram marked schematic. - core/DESIGN.md: drop nonexistent StreamChunk.is_complete field. - async_utils/DESIGN.md: EventPublisherService is a plain per-instance class (not a SingletonMixin singleton). - event_logger/DESIGN.md: "Not Yet Wired" → publishing is wired in BenchmarkSession. - ready_check_design.md: receiver does NOT close the socket on timeout. - openai/DESIGN.md: decode_sse_message returns SSEChoice (not str); document completions_adapter.py / OpenAITextCompletionsAdapter. - evaluation/DESIGN.md: Files table adds bfcl_v4_*; Scoring Methods registry adds agentic_inference_inline, vbench, bfcl_v4, legacy_mlperf_deepseek_r1. - dataset_manager/DESIGN.md: preset example gpqa::Qwen/Qwen3-8B → gpqa::gptoss. - load_generator/DESIGN.md: delay.py uniform_delay_fn → make_delay_fn. Co-Authored-By: Claude Opus 4.8 --- docs/CLIENT_PERFORMANCE_TUNING.md | 6 +-- docs/ENDPOINT_CLIENT.md | 54 +++++++++---------- docs/PERF_ARCHITECTURE.md | 10 ++-- docs/async_utils/DESIGN.md | 9 ++-- .../services/event_logger/DESIGN.md | 9 ++-- .../transport/zmq/ready_check_design.md | 2 +- docs/core/DESIGN.md | 1 - docs/dataset_manager/DESIGN.md | 9 ++-- docs/evaluation/DESIGN.md | 31 ++++++----- docs/load_generator/DESIGN.md | 2 +- docs/openai/DESIGN.md | 25 ++++----- 11 files changed, 84 insertions(+), 74 deletions(-) diff --git a/docs/CLIENT_PERFORMANCE_TUNING.md b/docs/CLIENT_PERFORMANCE_TUNING.md index 0c985cee2..2ad5902ef 100644 --- a/docs/CLIENT_PERFORMANCE_TUNING.md +++ b/docs/CLIENT_PERFORMANCE_TUNING.md @@ -32,12 +32,12 @@ enable_cpu_affinity: true # Auto-compute NUMA-aware plan (default) # enable_cpu_affinity: false # Disabled ``` -**Auto mode allocation** (default 6 physical cores for loadgen): +**Auto mode allocation** (default 5 physical cores for loadgen, `DEFAULT_LOADGEN_CORES`): - 1 core: Session thread (scheduler, busy-wait timing) - 1 core: Event loop thread (uvloop, response handling) -- 4 cores: ZMQ I/O threads -- Remaining physical cores: Workers (one per core with all SMT siblings) +- Remaining cores: ZMQ I/O threads (up to 4, sharing the leftover loadgen cores) +- All other physical cores: Workers (one per core with all SMT siblings) ## Platform Notes diff --git a/docs/ENDPOINT_CLIENT.md b/docs/ENDPOINT_CLIENT.md index 8b7982a76..3e8733be9 100644 --- a/docs/ENDPOINT_CLIENT.md +++ b/docs/ENDPOINT_CLIENT.md @@ -790,29 +790,29 @@ Each worker maintains its own `ConnectionPool` to its assigned endpoint. The poo **Idle Connection Validation (`is_stale`):** -When `acquire()` pops a connection from the idle stack, it must verify the server hasn't closed it. `PooledConnection.is_stale()` combines a fast-path skip with a zero-cost kernel probe: +When `acquire()` pops a connection from the idle stack, it must verify the server hasn't closed it. `PooledConnection.is_stale()` combines a fast-path skip with a zero-cost kernel probe. It uses `poll()` rather than `select()` to avoid the `FD_SETSIZE` limit on high fds, and reuses a persistent per-connection poller (registered lazily on first call, since the fd is stable per connection): ```python def is_stale(self) -> bool: - # Fast path: skip select() for recently-used connections. + # Fast path: skip the probe for recently-used connections. # A server won't close a connection within 1s of last use, # so the syscall is unnecessary. Saves ~1µs per acquire. if time.monotonic() - self.last_used < 1.0: return False - # Zero-timeout select(): asks the kernel if any data or - # errors are pending. A healthy idle socket has neither. - readable, _, exceptional = select.select([fd], [], [fd], 0) - return bool(readable or exceptional) + # Zero-timeout poll() on a persistent per-connection poller: + # asks the kernel if any data or errors are pending. A healthy + # idle socket has neither. A raised poll event means the server + # sent FIN (readable EOF) or the socket errored. + return bool(self._stale_poller.poll(0)) ``` -| `select()` Result | Kernel State | Pool Action | -| ------------------ | -------------------------------------------------------- | --------------------------------- | -| `readable=True` | Server sent FIN, indicating it is closing the connection | Discard connection, try next idle | -| `exceptional=True` | Socket error such as a TCP reset from the server | Discard connection, try next idle | -| Both empty | No pending data or errors on the socket | Connection is healthy, use it | +| `poll()` Result | Kernel State | Pool Action | +| --------------- | ---------------------------------------------------- | --------------------------------- | +| Event raised | Server sent FIN (readable EOF) or socket error/reset | Discard connection, try next idle | +| No event | No pending data or errors on the socket | Connection is healthy, use it | -The fast-path skip (`< 1.0s`) avoids the `select()` syscall entirely for connections in active rotation — reducing validation from ~1.2µs to ~161ns (see latencies below). +The fast-path skip (`< 1.0s`) avoids the `poll()` syscall entirely for connections in active rotation — reducing validation from ~1.2µs to ~161ns (see latencies below). **Operation Latencies:** @@ -839,17 +839,17 @@ Per-operation latency measurements for `ConnectionPool` (localhost TCP, uvloop): The `_SocketConfig` class (`http.py`) defines socket options applied to all TCP connections created by the connection pool. These options are tuned for low-latency streaming workloads where individual request latency directly impacts benchmark measurements. -| Option | Value | Effect | Interaction | -| ------------------ | ----- | ----------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | -| `TCP_NODELAY` | 1 | Disables Nagle's algorithm, allowing small packets to be sent immediately rather than being buffered for batching | With `TCP_QUICKACK`: eliminates both send batching (Nagle) and receive-side delayed ACK, removing the two primary sources of TCP-induced latency | -| `TCP_QUICKACK` | 1 | Immediately acknowledge received packets instead of delaying acknowledgments (Linux-specific) | Not sticky — kernel may revert to delayed ACK mode; re-applied per connection. Together with `TCP_NODELAY`, ensures neither side introduces artificial delays | -| `SO_KEEPALIVE` | 1 | Enable TCP keepalive probes at the socket level | Activates the kernel keepalive mechanism; actual probe timing controlled by `TCP_KEEPIDLE`, `TCP_KEEPCNT`, `TCP_KEEPINTVL` | -| `TCP_KEEPIDLE` | 1s | Start probing after 1 second idle (Linux-specific) | With `TCP_KEEPCNT=1` + `TCP_KEEPINTVL=1s`: total dead-connection detection time is 2 seconds (1s idle + 1 probe × 1s interval) | -| `TCP_KEEPCNT` | 1 | 1 failed probe = connection declared dead | Aggressive: default is 9. Appropriate here because a failed probe to a local/VPC endpoint indicates genuine failure, not transient packet loss | -| `TCP_KEEPINTVL` | 1s | 1 second between probes | With only 1 probe needed (`TCP_KEEPCNT=1`), a single missed probe triggers connection close | -| `SO_RCVBUF` | 4 MB | Receive buffer size | 4 MB buffer accommodates approximately 1M tokens (at 4 bytes per token) for offline-mode full responses. Prevents the kernel from dropping data when the application is briefly busy | -| `SO_SNDBUF` | 4 MB | Send buffer size | Sized for large request payloads (long prompts). Allows `write()` to complete without blocking even for large HTTP bodies | -| `TCP_USER_TIMEOUT` | 0 | Disabled — no timeout on unacknowledged sent data (Linux-specific) | Keepalive handles dead-connection detection; setting this to 0 avoids interfering with long-running SSE streams where the server may take seconds between chunks | +| Option | Value | Effect | Interaction | +| ------------------ | ------ | ----------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `TCP_NODELAY` | 1 | Disables Nagle's algorithm, allowing small packets to be sent immediately rather than being buffered for batching | With `TCP_QUICKACK`: eliminates both send batching (Nagle) and receive-side delayed ACK, removing the two primary sources of TCP-induced latency | +| `TCP_QUICKACK` | 1 | Immediately acknowledge received packets instead of delaying acknowledgments (Linux-specific) | Not sticky — kernel may revert to delayed ACK mode; re-applied per connection. Together with `TCP_NODELAY`, ensures neither side introduces artificial delays | +| `SO_KEEPALIVE` | 0 | Disabled — kernel TCP keepalive is turned off | Dead connections are detected in the pool via `connection_lost`/`eof_received`, not kernel keepalive; keepalive probes produced connection-timeout errors in offline and high-concurrency modes. The `TCP_KEEP*` options below are set but inert while this is 0 | +| `TCP_KEEPIDLE` | 1s | Start probing after 1 second idle (Linux-specific) | Inert while `SO_KEEPALIVE=0`; only takes effect if keepalive is re-enabled | +| `TCP_KEEPCNT` | 5 | 5 failed probes = connection declared dead | Inert while `SO_KEEPALIVE=0`; only takes effect if keepalive is re-enabled | +| `TCP_KEEPINTVL` | 1s | 1 second between probes | Inert while `SO_KEEPALIVE=0`; only takes effect if keepalive is re-enabled | +| `SO_RCVBUF` | 128 KB | Receive buffer size | Sliding-window buffer, not a full-message buffer: the event loop reads eagerly, so it only holds data between kernel delivery and application read (~one RTT). Larger responses stream through fine via the TCP sliding window | +| `SO_SNDBUF` | 128 KB | Send buffer size | Sliding-window send buffer; large request bodies stream through without blocking `write()` | +| `TCP_USER_TIMEOUT` | 0 | Disabled — no timeout on unacknowledged sent data (Linux-specific) | Keepalive handles dead-connection detection; setting this to 0 avoids interfering with long-running SSE streams where the server may take seconds between chunks | **Cross-platform compatibility:** Applied via `_SocketConfig.apply(sock)` with `hasattr()` checks for Linux-specific options (`TCP_KEEPIDLE`, `TCP_QUICKACK`, `TCP_USER_TIMEOUT`). On non-Linux platforms, these options are silently skipped — the system runs with reduced tuning but remains functional. @@ -861,9 +861,9 @@ The `_SocketConfig` class (`http.py`) defines socket options applied to all TCP | Idle connection strategy | LIFO stack | Random from idle list | Lower error rate: high load reuses hot connections; low load (long idle) stale ones sink to bottom | | Waiter queue | FIFO via `OrderedDict` | List, deque | Fair scheduling; O(1) insert/remove | | Connection limiting | `max_connections` cap; tracks `len(all_connections) + _creating` | Unlimited | Prevents ephemeral port exhaustion — a real production failure mode at high concurrency (see [§8.6](#86-benchmark-results-vs-aiohttp)). Counting in-progress connections (`_creating`) prevents race where concurrent `acquire()` calls overshoot the limit during TCP handshake | -| Staleness detection | `select()` probe on FD with zero timeout | Timeout-based, `poll()` | Detects server FIN without I/O. `select()` over `poll()` avoids poll object creation overhead | -| Preclose skip | `time.monotonic() - last_used < 1.0` → return not-stale immediately | Always probe | Server unlikely to close within 1s of last use; skips `select()` syscall entirely on hot connections under load | -| Socket tuning | `TCP_NODELAY` + `TCP_QUICKACK` [7], 4MB buffers | Defaults | `TCP_NODELAY` disables Nagle batching; `TCP_QUICKACK` disables delayed ACK; together they eliminate both send and receive latency sources. 4MB buffers sized for offline-mode full responses. See [§8.4](#84-socket-config) for full socket option table | +| Staleness detection | `poll()` on the FD with zero timeout via a persistent per-connection `_stale_poller` | Timeout-based, `select()` | Detects server FIN without I/O. A persistent `poll` object registered once per connection avoids rebuilding the fd set on every probe | +| Preclose skip | `time.monotonic() - last_used < 1.0` → return not-stale immediately | Always probe | Server unlikely to close within 1s of last use; skips the `poll()` syscall entirely on hot connections under load | +| Socket tuning | `TCP_NODELAY` + `TCP_QUICKACK` [7], 128KB buffers | Defaults | `TCP_NODELAY` disables Nagle batching; `TCP_QUICKACK` disables delayed ACK; together they eliminate both send and receive latency sources. 128KB sliding-window buffers (not full-message buffers — the event loop reads eagerly). See [§8.4](#84-socket-config) for full socket option table | | Ephemeral port detection | Read `/proc/sys/net/ipv4/ip_local_port_range` | Manual configuration | Auto-sizes `max_connections` to the port budget (range x distinct endpoints); raises `RuntimeError` if explicit value exceeds it. Live socket occupancy is not subtracted (racy, counts unrelated destinations) | | Connection warmup | Auto: 50% of an explicit pool, 25% of the port budget for the full-auto config; establishment paced by a per-pool connect limiter (`max_concurrent_warmup_connects`, default 128) | 0% or 100%, unbounded `gather` | 100% = SYN flood risk to server; 0% = ~100x cold-start penalty per [§8.3](#83-connection-pool). Pacing bounds in-flight `connect()` for warmup AND runtime pool growth, so bursts can't overflow the server accept queue or exhaust ephemeral ports. `return_exceptions=True` ensures individual failures don't abort the warmup batch | | Idle connection discard | `max_idle_time=4.0s` proactive close | Rely on staleness only | Proactive discard avoids keepalive race with server timeout; 4s chosen to be shorter than typical server keepalive (5-60s) | @@ -1351,7 +1351,7 @@ Per-message round-trip: **4 context switches** (sender↔IO thread↔kernel↔IO | Event loop daemon | 1 | uvloop in `HTTPEndpointClient` | | Transport I/O | `io_threads` (default 4) | Background threads for IPC | -Default `loadgen_cores=2` accommodates the 2 Python threads (Session + Event loop). +Default `loadgen_cores=5` (`DEFAULT_LOADGEN_CORES` in `cpu_affinity.py`) reserves headroom for the loadgen-side threads (Session + event loop) plus the transport I/O threads that share those cores. **Performance ranking sources (checked in order):** diff --git a/docs/PERF_ARCHITECTURE.md b/docs/PERF_ARCHITECTURE.md index c7a31d61a..64efea990 100644 --- a/docs/PERF_ARCHITECTURE.md +++ b/docs/PERF_ARCHITECTURE.md @@ -367,10 +367,12 @@ Workers are mostly I/O-bound (waiting on HTTP responses). └─────────────────────────────────────────────────────────────────────┘ ``` -| Component | Allocation | Rationale | -| ----------- | -------------------------------- | ------------------------------------------------------------------------- | -| **LoadGen** | 2 physical cores (4 logical) | Session thread + event loop thread. Bottleneck - gets fastest cores. | -| **Workers** | 1 physical core each (2 logical) | I/O-bound. Full core isolation prevents context switches, reduces jitter. | +| Component | Allocation | Rationale | +| ----------- | ------------------------------------------ | ---------------------------------------------------------------------------------------------- | +| **LoadGen** | `DEFAULT_LOADGEN_CORES` = 5 physical cores | Session thread + event loop thread + up to 4 ZMQ I/O threads. Bottleneck - gets fastest cores. | +| **Workers** | 1 physical core each (2 logical) | I/O-bound. Full core isolation prevents context switches, reduces jitter. | + +> The core-ordering diagram above is schematic — it shows two `LG` cells for illustration; the actual default reserves 5 physical cores for loadgen (`DEFAULT_LOADGEN_CORES` in `cpu_affinity.py`). ### Why Both Hyperthreads Per Core diff --git a/docs/async_utils/DESIGN.md b/docs/async_utils/DESIGN.md index 8b1da236a..9e1020613 100644 --- a/docs/async_utils/DESIGN.md +++ b/docs/async_utils/DESIGN.md @@ -69,17 +69,18 @@ short-lived coroutines on the hot path. ### `EventPublisherService` -Singleton via `SingletonMixin` — after the first construction, subsequent calls return the -cached instance. The first construction requires a `ManagedZMQContext`. Subscribers receive -`EventRecord` messages over a ZMQ SUB socket. +A plain per-instance class wrapping `ZmqMessagePublisher[EventRecord]` with `LoopManager` +integration and auto-generated socket names. Each construction requires a `ManagedZMQContext`. +Subscribers receive `EventRecord` messages over a ZMQ SUB socket. ```python -class EventPublisherService(SingletonMixin, ZmqEventRecordPublisher): +class EventPublisherService(ZmqMessagePublisher[EventRecord]): def __init__( self, managed_zmq_context: ManagedZMQContext, extra_eager: bool = False, isolated_event_loop: bool = False, + send_threshold: int = 1000, ) -> None def publish(self, record: EventRecord) -> None diff --git a/docs/async_utils/services/event_logger/DESIGN.md b/docs/async_utils/services/event_logger/DESIGN.md index 868b5bbcf..b5da4501a 100644 --- a/docs/async_utils/services/event_logger/DESIGN.md +++ b/docs/async_utils/services/event_logger/DESIGN.md @@ -212,9 +212,8 @@ usage: uv run python -m inference_endpoint.async_utils.services.event_logger | `--socket-name` | Yes | — | Socket name within socket-dir | | `--writers` | No | `jsonl` | Writer backends: `jsonl`, `sql`, or both | -## Not Yet Wired +## Wiring -The EventRecord pub/sub infrastructure is ready, but actual `publish(EventRecord(...))` -calls have not been connected in the load generator or worker processes. Once wired, -the event logger will receive and persist all session/sample/error events published -during a benchmark run. +`publish(EventRecord(...))` calls are connected in the load generator (`BenchmarkSession` +publishes STARTED / PROMPT / COMPLETE / ERROR events via its `EventPublisher`), so the event +logger receives and persists all session/sample/error events published during a benchmark run. diff --git a/docs/async_utils/transport/zmq/ready_check_design.md b/docs/async_utils/transport/zmq/ready_check_design.md index 15e979689..1af35ddd8 100644 --- a/docs/async_utils/transport/zmq/ready_check_design.md +++ b/docs/async_utils/transport/zmq/ready_check_design.md @@ -40,7 +40,7 @@ This means one receiver socket handles readiness from all subprocesses. - Binds a ZMQ PULL socket on an IPC path - `wait(timeout)` blocks until `count` signals arrive - Returns list of identities in arrival order -- Closes socket after all signals received (or on timeout) +- Closes the socket after all signals are received, but deliberately **not** on timeout (the caller may retry) - Timeout is a total deadline, not per-message ### `send_ready_signal()` (subprocess side) diff --git a/docs/core/DESIGN.md b/docs/core/DESIGN.md index 99062ab99..17a15362b 100644 --- a/docs/core/DESIGN.md +++ b/docs/core/DESIGN.md @@ -52,7 +52,6 @@ Represents one SSE delta from a streaming response. | ---------------- | ---------------- | ------------------------------ | | `id` | `str` | Matches originating `Query.id` | | `response_chunk` | `str` | Incremental token text | -| `is_complete` | `bool` | True for the final chunk | | `metadata` | `dict[str, Any]` | Per-chunk metadata | ### `TextModelOutput` diff --git a/docs/dataset_manager/DESIGN.md b/docs/dataset_manager/DESIGN.md index e1c8c6d1a..a8b083db5 100644 --- a/docs/dataset_manager/DESIGN.md +++ b/docs/dataset_manager/DESIGN.md @@ -77,8 +77,9 @@ parser/remap config, and dataset name. Format is inferred from file extension wh - `.parquet` → `PARQUET` - explicit `format=huggingface` → `HF` -Presets (e.g. `"gpqa::Qwen/Qwen3-8B"`) are encoded in `config.name` as a `"::"` split — the -factory resolves them to a predefined dataset class with a model-specific transform stack. +Presets (e.g. `"gpqa::gptoss"`) are encoded in `config.name` as a `"::"` split — `::` — +where the factory resolves the first segment to a predefined dataset class and the second to a named +preset with its transform stack. ### `Transform` (abstract base) @@ -116,8 +117,8 @@ configs. Each predefined dataset ships with default transforms for supported mod ## Preset System -A preset string like `"gpqa::Qwen/Qwen3-8B"` resolves to a predefined dataset with a -model-specific transform stack pre-applied. This is used by rulesets to ensure consistent +A preset string like `"gpqa::gptoss"` (`::`) resolves to a predefined dataset with a +named preset's transform stack pre-applied. This is used by rulesets to ensure consistent prompt formatting across submissions. ## Design Decisions diff --git a/docs/evaluation/DESIGN.md b/docs/evaluation/DESIGN.md index 2ec023884..1074c343d 100644 --- a/docs/evaluation/DESIGN.md +++ b/docs/evaluation/DESIGN.md @@ -35,11 +35,12 @@ accuracy summary written into benchmark results ## Files -| File | Purpose | -| ---------------- | ----------------------------------------------------------------- | -| `extractor.py` | Extracts model answer from raw text (regex, boxed-answer parsing) | -| `scoring.py` | Compares extracted answer to ground truth label | -| `livecodebench/` | LiveCodeBench-specific code execution pipeline | +| File | Purpose | +| ---------------- | -------------------------------------------------------------------------------------------------------------------------------------------- | +| `extractor.py` | Extracts model answer from raw text (regex, boxed-answer parsing) | +| `scoring.py` | Compares extracted answer to ground truth label | +| `bfcl_v4_*.py` | BFCL v4 function-calling eval: `bfcl_v4_execution.py`, `bfcl_v4_scorer.py`, `bfcl_v4_metrics.py`, and the multi-turn `runner`/`scorer`/`cli` | +| `livecodebench/` | LiveCodeBench-specific code execution pipeline | ## LiveCodeBench @@ -60,13 +61,19 @@ Files: The scorer registry in `evaluation/scoring.py` currently includes: -| Method | Description | -| --------------------- | -------------------------------------------------------------- | -| `pass_at_1` | Exact-match style scoring; also used by the LiveCodeBench path | -| `string_match` | Whitespace-trimmed string equality | -| `rouge` | ROUGE-based text generation scoring | -| `code_bench_scorer` | LiveCodeBench code-execution scoring | -| `shopify_category_f1` | Shopify category F1 evaluation | +| Method | Description | +| --------------------------- | -------------------------------------------------------------- | +| `pass_at_1` | Exact-match style scoring; also used by the LiveCodeBench path | +| `string_match` | Whitespace-trimmed string equality | +| `rouge` | ROUGE-based text generation scoring | +| `code_bench_scorer` | LiveCodeBench code-execution scoring | +| `shopify_category_f1` | Shopify category F1 evaluation | +| `agentic_inference_inline` | Inline scoring for agentic multi-turn inference | +| `vbench` | VBench video-generation accuracy (WAN 2.2 T2V) | +| `bfcl_v4` | BFCL v4 function-calling accuracy | +| `legacy_mlperf_deepseek_r1` | MLPerf DeepSeek-R1 combined multi-subset accuracy | + +The registry (`Scorer.PREDEFINED`) is auto-populated from `Scorer` subclasses via `__init_subclass__`. The scoring configuration used by benchmark execution is specified per accuracy dataset under `datasets[].accuracy_config`, including `accuracy_config.eval_method`, diff --git a/docs/load_generator/DESIGN.md b/docs/load_generator/DESIGN.md index 1c0fa4435..b717545ef 100644 --- a/docs/load_generator/DESIGN.md +++ b/docs/load_generator/DESIGN.md @@ -16,7 +16,7 @@ src/inference_endpoint/load_generator/ │ # BurstStrategy, ConcurrencyStrategy, │ # create_load_strategy() ├── sample_order.py # SampleOrder, WithoutReplacement, WithReplacement -└── delay.py # poisson_delay_fn, uniform_delay_fn +└── delay.py # poisson_delay_fn, make_delay_fn ``` ## Architecture diff --git a/docs/openai/DESIGN.md b/docs/openai/DESIGN.md index 880ae7b75..a52c0bc7f 100644 --- a/docs/openai/DESIGN.md +++ b/docs/openai/DESIGN.md @@ -61,13 +61,13 @@ class HttpRequestAdapter(ABC): def decode_response(cls, response_bytes: bytes, query_id: str) -> QueryResult: ... @classmethod - def decode_sse_message(cls, json_bytes: bytes) -> str: ... + def decode_sse_message(cls, json_bytes: bytes) -> SSEChoice: ... ``` `dataset_transforms()` returns adapter-specific transforms that shape dataset rows into the expected `Query.data` schema. `encode_query()` serialises a `Query` to HTTP request bytes. -`decode_response()` parses a non-streaming response. `decode_sse_message()` extracts the content -string from a single SSE JSON payload; `parse_sse_chunk()` (concrete, on the base class) iterates +`decode_response()` parses a non-streaming response. `decode_sse_message()` decodes a single SSE +JSON payload into an `SSEChoice`; `parse_sse_chunk()` (concrete, on the base class) iterates the SSE buffer and calls it repeatedly. ### `SSEAccumulatorProtocol` (protocol, defined in `endpoint_client/accumulator_protocol.py`) @@ -87,15 +87,16 @@ than shared across a connection. ## Key Files -| File | Purpose | -| --------------------------- | ------------------------------------------------------------ | -| `openai_msgspec_adapter.py` | Hot-path adapter; uses msgspec for request encoding | -| `openai_adapter.py` | Standard adapter; uses stdlib json | -| `accumulator.py` | Per-request streaming accumulator for OpenAI SSE deltas | -| `types.py` | Python type annotations for OpenAI response objects | -| `openai_types_gen.py` | Auto-generated from `openapi.yaml`; do not edit manually | -| `harmony.py` | Optional `openai-harmony` integration for compatibility shim | -| `openapi.yaml` | OpenAI API spec snapshot; excluded from pre-commit | +| File | Purpose | +| --------------------------- | ----------------------------------------------------------------------------------------------------------------------------- | +| `openai_msgspec_adapter.py` | Hot-path chat adapter; uses msgspec for request encoding | +| `openai_adapter.py` | Standard chat adapter; uses stdlib json | +| `completions_adapter.py` | Pre-tokenized `/v1/completions` adapter (`OpenAITextCompletionsAdapter`) — sends token IDs, bypasses the server chat template | +| `accumulator.py` | Per-request streaming accumulator for OpenAI SSE deltas | +| `types.py` | Python type annotations for OpenAI response objects | +| `openai_types_gen.py` | Auto-generated from `openapi.yaml`; do not edit manually | +| `harmony.py` | Optional `openai-harmony` integration for compatibility shim | +| `openapi.yaml` | OpenAI API spec snapshot; excluded from pre-commit | ## Design Decisions From 8b54164ee9e626a5cb54106f0bbf23015356ef23 Mon Sep 17 00:00:00 2001 From: Alice Cheng Date: Mon, 20 Jul 2026 11:12:04 -0700 Subject: [PATCH 2/6] docs(metrics): rewrite design docs for the aggregator snapshot pipeline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The metrics design docs described a defunct SQLite EventRecorder + duckdb MetricsReporter write/read architecture (and a later KVStore BasicKVStoreReader build_report variant) that no longer exists in the tree. Rewrite both to the current design: - metrics/DESIGN.md: metrics/ is the reporting side only — it consumes the MetricsSnapshot produced by the MetricsAggregatorService subprocess (events PUB → aggregator → final_snapshot.json) and builds the Report via Report.from_snapshot(dict). Documents the data flow, files (report/metric/ early_stopping/results_plots), the Report struct + from_snapshot QPS/TPS window selection, and integration points; cross-links the aggregator spec. - metrics/report_design.md: builder-level reference for report.py — from_snapshot's dict contract, _series_to_metric_dict rollup shape, the complete flag, early-stopping placement, and QPS/TPS window selection. No remaining references to EventRecorder/MetricsReporter/KVStore/SQLite/duckdb. Co-Authored-By: Claude Opus 4.8 --- docs/metrics/DESIGN.md | 259 ++++++++++++++++------------------ docs/metrics/report_design.md | 134 +++++++++--------- 2 files changed, 183 insertions(+), 210 deletions(-) diff --git a/docs/metrics/DESIGN.md b/docs/metrics/DESIGN.md index 94b880bee..37593c601 100644 --- a/docs/metrics/DESIGN.md +++ b/docs/metrics/DESIGN.md @@ -1,6 +1,6 @@ # Metrics — Design Spec -> Records per-sample timing events to SQLite during a run (write path), then aggregates them into QPS, latency percentiles, TTFT, and TPOT after the run (read path). +> Builds the final benchmark `Report` from the metrics-aggregator's rollup snapshot and renders it (console + `result_summary.json`). Also holds the metric target types, the MLPerf early-stopping percentile math, and the standardized run-artifact plots. **Component specs:** [async_utils](../async_utils/DESIGN.md) · [commands](../commands/DESIGN.md) · [config](../config/DESIGN.md) · [core](../core/DESIGN.md) · [dataset_manager](../dataset_manager/DESIGN.md) · [endpoint_client](../endpoint_client/DESIGN.md) · [evaluation](../evaluation/DESIGN.md) · [load_generator](../load_generator/DESIGN.md) · **metrics** · [openai](../openai/DESIGN.md) · [plugins](../plugins/DESIGN.md) · [profiling](../profiling/DESIGN.md) · [sglang](../sglang/DESIGN.md) · [testing](../testing/DESIGN.md) · [utils](../utils/DESIGN.md) @@ -8,168 +8,149 @@ ## Overview -`metrics/` records benchmark events during a run and aggregates them into performance metrics -afterwards. It is split into two parts with a clean boundary: `EventRecorder` writes; nothing -else does. `MetricsReporter` reads. +`metrics/` is the **reporting** side of the pipeline: it consumes the rollup snapshot produced by +the metrics-aggregator subprocess and turns it into a human-readable console report and a +machine-readable `result_summary.json`. + +It does **not** record or aggregate events itself. During a run, events are published over ZMQ by +the load generator and folded into per-sample rollups by a separate +[`MetricsAggregatorService`](../async_utils/services/metrics_aggregator/DESIGN.md) subprocess, +which publishes `MetricsSnapshot` messages and atomically writes the terminal snapshot to +`final_snapshot.json`. `metrics/` reads that snapshot and builds the `Report`. ## Responsibilities -- Persist every timing event to SQLite during the run (write path) -- Aggregate events into QPS, latency percentiles, TTFT, and TPOT after the run (read path) -- Validate results against metric targets from the active ruleset -- Produce human-readable console output and machine-readable JSON reports +- Turn a metrics-aggregator snapshot dict into a `Report` (`report.py`, `Report.from_snapshot`) +- Derive headline QPS/TPS once, so the serialized report is self-complete +- Render the report to console (`display`) and to `result_summary.json` (`to_json`) +- Define the metric **target** types used by rulesets (`metric.py`) +- Compute MLPerf LoadGen early-stopping percentile estimates (`early_stopping.py`) +- Produce standardized run-artifact plots (`results_plots.py`) -## Component Map +## Data Flow ``` -SampleEventHandler ──► EventRecorder (SQLite, queue-backed) - │ - ▼ - MetricsReporter - │ - ┌───────────┴──────────┐ - ▼ ▼ - console output JSON report +load_generator ──► EventPublisher (events PUB) + │ + ▼ + MetricsAggregatorService (subprocess) + folds events → MetricsRegistry + publishes MetricsSnapshot (metrics PUB) + atomically writes final_snapshot.json ◄── primary Report source + │ + ▼ + main process (commands/benchmark/execute.py) + json.loads(final_snapshot.json) → dict + │ + ▼ + Report.from_snapshot(dict, run_config=...) + │ + ┌───────────┴───────────┐ + ▼ ▼ + Report.display(fn) Report.to_json(result_summary.json) ``` -## Public Interface - -### `EventRecorder` +The snapshot is consumed as a **dict** (`json.loads`), not decoded back into the wire +`MetricsSnapshot` Struct — the wire form is `array_like=True` for compact msgpack, and the dict +form (`snapshot_to_dict`) is the canonical consumer contract. If `final_snapshot.json` is missing +(e.g. the aggregator was SIGKILLed before its signal handler ran), the caller falls back to the +subscriber's last live snapshot and the resulting `Report.complete` is `False`. -Only one `EventRecorder` may be actively writing at a time per process. The live instance is -accessible via the class variable `EventRecorder.LIVE`. +## Files -```python -class EventRecorder: - LIVE: "EventRecorder | None" # class variable; set on construction, cleared on close - - @classmethod - def record_event( - cls, - ev_type: Event, - timestamp_ns: int, - sample_uuid: str = "", - force_commit: bool = False, - assert_active: bool = True, - data: Any = None, - ) -> bool - - def wait_for_writes(self, force_commit: bool = True) -> None - # Blocks until the background writer thread has flushed all queued events - - @staticmethod - def db_path(session_id: str) -> Path -``` +| File | Purpose | +| ------------------- | ------------------------------------------------------------------------------------------------------------------ | +| `report.py` | `Report` Struct + `Report.from_snapshot(dict)`; console `display()` and `to_json()` serializer | +| `metric.py` | Metric target types (`Throughput`, `QueryLatency`, `TTFT`, `TPOT`) used by rulesets | +| `early_stopping.py` | MLPerf LoadGen early-stopping percentile estimates (pure math); see [docs/early_stopping.md](../early_stopping.md) | +| `results_plots.py` | Standardized run-artifact plots (matplotlib-guarded); CLI: `scripts/plot_results.py` | -`record_event()` is a **classmethod** and is **non-blocking**: events are placed on a queue and -written by a background thread. Returns `True` if recorded, `False` if no recorder is active -(when `assert_active=False`). - -### `MetricsReporter` - -```python -class MetricsReporter: - def __init__( - self, - connection_name: os.PathLike, - client_type: str = "duckdb", - ) -> None - - def create_report( - self, - tokenizer: Tokenizer | None = None, - tpot_reporting_mode: TPOTReportingMode = TPOTReportingMode.REQUEST_WEIGHTED, - ) -> Report - - def dump_to_json(self, json_path: Path) -> None -``` +## Public Interface -`create_report()` executes SQL aggregation over the events database and returns a `Report` -object. The optional `tokenizer` enables output sequence length (OSL) computation. `Report` -itself exposes `display(fn=print, summary_only=False)` for console output. +### `Report` (`report.py`) -### Metric Types (`metric.py`) +`Report` is a frozen `msgspec.Struct`. It is built by `from_snapshot` and is the single object that +`display()` and `to_json()` render. ```python -class Throughput(Metric): - REL_TOL = 0.1 # ±10% relative tolerance - def __init__(self, target_qps: float): ... # stored as self.target - -class QueryLatency(Metric): - REL_TOL = 0.1 - def __init__(self, target_latency_ms: float | None = None, - target_qps: float | None = None): ... - -class TTFT(Metric): - def __init__(self, max_ttft_latency_ms: float): ... # hard ceiling +class Report(msgspec.Struct, frozen=True): + version: str + git_sha: str | None + test_started_at: int + n_samples_issued: int + n_samples_completed: int + n_samples_failed: int + duration_ns: int | None + state: str # terminal SessionState string ("complete" / "interrupted") + complete: bool # True iff state == "complete" AND n_pending_tasks == 0 + ttft: dict # per-metric rollup dicts (min/max/mean/percentiles/...) + tpot: dict + latency: dict + output_sequence_lengths: dict + legacy_loadgen_window_duration_ns: int | None = None + qps: float | None = None + tps: float | None = None + finish_reason_counts: dict[str, int] = {} + run_config: dict | None = None # config (load pattern, warmup, RNG seeds) + accuracy: list[dict] = [] # per-dataset accuracy, attached post-scoring -class TPOT(Metric): - def __init__(self, max_tpot_latency_ms: float): ... # hard ceiling -``` - -Each metric exposes `is_valid(measurement) -> bool`. The target value is stored as -`self.target` on the base `Metric` class. - -## Data Flow - -### Write Path (during run) + @classmethod + def from_snapshot( + cls, snap: dict, *, run_config: dict | None = None, + use_legacy_loadgen_qps_metrics: bool = True, + ) -> "Report": ... -``` -SampleEventHandler.query_result_complete(result) - → EventRecorder.record_event( - SampleEvent.COMPLETE, - time.monotonic_ns(), - sample_uuid=result.id, - data={...}, - ) - → queue.put(EventRow(...)) # non-blocking - → background thread: INSERT INTO events + def to_json(self, save_to: os.PathLike | None = None) -> bytes: ... + def display(self, ...) -> None: ... ``` -### Read Path (after run) - -``` -MetricsReporter.create_report() - → SELECT / GROUP BY on events table (DuckDB) - → compute percentiles (p50, p90, p99, p999) - → compute TTFT = time from LOADGEN_ISSUE_CALLED to FIRST_CHUNK - → compute TPOT = (COMPLETE.ts - FIRST_CHUNK.ts) / output_tokens - → compute tracked duration from TEST_STARTED / STOP_PERFORMANCE_TRACKING windows - → compute QPS = tracked completed samples / tracked duration - → validate each metric against RuntimeSettings.reported_metrics -``` +**`from_snapshot`** reads counters and per-series rollups out of the snapshot dict, all via +`.get(...)` with safe defaults so a partial/malformed snapshot yields an honest _incomplete_ +report instead of crashing (missing `state` defaults to `"interrupted"`). It derives headline +throughput once: + +- The snapshot always carries **both** `tracked_duration_ns` (endpoints-native full-run window) + and `legacy_loadgen_window_duration_ns` (MLPerf LoadGen "completed" window, poisson only). It is + therefore config-agnostic and reinterpretable either way. +- Which window drives the headline QPS/TPS is decided by the run config + (`use_legacy_loadgen_qps_metrics`), not by the snapshot. The legacy window is used only when it + is enabled, available, and there are ≥2 completions (`QPS = (completed-1)/window`); otherwise + both QPS and TPS fall back to the native window so they always share one window, and + `legacy_loadgen_window_duration_ns` is left `None` so the serialized report honestly records + which view it holds. +- `tps` is additionally `None` when no OSL was recorded (non-streaming, or tokenizer unavailable). + +**Accuracy** is not part of the metrics snapshot; `from_snapshot` leaves `accuracy` empty and the +finalizer attaches per-dataset accuracy entries after scoring. `run_config` is supplied by the +caller (it is config, not a measured metric). + +### Metric target types (`metric.py`) + +`Throughput`, `QueryLatency`, `TTFT`, and `TPOT` are the metric **targets** a ruleset validates a +run against — not the measured rollups (those live in the snapshot / `Report`). Each stores its +target on the base `Metric` and exposes `is_valid(measurement) -> bool`; `Throughput`/`QueryLatency` +apply a relative tolerance, while `TTFT`/`TPOT` are hard latency ceilings. ## Design Decisions -**SQLite as the event store** - -SQLite gives durable, queryable storage with no external dependencies. The write path uses a -single background writer thread (SQLite's WAL mode is single-writer) to avoid contention. -Aggregation uses DuckDB for columnar SQL performance over the file written by SQLite. - -**Queue-backed writes to decouple hot path** - -The `record_event()` call must not block the load generator thread. Events are placed on a -`queue.Queue` and consumed by a dedicated writer thread. The queue is unbounded; back-pressure -is not a concern because write throughput (SQLite) exceeds event rate in all tested scenarios. - -**Singleton enforcement** - -Only one `EventRecorder` may exist per process. The singleton is enforced at construction time -with a class-level flag. This prevents double-counting if code accidentally constructs a second -recorder. +**Reporting is decoupled from aggregation.** Event recording, tokenization, and rollup live in the +aggregator subprocess; `metrics/` only consumes the finished snapshot. This keeps the hot path (the +aggregator) isolated from report formatting and lets the report be rebuilt offline from a persisted +`final_snapshot.json`. -**TPOT calculation modes** +**The dict snapshot is the contract.** `from_snapshot` takes the `snapshot_to_dict` shape and never +decodes the wire Struct — see the data-flow note above. `result_summary.json` is self-complete +(carries derived QPS/TPS and `run_config`) so a valid run is fully described by its own artifact. -TPOT can be weighted by request (each request contributes equally) or by output token count -(each token contributes equally). The default is request-weighted. The `TPOTReportingMode` enum -controls this at report time without re-running the benchmark. +**Honest incompleteness over crashes.** Every snapshot read defaults safely; drain-timeout and +interrupted runs produce a `Report` with `complete=False` and an explicit indicator in `display()` +rather than a partial success that looks clean. ## Integration Points -| Component | Role | -| ---------------------------- | --------------------------------------------------------- | -| `load_generator/sample.py` | Calls `record_event()` for every state transition | -| `load_generator/session.py` | Calls `create_report()` at run end; saves output | -| `config/runtime_settings.py` | `reported_metrics` list drives which metrics are computed | -| `config/ruleset_base.py` | Provides `Metric` targets for validation | +| Component | Role | +| ---------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------- | +| [`metrics_aggregator`](../async_utils/services/metrics_aggregator/DESIGN.md) | Produces the `MetricsSnapshot` / `final_snapshot.json` that `from_snapshot` consumes | +| `commands/benchmark/execute.py` | Launches the aggregator subprocess, reads `final_snapshot.json`, builds/attaches the `Report`, attaches accuracy | +| `config/` (`use_legacy_loadgen_qps_metrics`) | Selects which duration window drives headline QPS/TPS | +| `scripts/plot_results.py` | CLI entry point over `results_plots.py` | diff --git a/docs/metrics/report_design.md b/docs/metrics/report_design.md index 7f663c1a8..42142dce2 100644 --- a/docs/metrics/report_design.md +++ b/docs/metrics/report_design.md @@ -1,106 +1,98 @@ # Report Design +Builder-level reference for `metrics/report.py`. For where the report sits in the run pipeline, +see [metrics/DESIGN.md](DESIGN.md); for the snapshot it consumes, see the +[metrics aggregator spec](../async_utils/services/metrics_aggregator/DESIGN.md). + ## Overview -The report module provides benchmark result summarization, display, and -serialization. It reads from the KVStore (via `BasicKVStoreReader`) and -produces a `Report` with rollup statistics, percentiles, and histograms. +`report.py` provides benchmark result summarization, display, and serialization. Its input is the +terminal metrics-aggregator snapshot in **dict** form (`snapshot_to_dict`, as persisted to +`final_snapshot.json`); its output is a `Report` carrying rollup statistics, percentiles, and +histograms, rendered to console and to `result_summary.json`. ## Architecture ``` -BasicKVStoreReader.snapshot() +final_snapshot.json (dict form of MetricsSnapshot) │ - ▼ - build_report(reader) + ▼ json.loads(...) → dict + Report.from_snapshot(snap, run_config=..., use_legacy_loadgen_qps_metrics=...) + │ + ├── counters → n_samples_issued/completed/failed, tracked_duration_ns, + │ legacy_loadgen_window_duration_ns, finish-reason counts │ - ├── counters → n_issued, n_completed, n_failed, duration_ns + ├── for each series (ttft, tpot, latency, osl): + │ _series_to_metric_dict(stat) → rollup dict │ - └── for each series metric: - SeriesStats.values → compute_summary() → dict + └── derive qps / tps once (window chosen by run config) │ ▼ - Report - ├── .display(fn) → human-readable output - ├── .to_json(path) → JSON serialization - ├── .qps → computed from n_completed / duration - └── .tps → computed from osl total / duration + Report (frozen msgspec.Struct) + ├── .display(fn) → human-readable output with histograms + └── .to_json(path) → result_summary.json (QPS/TPS + run_config included) ``` ## Design Principles -**No SQL, no UUID tracking, no deduplication.** - -The old `MetricsReporter` queried SQLite via duckdb and built `RollupQueryTable` -objects with UUID-indexed rows, repeat counts, and numpy sorted arrays. None of -this complexity is needed when the input is a `list[float]` from the KVStore. - -The entire rollup is a single function: `compute_summary(values) → dict`. -It calls numpy for percentiles and histograms. No classes, no state. - -**Reports are reproducible from the event log.** +**The dict snapshot is the sole input.** `from_snapshot` consumes the `snapshot_to_dict` shape +directly and never decodes the wire `MetricsSnapshot` Struct (which is `array_like=True` for compact +msgpack). A consumer feeds `json.loads(path.read_bytes())` straight in. This lets a report be +rebuilt offline from any persisted `final_snapshot.json`. -The KVStore is lossy aggregation — it stores per-metric series, not per-sample -provenance. The authoritative record of what happened during a run is the event -log written by the `EventLoggerService`. Every number in a `Report` can be -recomputed by replaying the event log through the same aggregator logic: if a -production report shows a TTFT spike, the event log is the ground truth a user -can mine to attribute the spike to specific samples or time windows. +**Honest incompleteness over crashes.** Every counter and series read uses `.get(...)` with a safe +default, so a truncated or partial (INTERRUPTED) snapshot produces an honest _empty_ rollup and a +`complete=False` report rather than raising. Missing `state` defaults to `"interrupted"`. -New metrics must preserve this property: the aggregator may only derive values -from event fields, never from out-of-band state. If a metric cannot be rebuilt -from the event log alone, it does not belong in the KVStore. +**Self-describing artifacts.** Derived QPS/TPS, the window that produced them +(`legacy_loadgen_window_duration_ns`), and `run_config` are all serialized, so a valid run is fully +identified by its own `result_summary.json`. ## Components -### `compute_summary(values, percentiles, n_histogram_buckets) → dict` +### `_series_to_metric_dict(stat) → dict` -Takes a `list[float]`, returns a dict with: +Converts one series-stat dict from the snapshot into the rollup shape `display()` expects: - `total`, `min`, `max`, `avg`, `std_dev`, `median` -- `percentiles`: dict of `{str(p): float}` for each requested percentile +- `percentiles`: `{str(p): float}` for each requested percentile - `histogram`: `{"buckets": [(lo, hi), ...], "counts": [int, ...]}` +- `early_stopping_percentiles` (optional): MLPerf early-stopping estimate map, placed right after + `percentiles`; present only when early stopping is enabled (COMPLETE snapshots for ttft/tpot/latency) -Empty input returns zeros with empty histogram/percentiles. - -### `Report` (frozen dataclass) - -Fields: - -- `version`, `git_sha`, `test_started_at` -- `n_samples_issued`, `n_samples_completed`, `n_samples_failed` -- `duration_ns` -- `ttft`, `tpot`, `latency`, `output_sequence_lengths` — each a summary dict - -Properties: - -- `qps`: `n_samples_completed / (duration_ns / 1e9)`, or None -- `tps`: `osl_total / (duration_ns / 1e9)`, or None - -Methods: +`avg`/`std_dev`/`median` are derived from the cheap rollups + percentiles (integer-exact variance +form for ns series). `median` falls back to `(min + max) / 2` only for hand-crafted dicts that omit +p50. A zero-count series returns `{}` (or an all-null early-stopping map if the feature is enabled). -- `display(fn, summary_only, newline)` — human-readable output with histograms -- `to_json(save_to)` — JSON serialization with QPS/TPS included +### `Report` (frozen `msgspec.Struct`) -### `build_report(reader) → Report` +Fields: `version`, `git_sha`, `test_started_at`, `n_samples_issued/completed/failed`, +`duration_ns`, `state`, `complete`, the four rollup dicts (`ttft`, `tpot`, `latency`, +`output_sequence_lengths`), `legacy_loadgen_window_duration_ns`, `qps`, `tps`, +`finish_reason_counts`, `run_config`, and `accuracy`. -Reads a `BasicKVStoreReader` snapshot and constructs a `Report`. Works -identically for live metrics (mid-test) and final reports (post-drain) — -the caller decides when to call it. +- `complete` = `state == "complete" and n_pending_tasks == 0` — `False` marks partial async metrics + (drain timeout, interrupt, or a live-tick fallback when no final snapshot was found). +- `qps`/`tps` are computed once in `from_snapshot` (see below) rather than as live properties, so + the serialized report is self-complete. +- `accuracy` is empty from `from_snapshot`; per-dataset entries are attached after scoring. -Counter keys read: `n_samples_issued`, `n_samples_completed`, -`n_samples_failed`, `duration_ns`, `test_started_at`. +Methods: `display(fn, ...)` for console output with histograms; `to_json(save_to)` for +`result_summary.json`. -Series keys summarized: `ttft_ns`, `tpot_ns`, `sample_latency_ns`, `osl`. +### `Report.from_snapshot(snap, *, run_config=None, use_legacy_loadgen_qps_metrics=True) → Report` -## What Was Removed +Splits the snapshot's `metrics` list into counters and series, then builds the `Report`. -From the old `metrics/reporter.py`: +Counter keys read: `tracked_samples_issued`, `tracked_samples_completed`, +`tracked_samples_failed`, `tracked_duration_ns`, `legacy_loadgen_window_duration_ns`, and the +`tracked_finish_reason_*` counters. Series summarized: `ttft`, `tpot`, `latency`, `osl`. -- `MetricsReporter` — SQLite/duckdb query engine (replaced by KVStore) -- `RollupQueryTable` — UUID-indexed rollup table (replaced by `compute_summary`) -- `MetricRow` — per-row accessor (not needed) -- `TPOTReportingMode` — niche enum (can be re-added if needed) -- `SampleUUIDNotFoundError` — UUIDs not relevant in KVStore -- `output_sequence_from_data` — SQL event data parser (not needed) -- `dump_to_json` / `dump_all_to_csv` — event log export (handled by EventLoggerService) +**QPS/TPS window selection.** The snapshot always carries both a native window +(`tracked_duration_ns`) and the MLPerf LoadGen "completed" window +(`legacy_loadgen_window_duration_ns`, poisson only), so it stays reinterpretable either way. The +legacy window drives the headline QPS/TPS only when it is enabled +(`use_legacy_loadgen_qps_metrics`), available, and there are ≥2 completions +(`QPS = (completed - 1) / window`). Otherwise both QPS and TPS fall back to the native window so +they always share one window, and `legacy_loadgen_window_duration_ns` is left `None` so the +serialized report records which view it holds. `tps` is `None` when no OSL was recorded. From d28f9d5b99c8f19eb289396c5f2cd9b4ab14c09b Mon Sep 17 00:00:00 2001 From: Alice Cheng Date: Mon, 20 Jul 2026 11:20:47 -0700 Subject: [PATCH 3/6] docs(load_generator): document agentic support, per-phase drain, snapshot metrics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fills the three non-trivial gaps in the load-generator design doc flagged by the review: - Agentic benchmarking: add AgenticInferenceStrategy + ConversationManager to the file tree, load-strategy table, and a new component section (response-driven next-turn issuing, concurrency measured in active conversations, turn state via ConversationState, AgenticInferenceConfig knobs, AgenticInferenceDataset shape). Note create_load_strategy raises for AGENTIC_INFERENCE (wired separately). - Drain semantics: reframe draining as the per-phase PhaseConfig.drain_after flag (default True; warmup sets False) rather than a property intrinsic to PhaseType. Add drain_after/drain_timeout/strategy to the PhaseConfig sample. - Metrics integration: replace the removed KVStore/KVStoreReader/from_kv_reader path everywhere (lifecycle, sequence + topology diagrams, integration section, TUI-future section) with the current flow — EventPublisherService → MetricsAggregatorService → MetricsSnapshot PUB + final_snapshot.json → Report.from_snapshot. Fix the "COMPLETE is always published" stale-completions snippet (COMPLETE is suppressed for WARMUP; ERROR precedes COMPLETE). Co-Authored-By: Claude Opus 4.8 --- docs/load_generator/DESIGN.md | 206 +++++++++++++++++++++++----------- 1 file changed, 139 insertions(+), 67 deletions(-) diff --git a/docs/load_generator/DESIGN.md b/docs/load_generator/DESIGN.md index b717545ef..82ad211fb 100644 --- a/docs/load_generator/DESIGN.md +++ b/docs/load_generator/DESIGN.md @@ -11,12 +11,14 @@ single-thread, single-event-loop-per-process constraint. ``` src/inference_endpoint/load_generator/ ├── __init__.py # Public exports -├── session.py # BenchmarkSession, SessionResult -├── strategy.py # LoadStrategy protocol, TimedIssueStrategy, -│ # BurstStrategy, ConcurrencyStrategy, -│ # create_load_strategy() -├── sample_order.py # SampleOrder, WithoutReplacement, WithReplacement -└── delay.py # poisson_delay_fn, make_delay_fn +├── session.py # BenchmarkSession, PhaseIssuer, PhaseConfig, SessionResult +├── strategy.py # LoadStrategy protocol, TimedIssueStrategy, +│ # BurstStrategy, ConcurrencyStrategy, +│ # create_load_strategy() +├── agentic_inference_strategy.py # AgenticInferenceStrategy (multi-turn conversations) +├── conversation_manager.py # ConversationManager, ConversationState +├── sample_order.py # SampleOrder, WithoutReplacement, WithReplacement +└── delay.py # poisson_delay_fn, make_delay_fn ``` ## Architecture @@ -33,9 +35,9 @@ each producing an independent report. BenchmarkSession.run(phases) | +-- STARTED - +-- [warmup] strategy.execute() → NO drain (keep in-flight saturated) + +-- [warmup] strategy.execute() → drain_after=False (keep in-flight saturated) +-- [perf phase 1] START_PERFORMANCE_TRACKING → strategy.execute() → drain → STOP_PERFORMANCE_TRACKING - +-- [warmup] strategy.execute() → NO drain (keep in-flight saturated) + +-- [warmup] strategy.execute() → drain_after=False (keep in-flight saturated) +-- [perf phase 2] START_PERFORMANCE_TRACKING → strategy.execute() → drain → STOP_PERFORMANCE_TRACKING +-- [accuracy x N] strategy.execute() → drain (uuid maps collected) +-- ENDED @@ -43,10 +45,17 @@ BenchmarkSession.run(phases) +-- return SessionResult { perf_results: [PhaseResult, ...], accuracy_results: [...] } ``` +**Draining is per-phase, not intrinsic to the phase type.** Each `PhaseConfig` carries a +`drain_after: bool` (default `True`) and an optional `drain_timeout`; `run_phase` drains in-flight +requests iff `phase.drain_after`. Warmup phases are simply configured with `drain_after=False` so +the next phase starts with concurrency already saturated — nothing about `PhaseType.WARMUP` itself +disables draining. + Each performance phase is bracketed by `START_PERFORMANCE_TRACKING` / -`STOP_PERFORMANCE_TRACKING` events, which the `MetricsAggregator` uses to -scope its tracked counters and duration. At the end of each perf phase, -metrics are snapshotted from the KVStoreReader and a `Report` is built. +`STOP_PERFORMANCE_TRACKING` events, which the metrics aggregator uses to +scope its tracked counters and duration. The aggregator folds the event stream into rollups and, +at session end, atomically writes `final_snapshot.json`; the main process reads that snapshot and +builds the `Report` via `Report.from_snapshot` (see [Integration Points](#eventpublisher--metricsaggregator)). > **TODO:** The current `MetricsAggregator` does not support per-phase scoping. > It maintains a single set of counters and series across all tracking windows. @@ -68,15 +77,22 @@ already at the target level. Common uses: ### Load Strategies -Three load patterns, three implementations — each uses the optimal async primitive -for its scheduling semantics, validated by benchmarking: +Four load patterns — each uses the async primitive that fits its scheduling semantics, validated +by benchmarking: + +| LoadPatternType | Strategy | Mechanism | Best At | +| ----------------- | -------------------------- | ------------------------------- | ------------------------ | +| POISSON | `TimedIssueStrategy` | `loop.call_at` (default) | ≤50k QPS | +| POISSON (precise) | `TimedIssueStrategy` | `run_in_executor(busy_wait)` | Sub-100μs precision | +| MAX_THROUGHPUT | `BurstStrategy` | `loop.call_soon` | Max fire rate | +| CONCURRENCY | `ConcurrencyStrategy` | `asyncio.Semaphore` | Fixed concurrency | +| AGENTIC_INFERENCE | `AgenticInferenceStrategy` | response-driven next-turn issue | Multi-turn conversations | -| LoadPatternType | Strategy | Mechanism | Best At | -| ----------------- | --------------------- | ---------------------------- | ------------------- | -| POISSON | `TimedIssueStrategy` | `loop.call_at` (default) | ≤50k QPS | -| POISSON (precise) | `TimedIssueStrategy` | `run_in_executor(busy_wait)` | Sub-100μs precision | -| MAX_THROUGHPUT | `BurstStrategy` | `loop.call_soon` | Max fire rate | -| CONCURRENCY | `ConcurrencyStrategy` | `asyncio.Semaphore` | Fixed concurrency | +The first three are open-loop over independent samples. `AgenticInferenceStrategy` is different: +it drives multi-turn **conversations** where each turn depends on the previous response (see +[AgenticInferenceStrategy](#agenticinferencestrategy) below). It is not selected through +`create_load_strategy` (that path raises for `AGENTIC_INFERENCE`) — it is wired directly when the +phase uses an `AgenticInferenceDataset`. **Default for Poisson is `loop.call_at`:** Sub-millisecond timing precision (600–700μs) with zero GIL contention and low response latency (0.6–1.4ms). No thread pool overhead. @@ -135,6 +151,9 @@ class PhaseConfig: runtime_settings: RuntimeSettings dataset: Dataset phase_type: PhaseType = PhaseType.PERFORMANCE + drain_after: bool = True # drain in-flight at phase end (warmup sets False) + drain_timeout: float | None = None + strategy: LoadStrategy | None = None class BenchmarkSession: @@ -158,14 +177,15 @@ class BenchmarkSession: 3. For each phase: a. Create `SampleOrder` and `LoadStrategy` from phase settings b. Set `self._current_dataset` to phase dataset - c. **WARMUP**: execute strategy, **do not drain** in-flight. No tracking - events, no report. Purpose: bring endpoint to steady-state concurrency - (e.g., fill KV caches, warm up connection pools). The next phase starts + c. **WARMUP** (configured `drain_after=False`): execute strategy, do not drain + in-flight. No tracking events, no report. Purpose: bring endpoint to steady-state + concurrency (e.g., fill KV caches, warm up connection pools). The next phase starts immediately with concurrency already at the target level. - d. **PERFORMANCE**: publish `START_PERFORMANCE_TRACKING`, execute strategy, - drain in-flight, publish `STOP_PERFORMANCE_TRACKING`. Snapshot metrics - from KVStoreReader → build `PhaseResult`. - e. **ACCURACY**: execute strategy, drain in-flight. No tracking events. + d. **PERFORMANCE** (`drain_after=True`): publish `START_PERFORMANCE_TRACKING`, execute + strategy, drain in-flight, publish `STOP_PERFORMANCE_TRACKING`. Build `PhaseResult` + (issued count + uuid map); the aggregator scopes its tracked counters/duration to the + tracking window, and the `Report` is built once at session end from `final_snapshot.json`. + e. **ACCURACY** (`drain_after=True`): execute strategy, drain in-flight. No tracking events. UUID map collected for eval scoring. 4. Publish `SessionEventType.ENDED` 5. Return `SessionResult` (contains `PhaseResult` per perf phase + accuracy maps) @@ -454,6 +474,39 @@ class ConcurrencyStrategy(LoadStrategy): self._sem.release() ``` +### AgenticInferenceStrategy + +Handles `LoadPatternType.AGENTIC_INFERENCE` (`agentic_inference_strategy.py`). Unlike the open-loop +strategies, it is **response-driven**: each conversation is a fixed sequence of turns known upfront +from the dataset, and turn _N+1_ is issued only after turn _N_'s response arrives (its content +feeds the next turn's prompt). Concurrency is measured in **active conversations** +(`_target_concurrency`), not in-flight requests. + +- `execute(phase_issuer)` seeds up to `_target_concurrency` initial conversations, then awaits + completion of all of them. +- On each response, `on_sample_complete(result)` synchronously routes it to its conversation and + calls `_issue_next_turn()` → `_issue_turn_now()`, which stores the new `query_id → conversation_id` + mapping and issues the turn with zero event-loop delay (or an optional per-turn delay). When a + conversation's turn cursor reaches the end of its turn list — or a turn fails — the conversation + finishes and a fresh conversation is seeded to hold concurrency steady. + +Behavior is configured via `AgenticInferenceConfig` (`config/schema.py`): `turn_timeout_s`, +`enable_salt`, `inject_tool_delay`, `num_trajectories_to_issue`, and +`stop_issuing_on_first_user_complete`. The dataset is an `AgenticInferenceDataset` (JSONL with +`conversation_id` / `turn` / `role` / `content` rows grouped consecutively); its +`ConversationMetadata` supplies the per-turn pre-built message history and any per-turn delays. + +### ConversationManager + +`conversation_manager.py` tracks per-conversation progress so the strategy knows when each +conversation is done. `ConversationState` holds `completed_turns`, `failed_turns`, and +`expected_client_turns`; `ConversationManager` owns the map of conversation → state and exposes +`mark_turn_complete()` / `mark_turn_failed()` (both advance `completed_turns`; the latter also bumps +`failed_turns`). A conversation is complete once its completed turns reach the expected count. + +Every `EventRecord` the session publishes in agentic mode carries `conversation_id` and `turn`, so +the metrics aggregator and event log can attribute each sample to its conversation and turn index. + ### SampleIssuer (Protocol) ```python @@ -562,21 +615,19 @@ sequenceDiagram participant C as Caller (execute.py) participant B as BenchmarkSession participant E as EventPublisher - participant M as MetricsAggregator - participant K as KVStoreReader + participant M as MetricsAggregatorService C->>B: run(phases) B->>E: STARTED Note over B: === Saturation Phase === - B->>B: execute strategy (untracked, no drain) + B->>B: execute strategy (untracked, drain_after=False) Note over B: === Perf Phase 1 (e.g. QPS=1000) === B->>E: START_PERFORMANCE_TRACKING B->>B: execute strategy B->>B: drain in-flight B->>E: STOP_PERFORMANCE_TRACKING - B->>K: snapshot metrics → Report Note over B: PhaseResult("perf_qps1k", issued_count=N) Note over B: === Perf Phase 2 (e.g. QPS=5000) === @@ -584,7 +635,6 @@ sequenceDiagram B->>B: execute strategy B->>B: drain in-flight B->>E: STOP_PERFORMANCE_TRACKING - B->>K: snapshot metrics → Report Note over B: PhaseResult("perf_qps5k", issued_count=N) Note over B: === Accuracy Phase === @@ -593,7 +643,10 @@ sequenceDiagram Note over B: PhaseResult("accuracy", uuid_map) B->>E: ENDED + E->>M: (event stream throughout) + M->>M: write final_snapshot.json B-->>C: SessionResult + C->>C: Report.from_snapshot(final_snapshot.json) ``` ### Separate Timer Process Data Flow @@ -647,9 +700,9 @@ graph TD subgraph "Worker Process N" WN["HTTP → endpoint"] end - subgraph "MetricsAggregator (subprocess)" - MA["ZmqEventRecordSubscriber"] - KB["KVStore (mmap)"] + subgraph "MetricsAggregatorService (subprocess)" + MA["EventRecord subscriber"] + KB["MetricsRegistry → MetricsSnapshot PUB + final_snapshot.json"] MA --> KB end @@ -685,8 +738,8 @@ graph TD subgraph "Worker Processes" W["HTTP → endpoint"] end - subgraph "MetricsAggregator" - MA["Subscriber → KVStore"] + subgraph "MetricsAggregatorService" + MA["Subscriber → MetricsRegistry → snapshot"] end T -->|"ZMQ IPC"| R @@ -719,6 +772,11 @@ def create_load_strategy( case LoadPatternType.CONCURRENCY: return ConcurrencyStrategy(lp.target_concurrency, sample_order) + + case LoadPatternType.AGENTIC_INFERENCE: + # Agentic runs are response-driven and need an AgenticInferenceDataset; + # they are wired up directly, not through this open-loop factory. + raise ValueError("AGENTIC_INFERENCE is not created via create_load_strategy") ``` --- @@ -783,15 +841,18 @@ to use `await shutdown_async()`. ### EventPublisher / MetricsAggregator -Session publishes `EventRecord` instances via `ZmqEventRecordPublisher`. The publisher -uses non-blocking ZMQ send with fd-based writer fallback — safe to call from sync -callbacks (like `call_at` fire functions). +The session publishes `EventRecord` instances via `EventPublisherService` (a per-instance wrapper +over `ZmqMessagePublisher[EventRecord]`). The publisher uses non-blocking ZMQ send with an fd-based +writer fallback — safe to call from sync callbacks (like `call_at` fire functions). -> **Key fix:** `Report.from_kv_reader` currently reads counter keys (`n_samples_issued`, -> `duration_ns`) that don't match the `MetricCounterKey` enum written by the aggregator -> (`total_samples_issued`, `tracked_duration_ns`). Must update `from_kv_reader` to use -> the actual key names. Performance reports should use `tracked_*` counters. A -> `test_started_at` counter must be added to the aggregator (set on `SessionEventType.STARTED`). +Those records flow to the [`MetricsAggregatorService`](../async_utils/services/metrics_aggregator/DESIGN.md) +subprocess, which folds them into a `MetricsRegistry` (counters + HDR/series rollups), publishes +live `MetricsSnapshot` messages, and — on `ENDED` / terminal state — atomically writes +`final_snapshot.json`. The main process reads that file (`json.loads` → dict) and builds the report +with `Report.from_snapshot(dict, run_config=...)`; if the file is missing (aggregator SIGKILLed) it +falls back to the subscriber's last live snapshot and the report is marked `complete=False`. The +report reads `tracked_*` counters (`tracked_samples_completed`, `tracked_duration_ns`, …) plus the +LoadGen "completed" window; see [metrics/report_design.md](../metrics/report_design.md). ### HttpClientSampleIssuer Migration @@ -844,23 +905,30 @@ phase. The receiver must distinguish stale vs current-phase completions: ```python def _handle_response(self, resp: QueryResult) -> None: query_id = resp.id - # Always publish the event (aggregator tracks all samples) - self._publisher.publish(EventRecord( - event_type=SampleEventType.COMPLETE, - timestamp_ns=resp.completed_at, - sample_uuid=query_id, - )) - # Only route to current phase strategy if this is a current-phase query - if query_id in self._current_phase_issuer.uuid_to_index: - self._current_phase_issuer.inflight -= 1 + # ERROR is published before COMPLETE (the aggregator depends on this order + # for tracked-failure counting). + if resp.error is not None: + self._publisher.publish(EventRecord(event_type=ErrorEventType.GENERIC, ...)) + # COMPLETE is NOT published for WARMUP phases — warmup samples must not + # land in the tracked counters/series. + if self._current_phase_type != PhaseType.WARMUP: + self._publisher.publish(EventRecord( + event_type=SampleEventType.COMPLETE, + timestamp_ns=resp.completed_at, + sample_uuid=query_id, + )) + # Only route to the current phase's strategy if this is a current-phase query. + # Stale completions (arrived from a prior no-drain phase) fall through here. + if phase_issuer is not None and query_id in phase_issuer.uuid_to_index: + phase_issuer.mark_inflight_complete() if self._current_strategy: self._current_strategy.on_query_complete(query_id) - # Stale completions: event published but strategy/inflight not affected ``` -Same guard applies to `StreamChunk` with `is_complete=True` — check -`uuid_to_index` membership before decrementing inflight. Non-final -StreamChunks don't affect inflight and can be published unconditionally. +The `uuid_to_index` membership gate keeps a stale completion (arriving from a prior warmup/no-drain +phase) from being attributed to the current phase's strategy or inflight count. Streaming responses +are handled on the same path: intermediate `StreamChunk`s emit `RECV_FIRST`/chunk events, and the +final response drives the COMPLETE/inflight bookkeeping above. ### Sync Per-Sample Work in Callbacks @@ -934,7 +1002,9 @@ async def _run_benchmark_async(ctx, loop) -> BenchmarkResult: await http_client.shutdown_async() publisher.close() await asyncio.to_thread(launcher.wait_for_exit, None) - report = Report.from_kv_reader(kv_reader) # after aggregator exits + # After the aggregator exits it has written final_snapshot.json: + snap = json.loads((metrics_output_dir / "final_snapshot.json").read_bytes()) + report = Report.from_snapshot(snap, run_config=run_config) return BenchmarkResult(session=result, collector=collector, report=report, ...) ``` @@ -1005,27 +1075,29 @@ to a child process, with the TUI as the foreground process reading periodic repo ``` TUI Process (foreground): - Renders live dashboard (throughput, latency, progress) - - Reads BasicKVStoreReader for real-time metrics from /dev/shm + - Subscribes to the aggregator's MetricsSnapshot PUB stream (metrics SUB) - Receives SessionResult via IPC on completion Benchmark Process (child): - Runs BenchmarkSession on its own event loop - - Writes metrics via EventPublisher -> MetricsAggregator -> KVStore + - Emits events via EventPublisher -> MetricsAggregatorService (which PUBs snapshots) - Returns SessionResult to parent via IPC (pickle over pipe / ZMQ) ``` This architecture is enabled by the current design's clean separation: -- **KVStore** is already cross-process readable (mmap on /dev/shm) +- **Metrics are already cross-process** — the aggregator publishes `MetricsSnapshot` frames over a + ZMQ PUB socket at the `--publish-interval` cadence, so any subscriber renders live percentiles + without touching the benchmark process. - **BenchmarkSession** has no UI dependencies — it takes callbacks - **SessionResult** is a frozen dataclass, trivially serializable -- The `on_sample_complete` callback would not be used in TUI mode (no - cross-process callback). Instead, the TUI polls KVStoreReader for - `tracked_samples_completed` to update the progress display. +- The `on_sample_complete` callback would not be used in TUI mode (no cross-process callback). + Instead, the TUI reads the latest `MetricsSnapshot` (e.g. `tracked_samples_completed`) to update + the progress display. -The TUI process can also read the `Report` from the KVStore at any time for -live intermediate reports (current QPS, latency distribution so far), not -just the final report. +The TUI can render live intermediate reports (current QPS, latency distribution so far) from each +live snapshot, and build the final `Report` from `final_snapshot.json` once the run ends — the same +`Report.from_snapshot` path used by the CLI. > **Constraint:** The benchmark child process must be a **non-daemon** OS process > (e.g., `subprocess.Popen` or `multiprocessing.Process(daemon=False)`). From de3607570492b71a2e7af1efaa5e20b3b719e464 Mon Sep 17 00:00:00 2001 From: Alice Cheng Date: Mon, 20 Jul 2026 11:30:59 -0700 Subject: [PATCH 4/6] docs(compliance): note per-phase test_mode override (perf-only is the default) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The audit plan and AGENTS.md described audit phases as unconditionally performance-only, but AuditRunSpec now carries a test_mode field and run_audit selects datasets per phase (perf-only for PERF, perf+accuracy for ACC/BOTH). Document that perf-only is the default and that the test_mode override exists but is currently unused — every registered audit (TEST04) runs perf-only. - compliance_audit_plan.md: mark the perf-only phase-config box with a footnote and expand the step-4 prose to describe the test_mode override. - AGENTS.md: update the compliance-audit rows accordingly. Co-Authored-By: Claude Opus 4.8 --- AGENTS.md | 34 +++++++++++++++++----------------- docs/compliance_audit_plan.md | 14 +++++++++++--- 2 files changed, 28 insertions(+), 20 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index bf87ebfec..a40ad9542 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -85,22 +85,22 @@ Dataset Manager --> Load Generator --> Endpoint Client --> External Endpoint ### Key Components -| Component | Location | Purpose | -| ----------------------------------- | ------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | -| **Load Generator** | `src/inference_endpoint/load_generator/` | Central orchestrator: `BenchmarkSession` owns the lifecycle, `PhaseIssuer` drives per-phase execution, `TimedIssueStrategy`/`BurstStrategy`/`ConcurrencyStrategy` control timing. Emits `ERROR` before `COMPLETE` for failed queries (metrics aggregator depends on this order). | -| **Endpoint Client** | `src/inference_endpoint/endpoint_client/` | Multi-process HTTP workers communicating via ZMQ IPC. `HTTPEndpointClient` is the main entry point | -| **Dataset Manager** | `src/inference_endpoint/dataset_manager/` | Loads JSONL, HuggingFace, CSV, JSON, Parquet datasets. `Dataset` base class with `load_sample()`/`num_samples()` interface | -| **Metrics Aggregator** | `src/inference_endpoint/async_utils/services/metrics_aggregator/` | Subprocess. Subscribes to events, aggregates per-sample metrics into a `MetricsRegistry` (counters + HDR-histogram series + raw values), publishes `MetricsSnapshot` over IPC PUB at a configurable cadence (`SessionState`: `INITIALIZE` → `LIVE` → `DRAINING` → {`COMPLETE` \| `INTERRUPTED`}). Final snapshot is atomically written to `final_snapshot.json` as the **primary** Report source; the terminal pub/sub frame is a TUI "run finished" signal only. | -| **Report** | `src/inference_endpoint/metrics/report.py` | `Report.from_snapshot(dict)` — pure-function builder consuming the dict form (`snapshot_to_dict`). Reads `final_snapshot.json` directly via `json.loads` (no Struct decode). Plumbs `complete = (state == "complete" and n_pending_tasks == 0)`; renders an explicit warning for `INTERRUPTED` runs. | -| **Config** | `src/inference_endpoint/config/`, `endpoint_client/config.py` | Pydantic-based YAML schema (`schema.py`), `HTTPClientConfig` (single Pydantic model for CLI/YAML/runtime), `RuntimeSettings` | -| **CLI** | `src/inference_endpoint/main.py`, `commands/benchmark/cli.py` | cyclopts-based, auto-generated from `schema.py` and `HTTPClientConfig` Pydantic models. Flat shorthands via `cyclopts.Parameter(alias=...)` | -| **Async Utils** | `src/inference_endpoint/async_utils/` | `LoopManager` (uvloop + eager_task_factory), ZMQ transport layer, generic `MessageCodec[T]`-parametrized pub/sub, event publisher | -| **OpenAI/SGLang** | `src/inference_endpoint/openai/`, `sglang/` | Protocol adapters and response accumulators for different API formats. `openai_completions` adapter (`completions_adapter.py`) sends pre-tokenized token IDs to `/v1/completions`, bypassing the server chat template — required for gpt-oss-120b on vLLM. `sglang` adapter sends to `/generate` via `input_ids`. Both apply `Harmonize()` client-side. | -| **TensorRT-LLM** | `src/inference_endpoint/trtllm/` | Adapter for TensorRT-LLM endpoints. `TRTLLMAdapter` sends requests; `TRTLLMSSEAccumulator` handles SSE streaming responses. | -| **DeepSeek-R1 (MLPerf)** | `src/inference_endpoint/evaluation/scoring.py` (`LegacyMLPerfDeepSeekR1Scorer`), `examples/07_DeepSeekR1_Example/` | MLPerf DeepSeek-R1 accuracy. TensorRT-LLM is OpenAI-compatible, so it is served via `api_type: openai` / `openai_completions` (no dedicated trtllm adapter). The combined multi-subset eval (`math500`/`aime`/`gpqa`/`mmlu_pro`/`livecodebench`) is the official MLCommons `eval_accuracy.py`, run out-of-process via `uv run --project` against the isolated subproject at `src/inference_endpoint/evaluation/legacy_mlperf_deepseek_r1/` (a uv subproject excluded from the parent wheel; mirrors the VBench pattern). The example feeds the exact MLPerf prompt via pre-tokenized `input_tokens` to `/v1/completions`. | -| **VideoGen** | `src/inference_endpoint/videogen/` | Adapter for video-generation endpoints (e.g. trtllm-serve `POST /v1/videos/generations`, used by MLPerf WAN2.2-T2V-A14B). Defaults to `response_format=video_path` (server saves video to shared storage and returns path) to avoid large byte payloads. Accuracy mode also runs on `video_path`: the adapter mirrors the path into `response_output` so the event log carries it to `VBenchScorer` (see `evaluation/scoring.py`), which scores videos via VBench from a sibling `uv` subproject at `examples/09_Wan22_VideoGen_Example/accuracy/` (vbench's `transformers==4.33.2` + `numpy<2` pins are incompatible with the parent env, so it runs out-of-process via `uv run --project`). Dataset is ingested via the generic JSONL loader. | -| **Compliance (submission checker)** | `src/inference_endpoint/compliance/checker.py`, `scripts/check_compliance.py` | Validates a completed run's report directory against a registered ruleset. `check_submission(report_dir, ruleset, model)` reads the resolved `config.yaml` plus scorer output (`accuracy/accuracy_results.json` for accuracy, `scores.json` for the agentic perf run) and runs config-lock (deterministic + single-stream), the accuracy gate (`score >= factor x reference`, factor 0.97 for Edge-Agentic), and run validity (0 dropped turns). Server-side launch flags (`--reasoning off`, `--ctx-size`) aren't in client artifacts, so they're surfaced as manual attestations. CLI: `scripts/check_compliance.py REPORT_DIR` (exit 0 = pass). | -| **Compliance (audit tests)** | `src/inference_endpoint/compliance/`, `commands/audit.py` | MLPerf compliance audits. `AuditTest` protocol + `AuditRunSpec`/`AuditRunArtifacts` + registry (`compliance/__init__.py`); `OutputCachingAudit` (`compliance/audit_test/output_caching_test.py`, which also owns the QPS-specific `AuditRunStats`) implements MLPerf **TEST04** output-caching detection — reference phase (distinct samples) vs. fixed-sample audit phase, comparing QPS against `threshold`. `commands/audit.py:run_audit` runs phases via `AuditTest.plan_runs`/`validate`, writing `audit_result.json`/`verify_.txt` atomically via `compliance/result.py`. Enabled by the `audit:` YAML block; `cli._run` runs it after the main benchmark (upstream MLPerf order: perf run, then TEST04), or standalone with `audit.only: true`. Performance-only. | +| Component | Location | Purpose | +| ----------------------------------- | ------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| **Load Generator** | `src/inference_endpoint/load_generator/` | Central orchestrator: `BenchmarkSession` owns the lifecycle, `PhaseIssuer` drives per-phase execution, `TimedIssueStrategy`/`BurstStrategy`/`ConcurrencyStrategy` control timing. Emits `ERROR` before `COMPLETE` for failed queries (metrics aggregator depends on this order). | +| **Endpoint Client** | `src/inference_endpoint/endpoint_client/` | Multi-process HTTP workers communicating via ZMQ IPC. `HTTPEndpointClient` is the main entry point | +| **Dataset Manager** | `src/inference_endpoint/dataset_manager/` | Loads JSONL, HuggingFace, CSV, JSON, Parquet datasets. `Dataset` base class with `load_sample()`/`num_samples()` interface | +| **Metrics Aggregator** | `src/inference_endpoint/async_utils/services/metrics_aggregator/` | Subprocess. Subscribes to events, aggregates per-sample metrics into a `MetricsRegistry` (counters + HDR-histogram series + raw values), publishes `MetricsSnapshot` over IPC PUB at a configurable cadence (`SessionState`: `INITIALIZE` → `LIVE` → `DRAINING` → {`COMPLETE` \| `INTERRUPTED`}). Final snapshot is atomically written to `final_snapshot.json` as the **primary** Report source; the terminal pub/sub frame is a TUI "run finished" signal only. | +| **Report** | `src/inference_endpoint/metrics/report.py` | `Report.from_snapshot(dict)` — pure-function builder consuming the dict form (`snapshot_to_dict`). Reads `final_snapshot.json` directly via `json.loads` (no Struct decode). Plumbs `complete = (state == "complete" and n_pending_tasks == 0)`; renders an explicit warning for `INTERRUPTED` runs. | +| **Config** | `src/inference_endpoint/config/`, `endpoint_client/config.py` | Pydantic-based YAML schema (`schema.py`), `HTTPClientConfig` (single Pydantic model for CLI/YAML/runtime), `RuntimeSettings` | +| **CLI** | `src/inference_endpoint/main.py`, `commands/benchmark/cli.py` | cyclopts-based, auto-generated from `schema.py` and `HTTPClientConfig` Pydantic models. Flat shorthands via `cyclopts.Parameter(alias=...)` | +| **Async Utils** | `src/inference_endpoint/async_utils/` | `LoopManager` (uvloop + eager_task_factory), ZMQ transport layer, generic `MessageCodec[T]`-parametrized pub/sub, event publisher | +| **OpenAI/SGLang** | `src/inference_endpoint/openai/`, `sglang/` | Protocol adapters and response accumulators for different API formats. `openai_completions` adapter (`completions_adapter.py`) sends pre-tokenized token IDs to `/v1/completions`, bypassing the server chat template — required for gpt-oss-120b on vLLM. `sglang` adapter sends to `/generate` via `input_ids`. Both apply `Harmonize()` client-side. | +| **TensorRT-LLM** | `src/inference_endpoint/trtllm/` | Adapter for TensorRT-LLM endpoints. `TRTLLMAdapter` sends requests; `TRTLLMSSEAccumulator` handles SSE streaming responses. | +| **DeepSeek-R1 (MLPerf)** | `src/inference_endpoint/evaluation/scoring.py` (`LegacyMLPerfDeepSeekR1Scorer`), `examples/07_DeepSeekR1_Example/` | MLPerf DeepSeek-R1 accuracy. TensorRT-LLM is OpenAI-compatible, so it is served via `api_type: openai` / `openai_completions` (no dedicated trtllm adapter). The combined multi-subset eval (`math500`/`aime`/`gpqa`/`mmlu_pro`/`livecodebench`) is the official MLCommons `eval_accuracy.py`, run out-of-process via `uv run --project` against the isolated subproject at `src/inference_endpoint/evaluation/legacy_mlperf_deepseek_r1/` (a uv subproject excluded from the parent wheel; mirrors the VBench pattern). The example feeds the exact MLPerf prompt via pre-tokenized `input_tokens` to `/v1/completions`. | +| **VideoGen** | `src/inference_endpoint/videogen/` | Adapter for video-generation endpoints (e.g. trtllm-serve `POST /v1/videos/generations`, used by MLPerf WAN2.2-T2V-A14B). Defaults to `response_format=video_path` (server saves video to shared storage and returns path) to avoid large byte payloads. Accuracy mode also runs on `video_path`: the adapter mirrors the path into `response_output` so the event log carries it to `VBenchScorer` (see `evaluation/scoring.py`), which scores videos via VBench from a sibling `uv` subproject at `examples/09_Wan22_VideoGen_Example/accuracy/` (vbench's `transformers==4.33.2` + `numpy<2` pins are incompatible with the parent env, so it runs out-of-process via `uv run --project`). Dataset is ingested via the generic JSONL loader. | +| **Compliance (submission checker)** | `src/inference_endpoint/compliance/checker.py`, `scripts/check_compliance.py` | Validates a completed run's report directory against a registered ruleset. `check_submission(report_dir, ruleset, model)` reads the resolved `config.yaml` plus scorer output (`accuracy/accuracy_results.json` for accuracy, `scores.json` for the agentic perf run) and runs config-lock (deterministic + single-stream), the accuracy gate (`score >= factor x reference`, factor 0.97 for Edge-Agentic), and run validity (0 dropped turns). Server-side launch flags (`--reasoning off`, `--ctx-size`) aren't in client artifacts, so they're surfaced as manual attestations. CLI: `scripts/check_compliance.py REPORT_DIR` (exit 0 = pass). | +| **Compliance (audit tests)** | `src/inference_endpoint/compliance/`, `commands/audit.py` | MLPerf compliance audits. `AuditTest` protocol + `AuditRunSpec`/`AuditRunArtifacts` + registry (`compliance/__init__.py`); `OutputCachingAudit` (`compliance/audit_test/output_caching_test.py`, which also owns the QPS-specific `AuditRunStats`) implements MLPerf **TEST04** output-caching detection — reference phase (distinct samples) vs. fixed-sample audit phase, comparing QPS against `threshold`. `commands/audit.py:run_audit` runs phases via `AuditTest.plan_runs`/`validate`, writing `audit_result.json`/`verify_.txt` atomically via `compliance/result.py`. Enabled by the `audit:` YAML block; `cli._run` runs it after the main benchmark (upstream MLPerf order: perf run, then TEST04), or standalone with `audit.only: true`. Perf-only by default (a phase may opt into accuracy via `AuditRunSpec.test_mode`, but this is unused today). | ### Hot-Path Architecture @@ -162,7 +162,7 @@ Validation is layered: Orthogonal to the main run: a YAML-only `audit:` block (`show=False`, no CLI flag) on `BenchmarkConfig` selects an `AuditTest`. If `audit:` is set, `cli._run` freezes a shared `report_dir`, runs the main benchmark, then calls `commands/audit.py:run_audit` — the upstream MLPerf order (perf run, then TEST04). Unlike upstream there is no SUT reset between stages; see docs/compliance_audit_plan.md ("Run ordering"). `run_audit`: -1. Builds a performance-only per-phase config (drops accuracy datasets so no phase re-issues or re-scores them) and, after the first phase's dataset loads, calls `AuditTest.validate(cfg, dataset_size, load_pattern)` to bounds-check the test's sample counts/indices and load pattern before any phase issues load (reuses the loaded dataset — no extra load). +1. Builds a per-phase config that is performance-only by default (drops accuracy datasets so no phase re-issues or re-scores them), overridable per-`AuditRunSpec` via `test_mode` (`ACC`/`BOTH` keeps accuracy datasets — supported but currently unused; every registered audit runs perf-only), and, after the first phase's dataset loads, calls `AuditTest.validate(cfg, dataset_size, load_pattern)` to bounds-check the test's sample counts/indices and load pattern before any phase issues load (reuses the loaded dataset — no extra load). 2. Runs each `AuditRunSpec` phase (from `AuditTest.plan_runs`) back-to-back under its own `/