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

core/libs/prx/libSceAgcDriver/CMakeLists.txt

@@ -1769,6 +1769,12 @@ 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)
+    set_tests_properties(agc_driver_shutdown_threads_joined agc_driver_shutdown_threads_compare_helpers 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 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/Driver/Shaders/ShaderRegistry.hpp

@@ -118,6 +118,7 @@ void PrioritizeRegistrationPreparation(PreparedShaders& prepared);
 void AwaitRegistrationPreparation();
 void SubmitLibraryPrebuild(std::function<void()> task);
 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/Lifecycle.cpp

@@ -6,6 +6,8 @@
 #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 <atomic>
 #include <cstdint>
@@ -118,6 +120,8 @@ void Driver::stop() {
     device.Reset();
     replacedDevices.clear();
     Graphics::ShutdownGuestBufferWorkers();
+    Graphics::StopPipelineLibraryWorker();
+    Graphics::StopCompareHelpers();
     PerformanceTrace::StopOutput();
     StopProfileOutput_nid_no_patch();
     stopped = true;

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;
@@ -1120,8 +1164,11 @@ void SubmitLibraryPrebuild(std::function<void()> task) {
 }
 
 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

@@ -41,6 +41,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;

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

@@ -5,21 +5,29 @@
 #include <cstring>
 #include <mutex>
 #include <thread>
+#include <vector>
 
 namespace AgcDriver::Graphics {
 namespace {
 
+std::atomic<unsigned> runningHelpers{0};
+
 class ComparePool {
 public:
     explicit ComparePool(unsigned count) {
-        for (unsigned i = 0; i < count; ++i) std::thread([this] { work(); }).detach();
+        stopping = count == 0;
+        for (unsigned i = 0; i < count; ++i) {
+            workers.emplace_back([this] { work(); });
+            runningHelpers.fetch_add(1, std::memory_order_acq_rel);
+        }
     }
 
     bool Run(std::span<const CompareSpan> spans, std::span<std::uint8_t> equal) {
         std::unique_lock caller(callers, std::try_to_lock);
         if (!caller.owns_lock()) return false;
         {
             std::unique_lock lock(mutex);
+            if (stopping) return false;
             finished.wait(lock, [&] { return working == 0; });
             jobSpans = spans;
             jobEqual = equal;
@@ -36,6 +44,21 @@ class ComparePool {
         return true;
     }
 
+    void Stop() {
+        std::vector<std::thread> joined;
+        {
+            std::lock_guard caller(callers);
+            std::lock_guard lock(mutex);
+            stopping = true;
+            joined.swap(workers);
+        }
+        wake.notify_all();
+        for (auto& worker : joined) {
+            if (worker.get_id() == std::this_thread::get_id()) worker.detach();
+            else worker.join();
+        }
+    }
+
 private:
     void drain(std::span<const CompareSpan> spans, std::span<std::uint8_t> equal) {
         std::size_t compared = 0;
@@ -51,7 +74,8 @@ class ComparePool {
         std::uint64_t seen = 0;
         std::unique_lock lock(mutex);
         for (;;) {
-            wake.wait(lock, [&] { return generation != seen; });
+            wake.wait(lock, [&] { return generation != seen || stopping; });
+            if (stopping) break;
             seen = generation;
             const auto spans = jobSpans;
             const auto equal = jobEqual;
@@ -62,6 +86,7 @@ class ComparePool {
             --working;
             if (working == 0) finished.notify_all();
         }
+        runningHelpers.fetch_sub(1, std::memory_order_acq_rel);
     }
 
     std::mutex callers;
@@ -74,10 +99,32 @@ class ComparePool {
     std::span<std::uint8_t> jobEqual;
     std::atomic<std::size_t> next{0};
     std::atomic<std::size_t> done{0};
+    std::vector<std::thread> workers;
+    bool stopping = false;
 };
 
+std::mutex& poolLifecycle() {
+    static std::mutex lifecycle;
+    return lifecycle;
+}
+
+bool& poolStopped() {
+    static bool stopped = false;
+    return stopped;
+}
+
+std::atomic<ComparePool*>& createdPool() {
+    static std::atomic<ComparePool*> created{nullptr};
+    return created;
+}
+
 ComparePool* sharedPool() {
-    static ComparePool* const pool = new ComparePool(CompareHelpers());
+    static ComparePool* const pool = [] {
+        std::lock_guard lifecycle(poolLifecycle());
+        auto* created = new ComparePool(poolStopped() ? 0u : CompareHelpers());
+        createdPool().store(created, std::memory_order_release);
+        return created;
+    }();
     return pool;
 }
 
@@ -97,6 +144,20 @@ void CompareSpansFrom(std::span<const CompareSpan> spans, std::span<std::uint8_t
     for (std::size_t index = 0; index < spans.size(); ++index) equal[index] = std::memcmp(spans[index].first, spans[index].second, spans[index].bytes) == 0 ? 1 : 0;
 }
 
+void StopCompareHelpers() {
+    ComparePool* pool = nullptr;
+    {
+        std::lock_guard lifecycle(poolLifecycle());
+        poolStopped() = true;
+        pool = createdPool().load(std::memory_order_acquire);
+    }
+    if (pool != nullptr) pool->Stop();
+}
+
+unsigned CompareHelpersRunning() {
+    return runningHelpers.load(std::memory_order_acquire);
+}
+
 void CompareSpans(std::span<const CompareSpan> spans, std::span<std::uint8_t> equal) {
     CompareSpansFrom(spans, equal, std::size_t{1} << 20u);
 }

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

@@ -1,6 +1,7 @@
 #include "prx/libSceAgcDriver/Graphics/include/PipelineLibrary.hpp"
 #include "prx/libSceAgcDriver/Execution/include/PerformanceTimer.hpp"
 #include <array>
+#include <atomic>
 #include <condition_variable>
 #include <cstdlib>
 #include <deque>
@@ -52,9 +53,12 @@ struct LibraryStore {
     std::condition_variable work;
     std::deque<std::pair<VkDevice, std::function<void()>>> jobs;
     std::map<VkDevice, std::size_t> pending;
-    bool started = false;
+    std::thread worker;
+    bool stopping = false;
 };
 
+std::atomic<unsigned> runningWorkers{0};
+
 LibraryStore& Libraries() {
     static auto* store = new LibraryStore();
     return *store;
@@ -63,20 +67,23 @@ LibraryStore& Libraries() {
 void runJobs(LibraryStore& store) {
     std::unique_lock lock(store.mutex);
     for (;;) {
-        store.work.wait(lock, [&] { return !store.jobs.empty(); });
+        store.work.wait(lock, [&] { return !store.jobs.empty() || store.stopping; });
+        if (store.stopping) break;
         auto [device, job] = std::move(store.jobs.front());
         store.jobs.pop_front();
         lock.unlock();
         job();
         lock.lock();
         if (--store.pending[device] == 0) store.idle.notify_all();
     }
+    runningWorkers.fetch_sub(1, std::memory_order_acq_rel);
 }
 
 void enqueue(LibraryStore& store, VkDevice device, std::function<void()> job) {
-    if (!store.started) {
-        std::thread(runJobs, std::ref(store)).detach();
-        store.started = true;
+    if (store.stopping) return;
+    if (!store.worker.joinable()) {
+        store.worker = std::thread(runJobs, std::ref(store));
+        runningWorkers.fetch_add(1, std::memory_order_acq_rel);
     }
     ++store.pending[device];
     store.jobs.emplace_back(device, std::move(job));
@@ -303,6 +310,27 @@ void ClearPipelineLibraries(VkDevice device) {
     store.devices.erase(found);
 }
 
+void StopPipelineLibraryWorker() {
+    auto& store = Libraries();
+    std::thread worker;
+    {
+        std::lock_guard lock(store.mutex);
+        store.stopping = true;
+        for (const auto& [device, job] : store.jobs) --store.pending[device];
+        store.jobs.clear();
+        worker = std::move(store.worker);
+    }
+    store.idle.notify_all();
+    store.work.notify_all();
+    if (!worker.joinable()) return;
+    if (worker.get_id() == std::this_thread::get_id()) worker.detach();
+    else worker.join();
+}
+
+unsigned PipelineLibraryWorkers() {
+    return runningWorkers.load(std::memory_order_acquire);
+}
+
 PipelineLibraryCounters PipelineLibraryCountersOf(VkDevice device) {
     auto& store = Libraries();
     std::lock_guard lock(store.mutex);