diff --git a/CMakeLists.txt b/CMakeLists.txt index 984f35f8..86165f7a 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -607,6 +607,9 @@ if (MOBILEGL_BUILD_DISAGGREGATED) MobileGL/MG_Remote/Transport/InProcessTransport.cpp MobileGL/MG_Remote/Transport/SocketTransport.cpp MobileGL/MG_Remote/Transport/ControlInbox.cpp + # P9 W3: the server's log forward - a queue and a sender thread, graded by severity + # (<= WARN lossy, >= ERROR lossless and paced), so no logging thread writes the socket. + MobileGL/MG_Remote/Transport/LogForward.cpp # Keeps MG_Util/Debug/Log.h - and through it the GL frontend's # umbrella header - out of the header-only wire code (WireLog.h). MobileGL/MG_Remote/Transport/WireLog.cpp diff --git a/MobileGL/MG_Pipe/MGPipeCallbacks.h b/MobileGL/MG_Pipe/MGPipeCallbacks.h index 1a012cb9..65036006 100644 --- a/MobileGL/MG_Pipe/MGPipeCallbacks.h +++ b/MobileGL/MG_Pipe/MGPipeCallbacks.h @@ -17,10 +17,10 @@ // Today this traffic is 95 call sites across 17 methods poked directly into frontend // objects. gallium has no vocabulary for shadow writeback, GPU-write notification, texture // re-send requests or default-framebuffer geometry, because in Mesa the state tracker and -// the driver share an address space. Naming them as nine callbacks plus one forward +// the driver share an address space. Naming them as eight callbacks plus one forward // terminator (MGPipeContext::ResourceSubDataComplete) is the deliberate deviation (D8). // -// NINE, NOT THE TEN PLAN B WROTE. The tenth was OnXfbScatterReady, and it went with the +// EIGHT, NOT THE TEN PLAN B WROTE. The tenth was OnXfbScatterReady, and it went with the // design it belonged to: plan B put the XFB scatter on the CLIENT (the server would hand // back the packed scratch and the client would run the patch loop over its own shadow), so // there had to be a callback that told the client the layout. P5c/P5f went the other way - @@ -31,6 +31,13 @@ // reader of this file has to rule out by hand, and the size assertion below made it look // load-bearing. // +// The ninth was OnLog, deleted in P9 W3 for the same reason: zero producers, zero consumers, +// no EventKind. Plan B pictured the server's log lines as reverse-channel records; they never +// were. They travel the CONTROL plane as LogLine frames (protocol.fbs), written by the server's +// logger through its log forward (MG_Remote/Transport/LogForward.h) - which is also where the +// severity policy this entry's comment promised ("<= WARN lossy, >= ERROR lossless and rate +// limited") now lives, on the level LogLine has always carried. +// // Installed at context creation. In a monolith these are direct calls; under split they are // records on the reverse channel, and their ORDER is a correctness requirement rather than // an optimization (section 7.4). @@ -55,13 +62,11 @@ namespace MobileGL::MG_Pipe { // pDefaultFramebufferInfo. void (*OnSurfaceChanged)(const MGPSurfaceInfo* info); void (*OnCapsInvalidated)(); - // <= WARN is lossy, >= ERROR is lossless and rate limited. - void (*OnLog)(Uint8 level, const char* text); }; - // Nine, and the count is asserted so a tenth cannot be added without touching the + // Eight, and the count is asserted so a ninth cannot be added without touching the // transport's reverse-channel record table. - inline constexpr SizeT kMGPipeCallbackCount = 9; + inline constexpr SizeT kMGPipeCallbackCount = 8; static_assert(sizeof(MGPipeCallbacks) == kMGPipeCallbackCount * sizeof(void (*)()), "MGPipeCallbacks gained or lost a callback"); diff --git a/MobileGL/MG_Remote/Client/ClientSession.h b/MobileGL/MG_Remote/Client/ClientSession.h index e9337ead..0bf582ac 100644 --- a/MobileGL/MG_Remote/Client/ClientSession.h +++ b/MobileGL/MG_Remote/Client/ClientSession.h @@ -99,6 +99,15 @@ namespace MobileGL::MG_Remote::Client { // m_remoteControlMutex and drains first; a caller of this does neither, which is the // point. Null unless spawned. Transport::SocketTransport* ControlSocketForTest() { return m_socketTransport.get(); } + // P9 W3's fault injection F2 ONLY (EventForfeitPeerTest): the control READER, so a TCP peer + // can stop taking anything off its control connection - the server's forwarded log lines + // included - and later resume (ControlInbox::PauseReadingForTest). False when there is no + // reader to pause (not a stream session). + bool PauseControlReaderForTest(bool paused) { + if (!m_controlInbox) return false; + m_controlInbox->PauseReadingForTest(paused); + return true; + } // CONTRACT-P6 4.3: THE PID THE SERVER STATED IN ITS Welcome, which is not the same fact // as the pid this session's launcher recorded - that one is local knowledge, this one diff --git a/MobileGL/MG_Remote/Server/ServerLoop.cpp b/MobileGL/MG_Remote/Server/ServerLoop.cpp index 93b7549f..509d3d54 100644 --- a/MobileGL/MG_Remote/Server/ServerLoop.cpp +++ b/MobileGL/MG_Remote/Server/ServerLoop.cpp @@ -162,6 +162,44 @@ namespace MobileGL::MG_Remote::Server { #endif } + // P9 W3, FAULT INJECTION F2 (HANDOFF-P9 §4): THE APPLY THREAD AS A LOG FIREHOSE. + // + // MOBILEGL_TEST_APPLY_LOG_FLOOD= makes the apply thread log, after EVERY record it applies, + // `n` WARN lines of about 600 bytes and one ERROR line, each numbered in its own series + // ("flood warn #k", "flood error #k"). It is the one producer F2 needs and nothing in the + // tree otherwise is: production logging is not allowed to repeat on a per-record path + // (Log.h), so no real verb can be made to fill a TCP control connection with lines at will - + // and "the apply thread is not held by logging while its client does not read" can only be + // asserted about a thread that is logging a lot. The numbering is what lets the test prove + // every ERROR line arrived (in order) while WARN lines were dropped. + // + // Meant for a server in its OWN process (a `--serve` session child). A TEST knob: read once + // per Start(), 0 unless set, capped, and it says so in the log when it is set. + Uint32 ApplyLogFloodLinesForTest() { + const char* text = std::getenv("MOBILEGL_TEST_APPLY_LOG_FLOOD"); + if (text == nullptr || *text == '\0') return 0; + const long lines = std::strtol(text, nullptr, 10); + if (lines <= 0) return 0; + const Uint32 capped = static_cast(std::min(lines, 4096)); + MGLOG_W("MG_Remote server: MOBILEGL_TEST_APPLY_LOG_FLOOD=%u - every applied record logs that many " + "WARN lines and one ERROR line. A test-only lever (fault injection F2); never set it in a " + "measured run", + static_cast(capped)); + return capped; + } + + void FloodLogForTest(Uint32 warnLines) { + // One apply thread per process (ServerLoopInstance's), so plain counters are enough. + static unsigned long long s_warn = 0; + static unsigned long long s_error = 0; + static const std::string kFiller(560, 'w'); + for (Uint32 i = 0; i < warnLines; ++i) { + MGLOG_W("MG_Remote server: MOBILEGL_TEST_APPLY_LOG_FLOOD flood warn #%llu %s", ++s_warn, + kFiller.c_str()); + } + MGLOG_E("MG_Remote server: MOBILEGL_TEST_APPLY_LOG_FLOOD flood error #%llu", ++s_error); + } + } // namespace // --------------------------------------------------------------------------------- @@ -236,6 +274,7 @@ namespace MobileGL::MG_Remote::Server { m_serverOwnedSurfaces.clear(); m_windowLostRequested.store(false, std::memory_order_release); m_serverWindowsLost.store(0, std::memory_order_release); + m_testLogFloodLines = ApplyLogFloodLinesForTest(); { const std::lock_guard lock(m_exitMutex); m_exited = false; @@ -783,6 +822,8 @@ namespace MobileGL::MG_Remote::Server { std::chrono::steady_clock::now() - applyStarted) .count()); ++sFrameRecords; + // P9 W3, F2: outside the apply timing above; 0 in every run that is not F2. + if (m_testLogFloodLines != 0) FloodLogForTest(m_testLogFloodLines); if (record.kind == static_cast(MG_Pipe::MGPWireOp::Present)) { const auto now = std::chrono::steady_clock::now(); const auto wallNs = static_cast( diff --git a/MobileGL/MG_Remote/Server/ServerLoop.h b/MobileGL/MG_Remote/Server/ServerLoop.h index ccd2fa36..752b1c64 100644 --- a/MobileGL/MG_Remote/Server/ServerLoop.h +++ b/MobileGL/MG_Remote/Server/ServerLoop.h @@ -569,6 +569,9 @@ namespace MobileGL::MG_Remote::Server { std::atomic m_parkBlocks{0}; std::atomic m_beforeRetireHook{nullptr}; std::atomic m_betweenRecordsHook{nullptr}; + // P9 W3, fault injection F2: MOBILEGL_TEST_APPLY_LOG_FLOOD, read by Start(). Zero - the only + // value a production run has - makes the per-record test a load of this and nothing else. + Uint32 m_testLogFloodLines = 0; // C7 / ID-54: the (dpy, draw, read, ctx) currently bound on the apply thread. Written and // read ONLY on the apply thread inside ApplyMakeCurrent, so it needs no lock; the two diff --git a/MobileGL/MG_Remote/Server/ServerMain.cpp b/MobileGL/MG_Remote/Server/ServerMain.cpp index 9e7e2c42..71b5a0d1 100644 --- a/MobileGL/MG_Remote/Server/ServerMain.cpp +++ b/MobileGL/MG_Remote/Server/ServerMain.cpp @@ -2,6 +2,7 @@ // SPDX-License-Identifier: LGPL-3.0-only #include "../Transport/Doorbell.h" #include "../Transport/FdPassing.h" +#include "../Transport/LogForward.h" #include "../Transport/SocketTransport.h" #include "../Transport/WireLog.h" #include "../Protocol/SurfaceOpCodec.h" @@ -55,13 +56,42 @@ void Refuse(SocketTransport& transport, Protocol::RefuseCode code, const char* d WireLogError("MG_Remote server: Refuse{%s} %s", Protocol::EnumNameRefuseCode(code), detail); } -void ForwardLog(void* user, const char* text) { - auto& transport = *static_cast(user); - flatbuffers::FlatBufferBuilder builder(512); - auto log = Protocol::CreateLogLine(builder, Protocol::LogLevel::Info, builder.CreateString(text)); - auto envelope = Protocol::CreateCtrlEnvelope(builder, Protocol::CtrlMsg::LogLine, log.Union()); - Protocol::FinishCtrlEnvelopeBuffer(builder, envelope); - (void)transport.SendFrame({builder.GetBufferPointer(), builder.GetSize()}); +// P9 W3. The log forward's unblock hook (LogForward.h, Close): the channel's sender is still inside a +// write to a client that has not read for the whole close budget, at the end of the session. Cutting +// the control socket is what returns that write (EPIPE) so the thread can be joined - the session is +// over by then (every caller is on RunSession's way out), and the socket is closed right after anyway. +// Not through SocketTransport::Shutdown: that takes the send mutex the blocked write holds. +void UnblockControlSend(void* user) { + const int fd = static_cast(user)->StreamFd(); + if (fd >= 0) (void)::shutdown(fd, SHUT_RDWR); +} + +// The forward's counters, once, into the server's OWN log - the forwarder is already gone by the time +// they are final, so this line is local; what the client saw of the same story is the in-stream +// LogForward{Dropped} / {Coalesced} lines at each gap. WARN when anything was lost, delayed past a +// deadline or cut, so it survives a production build; otherwise a D line. +void LogForwardSummary(const Transport::LogForwardCounters& counters) { + const bool eventful = counters.lossyDropped != 0 || counters.errorsCoalesced != 0 || counters.abandoned != 0 || + counters.fatalFlushTimeouts != 0 || counters.transportFailed; + if (eventful) { + MGLOG_W("MG_Remote server: log forwarding ended - forwarded=%llu dropped(<=WARN)=%llu (%llu bytes) " + "coalesced(ERROR)=%llu paced(ERROR)=%llu notices=%llu abandoned=%llu fatal-flush-timeouts=%llu " + "transport-failed=%d max-offer-us=%llu", + static_cast(counters.forwarded), + static_cast(counters.lossyDropped), + static_cast(counters.lossyDroppedBytes), + static_cast(counters.errorsCoalesced), + static_cast(counters.errorsPaced), + static_cast(counters.notices), + static_cast(counters.abandoned), + static_cast(counters.fatalFlushTimeouts), counters.transportFailed ? 1 : 0, + static_cast(counters.maxOfferNs / 1000)); + } else { + MGLOG_D("MG_Remote server: log forwarding ended - forwarded=%llu paced(ERROR)=%llu max-offer-us=%llu", + static_cast(counters.forwarded), + static_cast(counters.errorsPaced), + static_cast(counters.maxOfferNs / 1000)); + } } // p7/spawnhang. THE PROGRESS REPORT: Wire::SurfaceProgress, sent by the thread that posted the op @@ -192,13 +222,22 @@ Server::ServerSession::DataConnectionSource ListenerSource(int listener, int con }; } -struct FlushAck { ITransport* transport; std::uint64_t seq; }; +// `forward` is the session's log forward when there is one (P9 W3). The ack then goes out THROUGH it, +// queued behind every line forwarded before the flush, because a forwarded line is no longer on the +// wire when its MGLOG returns - sending the ack directly could overtake lines still queued, and +// MGPipeSyncPeerLog's promise is that they are in the client's file when the ack arrives. +struct FlushAck { ITransport* transport; Transport::LogForwardChannel* forward; std::uint64_t seq; }; void SendLogAck(void* pointer) { auto& ack = *static_cast(pointer); flatbuffers::FlatBufferBuilder builder(64); auto flush = Protocol::CreateLogFlush(builder, ack.seq, true); auto envelope = Protocol::CreateCtrlEnvelope(builder, Protocol::CtrlMsg::LogFlush, flush.Union()); Protocol::FinishCtrlEnvelopeBuffer(builder, envelope); + if (ack.forward != nullptr) { + ack.forward->OfferControlFrame( + std::vector(builder.GetBufferPointer(), builder.GetBufferPointer() + builder.GetSize())); + return; + } (void)ack.transport->SendFrame({builder.GetBufferPointer(), builder.GetSize()}); } @@ -429,11 +468,25 @@ int RunSession(std::unique_ptr control, std::vector int { - if (!inProcess) return code; - // Both point at `control`, which dies with this frame. - loop.SetControlProgressSink(nullptr, nullptr); + // P9 W3: the TCP session's log forward (LogForward.h), made where forwarding is switched on below. + // Declared up here because every way out has to take it down the same way: the logger stops + // calling it FIRST (so no line is offered to a channel being destroyed), then the channel drains + // within its close budget and its sender is joined, then its counters are said locally. Every + // shape, forked or not: a forked child's early return destroys the channel with this frame, and a + // forwarder left pointing at it would be a dangling pointer for however many lines remain. + std::unique_ptr logForward; + const auto stopLogForwarding = [&] { MobileGL::MG_Util::Debug::SetLogForwarder(nullptr, nullptr); + if (!logForward) return; + const auto counters = logForward->Close(); + logForward.reset(); + LogForwardSummary(counters); + }; + const auto endSession = [&](int code) -> int { + stopLogForwarding(); + if (!inProcess) return code; + // Points at `control`, which dies with this frame (the log forwarder did too, above). + loop.SetControlProgressSink(nullptr, nullptr); if (backendBuilt) { // Stop() joins a running apply thread (it destroys the backend on it), or - never // started - drops the backend here; either way the display lease goes with it. @@ -575,11 +628,18 @@ int RunSession(std::unique_ptr control, std::vector(*control, &UnblockControlSend, control.get()); // P12 review fix: in-process, only THIS session's threads (this one and its apply thread) // forward, and outside the log mutex - the UI thread, the listener and the next session's // refusals share this process and must neither reach this client nor wait on it. - if (inProcess) MobileGL::MG_Util::Debug::SetSessionLogForwarder(&ForwardLog, control.get()); - else MobileGL::MG_Util::Debug::SetLogForwarder(&ForwardLog, control.get()); + if (inProcess) + MobileGL::MG_Util::Debug::SetSessionLogForwarder(&Transport::LogForwardChannel::ForwardThunk, + logForward.get()); + else + MobileGL::MG_Util::Debug::SetLogForwarder(&Transport::LogForwardChannel::ForwardThunk, logForward.get()); } // p7/spawnhang: this thread posts every surface op the client sends (ServerApplyWireSurfaceOp, // below) and is the one that says "still running" while the apply thread runs it. Cleared @@ -700,7 +760,7 @@ int RunSession(std::unique_ptr control, std::vector(verbs.ExpectedApplierResetSerial()), static_cast(verbs.ObjectDeaths())); } - FlushAck ack{control.get(), flush->seq()}; + FlushAck ack{control.get(), logForward.get(), flush->seq()}; MobileGL::MG_Util::Debug::WithLogBarrier(&SendLogAck, &ack); continue; } @@ -749,7 +809,8 @@ int RunSession(std::unique_ptr control, std::vector(SessionLatchCount()), kSessionLatchedExitCode); } session.Close(); - MobileGL::MG_Util::Debug::SetLogForwarder(nullptr, nullptr); + // P9 W3: the forwarder is cleared, the forward drains (bounded) and its counters are logged. + stopLogForwarding(); // P12 (D5): the backend loop.Stop() destroyed is not this session's to point at any more. if (inProcess) session.SetBackend(nullptr); // The caller's exit (a forked child's ExitSessionProcess) closes control after cleanup/flush - diff --git a/MobileGL/MG_Remote/Transport/ControlInbox.cpp b/MobileGL/MG_Remote/Transport/ControlInbox.cpp index 7e597199..79f62c18 100644 --- a/MobileGL/MG_Remote/Transport/ControlInbox.cpp +++ b/MobileGL/MG_Remote/Transport/ControlInbox.cpp @@ -1,5 +1,6 @@ #include "ControlInbox.h" #include "Doorbell.h" +#include "LogForward.h" #include "../Protocol/SurfaceOpCodec.h" #include #include @@ -18,11 +19,21 @@ namespace MobileGL::MG_Remote::Transport { if (m_reader.joinable()) m_reader.join(); } + void ControlInbox::PauseReadingForTest(bool paused) { + std::lock_guard lock(m_mutex); + m_pausedForTest = paused; + m_cv.notify_all(); + } + void ControlInbox::Read() { for (;;) { { - std::lock_guard lock(m_mutex); + std::unique_lock lock(m_mutex); if (m_stop) return; + if (m_pausedForTest) { + m_cv.wait(lock, [&] { return m_stop || !m_pausedForTest; }); + continue; + } } std::uint64_t size = 0; auto result = m_transport.ReceiveFrame({nullptr, 0}, &size, 100); @@ -38,7 +49,9 @@ namespace MobileGL::MG_Remote::Transport { else { const auto* envelope = ::MobileGL::Wire::GetCtrlEnvelope(frame.data()); if (const auto* log = envelope->msg_as_LogLine()) { - if (log->text()) MG_Util::Debug::WritePeerLog(log->text()->c_str()); + // P9 W3: filed with the severity the server sent (LogForward.h's mapping). + if (log->text()) + MG_Util::Debug::WritePeerLog(LogLevelFromWire(log->level()), log->text()->c_str()); continue; } } diff --git a/MobileGL/MG_Remote/Transport/ControlInbox.h b/MobileGL/MG_Remote/Transport/ControlInbox.h index 437b51b8..a054efc9 100644 --- a/MobileGL/MG_Remote/Transport/ControlInbox.h +++ b/MobileGL/MG_Remote/Transport/ControlInbox.h @@ -18,6 +18,11 @@ namespace MobileGL::MG_Remote::Transport { MobileGLResult Receive(std::vector& frame, std::uint32_t timeoutMs); std::uint64_t Peek(); bool WaitClosed(std::uint32_t timeoutMs); + // TEST-ONLY (P9 W3, fault injection F2). While paused the reader takes NOTHING off the + // socket - not a LogLine, not a reply - which is exactly a client that stopped reading its + // control connection, the peer the server's log forward (LogForward.h) must not be held by. + // Frames already queued stay queued; resuming picks up where the socket is. + void PauseReadingForTest(bool paused); private: void Read(); ITransport& m_transport; @@ -26,6 +31,7 @@ namespace MobileGL::MG_Remote::Transport { std::deque> m_frames; std::uint64_t m_bytes = 0; bool m_stop = false; + bool m_pausedForTest = false; MobileGLResult m_result = MOBILEGL_OK; std::thread m_reader; }; diff --git a/MobileGL/MG_Remote/Transport/LogForward.cpp b/MobileGL/MG_Remote/Transport/LogForward.cpp new file mode 100644 index 00000000..03a5a55d --- /dev/null +++ b/MobileGL/MG_Remote/Transport/LogForward.cpp @@ -0,0 +1,343 @@ +// MobileGL - MobileGL/MG_Remote/Transport/LogForward.cpp +// Copyright (c) 2025-2026 MobileGL-Dev +// Licensed under the GNU Lesser General Public License v3.0: +// https://www.gnu.org/licenses/gpl-3.0.txt +// https://www.gnu.org/licenses/lgpl-3.0.txt +// SPDX-License-Identifier: LGPL-3.0-only +// End of Source File Header + +// P9 W3: the graded, non-blocking log forward. The policy and its reasons are in LogForward.h. + +#include "LogForward.h" + +#include "../Protocol/generated/protocol_generated.h" + +// The umbrella, for MOBILEGL_LOG_LEVEL_* and GetCurrentTime - a .cpp may pull it (WireLog.h). +#include + +#include +#include +#include + +#if defined(__linux__) +#include +#endif + +namespace MobileGL::MG_Remote::Transport { + + namespace Wire = ::MobileGL::Wire; + + // THE TWO SCALES ARE ONE SCALE, and these are what would notice if either moved: a sixth + // MobileGL level, a reordered wire enum, or Log.h's 2026-08-13 inversion coming back. + static_assert(MOBILEGL_LOG_LEVEL_DEBUG == 0 && MOBILEGL_LOG_LEVEL_INFO == 1 && MOBILEGL_LOG_LEVEL_WARN == 2 && + MOBILEGL_LOG_LEVEL_ERROR == 3 && MOBILEGL_LOG_LEVEL_FATAL == 4, + "MobileGL's log levels moved; re-read WireLogLevelFor / LogLevelFromWire"); + static_assert(static_cast(Wire::LogLevel::Debug) == 0 && static_cast(Wire::LogLevel::Info) == 1 && + static_cast(Wire::LogLevel::Warn) == 2 && static_cast(Wire::LogLevel::Error) == 3 && + static_cast(Wire::LogLevel::Fatal) == 4, + "protocol.fbs's LogLevel moved; re-read WireLogLevelFor / LogLevelFromWire"); + + Wire::LogLevel WireLogLevelFor(int level) { + if (level <= MOBILEGL_LOG_LEVEL_DEBUG) return Wire::LogLevel::Debug; + switch (level) { + case MOBILEGL_LOG_LEVEL_INFO: return Wire::LogLevel::Info; + case MOBILEGL_LOG_LEVEL_WARN: return Wire::LogLevel::Warn; + case MOBILEGL_LOG_LEVEL_ERROR: return Wire::LogLevel::Error; + default: return Wire::LogLevel::Fatal; + } + } + + int LogLevelFromWire(Wire::LogLevel level) { + switch (level) { + case Wire::LogLevel::Debug: return MOBILEGL_LOG_LEVEL_DEBUG; + case Wire::LogLevel::Info: return MOBILEGL_LOG_LEVEL_INFO; + case Wire::LogLevel::Warn: return MOBILEGL_LOG_LEVEL_WARN; + case Wire::LogLevel::Error: return MOBILEGL_LOG_LEVEL_ERROR; + case Wire::LogLevel::Fatal: return MOBILEGL_LOG_LEVEL_FATAL; + } + // A value this build's schema does not name: a newer peer's severity. Never demoted into + // the lossy class - an unknown severity is filed as an error. + return MOBILEGL_LOG_LEVEL_ERROR; + } + + namespace { + // The header a notice line carries: the shape of Log()'s own ("[time] [OS thread/LEVEL]: "), + // so it reads like any other forwarded line and starts with '[' (WritePeerLog writes it as it + // came). The "OS" slot names the layer instead: GetOSName is not callable outside Log.cpp. + std::string NoticeHeader(int level) { + return "[" + MG_Util::Debug::GetCurrentTime() + "] [MG_Remote mgl-log-fwd/" + + MG_Util::Debug::LogLevelTag(level) + "]: "; + } + } // namespace + + LogForwardChannel::LogForwardChannel(ITransport& transport, Unblock unblock, void* unblockUser, + const LogForwardLimits& limits) + : m_transport(transport), m_unblock(unblock), m_unblockUser(unblockUser), m_limits(limits), + m_tokens(static_cast(limits.errorBurstLines)), m_tokensAt(std::chrono::steady_clock::now()) { + m_sender = std::thread([this] { Run(); }); + } + + LogForwardChannel::~LogForwardChannel() { (void)Close(); } + + void LogForwardChannel::ForwardThunk(void* user, int level, const char* line) { + static_cast(user)->Offer(level, line); + } + + bool LogForwardChannel::IsUrgentLocked() const { return m_urgent > 0 || m_closing; } + + void LogForwardChannel::QueueGapNoticesLocked(bool lossyGapToo) { + // ONE NOTICE PER GAP, however many lines it swallowed, so a notice never needs room of its + // own worth refusing and the number of notices is bounded by the number of lines that did + // get in. WHERE each class's gap is closed differs, on purpose: + // * an ERROR gap is marked before the very next item queued - the next ERROR line that + // fits, a LogFlush ack, Close - because an error gap is rare and where it sits matters; + // * a <= WARN gap is marked when the lossy class RECOVERS (the next line at or below WARN + // that fits), or at a LogFlush ack or Close. Not before every ERROR line that squeezes + // through in the middle of it: under a sustained stall that was one notice per ERROR + // line (F2 measured 373 in one run), which buries the errors it sits between. The notice + // names the time the gap began instead, so it still says where it started. + char text[512]; + if (lossyGapToo && m_gapLossyLines != 0) { + std::snprintf(text, sizeof(text), + "MG_Remote server: LogForward{Dropped} - %llu line(s) at or below WARN (%llu bytes) were " + "not forwarded, from %s until here: the control connection was not taking them and the " + "forward queue held its %llu-byte cap. The server's own log has every one\n", + static_cast(m_gapLossyLines), + static_cast(m_gapLossyBytes), m_gapLossySince.c_str(), + static_cast(m_limits.lossyQueueBytes)); + Item notice; + notice.kind = Kind::Notice; + notice.level = MOBILEGL_LOG_LEVEL_WARN; + notice.seq = m_nextSeq++; + notice.text = NoticeHeader(notice.level) + text; + m_queue.push_back(std::move(notice)); + ++m_counters.notices; + m_gapLossyLines = 0; + m_gapLossyBytes = 0; + } + if (m_gapErrorLines != 0) { + std::snprintf(text, sizeof(text), + "MG_Remote server: LogForward{Coalesced} - %llu ERROR line(s) were not forwarded here: the " + "forward queue held its %llu-byte cap of error lines the control connection had not " + "taken. The server's own log has every one\n", + static_cast(m_gapErrorLines), + static_cast(m_limits.errorQueueBytes)); + Item notice; + notice.kind = Kind::Notice; + notice.level = MOBILEGL_LOG_LEVEL_ERROR; + notice.seq = m_nextSeq++; + notice.text = NoticeHeader(notice.level) + text; + m_queue.push_back(std::move(notice)); + ++m_counters.notices; + m_gapErrorLines = 0; + } + } + + void LogForwardChannel::Offer(int level, const char* line) { + if (line == nullptr) return; + const auto started = std::chrono::steady_clock::now(); + const std::size_t length = std::strlen(line); + std::unique_lock lock(m_mutex); + // The bound this whole file exists for, measured where it is spent: everything below FATAL + // returns through here with the lock still held and no wait taken. + const auto noteOfferTime = [this, started] { + const auto spent = static_cast( + std::chrono::duration_cast(std::chrono::steady_clock::now() - started) + .count()); + m_counters.maxOfferNs = std::max(m_counters.maxOfferNs, spent); + }; + if (m_closing || m_counters.transportFailed) { + // Offered after Close() began or after the peer went: it can never be written. + ++m_counters.abandoned; + return; + } + const bool lossy = level <= MOBILEGL_LOG_LEVEL_WARN; + const bool fatal = level >= MOBILEGL_LOG_LEVEL_FATAL; + if (lossy && m_lossyBytes + length > m_limits.lossyQueueBytes) { + // The first drop of a gap stamps when it began; the notice that closes it names that. + if (m_gapLossyLines == 0) m_gapLossySince = MG_Util::Debug::GetCurrentTime(); + ++m_counters.lossyDropped; + m_counters.lossyDroppedBytes += length; + ++m_gapLossyLines; + m_gapLossyBytes += length; + noteOfferTime(); + return; + } + // FATAL is never refused: the process is about to end, and its last word is the one line + // the cap must not cost. The cap still bounds everything that can repeat. + if (!lossy && !fatal && m_errorBytes + length > m_limits.errorQueueBytes) { + ++m_counters.errorsCoalesced; + ++m_gapErrorLines; + noteOfferTime(); + return; + } + // A lossy line that fits closes a <= WARN gap; any line closes an ERROR gap. + QueueGapNoticesLocked(/*lossyGapToo=*/lossy); + Item item; + item.kind = Kind::Line; + item.level = level; + item.seq = m_nextSeq++; + item.text.assign(line, length); + const std::uint64_t seq = item.seq; + m_queue.push_back(std::move(item)); + if (lossy) m_lossyBytes += length; + else m_errorBytes += length; + if (fatal) ++m_urgent; + m_cv.notify_all(); + if (!fatal) { + noteOfferTime(); + return; + } + // FATAL: wait - bounded - until the sender has written this line and everything before it. + const auto deadline = started + std::chrono::milliseconds(m_limits.fatalFlushMs); + if (!m_cv.wait_until(lock, deadline, + [this, seq] { return m_writtenSeq >= seq || m_counters.transportFailed; })) { + ++m_counters.fatalFlushTimeouts; + } + } + + void LogForwardChannel::OfferControlFrame(std::vector frame) { + std::lock_guard lock(m_mutex); + if (m_closing || m_counters.transportFailed) { + ++m_counters.abandoned; + return; + } + // A gap before the acknowledgement is marked before it: the client reads the file after the + // ack arrives, and a gap it cannot see there is a gap it will read as "nothing was said". + QueueGapNoticesLocked(/*lossyGapToo=*/true); + Item item; + item.kind = Kind::Control; + item.seq = m_nextSeq++; + item.frame = std::move(frame); + m_queue.push_back(std::move(item)); + ++m_urgent; + m_cv.notify_all(); + } + + MobileGLResult LogForwardChannel::Send(const Item& item) { + if (item.kind == Kind::Control) { + return m_transport.SendFrame(MobileGLByteSpan{item.frame.data(), item.frame.size()}); + } + flatbuffers::FlatBufferBuilder builder(item.text.size() + 64); + const auto text = builder.CreateString(item.text); + const auto log = Wire::CreateLogLine(builder, WireLogLevelFor(item.level), text); + const auto envelope = Wire::CreateCtrlEnvelope(builder, Wire::CtrlMsg::LogLine, log.Union()); + Wire::FinishCtrlEnvelopeBuffer(builder, envelope); + return m_transport.SendFrame(MobileGLByteSpan{builder.GetBufferPointer(), builder.GetSize()}); + } + + void LogForwardChannel::Run() { +#if defined(__linux__) + pthread_setname_np(pthread_self(), "mgl-log-fwd"); +#endif + // THIS THREAD NEVER LOGS. It is the logger's sink: a line it logged would come straight back + // to its own queue (and in the non-session shape, through the log mutex a FATAL Offer can be + // waiting behind). What it has to say is in the counters, which Close() hands to its owner. + std::unique_lock lock(m_mutex); + for (;;) { + m_cv.wait(lock, [this] { return m_stop || !m_queue.empty(); }); + if (m_queue.empty()) return; // stopped, and nothing left + if (m_counters.transportFailed) { + // The peer is gone (a write failed): nothing queued can be written any more. + m_counters.abandoned += m_queue.size(); + m_writtenSeq = m_queue.back().seq; + m_queue.clear(); + m_lossyBytes = 0; + m_errorBytes = 0; + m_urgent = 0; + m_cv.notify_all(); + continue; + } + Item& front = m_queue.front(); + if (front.kind == Kind::Line && front.level == MOBILEGL_LOG_LEVEL_ERROR && !IsUrgentLocked()) { + // THE ERROR PACE. A token bucket, refilled by the time since it was last looked at. + // A line with no token waits - in the queue, in order - until one accrues, or until + // something urgent (a FATAL line, a LogFlush ack, Close) makes the pace moot. + const auto now = std::chrono::steady_clock::now(); + const double elapsed = std::chrono::duration(now - m_tokensAt).count(); + m_tokens = std::min(static_cast(m_limits.errorBurstLines), + m_tokens + elapsed * static_cast(m_limits.errorLinesPerSecond)); + m_tokensAt = now; + if (m_tokens < 1.0) { + if (!front.paced) { + front.paced = true; + ++m_counters.errorsPaced; + } + const double rate = std::max(1.0, static_cast(m_limits.errorLinesPerSecond)); + const auto untilToken = std::chrono::duration_cast( + std::chrono::duration((1.0 - m_tokens) / rate)); + m_cv.wait_for(lock, untilToken, [this] { return IsUrgentLocked() || m_stop; }); + continue; + } + m_tokens -= 1.0; + } + Item item = std::move(front); + m_queue.pop_front(); + if (item.kind == Kind::Line) { + if (item.level <= MOBILEGL_LOG_LEVEL_WARN) m_lossyBytes -= item.text.size(); + else m_errorBytes -= item.text.size(); + } + if (item.kind == Kind::Control || (item.kind == Kind::Line && item.level >= MOBILEGL_LOG_LEVEL_FATAL)) { + --m_urgent; + } + m_sending = true; + lock.unlock(); + const MobileGLResult sent = Send(item); + lock.lock(); + m_sending = false; + m_writtenSeq = item.seq; + if (sent == MOBILEGL_OK) { + if (item.kind == Kind::Line) ++m_counters.forwarded; + } else if (sent == MOBILEGL_ERR_INVALID_ARGUMENT) { + // Refused on its own shape (over the frame cap), not because the peer is gone. + ++m_counters.abandoned; + } else { + ++m_counters.abandoned; + m_counters.transportFailed = true; + } + m_cv.notify_all(); + } + } + + LogForwardCounters LogForwardChannel::Close() { + bool unblock = false; + { + std::unique_lock lock(m_mutex); + if (m_closed) return m_counters; + if (!m_closing) { + m_closing = true; + // The last gap is marked too: a client reading the file after the session sees + // that the tail it has is not the whole tail. + QueueGapNoticesLocked(/*lossyGapToo=*/true); + m_cv.notify_all(); + const auto drained = [this] { + return (m_queue.empty() && !m_sending) || m_counters.transportFailed; + }; + if (!m_cv.wait_for(lock, std::chrono::milliseconds(m_limits.closeDrainMs), drained)) { + // Not taken within the budget: the peer is not reading. The rest is abandoned, + // and a sender inside a write is blocked on that peer and needs unblocking. + m_counters.abandoned += m_queue.size(); + if (!m_queue.empty()) m_writtenSeq = m_queue.back().seq; + m_queue.clear(); + m_lossyBytes = 0; + m_errorBytes = 0; + m_urgent = 0; + unblock = m_sending; + } + } + m_stop = true; + m_cv.notify_all(); + } + if (unblock && m_unblock != nullptr) m_unblock(m_unblockUser); + if (m_sender.joinable()) m_sender.join(); + std::lock_guard lock(m_mutex); + m_closed = true; + return m_counters; + } + + LogForwardCounters LogForwardChannel::Counters() const { + std::lock_guard lock(m_mutex); + return m_counters; + } + +} // namespace MobileGL::MG_Remote::Transport diff --git a/MobileGL/MG_Remote/Transport/LogForward.h b/MobileGL/MG_Remote/Transport/LogForward.h new file mode 100644 index 00000000..7c5c6066 --- /dev/null +++ b/MobileGL/MG_Remote/Transport/LogForward.h @@ -0,0 +1,195 @@ +// MobileGL - MobileGL/MG_Remote/Transport/LogForward.h +// Copyright (c) 2025-2026 MobileGL-Dev +// Licensed under the GNU Lesser General Public License v3.0: +// https://www.gnu.org/licenses/gpl-3.0.txt +// https://www.gnu.org/licenses/lgpl-3.0.txt +// SPDX-License-Identifier: LGPL-3.0-only +// End of Source File Header + +// P9 W3. THE SERVER'S LOG LINES ON THE CONTROL PLANE, GRADED, AND NEVER A REASON TO STOP APPLYING. +// +// On TCP the server forwards every line it logs to its client as a LogLine control frame, and the +// client files it under `.server.log` (ControlInbox.cpp). Until this file the forward was a +// blocking SocketTransport::SendFrame made by the LOGGING THREAD, inside the logger: a client that +// stopped reading its control connection - a GL thread parked in a debugger, a ControlInbox whose +// bounded queue filled, a peer that simply stopped - filled the socket, and the next MGLOG on the +// apply thread never returned. Every record behind it waited; a forfeit that would have ended the +// session by name (ServerSession::ForfeitReverseChannel) was itself a log line and never finished +// being said. Design/04 §8.1 had written the policy down from the start (<= WARN lossy, >= ERROR +// lossless and rate limited), and LogLine.level has been on the wire for it since P6; nothing +// used either, and every line went out as Info. +// +// THE SHAPE. A LogForwardChannel owns a queue and ONE sender thread; the forwarder the logger +// calls (Offer) only enqueues, so the logging thread never touches the socket. The sender is the +// only thread that writes log frames, and it writes them in the order they were offered. +// +// THE POLICY, per severity (MOBILEGL_LOG_LEVEL_*): +// * DEBUG / INFO / WARN are LOSSY. A line that would take the queue's lossy bytes past +// LogForwardLimits::lossyQueueBytes is dropped and COUNTED. The drop is not silent to the +// client either: when the gap closes - the next line at or below WARN that fits, a LogFlush +// ack, or Close - the channel puts a WARN line of its own in the stream, `LogForward{Dropped} - +// N line(s) at or below WARN (B bytes) were not forwarded, from