perf(agc): link pipelines with link-time optimization on half the hardware threads

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

@@ -1,17 +1,23 @@
 #include "prx/libSceAgcDriver/Graphics/include/PipelineLibrary.hpp"
 #include "prx/libSceAgcDriver/Graphics/include/PipelineJournal.hpp"
 #include "prx/libSceAgcDriver/Execution/include/PerformanceTimer.hpp"
+#include <algorithm>
 #include <array>
 #include <atomic>
+#include <charconv>
 #include <condition_variable>
 #include <cstdlib>
+#include <cstring>
 #include <deque>
 #include <functional>
 #include <thread>
 #include <map>
 #include <mutex>
 #include <optional>
+#include <stdexcept>
+#include <string>
 #include <string_view>
+#include <vector>
 
 namespace AgcDriver::Graphics {
 
@@ -55,12 +61,27 @@ struct LibraryStore {
     std::deque<std::pair<VkDevice, std::function<void()>>> jobs;
     std::map<VkDevice, std::size_t> pending;
     std::map<VkDevice, std::size_t> activeBuilds;
-    std::thread worker;
+    std::vector<std::thread> workers;
+    std::size_t waitingWorkers = 0;
     bool stopping = false;
 };
 
 std::atomic<unsigned> runningWorkers{0};
 
+std::size_t WorkerLimit() {
+    static const std::size_t limit = [] {
+        const std::size_t hardware = std::max<std::size_t>(1, std::thread::hardware_concurrency());
+        const char* text = std::getenv("APS5_PIPELINE_LTO_THREADS");
+        if (text == nullptr) return std::max<std::size_t>(2, hardware / 2);
+        const char* end = text + std::strlen(text);
+        std::size_t value = 0;
+        const auto [last, error] = std::from_chars(text, end, value);
+        if (error != std::errc{} || last != end || value == 0 || value > hardware) throw std::runtime_error("APS5_PIPELINE_LTO_THREADS must be a whole number from 1 through " + std::to_string(hardware) + ", the hardware thread count");
+        return value;
+    }();
+    return limit;
+}
+
 LibraryStore& Libraries() {
     static auto* store = new LibraryStore();
     return *store;
@@ -69,7 +90,9 @@ LibraryStore& Libraries() {
 void runJobs(LibraryStore& store) {
     std::unique_lock lock(store.mutex);
     for (;;) {
+        ++store.waitingWorkers;
         store.work.wait(lock, [&] { return !store.jobs.empty() || store.stopping; });
+        --store.waitingWorkers;
         if (store.stopping) break;
         auto [device, job] = std::move(store.jobs.front());
         store.jobs.pop_front();
@@ -83,8 +106,8 @@ void runJobs(LibraryStore& store) {
 
 void enqueue(LibraryStore& store, VkDevice device, std::function<void()> job) {
     if (store.stopping) return;
-    if (!store.worker.joinable()) {
-        store.worker = std::thread(runJobs, std::ref(store));
+    if (store.waitingWorkers <= store.jobs.size() && store.workers.size() < WorkerLimit()) {
+        store.workers.emplace_back(runJobs, std::ref(store));
         runningWorkers.fetch_add(1, std::memory_order_acq_rel);
     }
     auto& pending = store.pending[device];
@@ -367,19 +390,21 @@ void ClearPipelineLibraries(VkDevice device) {
 
 void StopPipelineLibraryWorker() {
     auto& store = Libraries();
-    std::thread worker;
+    std::vector<std::thread> workers;
     {
         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);
+        workers = std::move(store.workers);
     }
     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();
+    for (auto& worker : workers) {
+        if (!worker.joinable()) continue;
+        if (worker.get_id() == std::this_thread::get_id()) worker.detach();
+        else worker.join();
+    }
 }
 
 unsigned PipelineLibraryWorkers() {

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

@@ -0,0 +1,121 @@
+#include "prx/libSceAgcDriver/Graphics/include/PipelineLibrary.hpp"
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <cstdint>
+#include <cstdio>
+#include <cstdlib>
+#include <functional>
+#include <mutex>
+#include <stdexcept>
+#include <string_view>
+#include <utility>
+
+using namespace AgcDriver::Graphics;
+using namespace std::chrono_literals;
+
+namespace {
+
+void Check(bool condition, const char* message) {
+    if (condition) return;
+    std::fprintf(stderr, "FAIL: %s\n", message);
+    std::fflush(stderr);
+    std::_Exit(1);
+}
+
+void Schedule(VkDevice device, std::function<void()> work) {
+    OptimizedPipeline pipeline;
+    pipeline.device = device;
+    pipeline.relink = std::move(work);
+    pipeline.deferred = true;
+    pipeline.Schedule();
+}
+
+}
+
+int main(int argc, char** argv) {
+    const auto device = reinterpret_cast<VkDevice>(std::uintptr_t{1});
+    if (argc == 2 && std::string_view(argv[1]) == "invalid") {
+        bool rejected = false;
+        try {
+            Schedule(device, [] {});
+        } catch (const std::runtime_error& error) {
+            rejected = std::string_view(error.what()).find("APS5_PIPELINE_LTO_THREADS") != std::string_view::npos;
+        }
+        Check(rejected, "an invalid worker count was accepted");
+        Check(PipelineLibraryWorkers() == 0, "a worker started for an invalid worker count");
+        std::printf("invalid worker count rejected before any worker started\n");
+        return 0;
+    }
+    std::mutex mutex;
+    std::condition_variable changed;
+    bool entered = false;
+    bool released = false;
+    std::atomic<unsigned> completed{0};
+    if (argc == 2 && std::string_view(argv[1]) == "burst") {
+        Schedule(device, [&] { ++completed; });
+        WaitForOptimizedPipelines(device);
+        Check(completed == 1 && PipelineLibraryWorkers() == 1, "the first job did not run on one worker");
+        Schedule(device, [&] {
+            std::unique_lock lock(mutex);
+            changed.wait(lock, [&] { return released; });
+            ++completed;
+        });
+        Schedule(device, [&] {
+            std::lock_guard lock(mutex);
+            ++completed;
+            changed.notify_all();
+        });
+        {
+            std::unique_lock lock(mutex);
+            Check(changed.wait_for(lock, 3s, [&] { return completed == 2; }), "a burst queued behind the idle worker it woke with a second worker allowed");
+            released = true;
+        }
+        changed.notify_all();
+        WaitForOptimizedPipelines(device);
+        Check(completed == 3 && PipelineLibraryWorkers() == 2, "the burst did not run once each on two workers");
+        StopPipelineLibraryWorker();
+        ClearPipelineLibraries(device);
+        Check(PipelineLibraryWorkers() == 0, "the pipeline workers did not stop");
+        std::printf("two jobs queued at once behind one idle worker ran on two workers\n");
+        return 0;
+    }
+    Check(argc == 1, "expected no argument, invalid or burst");
+    Schedule(device, [&] {
+        std::unique_lock lock(mutex);
+        entered = true;
+        changed.notify_all();
+        changed.wait(lock, [&] { return released; });
+        ++completed;
+    });
+    {
+        std::unique_lock lock(mutex);
+        Check(changed.wait_for(lock, 3s, [&] { return entered; }), "the first worker did not start");
+    }
+
+    Schedule(device, [&] {
+        ++completed;
+        changed.notify_all();
+    });
+    {
+        std::unique_lock lock(mutex);
+        Check(changed.wait_for(lock, 3s, [&] { return completed == 1; }), "a job waited behind a held job with a second worker allowed");
+    }
+    Check(PipelineLibraryWorkers() == 2, "the queue did not start a second worker");
+
+    Schedule(device, [&] { ++completed; });
+    {
+        std::lock_guard lock(mutex);
+        released = true;
+    }
+    changed.notify_all();
+    WaitForOptimizedPipelines(device);
+    Check(completed == 3, "a job was lost or ran twice");
+    Check(PipelineLibraryWorkers() == 2, "the queue started more workers than allowed");
+
+    StopPipelineLibraryWorker();
+    ClearPipelineLibraries(device);
+    Check(PipelineLibraryWorkers() == 0, "the pipeline workers did not stop");
+    std::printf("two workers: a job finished while another was held, three jobs ran once each, both workers stopped\n");
+    return 0;
+}

core/libs/prx/libSceAgcDriver/tests/cmake/agc_pipeline_library_workers_tests.cmake

@@ -0,0 +1,27 @@
+add_test_executable(agc_pipeline_library_workers_tests EXCLUDE_FROM_ALL tests/PipelineLibraryWorkers.cpp Graphics/src/PipelineLibrary.cpp)
+target_include_directories(agc_pipeline_library_workers_tests PRIVATE ${LIBS_INCLUDE_DIR} ${CMAKE_SOURCE_DIR}/core/shader/recompiler ${ANYPS5_VULKAN_HEADERS_INCLUDE_DIR})
+target_link_libraries(agc_pipeline_library_workers_tests PRIVATE libSceAgcDriver libc)
+set_target_properties(agc_pipeline_library_workers_tests PROPERTIES CXX_EXTENSIONS OFF)
+configure_windows_unwind(agc_pipeline_library_workers_tests)
+
+if(BUILD_TESTING)
+    set_target_properties(agc_pipeline_library_workers_tests PROPERTIES EXCLUDE_FROM_ALL OFF)
+    add_test(NAME agc_pipeline_library_workers COMMAND agc_pipeline_library_workers_tests)
+    set_tests_properties(agc_pipeline_library_workers PROPERTIES TIMEOUT 15 ENVIRONMENT "APS5_PIPELINE_LTO_THREADS=2")
+    add_test(NAME agc_pipeline_library_workers_burst COMMAND agc_pipeline_library_workers_tests burst)
+    set_tests_properties(agc_pipeline_library_workers_burst PROPERTIES TIMEOUT 15 ENVIRONMENT "APS5_PIPELINE_LTO_THREADS=2")
+    set(workerTests agc_pipeline_library_workers agc_pipeline_library_workers_burst)
+    foreach(case negative=-1 text=abc zero=0 suffix=8x above_hardware=100000)
+        string(REPLACE "=" ";" parts "${case}")
+        list(GET parts 0 name)
+        list(GET parts 1 value)
+        add_test(NAME agc_pipeline_library_workers_invalid_${name} COMMAND agc_pipeline_library_workers_tests invalid)
+        set_tests_properties(agc_pipeline_library_workers_invalid_${name} PROPERTIES TIMEOUT 15 ENVIRONMENT "APS5_PIPELINE_LTO_THREADS=${value}")
+        list(APPEND workerTests agc_pipeline_library_workers_invalid_${name})
+    endforeach()
+    if(WIN32)
+        foreach(test IN LISTS workerTests)
+            set_tests_properties(${test} PROPERTIES ENVIRONMENT_MODIFICATION "PATH=path_list_prepend:$<TARGET_FILE_DIR:libc>")
+        endforeach()
+    endif()
+endif()

core/libs/prx/libSceAgcDriver/tests/cmake/agc_pipeline_queue_failure_tests.cmake

@@ -9,7 +9,7 @@ if(BUILD_TESTING)
     set_target_properties(agc_pipeline_queue_failure_tests PROPERTIES EXCLUDE_FROM_ALL OFF)
     foreach(mode wait clear stop)
         add_test(NAME agc_pipeline_queue_failure_${mode} COMMAND agc_pipeline_queue_failure_tests ${mode})
-        set_tests_properties(agc_pipeline_queue_failure_${mode} PROPERTIES TIMEOUT 15)
+        set_tests_properties(agc_pipeline_queue_failure_${mode} PROPERTIES TIMEOUT 15 ENVIRONMENT "APS5_PIPELINE_LTO_THREADS=1")
         if(WIN32)
             set_tests_properties(agc_pipeline_queue_failure_${mode} PROPERTIES ENVIRONMENT_MODIFICATION
                 "PATH=path_list_prepend:$<TARGET_FILE_DIR:libc>")

docs/dev/ENVIRONMENT.md

@@ -403,6 +403,7 @@ The code reads environment names through `getenv`, `IsEnvironmentSet`, `kibSetti
 | `APS5_PIPELINE_JOURNAL_ENTRIES` | integer | `core/libs/prx/libSceAgcDriver/Graphics/src/PipelineJournal.cpp` | 4096 | number of graphics pipelines a replay builds at most, the last recorded; a log line names how many of each kind were left out |
 | `APS5_PIPELINE_JOURNAL_MAX_MIB` | integer | `core/libs/prx/libSceAgcDriver/Graphics/src/PipelineJournal.cpp` | 1024 | size cap in MiB on the pipeline journal; the file is rewritten down to three quarters of the cap when it is reached, keeping the newest pipelines and the modules they name and leaving room for later records |
 | `APS5_PIPELINE_JOURNAL_THREADS` | integer | `core/libs/prx/libSceAgcDriver/Graphics/src/PipelineJournal.cpp` |  | threads that replay the journal's pipelines, 0 disables the replay; unset uses half the hardware threads, at least one |
+| `APS5_PIPELINE_LTO_THREADS` | integer | `core/libs/prx/libSceAgcDriver/Graphics/src/PipelineLibrary.cpp` |  | threads that link graphics pipelines with link-time optimization in the background, a whole number from 1 through the hardware thread count; any other value fails at the first background link, 0 included (it does not disable the link, unlike `APS5_PIPELINE_JOURNAL_THREADS`); unset uses half the hardware threads, at least two |
 | `APS5_PIPELINE_STATS` | string | `core/libs/prx/libSceAgcDriver/Execution/src/VulkanDevice.cpp` |  | captures and prints driver statistics for each new pipeline, requiring `VK_KHR_pipeline_executable_properties`; `1` enables |
 | `APS5_PIPELINE_STORE_ENTRIES` | integer | `core/libs/prx/libSceAgcDriver/Graphics/src/Pipeline.cpp` | adaptive | fixes the LRU bound on the entries of the pipeline store, which otherwise starts at 1024 and doubles up to 4096 for a journal replay that does not fit and while evicted pipelines keep coming back; an entry is evicted only once no recorded draw holds it |
 | `APS5_PM4_RING` | integer | `core/libs/prx/libSceAgcDriver/Execution/src/Driver/Queues/SubmissionRing.cpp` | 16 | number of submissions the PM4 ring keeps for dumps, from 0 to 4096; 0 disables the ring |