如何优化基于OpenCL和C实现的奇偶排序算法的运行速度？

核心性能瓶颈 #

• 内核启动开销过高：主机端每轮迭代都要入队2次内核，数组长度为N时就要启动N次内核，OpenCL内核启动的固定开销会占用绝大多数运行时间
• 全局内存访问频繁：所有交换操作直接读写全局内存，没有利用速度比全局内存高数十倍的局部内存做缓存
• 冗余工作项过多：当前全局工作大小计算逻辑会生成大量空跑的无效工作项，浪费计算单元资源
• 无提前终止逻辑：即使数组已经完成排序，仍会跑完所有预设迭代，产生大量无用计算

具体优化方案 #

1. 内核逻辑合并 #

将奇偶排序的两层逻辑合并到同一个内核中，把迭代循环放到内核内部执行，主机端仅需入队1次内核，彻底消除重复的内核启动开销。

2. 新增提前终止机制 #

新增全局交换标志位，某轮迭代没有发生任何交换时直接跳出循环，对于接近有序的数组可减少90%以上的迭代次数。

3. 利用局部内存缓存数据 #

将单个工作组负责的数组段先拷贝到__local内存，在局部内存完成多轮排序后再统一写回全局，大幅降低全局内存访问带来的延迟损耗。

4. 裁剪冗余工作项 #

全局工作大小调整为ceil(array_size / 2)，仅保留实际需要的工作项，避免计算资源浪费。

优化后代码示例 #

主机端修改部分 #

// 调整全局工作大小计算逻辑
size_t global_work_size = (array_size + 1) / 2;
size_t local_work_size = ((size_t)global_work_size > max_work_group_size) ? max_work_group_size : (size_t)global_work_size;
global_work_size = (global_work_size + local_work_size - 1) / local_work_size * local_work_size;

// 创建交换标志位缓冲区
int init_flag = 1;
cl_mem swap_flag_buf = clCreateBuffer(context, CL_MEM_READ_WRITE | CL_MEM_COPY_HOST_PTR, sizeof(int), &init_flag, &err);
// 给内核设置参数
err  = clSetKernelArg(sort_kernel, 0, sizeof(cl_mem), &array_buf);
err |= clSetKernelArg(sort_kernel, 1, sizeof(int), &array_size);
err |= clSetKernelArg(sort_kernel, 2, sizeof(cl_mem), &swap_flag_buf);
err |= clSetKernelArg(sort_kernel, 3, local_work_size * 2 * sizeof(int), NULL); // 局部内存参数
check_err(err, "set kernel arg");

// 仅入队一次内核即可完成全量排序
err = clEnqueueNDRangeKernel(command_queue, sort_kernel, 1, NULL, &global_work_size, &local_work_size, 0, NULL, NULL);
check_err(err, "enqueue kernel");

内核修改部分 #

__kernel void sort_kernel(
    __global int *array, 
    int array_size,
    __global int *swap_flag,
    __local int *local_buf
) {
    int gid = get_global_id(0);
    int lid = get_local_id(0);
    int local_size = get_local_size(0);
    int group_id = get_group_id(0);
    int group_start = group_id * local_size * 2;
    int local_elem_cnt = min(local_size * 2, array_size - group_start);
    
    // 加载当前工作组负责的数组段到局部内存
    if (group_start + lid < array_size) local_buf[lid] = array[group_start + lid];
    if (group_start + lid + local_size < array_size) local_buf[lid + local_size] = array[group_start + lid + local_size];
    barrier(CLK_LOCAL_MEM_FENCE);

    int local_swap;
    do {
        // 每轮迭代开始重置全局交换标志
        if (lid == 0) {
            *swap_flag = 0;
        }
        barrier(CLK_GLOBAL_MEM_FENCE);
        local_swap = 0;

        // 偶排序阶段
        if (lid * 2 < local_elem_cnt - 1) {
            int idx = lid * 2;
            if (local_buf[idx] > local_buf[idx + 1]) {
                int tmp = local_buf[idx];
                local_buf[idx] = local_buf[idx + 1];
                local_buf[idx + 1] = tmp;
                local_swap = 1;
            }
        }
        barrier(CLK_LOCAL_MEM_FENCE);

        // 奇排序阶段
        if (lid * 2 + 1 < local_elem_cnt - 1) {
            int idx = lid * 2 + 1;
            if (local_buf[idx] > local_buf[idx + 1]) {
                int tmp = local_buf[idx];
                local_buf[idx] = local_buf[idx + 1];
                local_buf[idx + 1] = tmp;
                local_swap = 1;
            }
        }
        barrier(CLK_LOCAL_MEM_FENCE);

        // 上报当前工作组的交换状态
        if (local_swap) {
            atomic_inc(swap_flag);
        }
        barrier(CLK_GLOBAL_MEM_FENCE);
    } while (*swap_flag > 0);

    // 排序完成后写回全局内存
    if (group_start + lid < array_size) array[group_start + lid] = local_buf[lid];
    if (group_start + lid + local_size < array_size) array[group_start + lid + local_size] = local_buf[lid + local_size];
}

补充优化方向 #

如果数组长度远大于设备显存容量，可采用分块排序策略，将数组拆分为多个适合设备处理的块，分别排序后再做归并即可。

内容的提问来源于stack exchange，提问作者jack