fix(agc): run a wait-free submission of another queue after queue 0's earlier submissions

core/libs/prx/libSceAgcDriver/Execution/include/Driver/Driver.hpp

@@ -74,6 +74,10 @@ class Driver {
     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 @@ class Driver {
     bool resetGraphics = false;
 
     std::uint32_t idleWaiters = 0;
+    std::uint64_t queue0Executing = 0;
+    std::atomic<std::uint32_t> orderHolders{0};
+    std::atomic<std::uint32_t> runningWorkers{0};
+    std::atomic<std::uint64_t> queue0Awaited{0};
 
     std::atomic<std::uint64_t> evidenceReads{0};
     std::atomic<std::uint64_t> evidenceValidations{0};

core/libs/prx/libSceAgcDriver/Execution/include/Driver/Queues/Submission.hpp

@@ -26,6 +26,7 @@ struct Submission {
     std::map<std::size_t, std::shared_ptr<IFlipRequest>> flips;
     std::map<std::size_t, std::shared_ptr<IRenderingWait>> renderingWaits;
     bool suspend = false;
+    bool waitFree = false;
 
     std::uint64_t received = 0;
     std::vector<std::uint64_t> labelWrites;

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::uint64_t>(std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::steady_clock::now() - submission.enqueuedAt).count());
             }
-            execute(submission);
+            {
+                struct Running {
+                    std::atomic<std::uint32_t>& count;
+                    explicit Running(std::atomic<std::uint32_t>& 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<unsigned long long>(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;

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<std::uint32_t> control(*const_cast<std::uint32_t*>(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<std::uint64_t>(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<std::size_t>(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));

core/libs/prx/libSceAgcDriver/Execution/src/Driver/Synchronization/WaitMemory.cpp

@@ -254,6 +254,12 @@ void Driver::waitMemory(std::span<const std::uint32_t> 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();

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<std::uint32_t> 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<std::uint32_t, 5> words{0xc0031500, 1, 1, 1, 0x41};
     Packet packet{words.data(), static_cast<std::uint32_t>(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");