swung0x48/MobileGL
1
0
Fork 0
mirror of https://github.com/MobileGL-Dev/MobileGL synced 2026-11-07 01:53:24 +09:00
Code Issues Packages Projects Releases Wiki Activity

[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

Browse Source 
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.
This commit is contained in:
BZLZHH
2026-09-26 00:47:06 +08:00
parent 82ed1df90e
commit a2c1f5f292
2 changed files with 88 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MobileGL/MG_Remote/Transport/StreamLink.cpp
+50 -10
View File
@@ -6,6 +6,7 @@
#include "ReplySlot.h"
#include "Framing.h"
#include <algorithm>
#include <array>
#include <chrono>
#include <condition_variable>
#include <cstring>
@@ -91,9 +92,33 @@ namespace MobileGL::MG_Remote::Transport {
std::mutex writeMutex, replyMutex;
std::condition_variable replyReady;
std::vector<LinkSpan> stage;
std::vector<std::uint8_t> reply, readReply;
std::vector<std::uint8_t> 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<std::uint8_t> bytes;
};
std::array<StoredReply, kReplyRing> 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<std::mutex> lock(replyMutex);
if (a <= replySeq) Corrupt("reply sequence moved backwards");
replySeq = a;
replyStatus = static_cast<std::int32_t>(static_cast<std::uint32_t>(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<std::int32_t>(static_cast<std::uint32_t>(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;
MobileGL/MG_Test/Wire/StreamLinkTest.cpp
+38
View File
@@ -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);