您现在的位置：程式碼之家 >> 編程語言 > >> C語言 >> C++ >> 關於C++

OpenCL Reduction 規約操作與工作組同步機制詳解

編輯：關於C++

Reduction 操作核心概念

Reduction（規約）是將多個資料元素歸約為單一結果的並行模式：求和、最大值、最小值、點積等。GPU 的海量並行能力讓它成為大規模 Reduction 的理想平台，但關鍵挑戰在於工作組內部同步與跨工作組歸約的協調。

OpenCL 同步原語：barrier

void barrier(cl_mem_fence_flags flags);

參數選項：

CLK_LOCAL_MEM_FENCE：確保本地記憶體讀寫順序，所有 work-item 看到一致的本地記憶體狀態
CLK_GLOBAL_MEM_FENCE：確保全域記憶體訪問順序
可組合使用：CLK_LOCAL_MEM_FENCE | CLK_GLOBAL_MEM_FENCE

執行語義：工作組內所有 work-item 執行到 barrier 時會阻塞，直到最後一個 work-item 到達，然後同時繼續。這是工作組級別的屏障同步，OpenCL 不提供跨工作組的全域同步機制。

樹狀歸約算法實現

經典的並行 Reduction 採用「樹狀歸約」：每輪將活躍執行緒數量減半，配對累加，直到剩下一個結果。

__kernel void reduce(__global uint4* input, __global uint4* output, int NUM) {
    NUM = NUM / 4;  // 每 4 個整數打包為 uint4
    
    unsigned int tid = get_local_id(0);
    unsigned int local_size = get_local_size(0);
    unsigned int global_size = get_global_size(0);
    
    uint4 res = (uint4)(0, 0, 0, 0);
    __local uint4 shared_data[64];  // 本地記憶體緩衝區，大小 = local_size
    
    // 階段 1：每個 work-item 累加多個元素 (stride loop)
    unsigned int i = get_global_id(0);
    while (i < NUM) {
        res += input[i];
        i += global_size;
    }
    
    // 階段 2：存入本地記憶體並同步
    shared_data[tid] = res;
    barrier(CLK_LOCAL_MEM_FENCE);  // 關鍵：確保所有寫入完成
    
    // 階段 3：樹狀歸約在本地記憶體中進行
    for (unsigned int s = local_size >> 1; s > 0; s >>= 1) {
        if (tid < s) {
            shared_data[tid] += shared_data[tid + s];
        }
        barrier(CLK_LOCAL_MEM_FENCE);  // 每輪歸約後必須同步
    }
    
    // 階段 4：工作組結果寫回全域記憶體
    if (tid == 0) {
        output[get_group_id(0)] = shared_data[0];
    }
}

完整主機端程式碼

#include <CL/cl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#define CHECK_ERR(err, msg) \
    if (err != CL_SUCCESS) { fprintf(stderr, "%s: %d\n", msg, err); exit(1); }

int main() {
    cl_int err;
    
    // 1. 選擇平台與設備
    cl_platform_id platform;
    clGetPlatformIDs(1, &platform, NULL);
    cl_device_id device;
    clGetDeviceIDs(platform, CL_DEVICE_TYPE_GPU, 1, &device, NULL);
    
    // 2. 建立 Context 與 Command Queue
    cl_context ctx = clCreateContext(NULL, 1, &device, NULL, NULL, &err);
    CHECK_ERR(err, "CreateContext");
    cl_command_queue queue = clCreateCommandQueue(ctx, device, 0, &err);
    CHECK_ERR(err, "CreateCommandQueue");
    
    // 3. 載入並建置 Kernel
    FILE* fp = fopen("reduce.cl", "r");
    fseek(fp, 0, SEEK_END);
    size_t src_len = ftell(fp);
    rewind(fp);
    char* src = (char*)malloc(src_len + 1);
    fread(src, 1, src_len, fp);
    src[src_len] = '\0';
    fclose(fp);
    
    cl_program prog = clCreateProgramWithSource(ctx, 1, (const char**)&src, &src_len, &err);
    CHECK_ERR(err, "CreateProgram");
    err = clBuildProgram(prog, 1, &device, NULL, NULL, NULL);
    if (err != CL_SUCCESS) {
        size_t log_size;
        clGetProgramBuildInfo(prog, device, CL_PROGRAM_BUILD_LOG, 0, NULL, &log_size);
        char* log = (char*)malloc(log_size);
        clGetProgramBuildInfo(prog, device, CL_PROGRAM_BUILD_LOG, log_size, log, NULL);
        fprintf(stderr, "Build Log:\n%s\n", log);
        free(log);
        exit(1);
    }
    free(src);
    
    cl_kernel kernel = clCreateKernel(prog, "reduce", &err);
    CHECK_ERR(err, "CreateKernel");
    
    // 4. 準備資料
    const int NUM = 25600;
    const int LOCAL_SIZE = 64;
    const int GLOBAL_SIZE = 640;  // 10 個工作組
    const int GROUP_NUM = GLOBAL_SIZE / LOCAL_SIZE;
    
    uint4* input = (uint4*)malloc(NUM * sizeof(uint4));
    for (int i = 0; i < NUM; i++) {
        input[i] = (uint4)(i*4+1, i*4+2, i*4+3, i*4+4);
    }
    uint4* output = (uint4*)calloc(GROUP_NUM, sizeof(uint4));
    
    cl_mem in_buf = clCreateBuffer(ctx, CL_MEM_READ_ONLY | CL_MEM_COPY_HOST_PTR,
                                   NUM * sizeof(uint4), input, &err);
    CHECK_ERR(err, "CreateBuffer input");
    cl_mem out_buf = clCreateBuffer(ctx, CL_MEM_WRITE_ONLY,
                                    GROUP_NUM * sizeof(uint4), NULL, &err);
    CHECK_ERR(err, "CreateBuffer output");
    
    // 5. 設定 Kernel 參數
    clSetKernelArg(kernel, 0, sizeof(cl_mem), &in_buf);
    clSetKernelArg(kernel, 1, sizeof(cl_mem), &out_buf);
    clSetKernelArg(kernel, 2, sizeof(int), &NUM);
    // 本地記憶體參數：大小 = LOCAL_SIZE * sizeof(uint4)
    clSetKernelArg(kernel, 3, LOCAL_SIZE * sizeof(uint4), NULL);
    
    // 6. 執行 Kernel
    size_t global_ws = GLOBAL_SIZE;
    size_t local_ws = LOCAL_SIZE;
    err = clEnqueueNDRangeKernel(queue, kernel, 1, NULL, &global_ws, &local_ws, 0, NULL, NULL);
    CHECK_ERR(err, "EnqueueNDRangeKernel");
    clFinish(queue);
    
    // 7. 讀回結果並驗證
    clEnqueueReadBuffer(queue, out_buf, CL_TRUE, 0, GROUP_NUM * sizeof(uint4), output, 0, NULL, NULL);
    
    // CPU 端驗證：計算總和
    long long cpu_sum = 0;
    for (int i = 0; i < NUM * 4; i++) cpu_sum += i + 1;
    
    long long gpu_sum = 0;
    for (int i = 0; i < GROUP_NUM; i++) {
        gpu_sum += output[i].s0 + output[i].s1 + output[i].s2 + output[i].s3;
    }
    
    printf("CPU Sum: %lld\nGPU Sum: %lld\nMatch: %s\n", 
           cpu_sum, gpu_sum, cpu_sum == gpu_sum ? "YES" : "NO");
    
    // 8. 清理資源
    clReleaseMemObject(in_buf);
    clReleaseMemObject(out_buf);
    clReleaseKernel(kernel);
    clReleaseProgram(prog);
    clReleaseCommandQueue(queue);
    clReleaseContext(ctx);
    free(input);
    free(output);
    
    return 0;
}

關鍵優化要點

向量化載入：使用 uint4 等向量類型，提升記憶體頻寬利用率
Stride Loop：每個 work-item 處理多個元素，隱藏記憶體延遲，提高佔用率
本地記憶體 Bank Conflict 避免：陣列大小設為 2 的冪次，且 local_size 為 32 的倍數 (warp 大小)
最小同步範圍：僅使用 CLK_LOCAL_MEM_FENCE，避免不必要的全域記憶體同步開銷

跨工作組的最終歸約

上述 Kernel 輸出每個工作組的部分和。最終結果需在主機端或啟動第二個單工作組 Kernel 完成最後歸約：

// 主機端完成最終歸約 (資料量小，開銷可忽略)
long long final_sum = 0;
for (int i = 0; i < GROUP_NUM; i++) {
    final_sum += output[i].s0 + output[i].s1 + output[i].s2 + output[i].s3;
}

Reduction 並行模式示意圖