fix(libSceJobManager): batch compute completion in caller storage

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

@@ -1,5 +1,6 @@
 #include "VulkanTestDevice.hpp"
 #include "SceTypes.hpp"
+#include "prx/libc/include/GuestArena.hpp"
 #include <array>
 #include <atomic>
 #include <chrono>
@@ -11,6 +12,7 @@
 #include <stdexcept>
 #include <string>
 #include <thread>
+#include <vector>
 
 extern "C" void AgcDriverShutdown_nid_postfix();
 
@@ -60,6 +62,9 @@ struct SequenceFactory {
 };
 
 extern "C" {
+int APS5_VABI sceKernelAllocateDirectMemory(std::int64_t, std::int64_t, std::size_t, std::size_t, int, std::int64_t*);
+int APS5_VABI sceKernelMapDirectMemory(void**, std::size_t, int, int, std::int64_t, std::size_t);
+int APS5_VABI sceKernelReleaseDirectMemory(std::int64_t, std::size_t);
 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*);
@@ -94,10 +99,10 @@ 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{};
+alignas(16) std::array<std::uint32_t, 8> SpecialsDefaultModifier{};
 std::array<ShaderRegister, 9> Registers{};
 alignas(16) std::array<unsigned char, 0x80> Shader{};
-alignas(16) std::array<unsigned char, 0x80> ShaderWithoutDispatch{};
+alignas(16) std::array<unsigned char, 0x80> ShaderDefaultModifier{};
 
 std::uint32_t Low(const volatile void* pointer) {
     return static_cast<std::uint32_t>(reinterpret_cast<std::uintptr_t>(pointer));
@@ -115,8 +120,8 @@ void BuildShader() {
                   {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();
+    for (auto* shader : {&Shader, &ShaderDefaultModifier}) {
+        const std::uint32_t* specials = shader == &Shader ? Specials.data() : SpecialsDefaultModifier.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());
@@ -144,10 +149,10 @@ 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(1, 1, false) == 0x29fd8 && SizeFor(1, 1, true) == 0x2a3c8, "size with one compute sequence");
+    Require(SizeFor(2, 3, false) == 0x52de0 && SizeFor(2, 3, true) == 0x531d0, "size with two compute sequences");
+    Require(SizeFor(3, 7, false) == 0x7bb28, "size with three compute sequences");
+    Require(SizeFor(1, 3, true) == 0x53188, "size with two queues for one compute sequence");
     Require(SizeFor(2, 1, true) == 0, "size with fewer queues than compute sequences");
 }
 
@@ -192,24 +197,30 @@ void RunCase() {
     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->DispatchCompute(ShaderDefaultModifier.data(), Table.data(), 1, 1, 1, nullptr) == 0, "the dispatch with a default modifier failed");
+    Require(sequence->Wait() == 0, "the wait after a default modifier dispatch failed");
+    Require(Output[0] == 0x33330003u, "a default modifier dispatch did not run");
 
+    Value[0] = 0x33330004u;
+    Require(sequence->DispatchComputeIndirect(ShaderDefaultModifier.data(), Table.data(), IndirectArguments.data(), nullptr) == 0, "the indirect dispatch with a default modifier failed");
+    Require(sequence->Wait() == 0, "the wait after an indirect default modifier dispatch failed");
+    Require(Output[0] == 0x33330004u, "an indirect default modifier dispatch did not run");
+
+    Value[0] = 0x33330005u;
     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");
+    Require(Output[0] == 0x33330004u, "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");
+    Require(Output[0] == 0x33330005u, "the compute dispatch behind a sync point did not run");
+    Require(SeenByJob == 0x33330005u, "a CPU job behind a sync point ran before the compute dispatch ahead of it finished");
 
     Value[0] = 0x44440004u;
-    Require(sequence->DispatchCompute(ShaderWithoutDispatch.data(), Table.data(), 1, 1, 1, nullptr) == 0, "the leading dispatch failed");
+    Require(sequence->DispatchCompute(ShaderDefaultModifier.data(), Table.data(), 1, 1, 1, nullptr) == 0, "the leading dispatch failed");
     Require(sequence->SyncPoint() == 0, "the sync point between compute dispatches failed");
     Require(sequence->DispatchCompute(Shader.data(), Table.data(), 1, 1, 1, "behind") == 0, "the compute dispatch behind a compute dispatch failed");
     Require(sequence->Wait() == 0, "the wait after chained compute dispatches failed");
@@ -229,16 +240,83 @@ void RunCase() {
     ::operator delete(memory, std::align_val_t{256});
 }
 
+void StorageCase(bool directMemory) {
+    constexpr std::uint64_t mask = (1u << 1) | (1u << 5) | (1u << 8);
+    const MemorySizeQueryParams query{1, 2, 0, 0, 32, 8, 4, 2, mask};
+    const auto size = RequiredSize(&query);
+    const auto allocationSize = (size + 0xffff) & ~std::uint64_t{0xffff};
+    std::int64_t physical = 0;
+    void* memory = nullptr;
+    if (directMemory) {
+        Require(sceKernelAllocateDirectMemory(0, 0x7fffffffffll, allocationSize, 0x10000, 0, &physical) == 0, "direct work area allocation failed");
+        Require(sceKernelMapDirectMemory(&memory, allocationSize, 0x33, 0, physical, 0x10000) == 0, "direct work area mapping failed");
+    } else {
+        memory = ::operator new(allocationSize, std::align_val_t{256});
+    }
+    {
+        const GuestArena::HostWrite destination(memory, allocationSize);
+        std::memset(memory, 0, allocationSize);
+        std::memset(static_cast<std::uint8_t*>(memory) + size, 0xa5, allocationSize - size);
+    }
+    alignas(64) unsigned char manager[0x100]{};
+    Construct(manager);
+    Require(Initialize(manager, memory, size, 2, 0, 2, mask, nullptr) == 0, "storage manager initialization failed");
+    auto* factory = Factories(manager, 0);
+    std::array<Sequence*, 2> sequences{};
+    for (auto& sequence : sequences) Require(factory->CreateCompute(&sequence, 0, 16, 4, 2, nullptr) == 0, "storage sequence creation failed");
+    for (std::uint32_t round = 0; round < 3; ++round) {
+        for (auto* sequence : sequences) {
+            Value[0] = 0x56780000u + round;
+            for (int dispatch = 0; dispatch < 8; ++dispatch)
+                Require(sequence->DispatchCompute(Shader.data(), Table.data(), 1, 1, 1, "batch") == 0, "batched dispatch failed");
+            Require(sequence->Wait() == 0, "batched wait failed");
+            Require(Output[0] == Value[0], "batched work was not visible after wait");
+            Require(sequence->Wait() == 0, "empty wait after a batch failed");
+        }
+        const auto* words = static_cast<const std::uint32_t*>(memory);
+        const auto base = reinterpret_cast<std::uintptr_t>(memory);
+        std::vector<std::uintptr_t> labels;
+        for (std::size_t index = 0; index + 8 <= size / 4; ++index) {
+            if (words[index] != 0xc0064900u || (words[index + 2] & 0xe7000000u) != 0x22000000u) continue;
+            const auto address = words[index + 3] | (std::uint64_t{words[index + 4]} << 32);
+            Require(address >= base && address + 8 <= base + size && (address & 7) == 0, "a completion label is outside the caller's work area");
+            Require(words[index + 5] == 8 * (round + 1) && words[index + 6] == 0, "a completion did not publish the batched 32-bit count");
+            Require(*reinterpret_cast<const volatile std::uint64_t*>(address) == words[index + 5], "a published batch was not retired");
+            labels.push_back(address);
+        }
+        Require(labels.size() == 2 && labels[0] != labels[1], "each queue must store one batch completion in the caller's work area");
+    }
+    for (auto* sequence : sequences) Require(factory->DestroyCompute(sequence) == 0, "storage sequence destruction failed");
+    for (auto& sequence : sequences) {
+        Require(factory->CreateCompute(&sequence, 0, 16, 4, 2, nullptr) == 0, "storage sequence reuse failed");
+        Require(sequence->DispatchCompute(Shader.data(), Table.data(), 1, 1, 1, nullptr) == 0, "dispatch after sequence reuse failed");
+        Require(factory->DestroyCompute(sequence) == 0, "destroy did not wait for reused sequence work");
+    }
+    Require(Shutdown(manager) == 0, "storage manager shutdown failed");
+    const auto* bytes = static_cast<const std::uint8_t*>(memory);
+    for (auto offset = size; offset < allocationSize; ++offset)
+        Require(bytes[offset] == 0xa5, "the manager wrote beyond its supplied work area");
+    if (directMemory) Require(sceKernelReleaseDirectMemory(physical, allocationSize) == 0, "direct work area release failed");
+    else ::operator delete(memory, std::align_val_t{256});
 }
 
-int main() {
+}
+
+int RunJobManagerCase(const char* name) {
     try {
+        Require(std::strcmp(name, "compute") == 0 || std::strcmp(name, "size") == 0 || std::strcmp(name, "execution") == 0 || std::strcmp(name, "storage") == 0, "unknown compute fixture");
+#ifndef ANYPS5_JOB_MANAGER_REFERENCE
         auto device = OpenVulkanTestDevice();
         if (!device) return VulkanTestSkipped;
         device.reset();
+#endif
         BuildShader();
-        SizeCase();
-        RunCase();
+        if (std::strcmp(name, "compute") == 0 || std::strcmp(name, "size") == 0) SizeCase();
+        if (std::strcmp(name, "compute") == 0 || std::strcmp(name, "execution") == 0) RunCase();
+        if (std::strcmp(name, "compute") == 0 || std::strcmp(name, "storage") == 0) {
+            StorageCase(false);
+            StorageCase(true);
+        }
         AgcDriverShutdown_nid_postfix();
         std::cout << "job manager compute test passed\n";
         return 0;
@@ -252,3 +330,10 @@ int main() {
         return 1;
     }
 }
+
+#ifndef ANYPS5_JOB_MANAGER_REFERENCE
+int main(int argc, char** argv) {
+    if (argc > 2) return 2;
+    return RunJobManagerCase(argc == 2 ? argv[1] : "compute");
+}
+#endif

core/libs/prx/libSceJobManager/CMakeLists.txt

@@ -4,6 +4,16 @@ target_link_libraries(libSceJobManager PRIVATE libSceAgc libSceAgcDriver)
 configure_windows_unwind(libSceJobManager)
 
 if(BUILD_TESTING)
+    add_test_executable(job_manager_compute_progress_tests ${LIBS_INCLUDE_DIR}/tests/JobManagerComputeProgress.cpp)
+    target_include_directories(job_manager_compute_progress_tests PRIVATE ${LIBS_INCLUDE_DIR})
+    target_link_libraries(job_manager_compute_progress_tests PRIVATE libc)
+    configure_windows_unwind(job_manager_compute_progress_tests)
+    add_test(NAME guest_job_manager_compute_progress COMMAND job_manager_compute_progress_tests)
+    set_tests_properties(guest_job_manager_compute_progress PROPERTIES TIMEOUT 20)
+    if(WIN32)
+        set_tests_properties(guest_job_manager_compute_progress PROPERTIES ENVIRONMENT_MODIFICATION
+            "PATH=path_list_prepend:$<TARGET_FILE_DIR:libc>")
+    endif()
     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 libSceAgc libSceAgcDriver libkernel libc)

core/libs/prx/libSceJobManager/ComputeProgress.hpp

@@ -0,0 +1,31 @@
+#ifndef CORE_LIBS_PRX_LIBSCEJOBMANAGER_COMPUTEPROGRESS_HPP
+#define CORE_LIBS_PRX_LIBSCEJOBMANAGER_COMPUTEPROGRESS_HPP
+
+#include <cstdint>
+#include <stdexcept>
+
+namespace JobManager {
+
+class ComputeProgress {
+public:
+    std::uint32_t Publish(std::uint32_t count) {
+        total += count;
+        return total;
+    }
+
+    std::uint32_t Collect(std::uint64_t value) {
+        const auto current = static_cast<std::uint32_t>(value);
+        const std::uint32_t finished = current - seen;
+        if (finished > std::uint32_t{total - seen}) throw std::runtime_error("libSceJobManager: a compute label advanced beyond its published work");
+        seen = current;
+        return finished;
+    }
+
+private:
+    std::uint32_t total = 0;
+    std::uint32_t seen = 0;
+};
+
+}
+
+#endif

core/libs/prx/libSceJobManager/Export.cpp

@@ -13,7 +13,9 @@
 #include <vector>
 
 #include "SceTypes.hpp"
+#include "ComputeProgress.hpp"
 #include "prx/libc/include/General.hpp"
+#include "prx/libc/include/GuestArena.hpp"
 
 extern "C" {
 int APS5_VABI scePthreadAttrInit(PthreadAttr* attr);
@@ -43,6 +45,9 @@ 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);
+std::uint32_t* APS5_VABI sceAgcAcbWaitRegMem(CommandBuffer* buf, std::uint8_t size, std::uint8_t compareFunction, std::uint8_t cachePolicy, const volatile void* address, std::uint64_t reference, std::uint64_t mask, std::uint32_t pollCycles);
+std::uint32_t* APS5_VABI sceAgcAcbRewind(CommandBuffer* buf, std::uint32_t initialState);
+std::uint32_t* APS5_VABI sceAgcAcbWriteData(CommandBuffer* buf, std::uint8_t dst, std::uint8_t cachePolicy, std::uint64_t address, const void* data, std::uint32_t numDwords, std::uint8_t increment, std::uint8_t writeConfirm);
 }
 
 namespace {
@@ -72,7 +77,6 @@ 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;
@@ -117,6 +121,32 @@ std::uint64_t AlignUp(std::uint64_t value, std::uint64_t alignment) {
     return (value + alignment - 1) & ~(alignment - 1);
 }
 
+std::uint64_t ComputeCommandBytes() {
+    std::array<std::uint32_t, 256> words{};
+    CommandBuffer buffer{words.data(), words.data() + words.size(), words.data(), words.data() + words.size(), nullptr, nullptr, 0};
+    const ShaderRegister shaderRegister{};
+    const std::array<std::uint32_t, 2> userData{};
+    bool encoded = sceAgcAcbAcquireMem(&buffer, 0x20380, nullptr, 0, 0x190) != nullptr;
+    encoded = encoded && sceAgcCbSetShRegistersDirect(&buffer, &shaderRegister, 1) != nullptr;
+    encoded = encoded && sceAgcCbSetShRegisterRangeDirect(&buffer, UserDataRegister, userData.data(), 2) != nullptr;
+    encoded = encoded && sceAgcAcbPushMarker(&buffer, "1234567890123456789012345678901", 0) != nullptr;
+    encoded = encoded && sceAgcCbDispatch(&buffer, 1, 1, 1, 0) != nullptr;
+    encoded = encoded && sceAgcAcbPopMarker(&buffer) != nullptr;
+    const auto dispatchBytes = static_cast<std::uint64_t>(buffer.cursor_up - words.data()) * 4;
+    buffer.cursor_up = words.data();
+    encoded = encoded && sceAgcCbReleaseMem(&buffer, 0x2f, 0, 1, 3, nullptr, 1, 0, 0, 0, 0, 0) != nullptr;
+    encoded = encoded && sceAgcAcbWaitRegMem(&buffer, 0, 3, 3, nullptr, 0, ~std::uint64_t{0}, 0x190) != nullptr;
+    encoded = encoded && sceAgcAcbRewind(&buffer, 1) != nullptr;
+    encoded = encoded && sceAgcCbReleaseMem(&buffer, 0x2f, 0x200, 1, 3, nullptr, 1, 0, 0, 0, 0, 0) != nullptr;
+    encoded = encoded && sceAgcAcbWriteData(&buffer, 2, 3, 0, userData.data(), 1, 1, 0) != nullptr;
+    const auto syncBytes = static_cast<std::uint64_t>(buffer.cursor_up - words.data()) * 4;
+    buffer.cursor_up = words.data();
+    encoded = encoded && sceAgcCbReleaseMem(&buffer, 0x2f, 0x200, 1, 3, nullptr, 1, 0, 0, 0, 2, 0) != nullptr;
+    const auto waitBytes = static_cast<std::uint64_t>(buffer.cursor_up - words.data()) * 4;
+    if (!encoded || dispatchBytes > 255 || syncBytes > 255 || waitBytes > 255) throw std::runtime_error("libSceJobManager: compute command size measurement failed");
+    return MaxDispatches * dispatchBytes + MaxSyncPoints * syncBytes + MaxWaits * waitBytes;
+}
+
 struct Manager;
 class Sequence;
 
@@ -179,6 +209,7 @@ public:
     void Advance();
     void DispatchCompleted(std::uint32_t invocations);
     void CollectCompute();
+    void SetComputeStorage(std::uint32_t* commands, std::size_t words, Label* completion);
     bool Busy() const { return outstanding != 0; }
 
     Manager* manager = nullptr;
@@ -192,6 +223,7 @@ private:
     int WaitLocked(std::unique_lock<std::mutex>& guard);
     int AppendCompute(Item item);
     void SubmitCompute(const Item& item);
+    void FlushCompute();
 
     Counters limits;
     Counters submitted;
@@ -202,11 +234,13 @@ private:
     std::size_t cursor = 0;
     std::deque<Item> items;
     std::condition_variable passed;
-    Label label{};
-    std::uint64_t labelSeen = 0;
-    std::uint64_t labelTotal = 0;
+    Label* label = nullptr;
+    JobManager::ComputeProgress progress;
+    std::uint32_t pendingCompute = 0;
     bool batchStarted = false;
-    std::deque<std::vector<std::uint32_t>> commands;
+    std::uint32_t* commandBegin = nullptr;
+    std::size_t commandWords = 0;
+    std::size_t commandUsed = 0;
 };
 
 class Factory final : public ISequenceFactory {
@@ -282,7 +316,7 @@ bool Sequence::Open(Manager* owner, const Counters& limit, std::uint32_t runQueu
     outstanding = 0;
     cursor = 0;
     items.clear();
-    commands.clear();
+    commandUsed = 0;
     batchStarted = false;
     if (!TakeNode(*owner)) return false;
     open = true;
@@ -310,6 +344,7 @@ void Sequence::Advance() {
             SubmitCompute(item);
             continue;
         }
+        FlushCompute();
         if (outstanding != 0) return;
         if (item.type == ItemType::SyncPoint) {
             ++completed.syncPoints;
@@ -320,7 +355,7 @@ void Sequence::Advance() {
         const auto retired = static_cast<std::uint32_t>(cursor);
         items.erase(items.begin(), items.begin() + static_cast<std::ptrdiff_t>(retired + 1));
         cursor = 0;
-        commands.clear();
+        commandUsed = 0;
         batchStarted = false;
         if (error == 0) {
             submitted.invocations -= completed.invocations;
@@ -370,31 +405,57 @@ void Sequence::SubmitCompute(const Item& item) {
     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) {
+    const auto capacity = std::min(commandWords - commandUsed, 64 + 3 * registerCount + item.name.size() / 4);
+    auto* const words = commandBegin + commandUsed;
+    CommandBuffer buffer{words, words + capacity, words, words + capacity, nullptr, nullptr, 0};
+    {
+        const GuestArena::HostWrite destination(words, capacity * 4);
+        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 (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;
+        if (!encoded) throw std::runtime_error("libSceJobManager: a compute dispatch did not fit its command buffer");
     }
-    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, {}};
+    ++pendingCompute;
+    const Packet packet{words, static_cast<std::uint32_t>(buffer.cursor_up - words), 0, {}};
+    commandUsed += packet.dw_num;
     if (sceAgcDriverSubmitAcb(queue, &packet) != 0) throw std::runtime_error("libSceJobManager: sceAgcDriverSubmitAcb failed");
 }
 
+void Sequence::SetComputeStorage(std::uint32_t* commands, std::size_t words, Label* completion) {
+    commandBegin = commands;
+    commandWords = words;
+    label = completion;
+    const GuestArena::HostWrite destination(label, sizeof(*label));
+    label->value = 0;
+}
+
+void Sequence::FlushCompute() {
+    if (pendingCompute == 0) return;
+    if (commandWords - commandUsed < 8) throw std::runtime_error("libSceJobManager: a compute completion did not fit its command buffer");
+    auto* const words = commandBegin + commandUsed;
+    CommandBuffer buffer{words, words + 8, words, words + 8, nullptr, nullptr, 0};
+    {
+        const GuestArena::HostWrite destination(words, 32);
+        if (sceAgcCbReleaseMem(&buffer, 0x2f, 0x200, 1, 3, label, 1, progress.Publish(pendingCompute), 0, 0, 2, 0) == nullptr) {
+            throw std::runtime_error("libSceJobManager: a compute completion could not be encoded");
+        }
+    }
+    pendingCompute = 0;
+    const Packet packet{words, static_cast<std::uint32_t>(buffer.cursor_up - words), 0, {}};
+    commandUsed += packet.dw_num;
+    if (sceAgcDriverSubmitAcb(queue, &packet) != 0) throw std::runtime_error("libSceJobManager: sceAgcDriverSubmitAcb failed for a completion");
+}
+
 void Sequence::CollectCompute() {
-    const std::uint64_t value = label.value;
-    const auto finished = static_cast<std::uint32_t>(value - labelSeen);
-    labelSeen = value;
+    const auto finished = progress.Collect(label->value);
     if (finished == 0) return;
     completed.dispatches += finished;
     outstanding -= finished;
@@ -457,7 +518,7 @@ int Sequence::WaitAndClose(std::uint32_t index) {
     const auto remaining = static_cast<std::uint32_t>(std::count_if(items.begin(), items.end(), [](const Item& item) { return item.type != ItemType::Wait; }));
     ReturnNodes(*manager, remaining + 1);
     items.clear();
-    commands.clear();
+    commandUsed = 0;
     cursor = 0;
     outstanding = 0;
     open = false;
@@ -592,18 +653,28 @@ 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, std::uint32_t computeQueues, bool* fits) {
+struct WorkLayout {
+    std::uint64_t commands = 0;
+    std::uint64_t commandBytes = 0;
+    std::uint64_t labels = 0;
+    std::uint32_t nodes = 0;
+};
+
+WorkLayout LayoutFor(std::uint64_t memory, std::uint64_t size, std::uint32_t workers, std::uint32_t sequences, std::uint32_t computeQueues) {
+    WorkLayout layout;
     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(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};
+        layout.commandBytes = ComputeCommandBytes();
+        layout.commands = AlignUp(address + ComputeContextBytes, 0x100);
+        layout.labels = AlignUp(layout.commands + layout.commandBytes * computeQueues + 4 * std::uint64_t{computeQueues}, 8);
+        address = layout.labels + 8 * std::uint64_t{computeQueues};
     }
     address = AlignUp(address, 64);
-    *fits = address <= end;
-    return *fits ? static_cast<std::uint32_t>(((end - address) / NodeBytes) & 0xffff) : 0;
+    if (address <= end) layout.nodes = static_cast<std::uint32_t>(((end - address) / NodeBytes) & 0xffff);
+    return layout;
 }
 
 int ValidateWorkers(const ConfigParams& config, std::uint32_t workers) {
@@ -638,7 +709,7 @@ 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) {
+void StartCompute(Manager& manager, const WorkLayout& layout, 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));
@@ -649,6 +720,8 @@ void StartCompute(Manager& manager, std::uint32_t computeSequences, std::uint64_
     for (std::uint32_t index = 0; index < computeSequences; ++index) {
         manager.computeSequences[index].compute = true;
         manager.computeSequences[index].queue = manager.computeQueues[index];
+        manager.computeSequences[index].SetComputeStorage(reinterpret_cast<std::uint32_t*>(layout.commands + index * layout.commandBytes),
+            static_cast<std::size_t>(layout.commandBytes / 4), reinterpret_cast<Label*>(layout.labels + index * sizeof(Label)));
     }
     manager.freeComputeSequences = computeSequences == MaxSequences ? ~std::uint32_t{0} : (std::uint32_t{1} << computeSequences) - 1;
     const KernelSchedParam schedule{static_cast<int>(priority)};
@@ -675,15 +748,17 @@ int Initialize(void* self, void* memory, std::uint64_t size, std::uint32_t worke
     }
     if (size < StateOffset) return ManagerMemoryTooSmall;
     if (size < StateEnd) throw std::runtime_error("sce::Job::JobManager::initialize: a memory block smaller than 0x2d8 bytes is not supported");
-    std::memset(static_cast<std::uint8_t*>(memory) + StateOffset, 0, StateEnd - StateOffset);
-    const BlockHeader header{0x4d4a, static_cast<std::uint16_t>(StateOffset), 0x05000001, static_cast<std::uint32_t>(size), 0};
-    std::memcpy(memory, &header, sizeof(header));
+    {
+        const GuestArena::HostWrite destination(memory, StateEnd);
+        std::memset(static_cast<std::uint8_t*>(memory) + StateOffset, 0, StateEnd - StateOffset);
+        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;
     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, computeQueues, &fits);
-    if (!fits || nodes == 0) return ManagerMemoryTooSmall;
+    const auto layout = LayoutFor(base, size, workers, sequences, computeQueues);
+    if (layout.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;
@@ -693,7 +768,7 @@ int Initialize(void* self, void* memory, std::uint64_t size, std::uint32_t worke
     manager->running = true;
     manager->workerCount = workers;
     manager->spinTicks = sceKernelGetTscFrequency() / 1000000 * settings.spinMicroseconds;
-    manager->freeNodes = nodes - 1;
+    manager->freeNodes = layout.nodes - 1;
     manager->freeSequences = sequences == MaxSequences ? ~std::uint32_t{0} : (std::uint32_t{1} << sequences) - 1;
     manager->sequences = std::vector<Sequence>(sequences);
     for (std::uint32_t priority = 0; priority < 2; ++priority) {
@@ -702,7 +777,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);
+    if (computeSequences != 0) StartCompute(*manager, layout, computeSequences, computeQueueMask, handlerPriority, handlerAffinity);
     Implementation(self) = manager;
     StartWorkers(*manager, settings, workers);
     return 0;
@@ -759,7 +834,7 @@ std::uint64_t APS5_VABI _ZN3sce3Job10JobManager27calculateRequiredMemorySizeEPKN
     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 + ComputeCommandBytes() * queues + 3) & ~std::uint64_t{3};
         bytes = (bytes + 4 * queues + 7) & ~std::uint64_t{7};
         bytes += 8 * queues;
     } else {

core/libs/tests/JobManagerComputeProgress.cpp

@@ -0,0 +1,27 @@
+#include "prx/libSceJobManager/ComputeProgress.hpp"
+#include <cstdint>
+#include <iostream>
+#include <stdexcept>
+
+int main() {
+    try {
+        JobManager::ComputeProgress progress;
+        if (progress.Publish(0xfffffffd) != 0xfffffffd || progress.Collect(0xfffffffc) != 0xfffffffc)
+            throw std::runtime_error("partial retirement before wrap failed");
+        if (progress.Publish(5) != 2 || progress.Collect(2) != 6 || progress.Collect(2) != 0)
+            throw std::runtime_error("batched completion across wrap failed");
+        if (progress.Publish(9) != 11 || progress.Collect(5) != 3 || progress.Collect(11) != 6)
+            throw std::runtime_error("partial retirement after wrap failed");
+        bool rejected = false;
+        try {
+            progress.Collect(12);
+        } catch (const std::runtime_error&) {
+            rejected = true;
+        }
+        if (!rejected || progress.Collect(11) != 0) throw std::runtime_error("an invalid completion changed retirement state");
+        return 0;
+    } catch (const std::exception& error) {
+        std::cerr << error.what() << '\n';
+        return 1;
+    }
+}

docs/dev/debt/libSceJobManager.md

@@ -2,8 +2,8 @@
 
 ### Functional
 
-- [sce::Job::JobManager](../../../core/libs/prx/libSceJobManager/Export.cpp) submits every compute dispatch with its own label write and interrupt, while the original batches label writes and rewinds commands encoded ahead. Command buffers and labels live on the host heap; labels outside a GPU import force the driver to drain recorded work before each write. Completion labels are written as 32 bits; wrap behavior is unverified. The per-queue command-size formula uses fixed builder sizes of 104, 148 and 32 bytes per dispatch, sync point and wait.
-- The [manager](../../../core/libs/prx/libSceJobManager/Export.cpp) keeps its state on the host heap rather than in the caller's block, apart from the block header. Blocks smaller than 0x2d8 bytes throw. Worker parking uses a host condition variable and mutex rather than the original semaphore and atomic queues; console scheduling and spin timing have not been measured. The fallback uses raw host TSC for its spin interval, so APS5_TIME_SCALE does not change host CPU use; the original module reads the scaled guest TSC when run through the host kernel.
+- [sce::Job::JobManager](../../../core/libs/prx/libSceJobManager/Export.cpp) submits commands as work becomes runnable rather than encoding ahead with the rewind protocol recovered from PPSA06888's module (offsets 0x2210 and 0x2650). Commands are recycled after a completed wait; exhausting the bounded command area throws. Native recovery from command exhaustion and console timing equivalence remain unverified.
+- The [manager](../../../core/libs/prx/libSceJobManager/Export.cpp) keeps its state on the host heap rather than in the caller's block, apart from the block header, compute commands and completion labels. Blocks smaller than 0x2d8 bytes throw. Worker parking uses a host condition variable and mutex rather than the original semaphore and atomic queues; console scheduling and spin timing have not been measured. The fallback uses raw host TSC for its spin interval, so APS5_TIME_SCALE does not change host CPU use; the original module reads the scaled guest TSC when run through the host kernel.
 - [Sequence and manager operations](../../../core/libs/prx/libSceJobManager/Export.cpp) throw on a compute dispatch on a CPU sequence, a failing `sceAgcDriverAddEqEvent`, a factory priority above 1, and use of a destroyed sequence. `shutdown` throws while compute work is in flight or a caller is blocked in a sequence wait. Worker creation and attribute failures throw where the original does not check.
 - [`initialize`](../../../core/libs/prx/libSceJobManager/Export.cpp) returns 0 instead of the result of `_sceUlobjmgrRegisterObject`. The `sceRazorCpuJobManager*` profiler calls are not made. Sequence and factory virtual-function order is recovered from module vtables; no calling title has been run.
 - The [fallback](../../../core/libs/prx/libSceJobManager/Export.cpp) and Windows [reference runner](../../../tools/run_job_manager_reference.py) disagree: the runner times out in the node-pool exhaustion fixture against the PPSA06888 module with the host kernel; the fallback returns `SequenceInvalid` and recycles its nodes. The original Wait spins while allocating a node from the exhausted global pool and resumes when another sequence returns one; the fallback deliberately needs no Wait node. Original-module execution on Linux and console timing equivalence are not covered by this runner.