feat(libSceJobManager): implement compute sequences and compute dispatches

core/libs/prx/libSceAgcDriver/CMakeLists.txt

@@ -1564,6 +1564,8 @@ target_include_directories(agc_driver_swap_alt_blend_tests PRIVATE $<TARGET_PROP
 
 target_include_directories(agc_driver_packed_float_color_target_tests PRIVATE $<TARGET_PROPERTY:shader_recompiler_agc_driver,INTERFACE_INCLUDE_DIRECTORIES>)
 
+target_link_libraries(agc_driver_job_manager_compute_tests PRIVATE libSceJobManager libSceAgc libkernel)
+
 target_include_directories(agc_driver_packed_unorm_color_target_tests PRIVATE $<TARGET_PROPERTY:shader_recompiler_agc_driver,INTERFACE_INCLUDE_DIRECTORIES>)
 
 target_include_directories(agc_driver_color1555_target_tests PRIVATE $<TARGET_PROPERTY:shader_recompiler_agc_driver,INTERFACE_INCLUDE_DIRECTORIES>)

core/libs/prx/libSceAgcDriver/tests/execution/JobManagerCompute.cpp

@@ -0,0 +1,242 @@
+#include "VulkanTestDevice.hpp"
+#include "SceTypes.hpp"
+#include <array>
+#include <atomic>
+#include <chrono>
+#include <cstdint>
+#include <cstdlib>
+#include <cstring>
+#include <iostream>
+#include <new>
+#include <stdexcept>
+#include <string>
+#include <thread>
+
+extern "C" void AgcDriverShutdown_nid_postfix();
+
+struct MemorySizeQueryParams {
+    std::uint32_t version;
+    std::uint32_t workers;
+    std::uint32_t sequences;
+    std::uint32_t invocations;
+    std::uint32_t dispatches;
+    std::uint32_t syncPoints;
+    std::uint32_t waits;
+    std::uint32_t computeSequences;
+    std::uint64_t computeQueueMask;
+};
+
+struct ConfigParams {
+    std::uint32_t version;
+    std::uint32_t threadFlags;
+    std::uint32_t stackSize;
+    std::uint32_t priority;
+    const char* threadName;
+    std::uint32_t spinMicroseconds;
+    std::uint32_t computeAffinity;
+    std::uint32_t computePriority;
+    std::uint32_t reserved;
+};
+
+using JobEntry = void (APS5_VABI*)(void*, std::uint32_t);
+
+struct Sequence {
+    APS5_VABI virtual int Dispatch(JobEntry, void*, std::uint32_t, const char*) = 0;
+    APS5_VABI virtual int SyncPoint() = 0;
+    APS5_VABI virtual int Wait() = 0;
+    APS5_VABI virtual void CompleteDestructor() = 0;
+    APS5_VABI virtual void DeletingDestructor() = 0;
+    APS5_VABI virtual int DispatchCompute(const void*, void*, std::int32_t, std::int32_t, std::int32_t, const char*) = 0;
+    APS5_VABI virtual int DispatchComputeIndirect(const void*, void*, const void*, const char*) = 0;
+};
+
+struct SequenceFactory {
+    APS5_VABI virtual int Create(Sequence**, std::uint32_t, std::uint32_t, std::uint32_t, std::uint32_t, const char*) = 0;
+    APS5_VABI virtual int Destroy(Sequence*) = 0;
+    APS5_VABI virtual void CompleteDestructor() = 0;
+    APS5_VABI virtual void DeletingDestructor() = 0;
+    APS5_VABI virtual int CreateCompute(Sequence**, std::uint32_t, std::uint32_t, std::uint32_t, std::uint32_t, const char*) = 0;
+    APS5_VABI virtual int DestroyCompute(Sequence*) = 0;
+};
+
+extern "C" {
+void APS5_VABI _ZN3sce3Job10JobManagerC1Ev(void*);
+std::uint64_t APS5_VABI _ZN3sce3Job10JobManager27calculateRequiredMemorySizeEPKNS1_21MemorySizeQueryParamsE(const MemorySizeQueryParams*);
+int APS5_VABI _ZN3sce3Job10JobManager10initializeEPvmjjjmPKNS1_12ConfigParamsE(void*, void*, std::uint64_t, std::uint32_t, std::uint32_t, std::uint32_t, std::uint64_t, const ConfigParams*);
+int APS5_VABI _ZN3sce3Job10JobManager8shutdownEv(void*);
+SequenceFactory* APS5_VABI _ZN3sce3Job10JobManager34getUnifiedSequenceFactoryInterfaceENS1_8PriorityE(void*, std::uint32_t);
+}
+
+#define Construct _ZN3sce3Job10JobManagerC1Ev
+#define RequiredSize _ZN3sce3Job10JobManager27calculateRequiredMemorySizeEPKNS1_21MemorySizeQueryParamsE
+#define Initialize _ZN3sce3Job10JobManager10initializeEPvmjjjmPKNS1_12ConfigParamsE
+#define Shutdown _ZN3sce3Job10JobManager8shutdownEv
+#define Factories _ZN3sce3Job10JobManager34getUnifiedSequenceFactoryInterfaceENS1_8PriorityE
+
+namespace {
+
+constexpr int InvalidArgument = static_cast<int>(0x81581000u);
+constexpr int OutOfRange = static_cast<int>(0x81580000u);
+constexpr int NullArgument = static_cast<int>(0x81580001u);
+constexpr int NoNodes = static_cast<int>(0x81580002u);
+constexpr int LimitExceeded = static_cast<int>(0x81580003u);
+constexpr int NoneFree = static_cast<int>(0x81580006u);
+constexpr int Invalid = static_cast<int>(0x81580007u);
+
+void Require(bool condition, const std::string& message) {
+    if (!condition) throw std::runtime_error(message);
+}
+
+alignas(256) std::array<std::uint32_t, 12> ReaderCode{0xf4040100u, 0xfa000000u, 0xf4080201u, 0xfa000000u, 0xbf8cc07fu, 0xf4000182u, 0xfa000000u, 0xbf8cc07fu, 0x7e000206u, 0xe0700000u, 0x80020000u, 0xbf810000u};
+alignas(256) std::array<std::uint32_t, 4> Value{};
+alignas(256) std::array<std::uint32_t, 2> Table{};
+alignas(256) volatile std::uint32_t Output[4]{};
+alignas(256) std::array<std::uint32_t, 4> OutputDescriptor{};
+alignas(32) std::array<std::uint32_t, 8> IndirectArguments{1, 1, 1, 0, 0, 0, 0, 0};
+alignas(16) std::array<std::uint32_t, 8> Specials{};
+alignas(16) std::array<std::uint32_t, 8> SpecialsWithoutDispatch{};
+std::array<ShaderRegister, 9> Registers{};
+alignas(16) std::array<unsigned char, 0x80> Shader{};
+alignas(16) std::array<unsigned char, 0x80> ShaderWithoutDispatch{};
+
+std::uint32_t Low(const volatile void* pointer) {
+    return static_cast<std::uint32_t>(reinterpret_cast<std::uintptr_t>(pointer));
+}
+
+std::uint32_t High(const volatile void* pointer) {
+    return static_cast<std::uint32_t>(static_cast<std::uint64_t>(reinterpret_cast<std::uintptr_t>(pointer)) >> 32u);
+}
+
+void BuildShader() {
+    Table = {Low(Value.data()), High(Value.data())};
+    OutputDescriptor = {Low(Output), High(Output) & 0xffffu, sizeof(Output), 0x31016facu};
+    const auto code = static_cast<std::uint64_t>(reinterpret_cast<std::uintptr_t>(ReaderCode.data()));
+    Registers = {{{0x20c, static_cast<std::uint32_t>(code >> 8u)}, {0x20d, static_cast<std::uint32_t>(code >> 40u) & 0xffu}, {0x207, 1}, {0x208, 1}, {0x209, 1}, {0x212, 0}, {0x213, 4u << 1u},
+                  {0x242, Low(OutputDescriptor.data())}, {0x243, High(OutputDescriptor.data())}}};
+    Specials[4] = 1;
+    const ShaderRegister* registers = Registers.data();
+    for (auto* shader : {&Shader, &ShaderWithoutDispatch}) {
+        const std::uint32_t* specials = shader == &Shader ? Specials.data() : SpecialsWithoutDispatch.data();
+        std::memcpy(shader->data() + 0x20, &registers, sizeof(registers));
+        std::memcpy(shader->data() + 0x28, &specials, sizeof(specials));
+        (*shader)[0x5c] = static_cast<unsigned char>(Registers.size());
+    }
+}
+
+std::uint64_t SizeFor(std::uint32_t computeSequences, std::uint64_t mask, bool limits) {
+    const MemorySizeQueryParams query{1, 2, 1, limits ? 100u : 0u, limits ? 8u : 0u, limits ? 4u : 0u, limits ? 2u : 0u, computeSequences, mask};
+    return RequiredSize(&query);
+}
+
+std::atomic<bool> Released{false};
+std::atomic<std::uint32_t> SeenByJob{0};
+
+void APS5_VABI BlockingJob(void*, std::uint32_t) {
+    while (!Released) std::this_thread::yield();
+}
+
+void APS5_VABI ReadOutputJob(void*, std::uint32_t) {
+    SeenByJob = Output[0];
+}
+
+void APS5_VABI Nothing(void*, std::uint32_t) {
+}
+
+void SizeCase() {
+    Require(SizeFor(0, 0, false) == 0x10d0 && SizeFor(0, 0, true) == 0x14c0, "size without compute sequences");
+    Require(SizeFor(1, 1, false) == 0x24fd8 && SizeFor(1, 1, true) == 0x253c8, "size with one compute sequence");
+    Require(SizeFor(2, 3, false) == 0x48e20 && SizeFor(2, 3, true) == 0x49210, "size with two compute sequences");
+    Require(SizeFor(3, 7, false) == 0x6cb68, "size with three compute sequences");
+    Require(SizeFor(1, 3, true) == 0x491c8, "size with two queues for one compute sequence");
+    Require(SizeFor(2, 1, true) == 0, "size with fewer queues than compute sequences");
+}
+
+void RunCase() {
+    const std::uint64_t size = SizeFor(1, 1, true);
+    void* memory = ::operator new((size + 255) & ~std::uint64_t{255}, std::align_val_t{256});
+    Require(memory != nullptr, "allocation failed");
+    alignas(64) static unsigned char manager[0x100];
+    Construct(manager);
+    const ConfigParams badHandler{1, 0x00021fff, 0x00010000, 0x2bc, nullptr, 0, 0x1fff, 0x50, 0};
+    Require(Initialize(manager, memory, size, 2, 1, 1, 1, &badHandler) == InvalidArgument, "a handler priority below 0x100 was accepted");
+    Require(Initialize(manager, memory, size, 2, 1, 1, 1, nullptr) == 0, "initialize with one compute sequence failed");
+    SequenceFactory* factory = Factories(manager, 0);
+    Sequence* sequence = nullptr;
+    Sequence* other = nullptr;
+    Require(factory->CreateCompute(&sequence, 100, 4, 4, 2, "compute") == 0, "the compute sequence was not created");
+    Require(factory->CreateCompute(&other, 1, 1, 1, 1, "second") == NoneFree, "a second compute sequence was created");
+    Require(factory->Create(&other, 1, 1, 1, 1, "cpu") == 0, "the CPU sequence was not created");
+    Require(factory->DestroyCompute(other) == Invalid, "a CPU sequence was destroyed as a compute sequence");
+    Require(factory->Destroy(sequence) == Invalid, "a compute sequence was destroyed as a CPU sequence");
+    Require(factory->Destroy(other) == 0, "the CPU sequence was not destroyed");
+
+    Require(sequence->DispatchCompute(nullptr, Table.data(), 1, 1, 1, nullptr) == NullArgument, "a null shader was accepted");
+    Require(sequence->DispatchCompute(Shader.data(), Table.data(), 0, 1, 1, nullptr) == OutOfRange, "a zero group count was accepted");
+    Require(sequence->DispatchCompute(Shader.data(), Table.data(), 1, 1, -1, nullptr) == OutOfRange, "a negative group count was accepted");
+    Require(sequence->DispatchComputeIndirect(Shader.data(), Table.data(), nullptr, nullptr) == NullArgument, "null indirect arguments were accepted");
+    Require(sequence->DispatchComputeIndirect(Shader.data(), Table.data(), reinterpret_cast<const char*>(IndirectArguments.data()) + 4, nullptr) == NullArgument, "misaligned indirect arguments were accepted");
+
+    Value[0] = 0x11110001u;
+    Require(sequence->DispatchCompute(Shader.data(), Table.data(), 1, 1, 1, "direct") == 0, "the compute dispatch failed");
+    Require(sequence->Wait() == 0, "the wait after a compute dispatch failed");
+    Require(Output[0] == 0x11110001u, "the compute dispatch had not run when the wait returned: " + std::to_string(Output[0]));
+
+    Value[0] = 0x22220002u;
+    Require(sequence->DispatchComputeIndirect(Shader.data(), Table.data(), IndirectArguments.data(), nullptr) == 0, "the indirect compute dispatch failed");
+    Require(sequence->Wait() == 0, "the wait after an indirect compute dispatch failed");
+    Require(Output[0] == 0x22220002u, "the indirect compute dispatch had not run when the wait returned: " + std::to_string(Output[0]));
+
+    Value[0] = 0x33330003u;
+    Require(sequence->DispatchCompute(ShaderWithoutDispatch.data(), Table.data(), 1, 1, 1, nullptr) == 0, "the dispatch of a shader without the dispatch bit failed");
+    Require(sequence->Wait() == 0, "the wait after a shader without the dispatch bit failed");
+    Require(Output[0] == 0x22220002u, "a shader without the dispatch bit ran");
+
+    Require(sequence->Dispatch(BlockingJob, nullptr, 1, "blocking") == 0, "the blocking job was not dispatched");
+    Require(sequence->SyncPoint() == 0, "the sync point failed");
+    Require(sequence->DispatchCompute(Shader.data(), Table.data(), 1, 1, 1, nullptr) == 0, "the compute dispatch behind a sync point failed");
+    std::this_thread::sleep_for(std::chrono::milliseconds(200));
+    Require(Output[0] == 0x22220002u, "a compute dispatch ran before the CPU job ahead of its sync point finished");
+    Released = true;
+    Require(sequence->SyncPoint() == 0, "the second sync point failed");
+    Require(sequence->Dispatch(ReadOutputJob, nullptr, 1, "reader") == 0, "the reading job was not dispatched");
+    Require(sequence->Wait() == 0, "the wait after the ordered dispatches failed");
+    Require(Output[0] == 0x33330003u, "the compute dispatch behind a sync point did not run");
+    Require(SeenByJob == 0x33330003u, "a CPU job behind a sync point ran before the compute dispatch ahead of it finished");
+
+    for (int index = 0; index < 4; ++index) Require(sequence->DispatchCompute(Shader.data(), Table.data(), 1, 1, 1, nullptr) == 0, "a compute dispatch within the limit failed");
+    Require(sequence->DispatchCompute(Shader.data(), Table.data(), 1, 1, 1, nullptr) == LimitExceeded, "a compute dispatch past the dispatch limit was accepted");
+    Require(sequence->DispatchCompute(Shader.data(), Table.data(), 1, 1, 1, nullptr) == Invalid, "a latched compute sequence accepted a dispatch");
+    Require(factory->DestroyCompute(sequence) == 0, "the compute sequence was not destroyed");
+
+    Require(factory->Create(&other, 0x7fff, 0x3ff, 0xff, 1, "nodes") == 0, "the node counting sequence was not created");
+    int accepted = 0;
+    int result = 0;
+    while (accepted < 0x3ff && (result = other->Dispatch(Nothing, nullptr, 1, nullptr)) == 0) ++accepted;
+    Require(result == NoNodes && accepted == 43, "the node pool of a compute configuration holds " + std::to_string(accepted) + " dispatches");
+    Require(Shutdown(manager) == 0, "shutdown failed");
+    ::operator delete(memory, std::align_val_t{256});
+}
+
+}
+
+int main() {
+    try {
+        auto device = OpenVulkanTestDevice();
+        if (!device) return VulkanTestSkipped;
+        device.reset();
+        BuildShader();
+        SizeCase();
+        RunCase();
+        AgcDriverShutdown_nid_postfix();
+        std::cout << "job manager compute test passed\n";
+        return 0;
+    } catch (const std::exception& error) {
+        std::cerr << error.what() << '\n';
+        try {
+            AgcDriverShutdown_nid_postfix();
+        } catch (const std::exception& shutdownError) {
+            std::cerr << shutdownError.what() << '\n';
+        }
+        return 1;
+    }
+}

core/libs/prx/libSceJobManager/CMakeLists.txt

@@ -1,11 +1,12 @@
 add_library(libSceJobManager SHARED EXCLUDE_FROM_ALL Export.cpp)
 target_include_directories(libSceJobManager PRIVATE ${LIBS_INCLUDE_DIR})
+target_link_libraries(libSceJobManager PRIVATE libSceAgc libSceAgcDriver)
 configure_windows_unwind(libSceJobManager)
 
 if(BUILD_TESTING)
     add_test_executable(guest_job_manager_tests ${LIBS_INCLUDE_DIR}/tests/GuestJobManager.cpp)
     target_include_directories(guest_job_manager_tests PRIVATE ${LIBS_INCLUDE_DIR})
-    target_link_libraries(guest_job_manager_tests PRIVATE libSceJobManager libc)
+    target_link_libraries(guest_job_manager_tests PRIVATE libSceJobManager libSceAgc libSceAgcDriver libkernel libc)
     configure_windows_unwind(guest_job_manager_tests)
     foreach(testCase IN ITEMS memory validation dispatch ordering nested limits priority shutdown)
         add_test(NAME guest_job_manager_${testCase} COMMAND guest_job_manager_tests ${testCase})

core/libs/prx/libSceJobManager/Export.cpp

@@ -22,6 +22,24 @@ int APS5_VABI scePthreadAttrSetstacksize(PthreadAttr* attr, std::size_t stackSiz
 int APS5_VABI scePthreadAttrSetaffinity(PthreadAttr* attr, KernelCpumask mask);
 int APS5_VABI scePthreadCreate(Pthread* thread, const PthreadAttr* attr, PthreadEntry entry, void* arg, const char* name);
 int APS5_VABI scePthreadJoin(Pthread thread, void** result);
+int APS5_VABI sceKernelCreateEqueue(KernelEqueue* eq, const char* name);
+int APS5_VABI sceKernelDeleteEqueue(KernelEqueue eq);
+int APS5_VABI sceKernelWaitEqueue(KernelEqueue eq, KernelEvent* ev, int num, int* out, const KernelUseconds* timo);
+int APS5_VABI sceKernelAddUserEventEdge(KernelEqueue eq, int id);
+int APS5_VABI sceKernelTriggerUserEvent(KernelEqueue eq, int id, void* udata);
+int APS5_VABI sceKernelGetEventFilter(const KernelEvent* ev);
+std::uintptr_t APS5_VABI sceKernelGetEventId(const KernelEvent* ev);
+int APS5_VABI sceAgcDriverAddEqEvent(KernelEqueue eq, int id, void* udata);
+int APS5_VABI sceAgcDriverDeleteEqEvent(KernelEqueue eq, int id);
+int APS5_VABI sceAgcDriverSubmitAcb(std::uint32_t queue, const Packet* packet);
+std::uint32_t* APS5_VABI sceAgcAcbAcquireMem(CommandBuffer* buf, std::uint32_t gcrControl, const volatile void* base, std::uint64_t sizeBytes, std::uint32_t pollCycles);
+std::uint32_t* APS5_VABI sceAgcCbSetShRegistersDirect(CommandBuffer* buf, const volatile ShaderRegister* regs, std::uint32_t numRegs);
+std::uint32_t* APS5_VABI sceAgcCbSetShRegisterRangeDirect(CommandBuffer* buf, std::uint32_t offset, const std::uint32_t* values, std::uint32_t numValues);
+std::uint32_t* APS5_VABI sceAgcAcbPushMarker(CommandBuffer* buf, const char* str, std::uint32_t color);
+std::uint32_t* APS5_VABI sceAgcAcbPopMarker(CommandBuffer* buf);
+std::uint32_t* APS5_VABI sceAgcCbDispatch(CommandBuffer* buf, std::uint32_t threadGroupX, std::uint32_t threadGroupY, std::uint32_t threadGroupZ, std::uint32_t modifier);
+std::uint32_t* APS5_VABI sceAgcAcbDispatchIndirect(CommandBuffer* buf, const volatile void* indirectArguments, std::uint32_t modifier);
+std::uint32_t* APS5_VABI sceAgcCbReleaseMem(CommandBuffer* buf, std::uint8_t action, std::uint16_t gcrControl, std::uint8_t dst, std::uint8_t cachePolicy, const volatile Label* address, std::uint8_t dataSelect, std::uint64_t data, std::uint16_t gdsOffset, std::uint16_t gdsSize, std::uint8_t interrupt, std::uint32_t interruptContextId);
 }
 
 namespace {
@@ -50,6 +68,11 @@ constexpr std::uint64_t StateEnd = 0x2d8;
 constexpr std::uint64_t ThreadRecordBytes = 0x48;
 constexpr std::uint64_t SequenceBytes = 0x240;
 constexpr std::uint64_t NodeBytes = 72;
+constexpr std::uint64_t ComputeContextBytes = 0x168;
+constexpr std::uint64_t ComputeCommandBytes = 0x3ff * 104 + 0xff * 148 + 0x3f * 32;
+constexpr std::uint32_t FirstComputeQueue = 0x20;
+constexpr std::uint32_t UserDataRegister = 0x240;
+constexpr int UserEventFilter = -11;
 
 struct MemorySizeQueryParams {
     std::uint32_t version;
@@ -94,7 +117,7 @@ std::uint64_t AlignUp(std::uint64_t value, std::uint64_t alignment) {
 struct Manager;
 class Sequence;
 
-enum class ItemType { Dispatch, SyncPoint, Wait };
+enum class ItemType { Dispatch, SyncPoint, Wait, Compute, ComputeIndirect };
 
 struct Item {
     ItemType type;
@@ -105,6 +128,11 @@ struct Item {
     std::uint32_t handedOut;
     std::uint32_t completed;
     bool* passed;
+    const void* shader = nullptr;
+    std::array<std::uint32_t, 3> groups{};
+    const void* indirectArguments = nullptr;
+    std::string name;
+    bool named = false;
 };
 
 struct Counters {
@@ -140,27 +168,26 @@ public:
     APS5_VABI int Dispatch(JobEntry entry, void* argument, std::uint32_t count, const char* name) override;
     APS5_VABI int SyncPoint() override;
     APS5_VABI int Wait() override;
-    APS5_VABI int DispatchCompute(const void*, void*, std::int32_t, std::int32_t, std::int32_t, const char*) override {
-        NotImplemented_nid_no_patch("sce::Job compute dispatch");
-        return 0;
-    }
-    APS5_VABI int DispatchComputeIndirect(const void*, void*, const void*, const char*) override {
-        NotImplemented_nid_no_patch("sce::Job indirect compute dispatch");
-        return 0;
-    }
+    APS5_VABI int DispatchCompute(const void* shader, void* userData, std::int32_t x, std::int32_t y, std::int32_t z, const char* name) override;
+    APS5_VABI int DispatchComputeIndirect(const void* shader, void* userData, const void* indirectArguments, const char* name) override;
 
     bool Open(Manager* owner, const Counters& limit, std::uint32_t runQueue);
     int WaitAndClose(std::uint32_t index);
     void Advance();
     void DispatchCompleted(std::uint32_t invocations);
+    void CollectCompute();
 
     Manager* manager = nullptr;
     bool open = false;
+    bool compute = false;
+    std::uint32_t queue = 0;
 
 private:
     int Latch(int code);
     void RequireOpen() const;
     int WaitLocked(std::unique_lock<std::mutex>& guard);
+    int AppendCompute(Item item);
+    void SubmitCompute(const Item& item);
 
     Counters limits;
     Counters submitted;
@@ -172,6 +199,11 @@ private:
     std::deque<Item> items;
     std::mutex waiters;
     std::condition_variable passed;
+    Label label{};
+    std::uint64_t labelSeen = 0;
+    std::uint64_t labelTotal = 0;
+    bool batchStarted = false;
+    std::deque<std::vector<std::uint32_t>> commands;
 };
 
 class Factory final : public ISequenceFactory {
@@ -196,6 +228,11 @@ struct Manager {
     std::uint32_t persistentThreads = 0;
     std::array<std::deque<Item*>, 2> runQueues;
     std::vector<Sequence> sequences;
+    std::deque<Sequence> computeSequences;
+    std::uint32_t freeComputeSequences = 0;
+    std::vector<std::uint32_t> computeQueues;
+    KernelEqueue computeEvents = 0;
+    Pthread computeHandler = nullptr;
     std::array<Factory, 2> factories;
     std::array<Factory, 2> unifiedFactories;
     std::vector<Pthread> workers;
@@ -231,6 +268,8 @@ bool Sequence::Open(Manager* owner, const Counters& limit, std::uint32_t runQueu
     outstanding = 0;
     cursor = 0;
     items.clear();
+    commands.clear();
+    batchStarted = false;
     if (!TakeNode(*owner)) return false;
     open = true;
     return true;
@@ -250,6 +289,12 @@ void Sequence::Advance() {
             manager->work.notify_all();
             continue;
         }
+        if (item.type == ItemType::Compute || item.type == ItemType::ComputeIndirect) {
+            ++outstanding;
+            ++cursor;
+            SubmitCompute(item);
+            continue;
+        }
         if (outstanding != 0) return;
         if (item.type == ItemType::SyncPoint) {
             ++completed.syncPoints;
@@ -260,6 +305,8 @@ void Sequence::Advance() {
         const auto retired = static_cast<std::uint32_t>(cursor) + 1;
         items.erase(items.begin(), items.begin() + static_cast<std::ptrdiff_t>(retired));
         cursor = 0;
+        commands.clear();
+        batchStarted = false;
         if (error == 0) {
             submitted.invocations -= completed.invocations;
             submitted.dispatches -= completed.dispatches;
@@ -296,6 +343,81 @@ int Sequence::Dispatch(JobEntry entry, void* argument, std::uint32_t count, cons
     return 0;
 }
 
+void Sequence::SubmitCompute(const Item& item) {
+    const auto* bytes = static_cast<const std::uint8_t*>(item.shader);
+    const ShaderRegister* registers = nullptr;
+    const std::uint8_t* specials = nullptr;
+    std::memcpy(&registers, bytes + 0x20, sizeof(registers));
+    std::memcpy(&specials, bytes + 0x28, sizeof(specials));
+    const std::uint32_t registerCount = bytes[0x5c];
+    std::uint32_t modifier = 0;
+    std::memcpy(&modifier, specials + 0x10, sizeof(modifier));
+    const auto userData = reinterpret_cast<std::uint64_t>(item.argument);
+    const std::array<std::uint32_t, 2> userDataWords{static_cast<std::uint32_t>(userData), static_cast<std::uint32_t>(userData >> 32)};
+    auto& words = commands.emplace_back(64 + 3 * registerCount + item.name.size() / 4);
+    CommandBuffer buffer{words.data(), words.data() + words.size(), words.data(), words.data() + words.size(), nullptr, nullptr, 0};
+    bool encoded = sceAgcAcbAcquireMem(&buffer, 0x20380u | (batchStarted ? 0u : 1u), nullptr, 0, 0x190) != nullptr;
+    encoded = encoded && sceAgcCbSetShRegistersDirect(&buffer, registers, registerCount) != nullptr;
+    encoded = encoded && sceAgcCbSetShRegisterRangeDirect(&buffer, UserDataRegister, userDataWords.data(), 2) != nullptr;
+    if (item.named) encoded = encoded && sceAgcAcbPushMarker(&buffer, item.name.c_str(), 0) != nullptr;
+    if ((modifier & 1) != 0) {
+        if (item.type == ItemType::Compute) {
+            encoded = encoded && sceAgcCbDispatch(&buffer, item.groups[0], item.groups[1], item.groups[2], modifier) != nullptr;
+        } else {
+            encoded = encoded && sceAgcAcbDispatchIndirect(&buffer, item.indirectArguments, modifier) != nullptr;
+        }
+    }
+    if (item.named) encoded = encoded && sceAgcAcbPopMarker(&buffer) != nullptr;
+    encoded = encoded && sceAgcCbReleaseMem(&buffer, 0x2f, 0x620, 0, 0, &label, 1, ++labelTotal, 0, 0, 2, 0) != nullptr;
+    if (!encoded) throw std::runtime_error("libSceJobManager: a compute dispatch did not fit its command buffer");
+    batchStarted = true;
+    const Packet packet{words.data(), static_cast<std::uint32_t>(buffer.cursor_up - words.data()), 0, {}};
+    if (sceAgcDriverSubmitAcb(queue, &packet) != 0) throw std::runtime_error("libSceJobManager: sceAgcDriverSubmitAcb failed");
+}
+
+void Sequence::CollectCompute() {
+    const std::uint64_t value = label.value;
+    const auto finished = static_cast<std::uint32_t>(value - labelSeen);
+    labelSeen = value;
+    if (finished == 0) return;
+    completed.dispatches += finished;
+    outstanding -= finished;
+    if (outstanding == 0) Advance();
+}
+
+int Sequence::AppendCompute(Item item) {
+    const std::lock_guard guard(manager->lock);
+    RequireOpen();
+    if (!compute) throw std::runtime_error("libSceJobManager: a compute dispatch was made on a CPU sequence");
+    if (error != 0) return SequenceInvalid;
+    if (submitted.dispatches++ >= limits.dispatches) return Latch(SequenceLimitExceeded);
+    if (!TakeNode(*manager)) return Latch(SequenceNoNodes);
+    items.push_back(std::move(item));
+    Advance();
+    return 0;
+}
+
+int Sequence::DispatchCompute(const void* shader, void* userData, std::int32_t x, std::int32_t y, std::int32_t z, const char* name) {
+    if (shader == nullptr) return SequenceNullArgument;
+    if (x <= 0 || y <= 0 || z <= 0) return SequenceOutOfRange;
+    Item item{ItemType::Compute, this, nullptr, userData, 0, 0, 0, nullptr};
+    item.shader = shader;
+    item.groups = {static_cast<std::uint32_t>(x), static_cast<std::uint32_t>(y), static_cast<std::uint32_t>(z)};
+    item.named = name != nullptr;
+    if (item.named) item.name = name;
+    return AppendCompute(std::move(item));
+}
+
+int Sequence::DispatchComputeIndirect(const void* shader, void* userData, const void* indirectArguments, const char* name) {
+    if (shader == nullptr || indirectArguments == nullptr || (reinterpret_cast<std::uint64_t>(indirectArguments) & 0x1f) != 0) return SequenceNullArgument;
+    Item item{ItemType::ComputeIndirect, this, nullptr, userData, 0, 0, 0, nullptr};
+    item.shader = shader;
+    item.indirectArguments = indirectArguments;
+    item.named = name != nullptr;
+    if (item.named) item.name = name;
+    return AppendCompute(std::move(item));
+}
+
 int Sequence::SyncPoint() {
     const std::lock_guard guard(manager->lock);
     RequireOpen();
@@ -318,10 +440,11 @@ int Sequence::WaitAndClose(std::uint32_t index) {
     const int result = WaitLocked(guard);
     ReturnNodes(*manager, static_cast<std::uint32_t>(items.size()) + 1);
     items.clear();
+    commands.clear();
     cursor = 0;
     outstanding = 0;
     open = false;
-    manager->freeSequences |= std::uint32_t{1} << index;
+    (compute ? manager->freeComputeSequences : manager->freeSequences) |= std::uint32_t{1} << index;
     return result;
 }
 
@@ -353,42 +476,63 @@ bool ValidLimits(std::uint32_t invocations, std::uint32_t dispatches, std::uint3
     return invocations != 0 || dispatches != 0 || syncPoints != 0 || waits != 0;
 }
 
-int Factory::Create(ISequence** out, std::uint32_t invocations, std::uint32_t dispatches, std::uint32_t syncPoints, std::uint32_t waits, const char*) {
+template <typename TSequences>
+int CreateIn(Manager* manager, TSequences& sequences, std::uint32_t& free, std::uint32_t priority, ISequence** out, const Counters& limits) {
     if (out == nullptr) return SequenceNullArgument;
-    if (!ValidLimits(invocations, dispatches, syncPoints, waits)) return SequenceOutOfRange;
+    if (!ValidLimits(limits.invocations, limits.dispatches, limits.syncPoints, limits.waits)) return SequenceOutOfRange;
     const std::lock_guard guard(manager->lock);
-    if (manager->freeSequences == 0) return SequenceNoneFree;
-    const int index = std::countr_zero(manager->freeSequences);
-    manager->freeSequences &= ~(std::uint32_t{1} << index);
-    Sequence& sequence = manager->sequences[static_cast<std::size_t>(index)];
-    if (!sequence.Open(manager, {invocations, dispatches, syncPoints, waits}, priority)) return SequenceNoNodes;
+    if (free == 0) return SequenceNoneFree;
+    const int index = std::countr_zero(free);
+    free &= ~(std::uint32_t{1} << index);
+    Sequence& sequence = sequences[static_cast<std::size_t>(index)];
+    if (!sequence.Open(manager, limits, priority)) return SequenceNoNodes;
     *out = &sequence;
     return 0;
 }
 
-int Factory::Destroy(ISequence* sequence) {
+template <typename TSequences>
+int DestroyIn(Manager* manager, TSequences& sequences, const std::uint32_t& free, ISequence* sequence) {
     Sequence* found = nullptr;
     std::uint32_t index = 0;
     {
         const std::lock_guard guard(manager->lock);
-        for (; index < manager->sequences.size(); ++index) {
-            if (&manager->sequences[index] == sequence) break;
+        for (; index < sequences.size(); ++index) {
+            if (&sequences[index] == sequence) break;
         }
-        if (index == manager->sequences.size() || (manager->freeSequences & (std::uint32_t{1} << index)) != 0) return SequenceInvalid;
-        found = &manager->sequences[index];
+        if (index == sequences.size() || (free & (std::uint32_t{1} << index)) != 0) return SequenceInvalid;
+        found = &sequences[index];
     }
     static_cast<void>(found->WaitAndClose(index));
     return 0;
 }
 
-int Factory::CreateCompute(ISequence** out, std::uint32_t invocations, std::uint32_t dispatches, std::uint32_t syncPoints, std::uint32_t waits, const char*) {
-    if (out == nullptr) return SequenceNullArgument;
-    if (!ValidLimits(invocations, dispatches, syncPoints, waits)) return SequenceOutOfRange;
-    return SequenceNoneFree;
+int Factory::Create(ISequence** out, std::uint32_t invocations, std::uint32_t dispatches, std::uint32_t syncPoints, std::uint32_t waits, const char*) {
+    return CreateIn(manager, manager->sequences, manager->freeSequences, priority, out, {invocations, dispatches, syncPoints, waits});
 }
 
-int Factory::DestroyCompute(ISequence*) {
-    return SequenceInvalid;
+int Factory::Destroy(ISequence* sequence) {
+    return DestroyIn(manager, manager->sequences, manager->freeSequences, sequence);
+}
+
+int Factory::CreateCompute(ISequence** out, std::uint32_t invocations, std::uint32_t dispatches, std::uint32_t syncPoints, std::uint32_t waits, const char*) {
+    return CreateIn(manager, manager->computeSequences, manager->freeComputeSequences, priority, out, {invocations, dispatches, syncPoints, waits});
+}
+
+int Factory::DestroyCompute(ISequence* sequence) {
+    return DestroyIn(manager, manager->computeSequences, manager->freeComputeSequences, sequence);
+}
+
+void* APS5_VABI ComputeHandlerEntry(void* argument) {
+    Manager& manager = *static_cast<Manager*>(argument);
+    for (;;) {
+        KernelEvent event{};
+        int count = 0;
+        static_cast<void>(sceKernelWaitEqueue(manager.computeEvents, &event, 1, &count, nullptr));
+        if (count <= 0) continue;
+        if (sceKernelGetEventFilter(&event) == UserEventFilter && sceKernelGetEventId(&event) == 0) return nullptr;
+        const std::lock_guard guard(manager.lock);
+        for (Sequence& sequence : manager.computeSequences) sequence.CollectCompute();
+    }
 }
 
 bool RunOne(Manager& manager, std::unique_lock<std::mutex>& guard) {
@@ -421,11 +565,16 @@ void* APS5_VABI WorkerEntry(void* argument) {
     return nullptr;
 }
 
-std::uint32_t NodeCount(std::uint64_t memory, std::uint64_t size, std::uint32_t workers, std::uint32_t sequences, bool* fits) {
+std::uint32_t NodeCount(std::uint64_t memory, std::uint64_t size, std::uint32_t workers, std::uint32_t sequences, std::uint32_t computeQueues, bool* fits) {
     const std::uint64_t end = memory + size;
     std::uint64_t address = AlignUp(memory + StateEnd, 8) + workers * ThreadRecordBytes;
     for (const std::uint64_t bytes : {std::uint64_t{0x100}, std::uint64_t{0x80}, std::uint64_t{0x80}, std::uint64_t{0x80}}) address = AlignUp(address, 64) + bytes;
-    address = AlignUp(AlignUp(address, 64) + sequences * SequenceBytes, 64);
+    address = AlignUp(address, 64) + (sequences + computeQueues) * SequenceBytes;
+    if (computeQueues != 0) {
+        address = AlignUp(address + ComputeContextBytes, 0x100) + AlignUp(ComputeCommandBytes * computeQueues, 8);
+        address += AlignUp(4 * std::uint64_t{computeQueues}, 8) + 8 * std::uint64_t{computeQueues};
+    }
+    address = AlignUp(address, 64);
     *fits = address <= end;
     return *fits ? static_cast<std::uint32_t>(((end - address) / NodeBytes) & 0xffff) : 0;
 }
@@ -462,6 +611,30 @@ void StartWorkers(Manager& manager, const ConfigParams& config, std::uint32_t wo
     }
 }
 
+void StartCompute(Manager& manager, std::uint32_t computeSequences, std::uint64_t computeQueueMask, std::uint32_t priority, std::uint32_t affinity) {
+    if (sceKernelCreateEqueue(&manager.computeEvents, "SceJobManagerComputeEventQueue") != 0) throw std::runtime_error("libSceJobManager: sceKernelCreateEqueue failed");
+    for (std::uint64_t mask = computeQueueMask; mask != 0; mask &= mask - 1) {
+        const std::uint32_t queue = FirstComputeQueue + static_cast<std::uint32_t>(std::countr_zero(mask));
+        if (sceAgcDriverAddEqEvent(manager.computeEvents, static_cast<int>(queue), nullptr) != 0) throw std::runtime_error("libSceJobManager: sceAgcDriverAddEqEvent failed");
+        manager.computeQueues.push_back(queue);
+    }
+    manager.computeSequences.resize(computeSequences);
+    for (std::uint32_t index = 0; index < computeSequences; ++index) {
+        manager.computeSequences[index].compute = true;
+        manager.computeSequences[index].queue = manager.computeQueues[index];
+    }
+    manager.freeComputeSequences = computeSequences == MaxSequences ? ~std::uint32_t{0} : (std::uint32_t{1} << computeSequences) - 1;
+    const KernelSchedParam schedule{static_cast<int>(priority)};
+    PthreadAttr attributes = nullptr;
+    if (scePthreadAttrInit(&attributes) != 0 || scePthreadAttrSetinheritsched(&attributes, 0) != 0 || scePthreadAttrSetschedparam(&attributes, &schedule) != 0 ||
+        scePthreadAttrSetstacksize(&attributes, 0x4000) != 0 || scePthreadAttrSetaffinity(&attributes, affinity) != 0) {
+        throw std::runtime_error("libSceJobManager: a compute handler thread attribute was rejected");
+    }
+    const int created = scePthreadCreate(&manager.computeHandler, &attributes, ComputeHandlerEntry, &manager, "SceJobManagerComputeEventHdlr");
+    scePthreadAttrDestroy(&attributes);
+    if (created != 0) throw std::runtime_error("libSceJobManager: scePthreadCreate failed for the compute handler");
+}
+
 int Initialize(void* self, void* memory, std::uint64_t size, std::uint32_t workers, std::uint32_t sequences, std::uint32_t computeSequences, std::uint64_t computeQueueMask, const ConfigParams* config, bool unified) {
     if (Implementation(self) != nullptr) return ManagerAlreadyInitialized;
     if (memory == nullptr) return ManagerNullMemory;
@@ -479,12 +652,16 @@ int Initialize(void* self, void* memory, std::uint64_t size, std::uint32_t worke
     const BlockHeader header{0x4d4a, static_cast<std::uint16_t>(StateOffset), 0x05000001, static_cast<std::uint32_t>(size), 0};
     std::memcpy(memory, &header, sizeof(header));
     if (config != nullptr && config->version > 1) return ManagerInvalidArgument;
-    if (computeSequences != 0) NotImplemented_nid_no_patch("sce::Job::JobManager::initialize with compute sequences");
     const ConfigParams& settings = config != nullptr ? *config : DefaultConfig;
+    const auto computeQueues = computeSequences != 0 ? static_cast<std::uint32_t>(std::popcount(computeQueueMask)) : 0;
     bool fits = false;
-    const std::uint32_t nodes = NodeCount(base, size, workers, sequences, &fits);
+    const std::uint32_t nodes = NodeCount(base, size, workers, sequences, computeQueues, &fits);
     if (!fits || nodes == 0) return ManagerMemoryTooSmall;
     if (const int invalid = ValidateWorkers(settings, workers); invalid != 0) return invalid;
+    const bool ownHandlerSettings = (settings.version & 1) != 0;
+    const std::uint32_t handlerPriority = ownHandlerSettings ? settings.computePriority : 0x100;
+    const std::uint32_t handlerAffinity = ownHandlerSettings ? settings.computeAffinity : 0x1fff;
+    if (computeSequences != 0 && (handlerPriority < 0x100 || handlerPriority > 0x2ff || handlerAffinity == 0 || handlerAffinity > 0x1fff)) return ManagerInvalidArgument;
     auto* manager = new Manager;
     manager->running = true;
     manager->freeNodes = nodes - 1;
@@ -496,6 +673,7 @@ int Initialize(void* self, void* memory, std::uint64_t size, std::uint32_t worke
         manager->unifiedFactories[priority].manager = manager;
         manager->unifiedFactories[priority].priority = priority;
     }
+    if (computeSequences != 0) StartCompute(*manager, computeSequences, computeQueueMask, handlerPriority, handlerAffinity);
     Implementation(self) = manager;
     StartWorkers(*manager, settings, workers);
     return 0;
@@ -539,12 +717,19 @@ std::uint64_t APS5_VABI _ZN3sce3Job10JobManager27calculateRequiredMemorySizeEPKN
     }
     const std::uint64_t total = params->sequences + computeSequences;
     if (params->invocations > total * MaxInvocations || params->dispatches > total * MaxDispatches || params->syncPoints > total * MaxSyncPoints || params->waits > total * MaxWaits) return 0;
-    if (computeSequences != 0) NotImplemented_nid_no_patch("sce::Job::JobManager::calculateRequiredMemorySize with compute sequences");
     std::uint64_t bytes = (NodeBytes * params->workers + 0x317) & 0xfc0;
     bytes = (bytes + 0x13f) & ~std::uint64_t{0x3f};
     bytes = (bytes + 0x0bf) & ~std::uint64_t{0x3f};
     bytes = (bytes + 0x13f) & ~std::uint64_t{0x3f};
-    bytes += params->sequences * SequenceBytes;
+    if (computeSequences != 0) {
+        const auto queues = static_cast<std::uint64_t>(std::popcount(params->computeQueueMask));
+        bytes = (bytes + (params->sequences + queues) * SequenceBytes + 0x267) & ~std::uint64_t{0xff};
+        bytes = (bytes + ComputeCommandBytes * queues + 3) & ~std::uint64_t{3};
+        bytes = (bytes + 4 * queues + 7) & ~std::uint64_t{7};
+        bytes += 8 * queues;
+    } else {
+        bytes += params->sequences * SequenceBytes;
+    }
     const std::uint32_t nodes = params->dispatches + params->syncPoints + params->waits + static_cast<std::uint32_t>(total) + 1;
     return AlignUp(bytes, 64) + std::uint64_t{nodes} * NodeBytes + 0x800;
 }
@@ -569,6 +754,14 @@ int APS5_VABI _ZN3sce3Job10JobManager8shutdownEv(void* self) {
     for (const Pthread worker : manager->workers) {
         if (scePthreadJoin(worker, nullptr) != 0) throw std::runtime_error("sce::Job::JobManager::shutdown: scePthreadJoin failed");
     }
+    if (manager->computeHandler != nullptr) {
+        if (sceKernelAddUserEventEdge(manager->computeEvents, 0) != 0 || sceKernelTriggerUserEvent(manager->computeEvents, 0, nullptr) != 0) {
+            throw std::runtime_error("sce::Job::JobManager::shutdown: the compute handler could not be signalled");
+        }
+        if (scePthreadJoin(manager->computeHandler, nullptr) != 0) throw std::runtime_error("sce::Job::JobManager::shutdown: scePthreadJoin failed");
+        for (const std::uint32_t queue : manager->computeQueues) static_cast<void>(sceAgcDriverDeleteEqEvent(manager->computeEvents, static_cast<int>(queue)));
+        if (sceKernelDeleteEqueue(manager->computeEvents) != 0) throw std::runtime_error("sce::Job::JobManager::shutdown: sceKernelDeleteEqueue failed");
+    }
     Implementation(self) = nullptr;
     delete manager;
     return 0;

docs/dev/debt/libSceJobManager.md

@@ -2,4 +2,4 @@
 
 ### Functional
 
-- [sce::Job::JobManager](../../../core/libs/prx/libSceJobManager/Export.cpp) (libSceJobManager) is written from the disassembly of the copies PPSA01474 and PPSA36082 ship in `sce_module`, and `guest_job_manager_tests` passes against both of those modules as well: the memory size formula, the error codes, sync point ordering, the per-sequence limits and the priority order are the original's. Only CPU sequences exist: `initialize` and `calculateRequiredMemorySize` with compute sequences, and both compute dispatches, are not implemented, and `createCompute` and `destroyCompute` return the errors the original returns when no compute sequence was configured. The manager's state lives on the host heap, with only the block header written to the title's memory; the original keeps everything in that block, and a block smaller than 0x2d8 bytes throws here. The worker sleep policy, the spin count and the persistent thread count are validated and stored but change nothing, a factory priority above 1 throws, `shutdown` throws while a thread is blocked in a sequence wait, and using a sequence after its `destroy` throws. `initialize` returns 0 where the original returns the result of its last import (`BG26hBGiNlw`, not resolved to a name), the `sceRazorCpuJobManager*` profiler calls are not made, and a worker thread that cannot be created or configured throws where the original does not check. The order of the sequence and factory virtual functions is read from the modules' vtables only, since no title that calls them has been run
+- [sce::Job::JobManager](../../../core/libs/prx/libSceJobManager/Export.cpp) (libSceJobManager) is written from the disassembly of the copies PPSA01474 and PPSA36082 ship in `sce_module`, and `guest_job_manager_tests` passes against both of those modules as well: the memory size formula, the error codes, sync point ordering, the per-sequence limits and the priority order are the original's. Compute sequences submit through `sceAgcDriverSubmitAcb` on the queue their mask bit selects and are checked the same way by `agc_driver_job_manager_compute_tests`: results, ordering against CPU dispatches and sync points, limits and sizes match the two modules. The command stream is simpler than the original's: every compute dispatch is submitted with its own label write and interrupt, where the original batches the label write and rewinds commands it encoded ahead, and the command buffers and labels live on the host heap. The per-queue command size in `calculateRequiredMemorySize` is a constant (104, 148 and 32 bytes per dispatch, sync point and wait), which is what the original measures by running the builders of this project's libSceAgc. A compute dispatch on a CPU sequence throws (the original does not check), and a failing `sceAgcDriverAddEqEvent` throws. The manager's state lives on the host heap, with only the block header written to the title's memory; the original keeps everything in that block, and a block smaller than 0x2d8 bytes throws here. The worker sleep policy, the spin count and the persistent thread count are validated and stored but change nothing, a factory priority above 1 throws, `shutdown` throws while a thread is blocked in a sequence wait, and using a sequence after its `destroy` throws. `initialize` returns 0 where the original returns the result of its last import (`BG26hBGiNlw`, not resolved to a name), the `sceRazorCpuJobManager*` profiler calls are not made, and a worker thread that cannot be created or configured throws where the original does not check. The order of the sequence and factory virtual functions is read from the modules' vtables only, since no title that calls them has been run