fix(agc): don't release a label wait on a store its queue already took and reset

A WAIT_REG_MEM passes when the label history holds a store recorded after
the wait's submission entered the driver with a value the compare accepts
(storedSince). A title that resets a label right after each wait on it,
with several submissions of the waiting queue queued at once (each waiting
for the label at its start), had all of those waits released by one store
of the other queue: after the first wait took it and the queue reset the
label, the history still held the store, newer than the later submissions.

Each queue now notes when a wait of it on a dword passed and, when it then
stores to that dword itself, that a store of another queue recorded before
the pass was taken: such a store no longer releases the queue's later waits
on those bytes. Stores the queue makes itself, and the other routes (memory,
the pending-label table, held at submit), are unchanged.

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

@@ -217,7 +217,9 @@ private:
     static PendingDrawPhases& pendingDrawPhases();
     void addDrawPhases(const std::array<double, DrawDriverPhaseCount>& ms, bool drawn, std::uint64_t captures);
     void noteLabelStore(std::uint64_t address, std::span<const std::byte> bytes, std::uint64_t stamp, std::uint32_t queue);
-    bool storedSince(std::span<const std::uint32_t> packet, std::uint64_t address, std::size_t bytes, std::uint64_t received, std::uint32_t* writer = nullptr);
+    void noteLabelPass(std::span<const std::uint32_t> packet, std::uint32_t queue);
+    void noteLabelIssue(std::span<const std::uint32_t> packet, std::uint32_t queue);
+    bool storedSince(std::span<const std::uint32_t> packet, std::uint64_t address, std::size_t bytes, std::uint64_t received, std::uint32_t queue, std::uint32_t* writer = nullptr);
     static void traceLabel(std::span<const std::uint32_t> packet, std::uint32_t queue);
     static std::array<WriteRecord, 16384>& writeHistory();
     static std::size_t& writeCursor();
@@ -458,6 +460,8 @@ private:
 
     AgcDriver::Mutex labelStoresMutex;
     std::unordered_map<std::uint64_t, std::array<LabelStore, LabelStoreHistory>> labelStores;
+    std::unordered_map<std::uint64_t, std::uint64_t> labelPasses;
+    std::unordered_map<std::uint64_t, std::uint64_t> labelTaken;
 
     std::vector<std::shared_ptr<VulkanDevice>> replacedDevices;
     std::stop_token shutdownToken = LibcShutdownToken_nid_postfix();

core/libs/prx/libSceAgcDriver/Execution/src/Driver/Packets/PacketExecution.cpp

@@ -274,6 +274,7 @@ void Driver::execute(const Submission& submission) {
         if (APS5_ENABLE_TIMING_LOG && submission.queue == 0 && pendingFrameTiming == nullptr) includeTimingSubmission(submission, false);
         PerformanceContext timingContext(submission.queue == 0 ? pendingFrameTiming.get() : FrameTiming::Async());
         PerformanceTimer timing("Driver.Packet");
+        if (!predicated) noteLabelIssue(packet, submission.queue);
         if (pipelined && !predicated && header != FlipPacketHeader && header != RenderingWaitPacketHeader && !submission.flips.contains(cursor) && (enqueueLabelPacket(packet, opcode, submission.queue) || enqueueDmaPacket(packet, opcode, submission.queue, queue))) {
             traceLabel(packet, submission.queue);
             recent.Record(cursor);
@@ -287,6 +288,7 @@ void Driver::execute(const Submission& submission) {
             if (value.has_value() && Pm4::WaitComparesValue(packet, *value)) {
                 ++waitOutcomes().fromRecorderSameQueue;
                 gateOpened() = true;
+                noteLabelPass(packet, submission.queue);
                 recent.Record(cursor);
                 cursor = nextCursor;
                 continue;
@@ -295,6 +297,7 @@ void Driver::execute(const Submission& submission) {
                 bumpEpoch(&EpochBumps::waits);
                 ++waitOutcomes().atEntry;
                 gateOpened() = true;
+                noteLabelPass(packet, submission.queue);
                 recent.Record(cursor);
                 cursor = nextCursor;
                 continue;
@@ -317,6 +320,7 @@ void Driver::execute(const Submission& submission) {
                 cursor = opcode == 0x3f ? submission.conditionalEnds.at(cursor) : nextCursor;
                 continue;
             }
+            noteLabelIssue(packet, submission.queue);
         }
         if (opcode == 0x3f) {
             cursor = nextCursor;
@@ -455,6 +459,7 @@ void Driver::execute(const Submission& submission) {
             static const bool traceGpu = std::getenv("APS5_TRACE_GPU") != nullptr;
             const auto waitStart = traceGpu ? std::chrono::steady_clock::now() : std::chrono::steady_clock::time_point{};
             timed(&WorkerProfile::waitMs, [&] { waitMemory(packet, submission.queue, recent, submission.received, submission.heldAtSubmit.contains(cursor)); });
+            noteLabelPass(packet, submission.queue);
             gateOpened() = true;
             if (traceGpu && std::chrono::steady_clock::now() - waitStart > std::chrono::milliseconds(200)) {
 

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

@@ -1,6 +1,7 @@
 #include "prx/libSceAgcDriver/Execution/include/Driver/Driver.hpp"
 #include "prx/libSceAgcDriver/Execution/include/Pm4.hpp"
 #include "prx/libSceAgcDriver/Execution/include/Mutex.hpp"
+#include <algorithm>
 #include <cstdlib>
 #include <cstring>
 #include <string_view>
@@ -19,14 +20,54 @@ void Driver::noteLabelStore(std::uint64_t address, std::span<const std::byte> by
     }
 }
 
-bool Driver::storedSince(std::span<const std::uint32_t> packet, std::uint64_t address, std::size_t bytes, std::uint64_t received, std::uint32_t* writer) {
+namespace {
+
+std::uint64_t queueDword(std::uint32_t queue, std::uint64_t dword) {
+    return dword | (static_cast<std::uint64_t>(queue) << 56u);
+}
+
+}
+
+void Driver::noteLabelPass(std::span<const std::uint32_t> packet, std::uint32_t queue) {
+    const auto opcode = (packet[0] >> 8u) & 0xffu;
+    if ((opcode != 0x3c && opcode != 0x93) || packet.size() < 7 || ((packet[1] >> 4u) & 3u) != 1u) return;
+    const auto bytes = Pm4::WaitAwaitedBytes(packet);
+    const std::uint64_t address = (packet[2] & (bytes == 8 ? ~7u : ~3u)) | (static_cast<std::uint64_t>(packet[3]) << 32u);
+    const auto passed = eventSerial.load(std::memory_order_acquire);
+    std::lock_guard lock(labelStoresMutex);
+    for (std::size_t offset = 0; offset < bytes; offset += 4) labelPasses[queueDword(queue, address + offset)] = passed;
+}
+
+void Driver::noteLabelIssue(std::span<const std::uint32_t> packet, std::uint32_t queue) {
+    switch ((packet[0] >> 8u) & 0xffu) {
+        case 0x37: case 0x40: case 0x49: case 0x50: case 0x83: break;
+        default: return;
+    }
+    const auto stores = Pm4::DecodeStores(packet);
+    if (stores.empty()) return;
+    std::lock_guard lock(labelStoresMutex);
+    if (labelPasses.empty()) return;
+    for (const auto& store : stores) {
+        if (store.conditional || !KeepsLabelStore(store.address, static_cast<std::size_t>(store.bytes))) continue;
+        for (std::uint64_t offset = 0; offset < store.bytes; offset += 4) {
+            const auto key = queueDword(queue, store.address + offset);
+            if (const auto pass = labelPasses.find(key); pass != labelPasses.end()) labelTaken[key] = pass->second;
+        }
+    }
+}
+
+bool Driver::storedSince(std::span<const std::uint32_t> packet, std::uint64_t address, std::size_t bytes, std::uint64_t received, std::uint32_t queue, std::uint32_t* writer) {
     std::lock_guard lock(labelStoresMutex);
     const auto low = labelStores.find(address);
     if (low == labelStores.end()) return false;
     const auto high = bytes == 8 ? labelStores.find(address + 4) : labelStores.end();
     if (bytes == 8 && high == labelStores.end()) return false;
+    std::uint64_t taken = 0;
+    for (std::size_t offset = 0; offset < bytes; offset += 4) {
+        if (const auto found = labelTaken.find(queueDword(queue, address + offset)); found != labelTaken.end()) taken = std::max(taken, found->second);
+    }
     for (const auto& store : low->second) {
-        if (store.stamp <= received) continue;
+        if (store.stamp <= received || (store.queue != queue && store.stamp <= taken)) continue;
         std::uint64_t value = store.value;
         if (bytes == 8) {
             const auto upper = std::find_if(high->second.begin(), high->second.end(), [&](const LabelStore& other) { return other.stamp == store.stamp; });

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

@@ -137,7 +137,7 @@ void Driver::waitMemory(std::span<const std::uint32_t> packet, std::uint32_t que
     } epochPoint;
     const auto storedByOwnLabel = [&] {
         std::uint32_t writer = queue;
-        return !requireMemory && storedSince(packet, awaited, awaitedBytes, received, &writer) && writer == queue;
+        return !requireMemory && storedSince(packet, awaited, awaitedBytes, received, queue, &writer) && writer == queue;
     };
     GuestMemory::CheckRange(reinterpret_cast<const void*>(awaited), awaitedBytes, awaitedBytes);
     if (Pm4::WaitSatisfiedUnchecked(packet)) {
@@ -146,7 +146,7 @@ void Driver::waitMemory(std::span<const std::uint32_t> packet, std::uint32_t que
         return;
     }
     if (heldAtSubmit) return;
-    if (std::uint32_t writer = queue; !requireMemory && storedSince(packet, awaited, awaitedBytes, received, &writer)) {
+    if (std::uint32_t writer = queue; !requireMemory && storedSince(packet, awaited, awaitedBytes, received, queue, &writer)) {
         if (writer != queue) notePendingWait(packet, queue, received);
         else epochPoint.bump = false;
         return;
@@ -342,7 +342,7 @@ void Driver::waitMemory(std::span<const std::uint32_t> packet, std::uint32_t que
     std::uint32_t polls = 0;
     StallWatch stall(queue);
     while (!Pm4::WaitSatisfiedUnchecked(packet)) {
-        if (std::uint32_t writer = queue; !requireMemory && storedSince(packet, awaited, awaitedBytes, received, &writer)) {
+        if (std::uint32_t writer = queue; !requireMemory && storedSince(packet, awaited, awaitedBytes, received, queue, &writer)) {
             if (writer != queue) notePendingWait(packet, queue, received);
             else epochPoint.bump = false;
             return;

core/libs/prx/libSceAgcDriver/tests/Submit.cpp

@@ -422,6 +422,35 @@ void testLabelHeldAtSubmission() {
     AgcDriverWaitIdle_nid_postfix();
 }
 
+void testLabelTakenAndResetInOneSubmission() {
+    alignas(64) static volatile std::uint32_t gate = 0, label = 0, first = 0, waiting = 0, second = 0;
+    submit(0x20, commands(waitEqual(&gate, 1), waitEqual(&label, 1), writeData(&label, 0), writeData(&first, 1), writeData(&waiting, 1), waitEqual(&label, 1), writeData(&second, 1)));
+    submit(0, commands(writeData(&label, 1)));
+    waitFor(&label, 1, "producer label never landed");
+    gate = 1;
+    waitFor(&waiting, 1, "consumer never reached its second label wait");
+    std::this_thread::sleep_for(std::chrono::milliseconds(1200));
+    check(first == 1 && second == 0, "a label store its own queue had taken and reset satisfied the queue's next wait, or the wait expired");
+    submit(0, commands(writeData(&label, 1)));
+    waitFor(&second, 1, "consumer did not resume after a new producer label");
+    AgcDriverWaitIdle_nid_postfix();
+}
+
+void testLabelTakenAndResetByEarlierSubmission() {
+    alignas(64) static volatile std::uint32_t gate = 0, label = 0, first = 0, second = 0;
+    submit(0x20, commands(waitEqual(&gate, 1), waitEqual(&label, 1), writeData(&label, 0), writeData(&first, 1)));
+    submit(0x20, commands(waitEqual(&label, 1), writeData(&label, 0), writeData(&second, 1)));
+    submit(0, commands(writeData(&label, 1)));
+    waitFor(&label, 1, "producer label never landed");
+    gate = 1;
+    waitFor(&first, 1, "consumer never passed its first label wait");
+    std::this_thread::sleep_for(std::chrono::milliseconds(1200));
+    check(second == 0, "a label store an earlier submission of the queue had taken and reset satisfied a later submission's wait, or the wait expired");
+    submit(0, commands(writeData(&label, 1)));
+    waitFor(&second, 1, "the later submission did not resume after a new producer label");
+    AgcDriverWaitIdle_nid_postfix();
+}
+
 void testWaitFreeSubmissionAfterEarlierQueue0Work() {
     alignas(64) static volatile std::uint32_t first = 0, second = 0;
     constexpr std::uint32_t writes = 20000;
@@ -1072,6 +1101,8 @@ int main(int argc, char** argv) {
         testLabelStoredSinceSubmission();
         testEndOfPipeLabelsWithoutWork();
         testLabelHeldAtSubmission();
+        testLabelTakenAndResetInOneSubmission();
+        testLabelTakenAndResetByEarlierSubmission();
         testWideLabelStoredSinceSubmission();
         testWaitFreeSubmissionAfterEarlierQueue0Work();
         testWaitFreeSubmissionQueue0WaitsOn();