fix(agc): stop and join the driver's thread pools at shutdown

core/libs/prx/libSceAgcDriver/CMakeLists.txt

@@ -1493,6 +1493,8 @@ target_include_directories(agc_driver_pipeline_journal_tests PRIVATE $<TARGET_PR
 
 target_include_directories(agc_driver_buffer_descriptor_runtime_tests PRIVATE $<TARGET_PROPERTY:shader_recompiler_agc_driver,INTERFACE_INCLUDE_DIRECTORIES>)
 
+target_include_directories(agc_driver_shutdown_threads_tests PRIVATE $<TARGET_PROPERTY:shader_recompiler_agc_driver,INTERFACE_INCLUDE_DIRECTORIES>)
+
 target_include_directories(agc_driver_prepared_publication_tests PRIVATE $<TARGET_PROPERTY:shader_recompiler_agc_driver,INTERFACE_INCLUDE_DIRECTORIES>)
 
 target_include_directories(agc_driver_prepared_graphics_tests PRIVATE $<TARGET_PROPERTY:shader_recompiler_agc_driver,INTERFACE_INCLUDE_DIRECTORIES>)
@@ -1848,6 +1850,17 @@ if(BUILD_TESTING)
             set_tests_properties(${preparedTest} PROPERTIES ENVIRONMENT_MODIFICATION "PATH=path_list_prepend:$<TARGET_FILE_DIR:libc>")
         endif()
     endforeach()
+    add_test(NAME agc_driver_shutdown_threads_joined COMMAND agc_driver_shutdown_threads_tests --joined)
+    add_test(NAME agc_driver_shutdown_threads_compare_helpers COMMAND agc_driver_shutdown_threads_tests --compare-helpers)
+    add_test(NAME agc_driver_shutdown_threads_draw_committer COMMAND agc_driver_shutdown_threads_tests --draw-committer)
+    add_test(NAME agc_driver_shutdown_threads_shader_cache_writer COMMAND agc_driver_shutdown_threads_tests --shader-cache-writer)
+    add_test(NAME agc_driver_shutdown_threads_late_shader_cache COMMAND agc_driver_shutdown_threads_tests --late-shader-cache)
+    add_test(NAME agc_driver_shutdown_threads_async_log_held COMMAND agc_driver_shutdown_threads_tests --async-log-held)
+    add_test(NAME agc_driver_shutdown_threads_exit_after_start COMMAND agc_driver_shutdown_threads_tests --exit-after-start)
+    set_tests_properties(agc_driver_shutdown_threads_joined agc_driver_shutdown_threads_compare_helpers agc_driver_shutdown_threads_draw_committer agc_driver_shutdown_threads_shader_cache_writer agc_driver_shutdown_threads_late_shader_cache agc_driver_shutdown_threads_async_log_held agc_driver_shutdown_threads_exit_after_start PROPERTIES ENVIRONMENT "ANYPS5_NO_SHADER_CACHE=1" SKIP_RETURN_CODE 77 WORKING_DIRECTORY "${CMAKE_SOURCE_DIR}" TIMEOUT 30)
+    if(WIN32)
+        set_tests_properties(agc_driver_shutdown_threads_joined agc_driver_shutdown_threads_compare_helpers agc_driver_shutdown_threads_draw_committer agc_driver_shutdown_threads_shader_cache_writer agc_driver_shutdown_threads_late_shader_cache agc_driver_shutdown_threads_async_log_held agc_driver_shutdown_threads_exit_after_start PROPERTIES ENVIRONMENT_MODIFICATION "PATH=path_list_prepend:$<TARGET_FILE_DIR:libc>")
+    endif()
     add_test(NAME agc_driver_compute_user_data_calls_registered COMMAND agc_driver_compute_user_data_calls_tests --registered)
     set_tests_properties(agc_driver_compute_user_data_calls_registered PROPERTIES ENVIRONMENT "ANYPS5_NO_SHADER_CACHE=1" SKIP_RETURN_CODE 77 WORKING_DIRECTORY "${CMAKE_SOURCE_DIR}")
     if(WIN32)

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

@@ -10,6 +10,7 @@
 #include <cstdio>
 #include <cstdlib>
 #include <deque>
+#include <initializer_list>
 #include <map>
 #include <memory>
 #include <mutex>
@@ -171,6 +172,16 @@ class AsyncLog {
         return true;
     }
 
+    static void FlushAtEnd() {
+        auto& log = Get();
+        if (log.abandoned.load(std::memory_order_acquire)) return;
+        if (!log.FlushFor(EndFlushLimit)) log.abandoned.store(true, std::memory_order_release);
+    }
+
+    std::unique_lock<std::mutex> Hold() {
+        return std::unique_lock(mutex);
+    }
+
 private:
     struct Window {
         Clock::time_point first;
@@ -193,14 +204,14 @@ class AsyncLog {
         if (!asynchronous) return;
         LogSink().store(&AsyncLog::Sink, std::memory_order_release);
         std::atexit(&AsyncLog::FlushAtEnd);
+        InitializeWaits();
         std::thread([this] { Run(); }).detach();
         LibcSetEndFlush_nid_no_patch(&AsyncLog::FlushAtEnd);
     }
 
-    static void FlushAtEnd() {
-        auto& log = Get();
-        if (log.abandoned.load(std::memory_order_acquire)) return;
-        if (!log.FlushFor(EndFlushLimit)) log.abandoned.store(true, std::memory_order_release);
+    void InitializeWaits() {
+        std::unique_lock lock(mutex);
+        for (auto* condition : {&wake, &space, &idle}) static_cast<void>(condition->wait_for(lock, std::chrono::nanoseconds(0)));
     }
 
     static int Sink(std::FILE* stream, const char* text, std::size_t length) {
@@ -383,6 +394,12 @@ inline void FlushAsyncLog_nid_no_patch() {
     AsyncLog::Get().Flush();
 }
 
+inline void FlushAsyncLogAtEnd_nid_no_patch() {
+    AsyncLog::FlushAtEnd();
+}
+
+std::unique_lock<std::mutex> HoldAsyncLog();
+
 template<typename... TArgs>
 void LogCoalescedStderr_nid_no_patch(const char* format, TArgs... args) {
     char small[1024];

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

@@ -30,9 +30,11 @@ class DrawPipeline {
     static bool& Active();
     static std::atomic<std::uint64_t>& EpochToken();
     static void FollowEpoch(std::uint64_t token);
+    static unsigned Committers();
 
     void Enqueue(Commit commit, std::vector<Range> writes, std::uint64_t labelAddress = 0, std::vector<std::byte> labelBytes = {});
     void Drain(DrainReason reason, std::uint32_t opcode = 0x100);
+    void Stop();
     static std::uint64_t DrainRequestsByOpcode(std::uint32_t opcode);
     bool Busy() const { return outstanding.load(std::memory_order_acquire) != 0; }
     std::size_t Outstanding() const { return outstanding.load(std::memory_order_acquire); }
@@ -60,6 +62,7 @@ class DrawPipeline {
     std::uint32_t idleWaiters = 0;
     bool committerWaiting = false;
     bool stopped = false;
+    bool closing = false;
     std::thread thread;
     std::uint64_t commits = 0;
     std::uint64_t commitNs = 0;

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

@@ -119,6 +119,7 @@ void AwaitRegistrationPreparation();
 void SubmitLibraryPrebuild(std::function<void()> task);
 void AwaitLibraryPrebuilds();
 void CloseRegistrationPreparation();
+unsigned RegistrationPreparationThreads();
 
 void ResolvePreparedGraphics(const ShaderSnapshot& front, const std::shared_ptr<const ShaderSnapshot>& fragment, std::uint32_t primitiveType, const ShaderRecompiler::SpirvTarget& target);
 

core/libs/prx/libSceAgcDriver/Execution/src/Driver/Draw/DrawPipeline.cpp

@@ -14,6 +14,8 @@ namespace {
 
 constexpr std::array<const char*, static_cast<std::size_t>(DrawPipeline::DrainReason::Count)> DrainNames{"packet", "flush", "labels", "capture", "indirect", "submission"};
 
+std::atomic<unsigned> runningCommitters{0};
+
 std::uint64_t elapsedNs(std::chrono::steady_clock::time_point since) {
     return static_cast<std::uint64_t>(std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::steady_clock::now() - since).count());
 }
@@ -59,6 +61,10 @@ void DrawPipeline::FollowEpoch(std::uint64_t token) {
     GuestMemory::BumpCollectEpoch();
 }
 
+unsigned DrawPipeline::Committers() {
+    return runningCommitters.load(std::memory_order_acquire);
+}
+
 void DrawPipeline::rethrowFailure() {
     if (failure == nullptr) return;
     auto error = std::exchange(failure, nullptr);
@@ -69,6 +75,11 @@ void DrawPipeline::Enqueue(Commit commit, std::vector<Range> writes, std::uint64
     static const bool profile = std::getenv("APS5_PROFILE_DRAW") != nullptr;
     std::unique_lock lock(mutex);
     rethrowFailure();
+    if (closing) {
+        lock.unlock();
+        commit();
+        return;
+    }
     if (items.size() >= Depth()) {
         if (GuestMemory::GpuMutex().HeldByThisThread()) throw std::runtime_error("draw pipeline full while the enqueuing thread holds the GPU lock");
         const auto start = std::chrono::steady_clock::now();
@@ -82,7 +93,10 @@ void DrawPipeline::Enqueue(Commit commit, std::vector<Range> writes, std::uint64
     ++enqueued;
     items.push_back({std::move(commit), std::move(writes), labelAddress, std::move(labelBytes)});
     outstanding.fetch_add(1, std::memory_order_release);
-    if (!thread.joinable()) thread = std::thread([this] { run(); });
+    if (!thread.joinable()) {
+        thread = std::thread([this] { run(); });
+        runningCommitters.fetch_add(1, std::memory_order_acq_rel);
+    }
     if (committerWaiting) wake.notify_one();
     if (!profile) return;
     const auto now = std::chrono::steady_clock::now();
@@ -122,8 +136,9 @@ void DrawPipeline::run() {
         {
             std::unique_lock lock(mutex);
             committerWaiting = true;
-            wake.wait(lock, [&] { return !items.empty(); });
+            wake.wait(lock, [&] { return !items.empty() || closing; });
             committerWaiting = false;
+            if (items.empty()) break;
             commit = stopped ? nullptr : &items.front().commit;
         }
         const auto start = std::chrono::steady_clock::now();
@@ -155,6 +170,20 @@ void DrawPipeline::run() {
         outstanding.fetch_sub(1, std::memory_order_release);
         if (idleWaiters != 0) idle.notify_all();
     }
+    runningCommitters.fetch_sub(1, std::memory_order_acq_rel);
+}
+
+void DrawPipeline::Stop() {
+    std::thread committer;
+    {
+        std::lock_guard lock(mutex);
+        closing = true;
+        committer = std::move(thread);
+    }
+    wake.notify_all();
+    if (!committer.joinable()) return;
+    if (committer.get_id() == std::this_thread::get_id()) committer.detach();
+    else committer.join();
 }
 
 void DrawPipeline::Drain(DrainReason reason, std::uint32_t opcode) {

core/libs/prx/libSceAgcDriver/Execution/src/Driver/Lifecycle.cpp

@@ -2,13 +2,17 @@
 #include "prx/libSceAgcDriver/Graphics/include/PipelineJournal.hpp"
 #include "prx/libSceAgcDriver/Execution/include/Driver/Driver.hpp"
 #include "prx/libSceAgcDriver/Execution/include/Driver/Diagnostics.hpp"
+#include "prx/libSceAgcDriver/Execution/include/Driver/Draw/DrawPipeline.hpp"
 #include "prx/libSceAgcDriver/Execution/include/Driver/Shaders/ShaderRegistry.hpp"
 #include "prx/libSceAgcDriver/Execution/include/GuestMemory.hpp"
 #include "prx/libSceAgcDriver/Execution/include/WorkerSampler.hpp"
 #include "prx/libSceAgcDriver/Execution/include/PerformanceTrace.hpp"
 #include "prx/libSceAgcDriver/Execution/include/ProfileOutput.hpp"
 #include "prx/libSceAgcDriver/Graphics/include/GuestBufferMemory.hpp"
+#include "prx/libSceAgcDriver/Graphics/include/ParallelCompare.hpp"
+#include "prx/libSceAgcDriver/Graphics/include/PipelineLibrary.hpp"
 #include "prx/libc/include/GuestAllocations.hpp"
+#include "ShaderDiskCache.hpp"
 #include <atomic>
 #include <cstdint>
 #include <cstdio>
@@ -115,15 +119,19 @@ void Driver::stop() {
     for (auto& [queue, worker] : workers) {
         if (worker.thread.joinable()) worker.thread.join();
     }
+    DrawPipeline::Queue0().Stop();
     StopWorkerSampler();
     CloseRegistrationPreparation();
+    Graphics::StopPipelineLibraryWorker();
     std::lock_guard gpuLock(GuestMemory::GpuMutex());
     device.Reset();
     replacedDevices.clear();
     Graphics::FlushPipelineJournal();
     Graphics::ShutdownGuestBufferWorkers();
+    Graphics::StopCompareHelpers();
+    ShaderRecompiler::ShaderDiskCache::Shutdown();
     PerformanceTrace::StopOutput();
-    FlushAsyncLog_nid_no_patch();
+    FlushAsyncLogAtEnd_nid_no_patch();
     StopProfileOutput_nid_no_patch();
     stopped = true;
 }
@@ -222,3 +230,11 @@ bool& Driver::onWorkerThread() {
 }
 
 }
+
+namespace AgcDriver {
+
+std::unique_lock<std::mutex> HoldAsyncLog() {
+    return AsyncLog::Get().Hold();
+}
+
+}

core/libs/prx/libSceAgcDriver/Execution/src/Driver/Shaders/ShaderRegistry.cpp

@@ -11,6 +11,7 @@
 #include "CompiledVariant.hpp"
 #include "Optimization/ResourceProgram.hpp"
 #include <algorithm>
+#include <array>
 #include <atomic>
 #include <chrono>
 #include <set>
@@ -933,7 +934,8 @@ class RegistrationPreparePool {
 public:
     static RegistrationPreparePool& Instance() {
         static auto* pool = [] {
-            auto* created = new RegistrationPreparePool(std::max(2u, std::thread::hardware_concurrency() / 2u));
+            std::lock_guard lifecycle(Lifecycle());
+            auto* created = new RegistrationPreparePool(Closing() ? 0u : std::max(2u, std::thread::hardware_concurrency() / 2u));
             PreparationSlot().store(created);
             return created;
         }();
@@ -947,7 +949,8 @@ class RegistrationPreparePool {
     static RegistrationPreparePool& Prebuild() {
         static auto* pool = [] {
             const char* text = std::getenv("APS5_PREBUILD_THREADS");
-            auto* created = new RegistrationPreparePool(text != nullptr ? std::max(1u, static_cast<unsigned>(std::strtoul(text, nullptr, 10))) : 1u);
+            std::lock_guard lifecycle(Lifecycle());
+            auto* created = new RegistrationPreparePool(Closing() ? 0u : text != nullptr ? std::max(1u, static_cast<unsigned>(std::strtoul(text, nullptr, 10))) : 1u);
             PrebuildSlot().store(created);
             return created;
         }();
@@ -958,7 +961,33 @@ class RegistrationPreparePool {
         return PrebuildSlot().load();
     }
 
+    static unsigned LiveWorkers() {
+        return liveWorkers.load(std::memory_order_acquire);
+    }
+
+    static void CloseAll() {
+        std::array<RegistrationPreparePool*, 2> pools{};
+        {
+            std::lock_guard lifecycle(Lifecycle());
+            Closing() = true;
+            pools = {Existing(), ExistingPrebuild()};
+        }
+        for (auto* pool : pools) {
+            if (pool != nullptr) pool->Close();
+        }
+    }
+
 private:
+    static std::mutex& Lifecycle() {
+        static std::mutex lifecycle;
+        return lifecycle;
+    }
+
+    static bool& Closing() {
+        static bool closing = false;
+        return closing;
+    }
+
     static std::atomic<RegistrationPreparePool*>& PreparationSlot() {
         static std::atomic<RegistrationPreparePool*> slot{nullptr};
         return slot;
@@ -1002,9 +1031,17 @@ class RegistrationPreparePool {
     }
 
     void Close() {
-        std::unique_lock lock(mutex);
-        closed = true;
-        idle.wait(lock, [&] { return tasks.empty() && running == 0; });
+        std::vector<pthread_t> joined;
+        {
+            std::unique_lock lock(mutex);
+            closed = true;
+            idle.wait(lock, [&] { return tasks.empty() && running == 0; });
+            joined.swap(workers);
+        }
+        ready.notify_all();
+        for (const auto worker : joined) {
+            if (!pthread_equal(worker, pthread_self())) pthread_join(worker, nullptr);
+        }
     }
 
 private:
@@ -1015,8 +1052,9 @@ class RegistrationPreparePool {
         for (unsigned index = 0; index < count; ++index) {
             pthread_t thread;
             if (pthread_create(&thread, &attributes, &RegistrationPreparePool::Main, this) == 0) {
-                pthread_detach(thread);
+                workers.push_back(thread);
                 ++threads;
+                liveWorkers.fetch_add(1, std::memory_order_acq_rel);
             }
         }
         pthread_attr_destroy(&attributes);
@@ -1029,7 +1067,11 @@ class RegistrationPreparePool {
             Task task;
             {
                 std::unique_lock lock(pool.mutex);
-                pool.ready.wait(lock, [&] { return !pool.tasks.empty(); });
+                pool.ready.wait(lock, [&] { return !pool.tasks.empty() || pool.closed; });
+                if (pool.tasks.empty()) {
+                    liveWorkers.fetch_sub(1, std::memory_order_acq_rel);
+                    return nullptr;
+                }
                 task = std::move(pool.tasks.front());
                 pool.tasks.pop_front();
                 ++pool.running;
@@ -1062,7 +1104,9 @@ class RegistrationPreparePool {
     std::condition_variable ready;
     std::condition_variable idle;
     std::deque<Task> tasks;
+    std::vector<pthread_t> workers;
     unsigned threads = 0;
+    inline static std::atomic<unsigned> liveWorkers{0};
     unsigned running = 0;
     std::uint64_t submitted = 0;
     std::set<std::uint64_t> unfinished;
@@ -1124,8 +1168,11 @@ void AwaitLibraryPrebuilds() {
 }
 
 void CloseRegistrationPreparation() {
-    RegistrationPreparePool::Instance().Close();
-    if (auto* prebuild = RegistrationPreparePool::ExistingPrebuild()) prebuild->Close();
+    RegistrationPreparePool::CloseAll();
+}
+
+unsigned RegistrationPreparationThreads() {
+    return RegistrationPreparePool::LiveWorkers();
 }
 
 std::vector<PreparedGraphicsStage> PrepareGraphicsStages(const DrawDecode& decoded, const ShaderRecompiler::SpirvTarget& target) {

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

@@ -16,6 +16,8 @@ struct CompareSpan {
 void CompareSpans(std::span<const CompareSpan> spans, std::span<std::uint8_t> equal);
 void CompareSpansFrom(std::span<const CompareSpan> spans, std::span<std::uint8_t> equal, std::size_t parallelBytes);
 unsigned CompareHelpers();
+void StopCompareHelpers();
+unsigned CompareHelpersRunning();
 
 }
 

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

@@ -48,6 +48,9 @@ void WaitForOptimizedPipelines(VkDevice device);
 
 void ClearPipelineLibraries(VkDevice device);
 
+void StopPipelineLibraryWorker();
+unsigned PipelineLibraryWorkers();
+
 struct PipelineLibraryCounters {
     std::array<std::uint64_t, 4> built{};
     std::uint64_t linked = 0;