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

[MG_Remote, MG_Test, ci] (P7): red-once controls for the latch's decline-and-close half, a control pump that declines after the latch, the pre-gate ahead of the applier's stamp, and a PH-5 raw-peer row that discriminates

Browse Source
This commit is contained in:
swung0x48
2026-09-23 13:44:37 -04:00
parent ec3758d088
commit 8848ee6ccf
10 changed files with 803 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MobileGL/MG_Remote/Server/PipeApplier.cpp
+9
View File
@@ -1305,6 +1305,15 @@ namespace MobileGL::MG_Remote::Server {
// a short-circuit, which is ID-103's reason extended to the capability probe: they are
// exercised on every record for the whole phase rather than first running on the day
// they start deciding.
//
// PH-1 (3): BUT NOT BEFORE THE RECORD HAS BEEN ADMITTED. MGPipeBarriered reads payload
// fields (a DrawVbo's MGPDrawInfo::Flags) and the stamps index per-opcode tables, so a
// record whose ring header is shorter than its own type - or names no opcode - would be
// read past its end here, before DecodeAndApply's pre-gate could refuse it. In an armed
// session child the pre-gate therefore runs FIRST: a refused record latches by name and
// is declined with nothing stamped (ServerLoopTest's short-record case reads the stamp).
// Unarmed it admits everything and the generated gate keeps its death, as before.
if (!m_decoder.AdmitOrDecline(record)) return false;
const Bool wireSaysBarriered = MG_Pipe::MGPipeBarriered(
static_cast<MG_Pipe::MGPWireOp>(record.kind), record.payload, MG_Pipe::MGPipeApplier());
MG_Pipe::MGPipeApplierSetCurrentRecordBarriered(
MobileGL/MG_Remote/Server/ServerLoop.cpp
+35 -5
View File
@@ -430,6 +430,8 @@ namespace MobileGL::MG_Remote::Server {
// (and does not spin on a ring whose head is still ahead of a tail DrainRing will no
// longer move). The exit path below tears the backend down as for a Stop; the
// control connection's owner (ServerMain::RunSession) sees the latch and closes.
// Pinned by ServerLoopTest's latched-batch case: without this break the thread is
// still running, with no Stop(), three seconds after the latch.
if (SessionLatched()) {
MGLOG_E("MG_Remote server: mgl-srv-apply leaves its loop - the session latched a "
"named fault and declines the rest of the ring (PH-1)");
@@ -593,6 +595,7 @@ namespace MobileGL::MG_Remote::Server {
if (!m_controlPosted.load(std::memory_order_acquire)) return false;
SurfaceControlFrame frame;
Bool declinedOnLatch = false;
{
const std::lock_guard<std::mutex> lock(m_controlMutex);
if (!m_controlPending) {
@@ -609,6 +612,23 @@ namespace MobileGL::MG_Remote::Server {
// that would act on the shadow is THIS one and it is busy running the frame; leaving
// the shadow set would only make the post-work iteration spin instead of park.
m_controlPosted.store(false, std::memory_order_release);
// PH-1 (3), ID-P7-1: A LATCHED SESSION RUNS NO FURTHER CONTROL OP. A frame can be
// posted before the latch and taken after it - RunSession passed SurfaceOpCodec's
// latch check a moment before a record latched on this thread, and this thread's
// exit path pumps once more - and a MakeCurrent, a surface create or a swap has no
// business running on a backend whose session has already been declared over. It
// is answered here, without a dispatch, with the code SurfaceOpCodec's own latched
// arm gives. ServerLoopTest's post-after-latch case goes red without it.
if (SessionLatched()) {
m_controlResult = MOBILEGL_ERR_PROTOCOL_MISMATCH;
m_controlPending = false;
m_controlFinished = true;
declinedOnLatch = true;
}
}
if (declinedOnLatch) {
m_controlDone.notify_all();
return true;
}
// The dispatch runs OUTSIDE the lock, exactly as the old work(user) did, and fills the
// frame's reply half; the reply is then published back into the slot under the lock so
@@ -637,17 +657,25 @@ namespace MobileGL::MG_Remote::Server {
// allocation; `applied == 0` then skips the retire block below entirely. Keep it that
// way: anything added here runs ~900 times a frame doing nothing.
//
// PH-1 (3), ID-P7-1: THE LATCH CHECK, AT THE TOP AND BEFORE EVERY POP. Once a named fault
// has latched (an armed spawn / TCP session child only - unarmed this is one acquire load
// of a flag nobody sets), no further record is applied: the rest of the ring is declined
// unread, the apply thread leaves its loop, and RunSession closes the session by name.
// PH-1 (3), ID-P7-1: THE LATCH CHECK, TWICE, AND EACH ONE HAS A JOB THE OTHER CANNOT DO.
// Once a named fault has latched (an armed spawn / TCP session child only - unarmed this is
// one acquire load of a flag nobody sets), no further record is applied: the rest of the
// ring is declined unread, the apply thread leaves its loop, and RunSession closes the
// session by name.
// * HERE, at the top: a drain ENTERED latched pops nothing. That is the apply thread's
// exit-path drain after it left on the latch, and any drain after RunSession's thread
// latched a control op. ServerLoopTest's latched-batch case goes red without it (the
// exit-path drain applies the record behind the latched one).
// * AFTER EACH RECORD, at the bottom of the loop: a record that latched is the last one
// this drain pops, even when the peer published more behind it in the same batch
// (PeerLatchTest's two-record case and ServerLoopTest's latched-batch case).
// A check at the loop's head instead would only repeat this one on the first iteration.
if (SessionLatched()) return 0;
ServerSession& session = *m_session;
Transport::SessionConsumer& consumer = session.Consumer();
PipeApplier& applier = session.Applier();
Uint64 applied = 0;
for (;;) {
if (SessionLatched()) break;
bool corrupt = false;
const bool popped = consumer.ApplyOne(
[this, &applier](const Transport::RingRecordView& record) {
@@ -684,6 +712,8 @@ namespace MobileGL::MG_Remote::Server {
// session applies. Checked only AFTER a popped record, so the empty-ring answer above
// still costs its two loads and nothing else.
if (session.ReverseChannelForfeited()) break;
// PH-1 (3): the per-record latch check described at the top of this function.
if (SessionLatched()) break;
}
if (applied != 0) {
// THE TWO TALLIES MUST AGREE, AND THAT IS WHAT MAKES R-9's BATCHING BAN CHECKABLE.
MobileGL/MG_Remote/Wire/PipeWireCodec.cpp
+15
View File
@@ -1576,6 +1576,21 @@ namespace MobileGL::MG_Remote::Wire {
}
} // namespace
Bool PipeWireDecoder::AdmitOrDecline(const Transport::RingRecordView& record) {
if ((record.flags & Transport::kRecPad) != 0 ||
record.kind == Transport::kRingPadRecordKind) {
return true; // DecodeAndApply's pad arm names it (R-9)
}
const Uint64 size = record.payloadSize + sizeof(MGPWireRecHeader);
if (AdmitsTheGeneratedGate(static_cast<MGPWireOp>(record.kind), size)) return true;
// Latched by name. The record is declined unread and counted, exactly as DecodeAndApply
// counts one it declines: the tally is +1 per record whatever became of it.
m_resolvedCount = 0;
m_lastAcceptanceKnown = false;
++m_applySeq;
return false;
}
Bool PipeWireDecoder::DecodeAndApply(const Transport::RingRecordView& record) {
if (!Valid()) {
WireProtocolFatal("PipeWireDecoder::DecodeAndApply", "no control page or segment table");
MobileGL/MG_Remote/Wire/PipeWireCodec.h
+11
View File
@@ -670,6 +670,17 @@ namespace MobileGL::MG_Remote::Wire {
// here Fatals, because a pad that reached the decoder has already been counted.
Bool DecodeAndApply(const Transport::RingRecordView& record);
// PH-1 (3): THE PRE-GATE ON ITS OWN, for a caller that reads the record before it hands
// it to DecodeAndApply. PipeApplier::ApplyOne stamps the verb boundary and computes
// MGPipeBarriered - which reads payload fields (MGPDrawInfo::Flags) - BEFORE the decode,
// so a record shorter than its own type, or with an opcode no row names, has to be refused
// ahead of that read, not only inside DecodeAndApply. True: the record may be read (and
// DecodeAndApply asks the same questions again, with the same answer). False: an armed
// session child latched it by name, and it is COUNTED here as a declined record so this
// decoder's tally stays level with appliedSeq (R-9). Unarmed it always answers true: the
// generated gate keeps its own death. A pad is let through for DecodeAndApply's own arm.
Bool AdmitOrDecline(const Transport::RingRecordView& record);
// THE DECODER'S OWN TALLY, NOT THE SHARED WATERMARK. Advanced by exactly one per
// applied non-pad record.
//
MobileGL/MG_Test/Wire/CMakeLists.txt
+5
View File
@@ -176,6 +176,11 @@ if (Python3_Interpreter_FOUND)
add_test(NAME PeerLatch.SiteMap
COMMAND ${Python3_EXECUTABLE} ${MGL_ROOT}/scripts/ci/ph_latch_sites.py --self-test)
set_tests_properties(PeerLatch.SiteMap PROPERTIES LABELS unit)
# The census's rule 2 now covers SessionLatch as well as SessionFail / WireLogFatal; its
# self-test feeds it one unmarked call of each and requires every one to be reported.
add_test(NAME FatalCensus.RuleTwoSelfTest
COMMAND ${Python3_EXECUTABLE} ${MGL_ROOT}/scripts/ci/fatal_census.py --self-test)
set_tests_properties(FatalCensus.RuleTwoSelfTest PROPERTIES LABELS unit)
endif ()
# Ph fuzz arm 3 (PH-6, ID-P7-2): a `--serve` supervisor, a peer that stops draining SEG_EVENT, and
MobileGL/MG_Test/Wire/PeerLatchTest.cpp
+318 -56
View File
@@ -30,6 +30,15 @@
// WHY FORKED PEERS: ClientSession is a process singleton whose device-lost latch never resets,
// so a process that watched session A die cannot be session B. Each peer is a fork of this
// gtest process taken before any session exists; it reports through a pipe and _exits.
//
// THE DECLINE-AND-CLOSE HALF has its own cases below the table (they are not sites, so they are
// not rows; scripts/ci/ph_latch_sites.py's MECHANICS list names them): a latched record is the
// last one its batch applies (DrainRing's two latch checks), and a latch raised on the apply
// thread closes the session while the peer is still connected and silent (RunSession's sliced
// control wait). ServerLoopTest holds the in-process half of the same checks.
//
// A PASSING CASE LEAVES NOTHING IN /tmp: the socket and both role logs are removed with the
// supervisor. A failing case keeps its logs, since the assertion messages point at them.
#include <MG_Remote/Client/ClientSession.h>
#include <MG_Remote/FatalFunnel.h>
@@ -59,6 +68,7 @@
#include <fstream>
#include <functional>
#include <memory>
#include <ostream>
#include <sstream>
#include <string>
#include <thread>
@@ -69,6 +79,7 @@
#include <arpa/inet.h>
#include <netinet/in.h>
#include <poll.h>
#include <sys/resource.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <unistd.h>
@@ -169,29 +180,61 @@ namespace {
}
server.pid = -1;
}
~Supervisor() { Stop(); }
// SIGTERM ends the supervisor without its own unlink, so the socket is removed here; the
// role logs go too unless the case failed - then they are the evidence its messages
// quote, and they stay.
~Supervisor() {
Stop();
if (!tcp && !endpoint.empty()) ::unlink(endpoint.c_str());
if (logBase.empty() || ::testing::Test::HasFailure()) return;
namespace Debug = MobileGL::MG_Util::Debug;
std::error_code ec;
std::filesystem::remove(Debug::RoleLogPath(logBase.c_str(), Debug::LogRole::Server), ec);
std::filesystem::remove(Debug::RoleLogPath(logBase.c_str(), Debug::LogRole::Client), ec);
std::filesystem::remove(logBase, ec);
}
};
bool Launch(const std::string& label, bool tcp, Supervisor* sup) {
const std::string stem = "/tmp/mgl-latch-" + label + "-" + std::to_string(::getpid());
sup->logBase = stem + ".log";
sup->tcp = tcp;
if (tcp) {
const int port = FreeLoopbackPort();
if (port <= 0) return false;
sup->endpoint = "tcp://127.0.0.1:" + std::to_string(port);
} else {
sup->endpoint = stem + ".sock";
::unlink(sup->endpoint.c_str());
}
::setenv("MOBILEGL_LOG_FILE_PATH", sup->logBase.c_str(), 1);
MobileGL::MG_Util::Debug::TruncateRoleLogs(sup->logBase.c_str());
PinHeadlessEgl();
if (Server::LaunchServerWithArgs(ServerImage(), sup->endpoint, {"--serve"}, &sup->server) !=
MOBILEGL_OK) {
return false;
// FreeLoopbackPort's port can be taken by another process between its close() and the
// supervisor's bind (ctest -j on a shared host). A supervisor that could not listen exits
// 72 at once, so that is retried on a fresh port rather than read as a failure.
for (int attempt = 0; attempt < 5; ++attempt) {
if (tcp) {
const int port = FreeLoopbackPort();
if (port <= 0) return false;
sup->endpoint = "tcp://127.0.0.1:" + std::to_string(port);
} else {
sup->endpoint = stem + ".sock";
::unlink(sup->endpoint.c_str());
}
MobileGL::MG_Util::Debug::TruncateRoleLogs(sup->logBase.c_str());
if (Server::LaunchServerWithArgs(ServerImage(), sup->endpoint, {"--serve"}, &sup->server) !=
MOBILEGL_OK) {
return false;
}
bool listening = false;
bool exited = false;
(void)WaitFor(
[&] {
if (sup->Log().find("listening on") != std::string::npos) return listening = true;
int status = 0;
if (::waitpid(sup->server.pid, &status, WNOHANG) == sup->server.pid) {
sup->server.pid = -1; // reaped here; Stop() owes it nothing
return exited = true;
}
return false;
},
10000);
if (listening) return true;
if (!exited || !tcp) return false;
}
return WaitFor([&] { return sup->Log().find("listening on") != std::string::npos; }, 10000);
return false;
}
// ------------------------------------------------------------------------------------
@@ -217,8 +260,11 @@ namespace {
// Runs `body` in a forked peer connected to `sup`. `stopCleanly` sends the orderly EOF (a
// session that must end exit=0); otherwise the peer just exits once its session is gone.
// `holdMs` > 0: the peer reports as soon as `body` returns and then HOLDS its connections open
// and silent for that long before it exits; the caller gets its pid in `heldPeer` and owes it
// a kill + waitpid (the silent-peer cases' shape).
PeerReport RunPeer(const Supervisor& sup, const PeerBody& body, bool stopCleanly,
int timeoutMs = 30000) {
int timeoutMs = 30000, int holdMs = 0, pid_t* heldPeer = nullptr) {
PeerReport report{};
int fds[2] = {-1, -1};
if (::pipe(fds) != 0) {
@@ -251,6 +297,7 @@ namespace {
const ssize_t wrote = ::write(fds[1], &r, sizeof(r));
(void)wrote;
std::fflush(nullptr);
if (holdMs > 0) std::this_thread::sleep_for(std::chrono::milliseconds(holdMs));
::_exit(0);
}
::close(fds[1]);
@@ -272,11 +319,23 @@ namespace {
::kill(pid, SIGKILL);
}
::close(fds[0]);
if (holdMs > 0 && heldPeer != nullptr && ready > 0) {
*heldPeer = pid;
return report;
}
int status = 0;
::waitpid(pid, &status, 0);
return report;
}
// The held peer's end: it has done its part, so it is killed rather than waited out.
void ReleaseHeldPeer(pid_t peer) {
if (peer <= 0) return;
::kill(peer, SIGKILL);
int status = 0;
::waitpid(peer, &status, 0);
}
// After the row's bytes: give the session time to reach them. The peer does not have to
// outlive its session - the parent reads the supervisor's reap line for the verdict - and a
// server that has not read the bytes yet still does after the peer's EOF (RunSession stops the
@@ -331,47 +390,66 @@ namespace {
// the filler's bytes with the image, optionally mutates the header, and publishes.
// 16 bytes MemoryBarrier
// 24 .. 1560 bytes BindSamplerStates with (total - 24) / 8 handles in its tail
bool PublishImage(Client::ClientSession& client, const std::vector<Uint8>& image,
Uint64* outSeq,
const std::function<void(Transport::RingRecordHeader&)>& mutateHeader = {}) {
//
// SEVERAL IMAGES, ONE PUBLISH: each gets its own filler (its ring position is taken right
// after that filler is encoded, so a wrap pad between two of them cannot shift it), all are
// overwritten, and one PublishAndNotify makes them visible to the server together - the shape
// of a peer that batches, which is what DrainRing's per-record latch check is for.
// `mutateHeader` applies to the FIRST image's header; `outSeq` is the LAST record's seq.
bool PublishImages(Client::ClientSession& client, const std::vector<std::vector<Uint8>>& images,
Uint64* outSeq,
const std::function<void(Transport::RingRecordHeader&)>& mutateHeader = {}) {
if (outSeq != nullptr) *outSeq = Wire::kInvalidSeq;
auto* link = client.DataLink();
if (link == nullptr || !link->Attached()) return false;
const Uint64 total = image.size();
if (link == nullptr || !link->Attached() || images.empty()) return false;
auto& encoder = client.Encoder();
Uint64 seq = Wire::kInvalidSeq;
if (total == 16) {
P::MGPMemoryBarrier barrier{};
barrier.Bits = 0x2000u;
seq = encoder.EncodeRecord(P::MGPWireOp::MemoryBarrier, &barrier, sizeof(barrier));
} else {
const Uint64 base = Align8(sizeof(P::MGPWireRecHeader) + sizeof(P::MGPSamplerStates));
if (total < base || (total - base) % 8 != 0 || (total - base) / 8 > P::kMGPipeMaxTextureUnits)
return false;
P::MGPSamplerStates filler{};
filler.Start = 0;
filler.Count = static_cast<Uint32>((total - base) / 8);
std::vector<P::MGPipeHandle> handles(filler.Count);
seq = encoder.EncodeRecord(P::MGPWireOp::BindSamplerStates, &filler, sizeof(filler),
handles.empty() ? nullptr : handles.data(),
handles.size() * sizeof(P::MGPipeHandle));
}
if (seq == Wire::kInvalidSeq) return false;
auto& commands = link->CommandsOut();
const auto* arena = link->RecordArena();
const Uint64 head = commands.LocalHead();
if (arena == nullptr || arena->Base == nullptr || head < total) return false;
auto* bytes = static_cast<Uint8*>(arena->Base) + ((head - total) & arena->Mask);
auto* header = reinterpret_cast<Transport::RingRecordHeader*>(bytes);
if (header->size != total) return false;
std::memcpy(bytes, image.data(), image.size());
if (mutateHeader) mutateHeader(*header);
if (arena == nullptr || arena->Base == nullptr) return false;
std::vector<Uint8*> at;
Uint64 seq = Wire::kInvalidSeq;
for (const auto& image : images) {
const Uint64 total = image.size();
if (total == 16) {
P::MGPMemoryBarrier barrier{};
barrier.Bits = 0x2000u;
seq = encoder.EncodeRecord(P::MGPWireOp::MemoryBarrier, &barrier, sizeof(barrier));
} else {
const Uint64 base = Align8(sizeof(P::MGPWireRecHeader) + sizeof(P::MGPSamplerStates));
if (total < base || (total - base) % 8 != 0 ||
(total - base) / 8 > P::kMGPipeMaxTextureUnits)
return false;
P::MGPSamplerStates filler{};
filler.Start = 0;
filler.Count = static_cast<Uint32>((total - base) / 8);
std::vector<P::MGPipeHandle> handles(filler.Count);
seq = encoder.EncodeRecord(P::MGPWireOp::BindSamplerStates, &filler, sizeof(filler),
handles.empty() ? nullptr : handles.data(),
handles.size() * sizeof(P::MGPipeHandle));
}
if (seq == Wire::kInvalidSeq) return false;
const Uint64 head = commands.LocalHead();
if (head < total) return false;
auto* bytes = static_cast<Uint8*>(arena->Base) + ((head - total) & arena->Mask);
if (reinterpret_cast<Transport::RingRecordHeader*>(bytes)->size != total) return false;
at.push_back(bytes);
}
for (std::size_t i = 0; i < images.size(); ++i) {
std::memcpy(at[i], images[i].data(), images[i].size());
}
if (mutateHeader) mutateHeader(*reinterpret_cast<Transport::RingRecordHeader*>(at.front()));
client.Producer().PublishAndNotify(seq);
if (link->Flush() != MOBILEGL_OK) return false;
if (outSeq != nullptr) *outSeq = seq;
return true;
}
bool PublishImage(Client::ClientSession& client, const std::vector<Uint8>& image,
Uint64* outSeq,
const std::function<void(Transport::RingRecordHeader&)>& mutateHeader = {}) {
return PublishImages(client, {image}, outSeq, mutateHeader);
}
bool Forge(Client::ClientSession& client, PeerReport& r, P::MGPWireOp op,
const std::vector<Uint8>& payload, const std::vector<std::vector<Uint8>>& tails = {},
const std::function<void(Transport::RingRecordHeader&)>& mutateHeader = {}) {
@@ -532,6 +610,11 @@ namespace {
bool (*forge)(Client::ClientSession&, PeerReport&);
};
// A failed row names itself, not "56-byte object <...>".
void PrintTo(const Row& row, std::ostream* os) {
*os << row.name << " (" << row.file << ", " << (row.marker != nullptr ? row.marker : "refusal") << ")";
}
// Setup helpers the rows share.
P::MGPBlobRef Stage(Client::ClientSession& client, Uint64 bytes, Uint8 fill = 0x5A) {
std::vector<Uint8> data(static_cast<std::size_t>(bytes), fill);
@@ -665,23 +748,60 @@ namespace {
}
#if MOBILEGL_PIPE_PUSH
// PH-5's fixture, framed as the wire carries it: a valid archive whose first LinkArtifacts
// vector count is one element past what the remaining bytes can hold
// (ProgramArtifactsCodecTest.AVectorCountCannotReservePastTheRemainingArchiveBytes).
std::vector<Uint8> ArchiveWithAnOversizedVectorCount() {
// PH-5's RAW-PEER CONTROL, MADE TO DISCRIMINATE. What PH-5 changed (ProgramArtifactsCodec.cpp
// TakeCount, 1ae7474d) is the CHARGE for a vector count: it used to be one byte per element
// (`count <= Remaining()`), and PH-5 charges sizeof(value_type) per element before the
// resize. So the control is a count BETWEEN the two bounds - one the byte bound admits and
// PH-5 refuses: `count == Remaining()`, in an archive padded out to kArchiveBytes. Under PH-5
// the decode refuses before anything is allocated and the session latches by name. Under the
// byte bound the same count reaches `resize(count)` - Remaining() elements of
// sizeof(ResourceReflection) each, hundreds of MiB - which the session child's address space
// has been capped not to fit (CapTheSessionsAddressSpace), so it dies of std::bad_alloc
// (signal 6), with no `Fatal{` line, instead of latching: the row goes red on the reap line
// and on the first-fault line. (A count merely one element past PH-5's bound, the first
// version of this row, was refused by name either way: the element reads run out of bytes.)
constexpr Uint64 kArchiveBytes = 7ull << 20;
constexpr Uint64 kSessionHeadroom = 320ull << 20;
std::vector<Uint8> ArchiveWhoseCountOnlyTheByteBoundAdmits(std::size_t* elementBytes) {
using namespace MobileGL::MG_State::GLState;
MobileGL::Vector<Uint8> codec;
EncodeProgramArtifacts(LinkArtifacts{}, SpirvArtifacts{}, codec);
MobileGL::Vector<Uint8> archive;
EncodeProgramArchive(LinkArtifacts{}, SpirvArtifacts{}, MobileGL::Vector<Uint32>{}, archive);
const std::size_t prefix = archive.size() - codec.size();
// The codec's version (Uint32) and schema (Uint64) words, then LinkArtifacts' first
// field - uniformReflection's count.
const std::size_t countOffset = prefix + sizeof(Uint32) + sizeof(Uint64);
const std::size_t available = archive.size() - countOffset - sizeof(Uint64);
archive.resize(static_cast<std::size_t>(kArchiveBytes), 0);
const Uint64 remaining = archive.size() - countOffset - sizeof(Uint64);
std::memcpy(archive.data() + countOffset, &remaining, sizeof(remaining));
using Element = std::remove_reference_t<decltype(LinkArtifacts{}.uniformReflection)>::value_type;
const Uint64 forged = static_cast<Uint64>(available / sizeof(Element)) + 1;
std::memcpy(archive.data() + countOffset, &forged, sizeof(forged));
*elementBytes = sizeof(Element);
return std::vector<Uint8>(archive.begin(), archive.end());
}
// Caps the SESSION CHILD's address space (not this peer's, not the supervisor's) at what it
// maps now plus kSessionHeadroom: room for everything a latched session does on its way out,
// and far short of the resize the byte bound would allow. Same uid, so prlimit may.
bool CapTheSessionsAddressSpace(Uint32 sessionPid, PeerReport& r) {
std::ifstream status("/proc/" + std::to_string(sessionPid) + "/status");
std::string line;
Uint64 vmKb = 0;
while (std::getline(status, line)) {
if (line.rfind("VmSize:", 0) == 0) vmKb = std::strtoull(line.c_str() + 7, nullptr, 10);
}
if (vmKb == 0) return Note(r, "the session child's VmSize is unreadable"), false;
rlimit current{};
if (::prlimit(static_cast<pid_t>(sessionPid), RLIMIT_AS, nullptr, &current) != 0)
return Note(r, "prlimit(RLIMIT_AS) read failed"), false;
rlimit capped = current;
capped.rlim_cur = static_cast<rlim_t>(vmKb * 1024 + kSessionHeadroom);
if (current.rlim_max != RLIM_INFINITY && capped.rlim_cur > current.rlim_max) capped.rlim_cur = current.rlim_max;
if (::prlimit(static_cast<pid_t>(sessionPid), RLIMIT_AS, &capped, nullptr) != 0)
return Note(r, "prlimit(RLIMIT_AS) write failed"), false;
return true;
}
#endif
constexpr Uint8 kCreateWindowSurface = static_cast<Uint8>(::MobileGL::Wire::SurfaceOpKind::CreateWindowSurface);
@@ -1175,10 +1295,19 @@ namespace {
return true;
}},
#if MOBILEGL_PIPE_PUSH
{"D11ArchiveVectorCount", "PipeWireCodec.cpp", "CreateShaderState.Reflection (PH-5)", Outcome::Latched,
"Fatal{ProtocolCorruption, \"CreateShaderState.Reflection\"}", false,
// PH-5. Its file is the codec's, not PipeWireCodec.cpp's: the line it ends on is the
// decoder's CreateShaderState.Reflection latch, but what it CONTROLS is TakeCount's
// element charge (see ArchiveWhoseCountOnlyTheByteBoundAdmits), so the site map does
// not credit it to the decoder's site (CreateShaderStateReflectionGarbage covers that).
{"D11ArchiveVectorCount", "ProgramArtifactsCodec.cpp", "PH-5 TakeCount count x sizeof(element)",
Outcome::Latched, "Fatal{ProtocolCorruption, \"CreateShaderState.Reflection\"}", false,
[](Client::ClientSession& c, PeerReport& r) {
const std::vector<Uint8> archive = ArchiveWithAnOversizedVectorCount();
std::size_t element = 0;
const std::vector<Uint8> archive = ArchiveWhoseCountOnlyTheByteBoundAdmits(&element);
// The row proves nothing unless the resize the byte bound allows is far past the cap.
if (static_cast<Uint64>(archive.size()) * element < 2 * kSessionHeadroom)
return Note(r, "fixture: the byte-bound resize would fit the session's headroom"), false;
if (!CapTheSessionsAddressSpace(c.PeerServerPid(), r)) return false;
P::MGPProgramDesc d{};
d.Cso = {3u, 1u};
d.Reflection = c.Encoder().StageBytes(archive.data(), archive.size());
@@ -1330,6 +1459,8 @@ namespace {
return "?";
}
void ExpectARenderingNextSession(Supervisor& sup, const char* forgeName, Uint32 previousPid);
// The two-session walk every row and the PH-1 (4) headline share. `forgeName` labels the
// assertions.
void RunRowAgainst(Supervisor& sup, const char* forgeName, Outcome outcome, const char* marker,
@@ -1378,15 +1509,30 @@ namespace {
}
// ---- session B: the same supervisor serves the next connection, and it renders
ExpectARenderingNextSession(sup, forgeName, a.serverPid);
}
// SESSION B, shared by the rows and the mechanics cases: the same supervisor serves the next
// connection from a new session child, applies a record, reaps it clean, and - with headless
// EGL - reads back a green clear. The last step of every caller (it may GTEST_SKIP).
void ExpectARenderingNextSession(Supervisor& sup, const char* forgeName, Uint32 previousPid) {
const PeerReport b = RunPeer(sup, HealthyNextSession, /*stopCleanly=*/true);
ASSERT_EQ(b.started, 1) << forgeName << ": the next connection got no Welcome (" << b.note << ")";
EXPECT_NE(b.serverPid, a.serverPid) << forgeName << ": the next session is not a new session child";
EXPECT_NE(b.serverPid, previousPid) << forgeName << ": the next session is not a new session child";
EXPECT_EQ(b.applied, 1) << forgeName << ": the next session did not apply a record";
std::string reapB;
EXPECT_TRUE(WaitFor([&] { return !(reapB = sup.ReapLine(b.serverPid)).empty(); }, 15000))
<< forgeName << ": the next session was never reaped";
EXPECT_NE(reapB.find("exit=0 "), std::string::npos)
<< forgeName << ": the next session did not end cleanly: `" << reapB << "`";
// WHERE HEADLESS EGL IS ABSENT THE CASE REPORTS SKIPPED, AND ONLY AFTER THE LATCH HALF HAS
// RUN. Every assertion above - the reap line, the first named fault, the SessionLatch line,
// Welcome, the applied record and the clean reap of the next session - has been evaluated
// by now, and a GTEST_SKIP after a failed EXPECT still reports the case FAILED; so on such a
// host "Skipped" means "latched and served, render half not run". The CI unit job
// (.github/workflows/test.yml `test`) installs libegl1, whose Recommends brings the Mesa
// vendor library on an Ubuntu runner; a runner without it shows these rows as Skipped and
// PresentFrameSerialZero (which needs a context to reach its site) as skipped outright.
if (b.egl == 0) {
GTEST_SKIP() << forgeName << ": latch and next-session Welcome proven; the render half "
"needs headless EGL, absent here (" << b.note << ")";
@@ -1445,4 +1591,120 @@ TEST(PeerLatchTest, AMalformedControlFrameOverTcpLatchesAndTheNextConnectionRend
false, [](Client::ClientSession& c, PeerReport& r) { return SendSurfaceOp(c, r, 200, 0); });
}
// =====================================================================================
// THE DECLINE-AND-CLOSE HALF: "declines all further work, closes the session cleanly"
// =====================================================================================
//
// Every row above publishes ONE forged record as the ring's last entry and then lets go of the
// connection, so none of them can tell whether the session stopped applying at the latch or merely
// ran out of records, nor whether it closed on the latch or on the peer's EOF. These cases can.
// ServerLoopTest's latched-batch, post-after-latch and short-record cases hold the same checks in
// process; scripts/ci/ph_latch_sites.py's MECHANICS list maps each check to its cases.
namespace {
P::MGPHandleOnly DeathOfKind999() {
P::MGPHandleOnly death{};
death.Handle = {5u, 1u};
death.Kind = 999;
return death;
}
// Two records that each latch on their own - ObjectDeath with Kind 999, then ObjectDeath with
// a null handle - under ONE publish, so one DrainRing sees both.
bool ForgeTwoFaultsInOneBatch(Client::ClientSession& c, PeerReport& r) {
const P::MGPHandleOnly kind = DeathOfKind999();
const P::MGPHandleOnly null{};
Uint64 seq = Wire::kInvalidSeq;
if (!PublishImages(c,
{RecordImage(P::MGPWireOp::ObjectDeath, Bytes(kind)),
RecordImage(P::MGPWireOp::ObjectDeath, Bytes(null))},
&seq)) {
Note(r, "the two forged records could not be published");
return false;
}
r.forged = 1;
// Not WaitForApplied(seq): the second record is the one that must NOT be applied.
AwaitSessionEnd(c, Wire::kInvalidSeq);
return true;
}
// One latching record, then silence (RunPeer's holdMs keeps both connections open).
bool ForgeAndFallSilent(Client::ClientSession& c, PeerReport& r) {
const P::MGPHandleOnly kind = DeathOfKind999();
Uint64 seq = Wire::kInvalidSeq;
if (!PublishImage(c, RecordImage(P::MGPWireOp::ObjectDeath, Bytes(kind)), &seq)) {
Note(r, "the forged record could not be published");
return false;
}
r.forged = 1;
// The consumer publishes appliedSeq for the latched record too, so this returns at once.
(void)c.WaitForApplied(seq, 3000);
return true;
}
constexpr int kSilentHoldMs = 10000;
constexpr int kCloseDeadlineMs = 3000;
void ExpectTheLatchClosesTheSessionUnderASilentPeer(const char* label, bool tcp) {
Supervisor sup;
ASSERT_TRUE(Launch(label, tcp, &sup)) << "the --serve supervisor did not start";
pid_t held = -1;
const PeerReport a = RunPeer(
sup, [](Client::ClientSession& c, PeerReport& r) { (void)ForgeAndFallSilent(c, r); },
/*stopCleanly=*/false, /*timeoutMs=*/30000, kSilentHoldMs, &held);
if (a.started != 1 || a.forged != 1) ReleaseHeldPeer(held);
ASSERT_EQ(a.started, 1) << label << ": no Welcome (" << a.note << ")";
ASSERT_EQ(a.forged, 1) << label << ": the peer could not publish its bytes (" << a.note << ")";
std::string reap;
bool peerStillConnected = false;
const bool reaped = WaitFor(
[&] {
if ((reap = sup.ReapLine(a.serverPid)).empty()) return false;
peerStillConnected = held > 0 && ::waitpid(held, nullptr, WNOHANG) == 0;
return true;
},
kCloseDeadlineMs);
ReleaseHeldPeer(held);
ASSERT_TRUE(reaped)
<< label << ": the session latched on its apply thread and was still open " << kCloseDeadlineMs
<< " ms later while its peer held the connection, silent - RunSession is waiting for a control "
"frame or an EOF instead of noticing the latch (ServerMain.cpp's sliced control wait)\n"
<< sup.Log();
EXPECT_TRUE(peerStillConnected)
<< label << ": the session ended only once the peer had gone, which says nothing about the latch";
EXPECT_NE(reap.find("exit=" + std::to_string(kSessionLatchedExitCode) + " (latched fault)"),
std::string::npos)
<< label << ": `" << reap << "`";
ExpectARenderingNextSession(sup, label, a.serverPid);
}
} // namespace
// DrainRing's two latch checks, over the wire: the record behind the latched one in the same batch
// is never applied - its own named fault never reaches the log - and the session closes having
// latched exactly one. Red with either check deleted: the top one (the apply thread's exit-path
// drain then applies the second record) or the per-record one (the same drain does).
TEST(PeerLatchTest, ALatchedRecordIsTheLastRecordItsBatchApplies) {
Supervisor sup;
ASSERT_TRUE(Launch("batch", /*tcp=*/false, &sup));
RunRowAgainst(sup, "batch", Outcome::Latched, "Fatal{ProtocolCorruption, \"ObjectDeath.Kind\"} - 999", false,
&ForgeTwoFaultsInOneBatch);
const std::string log = sup.Log();
EXPECT_EQ(log.find("\"ObjectDeath.Handle\""), std::string::npos)
<< "the record published behind the latched one was applied after the latch:\n" << log;
EXPECT_NE(log.find("session closed on a latched fault (ProtocolCorruption; 1 named fault(s)"), std::string::npos)
<< "the session did not close on exactly one latched fault:\n" << log;
}
// RunSession's sliced control wait: a latch raised on the APPLY thread while the peer sits connected
// and silent still ends the session within a few slices - not at the peer's EOF, which here is ten
// seconds away. Red with the wait put back to kWaitForever. Over the unix pair and over TCP, whose
// control transports time out differently.
TEST(PeerLatchTest, ALatchOnTheApplyThreadClosesTheSessionWhileAUnixPeerHoldsItOpen) {
ExpectTheLatchClosesTheSessionUnderASilentPeer("silent-unix", /*tcp=*/false);
}
TEST(PeerLatchTest, ALatchOnTheApplyThreadClosesTheSessionWhileATcpPeerHoldsItOpen) {
ExpectTheLatchClosesTheSessionUnderASilentPeer("silent-tcp", /*tcp=*/true);
}
#endif // !_WIN32
MobileGL/MG_Test/Wire/ServerLoopTest.cpp
+257
View File
@@ -45,6 +45,7 @@
#include <MG_Pipe/PipeApply.h>
#include <MG_Remote/CapsCodec.h>
#include <MG_Remote/Client/ClientSession.h>
#include <MG_Remote/FatalFunnel.h>
#include <MG_Remote/Protocol/generated/protocol_generated.h>
#include <MG_Remote/Protocol/SurfaceOpCodec.h>
#include <MG_Remote/Server/PipeApplier.h>
@@ -2704,6 +2705,262 @@ TEST(P5fReverseChannel, CachedReverseCallbacksRejectAMissingSessionOwner) {
}
#endif
// =====================================================================================
// PH-1 (3), ID-P7-1: THE LATCH'S DECLINE HALF, IN PROCESS
// =====================================================================================
//
// The latch is ARMED only in a spawn / TCP session child (ServerMain::RunSession), and arming is
// one-way and process-wide, so every case here arms it in a FORKED CHILD (EXPECT_EXIT) and reports
// through the child's exit status: 0 is "the claim held", anything else is a bitmask naming each
// part that did not (the failure message decodes it). This process's own latch is never armed, so
// every other case in the binary keeps its deaths. PeerLatchTest holds the same checks across a
// real process boundary; scripts/ci/ph_latch_sites.py's MECHANICS list maps each check to both.
#if !defined(_WIN32)
namespace {
namespace Remote = MobileGL::MG_Remote;
bool PollUntil(const std::function<bool()>& predicate, int timeoutMs) {
const auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeoutMs);
while (std::chrono::steady_clock::now() < deadline) {
if (predicate()) return true;
std::this_thread::sleep_for(std::chrono::milliseconds(2));
}
return predicate();
}
// THREE RECORDS UNDER ONE PUBLISH, THE FIRST OF WHICH LATCHES: ObjectDeath with Kind 999, then
// ObjectDeath with a null handle (a latching fault of its own), then a legal MemoryBarrier.
enum : int {
kBatchThreadStayed = 1, // the apply thread was still running 3 s after the latch, no Stop()
kBatchDrainedMore = 2, // a record behind the latched one reached the applier
kBatchLatchedMore = 4, // a second named fault was latched (the null-handle record ran)
kBatchWrongFirst = 8, // the latched line is not the Kind fault's
kBatchWatermark = 16, // appliedSeq moved past the latched record
};
[[noreturn]] void RunALatchedBatchAndExit() {
Remote::ArmSessionLatch();
ServerFixture fixture;
if (!fixture.Handshake() || !fixture.StartLoop()) ::_exit(64);
if (!fixture.WaitUntilTrulyParked()) ::_exit(65);
Server::ServerLoop& loop = Server::ServerLoopInstance();
MG_Pipe::MGPHandleOnly kind{};
kind.Handle = MG_Pipe::MGPipeHandle{5u, 1u};
kind.Kind = 999;
const MG_Pipe::MGPHandleOnly nullHandle{};
MG_Pipe::MGPMemoryBarrier barrier{};
barrier.Bits = 0x2000u;
(void)fixture.encoder.EncodeRecord(MG_Pipe::MGPWireOp::ObjectDeath, &kind, sizeof(kind));
(void)fixture.encoder.EncodeRecord(MG_Pipe::MGPWireOp::ObjectDeath, &nullHandle, sizeof(nullHandle));
const Uint64 last =
fixture.encoder.EncodeRecord(MG_Pipe::MGPWireOp::MemoryBarrier, &barrier, sizeof(barrier));
if (last == Codec::kInvalidSeq) ::_exit(66);
fixture.encoder.Publish();
fixture.producer.PublishAndNotify(last);
int failed = 0;
if (!PollUntil([&] { return !loop.Running(); }, 3000)) failed |= kBatchThreadStayed;
if (loop.DrainedRecords() != 1) failed |= kBatchDrainedMore;
if (Remote::SessionLatchCount() != 1) failed |= kBatchLatchedMore;
if (std::strstr(Remote::SessionLatchedLine(), "\"ObjectDeath.Kind\"") == nullptr) failed |= kBatchWrongFirst;
if (fixture.session->Consumer().AppliedSeq() != 1) failed |= kBatchWatermark;
fixture.Stop();
::_exit(failed);
}
// A CONTROL FRAME POSTED AFTER THE LATCH, before the apply thread has looked at it.
enum : int {
kPostRan = 1, // the probe ran on the apply thread
kPostAnswered = 2, // the poster was not answered PROTOCOL_MISMATCH
kPostDispatched = 4, // ControlFramesDispatched moved
};
[[noreturn]] void PostAfterALatchAndExit() {
Remote::ArmSessionLatch();
ServerFixture fixture;
if (!fixture.Handshake() || !fixture.StartLoop()) ::_exit(64);
if (!fixture.WaitUntilTrulyParked()) ::_exit(65);
Server::ServerLoop& loop = Server::ServerLoopInstance();
const Uint64 dispatchedBefore = loop.ControlFramesDispatched();
// Latched OFF the apply thread - where RunSession's SurfaceOp arm latches for real - so
// the parked thread has not seen it: the latch is not in its park predicate, and the
// frame below is what wakes it.
(void)Remote::SessionLatch(Remote::MGFatalFamily::ProtocolCorruption,
"MGPipe: Fatal{ProtocolCorruption, \"unit.control\"} - latched off the "
"apply thread before the frame was taken");
std::atomic<int> ran{0};
const MobileGLResult rc = loop.RunProbeOnApplyThreadForTesting(
+[](void* user) -> MobileGLResult {
static_cast<std::atomic<int>*>(user)->fetch_add(1);
return MOBILEGL_OK;
},
&ran);
int failed = 0;
if (ran.load() != 0) failed |= kPostRan;
if (rc != MOBILEGL_ERR_PROTOCOL_MISMATCH) failed |= kPostAnswered;
if (loop.ControlFramesDispatched() != dispatchedBefore) failed |= kPostDispatched;
fixture.Stop();
::_exit(failed);
}
// TWO FAULTS, ONE LATCH: the funnel's own bookkeeping.
enum : int {
kSemReturned = 1, // an armed SessionLatch returned true (or did not return)
kSemNotLatched = 2, // SessionLatched() is false after a fault
kSemNotFirst = 4, // the latched family / line is not the FIRST fault's
kSemCount = 8, // SessionLatchCount / SessionFaultCount did not count both
kSemFrames = 16, // not exactly one SessionFault frame, naming the first fault
};
[[noreturn]] void LatchTwiceAndExit() {
Remote::ArmSessionLatch();
ServerFixture fixture;
if (!fixture.Handshake()) ::_exit(64);
std::vector<Uint8> buffer(64 * 1024);
const auto pump = [&](int* fatalFrames, bool* namesFirst) {
for (;;) {
std::uint64_t size = 0;
const MobileGLResult rc =
fixture.clientTransport->ReceiveFrame({buffer.data(), buffer.size()}, &size, 100);
if (rc == MOBILEGL_ERR_BUFFER_TOO_SMALL) {
buffer.resize(static_cast<SizeT>(size));
continue;
}
if (rc != MOBILEGL_OK) return;
if (fatalFrames == nullptr) continue;
const auto* envelope = ::MobileGL::Wire::GetCtrlEnvelope(buffer.data());
if (envelope == nullptr || envelope->msg_type() != ::MobileGL::Wire::CtrlMsg::Fatal) continue;
++*fatalFrames;
const auto* fatal = envelope->msg_as_Fatal();
if (fatal != nullptr && fatal->message() != nullptr &&
std::strstr(fatal->message()->c_str(), "unit.first") != nullptr) {
*namesFirst = true;
}
}
};
pump(nullptr, nullptr); // what the handshake left for the client (Welcome, caps)
const Uint64 faultsBefore = Remote::SessionFaultCount();
const bool first = Remote::SessionLatch(Remote::MGFatalFamily::ProtocolCorruption,
"MGPipe: Fatal{ProtocolCorruption, \"unit.first\"} - the first "
"named fault (%d)",
1);
const bool second = Remote::SessionLatch(Remote::MGFatalFamily::UnmigratedVerb,
"MGPipe: Fatal{UnmigratedVerb, \"unit.second\"} - a later one (%d)", 2);
int failed = 0;
if (first || second) failed |= kSemReturned;
if (!Remote::SessionLatched()) failed |= kSemNotLatched;
const char* line = Remote::SessionLatchedLine();
if (Remote::SessionLatchedFamily() != Remote::MGFatalFamily::ProtocolCorruption ||
std::strstr(line, "unit.first") == nullptr || std::strstr(line, "unit.second") != nullptr) {
failed |= kSemNotFirst;
}
if (Remote::SessionLatchCount() != 2 || Remote::SessionFaultCount() - faultsBefore != 2) failed |= kSemCount;
int fatalFrames = 0;
bool namesFirst = false;
pump(&fatalFrames, &namesFirst);
if (fatalFrames != 1 || !namesFirst) failed |= kSemFrames;
fixture.Stop();
::_exit(failed);
}
// A RECORD SHORTER THAN ITS OWN TYPE, refused before the applier stamps or reads it.
enum : int {
kShortStamped = 1, // the Clear verb was stamped from a record the pre-gate refuses
kShortWrongFault = 2, // the latched line is not record.Minimum
kShortDrained = 4, // not exactly the setup draw and the short record reached the applier
};
[[noreturn]] void AShortRecordIsLatchedBeforeItIsStampedAndExit() {
Remote::ArmSessionLatch();
ServerFixture fixture;
if (!fixture.Handshake() || !fixture.StartLoop()) ::_exit(64);
// A legal draw first: it stamps DrawArrays, so the verb read below is this child's own
// doing whatever the parent had stamped before the fork.
MG_Pipe::MGPDrawInfo info{};
info.Mode = 0x0004; // GL_TRIANGLES
info.InstanceCount = 1;
info.MinIndex = ~0u;
info.MaxIndex = ~0u;
info.NumDraws = 1;
const MG_Pipe::MGPDrawRange range{0, 3, 0};
if (!fixture.EmitAndWaitWithTail(MG_Pipe::MGPWireOp::DrawVbo, &info, sizeof(info), &range, sizeof(range)))
::_exit(65);
if (MG_Pipe::gPipeInputs.CurrentVerb() == MG_Pipe::MGPipeVerb::Clear) ::_exit(66);
// A Clear encoded legally and then SHORTENED in the ring before it is published - the
// raw-record peer driver's move, in process: its header says 8 bytes, a header and no
// MGPClear at all.
const MG_Pipe::MGPClear clear = WholeFramebufferClear();
const Uint64 seq = fixture.encoder.EncodeRecord(MG_Pipe::MGPWireOp::Clear, &clear, sizeof(clear));
if (seq == Codec::kInvalidSeq) ::_exit(67);
const Uint64 total = (sizeof(MG_Pipe::MGPWireRecHeader) + sizeof(clear) + 7u) & ~Uint64{7u};
auto* ring = static_cast<Uint8*>(fixture.clientSegments.CmdRingBase());
const Uint64 mask = fixture.clientSegments.CmdRingCapacity() - 1;
auto* header =
reinterpret_cast<Transport::RingRecordHeader*>(ring + ((fixture.cmd.LocalHead() - total) & mask));
if (header->size != total || header->kind != static_cast<std::uint16_t>(MG_Pipe::MGPWireOp::Clear))
::_exit(68);
header->size = 8;
fixture.encoder.Publish();
fixture.producer.PublishAndNotify(seq);
Server::ServerLoop& loop = Server::ServerLoopInstance();
(void)PollUntil([&] { return !loop.Running(); }, 3000);
int failed = 0;
if (MG_Pipe::gPipeInputs.CurrentVerb() == MG_Pipe::MGPipeVerb::Clear) failed |= kShortStamped;
if (!Remote::SessionLatched() || std::strstr(Remote::SessionLatchedLine(), "\"record.Minimum\"") == nullptr)
failed |= kShortWrongFault;
if (loop.DrainedRecords() != 2) failed |= kShortDrained;
fixture.Stop();
::_exit(failed);
}
} // namespace
// DrainRing's two latch checks and the apply thread's own exit, in process. Red with the top
// check deleted (the exit-path drain applies the null-handle record: bits 2|4|16), with the
// per-record check deleted (the same drain applies both records behind it: bits 2|4|16), and with
// ApplyThreadMain's latched break deleted (bit 1: the thread spins on a ring it will not drain).
TEST(ServerLoopLatchTest, ALatchedRecordEndsItsBatchAndTheApplyThreadLeavesWithoutAStop) {
EXPECT_EXIT(RunALatchedBatchAndExit(), ::testing::ExitedWithCode(0), ".*")
<< "bits: 1 = the apply thread was still running 3 s after the latch; 2 = a record behind the "
"latched one reached the applier; 4 = a second fault was latched; 8 = the latched line is "
"not ObjectDeath.Kind; 16 = appliedSeq moved past the latched record (64+ = setup)";
}
// ServerLoop::PumpControlRequest's latched arm. A frame taken after the latch - posted before it
// was seen, or pumped by the exit path - is answered PROTOCOL_MISMATCH and never dispatched. Red
// with the arm deleted: the probe runs (bits 1|2|4).
TEST(ServerLoopLatchTest, AControlFrameTakenAfterTheLatchIsAnsweredWithoutRunning) {
EXPECT_EXIT(PostAfterALatchAndExit(), ::testing::ExitedWithCode(0), ".*")
<< "bits: 1 = the probe ran; 2 = the poster was not answered PROTOCOL_MISMATCH; 4 = the dispatch "
"counter moved (64+ = setup)";
}
// FatalFunnel's SessionLatch: armed, every fault returns false and is counted, the FIRST one is the
// latched cause, and exactly one SessionFault frame - the first fault's - reaches the peer.
TEST(ServerLoopLatchTest, AnArmedSessionLatchKeepsTheFirstFaultCountsEveryOneAndPublishesOnce) {
EXPECT_EXIT(LatchTwiceAndExit(), ::testing::ExitedWithCode(0), ".*")
<< "bits: 1 = an armed SessionLatch did not return false; 2 = not latched; 4 = the latched "
"family/line is not the first fault's; 8 = not both faults counted; 16 = not exactly one "
"SessionFault frame naming the first fault (64+ = setup)";
}
// And unarmed - inproc, the client, every unit case - it IS SessionFail: the same line, and a death.
TEST(ServerLoopLatchTest, AnUnarmedSessionLatchDiesWithItsLineLikeSessionFail) {
EXPECT_EXIT((void)MobileGL::MG_Remote::SessionLatch(MobileGL::MG_Remote::MGFatalFamily::ProtocolCorruption,
"MGPipe: Fatal{ProtocolCorruption, \"unit.unarmed\"} - "
"no session child armed this process"),
::testing::KilledBySignal(SIGABRT), "unit\\.unarmed");
}
// PipeApplier::ApplyOne admits the record BEFORE it stamps the verb boundary or computes
// MGPipeBarriered (which reads payload fields). Red with the admission moved back behind the stamp:
// the short Clear stamps the Clear verb before DecodeAndApply's pre-gate refuses it (bit 1).
TEST(ServerLoopLatchTest, ARecordShorterThanItsTypeIsLatchedBeforeTheApplierStampsIt) {
EXPECT_EXIT(AShortRecordIsLatchedBeforeItIsStampedAndExit(), ::testing::ExitedWithCode(0), ".*")
<< "bits: 1 = the Clear verb was stamped from a record shorter than MGPClear; 2 = the latched "
"line is not record.Minimum; 4 = not exactly two records reached the applier (64+ = setup)";
}
#endif
int main(int argc, char** argv) {
// Before anything logs: MG_Util::Debug::InitFile() reads the variable once, on the first
// write, and caches the FILE*. The name carries this process's pid, because
MobileGL/MG_Test/Wire/SurfaceControlFrameTest.cpp
+31
View File
@@ -23,6 +23,7 @@
// ServerLoopTest's cases (AVoidForwarderCrossesAsOneDispatchedFrame and the whole
// ServerLoopEglTest fixture, which now drives every forwarder through the frame channel).
#include <MG_Remote/FatalFunnel.h>
#include <MG_Remote/Protocol/SurfaceOpCodec.h>
#include <MG_Util/Debug/Log.h>
#include <MG_Remote/Protocol/generated/protocol_generated.h>
@@ -402,6 +403,36 @@ TEST(SurfaceControlFrameTest, AnUnknownWireOpKindIsProtocolCorruptionByName) {
EXPECT_NE(log.find("Fatal{ProtocolCorruption, \"SurfaceOp\"}"), std::string::npos)
<< "the abort happened but not for this rule's reason; the log says: " << log;
}
// PH-1 (3), ID-P7-1: ServerApplyWireSurfaceOp's latched arm. Once the session has latched, a
// WELL-FORMED op is answered PROTOCOL_MISMATCH without being decoded into a dispatch - the op
// RunSession may already be holding when the apply thread latches. In a forked child, because
// arming is one-way and process-wide. Red with the arm deleted: the op goes on to
// RunSurfaceControlFrame, which (no apply thread in this process) answers NOT_INITIALIZED, exit 3.
namespace {
[[noreturn]] void ALatchedSessionAnswersAWellFormedOpAndExit() {
MobileGL::MG_Remote::ArmSessionLatch();
(void)MobileGL::MG_Remote::SessionLatch(MobileGL::MG_Remote::MGFatalFamily::ProtocolCorruption,
"MGPipe: Fatal{ProtocolCorruption, \"unit.surface\"} - the "
"session latched before this op arrived");
SurfaceControlFrame frame;
frame.kind = SurfaceControlOp::SetSwapInterval;
frame.seq = 77;
frame.swapInterval = 1;
flatbuffers::FlatBufferBuilder builder(256);
const ::MobileGL::Wire::SurfaceOp* wireOp = EncodeAndParse(frame, &builder);
if (wireOp == nullptr) ::_exit(64);
SurfaceControlFrame reply;
const MobileGLResult rc = ServerApplyWireSurfaceOp(*wireOp, &reply);
::_exit(rc == MOBILEGL_ERR_PROTOCOL_MISMATCH ? 0 : 3);
}
} // namespace
TEST(SurfaceControlFrameTest, ALatchedSessionAnswersAWellFormedSurfaceOpWithoutRunningIt) {
EXPECT_EXIT(ALatchedSessionAnswersAWellFormedOpAndExit(), ::testing::ExitedWithCode(0), ".*")
<< "exit 3: a latched session dispatched a well-formed SurfaceOp instead of declining it; "
"exit 64: the op did not encode";
}
#endif
int main(int argc, char** argv) {
scripts/ci/fatal_census.py
+54 -10
View File
@@ -286,6 +286,45 @@ def sanctioned(rel, anchor_window):
return any(anchor in anchor_window for anchor, _reason in FUNNEL_SITES.get(rel, ()))
def unnamed_funnel_call_lines(lines):
"""Rule 2 on one file's comment-stripped lines: the 0-based index of every WireLogFatal /
SessionFail / SessionLatch call whose window carries no `Fatal{Word` marker."""
found = []
for index, line in enumerate(lines):
if not (WIRE_LOG_FATAL_CALL.search(line) or SESSION_FAIL_CALL.search(line) or
SESSION_LATCH_CALL.search(line)):
continue
window = "\n".join(lines[index:index + kMarkerWindow])
if not MARKER.search(window):
found.append(index)
return found
# RULE 2's OWN NEGATIVE CONTROL (P7 F2 latch). Each case is a one-line call and the number of
# unmarked funnel calls rule 2 must find in it. The first row is the one PH-1 (3) added the
# SessionLatch pattern for: drop SESSION_LATCH_CALL from unnamed_funnel_call_lines and it reads 0,
# which is how a site converted from SessionFail to SessionLatch would silently leave the census.
# The last two pin the lookbehind: MG_Pipe's seams are funnels of their own, not calls to these.
SELF_TEST_CASES = (
('SessionLatch(MGFatalFamily::ProtocolCorruption, "MGPipe: no family word %d", x);', 1),
('SessionLatch(MGFatalFamily::ProtocolCorruption, "MGPipe: Fatal{ProtocolCorruption, \\"w\\"} %d", x);', 0),
('SessionFail(MGFatalFamily::ProtocolCorruption, "MGPipe: no family word");', 1),
('SessionFail(MGFatalFamily::ProtocolCorruption, "MGPipe: Fatal{ProtocolCorruption, \\"w\\"}");', 0),
('WireLogFatal("MGPipe: no family word");', 1),
('MGPipeSessionLatch(family, format);', 0),
('MGPipeSessionFail(family, format);', 0),
)
def self_test():
failures = []
for source, expected in SELF_TEST_CASES:
got = len(unnamed_funnel_call_lines(strip_comments(source).split("\n")))
if got != expected:
failures.append(f"rule 2 found {got} unmarked funnel call(s) in `{source}`, expected {expected}")
return failures
def census():
unmarked = []
families = set()
@@ -306,16 +345,11 @@ def census():
refusals.setdefault(word, []).append(f"{rel}:{index + 1}")
if rel not in FUNNEL_FILES:
# Rule 2: every WireLogFatal / SessionFail call carries a family word in its string.
# window covers its continuation lines.
for index, line in enumerate(lines):
if not (WIRE_LOG_FATAL_CALL.search(line) or SESSION_FAIL_CALL.search(line) or
SESSION_LATCH_CALL.search(line)):
continue
window = "\n".join(lines[index:index + kMarkerWindow])
if not MARKER.search(window):
unnamed_funnel_calls.append({"file": rel, "line": index + 1,
"text": line.strip()})
# Rule 2: every WireLogFatal / SessionFail / SessionLatch call carries a family word
# in its string; the window covers its continuation lines.
for index in unnamed_funnel_call_lines(lines):
unnamed_funnel_calls.append({"file": rel, "line": index + 1,
"text": lines[index].strip()})
for index, line in enumerate(lines):
if not ABORT.search(line):
@@ -388,8 +422,18 @@ def main():
parser.add_argument("--write-baseline", action="store_true",
help="Record today's numbers. Run it when a family is deliberately added.")
parser.add_argument("--json", action="store_true", help="Print the census and exit 0.")
parser.add_argument("--self-test", action="store_true",
help="Prove rule 2 still sees an unmarked SessionLatch / SessionFail / "
"WireLogFatal call, and nothing else; exit 1 if it does not.")
args = parser.parse_args()
if args.self_test:
failures = self_test()
for message in failures:
print(f"::error::{message}", file=sys.stderr)
print(f"fatal census self-test: {len(SELF_TEST_CASES)} cases, {len(failures)} failed")
return 1 if failures else 0
result = census()
if args.json:
print(json.dumps(result, indent=2))
scripts/ci/ph_latch_sites.py
Executable → Regular
+68 -3
View File
@@ -21,6 +21,12 @@ behind popcount(Mask) == Count, so only the mismatch is reachable).
`--table` prints the map as Markdown. `--self-test` proves the check can go red: it drops each
covering row in turn and requires its group to turn up unmapped.
THE DECLINE-AND-CLOSE HALF IS NOT A SITE, so it is not a row: MECHANICS below names each check
that makes a latched session decline the rest of its work and close (DrainRing's two latch checks,
the apply thread's own exit, the control pump's and SurfaceOpCodec's latched answers, RunSession's
sliced wait, the pre-gate ahead of the applier's stamp, SessionLatch's own bookkeeping), the code
that IS the check, and the cases that go red without it. The script requires both to exist.
"""
import argparse
import re
@@ -63,6 +69,63 @@ UNREACHABLE = [
"SurfaceOpCodec refuses an unknown wire kind (SurfaceOpUnknownKind) before the dispatch"),
]
REMOTE = ROOT / "MobileGL" / "MG_Remote"
WIRE_TESTS = ROOT / "MobileGL" / "MG_Test" / "Wire"
# (source, what the check does, a regex that IS the check, [(test file, case name)]) - each case goes
# red with the check deleted (the red-once is recorded in the package's NOTE).
MECHANICS = [
(REMOTE / "Server" / "ServerLoop.cpp", "DrainRing: a drain ENTERED latched pops nothing (the exit-path drain)",
r"Uint64 ServerLoop::DrainRing\(\) \{.*?if \(SessionLatched\(\)\) return 0;",
[("ServerLoopTest.cpp", "ALatchedRecordEndsItsBatchAndTheApplyThreadLeavesWithoutAStop"),
("PeerLatchTest.cpp", "ALatchedRecordIsTheLastRecordItsBatchApplies")]),
(REMOTE / "Server" / "ServerLoop.cpp", "DrainRing: a record that latched is the last one its batch applies",
r"\+\+applied;\s*if \(SessionLatched\(\)\) break;",
[("ServerLoopTest.cpp", "ALatchedRecordEndsItsBatchAndTheApplyThreadLeavesWithoutAStop"),
("PeerLatchTest.cpp", "ALatchedRecordIsTheLastRecordItsBatchApplies")]),
(REMOTE / "Server" / "ServerLoop.cpp", "ApplyThreadMain: the apply thread leaves its loop on the latch, no Stop()",
r"if \(SessionLatched\(\)\) \{\s*MGLOG_E\(\"MG_Remote server: mgl-srv-apply leaves its loop",
[("ServerLoopTest.cpp", "ALatchedRecordEndsItsBatchAndTheApplyThreadLeavesWithoutAStop")]),
(REMOTE / "Server" / "ServerLoop.cpp", "PumpControlRequest: a frame taken after the latch is answered, not run",
r"if \(SessionLatched\(\)\) \{\s*m_controlResult = MOBILEGL_ERR_PROTOCOL_MISMATCH;",
[("ServerLoopTest.cpp", "AControlFrameTakenAfterTheLatchIsAnsweredWithoutRunning")]),
(REMOTE / "Protocol" / "SurfaceOpCodec.cpp", "ServerApplyWireSurfaceOp: a latched session declines a well-formed op",
r"if \(SessionLatched\(\)\) return MOBILEGL_ERR_PROTOCOL_MISMATCH;",
[("SurfaceControlFrameTest.cpp", "ALatchedSessionAnswersAWellFormedSurfaceOpWithoutRunningIt")]),
(REMOTE / "Server" / "ServerMain.cpp", "RunSession: the control wait is sliced, so a latch closes a silent peer's session",
r"ReceiveFrame\(\{buffer\.data\(\), buffer\.size\(\)\}, &size, kControlSliceMs\)",
[("PeerLatchTest.cpp", "ALatchOnTheApplyThreadClosesTheSessionWhileAUnixPeerHoldsItOpen"),
("PeerLatchTest.cpp", "ALatchOnTheApplyThreadClosesTheSessionWhileATcpPeerHoldsItOpen")]),
(REMOTE / "Server" / "PipeApplier.cpp", "ApplyOne: the pre-gate runs before the verb stamp and the barrier read",
r"if \(!m_decoder\.AdmitOrDecline\(record\)\) return false;\s*const Bool wireSaysBarriered",
[("ServerLoopTest.cpp", "ARecordShorterThanItsTypeIsLatchedBeforeTheApplierStampsIt")]),
(REMOTE / "FatalFunnel.cpp", "SessionLatch: first fault wins, every fault counted, one SessionFault; unarmed it dies",
r"if \(!g_latchArmed\.load\(std::memory_order_acquire\)\) \{\s*SessionFail\(",
[("ServerLoopTest.cpp", "AnArmedSessionLatchKeepsTheFirstFaultCountsEveryOneAndPublishesOnce"),
("ServerLoopTest.cpp", "AnUnarmedSessionLatchDiesWithItsLineLikeSessionFail")]),
]
def mechanics_failures():
failures = []
for source, what, check, cases in MECHANICS:
if not re.search(check, source.read_text(encoding="utf-8"), re.DOTALL):
failures.append(f"MECHANICS: {source.name} no longer has the check for `{what}`")
for test_file, case in cases:
text = (WIRE_TESTS / test_file).read_text(encoding="utf-8")
if not re.search(r"\bTEST(?:_F|_P)?\(\s*\w+\s*,\s*" + case + r"\s*\)", text):
failures.append(f"MECHANICS: {test_file} has no case {case} (the control for `{what}`)")
return failures
def print_mechanics():
print()
print("| decline / close check | source | negative control(s) |")
print("|---|---|---|")
for source, what, _check, cases in MECHANICS:
print(f"| {what} | {source.name} | " + ", ".join(f"`{f.split('.')[0]}.{c}`" for f, c in cases) + " |")
LATCH_CALL = re.compile(r"(?<![A-Za-z0-9_])(SessionLatch|WireProtocolLatchAt|WireProtocolLatch)\s*\(")
STRING = re.compile(r'"((?:[^"\\]|\\.)*)"')
CONVERSION = re.compile(r"%[-+ #0-9.]*(?:hh|h|ll|l|z|j|t)?[diouxXcsp]")
@@ -330,7 +393,8 @@ def main():
mapping = build_map(sites, table)
if args.table:
print_table(mapping, table)
failures = []
print_mechanics()
failures = mechanics_failures()
if args.self_test:
failure = self_test(sites, table)
if failure:
@@ -349,8 +413,9 @@ def main():
argued = sum(1 for m in mapping if not m["rows"] and m["unreachable"])
print(f"ph latch sites: {len(sites)} latch calls = {len(mapping)} distinct lines in {len(FILES)} files; "
f"{covered} covered by PeerLatchTest rows ({len(table)} rows), {argued} argued unreachable, "
f"{len(unmapped(mapping))} unmapped" + ("; self-test red on every drop" if args.self_test and not
any("self-test" in f for f in failures) else ""))
f"{len(unmapped(mapping))} unmapped; {len(MECHANICS)} decline/close checks, each with its case(s)" +
("; self-test red on every drop" if args.self_test and not
any("self-test" in f for f in failures) else ""))
for message in failures:
print(f"::error::{message}", file=sys.stderr)
return 1 if failures else 0