Workers optimization and refactor

engine/modules/AssetManager.cppm

@@ -19,7 +19,7 @@ export
 
     struct AssetHandle
     {
-        inline static constexpr std::uint32_t INVALID_ASSET_ID = UINT32_MAX;
+        static constexpr std::uint32_t INVALID_ASSET_ID = UINT32_MAX;
         std::size_t slot = 0;
         std::uint32_t id = 0;
     };
@@ -96,7 +96,7 @@ export
     class AssetManager final : public ee_utils::ThreadedWorker<AssetStreamCommand, AssetStreamCommand::COMPLETED>
     {
     public:
-        AssetManager() : ee_utils::ThreadedWorker<AssetStreamCommand, AssetStreamCommand::COMPLETED>(L"Assets Stream")
+        AssetManager() : ThreadedWorker(L"Assets Stream", 250)
         {
             commandsMapping[AssetStreamCommand::LOAD] = std::bind_front(&AssetManager::LoadCmd, this);
             commandsMapping[AssetStreamCommand::FIND] = std::bind_front(&AssetManager::FindCmd, this);
@@ -105,7 +105,7 @@ export
             diskSystem = Allocate<NativeFileSystem>();
         }
 
-        ~AssetManager()
+        ~AssetManager() override
         {
             Deallocate(diskSystem);
         }
@@ -138,7 +138,7 @@ export
             command.out = &result;
 
             Request(command);
-            while (command.out->load() == nullptr);
+            while (command.out->load() == nullptr) {}
             return *reinterpret_cast<AssetHandle*>(command.out->load());
         }
 
@@ -167,9 +167,9 @@ export
             return Get<T>(Find(id));
         }
 
-        void Redirect(AssetHandle &handle, void *resource)
+        void Redirect(const AssetHandle &handle, void *resource)
         {
-            AssetInfo &asset = activeAssets[handle.slot];
+            const AssetInfo &asset = activeAssets[handle.slot];
             if (asset.resourceAddress == nullptr)
             {
                 LOG(WARNING, "Attempt to redirect empty handle");
@@ -262,7 +262,7 @@ export
 
         void Release() override
         {
-            for (AssetInfo &asset : activeAssets)
+            for (const AssetInfo &asset : activeAssets)
             {
                 //TODO: LEAK! Free resource
                 if (asset.resourceAddress->resource.load() != nullptr)
@@ -275,8 +275,8 @@ export
             }
             activeAssets.clear();
 
-            for (auto &mapping : assetsMapping)
+            for (AssetHandle *&handle: assetsMapping | std::views::values)
-                Deallocate(mapping.second);
+                Deallocate(handle);
             assetsMapping.clear();
 
             binaryStorage.clear();

engine/modules/Client.cppm

@@ -23,7 +23,7 @@ export
     {
     public:
         Client();
-        ~Client();
+        ~Client() override;
         void Init();
 
         //void Send(); -> Request(SEND)

engine/src/Client.cpp

@@ -3,7 +3,7 @@ import utils;
 import views;
 import Engine;
 
-Client::Client() : ee_utils::ThreadedWorker<ClientCommand, ClientCommand::COMPLETED>(L"Client")
+Client::Client() : ThreadedWorker(L"Client", 0)
 {
     commandsMapping[ClientCommand::SEND] = std::bind_front(&Client::SendCmd, this);
 }

engine/src/main.cpp

@@ -79,10 +79,10 @@ int main(int argc, char **argv)
     Engine::client->Start();
 
     // Game loop
-    ee_engine::RendererCommand renderCmdBuffer; std::vector<ee_engine::RendererCommand> renderIncompletesBuffer;
+    ee_engine::RendererCommand renderCmdBuffer;
     while(!glfwWindowShouldClose(window))
     {
-        Engine::renderer->ProcessQueue(renderCmdBuffer, renderIncompletesBuffer);
+        Engine::renderer->ProcessQueue(renderCmdBuffer);
         Engine::renderer->Draw();
 
         glfwSwapBuffers(window);

utils/modules/worker.cppm

@@ -11,48 +11,60 @@ export namespace ee_utils
     };
     template <typename T> concept IWorkerCommand = std::is_base_of_v<WorkerCommand, T>;
 
-    template <IWorkerCommand TCommand, std::size_t maxCmdType>
+    template <IWorkerCommand TCommand, std::size_t MaxCmdTypeId, bool IdleDetection = false>
     class Worker
     {
     public:
         void Request(const TCommand &command)
         {
-            commands.push(command);
+            commands.push(command); //TODO: filled queue case
         }
 
         void Request(const std::uint8_t type)
         {
-            commands.push(TCommand {.type = type});
+            commands.push(TCommand {.type = type}); //TODO: filled queue case
         }
 
-        void ProcessQueue(TCommand &buffer, std::vector<TCommand> &incompletesBuffer)
+        void ProcessQueue(TCommand &buffer) requires !IdleDetection
         {
             while (this->commands.pop(buffer))
             {
                 ProcessCommand(buffer);
-                if (buffer.type != maxCmdType)
-                    incompletesBuffer.push_back(buffer);
             }
-            for (TCommand &command : incompletesBuffer)
+        }
+
+        std::uint32_t ProcessQueue(TCommand &buffer) requires IdleDetection
         {
-                this->commands.push(command);
+            thread_local std::uint32_t idleCounter = 0;
+            if (!this->commands.pop(buffer))
+            {
+                idleCounter++;
+                return idleCounter;
             }
-            incompletesBuffer.clear();
+            idleCounter = 0;
+
+            do
+            {
+                ProcessCommand(buffer);
+            } while (this->commands.pop(buffer));
+
+            return 0;
         }
 
     protected:
-        std::array<std::function<void(TCommand &command)>, maxCmdType + 1> commandsMapping;
+        // TODO: replace std::function with more optimized ref
-        // TODO: maybe I should use lockfree::queue ?
+        std::array<std::function<void(TCommand &command)>, MaxCmdTypeId + 1> commandsMapping;
+        // TODO: maybe I should use lockfree::queue ? Otherwise make consumer thread explicit for safety
         boost::lockfree::spsc_queue<TCommand, boost::lockfree::capacity<512>> commands;
 
         Worker() {};
 
         void ProcessCommand(TCommand &command)
         {
-            if (command.type > maxCmdType || !commandsMapping[command.type])
+            if (command.type > MaxCmdTypeId || !commandsMapping[command.type])
             {
                 LOG(WARNING, "Unhandled command. Type id: %d", command.type);
-                command.type = maxCmdType;
+                command.type = MaxCmdTypeId;
                 return;
             }
 
@@ -68,7 +80,7 @@ export namespace ee_utils
 
         void Start()
         {
-            if (isRunning)
+            if (isRunning.load())
             {
                 LOG(CRITICAL, "Thread can be started only once before stop");
                 throw std::logic_error("Thread can be started only once before stop");
@@ -83,7 +95,7 @@ export namespace ee_utils
 
         void Shutdown()
         {
-            if (isRunning)
+            if (isRunning.load())
                 thread.request_stop();
             else LOG(WARNING, "Tried to shutdown stopped thread");
         }
@@ -96,15 +108,22 @@ export namespace ee_utils
     };
     template<typename T> concept IThreadedWorker = std::is_base_of_v<ThreadedWorkerBase, T>;
 
-    template <IWorkerCommand TCommand, std::size_t maxCmdType>
+    enum class IdleBackoffMode : std::uint8_t
-    class ThreadedWorker : public Worker<TCommand, maxCmdType>, public ThreadedWorkerBase
+    {
+        DISABLED,
+        YIELD
+    };
+
+    template <IWorkerCommand TCommand, std::size_t MaxCmdType, IdleBackoffMode IdleBackoffMode = IdleBackoffMode::DISABLED>
+    class ThreadedWorker : public Worker<TCommand, MaxCmdType, static_cast<bool>(IdleBackoffMode)>, public ThreadedWorkerBase
     {
     protected:
         const wchar_t *threadName;
 
-        explicit ThreadedWorker<TCommand, maxCmdType>(const wchar_t *threadName)
+        explicit ThreadedWorker(const wchar_t *threadName, std::uint32_t yieldIdleTime = 0)
         {
             this->threadName = threadName;
+            this->yieldIdleTime = yieldIdleTime;
         }
 
         void Threading()
@@ -120,11 +139,19 @@ export namespace ee_utils
         void Loop()
         {
             TCommand commandBuffer{};
-            // ReSharper disable once CppTooWideScope
-            std::vector<TCommand> incompleteCommands;
             while (!stoken->stop_requested())
             {
-                this->ProcessQueue(commandBuffer, incompleteCommands);
+                if constexpr (IdleBackoffMode == IdleBackoffMode::DISABLED)
+                {
+                    this->ProcessQueue(commandBuffer);
+                }
+                else
+                {
+                    const std::uint32_t idleCounter = this->ProcessQueue(commandBuffer);
+
+                    if (idleCounter > yieldIdleTime)
+                        std::this_thread::yield();
+                }
 
                 Update();
             }
@@ -134,6 +161,8 @@ export namespace ee_utils
         virtual void Release() = 0;
 
     private:
+        std::uint32_t yieldIdleTime;
+
         void ReleaseStop()
         {
             Release();