feat: add a job system owned by Application

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

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;
 

include/public/Core/JobSystem.hpp

@@ -0,0 +1,106 @@
+#ifndef _JOBSYSTEM_HPP_
+#define _JOBSYSTEM_HPP_
+
+#include <Core/OSDef.hpp>
+#include <condition_variable>
+#include <cstdint>
+#include <deque>
+#include <functional>
+#include <mutex>
+#include <thread>
+#include <vector>
+
+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<void()> 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<void(uint32_t)>& fn);
+
+    uint32_t GetWorkerCount() const {
+        return static_cast<uint32_t>(m_workers.size());
+    }
+
+    /// The pool Application created, or null outside of an Application.
+    static JobSystem* GetInstance() { return Instance; }
+
+   private:
+    struct Job {
+        std::function<void()> fn;
+        uint32_t counter;
+    };
+
+    struct Counter {
+        uint32_t pending = 0;
+        uint32_t generation = 0;
+    };
+
+    uint32_t AllocCounter(uint32_t pending);
+    void Push(std::function<void()> fn, uint32_t counter);
+    bool RunOne(std::unique_lock<std::mutex>& lock);
+    void WorkerLoop();
+
+    mutable std::mutex m_mutex;
+    std::condition_variable m_wake;
+    std::deque<Job> m_queue;
+    std::vector<Counter> m_counters;
+    std::vector<uint32_t> m_freeCounters;
+    std::vector<std::thread> m_workers;
+    bool m_stopping = false;
+
+    static JobSystem* Instance;
+};
+
+}  // namespace Sleak
+
+#endif  // _JOBSYSTEM_HPP_

include/public/ECS/Components/MeshComponent.hpp

@@ -8,10 +8,10 @@
 #include <Memory/RefPtr.hpp>
 
 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.

src/Assets/ModelLoader.cpp

@@ -1,31 +1,145 @@
-#include <Runtime/ModelLoader.hpp>
+#include <assimp/postprocess.h>
+#include <assimp/scene.h>
+#include <stb_image.h>
 
 #include <Core/GameObject.hpp>
+#include <Core/JobSystem.hpp>
+#include <Core/Logger.hpp>
+#include <ECS/Components/AnimatorComponent.hpp>
+#include <ECS/Components/MaterialComponent.hpp>
 #include <ECS/Components/MeshComponent.hpp>
 #include <ECS/Components/TransformComponent.hpp>
-#include <ECS/Components/MaterialComponent.hpp>
-#include <ECS/Components/AnimatorComponent.hpp>
-#include <Runtime/Material.hpp>
-#include <Runtime/Texture.hpp>
-#include <Runtime/Skeleton.hpp>
-#include <Runtime/AnimationClip.hpp>
+#include <Memory/RefPtr.hpp>
 #include <Physics/ColliderComponent.hpp>
 #include <Physics/RigidbodyComponent.hpp>
-#include <Memory/RefPtr.hpp>
-#include <Core/Logger.hpp>
-
+#include <Runtime/AnimationClip.hpp>
+#include <Runtime/Material.hpp>
 #include <Runtime/MeshData.hpp>
+#include <Runtime/ModelLoader.hpp>
+#include <Runtime/Skeleton.hpp>
+#include <Runtime/Texture.hpp>
+#include <algorithm>
+#include <assimp/Importer.hpp>
+#include <filesystem>
+
 #include "../../include/private/Graphics/Common/ResourceManager.hpp"
 
-#include <assimp/Importer.hpp>
-#include <assimp/scene.h>
-#include <assimp/postprocess.h>
+namespace Sleak {
 
-#include <stb_image.h>
+namespace {
 
-#include <filesystem>
+struct EmbeddedDecode {
+    const aiTexture* source = nullptr;
+    std::vector<std::string> cacheKeys;
+    unsigned char* pixels = nullptr;
+    int width = 0;
+    int height = 0;
+};
 
-namespace Sleak {
+void CollectUsedMaterials(const aiNode* node, const aiScene* scene,
+                          std::vector<bool>& 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<bool> used(scene->mNumMaterials, false);
+    CollectUsedMaterials(scene->mRootNode, scene, used);
+
+    std::vector<EmbeddedDecode> 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<uint32_t>(
+            std::min<size_t>(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<const unsigned char*>(d.source->pcData),
+                static_cast<int>(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<uint32_t>(d.width),
+                            static_cast<uint32_t>(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<AnimationClip*> ModelLoader::LoadAnimationsOnly(
     return result;
 }
 
-} // namespace Sleak
+}  // namespace Sleak