swung0x48/MobileGL
1
0
Fork 0
mirror of https://github.com/MobileGL-Dev/MobileGL synced 2026-09-17 00:28:31 +09:00
Code Issues Packages Projects Releases Wiki Activity

[Test] (MG_Test, Wire): pin the session pair - twenty thousand records across two real threads, one case per watermark rule, the kRecPad rule against the session's own counter, the reply slot's seq stamp and a shutdown whose join is bounded at five seconds so a lost wakeup fails red

Browse Source
This commit is contained in:
swung0x48
2026-09-11 14:13:20 -04:00
parent 558fb210dc
commit 1f953c09ea
2 changed files with 772 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MobileGL/MG_Test/Wire/CMakeLists.txt
+3
View File
@@ -9,6 +9,9 @@ set(MOBILEGL_WIRE_TESTS
RingTest RingTest
InProcessTransportTest InProcessTransportTest
ProtocolSmokeTest ProtocolSmokeTest
# P5 s1: the ring-owning session pair - four ShmSegment-backed segments, the five
# watermarks with real writers, the reply slot pool and the event ring.
SessionTest
) )
if (NOT WIN32) if (NOT WIN32)
MobileGL/MG_Test/Wire/SessionTest.cpp
+769
View File
@@ -0,0 +1,769 @@
// MobileGL - MobileGL/MG_Test/Wire/SessionTest.cpp
// Copyright (c) 2025-2026 MobileGL-Dev
// Licensed under the GNU Lesser General Public License v3.0:
// https://www.gnu.org/licenses/gpl-3.0.txt
// https://www.gnu.org/licenses/lgpl-3.0.txt
// SPDX-License-Identifier: LGPL-3.0-only
// End of Source File Header
// The ring-owning session pair: four ShmSegment-backed segments, two real threads over the
// SEG_CMD ring, the five watermarks with real writers, the SEG_REPLY slot pool, the SEG_EVENT
// reverse channel, and a shutdown that a lost wakeup turns RED rather than hanging.
//
// WHY THIS SUITE EXISTS ALONGSIDE RingTest. RingTest pins the ring's own mechanics against a
// fixture whose control page is on the stack and whose byte area is a std::vector, and it pins
// the five watermark RULES against nobody, because until P5 nothing in the tree wrote one
// (every watermark was declared, zeroed by InitRingControl and written by no code at all).
// This suite pins the WRITERS: SessionProducer, SessionConsumer and the Watermark namespace are
// the only things that advance them, and they do it over memory that came from ShmSegment.
//
// AND THAT LAST PART IS THE POINT. `inproc` allocating its rings with new[] would work, would
// be shorter, and would move every question about mapping, alignment, size rounding and
// lifetime into P6 - onto the day the second process appears. So the session uses ShmSegment in
// both delivery modes and SessionSegmentsAreRealSharedMemory below is the mechanical check that
// it still does.
#include <MG_Remote/Transport/Doorbell.h>
#include <MG_Remote/Transport/EventRing.h>
#include <MG_Remote/Transport/InProcessTransport.h>
#include <MG_Remote/Transport/ReplySlot.h>
#include <MG_Remote/Transport/Ring.h>
#include <MG_Remote/Transport/RoleMemory.h>
#include <MG_Remote/Transport/SessionRings.h>
#include <gtest/gtest.h>
#include <atomic>
#include <chrono>
#include <cstring>
#include <memory>
#include <string>
#include <thread>
#include <vector>
using namespace MobileGL::MG_Remote::Transport;
namespace {
// Small enough to be cheap in CI, large enough that 20 000 sixteen-byte records wrap the
// command ring many times over - which is what puts the kRecPad rule under load rather
// than under a contrived single wrap.
SessionSegmentSizes TestSizes() {
SessionSegmentSizes sizes;
sizes.CmdBytes = 64ull * 1024; // -> 32 KiB ring after the control page
sizes.StageBytes = 64ull * 1024; // -> 64 KiB, no control page of its own
sizes.ReplyBytes = 64ull * 1024; // -> 8 slots of 8 KiB
sizes.EventBytes = 32ull * 1024; // -> 16 KiB ring after the control page
sizes.ReplySlotCount = 8;
return sizes;
}
// One session's worth of everything, wired the way ClientSession and ServerSession wire it:
// the server owns the segments, the client attaches to the same mapping, and the two
// doorbells come from the transport while the RINGS come from here.
struct SessionFixture {
std::unique_ptr<InProcessTransport> clientTransport;
std::unique_ptr<InProcessTransport> serverTransport;
SessionSegments serverSegments;
SessionSegments clientSegments;
RingProducer cmdProducer;
RingProducer stageProducer;
RingConsumer cmdConsumer;
SessionProducer producer;
SessionConsumer consumer;
ReplySlotPool replies;
EventRingProducer eventOut;
EventRingConsumer eventIn;
bool Build(const SessionSegmentSizes& sizes) {
InProcessTransport::CreatePair(clientTransport, serverTransport);
if (serverSegments.Create(sizes, MemoryRole::Server) != MOBILEGL_OK) {
return false;
}
if (clientSegments.AttachInProcess(serverSegments, MemoryRole::Client) != MOBILEGL_OK) {
return false;
}
RingControl* control = serverSegments.CmdControl();
cmdProducer = RingProducer(control, clientSegments.CmdRingBase(),
clientSegments.CmdRingCapacity(), RingCursorSet::Cmd);
stageProducer = RingProducer(control, clientSegments.StageBase(),
clientSegments.StageCapacity(), RingCursorSet::Stage);
cmdConsumer = RingConsumer(control, serverSegments.CmdRingBase(),
serverSegments.CmdRingCapacity(), RingCursorSet::Cmd);
if (!cmdProducer.Valid() || !stageProducer.Valid() || !cmdConsumer.Valid()) {
return false;
}
// PeerDoorbell is the bell the OTHER end parks on; SelfDoorbell is this end's own.
// Which is which is the session's knowledge, never ITransport's (contract §3.9).
producer.Attach(control, &cmdProducer, &stageProducer, &clientTransport->PeerDoorbell(),
&clientTransport->SelfDoorbell(), kDefaultSpinUs);
consumer.Attach(control, &cmdConsumer, &serverTransport->PeerDoorbell(),
&serverTransport->SelfDoorbell(), kDefaultSpinUs);
replies = ReplySlotPool(serverSegments.ReplyBase(), serverSegments.ReplyBytes(),
serverSegments.ReplySlotCount());
replies.Clear();
eventOut = EventRingProducer(serverSegments.EventControl(), control,
serverSegments.EventRingBase(),
serverSegments.EventRingCapacity());
eventIn = EventRingConsumer(clientSegments.EventControl(), control,
clientSegments.EventRingBase(),
clientSegments.EventRingCapacity(),
clientSegments.EventSegmentBase());
return replies.Valid() && eventOut.Valid() && eventIn.Valid();
}
RingControl& Control() { return *serverSegments.CmdControl(); }
};
} // namespace
// ---------------------------------------------------------------------------
// The segments themselves
// ---------------------------------------------------------------------------
// `inproc` must not quietly become new[]. A descriptor (POSIX) or a native handle (Windows) is
// the mechanical difference between a session whose spawn path is the same code and one whose
// spawn path is written for the first time in P6.
TEST(SessionTest, SessionSegmentsAreRealSharedMemoryAndNotAHeapAllocation) {
SessionSegments segments;
ASSERT_EQ(segments.Create(TestSizes(), MemoryRole::Server), MOBILEGL_OK);
EXPECT_TRUE(segments.Valid());
#if !defined(_WIN32)
EXPECT_GE(segments.DescriptorFor(SessionSegmentSlot::Cmd), 0);
EXPECT_GE(segments.DescriptorFor(SessionSegmentSlot::Stage), 0);
EXPECT_GE(segments.DescriptorFor(SessionSegmentSlot::Reply), 0);
EXPECT_GE(segments.DescriptorFor(SessionSegmentSlot::Event), 0);
#endif
// Both control pages start initialised, with the two generations at 1 and every watermark
// at 0 - the two conventions are opposite on purpose (Ring.h).
ASSERT_NE(segments.CmdControl(), nullptr);
ASSERT_NE(segments.EventControl(), nullptr);
EXPECT_EQ(segments.CmdControl()->ringGeneration.load(), 1u);
EXPECT_EQ(segments.EventControl()->ringGeneration.load(), 1u);
EXPECT_EQ(segments.CmdControl()->appliedSeq.load(), 0u);
segments.Close();
EXPECT_FALSE(segments.Valid());
}
// The arithmetic the whole geometry rests on, and the one place CONTRACT-P5 §5's "8 MiB caps
// one record at 4 MiB" is corrected: the control page sits at the HEAD of SEG_CMD and the ring
// capacity must be a power of two, so an 8 MiB segment yields a 4 MiB ring and a 2 MiB record.
TEST(SessionTest, TheRingIsTheLargestPowerOfTwoLeftAfterTheControlPage) {
EXPECT_EQ(LargestPowerOfTwoAtMost(0u), 0u);
EXPECT_EQ(LargestPowerOfTwoAtMost(1u), 1u);
EXPECT_EQ(LargestPowerOfTwoAtMost(4095u), 2048u);
EXPECT_EQ(LargestPowerOfTwoAtMost(4096u), 4096u);
EXPECT_EQ(LargestPowerOfTwoAtMost(4097u), 4096u);
// The four contract sizes, which ProtocolSmokeTest.cpp:72 pins on the wire.
constexpr std::uint64_t kCmd = 8ull * 1024 * 1024;
constexpr std::uint64_t kEvent = 256ull * 1024;
EXPECT_EQ(RingCapacityForSegment(kCmd), 4ull * 1024 * 1024);
EXPECT_EQ(RingCapacityForSegment(kEvent), 128ull * 1024);
// ... and therefore the real cap on one record, which R-10 obliges the codec to prove it
// never approaches. Half of the ring, not half of the segment.
RingControl control{};
InitRingControl(control);
std::vector<std::uint8_t> bytes(static_cast<std::size_t>(RingCapacityForSegment(kCmd)));
RingProducer producer(&control, bytes.data(), RingCapacityForSegment(kCmd), RingCursorSet::Cmd);
ASSERT_TRUE(producer.Valid());
EXPECT_EQ(producer.MaxRecordBytes(), 2ull * 1024 * 1024);
// A segment that cannot hold the control page plus the smallest ring has NO ring, rather
// than a ring of some rounded-down nonsense.
EXPECT_EQ(RingCapacityForSegment(sizeof(RingControl)), 0u);
EXPECT_EQ(RingCapacityForSegment(sizeof(RingControl) + 8), 0u);
}
// The default geometry really allocates, and the announced sizes are the MAPPING sizes - what
// a spawn peer must map - not the ring capacity inside them.
TEST(SessionTest, TheDefaultGeometryIsTheFourContractSizes) {
SessionSegments segments;
ASSERT_EQ(segments.Create(SessionSegmentSizes{}, MemoryRole::Server), MOBILEGL_OK);
EXPECT_EQ(segments.AnnouncedSize(SessionSegmentSlot::Cmd), 8ull * 1024 * 1024);
EXPECT_EQ(segments.AnnouncedSize(SessionSegmentSlot::Stage), 32ull * 1024 * 1024);
EXPECT_EQ(segments.AnnouncedSize(SessionSegmentSlot::Reply), 8ull * 1024 * 1024);
EXPECT_EQ(segments.AnnouncedSize(SessionSegmentSlot::Event), 256ull * 1024);
EXPECT_EQ(segments.CmdRingCapacity(), 4ull * 1024 * 1024);
// SEG_STAGE carries no control page: RingControl holds both cursor triples, so the whole
// segment is ring and 32 MiB is already a power of two.
EXPECT_EQ(segments.StageCapacity(), 32ull * 1024 * 1024);
segments.Close();
}
// The ledger is per role and DOES double-count under inproc, deliberately: the two roles map
// the same pages here and will not under spawn, so the per-role numbers are what t1 subtracts
// with and a silently deduplicated total would hide exactly that difference.
TEST(SessionTest, TheMemoryLedgerIsPerRoleAndIsReleasedOnClose) {
const std::uint64_t clientBefore = LedgerMappedBytes(MemoryRole::Client);
const std::uint64_t serverBefore = LedgerMappedBytes(MemoryRole::Server);
{
SessionFixture session;
ASSERT_TRUE(session.Build(TestSizes()));
const std::uint64_t mapped = session.serverSegments.MappedBytes();
EXPECT_GT(mapped, 0u);
EXPECT_EQ(LedgerMappedBytes(MemoryRole::Server), serverBefore + mapped);
EXPECT_EQ(LedgerMappedBytes(MemoryRole::Client), clientBefore + mapped);
const RoleMemorySample sample = SampleRoleMemory(MemoryRole::Client);
EXPECT_EQ(sample.MappedSegmentBytes, clientBefore + mapped);
#if defined(__linux__) || defined(__ANDROID__)
// VmHWM is the PROCESS's high-water mark, so it is the same number for both roles and
// is only meaningful beside the ledger - which is why RoleMemorySample carries both.
EXPECT_GT(sample.PeakRssBytes, 0u);
EXPECT_GE(sample.PeakRssBytes, sample.CurrentRssBytes);
#endif
}
EXPECT_EQ(LedgerMappedBytes(MemoryRole::Client), clientBefore);
EXPECT_EQ(LedgerMappedBytes(MemoryRole::Server), serverBefore);
}
// ---------------------------------------------------------------------------
// Two real threads, twenty thousand records
// ---------------------------------------------------------------------------
// The acceptance run. A 32 KiB command ring and 16-byte records means this wraps about ten
// times, so the wrap filler is exercised under load rather than in one contrived case - and
// the invariant checked at the end is that appliedSeq counted the RECORDS and not the fillers.
TEST(SessionTest, TwoThreadsMoveTwentyThousandRecordsAndAgreeOnEveryOne) {
SessionFixture session;
ASSERT_TRUE(session.Build(TestSizes()));
constexpr std::uint32_t kRecords = 20000;
std::atomic<bool> ok{true};
std::atomic<std::uint32_t> consumed{0};
std::thread apply([&] {
std::uint32_t seen = 0;
while (seen < kRecords && ok.load()) {
const SessionWait woke = session.consumer.WaitForWork(5000);
if (woke == SessionWait::ShutDown) {
return;
}
if (woke == SessionWait::TimedOut) {
ok.store(false); // a lost wakeup, or the producer stalled: RED, never a hang
return;
}
bool corrupt = false;
while (session.consumer.ApplyOne(
[&](const RingRecordView& view) {
std::uint32_t value = 0;
std::memcpy(&value, view.payload, sizeof(value));
if (value != seen) {
ok.store(false);
}
// appliedSeq is advanced AFTER this returns, once, by the session - which
// is what the barrier's waiter is entitled to assume.
++seen;
consumed.store(seen);
},
&corrupt)) {
if (!ok.load()) {
return;
}
}
if (corrupt) {
ok.store(false);
return;
}
session.consumer.RetireThrough(session.consumer.AppliedSeq());
}
});
std::uint64_t emitted = 0;
for (std::uint32_t index = 0; index < kRecords && ok.load(); ++index) {
void* payload = nullptr;
while ((payload = session.cmdProducer.Reserve(1, kRecNone, sizeof(std::uint32_t))) ==
nullptr) {
// Never "wait" on a nullptr with enough free bytes - Ring.h:226-233 says that can
// only mean "too big, chunk", and a producer that waited there would stall for ever.
ASSERT_LT(session.cmdProducer.FreeBytes(), 16u);
if (session.producer.WaitForCmdSpace(16, 5000) != SessionWait::Reached) {
ok.store(false);
break;
}
}
if (payload == nullptr) {
break;
}
std::memcpy(payload, &index, sizeof(index));
++emitted;
session.producer.PublishAndNotify(emitted);
}
apply.join();
EXPECT_TRUE(ok.load());
EXPECT_EQ(consumed.load(), kRecords);
EXPECT_EQ(emitted, static_cast<std::uint64_t>(kRecords));
EXPECT_EQ(session.Control().submittedSeq.load(), static_cast<std::uint64_t>(kRecords));
// The whole point: the two sides' sequence spaces are identical after ten wraps' worth of
// fillers. A side that counted a kRecPad would land here off by the number of wraps.
EXPECT_EQ(session.Control().appliedSeq.load(), static_cast<std::uint64_t>(kRecords));
EXPECT_EQ(session.consumer.AppliedSeq(), static_cast<std::uint64_t>(kRecords));
EXPECT_TRUE(RingCursorsValid(session.Control(), RingCursorSet::Cmd,
session.serverSegments.CmdRingCapacity()));
}
// ---------------------------------------------------------------------------
// One case per watermark rule (R-9), against the writers rather than the rules
// ---------------------------------------------------------------------------
// submittedSeq: advanced by the PRODUCER after it publishes, and NOBODY WAITS ON IT. The order
// inside PublishAndNotify is publish -> watermark -> ring, and never any other: the doorbell's
// fence only orders what precedes it, so ringing first reopens the lost-wakeup window.
TEST(SessionTest, SubmittedSeqIsThePublishersAndIsPurelyDiagnostic) {
SessionFixture session;
ASSERT_TRUE(session.Build(TestSizes()));
for (std::uint64_t seq = 1; seq <= 4; ++seq) {
void* payload = session.cmdProducer.Reserve(1, kRecNone, sizeof(std::uint64_t));
ASSERT_NE(payload, nullptr);
std::memcpy(payload, &seq, sizeof(seq));
session.producer.PublishAndNotify(seq);
EXPECT_EQ(session.Control().submittedSeq.load(), seq);
// It says nothing about what has been APPLIED, which is the distinction a waiter that
// picked the wrong watermark would lose.
EXPECT_EQ(session.Control().appliedSeq.load(), 0u);
}
}
// appliedSeq: advanced by the CONSUMER for EVERY SINGLE RECORD. P5 forbids the sixty-four
// record batching this ring was designed for, because the verb barrier and every reply wait
// read it - a batched watermark makes a waiter block on work that already ran or, far worse,
// resume on work that has not. Checked after every record, not at the end.
TEST(SessionTest, AppliedSeqAdvancesExactlyOncePerRecordAndIsNeverBatched) {
SessionFixture session;
ASSERT_TRUE(session.Build(TestSizes()));
constexpr std::uint64_t kRecords = 12;
for (std::uint64_t seq = 1; seq <= kRecords; ++seq) {
void* payload = session.cmdProducer.Reserve(1, kRecNone, sizeof(std::uint64_t));
ASSERT_NE(payload, nullptr);
std::memcpy(payload, &seq, sizeof(seq));
}
session.producer.PublishAndNotify(kRecords);
std::uint64_t applied = 0;
while (session.consumer.ApplyOne([&](const RingRecordView&) { ++applied; })) {
EXPECT_EQ(session.Control().appliedSeq.load(), applied)
<< "appliedSeq did not move with the record; a barrier waiter would be blocked on "
"work that already ran";
EXPECT_EQ(session.consumer.AppliedSeq(), applied);
}
EXPECT_EQ(applied, kRecords);
}
// retiredSeq: advanced once the SEG_STAGE bytes a record referenced are finished with, and the
// staging allocator reclaims behind it. Late is merely slow; EARLY hands live bytes back to the
// producer, so the advance clamps to appliedSeq rather than believing its caller.
TEST(SessionTest, RetiredSeqMayTrailTheApplyButCanNeverOvertakeIt) {
SessionFixture session;
ASSERT_TRUE(session.Build(TestSizes()));
for (std::uint64_t seq = 1; seq <= 3; ++seq) {
ASSERT_NE(session.cmdProducer.Reserve(1, kRecNone, 8), nullptr);
}
session.producer.PublishAndNotify(3);
while (session.consumer.ApplyOne([](const RingRecordView&) {})) {
}
ASSERT_EQ(session.Control().appliedSeq.load(), 3u);
// Trailing is legal and is what "late" means.
Watermark::AdvanceRetired(session.Control(), 1);
EXPECT_EQ(session.Control().retiredSeq.load(), 1u);
// Running ahead is not: clamped to what has actually been applied.
Watermark::AdvanceRetired(session.Control(), 99);
EXPECT_EQ(session.Control().retiredSeq.load(), 3u);
// And it never goes backwards, because a waiter that already resumed on the higher value
// cannot be un-resumed.
Watermark::AdvanceRetired(session.Control(), 2);
EXPECT_EQ(session.Control().retiredSeq.load(), 3u);
}
// completedFrameSerial: the SERVER's, advanced when a present completes. It trails appliedSeq
// by the GPU's own depth and must never be conflated with it - recycling and ageing wait on
// this one and would free a resource the GPU is still reading if they waited on the other.
TEST(SessionTest, CompletedFrameSerialIsTheServersAndIsIndependentOfAppliedSeq) {
SessionFixture session;
ASSERT_TRUE(session.Build(TestSizes()));
for (std::uint64_t seq = 1; seq <= 5; ++seq) {
ASSERT_NE(session.cmdProducer.Reserve(1, kRecNone, 8), nullptr);
}
session.producer.PublishAndNotify(5);
while (session.consumer.ApplyOne([](const RingRecordView&) {})) {
}
EXPECT_EQ(session.Control().appliedSeq.load(), 5u);
// Five records applied, no frame completed: the two are not the same number and nothing
// may derive one from the other.
EXPECT_EQ(session.Control().completedFrameSerial.load(), 0u);
Watermark::AdvanceCompletedFrame(session.Control(), 2);
EXPECT_EQ(session.Control().completedFrameSerial.load(), 2u);
Watermark::AdvanceCompletedFrame(session.Control(), 1);
EXPECT_EQ(session.Control().completedFrameSerial.load(), 2u) << "a watermark went backwards";
}
// presentAckSerial: the only back-pressure that bounds LATENCY rather than bytes. A client
// throttled on it parks, and the server's advance plus the reverse doorbell is what releases
// it - which is the half of the doorbell design that exists so a client wait is not a
// cross-process spin on one shared cache line for a whole frame of a big core.
TEST(SessionTest, PresentAckSerialIsWaitedOnWithGreaterOrEqualAndWakesThroughTheReverseBell) {
auto session = std::make_shared<SessionFixture>();
ASSERT_TRUE(session->Build(TestSizes()));
std::atomic<bool> released{false};
std::atomic<SessionWait> result{SessionWait::TimedOut};
std::thread throttled([session, &released, &result] {
result.store(session->producer.WaitForPresentAck(4, 5000));
released.store(true, std::memory_order_release);
});
// Let it get past the spin and announce itself parked, so the wakeup really travels.
while (session->Control().producerParked.load() == 0 && !released.load()) {
std::this_thread::yield();
}
// The server jumps STRAIGHT PAST the value the waiter asked for. An equality waiter would
// still be asleep here; the >= waiter this contract mandates is released.
Watermark::AdvancePresentAck(session->Control(), 7);
session->consumer.NotifyClient();
throttled.join();
EXPECT_TRUE(released.load());
EXPECT_EQ(result.load(), SessionWait::Reached);
EXPECT_EQ(session->Control().presentAckSerial.load(), 7u);
EXPECT_EQ(session->Control().producerParked.load(), 0u);
}
// ---------------------------------------------------------------------------
// kRecPad (R-9's last sentence, and the one with no other detector)
// ---------------------------------------------------------------------------
// A wrap filler is FRAMING, not a record: no opcode, no payload meaning, no reply slot. If one
// side counts it and the other does not, the two sequence spaces drift by one per wrap, for
// ever - and because seq IS the reply-slot id (R-3), a drifted seq silently reads ANOTHER
// CALL'S ANSWER rather than failing. Nothing on this ring checksums that.
//
// Here the ring is driven right across the wrap boundary with a record size that cannot divide
// it, so fillers are certain; the session's own appliedSeq must count the records and not them.
TEST(SessionTest, AWrapFillerDoesNotAdvanceTheSessionsAppliedSeq) {
SessionSegmentSizes sizes = TestSizes();
sizes.CmdBytes = 8192; // -> a 4 KiB ring, so a handful of records wraps it
SessionFixture session;
ASSERT_TRUE(session.Build(sizes));
ASSERT_EQ(session.serverSegments.CmdRingCapacity(), 4096u);
// 104 bytes + the 8-byte header = 112, and 4096 / 112 is not an integer, so the boundary
// falls inside a record and the producer must emit a filler on every lap.
constexpr std::uint64_t kPayload = 104;
constexpr std::uint64_t kRecords = 200; // ~5 laps
std::uint64_t emitted = 0;
std::uint64_t applied = 0;
std::uint64_t fillerBytes = 0;
std::uint64_t headBefore = 0;
while (emitted < kRecords) {
headBefore = session.cmdProducer.LocalHead();
void* payload = session.cmdProducer.Reserve(1, kRecNone, kPayload);
if (payload == nullptr) {
// Drain and try again; no doorbell needed, this is one thread.
session.producer.PublishAndNotify(emitted);
while (session.consumer.ApplyOne([&](const RingRecordView& view) {
// A filler must NEVER reach the thing that is about to number it.
EXPECT_EQ(view.flags & kRecPad, 0u) << "a wrap filler reached the record counter";
EXPECT_NE(view.kind, kRingPadRecordKind);
++applied;
})) {
}
session.consumer.RetireThrough(session.consumer.AppliedSeq());
continue;
}
std::memset(payload, static_cast<int>(emitted & 0xFF), static_cast<std::size_t>(kPayload));
const std::uint64_t grew = session.cmdProducer.LocalHead() - headBefore;
if (grew > kPayload + sizeof(RingRecordHeader)) {
fillerBytes += grew - (kPayload + sizeof(RingRecordHeader));
}
++emitted;
}
session.producer.PublishAndNotify(emitted);
while (session.consumer.ApplyOne([&](const RingRecordView& view) {
EXPECT_EQ(view.flags & kRecPad, 0u) << "a wrap filler reached the record counter";
++applied;
})) {
}
EXPECT_GT(fillerBytes, 0u) << "the ring never wrapped, so this case proved nothing";
EXPECT_EQ(applied, emitted);
// The session's watermark - the number the barrier's waiter and every reply read use - has
// to be the record count, with the fillers' bytes invisible to it.
EXPECT_EQ(session.Control().appliedSeq.load(), emitted);
}
// ---------------------------------------------------------------------------
// SEG_REPLY: the slot pool
// ---------------------------------------------------------------------------
TEST(SessionTest, AReplyIsAddressedBySeqAndCarriesItsSeqBackForSelfCheck) {
SessionFixture session;
ASSERT_TRUE(session.Build(TestSizes()));
ASSERT_EQ(session.replies.SlotCount(), 8u);
EXPECT_EQ(session.replies.SlotBytes(), 64ull * 1024 / 8);
EXPECT_EQ(session.replies.MaxReplyBytes(),
session.replies.SlotBytes() - sizeof(ReplySlotHeader));
const std::uint32_t pixels[4] = {1, 2, 3, 4};
session.replies.Post(5, kReplyStatusOk, pixels, sizeof(pixels));
std::uint32_t out[4] = {};
std::int32_t status = -1;
std::uint64_t size = 0;
ASSERT_TRUE(session.replies.Read(5, out, sizeof(out), &status, &size));
EXPECT_EQ(status, kReplyStatusOk);
EXPECT_EQ(size, sizeof(pixels));
EXPECT_EQ(std::memcmp(out, pixels, sizeof(pixels)), 0);
// The stamp is the self-check. Seq 13 addresses the SAME slot (13 % 8 == 5), and reading it
// as seq 13 must FAIL rather than hand back seq 5's answer - which is exactly what a
// sequence space drifted by a counted kRecPad would do.
EXPECT_FALSE(session.replies.Read(13, out, sizeof(out), &status, &size));
}
// DECLINED IS A REAL ANSWER, not a failure: it is how MapPersistent says nullptr (R-6) and how
// the four Bool acceptance entry points say false (R-5). A client that folds it into "error"
// re-creates ID-39's 66 lost uploads from the other side.
TEST(SessionTest, DeclinedIsARealAnswerWithNoPayload) {
SessionFixture session;
ASSERT_TRUE(session.Build(TestSizes()));
session.replies.Post(1, kReplyStatusDeclined, nullptr, 0);
std::int32_t status = -1;
std::uint64_t size = 99;
EXPECT_TRUE(session.replies.Read(1, nullptr, 0, &status, &size));
EXPECT_EQ(status, kReplyStatusDeclined);
EXPECT_EQ(size, 0u);
session.replies.Post(2, kReplyStatusError, nullptr, 0);
EXPECT_TRUE(session.replies.Read(2, nullptr, 0, &status, &size));
EXPECT_EQ(status, kReplyStatusError);
// Seq 0 is "no record" and can never name a slot: seq is 1-based (R-3).
EXPECT_FALSE(session.replies.Read(0, nullptr, 0, &status, &size));
}
#if defined(GTEST_HAS_DEATH_TEST) && GTEST_HAS_DEATH_TEST
// A reply larger than a slot is FATAL, not chunked and not truncated: P5's only large answer is
// a blocking ReadPixels whose size the client knows before it emits, so an overflow means the
// two sides disagree about the frame. A gate that cannot go red is not a gate.
TEST(SessionTestDeath, AReplyLargerThanItsSlotIsFatalRatherThanTruncated) {
SessionSegments segments;
ASSERT_EQ(segments.Create(TestSizes(), MemoryRole::Server), MOBILEGL_OK);
ReplySlotPool pool(segments.ReplyBase(), segments.ReplyBytes(), segments.ReplySlotCount());
ASSERT_TRUE(pool.Valid());
std::vector<std::uint8_t> oversize(pool.SlotBytes() + 1, 0xAB);
EXPECT_DEATH(pool.Post(1, kReplyStatusOk, oversize.data(), oversize.size()), "");
}
#endif
// ---------------------------------------------------------------------------
// SEG_EVENT: the reverse channel
// ---------------------------------------------------------------------------
// P5 owes exactly this: the event ring can CARRY the three callbacks the reduced path needs.
// The overflow policy is P9's, so what is pinned here is the mechanism - a full ring latches
// eventRingFull and a dropped lossy event counts in eventDropped - and not a decision between
// them.
TEST(SessionTest, TheEventRingCarriesTheThreeReverseCallbacks) {
SessionFixture session;
ASSERT_TRUE(session.Build(TestSizes()));
// OnBufferWriteback: the bytes ride INSIDE the record, and the blobref the client hands the
// frontend names SEG_EVENT plus the in-segment offset of those bytes - never a host
// pointer (R-2's rule B).
const std::uint8_t written[8] = {9, 8, 7, 6, 5, 4, 3, 2};
{
void* slot = session.eventOut.Reserve(kEventBufferWriteback,
sizeof(EventBufferWritebackHead) + sizeof(written));
ASSERT_NE(slot, nullptr);
EventBufferWritebackHead head{};
head.Resource = EventHandle{7, 1};
head.Offset = 64;
head.Size = sizeof(written);
std::memcpy(slot, &head, sizeof(head));
std::memcpy(static_cast<std::uint8_t*>(slot) + sizeof(head), written, sizeof(written));
}
// OnGpuWritten: a count and a tail of ranges.
{
void* slot = session.eventOut.Reserve(kEventGpuWritten,
sizeof(EventGpuWrittenHead) + 2 * sizeof(EventRange));
ASSERT_NE(slot, nullptr);
EventGpuWrittenHead head{};
head.Resource = EventHandle{9, 2};
head.RangeCount = 2;
std::memcpy(slot, &head, sizeof(head));
const EventRange ranges[2] = {{0, 16}, {128, 32}};
std::memcpy(static_cast<std::uint8_t*>(slot) + sizeof(head), ranges, sizeof(ranges));
}
// OnSurfaceChanged: a fixed head, the MGPSurfaceInfo image.
{
void* slot = session.eventOut.Reserve(kEventSurfaceChanged, sizeof(EventSurfaceChangedHead));
ASSERT_NE(slot, nullptr);
EventSurfaceChangedHead head{};
head.Width = 1280;
head.Height = 720;
head.IsDefault = 1;
std::memcpy(slot, &head, sizeof(head));
}
session.eventOut.PublishAndNotify(session.clientTransport->SelfDoorbell(),
session.Control().producerParked);
RingRecordView view{};
ASSERT_TRUE(session.eventIn.Pop(view));
EXPECT_EQ(view.kind, kEventBufferWriteback);
EventBufferWritebackHead writeback{};
std::memcpy(&writeback, view.payload, sizeof(writeback));
EXPECT_EQ(writeback.Resource.Slot, 7u);
EXPECT_EQ(writeback.Size, sizeof(written));
const auto* inlineBytes = static_cast<const std::uint8_t*>(view.payload) + sizeof(writeback);
EXPECT_EQ(std::memcmp(inlineBytes, written, sizeof(written)), 0);
// The offset a blobref would carry: inside SEG_EVENT and past its control page, never a
// host address.
const std::uint64_t offset = session.eventIn.OffsetInSegment(inlineBytes);
EXPECT_GE(offset, sizeof(RingControl));
EXPECT_LT(offset, session.clientSegments.AnnouncedSize(SessionSegmentSlot::Event));
ASSERT_TRUE(session.eventIn.Pop(view));
EXPECT_EQ(view.kind, kEventGpuWritten);
EventGpuWrittenHead gpuWritten{};
std::memcpy(&gpuWritten, view.payload, sizeof(gpuWritten));
EXPECT_EQ(gpuWritten.RangeCount, 2u);
ASSERT_TRUE(session.eventIn.Pop(view));
EXPECT_EQ(view.kind, kEventSurfaceChanged);
EventSurfaceChangedHead surface{};
std::memcpy(&surface, view.payload, sizeof(surface));
EXPECT_EQ(surface.Width, 1280u);
EXPECT_EQ(surface.IsDefault, 1u);
EXPECT_FALSE(session.eventIn.Pop(view));
session.eventIn.Drained();
EXPECT_FALSE(session.eventIn.RingIsFull());
EXPECT_EQ(session.eventIn.DroppedEvents(), 0u);
}
TEST(SessionTest, AFullEventRingLatchesTheFlagRatherThanDecidingWhatToDoAboutIt) {
SessionFixture session;
ASSERT_TRUE(session.Build(TestSizes()));
// Fill it. Reserve refuses at half the ring per record, so this terminates.
const std::uint64_t capacity = session.serverSegments.EventRingCapacity();
std::uint64_t posted = 0;
while (session.eventOut.Reserve(kEventGpuWritten, 256) != nullptr) {
++posted;
ASSERT_LT(posted, capacity); // a producer that never fills is a broken case
}
EXPECT_TRUE(session.eventIn.RingIsFull());
session.eventOut.CountDrop();
EXPECT_EQ(session.eventIn.DroppedEvents(), 1u);
session.eventOut.PublishAndNotify(session.clientTransport->SelfDoorbell(),
session.Control().producerParked);
RingRecordView view{};
std::uint64_t drained = 0;
while (session.eventIn.Pop(view)) {
++drained;
}
EXPECT_EQ(drained, posted);
session.eventIn.Drained();
EXPECT_FALSE(session.eventIn.RingIsFull());
}
// ---------------------------------------------------------------------------
// Shutdown
// ---------------------------------------------------------------------------
// The design's own steady state: the apply thread spun, set consumerParked and blocked with NO
// DEADLINE. Only CondVarDoorbell::Kill() can bring it back - a single Notify is consumed by one
// Park, after which Doorbell::Wait re-tests a condition nothing published, finds the bell alive
// and parks again, forever. InProcessChannel::Close kills both bells, and SessionConsumer turns
// `Wait == false && Dead()` into SessionWait::ShutDown.
//
// A REGRESSION HERE IS A HANG, so the join is bounded at five seconds and the waiter is
// detached on timeout: the test goes red instead of wedging the CI job. That is
// InProcessTransportTest.cpp:344's shape, and it is copied on purpose.
TEST(SessionTest, ShutdownUnparksTheApplyThreadAndTheJoinIsBounded) {
struct Shared {
SessionFixture session;
std::atomic<bool> returned{false};
std::atomic<SessionWait> verdict{SessionWait::Reached};
std::atomic<bool> started{false};
};
auto shared = std::make_shared<Shared>();
ASSERT_TRUE(shared->session.Build(TestSizes()));
std::thread apply([shared] {
shared->started.store(true, std::memory_order_release);
// kWaitForever, exactly as the real apply loop parks.
shared->verdict.store(shared->session.consumer.WaitForWork(kWaitForever));
shared->returned.store(true, std::memory_order_release);
});
while (shared->session.Control().consumerParked.load() == 0 &&
!shared->returned.load(std::memory_order_acquire)) {
std::this_thread::yield();
}
std::this_thread::sleep_for(std::chrono::milliseconds(50));
ASSERT_FALSE(shared->returned.load(std::memory_order_acquire))
<< "the apply thread returned before anything shut the session down";
shared->session.clientTransport->Shutdown();
const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5);
while (!shared->returned.load(std::memory_order_acquire) &&
std::chrono::steady_clock::now() < deadline) {
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
if (!shared->returned.load(std::memory_order_acquire)) {
apply.detach();
FAIL() << "Shutdown did not unpark the apply thread within 5 s: without a doorbell death "
"state a waiter consumes the ring and parks again, and teardown can never join";
}
apply.join();
EXPECT_EQ(shared->verdict.load(), SessionWait::ShutDown);
EXPECT_TRUE(shared->session.serverTransport->SelfDoorbell().Dead());
EXPECT_EQ(shared->session.Control().consumerParked.load(), 0u);
// Sticky: a wait that ARRIVES after the shutdown returns at once rather than parking, so a
// late thread cannot hang either.
const auto start = std::chrono::steady_clock::now();
EXPECT_EQ(shared->session.consumer.WaitForWork(kWaitForever), SessionWait::ShutDown);
EXPECT_LT(std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::steady_clock::now() - start)
.count(),
1000);
// And so does a producer blocked in the verb barrier: without this, a client waiting for
// appliedSeq when the server died would sit in the barrier for ever.
EXPECT_EQ(shared->session.producer.WaitForApplied(1, kWaitForever), SessionWait::ShutDown);
}
// ---------------------------------------------------------------------------
// The ABI fingerprint's mixer
// ---------------------------------------------------------------------------
// A fingerprint that cannot be SHOWN to change is indistinguishable from one that is never
// compared, which is why the mixer takes its sizes as arguments instead of reading sizeof
// directly: a test can vary one byte and prove the answer moves.
TEST(SessionTest, TheAbiFingerprintChangesWhenAnyOfItsInputsDoes) {
const std::uint64_t base = MixAbiFingerprint(1024, 1080, 552, 0x00010000, "abc1234");
EXPECT_NE(base, 0u) << "0 is reserved for \"not stated\"";
EXPECT_EQ(base, MixAbiFingerprint(1024, 1080, 552, 0x00010000, "abc1234"));
EXPECT_NE(base, MixAbiFingerprint(1025, 1080, 552, 0x00010000, "abc1234"));
EXPECT_NE(base, MixAbiFingerprint(1024, 1081, 552, 0x00010000, "abc1234"));
EXPECT_NE(base, MixAbiFingerprint(1024, 1080, 553, 0x00010000, "abc1234"));
EXPECT_NE(base, MixAbiFingerprint(1024, 1080, 552, 0x00010001, "abc1234"));
EXPECT_NE(base, MixAbiFingerprint(1024, 1080, 552, 0x00010000, "abc1235"));
// A missing stamp is not the same as an empty one, and neither is the same as a real build.
EXPECT_NE(MixAbiFingerprint(1024, 1080, 552, 0x00010000, nullptr),
MixAbiFingerprint(1024, 1080, 552, 0x00010000, "abc1234"));
}