From a3466e6f4ca31da4d85e0d34445eb7eccc732939 Mon Sep 17 00:00:00 2001 From: Swung0x48 Date: Fri, 18 Sep 2026 07:57:33 -0400 Subject: [PATCH] [MG_Remote] (Disaggregated): P5e ra - SEG_EVENT stops aborting and starts flow-controlling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - CONTRACT-P5E §2.6. Fatal{EventRingOverflow}'s premise was "the client drains at every verb, so a full 256 KiB ring is a burst no workload has". A run-ahead client drains at its waits, which on the steady path are one present apart, so a full ring is an ordinary backlog and aborting on it would turn the reverse channel's capacity into a workload limit. - The producer blocks instead (ReserveEventOrBlock, one entry point for all four event kinds): publish what is in the ring, ring the client, park on this server's own bell until the latch clears, retry. A server that does not publish kCapRunAheadApply keeps P5C's Fatal byte for byte. - Two named refusals survive, and they are the impossible cases rather than the busy ones: a single event larger than half the ring (no drain can ever make room) and a 30 s wait for a client that is not draining because it is not running. - The deadlock argument needs both halves, so both are here. EventRingConsumer::Drained() now CLEARS AND RINGS - a cleared flag with no bell is the lost wakeup the forward direction's publish-then-ring order exists to prevent - and SessionProducer gains WaitForAppliedOrEventBacklog / WaitForPresentAckOrEventBacklog so a parked client is always wakeable by the server that is waiting for it. The old two waits keep their callers: teardown has no ring to drain and must not learn about one. - ServerLoop's park condition gains a fourth clause, and it is the one that says "not now" rather than "wake up": with the ring full a waiting record is not work yet, because applying it would only park the producer half way through. Stop and Control still win, so a teardown is never held up by a client that stopped draining. - Inert this phase: nothing publishes kCapRunAheadApply until the integration commit, so every arm above answers as it did before. --- MobileGL/MG_Remote/Server/ServerLoop.cpp | 18 +++- MobileGL/MG_Remote/Server/ServerSession.cpp | 105 +++++++++++++++++--- MobileGL/MG_Remote/Transport/EventRing.h | 20 +++- MobileGL/MG_Remote/Transport/Ring.cpp | 34 +++++++ MobileGL/MG_Remote/Transport/SessionRings.h | 16 +++ 5 files changed, 173 insertions(+), 20 deletions(-) diff --git a/MobileGL/MG_Remote/Server/ServerLoop.cpp b/MobileGL/MG_Remote/Server/ServerLoop.cpp index 1d56898d..114a9a08 100644 --- a/MobileGL/MG_Remote/Server/ServerLoop.cpp +++ b/MobileGL/MG_Remote/Server/ServerLoop.cpp @@ -377,15 +377,27 @@ namespace MobileGL::MG_Remote::Server { // CondVarDoorbell::Kill() breaks that, and Kill is the CLIENT's teardown call, not // something an EGL make-current request may perform. So the predicate carries all // three arming conditions and the doorbell wakes the thread for any of them. + // P5e (ra, CONTRACT-P5E §2.6) ADDS A FOURTH, AND IT IS THE ONE THAT SAYS "NOT NOW" + // RATHER THAN "WAKE UP". A full SEG_EVENT means the apply thread has nowhere to put + // the events the next record's apply would produce, so a waiting record is NOT work + // yet: applying it would only park the producer half way through. Stop and Control + // still win - a teardown may not be held up by a client that stopped draining - which + // is why they are tested first and the ring test sits between them and the queue. + // + // The wake comes from EventRingConsumer::Drained(), which clears the latch AND rings + // this bell (that pairing is the whole of the flow-control protocol; a cleared flag + // with no bell is the lost wakeup the forward direction's publish-then-ring order + // exists to prevent). const auto ready = [this, &control, &ring] { - return control.cmdHead.load(std::memory_order_acquire) != ring.LocalTail() || - m_stopRequested.load(std::memory_order_acquire) || ControlIsPending(); + if (m_stopRequested.load(std::memory_order_acquire) || ControlIsPending()) return true; + if (control.eventRingFull.load(std::memory_order_acquire) != 0) return false; + return control.cmdHead.load(std::memory_order_acquire) != ring.LocalTail(); }; for (;;) { PumpControlRequest(); if (m_stopRequested.load(std::memory_order_acquire)) break; - DrainRing(); + if (control.eventRingFull.load(std::memory_order_acquire) == 0) DrainRing(); if (m_stopRequested.load(std::memory_order_acquire)) break; if (ready()) continue; diff --git a/MobileGL/MG_Remote/Server/ServerSession.cpp b/MobileGL/MG_Remote/Server/ServerSession.cpp index 8401dbec..f8734b58 100644 --- a/MobileGL/MG_Remote/Server/ServerSession.cpp +++ b/MobileGL/MG_Remote/Server/ServerSession.cpp @@ -180,6 +180,87 @@ namespace MobileGL::MG_Remote::Server { std::abort(); } + // ---- P5e (ra), CONTRACT-P5E §2.6: the ring stops being a Fatal and becomes a queue -- + // + // THE FATAL'S PREMISE DIES WITH THE LOCKSTEP. "A full ring is a producer burst no + // workload has" was true because the client drained at EVERY verb: the ring was empty + // whenever the server started a record, so filling 256 KiB inside one apply meant a + // defect. A run-ahead client drains at its waits, which on the steady path are one + // present apart, so a full ring is now an ordinary backlog and aborting on it would + // turn the reverse channel's capacity into a workload limit. + // + // SO THE PRODUCER BLOCKS. It publishes what is already in the ring and rings the + // client (a client that has not yet noticed cannot drain), then parks on its own bell + // until the client's Drained() clears the latch and rings back. The deadlock argument + // is the contract's, and it needs BOTH halves: the server blocks only here, the client + // blocks only on watermarks the server advances, and every client park breaks on + // `eventRingFull` as well (SessionProducer::WaitFor*OrEventBacklog) - so a parked + // client is always wakeable by the server that is waiting for it. + // + // TWO NAMED REFUSALS SURVIVE, and they are the impossible cases rather than the busy + // ones: a single event larger than the ring can EVER hold (no amount of draining + // helps), and a wait that ran out of patience (the peer is gone or wedged). A server + // that does not publish kCapRunAheadApply keeps P5C's Fatal unchanged. + Bool ServerPublishesRunAhead(const ServerSession& session) { + if (!session.CallMaskIsSet()) return false; + return (session.CallMask() & static_cast(MG_Pipe::kCapRunAheadApply)) != 0; + } + + constexpr Uint32 kEventBacklogWaitMs = 30000; + + void* ReserveEventOrBlock(ServerSession& session, Transport::EventKind kind, + const char* eventName, Uint64 payloadBytes) { + void* slot = session.Events().Reserve(kind, payloadBytes); + if (slot != nullptr) return slot; + if (!ServerPublishesRunAhead(session)) FatalEventRingOverflow(eventName, payloadBytes); + + // Impossible case one: the record does not fit an EMPTY ring. Reserve caps one + // record at capacity/2 (Ring.h), so no drain can ever make room and blocking would + // be a hang with a comment on it. + const Uint64 cap = session.Events().Ring().MaxRecordBytes(); + if (payloadBytes + sizeof(Transport::RingRecordHeader) > cap) { + MGLOG_F("MGPipe: Fatal{EventRingOverflow} - %s needs %llu bytes and SEG_EVENT " + "caps ONE record at %llu (half its capacity). Flow control cannot help: " + "no amount of draining makes a record fit a ring that is too small for " + "it. Raise MOBILEGL_IPC_EVENT_KB or slice the event at its producer, as " + "the writeback path already does", + eventName, static_cast(payloadBytes), + static_cast(cap)); + std::abort(); + } + + Transport::RingControl& control = session.Control(); + Transport::Doorbell& bell = session.ConsumerDoorbell(); + const auto drained = [&control] { + return control.eventRingFull.load(std::memory_order_acquire) == 0; + }; + for (Uint32 round = 0; round < 2; ++round) { + // PUBLISH AND RING FIRST. The client cannot drain a ring whose head it has not + // been shown, and the latch Reserve just set is only useful to a client that + // gets a bell with it. + session.PublishEvents(); + if (!drained() && !bell.Wait(control.consumerParked, drained, /*spinUs=*/0, + kEventBacklogWaitMs)) { + MGLOG_F("MGPipe: Fatal{EventRingOverflow} - %s waited %u ms for the client " + "to drain SEG_EVENT and it never did. Under run-ahead the apply " + "thread parks on this ring (CONTRACT-P5E §2.6); a client that does " + "not drain is one that is not running", + eventName, kEventBacklogWaitMs); + std::abort(); + } + slot = session.Events().Reserve(kind, payloadBytes); + if (slot != nullptr) return slot; + } + // Two full rounds against a ring the record provably fits: something else is + // producing into it, which on this side is impossible (one producer, the apply + // thread, by construction). + MGLOG_F("MGPipe: Fatal{EventRingOverflow} - %s could not reserve %llu bytes on an " + "emptied SEG_EVENT. The apply thread is SEG_EVENT's only producer, so a " + "drained ring that still refuses a record it fits is a corrupt cursor set", + eventName, static_cast(payloadBytes)); + std::abort(); + } + void ServerOnBufferWriteback(MG_Pipe::MGPipeHandle res, Uint64 offset, MG_Pipe::MGPBlobRef bytes) { if (bytes.Seg != MG_Pipe::kMGHostSpanSegNone) { @@ -195,10 +276,8 @@ namespace MobileGL::MG_Remote::Server { } ServerSession& session = ServerSessionInstance(); const Uint64 payloadBytes = sizeof(Transport::EventBufferWritebackHead) + bytes.Size; - void* slot = session.Events().Reserve(Transport::kEventBufferWriteback, payloadBytes); - if (slot == nullptr) { - FatalEventRingOverflow("kEventBufferWriteback", payloadBytes); - } + void* slot = ReserveEventOrBlock(session, Transport::kEventBufferWriteback, + "kEventBufferWriteback", payloadBytes); Transport::EventBufferWritebackHead head{}; head.Resource = Transport::EventHandle{res.Slot, res.Gen}; head.Offset = offset; @@ -217,10 +296,8 @@ namespace MobileGL::MG_Remote::Server { ServerSession& session = ServerSessionInstance(); const Uint64 tailBytes = static_cast(rangeCount) * sizeof(Transport::EventRange); const Uint64 payloadBytes = sizeof(Transport::EventGpuWrittenHead) + tailBytes; - void* slot = session.Events().Reserve(Transport::kEventGpuWritten, payloadBytes); - if (slot == nullptr) { - FatalEventRingOverflow("kEventGpuWritten", payloadBytes); - } + void* slot = ReserveEventOrBlock(session, Transport::kEventGpuWritten, + "kEventGpuWritten", payloadBytes); Transport::EventGpuWrittenHead head{}; head.Resource = Transport::EventHandle{res.Slot, res.Gen}; head.RangeCount = static_cast(rangeCount); @@ -237,10 +314,8 @@ namespace MobileGL::MG_Remote::Server { void ServerOnSurfaceChanged(const MG_Pipe::MGPSurfaceInfo* info) { ServerSession& session = ServerSessionInstance(); constexpr Uint64 payloadBytes = sizeof(Transport::EventSurfaceChangedHead); - void* slot = session.Events().Reserve(Transport::kEventSurfaceChanged, payloadBytes); - if (slot == nullptr) { - FatalEventRingOverflow("kEventSurfaceChanged", payloadBytes); - } + void* slot = ReserveEventOrBlock(session, Transport::kEventSurfaceChanged, + "kEventSurfaceChanged", payloadBytes); Transport::EventSurfaceChangedHead head{}; if (info != nullptr) { head.Width = info->Width; @@ -649,10 +724,8 @@ namespace MobileGL::MG_Remote::Server { } const Uint32 messageBytes = static_cast(length) + 1; const Uint64 payloadBytes = sizeof(Transport::EventGlErrorHead) + messageBytes; - void* slot = m_events.Reserve(Transport::kEventGlError, payloadBytes); - if (slot == nullptr) { - FatalEventRingOverflow("kEventGlError", payloadBytes); - } + void* slot = ReserveEventOrBlock(*this, Transport::kEventGlError, "kEventGlError", + payloadBytes); Transport::EventGlErrorHead head{}; head.Code = code; head.MessageBytes = messageBytes; diff --git a/MobileGL/MG_Remote/Transport/EventRing.h b/MobileGL/MG_Remote/Transport/EventRing.h index 644664af..3911f592 100644 --- a/MobileGL/MG_Remote/Transport/EventRing.h +++ b/MobileGL/MG_Remote/Transport/EventRing.h @@ -189,10 +189,26 @@ namespace MobileGL::MG_Remote::Transport { void Drained() { m_consumer.PublishRetired(); if (m_cmdControl != nullptr) { - m_cmdControl->eventRingFull.store(0, std::memory_order_release); + // P5e (ra, CONTRACT-P5E §2.6): CLEARING THE LATCH IS ONLY HALF OF IT. Under + // flow control the server is PARKED on this flag - either inside a producer + // that could not Reserve, or at a record boundary refusing to start the next + // record - and a cleared flag with no bell is a wakeup lost exactly the way + // the forward direction's publish-then-ring order exists to prevent. The + // exchange is what makes the ring one-shot: only the drain that actually found + // the flag up rings, so a steady stream of drains costs one relaxed RMW. + const bool wasFull = + m_cmdControl->eventRingFull.exchange(0, std::memory_order_acq_rel) != 0; + if (wasFull && m_serverBell != nullptr) { + NotifyIfParked(*m_serverBell, m_cmdControl->consumerParked); + } } } + // The bell the SERVER parks on, lent by the session so that Drained() can ring it. + // Null in every fixture that consumes a ring without a server behind it, and the + // ring's clear then behaves exactly as it did before P5e. + void SetServerDoorbell(Doorbell* bell) { m_serverBell = bell; } + // Byte offset of `payload` inside SEG_EVENT, which is what an // OnBufferWriteback MGPBlobRef must carry (Seg = kSegEvent, Offset = // this, Size = the head's Size). Never a host address - R-2's rule B. @@ -217,6 +233,8 @@ namespace MobileGL::MG_Remote::Transport { RingControl* m_cmdControl = nullptr; RingConsumer m_consumer; const std::uint8_t* m_segmentBase = nullptr; + // P5e (ra): the server's bell, rung by Drained() when it cleared a full latch. + Doorbell* m_serverBell = nullptr; }; } // namespace MobileGL::MG_Remote::Transport diff --git a/MobileGL/MG_Remote/Transport/Ring.cpp b/MobileGL/MG_Remote/Transport/Ring.cpp index a1473b50..dec86938 100644 --- a/MobileGL/MG_Remote/Transport/Ring.cpp +++ b/MobileGL/MG_Remote/Transport/Ring.cpp @@ -501,6 +501,40 @@ namespace MobileGL::MG_Remote::Transport { timeoutMs); } + // P5e (ra), CONTRACT-P5E §2.6. Two predicates, one OR each, and the caller decides which + // of the two happened by reading the watermark it actually cares about. + bool SessionProducer::EventRingIsFull() const { + return m_control != nullptr && m_control->eventRingFull.load(std::memory_order_acquire) != 0; + } + + SessionWait SessionProducer::WaitForAppliedOrEventBacklog(std::uint64_t seq, + std::uint32_t timeoutMs) { + if (!Valid()) { + return SessionWait::TimedOut; + } + RingControl* control = m_control; + return Park( + [control, seq] { + return Watermark::Reached(control->appliedSeq, seq) || + control->eventRingFull.load(std::memory_order_acquire) != 0; + }, + timeoutMs); + } + + SessionWait SessionProducer::WaitForPresentAckOrEventBacklog(std::uint64_t serial, + std::uint32_t timeoutMs) { + if (!Valid()) { + return SessionWait::TimedOut; + } + RingControl* control = m_control; + return Park( + [control, serial] { + return Watermark::Reached(control->presentAckSerial, serial) || + control->eventRingFull.load(std::memory_order_acquire) != 0; + }, + timeoutMs); + } + SessionWait SessionProducer::WaitForCmdSpace(std::uint64_t bytes, std::uint32_t timeoutMs) { if (!Valid()) { return SessionWait::TimedOut; diff --git a/MobileGL/MG_Remote/Transport/SessionRings.h b/MobileGL/MG_Remote/Transport/SessionRings.h index b3fd1c35..bcb445ff 100644 --- a/MobileGL/MG_Remote/Transport/SessionRings.h +++ b/MobileGL/MG_Remote/Transport/SessionRings.h @@ -325,6 +325,22 @@ namespace MobileGL::MG_Remote::Transport { SessionWait WaitForApplied(std::uint64_t seq, std::uint32_t timeoutMs); // Present throttle. SessionWait WaitForPresentAck(std::uint64_t serial, std::uint32_t timeoutMs); + + // ---- P5e (ra), CONTRACT-P5E §2.6: the same two waits, PLUS the event ring --------- + // + // A parked client must be wakeable by a server that ran out of SEG_EVENT, or the + // flow-control deadlock is one burst wide: the server stops producing, so it stops + // applying, so appliedSeq never reaches the seq this waiter is parked on, so the ring + // is never drained. Both predicates therefore break on `eventRingFull` as well, and + // the CALLER re-reads the watermark to find out which of the two woke it - a third + // SessionWait value would have made every existing `== Reached` site a bug. + // + // The old entry points above are unchanged and keep their callers: teardown's + // WaitForApplied has no ring to drain and must not learn about one. + SessionWait WaitForAppliedOrEventBacklog(std::uint64_t seq, std::uint32_t timeoutMs); + SessionWait WaitForPresentAckOrEventBacklog(std::uint64_t serial, std::uint32_t timeoutMs); + // True when the peer latched "SEG_EVENT is full and I stopped producing". + bool EventRingIsFull() const; // Back-pressure when Reserve returned nullptr. NEVER call this when // FreeBytes() is already >= the record: Ring.h:226-233 - a nullptr with // enough free bytes can only mean "too big, chunk", and waiting on it