diff --git a/core/libs/prx/libSceAgcDriver/tests/execution/JobManagerCompute.cpp b/core/libs/prx/libSceAgcDriver/tests/execution/JobManagerCompute.cpp index c149aa41a9..d557f4e4c5 100644 --- a/core/libs/prx/libSceAgcDriver/tests/execution/JobManagerCompute.cpp +++ b/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 #include #include @@ -11,6 +12,7 @@ #include #include #include +#include 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 OutputDescriptor{}; alignas(32) std::array IndirectArguments{1, 1, 1, 0, 0, 0, 0, 0}; alignas(16) std::array Specials{}; -alignas(16) std::array SpecialsWithoutDispatch{}; +alignas(16) std::array SpecialsDefaultModifier{}; std::array Registers{}; alignas(16) std::array Shader{}; -alignas(16) std::array ShaderWithoutDispatch{}; +alignas(16) std::array ShaderDefaultModifier{}; std::uint32_t Low(const volatile void* pointer) { return static_cast(reinterpret_cast(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, ®isters, sizeof(registers)); std::memcpy(shader->data() + 0x28, &specials, sizeof(specials)); (*shader)[0x5c] = static_cast(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(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 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(memory); + const auto base = reinterpret_cast(memory); + std::vector 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(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(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 diff --git a/core/libs/prx/libSceJobManager/CMakeLists.txt b/core/libs/prx/libSceJobManager/CMakeLists.txt index 002ba69513..5e5a4083a9 100644 --- a/core/libs/prx/libSceJobManager/CMakeLists.txt +++ b/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:$") + 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) diff --git a/core/libs/prx/libSceJobManager/ComputeProgress.hpp b/core/libs/prx/libSceJobManager/ComputeProgress.hpp new file mode 100644 index 0000000000..aadd732245 --- /dev/null +++ b/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 +#include + +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(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 diff --git a/core/libs/prx/libSceJobManager/Export.cpp b/core/libs/prx/libSceJobManager/Export.cpp index 4e3f1395b3..ec5a45a6d2 100644 --- a/core/libs/prx/libSceJobManager/Export.cpp +++ b/core/libs/prx/libSceJobManager/Export.cpp @@ -13,7 +13,9 @@ #include #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 words{}; + CommandBuffer buffer{words.data(), words.data() + words.size(), words.data(), words.data() + words.size(), nullptr, nullptr, 0}; + const ShaderRegister shaderRegister{}; + const std::array 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(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(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(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& 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 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> 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(cursor); items.erase(items.begin(), items.begin() + static_cast(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(item.argument); const std::array userDataWords{static_cast(userData), static_cast(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(buffer.cursor_up - words.data()), 0, {}}; + ++pendingCompute; + const Packet packet{words, static_cast(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(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(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::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(((end - address) / NodeBytes) & 0xffff) : 0; + if (address <= end) layout.nodes = static_cast(((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::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(layout.commands + index * layout.commandBytes), + static_cast(layout.commandBytes / 4), reinterpret_cast(layout.labels + index * sizeof(Label))); } manager.freeComputeSequences = computeSequences == MaxSequences ? ~std::uint32_t{0} : (std::uint32_t{1} << computeSequences) - 1; const KernelSchedParam schedule{static_cast(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(memory) + StateOffset, 0, StateEnd - StateOffset); - const BlockHeader header{0x4d4a, static_cast(StateOffset), 0x05000001, static_cast(size), 0}; - std::memcpy(memory, &header, sizeof(header)); + { + const GuestArena::HostWrite destination(memory, StateEnd); + std::memset(static_cast(memory) + StateOffset, 0, StateEnd - StateOffset); + const BlockHeader header{0x4d4a, static_cast(StateOffset), 0x05000001, static_cast(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::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(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::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 { diff --git a/core/libs/tests/JobManagerComputeProgress.cpp b/core/libs/tests/JobManagerComputeProgress.cpp new file mode 100644 index 0000000000..fb0d90da1c --- /dev/null +++ b/core/libs/tests/JobManagerComputeProgress.cpp @@ -0,0 +1,27 @@ +#include "prx/libSceJobManager/ComputeProgress.hpp" +#include +#include +#include + +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; + } +} diff --git a/docs/dev/debt/libSceJobManager.md b/docs/dev/debt/libSceJobManager.md index 0ce05f3280..f5d33ff81a 100644 --- a/docs/dev/debt/libSceJobManager.md +++ b/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.