feat: add JobSystem

destrum/CMakeLists.txt

@@ -134,6 +134,8 @@ set(SRC_FILES
         "src/Util/DeltaTime.cpp"
         "src/Util/ImGuiUtils.cpp"
 
+        "src/Threading/JobSystem.cpp"
+
         "src/Physics/PhysicsWorld.cpp"
         "src/Physics/PhysicsSceneBridge.cpp"
         "src/Physics/JoltPhysicsWorld.cpp"

destrum/include/destrum/Threading/JobSystem.h

@@ -0,0 +1,95 @@
+#ifndef DESTRUM_JOBSYSTEM_H
+#define DESTRUM_JOBSYSTEM_H
+
+#include <atomic>
+#include <condition_variable>
+#include <cstdint>
+#include <deque>
+#include <functional>
+#include <future>
+#include <memory>
+#include <mutex>
+#include <thread>
+#include <type_traits>
+#include <vector>
+
+#include <destrum/Singleton.h>
+
+enum class JobPriority {
+    High,
+    Low,
+};
+
+class JobCounter {
+public:
+    JobCounter() = default;
+    JobCounter(const JobCounter&) = delete;
+    JobCounter& operator=(const JobCounter&) = delete;
+
+    [[nodiscard]] bool IsDone() const { return m_Pending.load(std::memory_order_acquire) == 0; }
+
+private:
+    friend class JobSystem;
+    std::atomic<int> m_Pending{0};
+};
+
+struct JobDispatchArgs {
+    std::uint32_t index{0};
+    std::uint32_t groupIndex{0};
+};
+
+class JobSystem final : public Singleton<JobSystem> {
+public:
+    friend class Singleton<JobSystem>;
+    using Job = std::function<void()>;
+
+    ~JobSystem() override;
+
+    // 0 workers picks hardware_concurrency - 1
+    void Init(std::uint32_t workerCount = 0);
+    void Shutdown();
+
+    [[nodiscard]] bool IsRunning() const { return !m_Workers.empty(); }
+    [[nodiscard]] std::uint32_t GetWorkerCount() const { return static_cast<std::uint32_t>(m_Workers.size()); }
+
+    void Execute(JobCounter& counter, Job job, JobPriority priority = JobPriority::High);
+    void Dispatch(JobCounter& counter, std::uint32_t count, std::uint32_t groupSize,
+                  std::function<void(JobDispatchArgs)> job);
+    void Wait(JobCounter& counter);
+
+    template <typename F>
+    auto Async(F&& fn) -> std::future<std::invoke_result_t<std::decay_t<F>>>
+    {
+        using Result = std::invoke_result_t<std::decay_t<F>>;
+        auto task = std::make_shared<std::packaged_task<Result()>>(std::forward<F>(fn));
+        auto future = task->get_future();
+        Submit([task] { (*task)(); }, JobPriority::Low);
+        return future;
+    }
+
+    void RunOnMainThread(Job job);
+    void DrainMainThreadQueue();
+
+private:
+    JobSystem() = default;
+
+    void Submit(Job job, JobPriority priority);
+    bool TryPopLocked(Job& job, bool& isLow);
+    [[nodiscard]] bool CanTakeLowLocked() const;
+    bool TryRunHighJob();
+    void WorkerLoop(std::uint32_t index);
+
+    std::vector<std::thread> m_Workers;
+    std::mutex m_Mutex;
+    std::condition_variable m_WorkAvailable;
+    std::deque<Job> m_HighQueue;
+    std::deque<Job> m_LowQueue;
+    std::uint32_t m_ActiveLowJobs{0};
+    std::uint32_t m_MaxActiveLowJobs{1};
+    bool m_Stopping{false};
+
+    std::mutex m_MainThreadMutex;
+    std::vector<Job> m_MainThreadQueue;
+};
+
+#endif // DESTRUM_JOBSYSTEM_H

destrum/src/App.cpp

@@ -16,6 +16,7 @@
 #include <destrum/Scene/ResourceCollector.h>
 #include <destrum/Scene/Scene.h>
 #include <destrum/Scene/SceneManager.h>
+#include <destrum/Threading/JobSystem.h>
 #include <destrum/Util/DeltaTime.h>
 
 namespace {
@@ -60,6 +61,7 @@ void App::init(const AppParams& params)
         tracy::SetThreadName("Main Thread");
         TracySetProgramName(params.appName.c_str());
         AssetFS::GetInstance().Init(params.exeDir);
+        JobSystem::GetInstance().Init();
 
         window.Create({
             .title = params.windowTitle,
@@ -171,6 +173,7 @@ void App::run()
             continue;
         }
 
+        JobSystem::GetInstance().DrainMainThreadQueue();
         if (layer) {
             layer->onBeginFrame();
         }
@@ -386,6 +389,7 @@ void App::cleanup()
     // The one place we wait for the GPU on shutdown: the cleanup calls below destroy
     // pipelines and other objects directly, which is only valid once nothing is in flight.
     attempt([this] { gfxDevice.waitIdle(); });
+    attempt([] { JobSystem::GetInstance().Shutdown(); });
 
     attempt([this] {
         if (layer) {

destrum/src/Threading/JobSystem.cpp

@@ -0,0 +1,230 @@
+#include <destrum/Threading/JobSystem.h>
+
+#include <algorithm>
+#include <exception>
+#include <string>
+
+#include <spdlog/spdlog.h>
+#include <tracy/Tracy.hpp>
+#include <common/TracySystem.hpp>
+
+JobSystem::~JobSystem()
+{
+    try {
+        Shutdown();
+    } catch (...) {
+    }
+}
+
+void JobSystem::Init(std::uint32_t workerCount)
+{
+    if (IsRunning()) {
+        return;
+    }
+
+    if (workerCount == 0) {
+        const std::uint32_t hardwareThreads = std::thread::hardware_concurrency();
+        workerCount = std::max<std::uint32_t>(1, hardwareThreads > 1 ? hardwareThreads - 1 : 1);
+    }
+
+    m_Stopping = false;
+    m_MaxActiveLowJobs = std::max<std::uint32_t>(1, workerCount / 2);
+
+    m_Workers.reserve(workerCount);
+    for (std::uint32_t i = 0; i < workerCount; ++i) {
+        m_Workers.emplace_back([this, i] { WorkerLoop(i); });
+    }
+    spdlog::info("JobSystem started with {} workers", workerCount);
+}
+
+void JobSystem::Shutdown()
+{
+    if (!IsRunning()) {
+        return;
+    }
+
+    {
+        std::lock_guard lock(m_Mutex);
+        m_Stopping = true;
+    }
+    m_WorkAvailable.notify_all();
+
+    for (std::thread& worker : m_Workers) {
+        if (worker.joinable()) {
+            worker.join();
+        }
+    }
+    m_Workers.clear();
+
+    {
+        std::lock_guard lock(m_Mutex);
+        m_HighQueue.clear();
+        m_LowQueue.clear();
+        m_ActiveLowJobs = 0;
+        m_Stopping = false;
+    }
+    {
+        std::lock_guard lock(m_MainThreadMutex);
+        m_MainThreadQueue.clear();
+    }
+}
+
+void JobSystem::Execute(JobCounter& counter, Job job, JobPriority priority)
+{
+    counter.m_Pending.fetch_add(1, std::memory_order_relaxed);
+    Submit([&counter, job = std::move(job)] {
+        try {
+            job();
+        } catch (const std::exception& e) {
+            spdlog::error("Job threw: {}", e.what());
+        } catch (...) {
+            spdlog::error("Job threw an unknown exception");
+        }
+        counter.m_Pending.fetch_sub(1, std::memory_order_release);
+    }, priority);
+}
+
+void JobSystem::Dispatch(JobCounter& counter, std::uint32_t count, std::uint32_t groupSize,
+                         std::function<void(JobDispatchArgs)> job)
+{
+    if (count == 0) {
+        return;
+    }
+    groupSize = std::max<std::uint32_t>(1, groupSize);
+
+    auto shared = std::make_shared<std::function<void(JobDispatchArgs)>>(std::move(job));
+    const std::uint32_t groupCount = (count + groupSize - 1) / groupSize;
+    for (std::uint32_t group = 0; group < groupCount; ++group) {
+        Execute(counter, [shared, group, groupSize, count] {
+            const std::uint32_t begin = group * groupSize;
+            const std::uint32_t end = std::min(begin + groupSize, count);
+            for (std::uint32_t i = begin; i < end; ++i) {
+                (*shared)(JobDispatchArgs{.index = i, .groupIndex = group});
+            }
+        });
+    }
+}
+
+void JobSystem::Wait(JobCounter& counter)
+{
+    ZoneScopedN("JobSystem::Wait");
+    while (!counter.IsDone()) {
+        if (!TryRunHighJob()) {
+            std::this_thread::yield();
+        }
+    }
+}
+
+void JobSystem::RunOnMainThread(Job job)
+{
+    std::lock_guard lock(m_MainThreadMutex);
+    m_MainThreadQueue.push_back(std::move(job));
+}
+
+void JobSystem::DrainMainThreadQueue()
+{
+    ZoneScopedN("JobSystem::DrainMainThreadQueue");
+    std::vector<Job> jobs;
+    {
+        std::lock_guard lock(m_MainThreadMutex);
+        jobs.swap(m_MainThreadQueue);
+    }
+    for (Job& job : jobs) {
+        try {
+            job();
+        } catch (const std::exception& e) {
+            spdlog::error("Main thread job threw: {}", e.what());
+        } catch (...) {
+            spdlog::error("Main thread job threw an unknown exception");
+        }
+    }
+}
+
+void JobSystem::Submit(Job job, JobPriority priority)
+{
+    if (!IsRunning()) {
+        job();
+        return;
+    }
+
+    {
+        std::lock_guard lock(m_Mutex);
+        if (priority == JobPriority::High) {
+            m_HighQueue.push_back(std::move(job));
+        } else {
+            m_LowQueue.push_back(std::move(job));
+        }
+    }
+    m_WorkAvailable.notify_one();
+}
+
+bool JobSystem::CanTakeLowLocked() const
+{
+    return !m_LowQueue.empty() && (m_Stopping || m_ActiveLowJobs < m_MaxActiveLowJobs);
+}
+
+bool JobSystem::TryPopLocked(Job& job, bool& isLow)
+{
+    if (!m_HighQueue.empty()) {
+        job = std::move(m_HighQueue.front());
+        m_HighQueue.pop_front();
+        isLow = false;
+        return true;
+    }
+    if (CanTakeLowLocked()) {
+        job = std::move(m_LowQueue.front());
+        m_LowQueue.pop_front();
+        ++m_ActiveLowJobs;
+        isLow = true;
+        return true;
+    }
+    return false;
+}
+
+bool JobSystem::TryRunHighJob()
+{
+    Job job;
+    {
+        std::lock_guard lock(m_Mutex);
+        if (m_HighQueue.empty()) {
+            return false;
+        }
+        job = std::move(m_HighQueue.front());
+        m_HighQueue.pop_front();
+    }
+    job();
+    return true;
+}
+
+void JobSystem::WorkerLoop(std::uint32_t index)
+{
+    const std::string name = "Job Worker " + std::to_string(index);
+    tracy::SetThreadName(name.c_str());
+
+    for (;;) {
+        Job job;
+        bool isLow = false;
+        {
+            std::unique_lock lock(m_Mutex);
+            m_WorkAvailable.wait(lock, [this] {
+                return m_Stopping || !m_HighQueue.empty() || CanTakeLowLocked();
+            });
+            if (!TryPopLocked(job, isLow)) {
+                return;
+            }
+        }
+
+        {
+            ZoneScopedN("Job");
+            job();
+        }
+
+        if (isLow) {
+            {
+                std::lock_guard lock(m_Mutex);
+                --m_ActiveLowJobs;
+            }
+            m_WorkAvailable.notify_one();
+        }
+    }
+}

tests/CMakeLists.txt

@@ -10,6 +10,7 @@ add_executable(destrum_tests
         gameplay_tests.cpp
         editor_tests.cpp
         audio_tests.cpp
+        jobs_tests.cpp
         ${CMAKE_SOURCE_DIR}/lightkeeper/src/components/Activatable.cpp
         ${CMAKE_SOURCE_DIR}/lightkeeper/src/components/Interactable.cpp
         ${CMAKE_SOURCE_DIR}/lightkeeper/src/components/InteractableButton.cpp
@@ -38,6 +39,6 @@ target_link_libraries(destrum_tests PRIVATE destrum::destrum destrum::editor)
 destrum_enable_warnings(destrum_tests)
 destrum_copy_runtime_dlls(destrum_tests)
 
-foreach(suite core serialization physics assets graphics ui gameplay editor audio)
+foreach(suite core serialization physics assets graphics ui gameplay editor audio jobs)
     add_test(NAME destrum_tests.${suite} COMMAND destrum_tests ${suite})
 endforeach()

tests/jobs_tests.cpp

@@ -0,0 +1,167 @@
+#include <atomic>
+#include <chrono>
+#include <stdexcept>
+#include <thread>
+#include <vector>
+
+#include <destrum/Threading/JobSystem.h>
+
+#include "TestUtils.h"
+
+using namespace test;
+
+namespace {
+    JobSystem& Jobs() { return JobSystem::GetInstance(); }
+
+    void testExecuteAndWait()
+    {
+        Jobs().Init(4);
+        JobCounter counter;
+        std::atomic<int> sum{0};
+        for (int i = 1; i <= 100; ++i) {
+            Jobs().Execute(counter, [&sum, i] { sum += i; });
+        }
+        Jobs().Wait(counter);
+        check(counter.IsDone(), "counter should be done after Wait");
+        check(sum == 5050, "every executed job should run before Wait returns");
+        Jobs().Shutdown();
+    }
+
+    void testDispatchCoversEveryIndexOnce()
+    {
+        Jobs().Init(4);
+        constexpr std::uint32_t count = 1003;
+        std::vector<std::atomic<int>> hits(count);
+        JobCounter counter;
+        Jobs().Dispatch(counter, count, 64, [&hits](JobDispatchArgs args) {
+            check(args.groupIndex == args.index / 64, "group index should match the index range");
+            ++hits[args.index];
+        });
+        Jobs().Wait(counter);
+        for (const auto& hit : hits) {
+            check(hit == 1, "Dispatch should visit each index exactly once");
+        }
+        Jobs().Shutdown();
+    }
+
+    void testNestedWaitDoesNotDeadlock()
+    {
+        Jobs().Init(2);
+        JobCounter outer;
+        std::atomic<int> innerRuns{0};
+        for (int i = 0; i < 8; ++i) {
+            Jobs().Execute(outer, [&innerRuns] {
+                JobCounter inner;
+                for (int j = 0; j < 8; ++j) {
+                    Jobs().Execute(inner, [&innerRuns] { ++innerRuns; });
+                }
+                Jobs().Wait(inner);
+            });
+        }
+        Jobs().Wait(outer);
+        check(innerRuns == 64, "nested jobs should all run");
+        Jobs().Shutdown();
+    }
+
+    void testBlockedLowJobDoesNotStallHigh()
+    {
+        Jobs().Init(2);
+        std::atomic<bool> release{false};
+        auto blocker = Jobs().Async([&release] {
+            while (!release) {
+                std::this_thread::sleep_for(std::chrono::milliseconds(1));
+            }
+            return 7;
+        });
+
+        JobCounter counter;
+        std::atomic<int> runs{0};
+        Jobs().Dispatch(counter, 32, 4, [&runs](JobDispatchArgs) { ++runs; });
+        Jobs().Wait(counter);
+        check(runs == 32, "high jobs should finish while a low job is blocked");
+
+        release = true;
+        check(blocker.get() == 7, "Async should return the job result");
+        Jobs().Shutdown();
+    }
+
+    void testAsyncRethrows()
+    {
+        Jobs().Init(2);
+        auto future = Jobs().Async([]() -> int { throw std::runtime_error("boom"); });
+        bool threw = false;
+        try {
+            (void)future.get();
+        } catch (const std::runtime_error&) {
+            threw = true;
+        }
+        check(threw, "Async should rethrow the job's exception on get");
+
+        JobCounter counter;
+        Jobs().Execute(counter, [] { throw std::runtime_error("logged"); });
+        Jobs().Wait(counter);
+        check(counter.IsDone(), "a throwing job should still complete its counter");
+        Jobs().Shutdown();
+    }
+
+    void testMainThreadQueue()
+    {
+        Jobs().Init(2);
+        const auto mainId = std::this_thread::get_id();
+        std::vector<int> order;
+        bool onMain = true;
+
+        JobCounter counter;
+        Jobs().Execute(counter, [&] {
+            for (int i = 0; i < 5; ++i) {
+                Jobs().RunOnMainThread([&, i] {
+                    order.push_back(i);
+                    onMain = onMain && std::this_thread::get_id() == mainId;
+                });
+            }
+        });
+        Jobs().Wait(counter);
+        check(order.empty(), "main thread jobs should wait for the drain");
+
+        Jobs().DrainMainThreadQueue();
+        check(order == std::vector<int>{0, 1, 2, 3, 4}, "main thread jobs should run in submission order");
+        check(onMain, "main thread jobs should run on the draining thread");
+        Jobs().Shutdown();
+    }
+
+    void testShutdownDrainsPendingJobs()
+    {
+        Jobs().Init(1);
+        std::atomic<int> runs{0};
+        std::vector<std::future<void>> futures;
+        for (int i = 0; i < 50; ++i) {
+            futures.push_back(Jobs().Async([&runs] { ++runs; }));
+        }
+        Jobs().Shutdown();
+        Jobs().Shutdown();
+        check(runs == 50, "Shutdown should finish queued jobs");
+        check(!Jobs().IsRunning(), "job system should be stopped");
+    }
+
+    void testRunsInlineWhenStopped()
+    {
+        check(!Jobs().IsRunning(), "job system should start stopped");
+        JobCounter counter;
+        int value = 0;
+        Jobs().Execute(counter, [&value] { value = 3; });
+        check(value == 3 && counter.IsDone(), "jobs should run inline without workers");
+        check(Jobs().Async([] { return 5; }).get() == 5, "Async should run inline without workers");
+    }
+}
+
+void RunJobsTests()
+{
+    testRunsInlineWhenStopped();
+    testExecuteAndWait();
+    testDispatchCoversEveryIndexOnce();
+    testNestedWaitDoesNotDeadlock();
+    testBlockedLowJobDoesNotStallHigh();
+    testAsyncRethrows();
+    testMainThreadQueue();
+    testShutdownDrainsPendingJobs();
+}

tests/main.cpp

@@ -11,6 +11,7 @@ void RunUiTests();
 void RunGameplayTests();
 void RunEditorTests();
 void RunAudioTests();
+void RunJobsTests();
 
 namespace {
     struct Suite {
@@ -28,6 +29,7 @@ namespace {
         {"gameplay", RunGameplayTests},
         {"editor", RunEditorTests},
         {"audio", RunAudioTests},
+        {"jobs", RunJobsTests},
     };
 }