diff --git a/core/libs/prx/libSceJobManager/CMakeLists.txt b/core/libs/prx/libSceJobManager/CMakeLists.txt index 002ba69513..02088aa791 100644 --- a/core/libs/prx/libSceJobManager/CMakeLists.txt +++ b/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) diff --git a/core/libs/prx/libSceJobManager/Export.cpp b/core/libs/prx/libSceJobManager/Export.cpp index 4e3f1395b3..ad0c9ed5ce 100644 --- a/core/libs/prx/libSceJobManager/Export.cpp +++ b/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& 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::min(runnable, std::uint64_t{manager->workerCount})), false); + } passed.wait(guard, [&done] { return done; }); --manager->blockedWaiters; return result; diff --git a/core/libs/tests/GuestJobManager.cpp b/core/libs/tests/GuestJobManager.cpp index 3d3644a538..e8bfeb526f 100644 --- a/core/libs/tests/GuestJobManager.cpp +++ b/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 publisherEntered{false}; + std::atomic spareEntered{false}; + std::atomic publish{false}; + std::atomic finished{false}; + std::atomic childCalls{0}; +}; + +static void APS5_VABI WorkerWaitChild(void* argument, std::uint32_t) { + ++static_cast(argument)->childCalls; +} + +static void APS5_VABI WorkerWaitPublisher(void* argument, std::uint32_t index) { + auto& state = *static_cast(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(); diff --git a/tools/run_job_manager_reference.py b/tools/run_job_manager_reference.py index 488916cba2..d804f3e573 100644 --- a/tools/run_job_manager_reference.py +++ b/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):