[Perf] (MG_Remote/Transport): the reply-wait metric gains a per-op tally, and the measurement says which rows actually buy the waits

- The handoff measured the total (55,428 reply waits in one Minecraft frame) and left "which rows"
  open: a total is the same number whether the waits are uploads, creates or parameter sets, and
  those three have different fixes - so knowing the split is what says whether P12 item B is the
  whole story or a fraction of one.
- LinkMetricsBeginReply now takes the op and Window carries a per-op tally, so
  LinkMetricsReplyWaitsFor(op) reads back what the P65LinkMetrics line prints as `wait_replies`
  split by row. Same activation as everything else in that file (MOBILEGL_PIPE_STATS=1), one array
  in the window, and a value past the table lands on the last slot so the per-op numbers still SUM
  to the total.
- RemoteClientControls.ReplyWaitsByOpForAnAtlasShapedLoad drives the two shapes a stitcher
  produces and prints the table: an atlas stitch (64 glTexSubImage2D into one level) goes from 64
  reply waits to ZERO, and a texture per sprite (64 textures, 256 glTexParameteri) from 384 to 320
  with the remainder all ResourceCreate 76 / ResourceRespecify 130 / SetTextureParams 130.
- It prints before it asserts: the table is the point of the case, and a gate that fired first
  would hide the numbers on exactly the run that needs them - the red one. The assertions are the
  structural ones that can be wrong about the mechanism (upload rows zero, the other rows not, the
  per-op tally summing to the total); the numbers themselves are recorded, not gated.
- The A/B is a real run: with the wire-arm arm test removed the case goes red with the full table,
  which is where the 64 and 384 above come from.

MobileGL/MG_Remote/Client/ClientSession.cpp

@@ -1877,7 +1877,10 @@ namespace MobileGL::MG_Remote::Client {
         // Publish the head, record submittedSeq, THEN ring - in that order, which is
         // SessionProducer's one job and RingTest.cpp:446's pin. Notify-then-publish loses the
         // wakeup.
-        const auto replyMetricStart = Transport::LinkMetricsBeginReply(ownsReplySlot);
+        // The op goes with the tally (P12): a total cannot say whether the waits are uploads,
+        // creates or parameter sets, and those three have different fixes.
+        const auto replyMetricStart =
+            Transport::LinkMetricsBeginReply(ownsReplySlot, static_cast<std::uint32_t>(op));
         m_producer.PublishAndNotify(seq);
         if (op == MG_Pipe::MGPWireOp::Present) {
             // A credit is permission to run ahead, not permission to retain the

MobileGL/MG_Remote/Transport/LinkMetrics.cpp

@@ -15,6 +15,9 @@ namespace {
     struct Window {
         std::uint64_t waitReplies = 0, samples = 0, replyNs = 0, stageBytes = 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.
+        std::array<std::uint64_t, LinkMetricsMaxOps> waitRepliesByOp{};
     };
     // The client already has exactly one record producer. The server never writes these.
     struct Metrics {
@@ -73,9 +76,11 @@ void LinkMetricsBegin() {
     metrics.started = metrics.lastFrame = ClockNs();
     metrics.cpuStarted = metrics.cpuLast = ThreadCpuNs();
 }
-std::uint64_t LinkMetricsBeginReply(bool wantsReply) {
+std::uint64_t LinkMetricsBeginReply(bool wantsReply, std::uint32_t op) {
     if (!metrics.active || !wantsReply) return 0;
+    const std::uint32_t slot = op < LinkMetricsMaxOps ? op : LinkMetricsMaxOps - 1;
     ++metrics.current.waitReplies; ++metrics.total.waitReplies;
+    ++metrics.current.waitRepliesByOp[slot]; ++metrics.total.waitRepliesByOp[slot];
     return ClockNs();
 }
 void LinkMetricsReplyApplied(std::uint64_t startedNs) {
@@ -89,6 +94,9 @@ void LinkMetricsReplyApplied(std::uint64_t startedNs) {
     }
 }
 std::uint64_t LinkMetricsReplyWaits() { return metrics.total.waitReplies; }
+std::uint64_t LinkMetricsReplyWaitsFor(std::uint32_t op) {
+    return metrics.total.waitRepliesByOp[op < LinkMetricsMaxOps ? op : LinkMetricsMaxOps - 1];
+}
 void LinkMetricsStageBytes(std::uint64_t bytes) {
     if (!metrics.active) return;
     metrics.current.stageBytes += bytes; metrics.total.stageBytes += bytes;

MobileGL/MG_Remote/Transport/LinkMetrics.h

@@ -6,7 +6,15 @@ namespace MobileGL::MG_Remote::Transport {
     // Disabled sessions never read a clock. Samples are a fixed 32-bucket histogram.
     void LinkMetricsBegin();
     void LinkMetricsEnd();
-    std::uint64_t LinkMetricsBeginReply(bool wantsReply);
+    // PER-OP: WHICH ROWS BOUGHT THE WAITS. The handoff measured the total (55,428 reply waits in
+    // one frame) and left "which rows" open, and a total cannot answer it: an upload's wait is what
+    // P12 item B removed, a create's is per resource and a parameter set's is per call - three
+    // different fixes, and the split is what says whether B is the whole fix or a fraction of one.
+    // `op` is an MGPWireOp value passed as an unsigned so this file keeps its single include; a
+    // value past the table lands on the last slot rather than being dropped, so a reader gets a
+    // number that does not add up instead of a silent zero.
+    inline constexpr std::uint32_t LinkMetricsMaxOps = 128;
+    std::uint64_t LinkMetricsBeginReply(bool wantsReply, std::uint32_t op);
     void LinkMetricsReplyApplied(std::uint64_t startedNs);
     void LinkMetricsStageBytes(std::uint64_t bytes);
 
@@ -17,6 +25,7 @@ namespace MobileGL::MG_Remote::Transport {
     // exactly why a case that asserts 0 for one row must ALSO assert that a reply-owning row
     // still counts, or it would pass on a session where nothing was counted at all.
     std::uint64_t LinkMetricsReplyWaits();
+    std::uint64_t LinkMetricsReplyWaitsFor(std::uint32_t op);
     void LinkMetricsPresent();
     void LinkMetricsServerPresent(std::uint64_t serial);
 }

MobileGL/MG_Test/Wire/RemoteClientControls.inc

@@ -910,4 +910,93 @@ TEST(RemoteClientControls, TextureUploadsStopOwningAReplyOnTheWireArm) {
     ExpectChildSuccess(child);
 }
 
+// =====================================================================================
+// P12 item B: WHICH ROWS BUY THE WAITS - the measurement the handoff left open (its 5.4)
+// =====================================================================================
+//
+// A total cannot answer it. 55,428 reply waits in one frame is the same number whether they are
+// uploads (item B), creates (per resource) or parameter sets (per call), and those three have
+// different fixes - so the split is what says whether B is the whole story or a fraction of one.
+// An atlas-shaped load makes it visible: one sprite per texture, one glTexImage2D and four
+// glTexParameteri, which is the call shape an atlas stitch has.
+//
+// A MEASUREMENT WITH A STRUCTURAL GATE UNDER IT, not a performance gate: the assertions are the
+// ones that can be WRONG about the mechanism (the upload rows contribute zero after item B, the
+// create / respecify / parameter rows do not, and the per-op tally sums to the total), and the
+// numbers themselves are printed for the register.
+TEST(RemoteClientControls, ReplyWaitsByOpForAnAtlasShapedLoad) {
+    const auto child = RunInChild([] {
+        ::setenv("MOBILEGL_PIPE_STATS", "1", 1);
+        CapsPeer backend;
+        StartControlSession(&backend);
+        MG_Config::Features.PipePush |= kMGPipeSubsystemResources |
+                                        kMGPipeSubsystemTextureResources | kMGPipeSubsystemSamplers;
+        MG_State::pGLContext = MakeUnique<MG_State::GLState::GLContext>();
+        constexpr int kSprites = 64;
+        Uint8 pixels[16 * 16 * 4];
+        std::memset(pixels, 0x5a, sizeof(pixels));
+
+        // SHAPE 1 - THE ATLAS STITCH, which is what item B was aimed at: ONE texture and many
+        // small writes into it. If B works, this shape's wait count stops tracking the upload
+        // count at all.
+        GLuint atlas = 0;
+        MG_Impl::GLImpl::GenTextures(1, &atlas);
+        MG_Impl::GLImpl::BindTexture(0x0DE1, atlas);
+        MG_Impl::GLImpl::TexImage2D(0x0DE1, 0, 0x8058, 256, 256, 0, 0x1908, 0x1401, nullptr);
+        MG_Impl::GLImpl::TexParameteri(0x0DE1, 0x2801, 0x2601);
+        MGP_FILL(DrawArrays);
+        const Uint64 afterAtlasSetup = Transport::LinkMetricsReplyWaits();
+        for (int i = 0; i < kSprites; ++i) {
+            MG_Impl::GLImpl::TexSubImage2D(0x0DE1, 0, (i % 8) * 16, (i / 8) * 16, 16, 16, 0x1908,
+                                           0x1401, pixels);
+            MGP_FILL(DrawArrays); // the drain point
+        }
+        const Uint64 atlasWaits = Transport::LinkMetricsReplyWaits() - afterAtlasSetup;
+        std::printf("shape 1 atlas stitch: %d sub-uploads into one level -> wait_replies=%llu\n",
+                    kSprites, static_cast<unsigned long long>(atlasWaits));
+        const Uint64 atlasWaitsBefore = Transport::LinkMetricsReplyWaits();
+
+        // SHAPE 2 - A TEXTURE PER SPRITE, the other thing a stitcher can do (and what the mipmap
+        // and dynamic-texture paths look like): every sprite pays a storage definition and its
+        // own parameter set.
+        for (int i = 0; i < kSprites; ++i) {
+            GLuint texture = 0;
+            MG_Impl::GLImpl::GenTextures(1, &texture);
+            MG_Impl::GLImpl::BindTexture(0x0DE1, texture);
+            MG_Impl::GLImpl::TexImage2D(0x0DE1, 0, 0x8058, 16, 16, 0, 0x1908, 0x1401, pixels);
+            MG_Impl::GLImpl::TexParameteri(0x0DE1, 0x2802, 0x2901); // WRAP_S = CLAMP_TO_EDGE
+            MG_Impl::GLImpl::TexParameteri(0x0DE1, 0x2803, 0x2901); // WRAP_T
+            MG_Impl::GLImpl::TexParameteri(0x0DE1, 0x2801, 0x2601); // MIN_FILTER = LINEAR
+            MG_Impl::GLImpl::TexParameteri(0x0DE1, 0x2800, 0x2601); // MAG_FILTER = LINEAR
+            MGP_FILL(DrawArrays); // the drain point
+        }
+        // PRINT FIRST, ASSERT SECOND: the table is the point of the case, and a gate that fired
+        // before it would hide the numbers on exactly the run that needs them - the red one.
+        std::printf("shape 2 a texture per sprite: %d textures -> wait_replies=%llu\n", kSprites,
+                    static_cast<unsigned long long>(Transport::LinkMetricsReplyWaits() - atlasWaitsBefore));
+        for (Uint32 op = 1; op < Transport::LinkMetricsMaxOps; ++op) {
+            const Uint64 waits = Transport::LinkMetricsReplyWaitsFor(op);
+            if (waits != 0)
+                std::printf("  op=%-24s waits=%llu\n", Codec::WireOpName(static_cast<MGPWireOp>(op)),
+                            static_cast<unsigned long long>(waits));
+        }
+        std::fflush(nullptr);
+        // THE STRUCTURAL GATE. The upload row must be free (item B) and every other row this load
+        // drives must NOT be, or the case would also pass on a session that counted nothing.
+        if (Transport::LinkMetricsReplyWaitsFor(static_cast<Uint32>(MGPWireOp::ResourceSubData)) != 0)
+            ::_exit(143);
+        if (Transport::LinkMetricsReplyWaitsFor(static_cast<Uint32>(MGPWireOp::ResourceCreate)) == 0)
+            ::_exit(144);
+        if (Transport::LinkMetricsReplyWaitsFor(static_cast<Uint32>(MGPWireOp::ResourceRespecify)) == 0)
+            ::_exit(145);
+        if (Transport::LinkMetricsReplyWaitsFor(static_cast<Uint32>(MGPWireOp::SetTextureParams)) == 0)
+            ::_exit(146);
+        Uint64 sum = 0;
+        for (Uint32 op = 0; op < Transport::LinkMetricsMaxOps; ++op)
+            sum += Transport::LinkMetricsReplyWaitsFor(op);
+        if (sum != Transport::LinkMetricsReplyWaits()) ::_exit(147);
+        ClientSessionInstance().Stop();
+    });
+    ExpectChildSuccess(child);
+}
 #endif