feat(agc): add opt-in bounded submission progress

core/libs/prx/libSceAgcDriver/CMakeLists.txt

@@ -1809,6 +1809,24 @@ target_include_directories(agc_driver_image_table_cache_bounds_tests PRIVATE $<T
 
 target_include_directories(agc_driver_draw_cache_relocation_tests PRIVATE $<TARGET_PROPERTY:shader_recompiler_agc_driver,INTERFACE_INCLUDE_DIRECTORIES>)
 
+if(BUILD_TESTING)
+    add_test(NAME agc_submit_progress COMMAND agc_driver_submit_progress_tests)
+    set_tests_properties(agc_submit_progress PROPERTIES
+            ENVIRONMENT "APS5_SUBMIT_PROGRESS=1;ANYPS5_REQUIRE_VULKAN=1" TIMEOUT 30 SKIP_RETURN_CODE 77 WORKING_DIRECTORY "${CMAKE_SOURCE_DIR}")
+    if(WIN32)
+        set_tests_properties(agc_submit_progress PROPERTIES ENVIRONMENT_MODIFICATION
+                "PATH=path_list_prepend:$<TARGET_FILE_DIR:libSceAgcDriver>;PATH=path_list_prepend:$<TARGET_FILE_DIR:libc>")
+    endif()
+    add_test(NAME agc_submit_progress_no_timeline COMMAND agc_driver_submit_progress_tests --fallback-only)
+    set_tests_properties(agc_submit_progress_no_timeline PROPERTIES
+            ENVIRONMENT "APS5_SUBMIT_PROGRESS=1;APS5_NO_TIMELINE=1;ANYPS5_REQUIRE_VULKAN=1"
+            TIMEOUT 30 SKIP_RETURN_CODE 77 WORKING_DIRECTORY "${CMAKE_SOURCE_DIR}")
+    if(WIN32)
+        set_tests_properties(agc_submit_progress_no_timeline PROPERTIES ENVIRONMENT_MODIFICATION
+                "PATH=path_list_prepend:$<TARGET_FILE_DIR:libSceAgcDriver>;PATH=path_list_prepend:$<TARGET_FILE_DIR:libc>")
+    endif()
+endif()
+
 if(BUILD_TESTING)
     add_test(NAME agc_driver_store_pending_identity COMMAND agc_driver_store_over_untracked_tests --pending-identity)
     set_tests_properties(agc_driver_store_pending_identity PROPERTIES

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

@@ -106,9 +106,9 @@ class Driver {
     static void forgetUnfinishedWrites(QueueWorker& worker, std::span<const std::uint64_t> labelWrites);
     static bool& completionDeferred();
     void deferCompletion(const Submission& submission);
-    void submitOpenWork();
+    void submitOpenWork(const std::shared_ptr<SubmissionProgress>& progress = {});
     void wakeQueueZeroForCompletions();
-    void completeSubmission(std::uint64_t serial, std::uint64_t received, std::span<const std::uint64_t> labelWrites);
+    void completeSubmission(std::uint64_t serial, std::uint64_t received, std::span<const std::uint64_t> labelWrites, const std::shared_ptr<SubmissionProgress>& progress);
     static bool waitFree(const Submission& submission);
     bool queue0Before(std::uint64_t received) const;
     bool orderReleased(std::uint32_t queue, std::uint64_t received) const;

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

@@ -3,6 +3,7 @@
 
 #include "prx/libSceAgcDriver/Execution/include/Driver/Shaders/ShaderRegistry.hpp"
 #include "prx/libSceAgcDriver/Execution/include/VideoOutput.hpp"
+#include "prx/libSceAgcDriver/Execution/include/Driver/Queues/SubmissionProgress.hpp"
 #include <atomic>
 #include <chrono>
 #include <cstddef>
@@ -22,6 +23,7 @@ struct Submission {
     std::uint64_t serial;
     std::uint32_t queue;
     std::vector<std::uint32_t> commands;
+    std::shared_ptr<SubmissionProgress> progress;
 
     std::shared_ptr<const ShaderRegistry> shaders;
     std::map<std::size_t, std::shared_ptr<IFlipRequest>> flips;

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

@@ -0,0 +1,44 @@
+#ifndef CORE_LIBS_PRX_LIBSCEAGCDRIVER_EXECUTION_INCLUDE_DRIVER_QUEUES_SUBMISSIONPROGRESS_HPP
+#define CORE_LIBS_PRX_LIBSCEAGCDRIVER_EXECUTION_INCLUDE_DRIVER_QUEUES_SUBMISSIONPROGRESS_HPP
+
+#include <chrono>
+#include <condition_variable>
+#include <cstdint>
+#include <memory>
+#include <mutex>
+#include <utility>
+
+namespace AgcDriver { class VulkanDevice; }
+
+namespace AgcDriver::DriverDetail {
+
+class SubmissionProgress {
+public:
+    using Clock = std::chrono::steady_clock;
+    using Ready = std::pair<std::shared_ptr<VulkanDevice>, std::uint64_t>;
+
+    void Publish(std::shared_ptr<VulkanDevice> device, std::uint64_t epoch) {
+        {
+            std::lock_guard lock(mutex);
+            result = {std::move(device), epoch};
+            ready = true;
+        }
+        changed.notify_one();
+    }
+
+    Ready WaitUntil(Clock::time_point deadline) {
+        std::unique_lock lock(mutex);
+        if (!changed.wait_until(lock, deadline, [&] { return ready; })) return {};
+        return result;
+    }
+
+private:
+    std::mutex mutex;
+    std::condition_variable changed;
+    Ready result;
+    bool ready = false;
+};
+
+}
+
+#endif

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

@@ -86,6 +86,8 @@ class VulkanDevice {
     bool CanWaitUnlocked() const;
     std::uint64_t SubmitAndEpoch();
     void WaitRecorded(std::uint64_t serial);
+    std::uint64_t SubmittedEpoch() const;
+    bool WaitRecordedFor(std::uint64_t serial, std::uint64_t timeoutNanos);
     void ReapRecorded(std::uint64_t serial);
     // Completes the batches that already finished (non-blocking); under the mutex.
     void ReapRecorded();

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

@@ -464,8 +464,8 @@ void Driver::execute(const Submission& submission) {
             return;
         }
         if ((std::atomic_ref<std::uint32_t>(*const_cast<std::uint32_t*>(submission.rewindTail - 1)).load(std::memory_order_acquire) & 0x80000000u) != 0) {
-            DrawPipeline::Queue0().Enqueue([this] {
-                submitOpenWork();
+            DrawPipeline::Queue0().Enqueue([this, progress = submission.progress] {
+                submitOpenWork(progress);
                 wakeQueueZeroForCompletions();
             }, {});
             executeRewindTail(submission);
@@ -488,7 +488,10 @@ void Driver::execute(const Submission& submission) {
         if (localDevice != nullptr) localDevice->SubmitRecorded(submission.queue == 0);
     }
     endTiming.Finish();
-    if (submission.rewindTail != nullptr) executeRewindTail(submission);
+    if (submission.rewindTail != nullptr) {
+        if (submission.progress) submitOpenWork(submission.progress);
+        executeRewindTail(submission);
+    }
 }
 
 }

core/libs/prx/libSceAgcDriver/Execution/src/Driver/Queues/QueueWorker.cpp

@@ -70,19 +70,23 @@ void Driver::deferCompletion(const Submission& submission) {
         queue0Uncommitted.push_back(received);
     }
     completionDeferred() = true;
-    DrawPipeline::Queue0().Enqueue([this, serial = submission.serial, received, labelWrites = submission.labelWrites] {
-        completeSubmission(serial, received, labelWrites);
+    DrawPipeline::Queue0().Enqueue([this, serial = submission.serial, received, labelWrites = submission.labelWrites, progress = submission.progress] {
+        completeSubmission(serial, received, labelWrites, progress);
     }, {});
 }
 
-void Driver::submitOpenWork() {
-    if (deferredLabels().labels.empty() && !Graphics::Recorder::PendingLabelSince().has_value() && Graphics::Recorder::RecordedWorkSinceSubmit() == 0) return;
+void Driver::submitOpenWork(const std::shared_ptr<SubmissionProgress>& progress) {
+    const bool work = !deferredLabels().labels.empty() || Graphics::Recorder::PendingLabelSince().has_value() || Graphics::Recorder::RecordedWorkSinceSubmit() != 0;
+    if (!work && !progress) return;
     ++submissionCosts(0).endSubmits;
     GuestMemory::TagGpuLockSite(GuestMemory::GpuLockSite::End);
     std::lock_guard gpuLock(GuestMemory::GpuMutex());
     const auto localDevice = device.Load();
-    recordDeferredLabels(localDevice.get(), 0);
-    if (localDevice != nullptr) localDevice->SubmitRecorded(true);
+    if (work) {
+        recordDeferredLabels(localDevice.get(), 0);
+        if (localDevice != nullptr) localDevice->SubmitRecorded(true);
+    }
+    if (progress) progress->Publish(localDevice, localDevice ? localDevice->SubmittedEpoch() : 0);
 }
 
 void Driver::wakeQueueZeroForCompletions() {
@@ -91,7 +95,7 @@ void Driver::wakeQueueZeroForCompletions() {
     changed.notify_all();
 }
 
-void Driver::completeSubmission(std::uint64_t serial, std::uint64_t received, std::span<const std::uint64_t> labelWrites) {
+void Driver::completeSubmission(std::uint64_t serial, std::uint64_t received, std::span<const std::uint64_t> labelWrites, const std::shared_ptr<SubmissionProgress>& progress) {
     struct Complete {
         Driver& driver;
         std::uint64_t serial, received;
@@ -109,7 +113,7 @@ void Driver::completeSubmission(std::uint64_t serial, std::uint64_t received, st
             if (notify) driver.changed.notify_all();
         }
     } complete{*this, serial, received, labelWrites};
-    submitOpenWork();
+    submitOpenWork(progress);
 }
 
 const std::atomic<std::uint64_t>*& Driver::workerQueued() {
@@ -261,6 +265,7 @@ void Driver::run(std::uint32_t id) noexcept {
                 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);
+            if (submission.progress && submission.rewindTail == nullptr && !completionDeferred()) submitOpenWork(submission.progress);
             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{};
             PerformanceContext completionContext(id == 0 ? frameTiming() : FrameTiming::Async());

core/libs/prx/libSceAgcDriver/Execution/src/Driver/Queues/Submission.cpp

@@ -1,3 +1,4 @@
+#include "prx/libSceAgcDriver/Execution/include/ProfileOutput.hpp"
 #include "prx/libSceAgcDriver/Execution/include/PerformanceTimer.hpp"
 #include "prx/libSceAgcDriver/Execution/include/Driver/Driver.hpp"
 #include "prx/libSceAgcDriver/Execution/include/Driver/Queues/Submission.hpp"
@@ -21,6 +22,56 @@ namespace AgcDriver::DriverDetail {
 
 namespace {
 
+bool& submitsQueue0() {
+    struct SubmitsQueue0Storage {};
+    return HostThreadLocal<bool, SubmitsQueue0Storage>();
+}
+
+bool progressEligible(const Submission& submission) {
+    if (submission.queue != 0 || !submission.flips.empty()
+        || !submission.renderingWaits.empty() || !submission.conditionalEnds.empty()) return false;
+    bool terminalRelease = false;
+    std::optional<Pm4::LabelWrite> knownWrite;
+    for (std::size_t at = 0; at < submission.commands.size();) {
+        const auto packet = std::span(submission.commands).subspan(at);
+        const auto count = Pm4::PacketWords(packet[0]);
+        if (count == 0 || count > packet.size()) return false;
+        const auto opcode = (packet[0] >> 8u) & 0xffu;
+        const bool type3 = (packet[0] >> 30u) == 3;
+        if (type3 && (opcode == 0x3c || opcode == 0x93)) {
+            if (!knownWrite || Pm4::Predicated(packet[0]) || ((packet[1] >> 4u) & 3u) != 1u) return false;
+            const auto wait = Pm4::DecodeWait(packet.first(count));
+            const auto bytes = knownWrite->Bytes();
+            if (wait.address < knownWrite->address) return false;
+            const auto offset = wait.address - knownWrite->address;
+            if (offset > bytes.size() || wait.bytes > bytes.size() - offset) return false;
+            std::uint64_t value = 0;
+            std::memcpy(&value, bytes.data() + offset, wait.bytes);
+            if (!Pm4::WaitComparesValue(packet.first(count), value)) return false;
+            terminalRelease = false;
+            at += count;
+            continue;
+        }
+        if (type3 && opcode == 0x59 && submission.rewindTail != nullptr && at + count == submission.commands.size()) {
+            at += count;
+            continue;
+        }
+        if (!Pm4::FillerPacket(packet[0]) && !(type3 && opcode == 0x10) && !Pm4::IsTagMarker(packet.first(count))) {
+            if (type3 && (opcode == 0x49 || opcode == 0x37)) {
+                const auto write = Pm4::DecodeLabelWrite(packet.first(count));
+                terminalRelease = opcode == 0x49 && write.has_value();
+                if (write && !Pm4::Predicated(packet[0]) && (opcode == 0x37 || (packet[2] >> 29u) < 3)) knownWrite = *write;
+                else if (opcode != 0x49 || (packet[2] >> 29u) != 0) knownWrite.reset();
+            } else {
+                terminalRelease = false;
+                if (!type3 || (opcode != 0x42 && opcode != 0x58)) knownWrite.reset();
+            }
+        }
+        at += count;
+    }
+    return terminalRelease;
+}
+
 std::uint64_t captureTime() {
     return static_cast<std::uint64_t>(std::llround(TraceMs() * 1000.0));
 }
@@ -38,11 +89,6 @@ bool EopRelease(std::span<const std::uint32_t> words) {
     return words.size() >= 8 && (words[0] >> 30u) == 3u && Pm4::PacketWords(words[0]) == 8u && ((words[0] >> 8u) & 0xffu) == 0x49u && ((words[2] >> 24u) & 7u) == 2u;
 }
 
-bool& submitsQueue0() {
-    struct SubmitsQueue0Storage {};
-    return HostThreadLocal<bool, SubmitsQueue0Storage>();
-}
-
 }
 
 std::deque<const std::uint32_t*>& Driver::releasedTails() {
@@ -366,6 +412,10 @@ void Driver::Submit(const Packet* packet, std::uint32_t queue, bool cpuFlip) {
     const auto validated = profile ? std::chrono::steady_clock::now() : start;
     auto* const capture = SubmissionCapture::Active();
     auto captured = capture != nullptr ? SubmissionCapture::Prepare(queue, SubmissionKind::Submit, submission.commands, submission.conditionalEnds) : CapturedSubmission{};
+    static const bool progressEnabled = std::getenv("APS5_SUBMIT_PROGRESS") != nullptr;
+    const auto progress = progressEnabled && progressEligible(submission) ? std::make_shared<SubmissionProgress>() : nullptr;
+    submission.progress = progress;
+    const auto progressDeadline = progress ? std::chrono::steady_clock::now() + std::chrono::milliseconds(2) : std::chrono::steady_clock::time_point{};
     {
         std::unique_lock lock(mutex);
         rethrowFailure();
@@ -406,6 +456,25 @@ void Driver::Submit(const Packet* packet, std::uint32_t queue, bool cpuFlip) {
     }
     changed.notify_all();
     if (capture != nullptr) writeCaptured(*this, *capture, captured);
+    if (progress) {
+        PerformanceTimer timing("Driver.SubmitProgress");
+        const auto [localDevice, epoch] = progress->WaitUntil(progressDeadline);
+        const auto now = std::chrono::steady_clock::now();
+        bool gpuReady = false;
+        if (localDevice && epoch != 0 && now < progressDeadline) {
+            const auto remaining = std::chrono::duration_cast<std::chrono::nanoseconds>(progressDeadline - now).count();
+            gpuReady = localDevice->WaitRecordedFor(epoch, static_cast<std::uint64_t>(remaining));
+        }
+        if (profile) {
+            static std::atomic<std::uint64_t> eligible{0}, ready{0}, expired{0};
+            if (gpuReady) ready.fetch_add(1, std::memory_order_relaxed);
+            else expired.fetch_add(1, std::memory_order_relaxed);
+            const auto count = eligible.fetch_add(1, std::memory_order_relaxed) + 1;
+            if (count % 1000 == 0) AgcDriver::ProfilePrint_nid_no_patch("[submit-progress] %llu eligible terminal releases, %llu GPU-ready returns, %llu expired or unsupported\n",
+                static_cast<unsigned long long>(count), static_cast<unsigned long long>(ready.load()), static_cast<unsigned long long>(expired.load()));
+        }
+        CheckFailure();
+    }
 }
 
 void Driver::SuspendPoint() {

core/libs/prx/libSceAgcDriver/Execution/src/VulkanDevice.cpp

@@ -2202,6 +2202,14 @@ void VulkanDevice::WaitRecorded(std::uint64_t serial) {
     if (state->recorder) state->recorder->WaitSerial(serial);
 }
 
+std::uint64_t VulkanDevice::SubmittedEpoch() const {
+    return state->recorder ? state->recorder->NewestSubmitted() : 0;
+}
+
+bool VulkanDevice::WaitRecordedFor(std::uint64_t serial, std::uint64_t timeoutNanos) {
+    return state->recorder && state->recorder->WaitSerialFor(serial, timeoutNanos);
+}
+
 void VulkanDevice::ReapRecorded(std::uint64_t serial) {
     if (!state->recorder) return;
     // The drain's site in the [recorder] table is this call's caller (Driver::execute's drain,

core/libs/prx/libSceAgcDriver/Graphics/include/Recorder.hpp

@@ -205,6 +205,7 @@ class Recorder {
     // it touches only the immutable timeline semaphore (the caller keeps the device alive). Requires
     // HasTimeline(); the debug switch APS5_NO_TIMELINE makes the device create the recorder without one.
     void WaitSerial(std::uint64_t serial);
+    bool WaitSerialFor(std::uint64_t serial, std::uint64_t timeoutNanos);
     // Under the mutex: finishes (completions, release) the in-flight batches up to `serial`, from the
     // front only; tolerates batches another thread finished meanwhile. Later batches stay in flight.
     void FinishUpTo(std::uint64_t serial);

core/libs/prx/libSceAgcDriver/Graphics/src/Recorder.cpp

@@ -3536,6 +3536,20 @@ void Recorder::WaitSerial(std::uint64_t serial) {
     Check(result, "vkWaitSemaphoresKHR recorder");
 }
 
+bool Recorder::WaitSerialFor(std::uint64_t serial, std::uint64_t timeoutNanos) {
+    PerformanceTimer frameTiming("Recorder.WaitSerialFor");
+    if (timeline == VK_NULL_HANDLE || serial == 0) return false;
+    const UnlockedWaiter waiter{id};
+    VkSemaphoreWaitInfoKHR wait{VK_STRUCTURE_TYPE_SEMAPHORE_WAIT_INFO_KHR};
+    wait.semaphoreCount = 1;
+    wait.pSemaphores = &timeline;
+    wait.pValues = &serial;
+    const auto result = context.Function<PFN_vkWaitSemaphoresKHR>("vkWaitSemaphoresKHR")(context.device, &wait, timeoutNanos);
+    if (result == VK_TIMEOUT) return false;
+    Check(result, "vkWaitSemaphoresKHR submission progress");
+    return true;
+}
+
 void Recorder::FinishUpTo(std::uint64_t serial) {
     // A drain: the wait already happened (unlocked), the fences are signaled, so the source-0 wait
     // recorded here is the bookkeeping only (a caller that did not wait first, WaitForLeases, shows