From 658697e26596052241ff6f876bd77a03d57355d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tur=C3=A1nszki=20J=C3=A1nos?= Date: Mon, 24 Mar 2025 08:24:25 +0100 Subject: [PATCH] job system updates --- WickedEngine/wiJobSystem.cpp | 62 ++++++++++++++++++++------------ WickedEngine/wiJobSystem.h | 5 ++- WickedEngine/wiLoadingScreen.cpp | 2 +- WickedEngine/wiVersion.cpp | 2 +- 4 files changed, 45 insertions(+), 26 deletions(-) diff --git a/WickedEngine/wiJobSystem.cpp b/WickedEngine/wiJobSystem.cpp index c930f24de..0d3d982e8 100644 --- a/WickedEngine/wiJobSystem.cpp +++ b/WickedEngine/wiJobSystem.cpp @@ -31,7 +31,7 @@ namespace wi::jobsystem uint32_t groupJobOffset; uint32_t groupJobEnd; uint32_t sharedmemory_size; - inline void execute() + inline uint32_t execute() { JobArgs args; args.groupID = groupID; @@ -53,7 +53,7 @@ namespace wi::jobsystem task(args); } - AtomicAdd(&ctx->counter, -1); + return ctx->counter.fetch_sub(1); // returns context counter's previous value } }; struct JobQueue @@ -84,8 +84,10 @@ namespace wi::jobsystem wi::vector threads; std::unique_ptr jobQueuePerThread; std::atomic nextQueue{ 0 }; - std::condition_variable wakeCondition; - std::mutex wakeMutex; + std::condition_variable sleepingCondition; // for workers that are sleeping + std::mutex sleepingMutex; // for workers that are sleeping + std::condition_variable waitingCondition; // for unblocking a Wait() + std::mutex waitingMutex; // for unblocking a Wait() // Start working on a job queue // After the job queue is finished, it can switch to an other queue and steal jobs from there @@ -97,7 +99,14 @@ namespace wi::jobsystem JobQueue& job_queue = jobQueuePerThread[startingQueue % numThreads]; while (job_queue.pop_front(job)) { - job.execute(); + uint32_t progress_before = job.execute(); + if (progress_before == 1) + { + // This is likely the last job because the counter was 1 before it was decremented in execute() + // So wake up the waiting threads here + std::unique_lock lock(waitingMutex); + waitingCondition.notify_all(); + } } startingQueue++; // go to next queue } @@ -122,7 +131,7 @@ namespace wi::jobsystem { for (auto& x : resources) { - x.wakeCondition.notify_all(); // wakes up sleeping worker threads + x.sleepingCondition.notify_all(); // wakes up sleeping worker threads } } }); @@ -187,9 +196,10 @@ namespace wi::jobsystem for (uint32_t threadID = 0; threadID < res.numThreads; ++threadID) { -#ifdef PLATFORM_LINUX std::thread& worker = res.threads.emplace_back([threadID, priority, &res] { +#ifdef PLATFORM_LINUX + // from the sched(2) manpage: // In the current [Linux 2.6.23+] implementation, each unit of // difference in the nice values of two processes results in a @@ -217,16 +227,15 @@ namespace wi::jobsystem default: assert(0); } -#else - std::thread& worker = res.threads.emplace_back([threadID, &res] { -#endif +#endif // PLATFORM_LINUX + while (internal_state.alive.load()) { res.work(threadID); // finished with jobs, put to sleep - std::unique_lock lock(res.wakeMutex); - res.wakeCondition.wait(lock); + std::unique_lock lock(res.sleepingMutex); + res.sleepingCondition.wait(lock); } }); @@ -345,7 +354,7 @@ namespace wi::jobsystem PriorityResources& res = internal_state.resources[int(ctx.priority)]; // Context state is updated: - AtomicAdd(&ctx.counter, 1); + ctx.counter.fetch_add(1); Job job; job.ctx = &ctx; @@ -363,7 +372,7 @@ namespace wi::jobsystem } res.jobQueuePerThread[res.nextQueue.fetch_add(1) % res.numThreads].push_back(job); - res.wakeCondition.notify_one(); + res.sleepingCondition.notify_one(); } void Dispatch(context& ctx, uint32_t jobCount, uint32_t groupSize, const std::function& task, size_t sharedmemory_size) @@ -377,7 +386,7 @@ namespace wi::jobsystem const uint32_t groupCount = DispatchGroupCount(jobCount, groupSize); // Context state is updated: - AtomicAdd(&ctx.counter, groupCount); + ctx.counter.fetch_add(groupCount); Job job; job.ctx = &ctx; @@ -404,7 +413,7 @@ namespace wi::jobsystem if (res.numThreads > 1) { - res.wakeCondition.notify_all(); + res.sleepingCondition.notify_all(); } } @@ -417,7 +426,7 @@ namespace wi::jobsystem bool IsBusy(const context& ctx) { // Whenever the context label is greater than zero, it means that there is still work that needs to be done - return AtomicLoad(&ctx.counter) > 0; + return ctx.counter.load() > 0; } void Wait(const context& ctx) @@ -427,19 +436,26 @@ namespace wi::jobsystem PriorityResources& res = internal_state.resources[int(ctx.priority)]; // Wake any threads that might be sleeping: - res.wakeCondition.notify_all(); + res.sleepingCondition.notify_all(); - // work() will pick up any jobs that are on stand by and execute them on this thread: + // work() will pick up any jobs that are on standby and execute them on this thread: res.work(res.nextQueue.fetch_add(1) % res.numThreads); while (IsBusy(ctx)) { // If we are here, then there are still remaining jobs that work() couldn't pick up. - // In this case those jobs are not standing by on a queue but currently executing - // on other threads, so they cannot be picked up by this thread. - // Allow to swap out this thread by OS to not spin endlessly for nothing - std::this_thread::yield(); + // The thread enters a sleep until the !IsBusy() waitCondition is signaled + std::unique_lock lock(res.waitingMutex); + if (IsBusy(ctx)) // check after locking, to not enter wait when it was completed after lock + { + res.waitingCondition.wait(lock, [&ctx] { return !IsBusy(ctx); }); + } } } } + + uint32_t GetRemainingJobCount(const context& ctx) + { + return ctx.counter.load(); + } } diff --git a/WickedEngine/wiJobSystem.h b/WickedEngine/wiJobSystem.h index 464ce2a2f..1e8bcac1d 100644 --- a/WickedEngine/wiJobSystem.h +++ b/WickedEngine/wiJobSystem.h @@ -33,7 +33,7 @@ namespace wi::jobsystem // Defines a state of execution, can be waited on struct context { - volatile long counter = 0; + std::atomic counter{ 0 }; Priority priority = Priority::High; }; @@ -57,4 +57,7 @@ namespace wi::jobsystem // Wait until all threads become idle // Current thread will become a worker thread, executing jobs void Wait(const context& ctx); + + // Returns the number of remaining jobs + uint32_t GetRemainingJobCount(const context& ctx); } diff --git a/WickedEngine/wiLoadingScreen.cpp b/WickedEngine/wiLoadingScreen.cpp index 7d34f4bfd..f49e2daff 100644 --- a/WickedEngine/wiLoadingScreen.cpp +++ b/WickedEngine/wiLoadingScreen.cpp @@ -23,7 +23,7 @@ namespace wi { if (launchedTasks == 0) return 100; - uint32_t counter = AtomicLoad(&ctx.counter); + uint32_t counter = wi::jobsystem::GetRemainingJobCount(ctx); float percent = 1 - float(counter) / float(launchedTasks); return (int)std::round(percent * 100); } diff --git a/WickedEngine/wiVersion.cpp b/WickedEngine/wiVersion.cpp index 3b585281f..5319c2fc4 100644 --- a/WickedEngine/wiVersion.cpp +++ b/WickedEngine/wiVersion.cpp @@ -9,7 +9,7 @@ namespace wi::version // minor features, major updates, breaking compatibility changes const int minor = 71; // minor bug fixes, alterations, refactors, updates - const int revision = 718; + const int revision = 719; const std::string version_string = std::to_string(major) + "." + std::to_string(minor) + "." + std::to_string(revision);