狸村 Mystic Web · 狸村
Home / Vulkan®教程镜像 / Multithreading
目录▶

引言

在本章中，我们将探讨如何利用 Vulkan 的多线程能力来提高应用程序的性能。现代 CPU 拥有多个核心，高效利用这些核心可以显著提升应用程序的性能，尤其是在计算密集型任务中。Vulkan 的显式设计使其非常适合多线程架构，允许对同步和资源访问进行细粒度控制。

概述

Vulkan 在设计时就考虑到了多线程，相比旧 API 具有几个优势：

  1. 线程安全的命令缓冲记录：多个线程可以同时向不同的命令缓冲记录命令。

  2. 显式同步：Vulkan 需要显式同步，让你能精确控制跨线程的资源访问。

  3. 基于队列的架构：不同的操作可以提交到不同的队列，可能并行执行。

然而，Vulkan 中的多线程需要仔细考虑：

  1. 资源共享：确保跨线程对共享资源的安全访问。

  2. 同步：在线程之间正确同步操作。

  3. 工作分配：有效分配工作以最大化并行性。

在本章中，我们将基于之前计算着色器的工作，实现一个多线程渲染系统。我们将创建一个粒子系统，其中：

  1. 一个线程处理窗口事件和呈现

  2. 多个工作线程为不同的粒子组记录命令缓冲

  3. 一个专用线程将工作提交给 GPU

实现

让我们逐步了解在 Vulkan 应用程序中实现多线程所需的关键组件：

线程安全的资源管理

首先，我们需要确保资源在线程间安全访问。我们将结合使用多种技术：

// 线程安全的资源管理器
class ThreadSafeResourceManager {
private:
    std::mutex resourceMutex;
    // 需要线程安全访问的资源
    std::vector<vk::raii::CommandPool> commandPools;
    std::vector<vk::raii::CommandBuffer> commandBuffers;

public:
    // 为每个工作线程创建命令池
    void createThreadCommandPools(vk::raii::Device& device, uint32_t queueFamilyIndex, uint32_t threadCount) {
        std::lock_guard<std::mutex> lock(resourceMutex);

        commandPools.clear();
        for (uint32_t i = 0; i < threadCount; i++) {
            vk::CommandPoolCreateInfo poolInfo{
                .flags = vk::CommandPoolCreateFlagBits::eResetCommandBuffer,
                .queueFamilyIndex = queueFamilyIndex
            };
            commandPools.emplace_back(device, poolInfo);
        }
    }

    // 获取特定线程的命令池
    vk::raii::CommandPool& getCommandPool(uint32_t threadIndex) {
        std::lock_guard<std::mutex> lock(resourceMutex);
        return commandPools[threadIndex];
    }

    // 为每个线程分配命令缓冲
    void allocateCommandBuffers(vk::raii::Device& device, uint32_t threadCount, uint32_t buffersPerThread) {
        std::lock_guard<std::mutex> lock(resourceMutex);

        commandBuffers.clear();
        for (uint32_t i = 0; i < threadCount; i++) {
            vk::CommandBufferAllocateInfo allocInfo{
                .commandPool = *commandPools[i],
                .level = vk::CommandBufferLevel::ePrimary,
                .commandBufferCount = buffersPerThread
            };
            auto threadBuffers = device.allocateCommandBuffers(allocInfo);
            for (auto& buffer : threadBuffers) {
                commandBuffers.emplace_back(std::move(buffer));
            }
        }
    }

    // 获取命令缓冲
    vk::raii::CommandBuffer& getCommandBuffer(uint32_t index) {
        std::lock_guard<std::mutex> lock(resourceMutex);
        return commandBuffers[index];
    }
};

工作线程实现

接下来，我们将实现为不同粒子组记录命令缓冲的工作线程：

class MultithreadedApplication {
private:
    // 线程相关成员
    uint32_t threadCount;
    std::vector<std::thread> workerThreads;
    std::atomic<bool> shouldExit{false};
    std::vector<std::atomic<bool>> threadWorkReady;
    std::vector<std::atomic<bool>> threadWorkDone;

    // 同步原语
    std::mutex queueSubmitMutex;
    std::condition_variable workCompleteCv;

    // 资源管理器
    ThreadSafeResourceManager resourceManager;

    // 粒子系统数据
    struct ParticleGroup {
        uint32_t startIndex;
        uint32_t count;
    };
    std::vector<ParticleGroup> particleGroups;

    // ... 其他 Vulkan 资源 ...

public:
    void initThreads() {
        // 确定要使用的线程数（为主线程保留一个核心）
        threadCount = std::max(1u, std::thread::hardware_concurrency() - 1);

        // 初始化同步原语
        threadWorkReady.resize(threadCount);
        threadWorkDone.resize(threadCount);

        for (uint32_t i = 0; i < threadCount; i++) {
            threadWorkReady[i] = false;
            threadWorkDone[i] = true;
        }

        // 为每个线程创建命令池
        resourceManager.createThreadCommandPools(device, graphicsQueueFamilyIndex, threadCount);

        // 将粒子划分为组，每组对应一个线程
        const uint32_t particlesPerThread = PARTICLE_COUNT / threadCount;
        particleGroups.resize(threadCount);

        for (uint32_t i = 0; i < threadCount; i++) {
            particleGroups[i].startIndex = i * particlesPerThread;
            particleGroups[i].count = (i == threadCount - 1) ?
                (PARTICLE_COUNT - i * particlesPerThread) : particlesPerThread;
        }

        // 启动工作线程
        for (uint32_t i = 0; i < threadCount; i++) {
            workerThreads.emplace_back(&MultithreadedApplication::workerThreadFunc, this, i);
        }
    }

    void workerThreadFunc(uint32_t threadIndex) {
        while (!shouldExit) {
            // 等待工作准备就绪
            if (!threadWorkReady[threadIndex]) {
                std::this_thread::yield();
                continue;
            }

            // 获取此线程的粒子组
            const ParticleGroup& group = particleGroups[threadIndex];

            // 获取此线程的命令缓冲
            vk::raii::CommandBuffer& cmdBuffer = resourceManager.getCommandBuffer(threadIndex);

            // 为此粒子组记录命令
            recordComputeCommandBuffer(cmdBuffer, group.startIndex, group.count);

            // 标记工作完成
            threadWorkDone[threadIndex] = true;
            threadWorkReady[threadIndex] = false;

            // 通知主线程
            workCompleteCv.notify_one();
        }
    }

    void recordComputeCommandBuffer(vk::raii::CommandBuffer& cmdBuffer, uint32_t startIndex, uint32_t count) {
        cmdBuffer.reset();
        cmdBuffer.begin({});

        // 绑定计算管线和描述符集
        cmdBuffer.bindPipeline(vk::PipelineBindPoint::eCompute, *computePipeline);
        cmdBuffer.bindDescriptorSets(vk::PipelineBindPoint::eCompute, *computePipelineLayout, 0, {*computeDescriptorSets[frameIndex]}, {});

        // 添加推送常量以指定此线程的粒子范围
        struct PushConstants {
            uint32_t startIndex;
            uint32_t count;
        } pushConstants{startIndex, count};

        cmdBuffer.pushConstants<PushConstants>(*computePipelineLayout, vk::ShaderStageFlagBits::eCompute, 0, pushConstants);

        // 调度计算工作
        uint32_t groupCount = (count + 255) / 256;
        cmdBuffer.dispatch(groupCount, 1, 1);

        cmdBuffer.end();
    }

    void signalThreadsToWork() {
        // 通知所有线程开始工作
        for (uint32_t i = 0; i < threadCount; i++) {
            threadWorkDone[i] = false;
            threadWorkReady[i] = true;
        }
    }

    void waitForThreadsToComplete() {
        // 等待所有线程完成工作
        std::unique_lock<std::mutex> lock(queueSubmitMutex);
        workCompleteCv.wait(lock, [this]() {
            for (uint32_t i = 0; i < threadCount; i++) {
                if (!threadWorkDone[i]) {
                    return false;
                }
            }
            return true;
        });
    }

    void cleanup() {
        // 通知线程退出并等待
        shouldExit = true;
        for (auto& thread : workerThreads) {
            if (thread.joinable()) {
                thread.join();
            }
        }

        // ... 清理其他资源 ...
    }
};

修改计算着色器

我们需要修改计算着色器，以处理由推送常量指定的粒子范围：

// 在计算着色器中（31_shader_compute.slang）
[[vk::push_constant]]
struct PushConstants {
    uint startIndex;
    uint count;
};

[[vk::binding(0, 0)]] ConstantBuffer<UniformBufferObject> ubo;
[[vk::binding(1, 0)]] RWStructuredBuffer<Particle> particlesIn;
[[vk::binding(2, 0)]] RWStructuredBuffer<Particle> particlesOut;
PushConstants pushConstants;

[numthreads(256,1,1)]
void compMain(uint3 threadId : SV_DispatchThreadID)
{
    uint index = threadId.x;

    // 只处理分配范围内的粒子
    if (index >= pushConstants.count) {
        return;
    }

    // 调整索引以从分配的起始索引开始
    uint globalIndex = pushConstants.startIndex + index;

    // 处理粒子
    Particle particle = particlesIn[globalIndex];

    // 根据速度和增量时间更新粒子位置
    particle.position += particle.velocity * ubo.deltaTime;

    // 简单边界检查，反转速度
    if (abs(particle.position.x) > 1.0) {
        particle.velocity.x *= -1.0;
    }
    if (abs(particle.position.y) > 1.0) {
        particle.velocity.y *= -1.0;
    }

    // 将更新后的粒子写入输出缓冲
    particlesOut[globalIndex] = particle;
}

更新主循环

最后，我们将更新主循环以协调工作线程：

void drawFrame() {
    // 等待上一帧完成
    auto fenceResult = device.waitForFences(*inFlightFences[frameIndex], vk::True, UINT64_MAX);
    if (fenceResult != vk::Result::eSuccess)
    {
        throw std::runtime_error("failed to wait for fence!");
    }

    // 获取下一张图像
    auto [result, imageIndex] = swapChain.acquireNextImage(UINT64_MAX, *imageAvailableSemaphores[frameIndex], nullptr);

    if (result == vk::Result::eErrorOutOfDateKHR || result == vk::Result::eSuboptimalKHR || framebufferResized) {
        framebufferResized = false;
        recreateSwapChain();
        return;
    }

    // 更新统一缓冲
    updateUniformBuffer(frameIndex);

    // 通知工作线程开始记录计算命令缓冲
    signalThreadsToWork();

    // 工作线程忙碌时，主线程记录图形命令缓冲
    recordGraphicsCommandBuffer(imageIndex);

    // 等待所有工作线程完成
    waitForThreadsToComplete();

    // 收集所有线程的命令缓冲
    std::vector<vk::CommandBuffer> computeCmdBuffers;
    for (uint32_t i = 0; i < threadCount; i++) {
        computeCmdBuffers.push_back(*resourceManager.getCommandBuffer(i));
    }

    // 提交计算工作
    vk::SubmitInfo computeSubmitInfo{
        .commandBufferCount = static_cast<uint32_t>(computeCmdBuffers.size()),
        .pCommandBuffers = computeCmdBuffers.data()
    };

    {
        std::lock_guard<std::mutex> lock(queueSubmitMutex);
        computeQueue.submit(computeSubmitInfo, nullptr);
    }

    // 等待计算完成后再进行图形
    vk::PipelineStageFlags waitStages[] = {vk::PipelineStageFlagBits::eVertexInput};

    // 提交图形工作
    vk::SubmitInfo graphisSubmitInfo{.waitSemaphoreCount   = 1,
                                      .pWaitSemaphores      = &*imageAvailableSemaphores[frameIndex],
                                      .pWaitDstStageMask    = &waitDestinationStageMask,
                                      .commandBufferCount   = 1,
                                      .pCommandBuffers      = &*graphicsCommandBuffers[frameIndex],
                                      .signalSemaphoreCount = 1,
                                      .pSignalSemaphores    = &*renderFinishedSemaphores[imageIndex]};

    {
        std::lock_guard<std::mutex> lock(queueSubmitMutex);
        device.resetFences(*inFlightFences[frameIndex]);
        graphicsQueue.submit(graphicsSubmitInfo, *inFlightFences[frameIndex]);
    }

    // 呈现图像
    vk::PresentInfoKHR presentInfo{
        .waitSemaphoreCount = 1,
        .pWaitSemaphores = &*renderFinishedSemaphores[frameIndex],
        .swapchainCount = 1,
        .pSwapchains = &*swapChain,
        .pImageIndices = &imageIndex
    };

    result = presentQueue.presentKHR(presentInfo);

    if (result == vk::Result::eErrorOutOfDateKHR || result == vk::Result::eSuboptimalKHR || framebufferResized) {
        framebufferResized = false;
        recreateSwapChain();
    } else if (result != vk::Result::eSuccess) {
        throw std::runtime_error("failed to present swap chain image!");
    }

    frameIndex = (frameIndex + 1) % MAX_FRAMES_IN_FLIGHT;
}

高级多线程技术

除了上述基本实现之外，还有几种高级技术可以进一步优化多线程 Vulkan 应用程序：

辅助命令缓冲

辅助命令缓冲可以并行记录，然后由主命令缓冲执行：

// 在工作线程中：
vk::CommandBufferInheritanceInfo inheritanceInfo{
    .renderPass = *renderPass,
    .subpass = 0,
    .framebuffer = *framebuffers[imageIndex]
};

vk::CommandBufferBeginInfo beginInfo{
    .flags = vk::CommandBufferUsageFlagBits::eRenderPassContinue,
    .pInheritanceInfo = &inheritanceInfo
};

secondaryCommandBuffer.begin(beginInfo);
// 记录渲染命令...
secondaryCommandBuffer.end();

// 在主线程中：
primaryCommandBuffer.begin({});
primaryCommandBuffer.beginRenderPass(...);
primaryCommandBuffer.executeCommands(secondaryCommandBuffers);
primaryCommandBuffer.endRenderPass();
primaryCommandBuffer.end();

用于动态工作分配的线程池

你可以使用线程池动态分配工作，而不是为每个线程分配固定工作：

class ThreadPool {
private:
    std::vector<std::thread> workers;
    std::queue<std::function<void()>> tasks;
    std::mutex queueMutex;
    std::condition_variable condition;
    bool stop;

public:
    ThreadPool(size_t threads) : stop(false) {
        for (size_t i = 0; i < threads; ++i) {
            workers.emplace_back([this] {
                while (true) {
                    std::function<void()> task;
                    {
                        std::unique_lock<std::mutex> lock(queueMutex);
                        condition.wait(lock, [this] { return stop || !tasks.empty(); });
                        if (stop && tasks.empty()) {
                            return;
                        }
                        task = std::move(tasks.front());
                        tasks.pop();
                    }
                    task();
                }
            });
        }
    }

    template<class F>
    void enqueue(F&& f) {
        {
            std::unique_lock<std::mutex> lock(queueMutex);
            tasks.emplace(std::forward<F>(f));
        }
        condition.notify_one();
    }

    ~ThreadPool() {
        {
            std::unique_lock<std::mutex> lock(queueMutex);
            stop = true;
        }
        condition.notify_all();
        for (std::thread& worker : workers) {
            worker.join();
        }
    }
};

异步资源加载

你可以使用多线程异步加载资源：

std::future<TextureData> loadTextureAsync(const std::string& filename) {
    return std::async(std::launch::async, [filename]() {
        TextureData data;
        // 从文件加载纹理数据
        return data;
    });
}

// 在代码中：
auto textureDataFuture = loadTextureAsync("texture.ktx");
// 执行其他工作...
TextureData textureData = textureDataFuture.get(); // 如果需要则等待完成
// 从加载的数据创建 Vulkan 纹理

性能考虑

在 Vulkan 中实现多线程时，请记住以下性能考虑：

  1. 线程创建开销：创建线程有开销，因此应在启动时创建一次，而不是每帧创建。

  2. 工作粒度：确保每个线程有足够的工作量来抵消线程开销。

  3. 伪共享：当多个线程访问相邻内存时，注意缓存行争用。

  4. 队列提交：队列提交应同步以避免竞争条件。

  5. 内存屏障：正确使用内存屏障以确保跨线程内存操作的可见性。

  6. 每线程命令池：每个线程应有自己的命令池以避免同步开销。

  7. 测量性能：始终测量以确保你的多线程确实提高了性能。

调试多线程 Vulkan 应用程序

调试多线程应用程序可能具有挑战性。以下是一些提示：

  1. 验证层：启用 Vulkan 验证层以捕获同步问题。

  2. 线程检测器：使用 ThreadSanitizer 等工具检测数据竞争。

  3. 日志记录：实现线程安全的日志记录以跟踪执行流程。

  4. 简化：从更简单的线程模型开始，逐步增加复杂性。

  5. 原子操作：使用原子操作实现线程安全的计数器和标志。

结论

在本章中，我们探讨了如何利用 Vulkan 的多线程来提高性能。我们实现了一个多线程粒子系统，其中：

  1. 多个工作线程并行记录命令缓冲

  2. 主线程协调工作并处理呈现

  3. 正确的同步确保线程安全

通过在多个 CPU 核心上分配工作，我们可以显著提高性能，尤其是对于计算密集型应用程序。Vulkan 的显式设计使其非常适合多线程架构，允许对同步和资源访问进行细粒度控制。

在你继续开发 Vulkan 应用程序时，请考虑多线程如何帮助你充分利用现代 CPU 的全部能力，并记住始终测量性能，以确保你的线程模型对你的特定用例确实有益。

C++ 代码

Vulkan 是 Khronos Group Inc. 的注册商标

教程内容版权归原作者，遵循 CC BY-SA 4.0；本站独立代码及设计除外。