fix(libSceJobManager): count a wait before serialising it and close under the lock

core/libs/prx/libSceJobManager/CMakeLists.txt

@@ -7,7 +7,7 @@ if(BUILD_TESTING)
     target_include_directories(guest_job_manager_tests PRIVATE ${LIBS_INCLUDE_DIR})
     target_link_libraries(guest_job_manager_tests PRIVATE libSceJobManager libc)
     configure_windows_unwind(guest_job_manager_tests)
-    foreach(testCase IN ITEMS memory validation dispatch ordering limits priority shutdown)
+    foreach(testCase IN ITEMS memory validation dispatch ordering nested limits priority shutdown)
         add_test(NAME guest_job_manager_${testCase} COMMAND guest_job_manager_tests ${testCase})
         set_tests_properties(guest_job_manager_${testCase} PROPERTIES TIMEOUT 20)
         if(WIN32)

core/libs/prx/libSceJobManager/Export.cpp

@@ -114,28 +114,28 @@ struct Counters {
     std::uint32_t waits = 0;
 };
 
-class SequenceInterface {
+class ISequence {
 public:
     APS5_VABI virtual int Dispatch(JobEntry entry, void* argument, std::uint32_t count, const char* name) = 0;
     APS5_VABI virtual int SyncPoint() = 0;
     APS5_VABI virtual int Wait() = 0;
-    APS5_VABI virtual ~SequenceInterface() = default;
+    APS5_VABI virtual void CompleteDestructor() {}
+    APS5_VABI virtual void DeletingDestructor() { throw std::runtime_error("libSceJobManager: a sequence was deleted by its caller"); }
     APS5_VABI virtual int DispatchCompute(const void* shader, void* userData, std::int32_t x, std::int32_t y, std::int32_t z, const char* name) = 0;
     APS5_VABI virtual int DispatchComputeIndirect(const void* shader, void* userData, const void* indirectArguments, const char* name) = 0;
-    static void operator delete(void*) { throw std::runtime_error("libSceJobManager: a sequence was deleted by its caller"); }
 };
 
-class SequenceFactoryInterface {
+class ISequenceFactory {
 public:
-    APS5_VABI virtual int Create(SequenceInterface** out, std::uint32_t invocations, std::uint32_t dispatches, std::uint32_t syncPoints, std::uint32_t waits, const char* name) = 0;
-    APS5_VABI virtual int Destroy(SequenceInterface* sequence) = 0;
-    APS5_VABI virtual ~SequenceFactoryInterface() = default;
-    APS5_VABI virtual int CreateCompute(SequenceInterface** out, std::uint32_t invocations, std::uint32_t dispatches, std::uint32_t syncPoints, std::uint32_t waits, const char* name) = 0;
-    APS5_VABI virtual int DestroyCompute(SequenceInterface* sequence) = 0;
-    static void operator delete(void*) { throw std::runtime_error("libSceJobManager: a sequence factory was deleted by its caller"); }
+    APS5_VABI virtual int Create(ISequence** out, std::uint32_t invocations, std::uint32_t dispatches, std::uint32_t syncPoints, std::uint32_t waits, const char* name) = 0;
+    APS5_VABI virtual int Destroy(ISequence* sequence) = 0;
+    APS5_VABI virtual void CompleteDestructor() {}
+    APS5_VABI virtual void DeletingDestructor() { throw std::runtime_error("libSceJobManager: a sequence factory was deleted by its caller"); }
+    APS5_VABI virtual int CreateCompute(ISequence** out, std::uint32_t invocations, std::uint32_t dispatches, std::uint32_t syncPoints, std::uint32_t waits, const char* name) = 0;
+    APS5_VABI virtual int DestroyCompute(ISequence* sequence) = 0;
 };
 
-class Sequence final : public SequenceInterface {
+class Sequence final : public ISequence {
 public:
     APS5_VABI int Dispatch(JobEntry entry, void* argument, std::uint32_t count, const char* name) override;
     APS5_VABI int SyncPoint() override;
@@ -150,7 +150,7 @@ public:
     }
 
     bool Open(Manager* owner, const Counters& limit, std::uint32_t runQueue);
-    void Close();
+    int WaitAndClose(std::uint32_t index);
     void Advance();
     void DispatchCompleted(std::uint32_t invocations);
 
@@ -159,6 +159,8 @@ public:
 
 private:
     int Latch(int code);
+    void RequireOpen() const;
+    int WaitLocked(std::unique_lock<std::mutex>& guard);
 
     Counters limits;
     Counters submitted;
@@ -172,12 +174,12 @@ private:
     std::condition_variable passed;
 };
 
-class Factory final : public SequenceFactoryInterface {
+class Factory final : public ISequenceFactory {
 public:
-    APS5_VABI int Create(SequenceInterface** out, std::uint32_t invocations, std::uint32_t dispatches, std::uint32_t syncPoints, std::uint32_t waits, const char* name) override;
-    APS5_VABI int Destroy(SequenceInterface* sequence) override;
-    APS5_VABI int CreateCompute(SequenceInterface** out, std::uint32_t invocations, std::uint32_t dispatches, std::uint32_t syncPoints, std::uint32_t waits, const char* name) override;
-    APS5_VABI int DestroyCompute(SequenceInterface* sequence) override;
+    APS5_VABI int Create(ISequence** out, std::uint32_t invocations, std::uint32_t dispatches, std::uint32_t syncPoints, std::uint32_t waits, const char* name) override;
+    APS5_VABI int Destroy(ISequence* sequence) override;
+    APS5_VABI int CreateCompute(ISequence** out, std::uint32_t invocations, std::uint32_t dispatches, std::uint32_t syncPoints, std::uint32_t waits, const char* name) override;
+    APS5_VABI int DestroyCompute(ISequence* sequence) override;
 
     Manager* manager = nullptr;
     std::uint32_t priority = 0;
@@ -234,13 +236,8 @@ bool Sequence::Open(Manager* owner, const Counters& limit, std::uint32_t runQueu
     return true;
 }
 
-void Sequence::Close() {
-    if (!open) return;
-    ReturnNodes(*manager, static_cast<std::uint32_t>(items.size()) + 1);
-    items.clear();
-    cursor = 0;
-    outstanding = 0;
-    open = false;
+void Sequence::RequireOpen() const {
+    if (!open) throw std::runtime_error("libSceJobManager: a destroyed sequence was used");
 }
 
 void Sequence::Advance() {
@@ -287,6 +284,7 @@ int Sequence::Dispatch(JobEntry entry, void* argument, std::uint32_t count, cons
     if (entry == nullptr) return SequenceNullArgument;
     if (count == 0 || count > MaxInvocations) return SequenceOutOfRange;
     const std::lock_guard guard(manager->lock);
+    RequireOpen();
     if (error != 0) return SequenceInvalid;
     const Counters before = submitted;
     submitted.invocations += count;
@@ -300,6 +298,7 @@ int Sequence::Dispatch(JobEntry entry, void* argument, std::uint32_t count, cons
 
 int Sequence::SyncPoint() {
     const std::lock_guard guard(manager->lock);
+    RequireOpen();
     if (error != 0) return SequenceInvalid;
     const std::uint32_t before = submitted.syncPoints++;
     if (before >= limits.syncPoints) return Latch(SequenceLimitExceeded);
@@ -310,8 +309,24 @@ int Sequence::SyncPoint() {
 }
 
 int Sequence::Wait() {
-    const std::lock_guard serial(waiters);
     std::unique_lock guard(manager->lock);
+    return WaitLocked(guard);
+}
+
+int Sequence::WaitAndClose(std::uint32_t index) {
+    std::unique_lock guard(manager->lock);
+    const int result = WaitLocked(guard);
+    ReturnNodes(*manager, static_cast<std::uint32_t>(items.size()) + 1);
+    items.clear();
+    cursor = 0;
+    outstanding = 0;
+    open = false;
+    manager->freeSequences |= std::uint32_t{1} << index;
+    return result;
+}
+
+int Sequence::WaitLocked(std::unique_lock<std::mutex>& guard) {
+    RequireOpen();
     int result = 0;
     if (error != 0) {
         result = SequenceInvalid;
@@ -322,6 +337,9 @@ int Sequence::Wait() {
     ++manager->blockedWaiters;
     manager->nodeFreed.wait(guard, [this] { return manager->freeNodes != 0; });
     --manager->freeNodes;
+    guard.unlock();
+    const std::lock_guard serial(waiters);
+    guard.lock();
     bool done = false;
     items.push_back({ItemType::Wait, this, nullptr, nullptr, 0, 0, 0, &done});
     Advance();
@@ -335,7 +353,7 @@ bool ValidLimits(std::uint32_t invocations, std::uint32_t dispatches, std::uint3
     return invocations != 0 || dispatches != 0 || syncPoints != 0 || waits != 0;
 }
 
-int Factory::Create(SequenceInterface** out, std::uint32_t invocations, std::uint32_t dispatches, std::uint32_t syncPoints, std::uint32_t waits, const char*) {
+int Factory::Create(ISequence** out, std::uint32_t invocations, std::uint32_t dispatches, std::uint32_t syncPoints, std::uint32_t waits, const char*) {
     if (out == nullptr) return SequenceNullArgument;
     if (!ValidLimits(invocations, dispatches, syncPoints, waits)) return SequenceOutOfRange;
     const std::lock_guard guard(manager->lock);
@@ -348,7 +366,7 @@ int Factory::Create(SequenceInterface** out, std::uint32_t invocations, std::uin
     return 0;
 }
 
-int Factory::Destroy(SequenceInterface* sequence) {
+int Factory::Destroy(ISequence* sequence) {
     Sequence* found = nullptr;
     std::uint32_t index = 0;
     {
@@ -359,20 +377,17 @@ int Factory::Destroy(SequenceInterface* sequence) {
         if (index == manager->sequences.size() || (manager->freeSequences & (std::uint32_t{1} << index)) != 0) return SequenceInvalid;
         found = &manager->sequences[index];
     }
-    static_cast<void>(found->Wait());
-    const std::lock_guard guard(manager->lock);
-    found->Close();
-    manager->freeSequences |= std::uint32_t{1} << index;
+    static_cast<void>(found->WaitAndClose(index));
     return 0;
 }
 
-int Factory::CreateCompute(SequenceInterface** out, std::uint32_t invocations, std::uint32_t dispatches, std::uint32_t syncPoints, std::uint32_t waits, const char*) {
+int Factory::CreateCompute(ISequence** out, std::uint32_t invocations, std::uint32_t dispatches, std::uint32_t syncPoints, std::uint32_t waits, const char*) {
     if (out == nullptr) return SequenceNullArgument;
     if (!ValidLimits(invocations, dispatches, syncPoints, waits)) return SequenceOutOfRange;
     return SequenceNoneFree;
 }
 
-int Factory::DestroyCompute(SequenceInterface*) {
+int Factory::DestroyCompute(ISequence*) {
     return SequenceInvalid;
 }
 
@@ -435,10 +450,10 @@ void StartWorkers(Manager& manager, const ConfigParams& config, std::uint32_t wo
         const KernelSchedParam schedule{static_cast<int>(config.priority)};
         PthreadAttr attributes = nullptr;
         if (scePthreadAttrInit(&attributes) != 0) throw std::runtime_error("libSceJobManager: scePthreadAttrInit failed");
-        scePthreadAttrSetinheritsched(&attributes, 0);
-        scePthreadAttrSetschedparam(&attributes, &schedule);
-        scePthreadAttrSetstacksize(&attributes, config.stackSize);
-        scePthreadAttrSetaffinity(&attributes, affinity);
+        if (scePthreadAttrSetinheritsched(&attributes, 0) != 0 || scePthreadAttrSetschedparam(&attributes, &schedule) != 0 ||
+            scePthreadAttrSetstacksize(&attributes, config.stackSize) != 0 || scePthreadAttrSetaffinity(&attributes, affinity) != 0) {
+            throw std::runtime_error("libSceJobManager: a worker thread attribute was rejected");
+        }
         Pthread thread = nullptr;
         const int created = scePthreadCreate(&thread, &attributes, WorkerEntry, &manager, name.c_str());
         scePthreadAttrDestroy(&attributes);
@@ -486,9 +501,10 @@ int Initialize(void* self, void* memory, std::uint64_t size, std::uint32_t worke
     return 0;
 }
 
-SequenceFactoryInterface* FactoryFor(void* self, std::uint32_t priority, bool unified) {
+ISequenceFactory* FactoryFor(void* self, std::uint32_t priority, bool unified) {
     Manager* manager = Implementation(self);
     if (manager == nullptr) return nullptr;
+    priority &= 0xffff;
     if (priority > 1) throw std::runtime_error("sce::Job::JobManager: sequence factory priority " + std::to_string(priority) + " does not exist");
     return unified ? &manager->unifiedFactories[priority] : &manager->factories[priority];
 }
@@ -558,11 +574,11 @@ int APS5_VABI _ZN3sce3Job10JobManager8shutdownEv(void* self) {
     return 0;
 }
 
-SequenceFactoryInterface* APS5_VABI _ZN3sce3Job10JobManager27getSequenceFactoryInterfaceENS1_8PriorityE(void* self, std::uint32_t priority) {
+ISequenceFactory* APS5_VABI _ZN3sce3Job10JobManager27getSequenceFactoryInterfaceENS1_8PriorityE(void* self, std::uint32_t priority) {
     return FactoryFor(self, priority, false);
 }
 
-SequenceFactoryInterface* APS5_VABI _ZN3sce3Job10JobManager34getUnifiedSequenceFactoryInterfaceENS1_8PriorityE(void* self, std::uint32_t priority) {
+ISequenceFactory* APS5_VABI _ZN3sce3Job10JobManager34getUnifiedSequenceFactoryInterfaceENS1_8PriorityE(void* self, std::uint32_t priority) {
     return FactoryFor(self, priority, true);
 }
 

core/libs/tests/GuestJobManager.cpp

@@ -39,13 +39,15 @@ struct Sequence {
     APS5_VABI virtual int Dispatch(JobEntry, void*, std::uint32_t, const char*) = 0;
     APS5_VABI virtual int SyncPoint() = 0;
     APS5_VABI virtual int Wait() = 0;
-    APS5_VABI virtual ~Sequence() = default;
+    APS5_VABI virtual void CompleteDestructor() = 0;
+    APS5_VABI virtual void DeletingDestructor() = 0;
 };
 
 struct SequenceFactory {
     APS5_VABI virtual int Create(Sequence**, std::uint32_t, std::uint32_t, std::uint32_t, std::uint32_t, const char*) = 0;
     APS5_VABI virtual int Destroy(Sequence*) = 0;
-    APS5_VABI virtual ~SequenceFactory() = default;
+    APS5_VABI virtual void CompleteDestructor() = 0;
+    APS5_VABI virtual void DeletingDestructor() = 0;
     APS5_VABI virtual int CreateCompute(Sequence**, std::uint32_t, std::uint32_t, std::uint32_t, std::uint32_t, const char*) = 0;
     APS5_VABI virtual int DestroyCompute(Sequence*) = 0;
 };
@@ -269,6 +271,53 @@ static void OrderingCase() {
     Require(Shutdown(manager.storage) == 0);
 }
 
+struct Nested {
+    Sequence* sequence;
+    std::atomic<bool> barrierAppended{false};
+    std::atomic<bool> followerAppended{false};
+    std::atomic<bool> leaderFinished{false};
+    std::atomic<int> followerResult{-1};
+    std::atomic<int> followerSawLeaderFinished{-1};
+};
+
+static void APS5_VABI FollowerJob(void* argument, std::uint32_t) {
+    auto& nested = *static_cast<Nested*>(argument);
+    nested.followerSawLeaderFinished = nested.leaderFinished ? 1 : 0;
+}
+
+static void APS5_VABI LeaderJob(void* argument, std::uint32_t) {
+    auto& nested = *static_cast<Nested*>(argument);
+    while (!nested.barrierAppended) std::this_thread::yield();
+    nested.followerResult = nested.sequence->Dispatch(FollowerJob, &nested, 1, "follower");
+    nested.followerAppended = true;
+    std::this_thread::sleep_for(std::chrono::milliseconds(20));
+    nested.leaderFinished = true;
+}
+
+static void NestedCase() {
+    Block block(SizeFor(2, 1, 4, 2, 2));
+    ManagerObject manager;
+    Require(Initialize(manager.storage, block.memory, block.bytes, 2, 1, nullptr) == 0);
+    SequenceFactory* factory = Factories(manager.storage, 0);
+    Nested nested{};
+    Require(factory->Create(&nested.sequence, 16, 4, 2, 2, "nested") == 0);
+    Require(nested.sequence->Dispatch(LeaderJob, &nested, 1, "leader") == 0);
+    Require(nested.sequence->SyncPoint() == 0);
+    nested.barrierAppended = true;
+    while (!nested.followerAppended) std::this_thread::yield();
+    Require(nested.sequence->Wait() == 0);
+    Require(nested.followerResult == 0 && nested.followerSawLeaderFinished == 1);
+    nested.sequence->CompleteDestructor();
+    Require(nested.sequence->Wait() == 0);
+    Require(factory->Destroy(nested.sequence) == 0);
+    Require(factory->Create(&nested.sequence, 1, 1, 0, 0, "no waits") == 0);
+    Require(nested.sequence->Wait() == LimitExceeded);
+    Require(nested.sequence->Dispatch(Nothing, nullptr, 1, nullptr) == Invalid);
+    Require(factory->Destroy(nested.sequence) == 0);
+    Require(Factories(manager.storage, 0x10000) == factory);
+    Require(Shutdown(manager.storage) == 0);
+}
+
 static void LimitsCase() {
     Block block(SizeFor(2, 2, 2, 1, 1));
     ManagerObject manager;
@@ -302,7 +351,7 @@ static void LimitsCase() {
     int result = 0;
     int accepted = 0;
     while (accepted < 0x3ff && (result = sequence->Dispatch(Nothing, nullptr, 1, nullptr)) == 0) ++accepted;
-    Require(result == NoNodes && accepted > 0);
+    Require(result == NoNodes && accepted == 33);
     Require(sequence->Dispatch(Nothing, nullptr, 1, nullptr) == Invalid);
     Require(Shutdown(manager.storage) == 0);
 }
@@ -377,6 +426,7 @@ int main(int argc, char** argv) {
     else if (name == "dispatch") DispatchCase();
     else if (name == "ordering") OrderingCase();
     else if (name == "limits") LimitsCase();
+    else if (name == "nested") NestedCase();
     else if (name == "priority") PriorityCase();
     else if (name == "shutdown") ShutdownCase();
     else return 2;

docs/dev/TechnicalDebt.md

@@ -521,4 +521,4 @@ Throughout the project, every function at every stage either **does exactly what
 - [Null vertex buffer V#](../../core/libs/prx/libSceAgcDriver/Graphics/include/VertexInput.hpp) (libSceAgcDriver) - a V# whose four dwords are all zero is bound to a zeroed vertex buffer of the attribute's size with stride 0, so every component of its attribute reads 0. This follows the V# fields in the [AMD RDNA2 ISA](https://gpuopen.com/download/rdna2-shader-instruction-set-architecture.pdf) (`NUM_RECORDS` 0, `DST_SEL` 0 selects 0), not measured on hardware.
 - [_fstatfs](../../core/libs/prx/libkernel/File/src/Stdio.cpp) (libkernel) fills the FreeBSD 11 `struct statfs` of the PS5 payload SDK with the host volume that holds the file: block size, transfer size, total, free and available blocks, maximum name length and the read-only, `nosuid` and `noexec` flags (on Windows only read-only, and `f_files` and `f_ffree` are 0, as NTFS has no fixed number of files). The filesystem type number and name, mount paths, owner, filesystem id and I/O counters are left 0 or empty, since they would describe a host mount, not a console one. Sockets and pipes return `EINVAL`, as FreeBSD's `getvnode` does; a Windows descriptor that is neither a file nor a pipe throws. Stremio Plus (PPSA74126) reads the free space before a download and only logs the other fields
 - [_WStoul](../../core/libs/prx/libc/src/Strings.cpp), [_Iswctype](../../core/libs/prx/libc/src/RuneLocale.cpp) and [swscanf_s](../../core/libs/prx/libc/src/ScanningWide.cpp) (libc) - `_WStoul` is `wcstoul` and throws for a base other than 0 or 2 to 36. `_Iswctype` takes the class numbers of Dinkumware's [xwcc.h](https://github.com/superna9999/ti-cgt6x/blob/cgt6x_7_3_4/include/xwcc.h) (1 `alnum` to 11 `xdigit`, 12 `blank`) and throws for any other value, including the FreeBSD masks `wctype` returns; the numbering is not confirmed on the console's `libc.prx`. `swscanf_s` follows C11 7.29.2.2 and K.3.9.1.5, but reads the size argument as `unsigned int` like `sscanf_s` where K.3.9.1.5 specifies `rsize_t`; the console's width is unverified. A runtime-constraint violation (null buffer, format or receiving pointer) only returns `EOF`, and a too small array gets a null first element as in `sscanf_s`. A literal mismatch is a matching failure and a field that is only a prefix, such as `0x` or `1e`, does not convert, as C11 requires; FreeBSD's `vfwscanf` reports an input failure for the first and converts the second. Integers out of range are truncated from `strtoll`/`strtoull`. It throws for `%p`, `%S`, `%C`, unknown length modifiers, scanset ranges, `%n` with suppression or width, and narrow `%s`, `%c` or `%[` input that the C-locale `wcrtomb` cannot convert
-- [sce::Job::JobManager](../../core/libs/prx/libSceJobManager/Export.cpp) (libSceJobManager) is written from the disassembly of the copies PPSA01474 and PPSA36082 ship in `sce_module`, and `guest_job_manager_tests` passes against both of those modules as well: the memory size formula, the error codes, sync point ordering, the per-sequence limits and the priority order are the original's. Only CPU sequences exist: `initialize` and `calculateRequiredMemorySize` with compute sequences, and both compute dispatches, are not implemented, and `createCompute` and `destroyCompute` return the errors the original returns when no compute sequence was configured. The manager's state lives on the host heap, with only the block header written to the title's memory; the original keeps everything in that block, and a block smaller than 0x2d8 bytes throws here. The worker sleep policy, the spin count and the persistent thread count are validated and stored but change nothing, a factory priority above 1 throws, and `shutdown` throws while a thread is blocked in a sequence wait. The order of the sequence and factory virtual functions is read from the modules' vtables only, since no title that calls them has been run
+- [sce::Job::JobManager](../../core/libs/prx/libSceJobManager/Export.cpp) (libSceJobManager) is written from the disassembly of the copies PPSA01474 and PPSA36082 ship in `sce_module`, and `guest_job_manager_tests` passes against both of those modules as well: the memory size formula, the error codes, sync point ordering, the per-sequence limits and the priority order are the original's. Only CPU sequences exist: `initialize` and `calculateRequiredMemorySize` with compute sequences, and both compute dispatches, are not implemented, and `createCompute` and `destroyCompute` return the errors the original returns when no compute sequence was configured. The manager's state lives on the host heap, with only the block header written to the title's memory; the original keeps everything in that block, and a block smaller than 0x2d8 bytes throws here. The worker sleep policy, the spin count and the persistent thread count are validated and stored but change nothing, a factory priority above 1 throws, `shutdown` throws while a thread is blocked in a sequence wait, and using a sequence after its `destroy` throws. `initialize` returns 0 where the original returns the result of its last import (`BG26hBGiNlw`, not resolved to a name), the `sceRazorCpuJobManager*` profiler calls are not made, and a worker thread that cannot be created or configured throws where the original does not check. The order of the sequence and factory virtual functions is read from the modules' vtables only, since no title that calls them has been run