diff --git a/CMakeLists.txt b/CMakeLists.txt index 2212728b..9a977ab4 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -73,6 +73,7 @@ add_subdirectory(vendors/assimp) # --- Find system packages --- find_package(OpenGL REQUIRED) +find_package(Threads REQUIRED) # --- Vulkan SDK (vendored) --- set(VULKAN_SDK "${VENDOR_DIR}/VulkanSdk") @@ -103,6 +104,7 @@ set(CORE_SOURCES src/Core/Application.cpp src/Core/CommandLine.cpp src/Core/File.cpp + src/Core/JobSystem.cpp src/Core/Logger.cpp src/Core/ScopedTimer.cpp src/Core/Serializable.cpp @@ -301,6 +303,7 @@ target_link_libraries(Engine PRIVATE ${OPENGL_LIBRARIES} freeglut assimp::assimp + Threads::Threads ) target_link_libraries(Engine PUBLIC yaml-cpp::yaml-cpp diff --git a/include/public/Core/Application.hpp b/include/public/Core/Application.hpp index 896cafaa..70e13463 100644 --- a/include/public/Core/Application.hpp +++ b/include/public/Core/Application.hpp @@ -57,6 +57,7 @@ struct ENGINE_API ApplicationDefaults { class Window; class DebugOverlay; +class JobSystem; namespace RenderEngine { class Renderer; } /// Owns the window, renderer, and game loop. One instance per process, @@ -210,6 +211,7 @@ class ENGINE_API Application { RenderEngine::Renderer* renderer; DebugOverlay* m_DebugOverlay = nullptr; Benchmark* m_benchmark = nullptr; + JobSystem* m_jobSystem = nullptr; GraphicsConfig m_graphicsConfig; float DeltaTime; diff --git a/include/public/Core/JobSystem.hpp b/include/public/Core/JobSystem.hpp new file mode 100644 index 00000000..aab904a9 --- /dev/null +++ b/include/public/Core/JobSystem.hpp @@ -0,0 +1,106 @@ +#ifndef _JOBSYSTEM_HPP_ +#define _JOBSYSTEM_HPP_ + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace Sleak { + +/// Refers to a submitted job (or a ParallelFor batch group) that can be +/// waited on. Copyable, and stays valid to query after the job finishes. +/// @ingroup core +struct JobHandle { + uint32_t index = 0; + uint32_t generation = 0; +}; + +/// Fixed pool of worker threads fed by one shared queue. +/// +/// Application owns the pool for the lifetime of the process and it is +/// reachable through GetInstance(). Jobs are plain callables. Submit() +/// returns a JobHandle to wait on, ParallelFor() splits an index range +/// into batches, and both waits run queued jobs on the calling thread +/// instead of sleeping, so a job may submit and wait on other jobs. +/// +/// Submit, Wait, IsDone and ParallelFor are safe to call from any thread, +/// including from inside a job. Jobs must not touch the GPU, the scene, +/// or anything else that is main-thread only. Keep jobs coarse (a +/// texture decode, a few hundred items of a ParallelFor), every job goes +/// through one mutex. +/// +/// @code{.cpp} +/// auto* jobs = Sleak::JobSystem::GetInstance(); +/// Sleak::JobHandle h = jobs->Submit([&] { DecodeSomething(); }); +/// jobs->ParallelFor(items.size(), 64, [&](uint32_t i) { Work(items[i]); }); +/// jobs->Wait(h); +/// @endcode +/// @ingroup core +class ENGINE_API JobSystem { + public: + /// Starts workerCount threads, or hardware_concurrency minus one + /// (at least one) when workerCount is 0. + explicit JobSystem(uint32_t workerCount = 0); + /// Runs every job still queued, then joins the workers. + ~JobSystem(); + + JobSystem(const JobSystem&) = delete; + JobSystem& operator=(const JobSystem&) = delete; + + /// Queues fn to run on a worker. + JobHandle Submit(std::function fn); + + /// Returns once the job behind handle has finished, running other + /// queued jobs on this thread in the meantime. + void Wait(JobHandle handle); + + /// True when the job behind handle has finished. + bool IsDone(JobHandle handle) const; + + /// Calls fn(i) for every i in [0, count) in batches of batchSize and + /// returns when all of them have run. The calling thread helps. + void ParallelFor(uint32_t count, uint32_t batchSize, + const std::function& fn); + + uint32_t GetWorkerCount() const { + return static_cast(m_workers.size()); + } + + /// The pool Application created, or null outside of an Application. + static JobSystem* GetInstance() { return Instance; } + + private: + struct Job { + std::function fn; + uint32_t counter; + }; + + struct Counter { + uint32_t pending = 0; + uint32_t generation = 0; + }; + + uint32_t AllocCounter(uint32_t pending); + void Push(std::function fn, uint32_t counter); + bool RunOne(std::unique_lock& lock); + void WorkerLoop(); + + mutable std::mutex m_mutex; + std::condition_variable m_wake; + std::deque m_queue; + std::vector m_counters; + std::vector m_freeCounters; + std::vector m_workers; + bool m_stopping = false; + + static JobSystem* Instance; +}; + +} // namespace Sleak + +#endif // _JOBSYSTEM_HPP_ diff --git a/include/public/ECS/Components/MeshComponent.hpp b/include/public/ECS/Components/MeshComponent.hpp index 1da11242..e911a11e 100644 --- a/include/public/ECS/Components/MeshComponent.hpp +++ b/include/public/ECS/Components/MeshComponent.hpp @@ -8,10 +8,10 @@ #include namespace Sleak { - class MeshData; - namespace RenderEngine { - class TransformBuffer; - class BufferBase; +struct MeshData; +namespace RenderEngine { +class TransformBuffer; +class BufferBase; } // namespace RenderEngine /// Untyped payload blob, currently unused by any buffer path. diff --git a/src/Assets/ModelLoader.cpp b/src/Assets/ModelLoader.cpp index b599a029..1c659264 100644 --- a/src/Assets/ModelLoader.cpp +++ b/src/Assets/ModelLoader.cpp @@ -1,31 +1,145 @@ -#include +#include +#include +#include #include +#include +#include +#include +#include #include #include -#include -#include -#include -#include -#include -#include +#include #include #include -#include -#include - +#include +#include #include +#include +#include +#include +#include +#include +#include + #include "../../include/private/Graphics/Common/ResourceManager.hpp" -#include -#include -#include +namespace Sleak { -#include +namespace { -#include +struct EmbeddedDecode { + const aiTexture* source = nullptr; + std::vector cacheKeys; + unsigned char* pixels = nullptr; + int width = 0; + int height = 0; +}; -namespace Sleak { +void CollectUsedMaterials(const aiNode* node, const aiScene* scene, + std::vector& used) { + for (unsigned int i = 0; i < node->mNumMeshes; ++i) { + const unsigned int mat = + scene->mMeshes[node->mMeshes[i]]->mMaterialIndex; + if (mat < used.size()) used[mat] = true; + } + for (unsigned int i = 0; i < node->mNumChildren; ++i) + CollectUsedMaterials(node->mChildren[i], scene, used); +} + +/// Decodes the compressed embedded textures the scene's materials use on the +/// job system and seeds textureCache with them, so LoadMaterialTexture finds +/// them already uploaded. Uploads stay on the calling thread. +void PredecodeEmbeddedTextures(const aiScene* scene, + TextureCache& textureCache) { + std::vector used(scene->mNumMaterials, false); + CollectUsedMaterials(scene->mRootNode, scene, used); + + std::vector decodes; + auto addSlot = [&](aiMaterial* mat, aiTextureType type) { + if (mat->GetTextureCount(type) == 0) return; + aiString aiPath; + mat->GetTexture(type, 0, &aiPath); + const aiTexture* embedded = scene->GetEmbeddedTexture(aiPath.C_Str()); + if (!embedded || embedded->mHeight != 0) return; + + std::string cacheKey = + std::string(aiPath.C_Str()) + ":" + std::to_string(type); + auto it = std::find_if( + decodes.begin(), decodes.end(), + [&](const EmbeddedDecode& d) { return d.source == embedded; }); + if (it == decodes.end()) { + decodes.push_back({}); + it = decodes.end() - 1; + it->source = embedded; + } + if (std::find(it->cacheKeys.begin(), it->cacheKeys.end(), cacheKey) == + it->cacheKeys.end()) + it->cacheKeys.push_back(std::move(cacheKey)); + }; + + for (unsigned int m = 0; m < scene->mNumMaterials; ++m) { + if (!used[m]) continue; + aiMaterial* mat = scene->mMaterials[m]; + addSlot(mat, mat->GetTextureCount(aiTextureType_DIFFUSE) > 0 + ? aiTextureType_DIFFUSE + : aiTextureType_BASE_COLOR); + addSlot(mat, aiTextureType_NORMALS); + addSlot(mat, aiTextureType_SPECULAR); + addSlot(mat, aiTextureType_EMISSIVE); + addSlot(mat, aiTextureType_METALNESS); + addSlot(mat, aiTextureType_DIFFUSE_ROUGHNESS); + } + if (decodes.empty()) return; + + // Decode in waves so only a handful of images are held in memory at once + JobSystem* jobs = JobSystem::GetInstance(); + const uint32_t wave = jobs ? jobs->GetWorkerCount() + 1 : 1; + + for (size_t first = 0; first < decodes.size(); first += wave) { + const uint32_t n = static_cast( + std::min(wave, decodes.size() - first)); + + auto decode = [&](uint32_t i) { + EmbeddedDecode& d = decodes[first + i]; + int channels = 0; + d.pixels = stbi_load_from_memory( + reinterpret_cast(d.source->pcData), + static_cast(d.source->mWidth), &d.width, &d.height, + &channels, 4); + }; + if (jobs) + jobs->ParallelFor(n, 1, decode); + else + for (uint32_t i = 0; i < n; ++i) decode(i); + + for (uint32_t i = 0; i < n; ++i) { + EmbeddedDecode& d = decodes[first + i]; + for (const std::string& key : d.cacheKeys) { + const std::string texPath = key.substr(0, key.rfind(':')); + ::Sleak::Texture* tex = nullptr; + if (!d.pixels) { + SLEAK_ERROR(" Failed to decode embedded texture: {}", + texPath); + } else { + tex = + RenderEngine::ResourceManager::CreateTextureFromMemory( + d.pixels, static_cast(d.width), + static_cast(d.height), + TextureFormat::RGBA8); + if (tex) + SLEAK_INFO(" Loaded embedded texture: {} ({}x{})", + texPath, d.width, d.height); + } + textureCache[key] = RefPtr<::Sleak::Texture>(tex); + } + stbi_image_free(d.pixels); + d.pixels = nullptr; + } + } +} + +} // namespace /// Recursively initialize a GameObject and all its children. static void InitializeRecursive(GameObject* obj) { @@ -95,6 +209,7 @@ GameObject* ModelLoader::Load(const std::string& filePath, return nullptr; } + PredecodeEmbeddedTextures(scene, textureCache); ProcessNode(scene->mRootNode, scene, root, directory, options, textureCache, materialCache); } else { @@ -105,6 +220,7 @@ GameObject* ModelLoader::Load(const std::string& filePath, SLEAK_INFO(" Skeleton: {} bones, {} animation clips", skeleton->GetBoneCount(), clips.size()); + PredecodeEmbeddedTextures(scene, textureCache); ProcessNodeAnimated(scene->mRootNode, scene, root, directory, options, textureCache, materialCache, skeleton, clips); } @@ -633,4 +749,4 @@ std::vector ModelLoader::LoadAnimationsOnly( return result; } -} // namespace Sleak +} // namespace Sleak diff --git a/src/Core/Application.cpp b/src/Core/Application.cpp index 83c0eae5..8f85a7ef 100644 --- a/src/Core/Application.cpp +++ b/src/Core/Application.cpp @@ -1,38 +1,39 @@ #include "../../include/public/Core/Application.hpp" -#include "../../include/public/Core/CommandLine.hpp" -#include "../../include/private/Graphics/Common/RendererFactory.hpp" -#include "../../include/private/Graphics/Common/RenderCommandQueue.hpp" -#include -#include -#include -#include -#include -#include -#include -#include "Graphics/Vulkan/VulkanRenderer.hpp" -#include "Core/Logger.hpp" -#include #include -#include #include - #include -#include -#include -#include +#include #include -#include -#include +#include +#include +#include #include +#include +#include +#include +#include +#include +#include +#include #include +#include +#include +#include +#include +#include +#include +#include + +#include "../../include/private/Graphics/Common/RenderCommandQueue.hpp" +#include "../../include/private/Graphics/Common/RendererFactory.hpp" +#include "../../include/public/Core/CommandLine.hpp" +#include "Core/Logger.hpp" +#include "ECS/Components/FirstPersonController.hpp" +#include "ECS/Components/FreeLookCameraController.hpp" #include "ECS/Components/MeshComponent.hpp" #include "ECS/Components/TransformComponent.hpp" -#include "ECS/Components/FreeLookCameraController.hpp" -#include "ECS/Components/FirstPersonController.hpp" -#include -#include -#include +#include "Graphics/Vulkan/VulkanRenderer.hpp" using namespace Sleak; using namespace Sleak::Math; @@ -54,6 +55,7 @@ namespace Sleak { throw std::runtime_error("The Application is already running!"); } Instance = this; + m_jobSystem = new JobSystem(); // Read all settings from CommandLine (parsed in main before Application) { @@ -108,6 +110,8 @@ namespace Sleak { if (renderer) renderer->WaitIdle(); delete Game; + delete m_jobSystem; + m_jobSystem = nullptr; Sleak::UI::ShutdownTextureCache(); // frees cached VkImage/memory pre-device-teardown Sleak::MeshBatch::Shutdown(); delete m_benchmark; diff --git a/src/Core/JobSystem.cpp b/src/Core/JobSystem.cpp new file mode 100644 index 00000000..f55a2a14 --- /dev/null +++ b/src/Core/JobSystem.cpp @@ -0,0 +1,142 @@ +#include +#include +#include +#include + +namespace Sleak { + +JobSystem* JobSystem::Instance = nullptr; + +JobSystem::JobSystem(uint32_t workerCount) { + if (workerCount == 0) { + const uint32_t hw = std::thread::hardware_concurrency(); + workerCount = hw > 1 ? hw - 1 : 1; + } + + m_workers.reserve(workerCount); + for (uint32_t i = 0; i < workerCount; ++i) + m_workers.emplace_back(&JobSystem::WorkerLoop, this); + + if (!Instance) Instance = this; + SLEAK_INFO("Job system started with {} workers", workerCount); +} + +JobSystem::~JobSystem() { + { + std::lock_guard lock(m_mutex); + m_stopping = true; + } + m_wake.notify_all(); + for (auto& worker : m_workers) worker.join(); + + if (Instance == this) Instance = nullptr; +} + +uint32_t JobSystem::AllocCounter(uint32_t pending) { + uint32_t index; + if (!m_freeCounters.empty()) { + index = m_freeCounters.back(); + m_freeCounters.pop_back(); + } else { + index = static_cast(m_counters.size()); + m_counters.push_back({0, 1}); + } + m_counters[index].pending = pending; + return index; +} + +void JobSystem::Push(std::function fn, uint32_t counter) { + m_queue.push_back({std::move(fn), counter}); +} + +JobHandle JobSystem::Submit(std::function fn) { + JobHandle handle; + { + std::lock_guard lock(m_mutex); + handle.index = AllocCounter(1); + handle.generation = m_counters[handle.index].generation; + Push(std::move(fn), handle.index); + } + m_wake.notify_one(); + return handle; +} + +bool JobSystem::RunOne(std::unique_lock& lock) { + if (m_queue.empty()) return false; + + Job job = std::move(m_queue.front()); + m_queue.pop_front(); + + lock.unlock(); + try { + job.fn(); + } catch (const std::exception& e) { + SLEAK_ERROR("Job threw an exception: {}", e.what()); + } catch (...) { + SLEAK_ERROR("Job threw an unknown exception"); + } + job.fn = nullptr; + lock.lock(); + + Counter& counter = m_counters[job.counter]; + if (--counter.pending == 0) { + if (++counter.generation == 0) counter.generation = 1; + m_freeCounters.push_back(job.counter); + m_wake.notify_all(); + } + return true; +} + +void JobSystem::WorkerLoop() { + std::unique_lock lock(m_mutex); + while (true) { + if (RunOne(lock)) continue; + if (m_stopping) break; + m_wake.wait(lock); + } +} + +bool JobSystem::IsDone(JobHandle handle) const { + std::lock_guard lock(m_mutex); + return handle.generation == 0 || handle.index >= m_counters.size() || + m_counters[handle.index].generation != handle.generation; +} + +void JobSystem::Wait(JobHandle handle) { + std::unique_lock lock(m_mutex); + if (handle.generation == 0 || handle.index >= m_counters.size()) return; + + while (m_counters[handle.index].generation == handle.generation) { + if (RunOne(lock)) continue; + m_wake.wait(lock); + } + // a push notification may have woken this waiter instead of a worker + if (!m_queue.empty()) m_wake.notify_one(); +} + +void JobSystem::ParallelFor(uint32_t count, uint32_t batchSize, + const std::function& fn) { + if (count == 0) return; + batchSize = std::max(batchSize, 1u); + const uint32_t batches = (count - 1) / batchSize + 1; + + JobHandle handle; + { + std::lock_guard lock(m_mutex); + handle.index = AllocCounter(batches); + handle.generation = m_counters[handle.index].generation; + for (uint32_t b = 0; b < batches; ++b) { + const uint32_t begin = b * batchSize; + const uint32_t end = begin + std::min(batchSize, count - begin); + Push( + [&fn, begin, end] { + for (uint32_t i = begin; i < end; ++i) fn(i); + }, + handle.index); + } + } + m_wake.notify_all(); + Wait(handle); +} + +} // namespace Sleak