将 OpenMP 应用于 C++ 中的特定嵌套循环

Posted 2023-02-22

【中文标题】将 OpenMP 应用于 C++ 中的特定嵌套循环

【英文标题】：Applying OpenMP to particular nested loops in C++

【发布时间】：2016-04-13 16:29:31

【问题描述】：

我在使用 openmp 并行化一段代码时遇到问题，我认为某些必须按顺序进行的操作存在概念问题。

else if (PERF_ROWS <= MAX_ROWS && function_switch == true)

    int array_dist_perf[PERF_ROWS];
    int array_dist[MAX_ROWS];

    #pragma omp parallel for collapse(2)
    for (int i = 0; i < MAX_COLUMNS;
                    i = i + 1 + (i % PERF_CLMN == 0 ? 1:0))
    
        for (int j = 0; j < PERF_ROWS; j++) //truncation perforation
        
            array_dist_perf[j] = abs(input[j] - input_matrix[j][i]);
        

        float av = mean(PERF_ROWS, array_dist_perf);

        float score = score_func(av);

        if (score > THRESHOLD_SCORE)
        
            for (int k = 0; k < MAX_ROWS; k++)
            
                array_dist[k] = abs(input[k] - input_matrix[k][i]);
            

            float av_real = mean(MAX_ROWS, array_dist);

            float score_real = score_func(av_real);

            rank_function(score_real, i);
        
    

错误是“折叠的循环没有完美嵌套”。我在 g++-5 上使用 Clion。提前致谢

【参考方案1】：

首先，完美嵌套的循环有以下形式：

for (init1; cond1; inc1)

   for (init2; cond2; inc2)
   
      ...
   

请注意，外循环的主体仅由内循环组成，没有其他内容。您的代码绝对不是这种情况 - 内部循环后面还有很多其他语句。

其次，您的外部循环不是 OpenMP 所需的规范形式。典型的循环是可以很容易地预先确定迭代次数和迭代步长的循环。由于您所做的是每次 i 是 PERF_CLMN 的倍数时跳过一次迭代，您可以将循环重写为：

for (int i = 0; i < MAX_COLUMNS; i++)

    if (i % PERF_CLMN == 1) continue;
    ...

这将造成工作不平衡，具体取决于MAX_COLUMNS 是否是线程数的倍数。但是还有另一个来源或不平衡，即rank_function() 的条件评估。因此，您应该使用动态调度。

现在，显然两个 array_dist* 循环都是私有的，但在您的情况下它们不是私有的，这将导致数据竞争。要么在循环体中移动数组的定义，要么使用private() 子句。

#pragma omp parallel for schedule(dynamic) private(array_dist_perf,array_dist)
for (int i = 0; i < MAX_COLUMNS; i++)

    if (i % PERF_CLMN == 1) continue;
    ...

现在，对于一些不请自来的优化建议：两个内部循环是多余的，因为第一个循环基本上是在做第二个工作的子集。您可以通过仅使用单个数组来优化计算并节省内存，并让第二个循环从第一个循环结束的地方继续。代码的最终版本应如下所示：

else if (PERF_ROWS <= MAX_ROWS && function_switch == true)

    int array_dist[MAX_ROWS];

    #pragma omp parallel for schedule(dynamic) private(array_dist)
    for (int i = 0; i < MAX_COLUMNS; i++)
    
        if (i % PERF_CLMN == 1) continue;

        for (int j = 0; j < PERF_ROWS; j++) //truncation perforation
        
            array_dist[j] = abs(input[j] - input_matrix[j][i]);
        

        float av = mean(PERF_ROWS, array_dist);

        float score = score_func(av);

        if (score > THRESHOLD_SCORE)
        
            for (int k = PERF_ROWS; k < MAX_ROWS; k++)
            
                array_dist[k] = abs(input[k] - input_matrix[k][i]);
            

            float av_real = mean(MAX_ROWS, array_dist);

            float score_real = score_func(av_real);

            rank_function(score_real, i);
        
    

另一个优化潜力在于input_matrix 不是以缓存友好的方式访问的。转置会导致列数据连续存储在内存中，提高内存访问局部性。

【讨论】：

非常感谢，缓存友好的方式是什么意思？ input_matrix 只是一个全局变量，因为维度很大。

您目前正在像... input_matrix[k-1][i] input_matrix[k][i] input_matrix[k+1][i] ...一样访问它。这些元素不会连续存储在内存中。如果您转置矩阵并像... input_matrix[i][k-1] input_matrix[i][k] input_matrix[i][k+1] ... 那样访问它，则元素将连续存储在内存中，并将作为同一缓存行的一部分加载，这将显着加快内存访问速度。

再次感谢，很遗憾我只有六个声誉，无法添加反馈，但您解决了我的问题。 =)