From 04f12028dce890b9d0e46ed617485852ca08a831 Mon Sep 17 00:00:00 2001 From: Dean Galvin Date: Sun, 4 Oct 2026 01:46:28 +1300 Subject: [PATCH] fix(agc): run a wait-free submission of another queue after queue 0's earlier submissions PPSA21564 submits the queue 0 work of each intro video frame F first, then 1 to 5 ms later a 12 MiB DMA_DATA on queue 0x3a that refills the video texture F sampled, for frame F + 2 (two textures, used in turn). Nothing orders them but time: no label, wait or flip lies between the copy and the draw, which is the first packet of the frame's first submission. A GPU that keeps up starts each submission as it arrives, so the draw reads its texture right after the frame is submitted, and the copy submitted after it lands later, every frame. Here the queue workers record independently, so the copy was recorded either before or after the draw depending on how far queue 0's worker lagged: the video showed the new frame or the one before, and every switch between the two was a frame shown backwards. A submission with no waits on another queue now starts only after queue 0 has run every submission it received earlier, as on a GPU that keeps up. While queue 0 is blocked in a wait, the held submission is released when its queue stores the awaited label (it or a later submission of that queue), or when nothing else could store it: no other worker is running and no recorded GPU work or completion writes it. A GPU that keeps up would be blocked at that wait too and run the other queue's submission meanwhile, so the order holds only while queue 0 still has work it would already have done, and no wait can deadlock on it. --- .../Execution/include/Driver/Driver.hpp | 8 +++ .../include/Driver/Queues/Submission.hpp | 1 + .../src/Driver/Queues/QueueWorker.cpp | 19 ++++++- .../src/Driver/Queues/Submission.cpp | 56 +++++++++++++++++-- .../src/Driver/Synchronization/WaitMemory.cpp | 6 ++ .../libs/prx/libSceAgcDriver/tests/Submit.cpp | 51 +++++++++++++++++ 6 files changed, 135 insertions(+), 6 deletions(-) diff --git a/core/libs/prx/libSceAgcDriver/Execution/include/Driver/Driver.hpp b/core/libs/prx/libSceAgcDriver/Execution/include/Driver/Driver.hpp index 895548081..cbb38b90e 100644 --- a/core/libs/prx/libSceAgcDriver/Execution/include/Driver/Driver.hpp +++ b/core/libs/prx/libSceAgcDriver/Execution/include/Driver/Driver.hpp @@ -74,6 +74,10 @@ private: void enqueue(Submission submission); void noteHeldAtSubmit(Submission& submission, std::size_t cursor); static void forgetUnfinishedWrites(QueueWorker& worker, const Submission& submission); + static bool waitFree(const Submission& submission); + bool queue0Before(std::uint64_t received) const; + bool orderReleased(std::uint32_t queue, std::uint64_t received) const; + void noteWaitBlocked(std::uint32_t queue, std::uint64_t awaited, bool blocked); void reportPresents(double waitedMs, std::size_t inFlight); static bool stampValidate(); static bool dataHits(); @@ -288,6 +292,10 @@ private: bool resetGraphics = false; std::uint32_t idleWaiters = 0; + std::uint64_t queue0Executing = 0; + std::atomic orderHolders{0}; + std::atomic runningWorkers{0}; + std::atomic queue0Awaited{0}; std::atomic evidenceReads{0}; std::atomic evidenceValidations{0}; diff --git a/core/libs/prx/libSceAgcDriver/Execution/include/Driver/Queues/Submission.hpp b/core/libs/prx/libSceAgcDriver/Execution/include/Driver/Queues/Submission.hpp index dd35572bc..ed7da06d9 100644 --- a/core/libs/prx/libSceAgcDriver/Execution/include/Driver/Queues/Submission.hpp +++ b/core/libs/prx/libSceAgcDriver/Execution/include/Driver/Queues/Submission.hpp @@ -26,6 +26,7 @@ struct Submission { std::map> flips; std::map> renderingWaits; bool suspend = false; + bool waitFree = false; std::uint64_t received = 0; std::vector labelWrites; diff --git a/core/libs/prx/libSceAgcDriver/Execution/src/Driver/Queues/QueueWorker.cpp b/core/libs/prx/libSceAgcDriver/Execution/src/Driver/Queues/QueueWorker.cpp index 41161f6e7..dca6dd040 100644 --- a/core/libs/prx/libSceAgcDriver/Execution/src/Driver/Queues/QueueWorker.cpp +++ b/core/libs/prx/libSceAgcDriver/Execution/src/Driver/Queues/QueueWorker.cpp @@ -77,10 +77,24 @@ void Driver::run(std::uint32_t id) noexcept { } submission = std::move(pending.front()); pending.pop_front(); + if (id == 0) queue0Executing = submission.suspend ? 0 : submission.received; + if (submission.waitFree) { + orderHolders.fetch_add(1, std::memory_order_acq_rel); + while (!failure && !stopping && !orderReleased(id, submission.received)) changed.wait_for(lock, std::chrono::milliseconds(1)); + orderHolders.fetch_sub(1, std::memory_order_acq_rel); + rethrowFailure(); + } worker.queued.fetch_sub(1, std::memory_order_acq_rel); if (profile && submission.enqueuedAt != std::chrono::steady_clock::time_point{}) costs.dequeueNs += static_cast(std::chrono::duration_cast(std::chrono::steady_clock::now() - submission.enqueuedAt).count()); } - execute(submission); + { + struct Running { + std::atomic& count; + explicit Running(std::atomic& count) : count(count) { count.fetch_add(1, std::memory_order_acq_rel); } + ~Running() { count.fetch_sub(1, std::memory_order_acq_rel); } + } running{runningWorkers}; + execute(submission); + } if (traceGpu) std::fprintf(stderr, "[gpu] %.1f done serial=%llu queue=0x%x\n", TraceMs(), static_cast(submission.serial), id); const auto completeStart = profile ? std::chrono::steady_clock::now() : std::chrono::steady_clock::time_point{}; bool notify = true; @@ -89,8 +103,9 @@ void Driver::run(std::uint32_t id) noexcept { rethrowFailure(); markCompleted(submission.serial); forgetUnfinishedWrites(workers.at(id), submission); + if (id == 0) queue0Executing = 0; - notify = idleWaiters != 0; + notify = idleWaiters != 0 || orderHolders.load(std::memory_order_acquire) != 0; } if (notify) changed.notify_all(); else ++costs.notifiesSkipped; diff --git a/core/libs/prx/libSceAgcDriver/Execution/src/Driver/Queues/Submission.cpp b/core/libs/prx/libSceAgcDriver/Execution/src/Driver/Queues/Submission.cpp index 2206962a3..4dbe52c02 100644 --- a/core/libs/prx/libSceAgcDriver/Execution/src/Driver/Queues/Submission.cpp +++ b/core/libs/prx/libSceAgcDriver/Execution/src/Driver/Queues/Submission.cpp @@ -88,10 +88,18 @@ void Driver::reserveOutputs(Submission& submission) { void Driver::executeRewindTail(const Submission& stalled) { std::atomic_ref control(*const_cast(stalled.rewindTail - 1)); - while ((control.load(std::memory_order_acquire) & 0x80000000u) == 0) { - CheckFailure(); - checkStopping(); - PollSleep(); + if ((control.load(std::memory_order_acquire) & 0x80000000u) == 0) { + noteWaitBlocked(stalled.queue, reinterpret_cast(stalled.rewindTail - 1), true); + struct BlockedWait { + Driver& driver; + std::uint32_t queue; + ~BlockedWait() { driver.noteWaitBlocked(queue, 0, false); } + } blocked{*this, stalled.queue}; + while ((control.load(std::memory_order_acquire) & 0x80000000u) == 0) { + CheckFailure(); + checkStopping(); + PollSleep(); + } } Submission tail{}; tail.queue = stalled.queue; @@ -173,8 +181,48 @@ void Driver::SuspendPoint() { changed.notify_all(); } +bool Driver::waitFree(const Submission& submission) { + if (submission.queue == 0 || submission.suspend || !submission.flips.empty() || !submission.renderingWaits.empty() || submission.rewindTail != nullptr) return false; + for (std::size_t at = 0; at < submission.commands.size(); at += std::max(1, Pm4::PacketWords(submission.commands[at]))) { + const auto header = submission.commands[at]; + const auto opcode = (header >> 8u) & 0xffu; + if ((header >> 30u) == 3u && (opcode == 0x3c || opcode == 0x93)) return false; + } + return true; +} + +bool Driver::queue0Before(std::uint64_t received) const { + if (queue0Executing != 0 && queue0Executing < received) return true; + const auto worker = workers.find(0); + if (worker == workers.end()) return false; + for (const auto& pending : worker->second.pending) { + if (pending.suspend) continue; + return pending.received < received; + } + return false; +} + +bool Driver::orderReleased(std::uint32_t queue, std::uint64_t received) const { + if (!queue0Before(received)) return true; + const auto awaited = queue0Awaited.load(std::memory_order_acquire); + if (awaited == 0) return false; + if (workers.at(queue).unfinishedWrites.contains(awaited & ~std::uint64_t{3})) return true; + return runningWorkers.load(std::memory_order_acquire) == 0 && !completionsPending() && !Graphics::Recorder::SnapshotWriteOverlaps(awaited, 4); +} + +void Driver::noteWaitBlocked(std::uint32_t queue, std::uint64_t awaited, bool blocked) { + if (blocked) runningWorkers.fetch_sub(1, std::memory_order_acq_rel); + else runningWorkers.fetch_add(1, std::memory_order_acq_rel); + if (queue == 0) queue0Awaited.store(blocked ? awaited : 0, std::memory_order_release); + if (blocked && orderHolders.load(std::memory_order_acquire) != 0) { + std::lock_guard lock(mutex); + changed.notify_all(); + } +} + void Driver::enqueue(Submission submission) { const auto queue = submission.queue; + submission.waitFree = waitFree(submission); auto& worker = workers[queue]; for (const auto dword : submission.labelWrites) ++worker.unfinishedWrites[dword]; worker.pending.push_back(std::move(submission)); diff --git a/core/libs/prx/libSceAgcDriver/Execution/src/Driver/Synchronization/WaitMemory.cpp b/core/libs/prx/libSceAgcDriver/Execution/src/Driver/Synchronization/WaitMemory.cpp index 20ba4c1aa..72a41535e 100644 --- a/core/libs/prx/libSceAgcDriver/Execution/src/Driver/Synchronization/WaitMemory.cpp +++ b/core/libs/prx/libSceAgcDriver/Execution/src/Driver/Synchronization/WaitMemory.cpp @@ -254,6 +254,12 @@ void Driver::waitMemory(std::span packet, std::uint32_t que if (entryTryFailed && pollService(false)) return; + noteWaitBlocked(queue, awaited, true); + struct BlockedWait { + Driver& driver; + std::uint32_t queue; + ~BlockedWait() { driver.noteWaitBlocked(queue, 0, false); } + } blocked{*this, queue}; static const bool pauseSpin = std::getenv("APS5_NO_PAUSE_SPIN") == nullptr; const auto spinLimit = queue == 0 ? std::chrono::microseconds(1500) : std::chrono::microseconds(100); const auto spinStart = std::chrono::steady_clock::now(); diff --git a/core/libs/prx/libSceAgcDriver/tests/Submit.cpp b/core/libs/prx/libSceAgcDriver/tests/Submit.cpp index 29b596b07..e2b8d991f 100644 --- a/core/libs/prx/libSceAgcDriver/tests/Submit.cpp +++ b/core/libs/prx/libSceAgcDriver/tests/Submit.cpp @@ -295,6 +295,53 @@ void testLabelHeldAtSubmission() { AgcDriverWaitIdle_nid_postfix(); } +void testWaitFreeSubmissionAfterEarlierQueue0Work() { + alignas(64) static volatile std::uint32_t first = 0, second = 0; + constexpr std::uint32_t writes = 20000; + std::vector words; + for (std::uint32_t i = 1; i <= writes; ++i) { + const auto write = writeData(&first, i); + words.insert(words.end(), write.begin(), write.end()); + } + submit(0, words); + submit(0x20, commands(writeData(&second, 1))); + waitFor(&second, 1, "the wait-free submission never ran"); + check(first == writes, "a wait-free submission ran before queue 0's earlier submission finished"); + AgcDriverWaitIdle_nid_postfix(); +} + +void testWaitFreeSubmissionQueue0WaitsOn() { + alignas(64) static volatile std::uint32_t label = 0, done = 0; + submit(0, commands(waitEqual(&label, 1), writeData(&done, 1))); + submit(0x20, commands(writeData(&label, 1))); + check(waitFor(&done, 1, "queue 0 never passed the wait the held submission satisfies") < std::chrono::milliseconds(500), "queue 0's wait on a held wait-free submission's label was not released at once"); + AgcDriverWaitIdle_nid_postfix(); +} + +void testWaitFreeSubmissionBehindHeldOne() { + alignas(64) static volatile std::uint32_t other = 0, label = 0, done = 0; + submit(0, commands(waitEqual(&label, 1), writeData(&done, 1))); + submit(0x20, commands(writeData(&other, 1))); + submit(0x20, commands(writeData(&label, 1))); + check(waitFor(&done, 1, "queue 0 never passed the wait a later submission of the held queue satisfies") < std::chrono::milliseconds(500), "queue 0's wait on a label stored behind a held submission was not released at once"); + check(other == 1, "the held submission did not run before the one behind it"); + AgcDriverWaitIdle_nid_postfix(); +} + +void testWaitFreeSubmissionTheCpuWaitsFor() { + alignas(64) static volatile std::uint32_t stored = 0, flag = 0, done = 0; + submit(0, commands(waitEqual(&flag, 1), writeData(&done, 1))); + submit(0x20, commands(writeData(&stored, 1))); + std::thread title([] { + waitFor(&stored, 1, "the held submission never ran while queue 0 waited on the CPU"); + flag = 1; + }); + const auto waited = waitFor(&done, 1, "queue 0 never passed a wait the CPU satisfies after the held submission"); + title.join(); + check(waited < std::chrono::milliseconds(500), "a held submission the CPU waits for was not released while queue 0 waited on the CPU"); + AgcDriverWaitIdle_nid_postfix(); +} + void testWorkerFailure() { std::array words{0xc0031500, 1, 1, 1, 0x41}; Packet packet{words.data(), static_cast(words.size()), 0, {}}; @@ -324,6 +371,10 @@ int main() { testEndOfPipeLabelsWithoutWork(); testLabelHeldAtSubmission(); testWideLabelStoredSinceSubmission(); + testWaitFreeSubmissionAfterEarlierQueue0Work(); + testWaitFreeSubmissionQueue0WaitsOn(); + testWaitFreeSubmissionBehindHeldOne(); + testWaitFreeSubmissionTheCpuWaitsFor(); testWorkerFailure(); check(expectFailure([] { LibcRunShutdown_nid_postfix(); }).find("required shader register") != std::string::npos, "shutdown lost worker failure"); std::puts("AGC driver submit tests passed");