diff --git a/MobileGL/MG_Remote/Server/ServerLoop.cpp b/MobileGL/MG_Remote/Server/ServerLoop.cpp index f3c4e675..d64e54bb 100644 --- a/MobileGL/MG_Remote/Server/ServerLoop.cpp +++ b/MobileGL/MG_Remote/Server/ServerLoop.cpp @@ -210,6 +210,7 @@ namespace MobileGL::MG_Remote::Server { } m_session = &session; m_stopRequested.store(false, std::memory_order_release); + m_abandonQueue.store(false, std::memory_order_release); // PH-6 fix round: the session's copy of the request, for ReserveEventOrBlock's wait. Lowered // with the loop's own, so a loop restarted on the same session does not forfeit at once. session.ClearApplyStopRequest(); @@ -506,13 +507,22 @@ namespace MobileGL::MG_Remote::Server { // a peer that keeps publishing could keep this final drain going as long as it liked, // which is exactly the unbounded hold the forfeit exists to end. The queue is left // unapplied; the rings die with the session. + // + // AND AFTER AbandonQueuedRecords (P12 review fix), for the forfeit's reason: the server is + // stopping under a client that may still be streaming, and draining what it keeps queuing + // is the unbounded hold Stop()'s bounded join turns into an abort. const Bool forfeited = session.ReverseChannelForfeited(); + const Bool abandoned = m_abandonQueue.load(std::memory_order_acquire); PumpControlRequest(); if (forfeited) { MGLOG_E("MG_Remote server: mgl-srv-apply stops on ReverseChannelForfeit - %llu reverse-channel " "event(s) dropped, applied seq %llu; the records still queued are not applied", static_cast(session.ForfeitDrops()), static_cast(session.Consumer().AppliedSeq())); + } else if (abandoned) { + MGLOG_W("MG_Remote server: mgl-srv-apply stops because the server is stopping - applied seq %llu; " + "the records still queued are not applied", + static_cast(session.Consumer().AppliedSeq())); } else { DrainRing(); } @@ -712,12 +722,25 @@ namespace MobileGL::MG_Remote::Server { // // The idle poll's cost is unchanged: this load replaces the function-top check the first // version made, and the empty-ring answer is still it plus Pop's cmdHead load. + // + // P12 (D6) REVIEW FIX: A LOST SERVER WINDOW IS ANSWERED BEFORE EVERY POP TOO, not only by + // PumpControlRequest between batches. A batch ends when the ring is empty, and a client + // streaming frames keeps it from emptying for seconds: surfaceDestroyed (ServerDisplay:: + // Detach) waited that long on the UI thread - past its 3 s bound, so it gave up and the + // session went on presenting into the destroyed window. Checked AFTER the latch check (the + // MECHANICS row in ph_latch_sites.py pins that one as the loop's first statement); the + // release latches ServerWindowLost, so `continue` takes the latch check's way out. One more + // acquire load per pop; the empty-ring answer pays it once. ServerSession& session = *m_session; Transport::SessionConsumer& consumer = session.Consumer(); PipeApplier& applier = session.Applier(); Uint64 applied = 0; for (;;) { if (SessionLatched()) break; + if (m_windowLostRequested.load(std::memory_order_acquire)) { + ReleaseLostServerWindow(); + continue; + } bool corrupt = false; const bool popped = consumer.ApplyOne( [this, &applier](const Transport::RingRecordView& record) { @@ -756,6 +779,9 @@ namespace MobileGL::MG_Remote::Server { // loop-head check on the next iteration - ph_latch_sites.py's MECHANICS row pins that // check as the first statement of this loop.) if (session.ReverseChannelForfeited()) break; + // P12 review fix: and the record in hand is the last one when the server is stopping + // (AbandonQueuedRecords); same placement, same free empty-ring answer. + if (m_abandonQueue.load(std::memory_order_acquire)) break; // TEST-ONLY scheduling point BETWEEN two records: after this record's own checks and // before the next pop's latch check. A control-thread latch stored here is seen by that // check (the first version let it through); one stored after the check and before the @@ -935,6 +961,8 @@ namespace MobileGL::MG_Remote::Server { m_exitCv.notify_all(); } + void ServerLoop::AbandonQueuedRecords() { m_abandonQueue.store(true, std::memory_order_release); } + void ServerLoop::Stop() { if (!m_thread.joinable()) { // Never started, or already stopped. The backend may still exist - the hook builds diff --git a/MobileGL/MG_Remote/Server/ServerLoop.h b/MobileGL/MG_Remote/Server/ServerLoop.h index e0699296..c1ebb5eb 100644 --- a/MobileGL/MG_Remote/Server/ServerLoop.h +++ b/MobileGL/MG_Remote/Server/ServerLoop.h @@ -216,6 +216,15 @@ namespace MobileGL::MG_Remote::Server { // Kill the doorbell, join the thread (bounded), then destroy the private backend object // ON THAT THREAD before it exits. Blocking by contract - see the header note. void Stop(); + // P12 review fix: THE NEXT Stop() LEAVES THE QUEUE UNAPPLIED. A Stop() normally lets the + // apply thread finish its batch and drain what is still queued - the client's own drain + // bounds that. A SERVER that is stopping (the in-process display server's + // mobilegl_server_stop_inprocess) has a client that may still be streaming, and a batch plus + // a final drain of a ring that client keeps refilling can outlast the bounded join + // (Fatal{ApplyThreadJoinTimeout} aborts the display Activity's process). Called before Stop(): + // the drain stops after the record in hand and the exit path declines the rest, as a + // ReverseChannelForfeit does. Reset by Start(). + void AbandonQueuedRecords(); Bool Running() const; @@ -486,6 +495,8 @@ namespace MobileGL::MG_Remote::Server { ServerSession* m_session = nullptr; std::atomic m_running{false}; std::atomic m_stopRequested{false}; + // P12 review fix: AbandonQueuedRecords() - read after every popped record and at the exit. + std::atomic m_abandonQueue{false}; std::thread m_thread; // The apply thread's identity used to live here as an atomic. It is // Detail::g_applyThreadKey now - see the block at the top of this header for why the diff --git a/MobileGL/MG_Remote/Server/ServerMain.cpp b/MobileGL/MG_Remote/Server/ServerMain.cpp index 951887aa..0d09eb9b 100644 --- a/MobileGL/MG_Remote/Server/ServerMain.cpp +++ b/MobileGL/MG_Remote/Server/ServerMain.cpp @@ -437,6 +437,7 @@ int RunSession(std::unique_ptr control, std::vector control, std::vector g_streamRecords{0}; + std::thread g_streamDetachThread; + std::atomic g_streamDetachResult{-1}; + std::atomic g_streamDetachMs{-1}; + + void StreamOneRecordAndLoseTheWindowAfterTheFirst() { + if (g_streamRecords.fetch_add(1, std::memory_order_acq_rel) == 0) { + g_streamDetachThread = std::thread([] { + const auto started = std::chrono::steady_clock::now(); + const Server::ServerWindowDetach detached = Server::ServerDisplayInstance().Detach(3000); + g_streamDetachMs.store(std::chrono::duration_cast( + std::chrono::steady_clock::now() - started) + .count(), + std::memory_order_release); + g_streamDetachResult.store(static_cast(detached), std::memory_order_release); + }); + // Detach clears the window and calls the lost hook under one lock hold, so once the window + // reads detached the request is published. + (void)PollUntil([] { return !Server::ServerDisplayInstance().Attached(); }, 1000); + } + std::this_thread::sleep_for(std::chrono::milliseconds(kStreamRecordMs)); + } + + [[noreturn]] void LoseTheServerWindowMidStreamAndExit() { + Remote::ArmSessionLatch(); + ServerFixture fixture; + if (!fixture.Handshake() || !fixture.StartLoop()) ::_exit(64); + if (!fixture.WaitUntilTrulyParked()) ::_exit(65); + Server::ServerLoop& loop = Server::ServerLoopInstance(); + Server::ServerDisplay& display = Server::ServerDisplayInstance(); + display.Install(CountingWindowHooks::Hooks()); + display.Attach(&g_fakeServerWindow, 64, 48); + ServerWindowProbe probe; + probe.width = 64; + probe.height = 48; + if (loop.RunProbeOnApplyThreadForTesting(&AcquireServerWindowOnTheApplyThread, &probe) != MOBILEGL_OK) ::_exit(66); + int failed = 0; + if (probe.rc != MOBILEGL_OK || probe.lease.window != &g_fakeServerWindow) failed |= kStreamNotLeased; + if (!fixture.WaitUntilTrulyParked()) ::_exit(67); + loop.SetBetweenRecordsHookForTesting(&StreamOneRecordAndLoseTheWindowAfterTheFirst); + // LEGAL draws under ONE publish (ALatchFromAnotherThread...'s records), so only the lost window + // can cut the batch short. + 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}; + Uint64 last = Codec::kInvalidSeq; + for (int i = 0; i < kStreamRecords; ++i) { + last = fixture.encoder.EncodeRecord(MG_Pipe::MGPWireOp::DrawVbo, &info, sizeof(info), &range, + sizeof(range)); + if (last == Codec::kInvalidSeq) ::_exit(68); + } + fixture.encoder.Publish(); + fixture.producer.PublishAndNotify(last); + + if (!PollUntil([] { return g_streamDetachResult.load(std::memory_order_acquire) >= 0; }, 8000)) ::_exit(69); + if (g_streamDetachThread.joinable()) g_streamDetachThread.join(); + if (g_streamDetachResult.load() != static_cast(Server::ServerWindowDetach::ReleasedBySession)) + failed |= kStreamDetach; + if (g_streamDetachMs.load() >= 1000) failed |= kStreamDetachSlow; + if (!PollUntil([&] { return !loop.Running(); }, 3000)) failed |= kStreamThreadStayed; + if (loop.DrainedRecords() > 2) failed |= kStreamDrainedOn; + if (!Remote::SessionLatched() || Remote::SessionLatchedFamily() != Remote::MGFatalFamily::ServerWindowLost) + failed |= kStreamNotLatched; + std::fprintf(stderr, "[stream] Detach answered %s in %lld ms after %llu of %d streamed records\n", + Server::ServerWindowDetachName( + static_cast(g_streamDetachResult.load())), + g_streamDetachMs.load(), static_cast(loop.DrainedRecords()), + kStreamRecords); + loop.SetBetweenRecordsHookForTesting(nullptr); + fixture.Stop(); + ::_exit(failed); + } +} // namespace + +// P12 review fix (major). Red with DrainRing's pre-pop lost-window check deleted: the whole 4 s batch +// runs under the UI thread's Detach, which gives up at its 3 s bound and answers TimedOut (bits 2|4|8). +TEST(ServerLoopLatchTest, ALostServerWindowIsReleasedMidBatchWhileTheClientStreams) { + EXPECT_EXIT(LoseTheServerWindowMidStreamAndExit(), ::testing::ExitedWithCode(0), ".*") + << "bits: 1 = the lease was not taken; 2 = Detach did not answer ReleasedBySession; 4 = Detach took " + "1 s or more; 8 = more than two records applied after the window went; 16 = no ServerWindowLost " + "latch; 32 = the apply thread stayed (64+ = setup)"; +} + +namespace { + // P12 review fix: A STOPPING SERVER UNDER A STREAMING CLIENT. 200 records at 40 ms each is an 8 s + // stream, longer than Stop()'s 5 s bounded join. The display server's stop marks the queue + // abandoned first; the join must then come back within a record or two. + std::atomic g_stopStreamRecords{0}; + void StreamSlowly() { + g_stopStreamRecords.fetch_add(1, std::memory_order_acq_rel); + std::this_thread::sleep_for(std::chrono::milliseconds(40)); + } + + enum : int { + kStopSlow = 1, // Stop() took 1 s or more + kStopDrainedOn = 2, // more than three records were applied + }; + + [[noreturn]] void StopTheServerMidStreamAndExit() { + Remote::ArmSessionLatch(); + ServerFixture fixture; + if (!fixture.Handshake() || !fixture.StartLoop()) ::_exit(64); + if (!fixture.WaitUntilTrulyParked()) ::_exit(65); + Server::ServerLoop& loop = Server::ServerLoopInstance(); + loop.SetBetweenRecordsHookForTesting(&StreamSlowly); + 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}; + Uint64 last = Codec::kInvalidSeq; + for (int i = 0; i < 200; ++i) { + last = fixture.encoder.EncodeRecord(MG_Pipe::MGPWireOp::DrawVbo, &info, sizeof(info), &range, + sizeof(range)); + if (last == Codec::kInvalidSeq) ::_exit(66); + } + fixture.encoder.Publish(); + fixture.producer.PublishAndNotify(last); + if (!PollUntil([] { return g_stopStreamRecords.load(std::memory_order_acquire) >= 1; }, 3000)) ::_exit(67); + // RunSession's order when mobilegl_server_stop_inprocess raised the stop. + const auto started = std::chrono::steady_clock::now(); + loop.AbandonQueuedRecords(); + loop.Stop(); + const auto stopMs = + std::chrono::duration_cast(std::chrono::steady_clock::now() - started).count(); + int failed = 0; + if (stopMs >= 1000) failed |= kStopSlow; + if (loop.DrainedRecords() > 3) failed |= kStopDrainedOn; + std::fprintf(stderr, "[stop] Stop() returned in %lld ms after %llu of 200 streamed records\n", + static_cast(stopMs), static_cast(loop.DrainedRecords())); + loop.SetBetweenRecordsHookForTesting(nullptr); + fixture.Stop(); + ::_exit(failed); + } +} // namespace + +// P12 review fix. Red with AbandonQueuedRecords a no-op: the batch and the exit-path drain apply the +// whole stream, Stop()'s join gives up at 5 s and Fatal{ApplyThreadJoinTimeout} aborts the process. +TEST(ServerLoopTest, StoppingTheServerUnderAStreamingClientLeavesTheQueueAndJoinsPromptly) { + EXPECT_EXIT(StopTheServerMidStreamAndExit(), ::testing::ExitedWithCode(0), ".*") + << "bits: 1 = Stop() took 1 s or more; 2 = more than three records applied after the stop " + "(64+ = setup; SIGABRT = Fatal{ApplyThreadJoinTimeout})"; +} + // D3. A server that owns no display - the host, the exec'd supervisor's children - refuses a request // for its window BY NAME and does NOT latch: it is the client's configuration, not corrupt bytes. Red // with the refusal turned into a latch: this unarmed process aborts.