[Perf] (MG_Remote): The producer's transport park is timed as its own per-frame number, and a reducer pairs the two role logs into a frame-time breakdown

transport_wait_ns measures wall time in Ring::Wait's flush/spin/sleep, so
the client's parked time is read directly instead of inferred by
subtracting CPU clocks that run concurrently with the server.
frame_stats.py matches client and server present serials into per-frame
avg/p50/p99 splits; the README documents the MOBILEGL_PIPE_STATS=1
workflow.

MobileGL/MG_Remote/Transport/LinkMetrics.cpp

@@ -14,6 +14,7 @@ namespace MobileGL::MG_Remote::Transport {
 namespace {
     struct Window {
         std::uint64_t waitReplies = 0, samples = 0, replyNs = 0, stageBytes = 0;
+        std::uint64_t transportWaitNs = 0;
         std::array<std::uint64_t, 32> histogram{};
         // One slot per wire op, plus a last slot for anything past the table. See
         // LinkMetricsMaxOps: the point is that the per-op numbers still SUM to waitReplies.
@@ -92,14 +93,16 @@ namespace {
         }
         MGLOG_I("P65LinkMetrics kind=%s frame=%llu wait_replies=%llu rtt_samples=%llu "
                 "rtt_mean_us=%.3f rtt_p50_upper_us=%llu rtt_p99_upper_us=%llu "
-                "stage_bytes=%llu wall_ns=%llu client_thread_cpu_ns=%llu rtt_hist_us_pow2=%s",
+                "stage_bytes=%llu wall_ns=%llu client_thread_cpu_ns=%llu "
+                "transport_wait_ns=%llu rtt_hist_us_pow2=%s",
             kind, static_cast<unsigned long long>(metrics.frame),
             static_cast<unsigned long long>(w.waitReplies), static_cast<unsigned long long>(w.samples),
             w.samples == 0 ? 0.0 : double(w.replyNs) / double(w.samples) / 1000.0,
             static_cast<unsigned long long>(QuantileUpperUs(w, 50)),
             static_cast<unsigned long long>(QuantileUpperUs(w, 99)),
             static_cast<unsigned long long>(w.stageBytes), static_cast<unsigned long long>(wallNs),
-            static_cast<unsigned long long>(cpuNs), histogram);
+            static_cast<unsigned long long>(cpuNs),
+            static_cast<unsigned long long>(w.transportWaitNs), histogram);
         // AND THE PER-OP SPLIT WITH EVERY FRAME. A whole-session total cannot name the rows behind
         // ONE frame, and one frame is the whole shape of this problem: the device run that
         // motivated this printed 43,232 of its 43,629 waits in a single atlas frame, and the exit
@@ -167,6 +170,15 @@ void LinkMetricsStageBytes(std::uint64_t bytes) {
     if (!metrics.active) return;
     metrics.current.stageBytes += bytes; metrics.total.stageBytes += bytes;
 }
+std::uint64_t LinkMetricsBeginTransportWait() {
+    return metrics.active ? ClockNs() : 0;
+}
+void LinkMetricsEndTransportWait(std::uint64_t startedNs) {
+    if (!metrics.active || startedNs == 0) return;
+    const auto elapsed = ClockNs() - startedNs;
+    metrics.current.transportWaitNs += elapsed;
+    metrics.total.transportWaitNs += elapsed;
+}
 void LinkMetricsPresent() {
     if (!metrics.active) return;
     ++metrics.frame;

MobileGL/MG_Remote/Transport/LinkMetrics.h

@@ -17,6 +17,11 @@ namespace MobileGL::MG_Remote::Transport {
     std::uint64_t LinkMetricsBeginReply(bool wantsReply, std::uint32_t op);
     void LinkMetricsReplyApplied(std::uint64_t startedNs);
     void LinkMetricsStageBytes(std::uint64_t bytes);
+    // Wall time spent in the producer's transport park (flush, spin and sleep).
+    // The server can run concurrently, so this is an independently measured
+    // duration, not a residual obtained by subtracting role CPU clocks.
+    std::uint64_t LinkMetricsBeginTransportWait();
+    void LinkMetricsEndTransportWait(std::uint64_t startedNs);
 
     // THE SAME NUMBER THE PER-FRAME LINE PRINTS AS `wait_replies` (P65LinkMetrics), read back
     // in-process. P12 item B needs it as a GATE rather than as a log line: "the texture half no

MobileGL/MG_Remote/Transport/Ring.cpp

@@ -9,6 +9,7 @@
 #include "Ring.h"
 
 #include "SessionRings.h"
+#include "LinkMetrics.h"
 #include "WireLog.h"
 
 #include <MG_Util/Debug/Log.h>
@@ -472,21 +473,27 @@ namespace MobileGL::MG_Remote::Transport {
         if (!Valid() || m_selfBell == nullptr) {
             return SessionWait::TimedOut;
         }
+        const auto transportStarted = LinkMetricsBeginTransportWait();
         // COUNTED BEFORE THE WAIT AND ONLY WHEN ONE REALLY HAPPENS - after the two refusals
         // above, which answer TimedOut without ever reaching the bell. See Waits()/Parks() in
         // the header for why the pair is the reading and either number alone is not.
-        if (m_link && m_link->Flush() != MOBILEGL_OK) return SessionWait::ShutDown;
+        if (m_link && m_link->Flush() != MOBILEGL_OK) {
+            LinkMetricsEndTransportWait(transportStarted);
+            return SessionWait::ShutDown;
+        }
         m_waits.fetch_add(1, std::memory_order_relaxed);
         // &m_parks is the OTHER half of the pair, and it is passed rather than read off the
         // bell: the bell is this endpoint's and the encoder waits on it too, so its own
         // ParkEntries() is not a subset of the waits counted one line above. See Parks() in
         // SessionRings.h.
         if (m_selfBell->Wait(m_control->producerParked, ready, m_spinUs, timeoutMs, &m_parks)) {
+            LinkMetricsEndTransportWait(transportStarted);
             return SessionWait::Reached;
         }
         // Wait == false && Dead() is "the session was shut down", and it is the
         // only thing that returns from a kWaitForever park. Anything else is the
         // deadline.
+        LinkMetricsEndTransportWait(transportStarted);
         return m_selfBell->Dead() ? SessionWait::ShutDown : SessionWait::TimedOut;
     }
 

tools/trace_replay/README.md

@@ -563,6 +563,25 @@ by name (`NoServerDisplay`) and that service goes on serving.
 
 ## TCP credit and Stage measurements
 
+For a paired frame-time breakdown, enable `MOBILEGL_PIPE_STATS=1` in both roles
+and keep their logs. The following reducer matches client and server present
+serials, drops the first (unpaired) server interval and one warmup frame, and
+reports avg/p50/p99 in milliseconds plus percentages of client present time:
+
+```bash
+python3 tools/trace_replay/frame_stats.py \
+  --client-log mobilegl.client.log --server-log mobilegl.server.log \
+  --warmup-frames 1 --output frame-stats.json
+```
+
+`transport_wait_ms` measures wall time in the client producer's transport park
+(flush, spin and sleep). It includes waiting for replies, present credit and
+ring space; it is not socket latency. Client and server CPU can overlap with
+each other and with a transport wait, so each percentage uses the client
+present interval as its denominator and the three percentages need not sum to
+100%. A missing frame is rejected. Percentiles use nearest rank over per-frame
+samples.
+
 `benchmark_tcp_credits.py` runs the complete OpenRA and rd12 traces at credits
 1, 2, and 3, then selects matching client/server tail frame IDs (100 and 200
 frames respectively). Extract the unchanged trace files as

tools/trace_replay/frame_stats.py

@@ -0,0 +1,128 @@
+#!/usr/bin/env python3
+"""Pair client/server present intervals and summarize their frame time.
+
+MOBILEGL_PIPE_STATS=1 enables the per-present clocks. Frame IDs are matched
+exactly; the first server interval is invalid because it has no preceding
+present. Transport wait measures wall time inside the client's transport park
+(flush, spin and sleep). It can overlap with server CPU and is not network latency.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+import re
+from pathlib import Path
+
+
+FIELDS = re.compile(r"(\w+)=([^\s]+)")
+CLIENT_MARKER = "P65LinkMetrics "
+SERVER_MARKER = "P65ServerMetrics "
+
+
+def read_frames(path: Path, marker: str, *, server: bool) -> dict[int, dict]:
+    rows = {}
+    for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
+        if marker not in line:
+            continue
+        values = dict(FIELDS.findall(line.split(marker, 1)[1]))
+        if server and values.get("valid") != "1":
+            continue
+        if not server and values.get("kind") != "frame":
+            continue
+        try:
+            number = int(values["frame"])
+            wall = int(values["wall_ns"])
+            cpu = int(values["apply_thread_cpu_ns" if server else "client_thread_cpu_ns"])
+            transport = 0 if server else int(values["transport_wait_ns"])
+        except (KeyError, ValueError) as error:
+            raise ValueError(f"{path}: malformed {marker.strip()} row") from error
+        if number in rows:
+            raise ValueError(f"{path}: duplicate frame {number}")
+        # The wall and thread-CPU clocks are sampled separately. Their nanosecond
+        # readings can differ slightly at a boundary even for a fully busy thread.
+        tolerance = max(100_000, wall // 100)
+        if (number <= 0 or wall <= 0 or cpu <= 0 or cpu > wall + tolerance or
+                transport < 0 or transport > wall + tolerance):
+            raise ValueError(f"{path}: invalid clocks in frame {number}")
+        rows[number] = {"wall_ns": wall, "cpu_ns": cpu, "transport_ns": transport}
+    if not rows:
+        raise ValueError(f"{path}: no usable {marker.strip()} frames")
+    return rows
+
+
+def quantile_ns(samples: list[int], percentile: int) -> int:
+    ordered = sorted(samples)
+    return ordered[max(0, math.ceil(len(ordered) * percentile / 100) - 1)]
+
+
+def distribution(samples: list[int]) -> dict[str, float]:
+    return {"avg": sum(samples) / len(samples) / 1e6,
+            "p50": quantile_ns(samples, 50) / 1e6,
+            "p99": quantile_ns(samples, 99) / 1e6}
+
+
+def summarize(client_path: Path, server_path: Path, *, warmup: int = 0,
+              tail: int | None = None) -> dict:
+    client = read_frames(client_path, CLIENT_MARKER, server=False)
+    server = read_frames(server_path, SERVER_MARKER, server=True)
+    selected = sorted(number for number in client if number > warmup)
+    if tail is not None:
+        selected = selected[-tail:]
+    if not selected:
+        raise ValueError("no frames remain after warmup/tail selection")
+    if selected != list(range(selected[0], selected[-1] + 1)):
+        raise ValueError("selected client frame IDs are not contiguous")
+    missing = [number for number in selected if number not in server]
+    if missing:
+        raise ValueError(f"server is missing matched frames: {missing[:10]}")
+
+    columns = {name: [] for name in ("client_present_ms", "server_present_ms",
+                                     "client_cpu_ms", "transport_wait_ms", "server_cpu_ms")}
+    for number in selected:
+        c, s = client[number], server[number]
+        for key, value in (("client_present_ms", c["wall_ns"]),
+                           ("server_present_ms", s["wall_ns"]),
+                           ("client_cpu_ms", c["cpu_ns"]),
+                           ("transport_wait_ms", c["transport_ns"]),
+                           ("server_cpu_ms", s["cpu_ns"])):
+            columns[key].append(value)
+
+    duration = sum(columns["client_present_ms"])
+    shares = {name: 100 * sum(columns[name]) / duration
+              for name in ("client_cpu_ms", "transport_wait_ms", "server_cpu_ms")}
+    return {"frames": len(selected), "first_frame": selected[0],
+            "last_frame": selected[-1], "warmup_frames": warmup,
+            "definition": "transport_wait is client wall time in SessionProducer::Park "
+                          "(flush, spin and sleep); role CPU and transport wait can overlap",
+            "percentile": "nearest rank", "unit": "ms",
+            "stats": {key: distribution(values) for key, values in columns.items()},
+            "share_percent_of_client_present": shares}
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--client-log", type=Path, required=True)
+    parser.add_argument("--server-log", type=Path, required=True)
+    parser.add_argument("--warmup-frames", type=int, default=1)
+    parser.add_argument("--tail-frames", type=int)
+    parser.add_argument("--output", type=Path)
+    args = parser.parse_args()
+    if args.warmup_frames < 0 or args.tail_frames is not None and args.tail_frames <= 0:
+        parser.error("warmup must be nonnegative and tail must be positive")
+    try:
+        result = summarize(args.client_log, args.server_log, warmup=args.warmup_frames,
+                           tail=args.tail_frames)
+    except (OSError, ValueError) as error:
+        parser.error(str(error))
+    rendered = json.dumps(result, ensure_ascii=False, indent=2) + "\n"
+    if args.output:
+        args.output.write_text(rendered, encoding="utf-8")
+    else:
+        print(rendered, end="")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tools/trace_replay/test_frame_stats.py

@@ -0,0 +1,51 @@
+#!/usr/bin/env python3
+"""Small fixture checks for frame ID pairing and additive frame attribution."""
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from frame_stats import summarize
+
+
+class FrameStatsTest(unittest.TestCase):
+    def logs(self, root, *, missing_server=False, overlapping=False):
+        client = root / "client.log"
+        server = root / "server.log"
+        client.write_text("\n".join(
+            f"P65LinkMetrics kind=frame frame={n} wall_ns={wall} "
+            f"client_thread_cpu_ns={cpu} transport_wait_ns={wait}"
+            for n, wall, cpu, wait in ((1, 10000000, 2000000, 1000000),
+                                       (2, 20000000, 5000000, 4000000),
+                                       (3, 40000000, 10000000, 7000000))) + "\n")
+        server.write_text("\n".join(
+            f"P65ServerMetrics frame={n} valid=1 wall_ns={wall} "
+            f"apply_thread_cpu_ns={cpu}"
+            for n, wall, cpu in ((2, 18000000, 17000000 if overlapping else 3000000),
+                                 (3, 35000000, 31000000 if overlapping else 6000000))
+            if not missing_server or n != 3) + "\n")
+        return client, server
+
+    def test_paired_percentiles_and_shares(self):
+        with tempfile.TemporaryDirectory() as directory:
+            result = summarize(*self.logs(Path(directory)), warmup=1)
+        self.assertEqual((result["first_frame"], result["last_frame"], result["frames"]),
+                         (2, 3, 2))
+        self.assertEqual(result["stats"]["transport_wait_ms"],
+                         {"avg": 5.5, "p50": 4.0, "p99": 7.0})
+        self.assertEqual(result["share_percent_of_client_present"]["transport_wait_ms"],
+                         100 * 11000000 / 60000000)
+
+    def test_missing_server_frame_is_error(self):
+        with tempfile.TemporaryDirectory() as directory:
+            with self.assertRaisesRegex(ValueError, "missing matched frames"):
+                summarize(*self.logs(Path(directory), missing_server=True), warmup=1)
+
+    def test_parallel_cpu_is_allowed(self):
+        with tempfile.TemporaryDirectory() as directory:
+            result = summarize(*self.logs(Path(directory), overlapping=True), warmup=1)
+        self.assertGreater(sum(result["share_percent_of_client_present"].values()), 100)
+
+
+if __name__ == "__main__":
+    unittest.main()