mirror of
https://github.com/MobileGL-Dev/MobileGL
synced 2026-11-07 01:53:24 +09:00
[MG_Remote] (Disaggregated): P5e ra - SEG_EVENT stops aborting and starts flow-controlling
- 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.
This commit is contained in:
@@ -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;
|
||||
|
||||
|
||||
@@ -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<Uint64>(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<unsigned long long>(payloadBytes),
|
||||
static_cast<unsigned long long>(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<unsigned long long>(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<Uint64>(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<std::uint32_t>(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<Uint32>(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;
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user