cuda编程笔记（8）--线程束warp

ouliten

1765人浏览 · 2025-07-17 23:14:15

ouliten · 2025-07-17 23:14:15 发布

CUDA 中的 Warp（线程束） 是 GPU 并行计算的最小执行调度单位

Warp 是由 32 个并行线程组成的执行单元，这些线程将以 SIMT（Single Instruction, Multiple Thread） 的方式同步执行同一条指令。

Grid
└── Block（多个）
    └── Warp（每 32 个线程组成一个）
        └── Thread（0 ~ 31）

SIMT 执行模型（Single Instruction Multiple Threads）

Warp 内 32 个线程 同时执行一条相同的指令
但每个线程可以对 不同数据 操作（类似 SIMD）
每次由 warp scheduler 发出一个 warp 执行一个指令周期

分支发散（Divergence）问题

如果 warp 内线程执行了不同的分支语句，会导致性能下降：

if (threadIdx.x % 2 == 0)
    a[threadIdx.x] = 1;
else
    a[threadIdx.x] = 2;

这时 warp 会分两轮执行，实际上还是串行，称为 分支发散（Branch Divergence）。

优化建议：

warp 内线程尽量走相同路径（控制流一致）
使用 warpSize 辅助判断
结构上分组 warp 级逻辑（例如每个 warp 处理一个任务）

为什么有Warp这个概念？

在之前，我们介绍GPU的线程模型的时候，只有Grid-Block-Thread。这个Warp是从何而来的呢？

Warp 是为了描述 GPU 的底层“调度单位和执行行为”。

概念	目的	属于
Block	程序员定义的逻辑并行单元，用于共享 memory / sync	程序模型层
Shared Memory	明确线程间的数据共享 + 手动同步	程序模型层
Warp	GPU 硬件内部的最小调度和执行单位（32 个线程）	硬件执行层

Warp 让你理解性能本质：GPU 是按 32 个线程批处理的

一个 block 可能有 128 或 1024 个线程，但 GPU 是以 32 个线程为单位调度和执行 的：

CUDA 内核 launch 时，会将一个 block 拆成若干个 warp
GPU 的 warp scheduler 一次调度一整个 warp
warp 内线程是 SIMT（Single Instruction Multiple Thread） 执行的

所以如果你不了解 warp：

你无法解释：为什么某些 if 分支会让 kernel 执行变慢（→ branch divergence）
你也不会理解：为什么 __shfl_sync() 比 shared memory 更快（→ warp 内通信）

warp-level 的通信操作比 shared memory 更快、更轻

对比	shared memory	warp shuffle
所有线程共享？	✅ 是的	❌ 仅 warp 内
是否需同步？	✅ 需要 `__syncthreads()`	❌ 不需要
使用内存吗？	✅ 存储在 shared memory	❌ 使用 warp register
开销	相对较高	更快、延迟更低
典型用途	block 级归约、tile GEMM	warp reduce、warp broadcast

例子：warp 级归约只用 5~6 次寄存器交换；shared memory 要写回再同步再读取。

warp-level 原语

在讲解具体函数前，先看四个原语共有的参数：

mask (unsigned int)：32 位的掩码。每一位代表 Warp 中的一个线程（Bit 0 对应 Lane 0）。只有掩码位置为 1 的线程才会参与操作。通常设置为 0xffffffff 表示所有线程参与。
var (T)：当前线程提供的变量。T 可以是 int, float, double 等（32/64位类型）。
width (int)：Shuffle 操作的范围跨度。必须是 2 的幂，且小于等于 warpSize（通常是 32）。如果 width=8，则每 8 个线程组成一个“小队”进行内部分组交换。

__shfl_sync

T __shfl_sync(unsigned mask, T var, int srcLane, int width=warpSize);

参数	类型	含义
`srcLane`	`int`	你希望当前线程从哪一个 lane（0~31）拿值
返回值	`int`	返回从lane线程中拿的值

所有线程调用这个函数后，会拿到 srcLane 线程的提供的 var 值。

int val = threadIdx.x;
int broadcasted = __shfl_sync(0xffffffff, val, 0); // 所有线程获取 lane 0 的值

其他变种：

__shfl_up_sync

T __shfl_up_sync(unsigned mask, T var, unsigned int delta, int width=warpSize);

参数	类型	含义
`delta`	`int`	当前线程获取编号为 `laneid - delta` 的线程中的数据。
返回值	`int`	返回`laneid - delta` 的线程中的数据

边界处理：如果 laneid - delta < 0（即前面没人了），则该线程维持原值。

__shfl_down_sync

T __shfl_up_sync(unsigned mask, T var, unsigned int delta, int width=warpSize);

参数	类型	含义
`delta`	`int`	当前线程获取编号为 `laneid + delta` 的线程中的数据。
返回值	`int`	返回`laneid + delta` 的线程中的数据

边界处理：如果 laneid + delta >= width（即后面没人了），则该线程维持原值。

__shfl_xor_sync

T __shfl_xor_sync(unsigned mask, T var, int laneMask, int width=warpSize);

参数	类型	含义
`laneMask`	`int`	当前线程获取编号为`laneid ^ laneMask`（按位异或）的线程中的数据。
返回值	`int`	返回对应的线程中的数据

关键特性：异或操作是对称的。如果 A 拿了 B 的数据，那么 B 一定同时也拿了 A 的数据。这种蝴蝶式（Butterfly）交换非常适合构建分治算法。

__ballot_sync

unsigned int __ballot_sync(unsigned mask, int predicate);

参数	类型	含义
`mask`	`unsigned`	有效线程掩码（0xFFFFFFFF）
`predicate`	`int`	当前线程的布尔值（非零为 true）

返回值是一个 32-bit 的整数，每一位表示该 lane 上线程是否为 true。

int x = threadIdx.x % 2;
unsigned int bitmask = __ballot_sync(0xFFFFFFFF, x == 0);
// 如果 0、2、4 线程满足，则 bitmask = 0b...101010

`__any_sync()` / `__all_sync()`

在 warp 内判断：是否有 / 是否所有线程满足某条件

int __any_sync(unsigned mask, int predicate);
int __all_sync(unsigned mask, int predicate);

参数	类型	含义
`mask`	`unsigned`	有效线程掩码
`predicate`	`int`	当前线程条件表达式

若 warp 中至少有一个(any)/所有(all) 线程 predicate 为 true返回非零，否则返回 0。

int is_positive = (threadIdx.x > 0);

if (__all_sync(0xFFFFFFFF, is_positive)) {
    // 所有线程都满足条件
}

if (__any_sync(0xFFFFFFFF, is_positive)) {
    // 至少有一个线程满足条件
}

__activemask

unsigned int __activemask(void);

返回当前 warp 中的 有效线程掩码，用于配合 __shfl_sync 等 warp 操作。

__syncwarp

void __syncwarp(unsigned mask = 0xffffffff);

__syncwarp() 是 CUDA 提供的 warp 内同步指令。
它的作用是：

让同一个 warp 中的线程在此处同步执行，保证该 warp 中的所有活跃线程都执行到这条语句之后再继续往下执行。

在旧架构中（比如 Pascal 之前），一个 warp 内的 32 个线程是严格锁步执行的（SIMT lockstep），也就是说：

你不需要任何同步操作，因为 warp 内的所有线程永远执行同一条指令。

但是从 Volta 架构（sm_70）开始，NVIDIA 引入了 Independent Thread Scheduling（独立线程调度）。
这意味着：

同一个 warp 内的线程可以在分支后被不同地调度执行，不再完全锁步。

要不要用，取决于你写的 kernel 是否涉及 warp 内通信或分支。

可以按以下规则判断：

情况	是否需要 `__syncwarp()`	原因
每个线程完全独立，不共享数据	❌ 不需要	warp 内本身执行同步，无依赖
线程之间通过寄存器通信（如 `__shfl_sync()`）	✅ 需要（配合 mask）	确保所有参与线程都同步到同一时刻
线程通过 shared memory 通信（写后读）	✅ 必须	防止乱序执行导致部分线程读到旧数据
warp 内有分支（`if/else`）然后再合流	✅ 建议	Volta 之后线程调度独立，分支可能乱序
warp 内循环中，线程间存在依赖	✅ 必须	防止未同步导致逻辑错误

有些情况，你可以省略它：

使用 __shfl_sync()、__ballot_sync() 这类 API（它们内部自带 warp 同步）；
在旧架构（Pascal 及以前）上运行；
或者确认 warp 内线程之间没有数据依赖（纯计算型）。

示例程序

warp实现规约 vs 共享内存

#ifndef __CUDACC__
#define __CUDACC__
#endif
#include <cuda_runtime.h>
#include <device_launch_parameters.h>

#include <iostream>
#include<cstdio>

#define N 256

void error_handling(cudaError_t res) {
    if (res !=cudaSuccess) {
        std::cout << "error!" << std::endl;
    }
}
__global__ void reduce_with_shared(float* in, float* out) {
    __shared__ float smen[N];
    int tid = threadIdx.x;
    smen[tid] = in[tid];
    __syncthreads();
    for (int stride = blockDim.x / 2; stride >= 1; stride >>=  1) {
        if (tid < stride)
            smen[tid] += smen[tid + stride];
        __syncthreads();
    }
    if (tid == 0) *out = smen[0];
}
__inline__ __device__ float warpReduce(float val) {
    // 每次从右边获取值，做加法
    //先广播自己当前的 val 值
    //然后从其他线程接收一个 val 值，再累加到自身的val
    for (int offset = warpSize / 2; offset >= 1; offset >>= 1)
        val += __shfl_down_sync(0xFFFFFFFF, val, offset);
    return val;
}
__global__ void reduce_with_warp(float* in, float* out) {
    int tid = threadIdx.x;
    float val = in[tid];
    float sum = warpReduce(val);
    //WarpSize是cuda提供的线程束大小的值
    if (tid % warpSize == 0)         // 只有每个 warp 的 lane 0 写结果
        out[tid / warpSize] = sum;
}
int main() {
    // 生成 N 个值初始化为 1.0f，期望和为 N
    float* h_in, * d_in, * d_out;
    cudaMallocHost(&h_in, N * sizeof(float));
    for (int i = 0; i < N; ++i) h_in[i] = 1.0f;
    cudaMalloc(&d_in, N * sizeof(float));
    cudaMalloc(&d_out, N * sizeof(float));
    cudaMemcpy(d_in, h_in, N * sizeof(float), cudaMemcpyHostToDevice);
    reduce_with_warp << <1, N >> > (d_in, d_out);
    cudaFreeHost(h_in);
    cudaFree(d_in);
    cudaFree(d_out);
}

warp规约有点难懂，需要解释一下

首先要理解warp的规约是在一个线程束内进行的规约（一般是32个线程），以及一个线程束内的线程是同时执行同一条指令的

也就是说32个线程同时走到了WarpReduce调用并进入，同时走到了__shfl_down_sync

val += __shfl_down_sync(0xFFFFFFFF, val, offset);

在这一行代码中，每个线程：

先广播自己当前的 val 值（传出去）；
同时从 lane = 自己的 lane + offset 那里接收一个 val 值；
然后将这个值加到自己的 val 上。

有个细节：如果 __shfl_down_sync 的目标 lane 超过了 31，CUDA 会自动忽略该调用

牢记所有线程都是同时执行该语句的，所以不会有有的线程先执行__shfl_down_sync修改了自己的val，再把val传出去的情况。

最后线程0会拿到32个线程值的总和

warp实现softmax

对于一组数 $x_{0},x_{1},...,x_{n-1}$ ，softmax 定义为：

$softmax(x_{i})=\frac{e^{x_{i}}}{\sum_{j}e^{x_{j}}}$

不过如果x很大，那么对应的e^x也会很大，一般会让所有x减去xi中的最大值再取e的指数幂（这样不会影响比例）

下面是一个warp内实现softmax的代码

__device__ float warp_softmax(float val) {
    // Step 1: 先减去最大值以提升数值稳定性
    float max_val = val;
    // 在warp内找最大值（规约）
    for (int offset = 16; offset > 0; offset /= 2) {
        float temp = __shfl_down_sync(0xffffffff, max_val, offset);
        max_val = fmaxf(max_val, temp);
    }

    // 广播最大值给所有线程（lane 0保存了最终结果）
    max_val = __shfl_sync(0xffffffff, max_val, 0);

    // Step 2: 计算指数
    float exp_val = expf(val - max_val);

    // Step 3: 求和所有exp
    float sum_exp = exp_val;
    for (int offset = 16; offset > 0; offset /= 2) {
        sum_exp += __shfl_down_sync(0xffffffff, sum_exp, offset);
    }

    // 广播sum给所有线程（lane 0保存了最终结果）
    sum_exp = __shfl_sync(0xffffffff, sum_exp, 0);

    // Step 4: softmax输出
    return exp_val / sum_exp;
}

下面是block级别的warp实现，不同warp之间还是需要共享内存来同步；求的是一个block内的softmax

#include <cstdio>
#include <cmath>

#define WARP_SIZE 32
//warp内规约求和（lane 0获取总和）
__inline__ __device__
float warpReduceSum(float val) {
    for (int offset = WARP_SIZE / 2; offset > 0; offset /= 2)
        val += __shfl_down_sync(0xffffffff, val, offset);
    return val;
}
//warp内规约求最大值（lane 0获取最大值）
__inline__ __device__
float warpReduceMax(float val) {
    for (int offset = WARP_SIZE / 2; offset > 0; offset /= 2)
        val = fmaxf(val, __shfl_down_sync(0xffffffff, val, offset));
    return val;
}

__global__ void block_softmax(float* input, float* output, int N) {
    int tid = threadIdx.x + blockIdx.x * blockDim.x;
    int lane = threadIdx.x % WARP_SIZE;//该线程在本warp内的编号
    int warp_id = threadIdx.x / WARP_SIZE;//该warp在整个block里的序号
    //每个warp对应的总和和最大值
    __shared__ float warp_max[32];
    __shared__ float warp_sum[32];
    __shared__ float smem_block_max;
    __shared__ float smem_block_sum;
    float x = (tid < N) ? input[tid] : -INFINITY;

    // Step 1: Warp内最大值
    float max_val = warpReduceMax(x);

    // Step 2: 线程0收集每个warp的最大值
    if (lane == 0)
        warp_max[warp_id] = max_val;

    __syncthreads();

    // Step 3: block范围内最大值（使用前WARP_SIZE线程处理）
    //整个block只有前32个线程干活了，根据warp_max进行规约求出block内的最大值
    float block_max = -INFINITY;
    if (threadIdx.x < WARP_SIZE) {
        block_max = warp_id < (blockDim.x + WARP_SIZE - 1) / WARP_SIZE ? warp_max[threadIdx.x] : -INFINITY;
        block_max = warpReduceMax(block_max);
    }

    // 广播block_max，整个block内的线程都会收到
    // thread 0 写入共享内存广播
    if (threadIdx.x == 0) smem_block_max = block_max;
    __syncthreads();
    block_max = smem_block_max;

    // Step 4: 减去最大值后求exp
    float exp_x = (tid < N) ? expf(x - block_max) : 0.0f;

    // Step 5: warp内规约求和
    float local_sum = warpReduceSum(exp_x);

    // Step 6: 每个warp写入warp_sum
    if (lane == 0)
        warp_sum[warp_id] = local_sum;

    __syncthreads();

    // Step 7: block范围内总和（again用前WARP_SIZE线程规约）
    float block_sum = 0.0f;
    if (threadIdx.x < WARP_SIZE) {
        block_sum = warp_id < (blockDim.x + WARP_SIZE - 1) / WARP_SIZE ? warp_sum[threadIdx.x] : 0.0f;
        block_sum = warpReduceSum(block_sum);
    }

    // 广播block_sum
    if (threadIdx.x == 0) smem_block_sum = block_sum;
    __syncthreads();
    block_sum = smem_block_sum;

    // Step 8: 输出softmax
    if (tid < N)
        output[tid] = exp_x / block_sum;
}

如果要整个 grid 中的所有线程输入数据 进行 softmax，由于block之间无法简便通信，只能将block级别的最大值、和先传回主机，主机先对每个block之间的最大值、和求出最大值、和，再启动新的核函数，也即

全局 Softmax 涉及三次 kernel 启动：

第一次归约 max。
第二次归约 exp 和。
第三次归一化。

warp实现前缀和Scan

__inline__ __device__
int warp_scan(int val) {
    for (int offset = 1; offset < 32; offset *= 2) {
        int y = __shfl_up_sync(0xFFFFFFFF, val, offset);
        if (threadIdx.x % 32 >= offset)
            val += y;
    }
    return val;
}

过程说明（以 8 个线程为例）：

ThreadID:  0   1   2   3   4   5   6   7
Input:     1   2   3   4   5   6   7   8

第一次循环：offset = 1

1 3 5 7 9 11 13 15

第二次循环：offset = 2

1 3 6 10 14 18 22 26
第三次循环：offset = 4

1 3 6 10 15 21 28 36

block级别的前缀和

__global__ void block_scan(const int* input, int* output) {
    __shared__ int warp_sum[32];//每个warp内的总和
    int tid = threadIdx.x;
    int lane = tid % warpSize;//该线程在本warp的lane
    int wid = tid / warpSize;//该线程所在warp的id
    int val = (tid < N) ? input[tid] : 0;
    val = warp_scan(val);//每个warp内部求出前缀和
    if (lane == warpSize - 1) warp_sum[wid] = val;//每个warp最后一个线程的值是warp内的总和
    __syncthreads();
    //以下只有第一个warp会执行
    if (wid == 0 && lane < (blockDim.x + warpSize - 1) / warpSize) {
        //在第一个warp内再执行一次求前缀和（对每个warp的总和）
        warp_sum[lane] = warp_scan(warp_sum[lane]);
    }
    __syncthreads();
    // 所有 warp 除了第一个 warp 加上前面 warp 的累加和
    if (wid > 0) {
        val += warp_sum[wid - 1];
    }
    if (tid < N) {
        output[tid] = val;
    }
}

AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐