From b94a44c7a4591ba330377cec7c9b8525a1550b8b Mon Sep 17 00:00:00 2001 From: legatt Date: Thu, 19 Feb 2026 16:18:01 +0300 Subject: [PATCH] Workers optimization and refactor --- engine/modules/AssetManager.cppm | 18 ++++---- engine/modules/Client.cppm | 4 +- engine/src/Client.cpp | 2 +- engine/src/main.cpp | 4 +- utils/modules/worker.cppm | 73 ++++++++++++++++++++++---------- 5 files changed, 65 insertions(+), 36 deletions(-) diff --git a/engine/modules/AssetManager.cppm b/engine/modules/AssetManager.cppm index 79002e5..1c87e11 100644 --- a/engine/modules/AssetManager.cppm +++ b/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 { public: - AssetManager() : ee_utils::ThreadedWorker(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(); } - ~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(command.out->load()); } @@ -167,9 +167,9 @@ export return Get(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) - Deallocate(mapping.second); + for (AssetHandle *&handle: assetsMapping | std::views::values) + Deallocate(handle); assetsMapping.clear(); binaryStorage.clear(); diff --git a/engine/modules/Client.cppm b/engine/modules/Client.cppm index 9249a32..4e1d4ac 100644 --- a/engine/modules/Client.cppm +++ b/engine/modules/Client.cppm @@ -23,7 +23,7 @@ export { public: Client(); - ~Client(); + ~Client() override; void Init(); //void Send(); -> Request(SEND) @@ -45,4 +45,4 @@ export void ViewModelSubject(ee_data::Subject subject); }; -} \ No newline at end of file +} diff --git a/engine/src/Client.cpp b/engine/src/Client.cpp index 0e76143..ced5b23 100644 --- a/engine/src/Client.cpp +++ b/engine/src/Client.cpp @@ -3,7 +3,7 @@ import utils; import views; import Engine; -Client::Client() : ee_utils::ThreadedWorker(L"Client") +Client::Client() : ThreadedWorker(L"Client", 0) { commandsMapping[ClientCommand::SEND] = std::bind_front(&Client::SendCmd, this); } diff --git a/engine/src/main.cpp b/engine/src/main.cpp index f9c14dd..d8119e4 100644 --- a/engine/src/main.cpp +++ b/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 renderIncompletesBuffer; + ee_engine::RendererCommand renderCmdBuffer; while(!glfwWindowShouldClose(window)) { - Engine::renderer->ProcessQueue(renderCmdBuffer, renderIncompletesBuffer); + Engine::renderer->ProcessQueue(renderCmdBuffer); Engine::renderer->Draw(); glfwSwapBuffers(window); diff --git a/utils/modules/worker.cppm b/utils/modules/worker.cppm index c8508fb..f32b036 100644 --- a/utils/modules/worker.cppm +++ b/utils/modules/worker.cppm @@ -9,50 +9,62 @@ export namespace ee_utils { std::uint8_t type = 0; }; - template concept IWorkerCommand = std::is_base_of_v; + template concept IWorkerCommand = std::is_base_of_v; - template + template 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 &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 + { + thread_local std::uint32_t idleCounter = 0; + if (!this->commands.pop(buffer)) { - this->commands.push(command); + idleCounter++; + return idleCounter; } - incompletesBuffer.clear(); + idleCounter = 0; + + do + { + ProcessCommand(buffer); + } while (this->commands.pop(buffer)); + + return 0; } protected: - std::array, maxCmdType + 1> commandsMapping; - // TODO: maybe I should use lockfree::queue ? + // TODO: replace std::function with more optimized ref + std::array, MaxCmdTypeId + 1> commandsMapping; + // TODO: maybe I should use lockfree::queue ? Otherwise make consumer thread explicit for safety boost::lockfree::spsc_queue> 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 concept IThreadedWorker = std::is_base_of_v; - template - class ThreadedWorker : public Worker, public ThreadedWorkerBase + enum class IdleBackoffMode : std::uint8_t + { + DISABLED, + YIELD + }; + + template + class ThreadedWorker : public Worker(IdleBackoffMode)>, public ThreadedWorkerBase { protected: const wchar_t *threadName; - explicit ThreadedWorker(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 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();