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");