feat(world): integrate thread pool and async chunk generation

include/Cubed/gameplay/chunk.hpp

@@ -23,6 +23,8 @@ private:
     std::atomic<bool> m_dirty{false};
     std::atomic<bool> m_need_upload{true};
     std::atomic<bool> m_is_on_gen_vertex_data{false};
+    std::atomic<bool> m_gening{false};
+    std::atomic<bool> m_gen_finish{false};
     std::atomic<BiomeType> m_biome = BiomeType::PLAIN;
     std::mutex m_vertexs_data_mutex;
 
@@ -127,6 +129,7 @@ public:
     // ensure thread safe!
     void gen_chunk();
 
+    bool is_gen_finish() const;
     ChunkPos chunk_pos() const;
     BiomeType biome() const;
     void biome(BiomeType b);

include/Cubed/gameplay/world.hpp

@@ -4,6 +4,7 @@
 #include "Cubed/gameplay/chunk.hpp"
 #include "Cubed/gameplay/game_time.hpp"
 #include "Cubed/gameplay/river_worm.hpp"
+#include "Cubed/tools/thread_pool.hpp"
 
 #include <atomic>
 #include <condition_variable>
@@ -34,15 +35,22 @@ class Player;
 class TextureManager;
 class World {
 private:
+    struct PendingChunk {
+        Chunk chunk;
+        std::future<void> future;
+    };
+
     using OptionalBlockVectorArray =
         std::array<std::optional<std::vector<BlockType>>, 4>;
     using ChunkPtrUpdateList = std::vector<std::pair<ChunkPos, Chunk*>>;
     using ChunkPairVector = std::vector<std::pair<ChunkPos, Chunk>>;
+    using ChunkPairQueue = std::queue<std::pair<ChunkPos, Chunk>>;
     using ConstChunkMap =
         std::unordered_map<ChunkPos, const Chunk*, ChunkPos::Hash>;
     using ChunkPosSet = std::unordered_set<ChunkPos, ChunkPos::Hash>;
     using ChunkHashMap = std::unordered_map<ChunkPos, Chunk, ChunkPos::Hash>;
-
+    using PendingChunkHashMap =
+        std::unordered_map<ChunkPos, PendingChunk, ChunkPos::Hash>;
     glm::vec3 m_gen_player_pos{0.0f, 0.0f, 0.0f};
     ChunkHashMap m_chunks;
     std::unordered_map<std::size_t, Player> m_players;
@@ -50,7 +58,7 @@ private:
 
     std::thread m_gen_thread;
     std::thread m_server_thread;
-
+    std::unique_ptr<ThreadPool> m_gen_thread_pool;
     std::stop_source m_server_stop_source;
 
     std::atomic<int> m_per_tick_time = DEFAULT_PER_TICK_TIME; // ms
@@ -59,7 +67,7 @@ private:
 
     mutable std::mutex m_chunks_mutex;
     std::mutex m_gen_signal_mutex;
-    std::mutex m_new_chunk_queue_mutex;
+    std::mutex m_new_chunk_mutex;
     std::mutex m_delete_vbo_mutex;
     std::mutex m_delete_vao_mutex;
     std::mutex m_gen_player_pos_mutex;
@@ -79,8 +87,9 @@ private:
 
     std::vector<ChunkPos> m_dirty_queue;
     std::vector<ChunkRenderSnapshot> m_render_snapshots;
-    std::vector<std::pair<ChunkPos, Chunk>> m_new_chunk;
+    std::vector<std::pair<ChunkPos, Chunk>> m_new_finished_chunk;
-    std::vector<std::pair<ChunkPos, Chunk>> m_new_chunk_queue;
+    // Can only be used in the gen thread
+    PendingChunkHashMap new_chunks;
 
     CaveCarver m_cave_carcer;
     RiverWorm m_river_worm;
@@ -88,15 +97,13 @@ private:
 
     void gen_chunks_internal();
     void sync_player_pos(glm::vec3& player_pos);
-    void
+    void compute_required_chunks(ChunkPosSet& required_chunks,
-    compute_required_chunks(ChunkPosSet& required_chunks,
+                                 ChunkPairVector& temp_neighbor);
-                            ChunkPairVector& temp_neighbor,
-                            std::vector<ChunkPos>& need_gen_temp_chunks_pos);
     void sync_and_collect_missing_chunks(std::vector<ChunkPos>&,
                                          const ChunkPosSet&);
-    void
+
-    build_neighbor_context_for_new_chunks(ConstChunkMap& new_chunks_neighbor,
+    void submit_new_chunks();
-                                          const ChunkPairVector& new_chunks);
+    void poll_finished_chunks();
 
 public:
     World();

include/Cubed/tools/thread_pool.hpp

@@ -0,0 +1,115 @@
+#pragma once
+
+#include <condition_variable>
+#include <cstddef>
+#include <functional>
+#include <future>
+#include <mutex>
+#include <queue>
+#include <thread>
+#include <vector>
+namespace Cubed {
+class ThreadPool {
+private:
+    std::vector<std::jthread> m_workers;
+    std::queue<std::function<void()>> 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};
+
+public:
+    ThreadPool(const ThreadPool&) = delete;
+    ThreadPool(ThreadPool&&) = delete;
+    ThreadPool& operator=(const ThreadPool&) = delete;
+    ThreadPool& operator=(ThreadPool&&) = delete;
+    explicit ThreadPool(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, stoken] { return !m_tasks.empty(); });
+                        if (stoken.stop_requested() && m_tasks.empty()) {
+                            return;
+                        }
+                        task = std::move(m_tasks.front());
+                        m_tasks.pop();
+                    }
+                    task();
+                }
+            });
+        }
+    }
+    ~ThreadPool() {
+        m_stopping = true;
+        for (auto& w : m_workers) {
+            w.request_stop();
+        }
+        m_cv.notify_all();
+    }
+    template <typename F> auto enqueue(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([task] { (*task)(); });
+        }
+        m_cv.notify_one();
+        return fut;
+    }
+    size_t thread_sum() const { return m_thread_sum.load(); }
+};
+
+template <std::random_access_iterator Iter, typename F>
+void parallel_do(ThreadPool& 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

lsan.supp

@@ -5,3 +5,4 @@ leak:libdecor-gtk.so
 leak:libgtk-3.so
 leak:libwayland-client.so
 leak:libglfw.so
+leak:libEGL_nvidia.so

src/gameplay/chunk.cpp

@@ -456,6 +456,9 @@ void Chunk::gen_cross_plane_vertices(int world_x, int world_y, int world_z,
 }
 
 void Chunk::gen_chunk() {
+    if (m_gening.exchange(true))
+        return;
+    m_gening = true;
     if (m_blocks.size() != 0) {
         Logger::warn(
             "Request Generator Chunk {} {} ,but the Blocks size is Not 0",
@@ -485,8 +488,9 @@ void Chunk::gen_chunk() {
         neightbor_blocks[i] = neighbor[i].get_chunk_blocks();
     }
     gen_vertex_data(neightbor_blocks);
+    m_gen_finish = true;
 }
-
+bool Chunk::is_gen_finish() const { return m_gen_finish.load(); }
 // Logger::info("Cross Sum {}", m_cross_vertices_sum.load());
 
 } // namespace Cubed

src/gameplay/world.cpp

@@ -5,11 +5,10 @@
 #include "Cubed/tools/cubed_assert.hpp"
 #include "Cubed/tools/cubed_hash.hpp"
 
-#include <execution>
 #include <glm/gtc/constants.hpp>
 #include <numbers>
 using namespace std::chrono;
-
+using namespace std::chrono_literals;
 namespace Cubed {
 
 struct ChunkRenderData {
@@ -77,10 +76,14 @@ void World::init_world() {
     m_cave_carcer.init(ChunkGenerator::seed());
     m_river_worm.init(ChunkGenerator::seed());
     m_chunks.reserve(MAX_DISTANCE * MAX_DISTANCE * 4);
+    int max_thread = std::thread::hardware_concurrency();
+    int used_thread = std::max(max_thread - 3, 1);
+    Logger::info("Max Support Thread is {}, use {} threads to gen", max_thread,
+                 used_thread);
+    m_gen_thread_pool = std::make_unique<ThreadPool>(used_thread);
+
     auto t1 = std::chrono::system_clock::now();
 
-    Logger::info("Max Support Thread is {}",
-                 std::thread::hardware_concurrency());
     // init players
     m_players.emplace(HASH::str("TestPlayer"), Player(*this, "TestPlayer"));
 
@@ -122,13 +125,19 @@ ChunkPos World::chunk_pos(int world_x, int world_z) {
 #pragma region ChunkGenerate
 
 void World::gen_chunks_internal() {
+    // Logger::info("gen_chunks_internal");
     m_chunk_gen_fraction = 0.0f;
     m_chunk_gen_finished = false;
+    /*
+    if (!new_chunks.empty()) {
+        submit_new_chunks();
+        return;
+    }*/
+
     ChunkPosSet required_chunks;
     ChunkPairVector temp_neighbor;
-    std::vector<ChunkPos> need_gen_temp_chunks_pos;
+
-    compute_required_chunks(required_chunks, temp_neighbor,
+    compute_required_chunks(required_chunks, temp_neighbor);
-                            need_gen_temp_chunks_pos);
 
     ASSERT_MSG(!required_chunks.empty(), "required chunks is empty!!");
 
@@ -145,45 +154,27 @@ void World::gen_chunks_internal() {
     }
 
     m_chunk_gen_fraction = 0.1f;
-
-    ChunkPairVector new_chunks;
-    ChunkPairVector new_temp_chunks;
     for (auto& pos : need_gen_chunks_pos) {
-        new_chunks.push_back({pos, Chunk(*this, pos)});
+        new_chunks.emplace(pos, Chunk(*this, pos));
     }
-    for (auto& pos : need_gen_temp_chunks_pos) {
+    auto t1 = system_clock::now();
-        new_temp_chunks.push_back({pos, Chunk(*this, pos)});
+    parallel_do(*m_gen_thread_pool, temp_neighbor.begin(), temp_neighbor.end(),
-    }
+                m_gen_thread_pool->thread_sum(),
-    ConstChunkMap new_chunks_neighbor;
-
-    build_neighbor_context_for_new_chunks(new_chunks_neighbor, new_chunks);
-
-    std::for_each(std::execution::par, new_temp_chunks.begin(),
-                  new_temp_chunks.end(),
                 [this](std::pair<ChunkPos, Chunk>& new_chunk) {
                     auto& [pos, chunk] = new_chunk;
                     chunk.gen_phase_one();
                     m_cave_carcer.try_to_add_path(pos, chunk.seed());
                     m_river_worm.try_to_add_path(pos, chunk.seed());
                 });
-
+    auto t2 = system_clock::now();
-    std::for_each(std::execution::par, new_chunks.begin(), new_chunks.end(),
+    Logger::info("Temp Neighbor Add Path Consum {}",
-                  [](std::pair<ChunkPos, Chunk>& new_chunk) {
+                 duration_cast<milliseconds>(t2 - t1));
-                      auto& [pos, chunk] = new_chunk;
-                      chunk.gen_chunk();
-                  });
-
     m_chunk_gen_fraction = 0.9f;
 
-    {
-        std::lock_guard lk(m_new_chunk_queue_mutex);
-        for (auto& x : new_chunks) {
-            m_new_chunk_queue.emplace_back(std::move(x));
-        }
-    }
     m_cave_carcer.cleanup_finished_caves();
     m_river_worm.cleanup_finished_rivers();
     m_chunk_gen_fraction = 1.0f;
+    submit_new_chunks();
     m_chunk_gen_finished = true;
 }
 
@@ -192,9 +183,8 @@ void World::sync_player_pos(glm::vec3& player_pos) {
     player_pos = m_gen_player_pos;
 }
 
-void World::compute_required_chunks(
+void World::compute_required_chunks(ChunkPosSet& required_chunks,
-    ChunkPosSet& required_chunks, ChunkPairVector& temp_neighbor,
+                                    ChunkPairVector& temp_neighbor) {
-    std::vector<ChunkPos>& need_gen_temp_chunks_pos) {
     glm::vec3 player_pos;
     sync_player_pos(player_pos);
 
@@ -212,18 +202,6 @@ void World::compute_required_chunks(
             }
         }
     }
-    int new_radius = radius + 1;
-    int new_r2 = new_radius * new_radius;
-    for (int dx = -new_radius; dx <= new_radius; ++dx) {
-        for (int dz = -new_radius; dz <= new_radius; ++dz) {
-            if (dx * dx + dz * dz <= new_r2) {
-                int nx = chunk_x + dx;
-                int nz = chunk_z + dz;
-
-                need_gen_temp_chunks_pos.push_back({nx, nz});
-            }
-        }
-    }
     int max_path_len = std::max(CavePath::step_max(), RiverPath::step_max());
     radius = max_path_len / 2;
     r2 = radius * radius;
@@ -261,22 +239,34 @@ void World::sync_and_collect_missing_chunks(
     }
 }
 
-void World::build_neighbor_context_for_new_chunks(
+void World::submit_new_chunks() {
-    ConstChunkMap& new_chunks_neighbor, const ChunkPairVector& new_chunks) {
+    std::lock_guard lock(m_new_chunk_mutex);
-    {
+    for (auto& [pos, task] : new_chunks) {
-        std::lock_guard lk(m_chunks_mutex);
+        if (!task.future.valid()) {
-        for (auto& [pos, chunk] : new_chunks) {
+            task.future = m_gen_thread_pool->enqueue(
-            for (auto& dir : CHUNK_DIR) {
+                [&task]() { task.chunk.gen_chunk(); });
-                auto it = m_chunks.find(pos + dir);
-                if (it != m_chunks.end()) {
-                    new_chunks_neighbor.insert({it->first, &(it->second)});
         }
     }
 }
+
+void World::poll_finished_chunks() {
+    m_new_finished_chunk.clear();
+    std::lock_guard lock(m_new_chunk_mutex);
+    std::erase_if(
+        new_chunks, [&](std::pair<const ChunkPos, PendingChunk>& pair) {
+            auto& pending = pair.second;
+            if (!pending.future.valid()) {
+                return false;
             }
-    for (auto& [pos, chunk] : new_chunks) {
+            if (pending.future.wait_for(0ms) != std::future_status::ready) {
-        new_chunks_neighbor.insert({pos, &chunk});
+                return false;
             }
+            pending.future.get();
+
+            m_new_finished_chunk.emplace_back(pair.first,
+                                              std::move(pending.chunk));
+            return true;
+        });
 }
 
 #pragma endregion
@@ -336,10 +326,12 @@ void World::serever_run(std::stop_token stoken) {
 }
 
 void World::need_gen() {
+
     if (!m_could_gen) {
         Logger::warn("It is generating or consuming new chunks");
         return;
     }
+
     m_could_gen = false;
     {
         std::lock_guard lk(m_gen_player_pos_mutex);
@@ -347,6 +339,7 @@ void World::need_gen() {
     }
 
     m_need_gen_chunk = true;
+
     m_gen_cv.notify_one();
 }
 
@@ -414,16 +407,16 @@ BlockType World::get_block_tpye(const glm::ivec3& block_pos) const {
     auto it = m_chunks.find(ChunkPos{chunk_x, chunk_z});
 
     if (it == m_chunks.end()) {
-        Logger::error("Can't Find Block {} {} {}", block_pos.x, block_pos.y,
+        // Logger::error("Can't Find Block {} {} {}", block_pos.x, block_pos.y,
-                      block_pos.z);
+        //               block_pos.z);
         return 0;
     }
     const auto& chunk_blocks = it->second.get_chunk_blocks();
     auto [x, y, z] = Chunk::world_to_block(block_pos, {chunk_x, chunk_z});
     if (x < 0 || y < 0 || z < 0 || x >= CHUNK_SIZE || y >= WORLD_SIZE_Y ||
         z >= CHUNK_SIZE) {
-        Logger::error("Can't Find Block {} {} {}", block_pos.x, block_pos.y,
+        // Logger::error("Can't Find Block {} {} {}", block_pos.x, block_pos.y,
-                      block_pos.z);
+        //               block_pos.z);
         return 0;
     }
     return chunk_blocks[Chunk::index(x, y, z)];
@@ -487,16 +480,9 @@ void World::update(float delta_time) {
         m_pending_delete_vao.clear();
     }
 
-    {
+    poll_finished_chunks();
-        std::scoped_lock lk(m_chunks_mutex, m_new_chunk_queue_mutex);
-        m_new_chunk.clear();
-        for (auto& x : m_new_chunk_queue) {
-            m_new_chunk.emplace_back(std::move(x));
-        }
-        m_new_chunk_queue.clear();
-    }
 
-    for (auto& x : m_new_chunk) {
+    for (auto& x : m_new_finished_chunk) {
         x.second.upload_to_gpu();
     }
 
@@ -505,7 +491,7 @@ void World::update(float delta_time) {
         std::lock_guard lk(m_chunks_mutex);
         bool consumed = false;
 
-        for (auto& x : m_new_chunk) {
+        for (auto& x : m_new_finished_chunk) {
             m_chunks.insert_or_assign(x.first, std::move(x.second));
             consumed = true;
         }
@@ -580,9 +566,9 @@ void World::rebuild_world() {
     m_cave_carcer.reload(ChunkGenerator::seed());
     m_river_worm.reload(ChunkGenerator::seed());
     {
-        std::scoped_lock lk(m_chunks_mutex, m_new_chunk_queue_mutex);
+        std::scoped_lock lk(m_chunks_mutex);
         m_chunks.clear();
-        m_new_chunk_queue.clear();
+        m_new_finished_chunk.clear();
     }
     m_could_gen = true;
     ChunkGenerator::reload();
@@ -592,20 +578,6 @@ void World::rebuild_world() {
     m_is_rebuilding = false;
 }
 
-float World::chunk_gen_fraction() const { return m_chunk_gen_fraction.load(); }
-
-int World::rendering_distance() const { return m_rendering_distance.load(); }
-
-void World::rendering_distance(int rendering_distance) {
-    m_rendering_distance = rendering_distance;
-}
-
-CaveCarver& World::cave_carcer() { return m_cave_carcer; }
-RiverWorm& World::river_worm() { return m_river_worm; }
-std::vector<glm::vec4>& World::planes() { return m_planes; }
-std::vector<ChunkRenderSnapshot>& World::render_snapshots() {
-    return m_render_snapshots;
-};
 /*
 glm::vec3 World::sunlight_dir() const {
     float t = static_cast<float>(m_day_tick) / DAY_TIME;
@@ -640,6 +612,21 @@ glm::vec3 World::sunlight_dir() const {
     return glm::normalize(-dir);
 }
 
+float World::chunk_gen_fraction() const { return m_chunk_gen_fraction.load(); }
+
+int World::rendering_distance() const { return m_rendering_distance.load(); }
+
+void World::rendering_distance(int rendering_distance) {
+    m_rendering_distance = rendering_distance;
+}
+
+CaveCarver& World::cave_carcer() { return m_cave_carcer; }
+RiverWorm& World::river_worm() { return m_river_worm; }
+std::vector<glm::vec4>& World::planes() { return m_planes; }
+std::vector<ChunkRenderSnapshot>& World::render_snapshots() {
+    return m_render_snapshots;
+};
+
 TickType World::game_tick() const { return m_game_ticks.load(); }
 TickType World::day_tick() const { return m_day_tick.load(); }
 void World::day_tick(TickType tick) {