From a2c1f5f292fbc933f139a44b3ea8c19ee8a01c46 Mon Sep 17 00:00:00 2001 From: BZLZHH Date: Sat, 26 Sep 2026 00:47:06 +0800 Subject: [PATCH] [Fix] (MG_Remote/Transport): StreamLink::ReadReply answers by seq instead of "is it the newest" - the contract ShmLink already kept, and the reason a deferred answer could never be read back ILink::ReadReply takes a SEQ (ILink.h:285) and ShmLink implements it against a ReplySlotPool, so any answer the pool still holds is readable. StreamLink kept ONE reply - replySeq/replyStatus/reply - and then asked the wrong question of it: x.replyReady.wait(lock, [&]{ return closed || x.replySeq >= seq || appliedSeq >= seq; }); if (x.replySeq != seq) return ... MOBILEGL_ERR_PROTOCOL_MISMATCH; The wait is written as "at least this seq"; the check is "exactly this seq". Every answer older than the newest was therefore unreadable, which nothing noticed because every caller read the answer it had just waited for - i.e. always the newest one. It is the transport half of the P12 create window's failure: the window defers up to four answers and collects them in order, and this link could not hand one back. THE RING, AT THE POOL'S OWN DEPTH. kReplyRing = kDefaultReplySlotCount (8), because that is the pool this link stands in for: an answer a reader has not taken by the time eight newer ones have arrived is one the shm arm has already overwritten too, so the two transports now lose the same answers and no more. Vectors are sized per reply, so an idle link pays eight empty ones. This does NOT make the create window work - on the device it changed nothing, because the pool itself is 8 slots against 59,673 replies in one frame (see the next commit). It is kept because the two transports disagreeing about ILink's contract is a bug in its own right, and because the single slot silently failed any read that raced a newer reply. Gate: StreamLinkTest.AnOlderReplyIsStillReadableByItsSeq posts three replies and reads them oldest-first - the order the window's drain takes them in. --- MobileGL/MG_Remote/Transport/StreamLink.cpp | 60 +++++++++++++++++---- MobileGL/MG_Test/Wire/StreamLinkTest.cpp | 38 +++++++++++++ 2 files changed, 88 insertions(+), 10 deletions(-) diff --git a/MobileGL/MG_Remote/Transport/StreamLink.cpp b/MobileGL/MG_Remote/Transport/StreamLink.cpp index ea093746..59f9b461 100644 --- a/MobileGL/MG_Remote/Transport/StreamLink.cpp +++ b/MobileGL/MG_Remote/Transport/StreamLink.cpp @@ -6,6 +6,7 @@ #include "ReplySlot.h" #include "Framing.h" #include +#include #include #include #include @@ -91,9 +92,33 @@ namespace MobileGL::MG_Remote::Transport { std::mutex writeMutex, replyMutex; std::condition_variable replyReady; std::vector stage; - std::vector reply, readReply; + std::vector readReply; std::uint64_t replySeq = 0, sentCmd = 0, receivedCmd = 0, sentEvent = 0, receivedEvent = 0; - std::int32_t replyStatus = 0; + // ---- THE RECENT-REPLY RING (P12) ------------------------------------------------- + // + // ILink::ReadReply takes a SEQ, and until P12 that was decoration: every caller read the + // answer it had just waited for, which is always the newest one, so a single slot answered + // every question and `replySeq != seq` could stand in for "no such answer". The create + // window is the first caller that reads an OLDER seq - it defers up to four answers and + // takes them in order - and against a single slot every one of those reads fails + // PROTOCOL_MISMATCH for ever. What that looked like on the device, measured: the window + // filled to 4 and STAYED there, `refusals=0` because no answer was ever read, every one of + // the frame's 4,536 creates fell through to the blocking path, and the wall clock did not + // move (29.0 s against the 30.1 s of the same binary with the window off). The deferral + // was pure overhead: four answers held unread, plus one failed drain attempt per record. + // + // BOUNDED AT kDefaultReplySlotCount, which is the pool ShmLink stands in for: an answer a + // reader has not taken by the time eight newer ones have arrived is one the shm arm has + // already overwritten too, so the two transports now lose the same answers and no more. + // The byte vectors are sized per reply, so an idle link pays eight empty ones. + static constexpr std::size_t kReplyRing = 8; + struct StoredReply { + std::uint64_t seq = 0; + std::int32_t status = 0; + std::vector bytes; + }; + std::array replies{}; + std::size_t replyNext = 0; std::uint64_t lastProgressSeq = 0; std::chrono::steady_clock::time_point lastProgress = std::chrono::steady_clock::now(); LinkArena records{}, events{}; @@ -309,8 +334,15 @@ namespace MobileGL::MG_Remote::Transport { std::lock_guard lock(replyMutex); if (a <= replySeq) Corrupt("reply sequence moved backwards"); replySeq = a; - replyStatus = static_cast(static_cast(b)); - reply.assign(payload, payload + size); + // RING, NOT REPLACE. replySeq stays the newest and stays monotone - + // the wait predicate below is written in terms of it - while the + // answers themselves are kept for the readers that come back for an + // older one (the window's drain, StreamLink.cpp's own note). + StoredReply& slot = replies[replyNext]; + replyNext = (replyNext + 1) % kReplyRing; + slot.seq = a; + slot.status = static_cast(static_cast(b)); + slot.bytes.assign(payload, payload + size); } replyReady.notify_all(); producer.Notify(); @@ -557,12 +589,20 @@ namespace MobileGL::MG_Remote::Transport { return x.closed.load(std::memory_order_acquire) || x.replySeq >= seq || x.C().Progress.appliedSeq.load(std::memory_order_acquire) >= seq; }); - if (x.replySeq != seq) return x.closed.load() ? MOBILEGL_ERR_TRANSPORT_CLOSED : MOBILEGL_ERR_PROTOCOL_MISMATCH; - x.readReply = x.reply; - *status = x.replyStatus; - *size = x.readReply.size(); - *p = x.readReply.data(); - return MOBILEGL_OK; + // BY SEQ, WHICH IS WHAT THE PARAMETER HAS ALWAYS SAID (P12). The equality this replaces + // was true only while every caller read the newest answer; it is what made a deferred + // answer unreadable and the create window inert on this transport. + for (const auto& slot : x.replies) { + if (slot.seq != seq) continue; + // Copied into the shared scratch so the pointer outlives the lock, exactly as the + // single-slot version did: the caller memcpys under its own frame. + x.readReply = slot.bytes; + *status = slot.status; + *size = x.readReply.size(); + *p = x.readReply.data(); + return MOBILEGL_OK; + } + return x.closed.load() ? MOBILEGL_ERR_TRANSPORT_CLOSED : MOBILEGL_ERR_PROTOCOL_MISMATCH; } std::uint64_t StreamLink::EventPublishedHead() const { return m_impl->memory ? m_impl->E().cmdHead.load(std::memory_order_acquire) : 0; diff --git a/MobileGL/MG_Test/Wire/StreamLinkTest.cpp b/MobileGL/MG_Test/Wire/StreamLinkTest.cpp index 7dab5059..df2f8f3e 100644 --- a/MobileGL/MG_Test/Wire/StreamLinkTest.cpp +++ b/MobileGL/MG_Test/Wire/StreamLinkTest.cpp @@ -137,6 +137,44 @@ namespace { EXPECT_EQ(value, 22u); } + // ---- P12: AND THE REPLY A READER COMES BACK FOR IS READABLE BY ITS SEQ ---- + // + // The case above reads the NEWEST answer, which is the only one a single-slot link could ever + // hand back. The create window defers up to four answers and takes them in order, so the OLDEST + // of them has to still be readable after newer ones have arrived - and against a + // replace-on-arrival slot it was not: every drain read failed PROTOCOL_MISMATCH for ever, the + // window stayed full, and every create fell back to the blocking path. Measured on the device + // before this: 4,536 creates, 4,536 of them blocking, `pending` pinned at 4, refusals 0, wall + // clock 29.0 s against 30.1 s with the window off. This is the transport half of that fix. + TEST_F(StreamPair, AnOlderReplyIsStillReadableByItsSeq) { + Queue(1, 1); + Queue(2, 2); + Queue(3, 3); + ASSERT_EQ(client.Flush(), MOBILEGL_OK); + ASSERT_EQ(apply.WaitForWork(2000), SessionWait::Reached); + for (std::uint64_t seq = 1; seq <= 3; ++seq) { + ASSERT_TRUE(apply.ApplyOne([&](const RingRecordView&) { + const std::uint64_t value = seq * 11; + EXPECT_EQ(server.PostReply(seq, 0, &value, sizeof value), MOBILEGL_OK); + })); + } + apply.RetireThrough(3); + ASSERT_EQ(server.FlushProgress(), MOBILEGL_OK); + ASSERT_EQ(publish.WaitForApplied(3, 2000), SessionWait::Reached); + // OLDEST FIRST, which is the order the window's drain takes them in. + for (std::uint64_t seq = 1; seq <= 3; ++seq) { + std::int32_t status = -1; + const void* p = nullptr; + std::uint64_t size = 0; + ASSERT_EQ(client.ReadReply(seq, &status, &p, &size), MOBILEGL_OK) << "seq " << seq; + ASSERT_EQ(size, sizeof(std::uint64_t)) << "seq " << seq; + std::uint64_t value = 0; + std::memcpy(&value, p, sizeof value); + EXPECT_EQ(value, seq * 11) << "seq " << seq; + EXPECT_EQ(status, 0) << "seq " << seq; + } + } + TEST_F(StreamPair, EventDrainReturnsCreditAndClearsRemoteFullLatch) { EventRingProducer events(serverMemory.EventControl(), serverMemory.CmdControl(), serverMemory.EventRingBase(), 1024);