fix(job-manager): wake runnable work before a worker waits

core/libs/prx/libSceJobManager/CMakeLists.txt

@@ -19,7 +19,7 @@ if(BUILD_TESTING)
             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 worker_spin)
+    foreach(testCase IN ITEMS memory validation dispatch ordering nested limits node_pool priority shutdown worker_policy worker_wake worker_wait worker_count worker_spin)
         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

@@ -247,8 +247,8 @@ struct Manager {
 
 thread_local Manager* currentWorker = nullptr;
 
-void WakeWorkers(Manager& manager, std::uint32_t invocations) {
-    const std::uint32_t available = invocations - (currentWorker == &manager ? 1u : 0u);
+void WakeWorkers(Manager& manager, std::uint32_t invocations, bool publisherCanRun = true) {
+    const std::uint32_t available = invocations - (publisherCanRun && currentWorker == &manager ? 1u : 0u);
     manager.wakeTokens += std::min(available, manager.sleepingWorkers - manager.wakeTokens);
     if (manager.wakeTokens != 0) manager.work.notify_all();
 }
@@ -477,6 +477,13 @@ int Sequence::WaitLocked(std::unique_lock<std::mutex>& guard) {
     bool done = false;
     items.push_back({ItemType::Wait, this, nullptr, nullptr, 0, 0, 0, &done});
     Advance();
+    if (!done && currentWorker == manager) {
+        std::uint64_t runnable = 0;
+        for (const auto& queue : manager->runQueues) {
+            for (const Item* item : queue) runnable += item->count - item->handedOut;
+        }
+        WakeWorkers(*manager, static_cast<std::uint32_t>(std::min(runnable, std::uint64_t{manager->workerCount})), false);
+    }
     passed.wait(guard, [&done] { return done; });
     --manager->blockedWaiters;
     return result;

core/libs/tests/GuestJobManager.cpp

@@ -303,6 +303,61 @@ static void APS5_VABI LeaderJob(void* argument, std::uint32_t) {
     nested.leaderFinished = true;
 }
 
+static void RequireParked(int clock);
+
+struct WorkerWait {
+    int spareClock = 0;
+    Sequence* child;
+    std::atomic<bool> publisherEntered{false};
+    std::atomic<bool> spareEntered{false};
+    std::atomic<bool> publish{false};
+    std::atomic<bool> finished{false};
+    std::atomic<unsigned> childCalls{0};
+};
+
+static void APS5_VABI WorkerWaitChild(void* argument, std::uint32_t) {
+    ++static_cast<WorkerWait*>(argument)->childCalls;
+}
+
+static void APS5_VABI WorkerWaitPublisher(void* argument, std::uint32_t index) {
+    auto& state = *static_cast<WorkerWait*>(argument);
+    if (index != 0) {
+        Require(pthread_getcpuclockid_nid_postfix(scePthreadSelf(), &state.spareClock) == 0);
+        state.spareEntered = true;
+        return;
+    }
+    state.publisherEntered = true;
+    while (!state.publish) std::this_thread::yield();
+    Require(state.child->Dispatch(WorkerWaitChild, &state, 1, "child") == 0);
+    Require(state.child->Wait() == 0);
+    state.finished = true;
+}
+
+static void WorkerWaitCase() {
+    Block block(SizeFor(2, 2, 4, 0, 4));
+    ManagerObject manager;
+    const ConfigParams config{0, 0x21fff, 0x10000, 0x2bc, "WorkerWait", 0, 0, 0, 0};
+    Require(Initialize(manager.storage, block.memory, block.bytes, 2, 2, &config) == 0);
+    Require(SetPersistentThreadCount(manager.storage, 0) == 0);
+    SequenceFactory* factory = Factories(manager.storage, 0);
+    Sequence* publisher = nullptr;
+    WorkerWait state{};
+    Require(factory->Create(&publisher, 2, 1, 0, 1, "publisher") == 0);
+    Require(factory->Create(&state.child, 1, 1, 0, 1, "child") == 0);
+    Require(publisher->Dispatch(WorkerWaitPublisher, &state, 2, "park spare") == 0);
+    auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(3);
+    while ((!state.publisherEntered || !state.spareEntered) && std::chrono::steady_clock::now() < deadline) std::this_thread::yield();
+    Require(state.publisherEntered && state.spareEntered);
+    RequireParked(state.spareClock);
+    state.publish = true;
+    deadline = std::chrono::steady_clock::now() + std::chrono::seconds(3);
+    while (!state.finished && std::chrono::steady_clock::now() < deadline) std::this_thread::yield();
+    Require(state.finished && state.childCalls == 1);
+    Require(publisher->Wait() == 0);
+    Require(factory->Destroy(publisher) == 0 && factory->Destroy(state.child) == 0);
+    Require(Shutdown(manager.storage) == 0);
+}
+
 static void NestedCase() {
     Block block(SizeFor(2, 1, 4, 2, 2));
     ManagerObject manager;
@@ -711,6 +766,7 @@ int RunJobManagerCase(const char* caseName) {
     else if (name == "priority") PriorityCase();
     else if (name == "shutdown") ShutdownCase();
     else if (name == "worker_policy") WorkerPolicyCase();
+    else if (name == "worker_wait") WorkerWaitCase();
     else if (name == "worker_wake") WorkerWakeCase();
     else if (name == "worker_spin") WorkerSpinCase();
     else if (name == "worker_count") WorkerCountCase();

tools/run_job_manager_reference.py

@@ -15,7 +15,7 @@ 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", "worker_spin")
+CASES = ("memory", "validation", "dispatch", "ordering", "nested", "limits", "node_pool", "priority", "shutdown", "worker_policy", "worker_wake", "worker_wait", "worker_count", "worker_spin")
 
 
 def stage_libraries(directory, build, toolchain, dependencies):