From fd459d96f795c8cefd90a035bf9db483fffb8ea6 Mon Sep 17 00:00:00 2001 From: Felipebmaitan Date: Fri, 23 Oct 2026 08:33:11 -0300 Subject: [PATCH] fix(libSceJobManager): honor worker spin and persistence policies --- core/libs/prx/libSceJobManager/CMakeLists.txt | 13 +- core/libs/prx/libSceJobManager/Export.cpp | 44 ++++- core/libs/tests/GuestJobManager.cpp | 150 +++++++++++++++++- core/libs/tests/JobManagerReference.cpp | 79 +++++++++ docs/dev/ENVIRONMENT.md | 2 + docs/dev/debt/libSceJobManager.md | 6 +- tools/test_job_manager_reference.py | 135 ++++++++++++++++ 7 files changed, 421 insertions(+), 8 deletions(-) create mode 100644 core/libs/tests/JobManagerReference.cpp create mode 100644 tools/test_job_manager_reference.py diff --git a/core/libs/prx/libSceJobManager/CMakeLists.txt b/core/libs/prx/libSceJobManager/CMakeLists.txt index 8539bbd0c2..fd14b28c4c 100644 --- a/core/libs/prx/libSceJobManager/CMakeLists.txt +++ b/core/libs/prx/libSceJobManager/CMakeLists.txt @@ -8,7 +8,18 @@ if(BUILD_TESTING) target_include_directories(guest_job_manager_tests PRIVATE ${LIBS_INCLUDE_DIR}) 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) + if(WIN32) + add_library(job_manager_reference SHARED EXCLUDE_FROM_ALL + ${LIBS_INCLUDE_DIR}/tests/GuestJobManager.cpp + ${LIBS_INCLUDE_DIR}/tests/JobManagerReference.cpp) + target_include_directories(job_manager_reference PRIVATE ${LIBS_INCLUDE_DIR}) + target_compile_definitions(job_manager_reference PRIVATE ANYPS5_JOB_MANAGER_REFERENCE) + target_link_libraries(job_manager_reference PRIVATE libkernel libc) + set_target_properties(job_manager_reference PROPERTIES PREFIX "" SUFFIX .prx + RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/tests) + configure_windows_unwind(job_manager_reference) + endif() + foreach(testCase IN ITEMS memory validation dispatch ordering nested limits node_pool priority shutdown worker_policy worker_wake worker_count) add_test(NAME guest_job_manager_${testCase} COMMAND guest_job_manager_tests ${testCase}) set_tests_properties(guest_job_manager_${testCase} PROPERTIES TIMEOUT 20) if(WIN32) diff --git a/core/libs/prx/libSceJobManager/Export.cpp b/core/libs/prx/libSceJobManager/Export.cpp index 155a95220d..18a578a6c7 100644 --- a/core/libs/prx/libSceJobManager/Export.cpp +++ b/core/libs/prx/libSceJobManager/Export.cpp @@ -6,6 +6,7 @@ #include #include #include +#include #include #include #include @@ -23,6 +24,8 @@ 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); +std::uint64_t APS5_VABI sceKernelReadTsc(); +std::uint64_t APS5_VABI sceKernelGetTscFrequency(); 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); @@ -226,6 +229,11 @@ struct Manager { std::uint32_t freeNodes = 0; std::uint32_t freeSequences = 0; std::uint32_t persistentThreads = 0; + std::uint32_t workerCount = 0; + std::uint32_t sleepingWorkers = 0; + std::uint32_t wakeTokens = 0; + std::uint64_t outstandingInvocations = 0; + std::uint64_t spinTicks = 0; std::array, 2> runQueues; std::vector sequences; std::deque computeSequences; @@ -238,6 +246,14 @@ struct Manager { std::vector workers; }; +thread_local Manager* currentWorker = nullptr; + +void WakeWorkers(Manager& manager, std::uint32_t invocations) { + const std::uint32_t available = invocations - (currentWorker == &manager ? 1u : 0u); + manager.wakeTokens += std::min(available, manager.sleepingWorkers - manager.wakeTokens); + if (manager.wakeTokens != 0) manager.work.notify_all(); +} + Manager*& Implementation(void* self) { return *static_cast(self); } @@ -285,7 +301,7 @@ void Sequence::Advance() { ++outstanding; ++cursor; manager->runQueues[priority].push_back(&item); - manager->work.notify_all(); + WakeWorkers(*manager, item.count); continue; } if (item.type == ItemType::Compute || item.type == ItemType::ComputeIndirect) { @@ -320,6 +336,7 @@ void Sequence::Advance() { } void Sequence::DispatchCompleted(std::uint32_t invocations) { + manager->outstandingInvocations -= invocations; completed.invocations += invocations; ++completed.dispatches; --outstanding; @@ -338,6 +355,7 @@ int Sequence::Dispatch(JobEntry entry, void* argument, std::uint32_t count, cons if (before.invocations + count > limits.invocations || before.dispatches >= limits.dispatches) return Latch(SequenceLimitExceeded); if (!TakeNode(*manager)) return Latch(SequenceNoNodes); items.push_back({ItemType::Dispatch, this, entry, argument, count, 0, 0, nullptr}); + manager->outstandingInvocations += count; Advance(); return 0; } @@ -551,11 +569,27 @@ bool RunOne(Manager& manager, std::unique_lock& guard) { void* APS5_VABI WorkerEntry(void* argument) { Manager& manager = *static_cast(argument); + currentWorker = &manager; + std::uint64_t lastWork = sceKernelReadTsc(); std::unique_lock guard(manager.lock); while (manager.running) { - if (RunOne(manager, guard)) continue; - manager.work.wait(guard, [&manager] { return !manager.running || !manager.runQueues[0].empty() || !manager.runQueues[1].empty(); }); + if (RunOne(manager, guard)) { + lastWork = sceKernelReadTsc(); + continue; + } + const bool spinning = manager.outstandingInvocations != 0 && sceKernelReadTsc() - lastWork < manager.spinTicks; + if (spinning || manager.sleepingWorkers >= manager.workerCount - manager.persistentThreads) { + guard.unlock(); + _mm_pause(); + guard.lock(); + continue; + } + ++manager.sleepingWorkers; + manager.work.wait(guard, [&manager] { return !manager.running || manager.wakeTokens != 0; }); + if (manager.wakeTokens != 0) --manager.wakeTokens; + --manager.sleepingWorkers; } + currentWorker = nullptr; return nullptr; } @@ -658,6 +692,8 @@ int Initialize(void* self, void* memory, std::uint64_t size, std::uint32_t worke if (computeSequences != 0 && (handlerPriority < 0x100 || handlerPriority > 0x2ff || handlerAffinity == 0 || handlerAffinity > 0x1fff)) return ManagerInvalidArgument; auto* manager = new Manager; manager->running = true; + manager->workerCount = workers; + manager->spinTicks = sceKernelGetTscFrequency() / 1000000 * settings.spinMicroseconds; manager->freeNodes = nodes - 1; manager->freeSequences = sequences == MaxSequences ? ~std::uint32_t{0} : (std::uint32_t{1} << sequences) - 1; manager->sequences = std::vector(sequences); @@ -781,7 +817,7 @@ ISequenceFactory* APS5_VABI _ZN3sce3Job10JobManager34getUnifiedSequenceFactoryIn int APS5_VABI _ZN3sce3Job10JobManager24setPersistentThreadCountEj(void* self, std::uint32_t count) { Manager* manager = Implementation(self); if (manager == nullptr) return ManagerNotInitialized; - if (count > manager->workers.size()) return ManagerInvalidArgument; + if (count > manager->workerCount) return ManagerInvalidArgument; const std::lock_guard guard(manager->lock); manager->persistentThreads = count; return 0; diff --git a/core/libs/tests/GuestJobManager.cpp b/core/libs/tests/GuestJobManager.cpp index 7258ae6694..9616edabf4 100644 --- a/core/libs/tests/GuestJobManager.cpp +++ b/core/libs/tests/GuestJobManager.cpp @@ -1,4 +1,5 @@ #include "SceTypes.hpp" +#include #include #include #include @@ -54,6 +55,9 @@ struct SequenceFactory { }; extern "C" { +Pthread APS5_VABI scePthreadSelf(); +int APS5_VABI pthread_getcpuclockid_nid_postfix(Pthread thread, int* clockId); +int APS5_VABI sceKernelClockGettime(KernelClockid clockId, KernelTimespec* time); void APS5_VABI _ZN3sce3Job10JobManagerC1Ev(void*); void APS5_VABI _ZN3sce3Job10JobManagerD1Ev(void*); std::uint64_t APS5_VABI _ZN3sce3Job10JobManager27calculateRequiredMemorySizeEPKNS1_21MemorySizeQueryParamsE(const MemorySizeQueryParams*); @@ -348,6 +352,15 @@ static void LimitsCase() { Require(factory->Create(&sequence, 4, 2, 1, 1, "waits") == 0); Require(sequence->Wait() == 0 && sequence->Wait() == 0); Require(factory->Destroy(sequence) == 0); + Require(Shutdown(manager.storage) == 0); +} + +static void NodePoolCase() { + Block block(SizeFor(2, 2, 2, 1, 1)); + ManagerObject manager; + Require(Initialize(manager.storage, block.memory, block.bytes, 2, 2, nullptr) == 0); + SequenceFactory* factory = Factories(manager.storage, 0); + Sequence* sequence = nullptr; Require(factory->Create(&sequence, 0x7fff, 0x3ff, 0xff, 1, "nodes") == 0); int result = 0; int accepted = 0; @@ -453,16 +466,149 @@ static void ShutdownCase() { } } -int main(int argc, char** argv) { - const std::string name = argc > 1 ? argv[1] : ""; +static void APS5_VABI CaptureWorkerClock(void* argument, std::uint32_t) { + Require(pthread_getcpuclockid_nid_postfix(scePthreadSelf(), static_cast(argument)) == 0); +} + +static std::int64_t CpuNanos(int clock) { + KernelTimespec time{}; + Require(sceKernelClockGettime(clock, &time) == 0); + return time.tv_sec * 1000000000LL + time.tv_nsec; +} + +static void RequireParked(int clock) { + std::this_thread::sleep_for(std::chrono::milliseconds(30)); + const auto start = CpuNanos(clock); + std::this_thread::sleep_for(std::chrono::milliseconds(80)); + Require(CpuNanos(clock) - start < 20000000); +} + +static void WorkerPolicyCase() { + for (const std::uint32_t mode : {0x10000u, 0x20000u}) { + for (const std::uint32_t spin : {0u, 50u}) { + Block block(SizeFor(1, 1, 8, 2, 2)); + ManagerObject manager; + const ConfigParams config{0, mode | 0x1fff, 0x10000, 0x2bc, "WorkerPolicy", spin, 0, 0, 0}; + Require(Initialize(manager.storage, block.memory, block.bytes, 1, 1, &config) == 0); + auto* factory = Factories(manager.storage, 0); + Sequence* sequence = nullptr; + Require(factory->Create(&sequence, 8, 8, 2, 2, "worker policy") == 0); + int clock = 0; + Require(sequence->Dispatch(CaptureWorkerClock, &clock, 1, nullptr) == 0 && sequence->Wait() == 0); + RequireParked(clock); + for (int round = 0; round < 2; ++round) { + Require(SetPersistentThreadCount(manager.storage, 1) == 0); + RequireParked(clock); + Require(sequence->Dispatch(Nothing, nullptr, 1, nullptr) == 0 && sequence->Wait() == 0); + const auto start = CpuNanos(clock); + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2); + while (CpuNanos(clock) - start < 30000000 && std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + Require(CpuNanos(clock) - start >= 30000000); + Require(SetPersistentThreadCount(manager.storage, 0) == 0); + RequireParked(clock); + } + Require(SetPersistentThreadCount(manager.storage, 1) == 0); + Require(sequence->Dispatch(Nothing, nullptr, 1, nullptr) == 0 && sequence->Wait() == 0); + Require(Shutdown(manager.storage) == 0); + } + } +} + +static void WorkerWakeCase() { + Block block(SizeFor(4, 1, 8, 2, 2)); + ManagerObject manager; + for (const std::uint32_t mode : {0x10000u, 0x20000u}) { + const ConfigParams config{0, mode | 0x1fff, 0x10000, 0x2bc, "WorkerWake", 50, 0, 0, 0}; + Require(Initialize(manager.storage, block.memory, block.bytes, 4, 1, &config) == 0); + Sequence* sequence = nullptr; + auto* factory = Factories(manager.storage, 0); + Require(factory->Create(&sequence, 1024, 8, 2, 2, "worker wake") == 0); + for (std::uint32_t round = 0; round < 60; ++round) { + Require(SetPersistentThreadCount(manager.storage, round % 5) == 0); + Tally tally(257); + Require(sequence->Dispatch(CountJob, &tally, 257, nullptr) == 0); + Require(sequence->SyncPoint() == 0); + Require(sequence->Dispatch(CountJob, &tally, 257, nullptr) == 0); + Require(sequence->Wait() == 0 && tally.calls == 514); + for (const auto& count : tally.seen) Require(count == 2); + } + Require(factory->Destroy(sequence) == 0); + Require(Shutdown(manager.storage) == 0); + } +} + +struct WorkerClocks { + std::array clocks{}; + std::atomic arrived{0}; + std::atomic release{false}; +}; + +static void APS5_VABI CaptureWorkerClocks(void* argument, std::uint32_t index) { + auto& workers = *static_cast(argument); + CaptureWorkerClock(&workers.clocks[index], 0); + ++workers.arrived; + while (!workers.release) std::this_thread::yield(); +} + +static void WorkerCountCase() { + Block block(SizeFor(4, 1, 8, 2, 2)); + ManagerObject manager; + const ConfigParams config{0, 0x1000f, 0x10000, 0x2bc, "WorkerCount", 0, 0, 0, 0}; + Require(Initialize(manager.storage, block.memory, block.bytes, 4, 1, &config) == 0); + Sequence* sequence = nullptr; + Require(Factories(manager.storage, 0)->Create(&sequence, 8, 8, 2, 2, "worker count") == 0); + WorkerClocks workers; + Require(sequence->Dispatch(CaptureWorkerClocks, &workers, 4, nullptr) == 0); + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2); + while (workers.arrived != 4 && std::chrono::steady_clock::now() < deadline) std::this_thread::yield(); + Require(workers.arrived == 4); + workers.release = true; + Require(sequence->Wait() == 0); + for (const std::uint32_t persistent : {2u, 4u}) { + Require(SetPersistentThreadCount(manager.storage, 0) == 0); + for (const int clock : workers.clocks) RequireParked(clock); + Require(SetPersistentThreadCount(manager.storage, persistent) == 0); + for (std::uint32_t job = 0; job < persistent; ++job) { + Require(sequence->Dispatch(Nothing, nullptr, 1, nullptr) == 0 && sequence->Wait() == 0); + } + std::array starts{}; + for (std::size_t index = 0; index < starts.size(); ++index) starts[index] = CpuNanos(workers.clocks[index]); + const auto until = std::chrono::steady_clock::now() + std::chrono::seconds(2); + unsigned int active = 0; + do { + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + active = 0; + for (std::size_t index = 0; index < starts.size(); ++index) { + if (CpuNanos(workers.clocks[index]) - starts[index] >= 30000000) ++active; + } + } while (active < persistent && std::chrono::steady_clock::now() < until); + Require(active == persistent); + } + Require(Shutdown(manager.storage) == 0); +} + +int RunJobManagerCase(const char* caseName) { + const std::string name = caseName; if (name == "memory") MemoryCase(); else if (name == "validation") ValidationCase(); else if (name == "dispatch") DispatchCase(); else if (name == "ordering") OrderingCase(); else if (name == "limits") LimitsCase(); + else if (name == "node_pool") NodePoolCase(); else if (name == "nested") NestedCase(); else if (name == "priority") PriorityCase(); else if (name == "shutdown") ShutdownCase(); + else if (name == "worker_policy") WorkerPolicyCase(); + else if (name == "worker_wake") WorkerWakeCase(); + else if (name == "worker_count") WorkerCountCase(); else return 2; return 0; } + +#ifndef ANYPS5_JOB_MANAGER_REFERENCE +int main(int argc, char** argv) { + return RunJobManagerCase(argc > 1 ? argv[1] : ""); +} +#endif diff --git a/core/libs/tests/JobManagerReference.cpp b/core/libs/tests/JobManagerReference.cpp new file mode 100644 index 0000000000..b808c60f2f --- /dev/null +++ b/core/libs/tests/JobManagerReference.cpp @@ -0,0 +1,79 @@ +#include "SceTypes.hpp" +#include +#include +#include +#include + +struct MemorySizeQueryParams; +struct ConfigParams; +struct SequenceFactory; + +int RunJobManagerCase(const char* name); + +template +static T Resolve(const char* nid) { + const bool host = std::getenv("ANYPS5_JOB_MANAGER_HOST") != nullptr; + const char* library = host ? "libSceJobManager.prx" : "libSceJobManager.prx.guest.prx"; + HMODULE module = GetModuleHandleA(library); + if (module == nullptr && host) module = LoadLibraryA("libs\\libSceJobManager.prx"); + const FARPROC symbol = module != nullptr ? GetProcAddress(module, nid) : nullptr; + if (symbol == nullptr) { + std::fprintf(stderr, "JobManager reference: missing %s in %s (Windows error %lu)\n", nid, library, GetLastError()); + std::exit(3); + } + return reinterpret_cast(symbol); +} + +extern "C" { + +void APS5_VABI _ZN3sce3Job10JobManagerC1Ev(void* self) { + static const auto call = Resolve("cwMzzdesLlY"); + call(self); +} + +void APS5_VABI _ZN3sce3Job10JobManagerD1Ev(void* self) { + static const auto call = Resolve("4g1ee34TMH8"); + call(self); +} + +std::uint64_t APS5_VABI _ZN3sce3Job10JobManager27calculateRequiredMemorySizeEPKNS1_21MemorySizeQueryParamsE(const MemorySizeQueryParams* query) { + static const auto call = Resolve("Nm2BhAwYNFk"); + return call(query); +} + +int APS5_VABI _ZN3sce3Job10JobManager10initializeEPvmjjjmPKNS1_12ConfigParamsE(void* self, void* memory, std::uint64_t size, std::uint32_t workers, std::uint32_t sequences, std::uint32_t computeSequences, std::uint64_t mask, const ConfigParams* config) { + static const auto call = Resolve("GS9DSbzC+MY"); + return call(self, memory, size, workers, sequences, computeSequences, mask, config); +} + +int APS5_VABI _ZN3sce3Job10JobManager10initializeEPvmjjPKNS1_12ConfigParamsE(void* self, void* memory, std::uint64_t size, std::uint32_t workers, std::uint32_t sequences, const ConfigParams* config) { + static const auto call = Resolve("d205iUFMyyE"); + return call(self, memory, size, workers, sequences, config); +} + +int APS5_VABI _ZN3sce3Job10JobManager8shutdownEv(void* self) { + static const auto call = Resolve("chxnvsxxfBY"); + return call(self); +} + +SequenceFactory* APS5_VABI _ZN3sce3Job10JobManager27getSequenceFactoryInterfaceENS1_8PriorityE(void* self, std::uint32_t priority) { + static const auto call = Resolve("W-yXgQmaozU"); + return call(self, priority); +} + +SequenceFactory* APS5_VABI _ZN3sce3Job10JobManager34getUnifiedSequenceFactoryInterfaceENS1_8PriorityE(void* self, std::uint32_t priority) { + static const auto call = Resolve("L5jTAtk3HKA"); + return call(self, priority); +} + +int APS5_VABI _ZN3sce3Job10JobManager24setPersistentThreadCountEj(void* self, std::uint32_t count) { + static const auto call = Resolve("VTKezfYu-v4"); + return call(self, count); +} + +int APS5_VABI Run_nid_no_patch() { + const char* name = std::getenv("ANYPS5_JOB_MANAGER_CASE"); + return name != nullptr ? RunJobManagerCase(name) : 2; +} + +} diff --git a/docs/dev/ENVIRONMENT.md b/docs/dev/ENVIRONMENT.md index ac824ed4c4..4b03f49404 100644 --- a/docs/dev/ENVIRONMENT.md +++ b/docs/dev/ENVIRONMENT.md @@ -6,6 +6,8 @@ The code reads environment names through `getenv`, `IsEnvironmentSet`, `kibSetti | Name | Kind | Source | Default | Description | |------|------|--------|---------|-------------| +| `ANYPS5_JOB_MANAGER_CASE` | string | `core/libs/tests/JobManagerReference.cpp` | | selects the CPU fixture run by `tools/test_job_manager_reference.py` | +| `ANYPS5_JOB_MANAGER_HOST` | presence | `core/libs/tests/JobManagerReference.cpp` | | runs the reference adapter against the host fallback instead of the loaded guest module | | `ANYPS5_LOADEXEC_PARENT` | string | `core/libs/tests/GuestLoadExecHost.cpp` | | variable the Windows entry stub reads; its wide name is built in `core/relinker/elfpatcher/src/windows/WindowsEntryStubBuilder.cpp` | | `ANYPS5_LOADEXEC_TEST_ROOT` | string | `core/libs/tests/GuestLoadExecHost.cpp` | | directory the load-exec host test reads its guest executable from | | `ANYPS5_TEST_QUOTED_ARGV0` | presence | `core/libs/tests/GuestLoadExecHost.cpp` | | makes the load-exec test's first argument one with quotes, a backslash and non-ASCII text | diff --git a/docs/dev/debt/libSceJobManager.md b/docs/dev/debt/libSceJobManager.md index 123977e935..b3286ee6dc 100644 --- a/docs/dev/debt/libSceJobManager.md +++ b/docs/dev/debt/libSceJobManager.md @@ -2,4 +2,8 @@ ### 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. 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. `shutdown` throws while a compute sequence has work in flight, a sequence's label total is written as 32 bits, and since the label is host memory no GPU import covers, the driver drains recorded work before each label write. 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) 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. +- [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 the original's final import (`BG26hBGiNlw`, unresolved). 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/test_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 cause remains unverified. Original-module execution on Linux and console timing equivalence are not covered by this runner. diff --git a/tools/test_job_manager_reference.py b/tools/test_job_manager_reference.py new file mode 100644 index 0000000000..d766bd51bc --- /dev/null +++ b/tools/test_job_manager_reference.py @@ -0,0 +1,135 @@ +import argparse +import hashlib +import json +import os +import shutil +import subprocess +import sys +import tempfile +from pathlib import Path + +from import_audit import read_host_imports +from unwrap_self import unwrap + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT / "core/relinker/relinker/tests")) +from guest_import_fixtures import executable + +CASES = ("memory", "validation", "dispatch", "ordering", "nested", "limits", "node_pool", "priority", "shutdown", "worker_policy", "worker_wake", "worker_count") + + +def stage_libraries(directory, build, toolchain, dependencies): + providers = {path.name.lower(): path for path in (build / "core/libs/libs").glob("*.prx")} + providers.update({path.name.lower(): path for path in toolchain.glob("*.dll")}) + pending = list(dependencies) + staged = set() + while pending: + name = pending.pop().lower() + if name in staged: + continue + staged.add(name) + source = providers.get(name) + if source is None: + if name.endswith(".prx"): + raise RuntimeError(f"Missing patched runtime {name}; build core/libs/libs/{name}") + continue + shutil.copy2(source, directory / source.name) + pending.extend(read_host_imports(source)) + + +def run(args, directory): + libraries = directory / "libs" + libraries.mkdir() + helper = args.build / "tests/job_manager_reference.prx" + if not helper.is_file(): + raise RuntimeError("Build the job_manager_reference target first") + shutil.copy2(helper, libraries / "jobref.prx") + subprocess.run([str(args.build / "core/libs/nid_patcher.exe"), "jobref", "--preserve-export-list", + str(args.build / "core/libs/libc.exports"), str(libraries / "jobref.prx")], check=True, capture_output=True) + module_hash = None + if args.module: + module_data = args.module.read_bytes() + module_hash = hashlib.sha256(module_data).hexdigest() + modules = directory / "sce_module" + modules.mkdir() + target = modules / "libSceJobManager.prx" + if module_data[:4] == b"\x7fELF": + shutil.copy2(args.module, target) + else: + unwrap(args.module, target) + source = directory / "input.elf" + source.write_bytes(executable("jobref.prx", symbol="Run_nid_no_patch")) + output = directory / "reference.exe" + options = ["--skip-sce-module"] if args.host else [] + result = subprocess.run([str(args.relinker), "--windows", "--registry", *options, str(source), str(output)], + cwd=directory, capture_output=True, timeout=60) + (directory / "relink.log").write_bytes(result.stdout + result.stderr) + if result.returncode: + raise RuntimeError(f"Relinker failed ({result.returncode}); see {directory / 'relink.log'}") + dependencies = set(read_host_imports(helper)) | {"libc.prx"} + for registry in directory.glob("*.registry.json"): + dependencies.update(entry["library"] for entry in json.loads(registry.read_text()) + if entry.get("library") not in ("jobref.prx", "libSceJobManager.prx")) + if args.host: + dependencies.add("libSceJobManager.prx") + stage_libraries(libraries, args.build, args.toolchain, dependencies) + environment = os.environ.copy() + environment["PATH"] = str(libraries) + os.pathsep + environment.get("PATH", "") + environment.pop("ANYPS5_JOB_MANAGER_HOST", None) + if args.host: + environment["ANYPS5_JOB_MANAGER_HOST"] = "1" + results = [] + for case in args.cases: + environment["ANYPS5_JOB_MANAGER_CASE"] = case + try: + result = subprocess.run([str(output)], cwd=directory, env=environment, + capture_output=True, timeout=args.timeout) + log = result.stdout + result.stderr + status = "passed" if result.returncode == 0 else "failed" + code = result.returncode + except subprocess.TimeoutExpired as error: + log = (error.stdout or b"") + (error.stderr or b"") + status, code = "timeout", None + (directory / f"{case}.log").write_bytes(log) + results.append({"case": case, "status": status, "returncode": code}) + print(json.dumps(results[-1]), flush=True) + summary = {"provider": "host" if args.host else "guest", "module_sha256": module_hash, "cases": results} + (directory / "results.json").write_text(json.dumps(summary, indent=2) + "\n") + return 0 if all(result["status"] == "passed" for result in results) else 1 + + +def main(): + parser = argparse.ArgumentParser(description="Run the CPU JobManager fixtures through Windows relinker startup using an owned module or the host fallback.") + provider = parser.add_mutually_exclusive_group(required=True) + provider.add_argument("--module", type=Path, help="Owned decrypted SELF or ELF libSceJobManager.prx") + provider.add_argument("--host", action="store_true", help="Validate the runner against the built host fallback") + parser.add_argument("--build", type=Path, required=True) + parser.add_argument("--relinker", type=Path, required=True) + parser.add_argument("--toolchain", type=Path, required=True, help="Directory containing the matching compiler runtime DLLs") + parser.add_argument("--output", type=Path, help="New directory for staged modules, per-case logs and results.json; defaults to a temporary directory") + parser.add_argument("--cases", nargs="+", choices=CASES, default=list(CASES)) + parser.add_argument("--timeout", type=float, default=20) + args = parser.parse_args() + if sys.platform != "win32": + parser.error("The reference adapter currently supports Windows only") + if args.timeout <= 0 or args.timeout > 30: + parser.error("--timeout must be greater than zero and at most 30 seconds") + try: + for name in ("build", "relinker", "toolchain", "module"): + value = getattr(args, name) + if value is not None: + setattr(args, name, value.resolve(strict=True)) + directory = args.output.resolve() if args.output else Path(tempfile.mkdtemp(prefix="anyps5-jobmanager-")) + if args.output: + directory.mkdir(parents=True, exist_ok=False) + print(f"Artifacts: {directory}", flush=True) + return run(args, directory) + except (OSError, RuntimeError, subprocess.SubprocessError) as error: + print(str(error), file=sys.stderr) + if isinstance(error, subprocess.CalledProcessError) and error.stderr: + print(error.stderr.decode(errors="replace"), file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main())