mirror of
https://github.com/zhenyan121/Cubed.git
synced 2026-08-09 02:07:04 +08:00
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.
This commit is contained in:
@@ -5,7 +5,7 @@
|
|||||||
#include "Cubed/gameplay/client_player.hpp"
|
#include "Cubed/gameplay/client_player.hpp"
|
||||||
#include "Cubed/gameplay/game_time.hpp"
|
#include "Cubed/gameplay/game_time.hpp"
|
||||||
#include "Cubed/gameplay/network_client.hpp"
|
#include "Cubed/gameplay/network_client.hpp"
|
||||||
#include "Cubed/tools/thread_pool.hpp"
|
#include "Cubed/tools/priority_thread_pool.hpp"
|
||||||
|
|
||||||
#include <absl/container/flat_hash_set.h>
|
#include <absl/container/flat_hash_set.h>
|
||||||
#include <deque>
|
#include <deque>
|
||||||
@@ -122,7 +122,7 @@ private:
|
|||||||
std::shared_ptr<NetworkClient> m_client;
|
std::shared_ptr<NetworkClient> m_client;
|
||||||
ChunkLoadStyle m_chunk_load_style{ChunkLoadStyle::CENTER};
|
ChunkLoadStyle m_chunk_load_style{ChunkLoadStyle::CENTER};
|
||||||
|
|
||||||
std::atomic<std::shared_ptr<ThreadPool>> m_thread_pool;
|
std::atomic<std::shared_ptr<PriorityThreadPool>> m_thread_pool;
|
||||||
|
|
||||||
void client_run(std::stop_token token);
|
void client_run(std::stop_token token);
|
||||||
|
|
||||||
|
|||||||
151
include/Cubed/tools/priority_thread_pool.hpp
Normal file
151
include/Cubed/tools/priority_thread_pool.hpp
Normal file
@@ -0,0 +1,151 @@
|
|||||||
|
#pragma once
|
||||||
|
|
||||||
|
#include <condition_variable>
|
||||||
|
#include <cstddef>
|
||||||
|
#include <functional>
|
||||||
|
#include <future>
|
||||||
|
#include <mutex>
|
||||||
|
#include <queue>
|
||||||
|
#include <thread>
|
||||||
|
#include <vector>
|
||||||
|
namespace Cubed {
|
||||||
|
class PriorityThreadPool {
|
||||||
|
private:
|
||||||
|
struct Task {
|
||||||
|
int priority = 10;
|
||||||
|
std::uint64_t sequence;
|
||||||
|
std::function<void()> task;
|
||||||
|
Task(int p, std::uint64_t seq, std::function<void()> 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<std::jthread> m_workers;
|
||||||
|
std::priority_queue<Task, std::vector<Task>, TaskCompare> m_tasks;
|
||||||
|
std::mutex m_mtx;
|
||||||
|
std::condition_variable_any m_cv;
|
||||||
|
std::atomic<bool> m_stopping{false};
|
||||||
|
std::atomic<size_t> 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<void()> 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 <typename F> auto enqueue(int priority, F&& f) {
|
||||||
|
|
||||||
|
using R = std::invoke_result_t<F>;
|
||||||
|
|
||||||
|
auto task =
|
||||||
|
std::make_shared<std::packaged_task<R()>>(std::forward<F>(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 <typename F> auto enqueue(F&& f) {
|
||||||
|
return enqueue(10, std::forward<F>(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 <std::random_access_iterator Iter, typename F>
|
||||||
|
void parallel_do(PriorityThreadPool& pool, Iter first, Iter last,
|
||||||
|
size_t max_threads, F&& f) {
|
||||||
|
max_threads = std::max<size_t>(1, max_threads);
|
||||||
|
max_threads = std::min(max_threads, pool.thread_sum());
|
||||||
|
std::decay_t<F> fn(std::forward<F>(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<size_t>(1, num_blocks);
|
||||||
|
size_t block_size = (length + num_blocks - 1) / num_blocks;
|
||||||
|
|
||||||
|
std::vector<std::future<void>> 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<size_t>(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
|
||||||
@@ -31,7 +31,7 @@ public:
|
|||||||
{
|
{
|
||||||
std::unique_lock lock(m_mtx);
|
std::unique_lock lock(m_mtx);
|
||||||
m_cv.wait(lock, stoken,
|
m_cv.wait(lock, stoken,
|
||||||
[this, stoken] { return !m_tasks.empty(); });
|
[this] { return !m_tasks.empty(); });
|
||||||
if (stoken.stop_requested() && m_tasks.empty()) {
|
if (stoken.stop_requested() && m_tasks.empty()) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
@@ -62,7 +62,9 @@ public:
|
|||||||
return fut;
|
return fut;
|
||||||
}
|
}
|
||||||
void stop() {
|
void stop() {
|
||||||
m_stopping = true;
|
if (m_stopping.exchange(true)) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
for (auto& w : m_workers) {
|
for (auto& w : m_workers) {
|
||||||
w.request_stop();
|
w.request_stop();
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -171,7 +171,7 @@ void ClientWorld::set_block(const glm::ivec3& block_pos, unsigned id) {
|
|||||||
|
|
||||||
auto pool = m_thread_pool.load();
|
auto pool = m_thread_pool.load();
|
||||||
|
|
||||||
pool->enqueue([this, pos]() {
|
pool->enqueue(0, [this, pos]() {
|
||||||
std::shared_ptr<ClientChunk> chunk;
|
std::shared_ptr<ClientChunk> chunk;
|
||||||
|
|
||||||
{
|
{
|
||||||
@@ -221,7 +221,7 @@ void ClientWorld::set_block(const glm::ivec3& block_pos, unsigned id) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
for (auto& npos : nposes) {
|
for (auto& npos : nposes) {
|
||||||
pool->enqueue([this, npos]() {
|
pool->enqueue(0, [this, npos]() {
|
||||||
std::shared_ptr<ClientChunk> chunk;
|
std::shared_ptr<ClientChunk> chunk;
|
||||||
|
|
||||||
{
|
{
|
||||||
@@ -388,7 +388,7 @@ void ClientWorld::change_pool_threads(int threads) {
|
|||||||
}
|
}
|
||||||
int used_thread = std::clamp(threads, 1, m_max_threads);
|
int used_thread = std::clamp(threads, 1, m_max_threads);
|
||||||
Logger::info("Create New Thread Pool Use {} Threads", used_thread);
|
Logger::info("Create New Thread Pool Use {} Threads", used_thread);
|
||||||
m_thread_pool.store(std::make_shared<ThreadPool>(used_thread));
|
m_thread_pool.store(std::make_shared<PriorityThreadPool>(used_thread));
|
||||||
}
|
}
|
||||||
|
|
||||||
void ClientWorld::hot_reload() {
|
void ClientWorld::hot_reload() {
|
||||||
|
|||||||
Reference in New Issue
Block a user