feat: extend ChunkPipeline

core/src/ChunkPipeline.cpp

@@ -161,3 +161,159 @@ std::vector<ChunkTask> ChunkPipeline::pendingTasks() const {
     return tasks;
 }
 
+void ChunkPipeline::stageGenerate(ChunkTaskResult& slot, Chunk* target,
+                                  std::atomic<bool>& cancelled) {
+    if (cancelled.load(std::memory_order_acquire)) {
+        slot.cancelled = true;
+        return;
+    }
+    auto start = std::chrono::steady_clock::now();
+    generator.generateChunk(*target, slot.coord);
+    slot.stage = ChunkStage::Generate;
+    slot.stageNanos[static_cast<int>(ChunkStage::Generate)] = nanosSince(start);
+}
+
+void ChunkPipeline::stageLight(ChunkTaskResult& slot, std::atomic<bool>& cancelled) {
+    if (cancelled.load(std::memory_order_acquire) || slot.cancelled) {
+        slot.cancelled = true;
+        return;
+    }
+    auto start = std::chrono::steady_clock::now();
+    Vec3i origin{slot.coord.x * chunkSize - 1, slot.coord.y * chunkSize - 1,
+                 slot.coord.z * chunkSize - 1};
+    Vec3i dims{chunkSize + 2, chunkSize + 2, chunkSize + 2};
+    slot.light = computeLight(world, origin, dims);
+    slot.stage = ChunkStage::Light;
+    slot.stageNanos[static_cast<int>(ChunkStage::Light)] = nanosSince(start);
+}
+
+void ChunkPipeline::stageMesh(ChunkTaskResult& slot, std::atomic<bool>& cancelled) {
+    if (cancelled.load(std::memory_order_acquire) || slot.cancelled) {
+        slot.cancelled = true;
+        return;
+    }
+    auto start = std::chrono::steady_clock::now();
+    slot.mesh = meshChunkLit(world, slot.coord, slot.light);
+    slot.stage = ChunkStage::Mesh;
+    slot.stageNanos[static_cast<int>(ChunkStage::Mesh)] = nanosSince(start);
+}
+
+void ChunkPipeline::stageReady(ChunkTaskResult& slot, const Chunk* source,
+                               std::atomic<bool>& cancelled) {
+    if (cancelled.load(std::memory_order_acquire) || slot.cancelled) {
+        slot.cancelled = true;
+        return;
+    }
+    auto start = std::chrono::steady_clock::now();
+    slot.chunk = *source;
+    slot.entryBytes = ChunkCache::entrySizeBytes(slot.chunk, slot.mesh);
+    slot.stage = ChunkStage::Ready;
+    slot.stageNanos[static_cast<int>(ChunkStage::Ready)] = nanosSince(start);
+}
+
+void ChunkPipeline::run() {
+    std::vector<ChunkTask> tasks = pendingTasks();
+    if (tasks.empty()) return;
+
+    std::size_t count = tasks.size();
+    std::vector<std::atomic<bool>*> flags(count, nullptr);
+    {
+        std::lock_guard<std::mutex> lock(controlMutex);
+        activeIndex.clear();
+        cancelFlags.clear();
+        cancelFlags.reserve(count);
+        for (std::size_t i = 0; i < count; ++i) {
+            activeIndex[tasks[i].coord] = i;
+            bool cancelled = !tasks[i].inRange || cancelledCoords.count(tasks[i].coord) != 0;
+            cancelFlags.push_back(std::unique_ptr<std::atomic<bool>>(
+                new std::atomic<bool>(cancelled)));
+            flags[i] = cancelFlags.back().get();
+        }
+    }
+
+    std::vector<ChunkTaskResult> slots(count);
+    std::vector<Chunk*> targets(count, nullptr);
+    std::unordered_map<Vec3i, std::size_t, Vec3iHash> indexOf;
+    indexOf.reserve(count * 2);
+    for (std::size_t i = 0; i < count; ++i) {
+        slots[i].coord = tasks[i].coord;
+        indexOf[tasks[i].coord] = i;
+        if (!flags[i]->load(std::memory_order_acquire))
+            targets[i] = &world.ensureChunk(tasks[i].coord);
+    }
+
+    JobGraph graph;
+    std::vector<std::array<std::size_t, chunkStageCount>> nodes(count);
+    for (std::size_t i = 0; i < count; ++i) {
+        JobPriority priority = tasks[i].priority;
+        std::string label = coordLabel(tasks[i].coord);
+        nodes[i][0] = graph.addNode(label + " generate",
+                                    [this, i, &slots, &targets, &flags] {
+                                        stageGenerate(slots[i], targets[i], *flags[i]);
+                                    },
+                                    priority);
+        nodes[i][1] = graph.addNode(label + " light",
+                                    [this, i, &slots, &flags] {
+                                        stageLight(slots[i], *flags[i]);
+                                    },
+                                    priority);
+        nodes[i][2] = graph.addNode(label + " mesh",
+                                    [this, i, &slots, &flags] {
+                                        stageMesh(slots[i], *flags[i]);
+                                    },
+                                    priority);
+        nodes[i][3] = graph.addNode(label + " ready",
+                                    [this, i, &slots, &targets, &flags] {
+                                        stageReady(slots[i], targets[i], *flags[i]);
+                                    },
+                                    priority);
+    }
+
+    for (std::size_t i = 0; i < count; ++i) {
+        graph.addDependency(nodes[i][1], nodes[i][0]);
+        graph.addDependency(nodes[i][2], nodes[i][1]);
+        graph.addDependency(nodes[i][3], nodes[i][2]);
+        for (int dx = -1; dx <= 1; ++dx) {
+            for (int dy = -1; dy <= 1; ++dy) {
+                for (int dz = -1; dz <= 1; ++dz) {
+                    if (dx == 0 && dy == 0 && dz == 0) continue;
+                    Vec3i neighbour = tasks[i].coord + Vec3i{dx, dy, dz};
+                    auto found = indexOf.find(neighbour);
+                    if (found == indexOf.end()) continue;
+                    graph.addDependency(nodes[i][1], nodes[found->second][0]);
+                    graph.addDependency(nodes[i][2], nodes[found->second][0]);
+                }
+            }
+        }
+    }
+
+    JobBatch batch = jobs.runGraph(graph);
+    batch.wait();
+    jobs.waitAll();
+
+    std::vector<std::size_t> mergeOrder(count);
+    std::iota(mergeOrder.begin(), mergeOrder.end(), std::size_t{0});
+    std::sort(mergeOrder.begin(), mergeOrder.end(), [&tasks](std::size_t a, std::size_t b) {
+        return coordLess(tasks[a].coord, tasks[b].coord);
+    });
+
+    PipelineStats delta;
+    completedResults.clear();
+    resultIndex.clear();
+    completedResults.reserve(count);
+    for (std::size_t index : mergeOrder) {
+        ChunkTaskResult& slot = slots[index];
+        for (int stage = 0; stage < chunkStageCount; ++stage)
+            delta.stageNanos[static_cast<std::size_t>(stage)] += slot.stageNanos[stage];
+        if (slot.cancelled) {
+            ++delta.cancelled;
+            continue;
+        }
+        for (int stage = 0; stage <= static_cast<int>(slot.stage); ++stage)
+            ++delta.stageCompleted[static_cast<std::size_t>(stage)];
+        ++delta.chunksReady;
+        chunkStore.store(slot.coord, slot.chunk, slot.mesh);
+        resultIndex[slot.coord] = completedResults.size();
+        completedResults.push_back(std::move(slot));
+    }
+