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);