From 45ad6f2d3fd1b8fb0cab7e551d0ad95e875e64c4 Mon Sep 17 00:00:00 2001 From: zhenyan121 <3367366583@qq.com> Date: Wed, 1 Jul 2026 13:56:26 +0800 Subject: [PATCH] feat(tools): add PriorityThreadPool with priority scheduling Implement PriorityThreadPool supporting task priorities and FIFO ordering for same priority. Update ClientWorld to use the new pool with explicit priority for chunk operations. Fix ThreadPool stop logic with atomic exchange and remove unnecessary lambda capture. --- include/Cubed/gameplay/client_world.hpp | 4 +- include/Cubed/tools/priority_thread_pool.hpp | 151 +++++++++++++++++++ include/Cubed/tools/thread_pool.hpp | 6 +- src/gameplay/client_world.cpp | 6 +- 4 files changed, 160 insertions(+), 7 deletions(-) create mode 100644 include/Cubed/tools/priority_thread_pool.hpp diff --git a/include/Cubed/gameplay/client_world.hpp b/include/Cubed/gameplay/client_world.hpp index efcf8f2..f5b96f5 100644 --- a/include/Cubed/gameplay/client_world.hpp +++ b/include/Cubed/gameplay/client_world.hpp @@ -5,7 +5,7 @@ #include "Cubed/gameplay/client_player.hpp" #include "Cubed/gameplay/game_time.hpp" #include "Cubed/gameplay/network_client.hpp" -#include "Cubed/tools/thread_pool.hpp" +#include "Cubed/tools/priority_thread_pool.hpp" #include #include @@ -122,7 +122,7 @@ private: std::shared_ptr m_client; ChunkLoadStyle m_chunk_load_style{ChunkLoadStyle::CENTER}; - std::atomic> m_thread_pool; + std::atomic> m_thread_pool; void client_run(std::stop_token token); diff --git a/include/Cubed/tools/priority_thread_pool.hpp b/include/Cubed/tools/priority_thread_pool.hpp new file mode 100644 index 0000000..a1059ce --- /dev/null +++ b/include/Cubed/tools/priority_thread_pool.hpp @@ -0,0 +1,151 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +namespace Cubed { +class PriorityThreadPool { +private: + struct Task { + int priority = 10; + std::uint64_t sequence; + std::function task; + Task(int p, std::uint64_t seq, std::function t) + : priority(p), sequence(seq), task(std::move(t)) {} + }; + + struct TaskCompare { + bool operator()(const Task& a, const Task& b) const { + + if (a.priority != b.priority) { + return a.priority > b.priority; + } + + return a.sequence > b.sequence; + } + }; + + std::vector m_workers; + std::priority_queue, TaskCompare> m_tasks; + std::mutex m_mtx; + std::condition_variable_any m_cv; + std::atomic m_stopping{false}; + std::atomic m_thread_sum{0}; + std::atomic_uint64_t m_sequence{0}; + +public: + PriorityThreadPool(const PriorityThreadPool&) = delete; + PriorityThreadPool(PriorityThreadPool&&) = delete; + PriorityThreadPool& operator=(const PriorityThreadPool&) = delete; + PriorityThreadPool& operator=(PriorityThreadPool&&) = delete; + explicit PriorityThreadPool(size_t thread_sum) : m_thread_sum(thread_sum) { + for (size_t i = 0; i < thread_sum; i++) { + m_workers.emplace_back([this](std::stop_token stoken) { + while (true) { + std::function task; + { + std::unique_lock lock(m_mtx); + m_cv.wait(lock, stoken, + [this] { return !m_tasks.empty(); }); + if (stoken.stop_requested() && m_tasks.empty()) { + return; + } + task = std::move(m_tasks.top().task); + m_tasks.pop(); + } + task(); + } + }); + } + } + ~PriorityThreadPool() { stop(); } + template auto enqueue(int priority, F&& f) { + + using R = std::invoke_result_t; + + auto task = + std::make_shared>(std::forward(f)); + auto fut = task->get_future(); + + { + std::lock_guard lock(m_mtx); + if (m_stopping) + throw std::runtime_error("thread pool stopped"); + m_tasks.emplace(priority, m_sequence++, [task] { (*task)(); }); + } + m_cv.notify_one(); + return fut; + } + + template auto enqueue(F&& f) { + return enqueue(10, std::forward(f)); + } + + void stop() { + if (m_stopping.exchange(true)) { + return; + } + + for (auto& w : m_workers) { + w.request_stop(); + } + + m_cv.notify_all(); + + for (auto& w : m_workers) { + if (w.joinable()) { + w.join(); + } + } + } + size_t thread_sum() const { return m_thread_sum.load(); } +}; + +template +void parallel_do(PriorityThreadPool& pool, Iter first, Iter last, + size_t max_threads, F&& f) { + max_threads = std::max(1, max_threads); + max_threads = std::min(max_threads, pool.thread_sum()); + std::decay_t fn(std::forward(f)); + size_t length = std::distance(first, last); + if (!length) { + return; + } + + constexpr size_t MIN_PER_THREAD = 25; + size_t num_blocks = + std::min(max_threads, (length + MIN_PER_THREAD - 1) / MIN_PER_THREAD); + num_blocks = std::max(1, num_blocks); + size_t block_size = (length + num_blocks - 1) / num_blocks; + + std::vector> futures; + futures.reserve(num_blocks - 1); + Iter block_start = first; + for (size_t i = 0; i < num_blocks - 1; ++i) { + Iter block_end = block_start; + auto remain = std::distance(block_start, last); + std::advance(block_end, std::min(block_size, remain)); + + futures.emplace_back(pool.enqueue([block_start, block_end, &fn]() { + for (auto it = block_start; it != block_end; ++it) { + fn(*it); + } + })); + + block_start = block_end; + } + for (auto it = block_start; it != last; ++it) { + fn(*it); + } + + for (auto& fut : futures) { + fut.get(); + } +}; + +} // namespace Cubed diff --git a/include/Cubed/tools/thread_pool.hpp b/include/Cubed/tools/thread_pool.hpp index 5b53eaa..d55c009 100644 --- a/include/Cubed/tools/thread_pool.hpp +++ b/include/Cubed/tools/thread_pool.hpp @@ -31,7 +31,7 @@ public: { std::unique_lock lock(m_mtx); m_cv.wait(lock, stoken, - [this, stoken] { return !m_tasks.empty(); }); + [this] { return !m_tasks.empty(); }); if (stoken.stop_requested() && m_tasks.empty()) { return; } @@ -62,7 +62,9 @@ public: return fut; } void stop() { - m_stopping = true; + if (m_stopping.exchange(true)) { + return; + } for (auto& w : m_workers) { w.request_stop(); } diff --git a/src/gameplay/client_world.cpp b/src/gameplay/client_world.cpp index 4f62815..633eb83 100644 --- a/src/gameplay/client_world.cpp +++ b/src/gameplay/client_world.cpp @@ -171,7 +171,7 @@ void ClientWorld::set_block(const glm::ivec3& block_pos, unsigned id) { auto pool = m_thread_pool.load(); - pool->enqueue([this, pos]() { + pool->enqueue(0, [this, pos]() { std::shared_ptr chunk; { @@ -221,7 +221,7 @@ void ClientWorld::set_block(const glm::ivec3& block_pos, unsigned id) { } for (auto& npos : nposes) { - pool->enqueue([this, npos]() { + pool->enqueue(0, [this, npos]() { std::shared_ptr chunk; { @@ -388,7 +388,7 @@ void ClientWorld::change_pool_threads(int threads) { } int used_thread = std::clamp(threads, 1, m_max_threads); Logger::info("Create New Thread Pool Use {} Threads", used_thread); - m_thread_pool.store(std::make_shared(used_thread)); + m_thread_pool.store(std::make_shared(used_thread)); } void ClientWorld::hot_reload() {