diff --git a/destrum/CMakeLists.txt b/destrum/CMakeLists.txt index f416df4..87d77a2 100644 --- a/destrum/CMakeLists.txt +++ b/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" diff --git a/destrum/include/destrum/Threading/JobSystem.h b/destrum/include/destrum/Threading/JobSystem.h new file mode 100644 index 0000000..688db7a --- /dev/null +++ b/destrum/include/destrum/Threading/JobSystem.h @@ -0,0 +1,95 @@ +#ifndef DESTRUM_JOBSYSTEM_H +#define DESTRUM_JOBSYSTEM_H + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +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 m_Pending{0}; +}; + +struct JobDispatchArgs { + std::uint32_t index{0}; + std::uint32_t groupIndex{0}; +}; + +class JobSystem final : public Singleton { +public: + friend class Singleton; + using Job = std::function; + + ~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(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 job); + void Wait(JobCounter& counter); + + template + auto Async(F&& fn) -> std::future>> + { + using Result = std::invoke_result_t>; + auto task = std::make_shared>(std::forward(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 m_Workers; + std::mutex m_Mutex; + std::condition_variable m_WorkAvailable; + std::deque m_HighQueue; + std::deque m_LowQueue; + std::uint32_t m_ActiveLowJobs{0}; + std::uint32_t m_MaxActiveLowJobs{1}; + bool m_Stopping{false}; + + std::mutex m_MainThreadMutex; + std::vector m_MainThreadQueue; +}; + +#endif // DESTRUM_JOBSYSTEM_H diff --git a/destrum/src/App.cpp b/destrum/src/App.cpp index cee4cbc..47ea3e3 100644 --- a/destrum/src/App.cpp +++ b/destrum/src/App.cpp @@ -16,6 +16,7 @@ #include #include #include +#include #include 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) { diff --git a/destrum/src/Threading/JobSystem.cpp b/destrum/src/Threading/JobSystem.cpp new file mode 100644 index 0000000..89b7075 --- /dev/null +++ b/destrum/src/Threading/JobSystem.cpp @@ -0,0 +1,230 @@ +#include + +#include +#include +#include + +#include +#include +#include + +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(1, hardwareThreads > 1 ? hardwareThreads - 1 : 1); + } + + m_Stopping = false; + m_MaxActiveLowJobs = std::max(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 job) +{ + if (count == 0) { + return; + } + groupSize = std::max(1, groupSize); + + auto shared = std::make_shared>(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 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(); + } + } +} diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index bd37957..b76135b 100644 --- a/tests/CMakeLists.txt +++ b/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() diff --git a/tests/jobs_tests.cpp b/tests/jobs_tests.cpp new file mode 100644 index 0000000..e11b775 --- /dev/null +++ b/tests/jobs_tests.cpp @@ -0,0 +1,167 @@ +#include +#include +#include +#include +#include + +#include + +#include "TestUtils.h" + +using namespace test; + +namespace { + JobSystem& Jobs() { return JobSystem::GetInstance(); } + + void testExecuteAndWait() + { + Jobs().Init(4); + JobCounter counter; + std::atomic 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> 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 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 release{false}; + auto blocker = Jobs().Async([&release] { + while (!release) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + return 7; + }); + + JobCounter counter; + std::atomic 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 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{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 runs{0}; + std::vector> 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(); +} diff --git a/tests/main.cpp b/tests/main.cpp index 3a30944..9d6a421 100644 --- a/tests/main.cpp +++ b/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}, }; }