1

我试图在我的内部循环中减少对 std::max 的调用次数，因为我调用它数百万次（毫不夸张！），这使得我的并行代码比顺序代码运行得慢。基本思想（是的，这是一个分配）是代码计算某个网格点的温度，一次又一次地迭代，直到最大变化不超过某个非常小的数字（例如 0.01）。新温度是其上方、下方和旁边的单元格中温度的平均值。结果，每个单元格都有不同的值，我想返回给定网格块的任何单元格中的最大变化。

我已经让代码工作了，但是速度很慢，因为我在内循环中对 std::max 进行了大量（过多）调用，而且它是 O(n*n)。我使用了一维域分解

注意： tdiff 不依赖于任何东西，只依赖于矩阵中的内容

归约函数的输入是 lambda 函数的结果

diff 是该网格块中单个单元格在 1 次迭代中的最大变化

阻塞范围在代码前面定义

t_new 是该网格点的新温度，t_old 是旧温度

max_diff = parallel_reduce(range, 0.0,
        //lambda function returns local max
        [&](blocked_range<size_t> range, double diff)-> double
        {
            for (size_t j = range.begin(); j<range.end(); j++)
            {
                for (size_t i = 1; i < n_x-1; i++)
                {
                    t_new[j*n_x+i]=0.25*(t_old[j*n_x+i+1]+t_old[j*n_x+i-1]+t_old[(j+1)*n_x+i]+t_old[(j-1)*n_x+i]);
                    tdiff = fabs(t_old[j*n_x+i] - t_new[j*n_x+i]);
                    diff = std::max(diff, tdiff);
                }   
            }
            return diff;    //return biggest value of tdiff for that iteration - once per 'i'
        },
        //reduction function - takes in all the max diffs for each iteration, picks the largest
        [&](double a, double b)-> double
        {
            convergence = std::max(a,b);
            return convergence;
        }
    );

如何让我的代码更有效率？我想减少对 std::max 的调用，但需要保持正确的值。使用 gprof 我得到：

Each sample counts as 0.01 seconds.
  %   cumulative   self              self     total           
 time   seconds   seconds    calls  ms/call  ms/call  name    
 61.66      3.47     3.47  3330884     0.00     0.00  double const& std::max<double>(double const&, double const&)
 38.03      5.61     2.14     5839     0.37     0.96  _ZZ4mainENKUlN3tbb13blocked_rangeImEEdE_clES1_d

ETA：执行我的代码所花费的时间中有 61.66% 是在 std::max 调用上，它调用了超过 300 万次。lambda函数的每个输出都会调用reduce函数，因此减少lambda函数中对std::max的调用次数也会减少对reduce函数的调用次数

4

2 回答 2

2

首先，我希望std::max被内联到它的调用者中，所以 gprof 指出它是一个单独的热点是很可疑的。您是否可能分析调试配置？

另外，我认为这不是std::max罪魁祸首。除非在其实现中启用了一些特殊检查，否则我认为它应该等效于(diff<tdiff)?tdiff:diff. 由于 std::max 的参数之一是您更新的变量，因此您可以尝试if (tdiff>diff) diff = tdiff;改用，但我怀疑它会给您带来很多好处（也许编译器可以自己进行此类优化）。

最有可能，作为采样打滑std::max的结果突出显示；即真正的热点在上面的计算中，这是非常有意义的，因为更多的工作和对可能具有更长延迟的非本地数据（数组）的访问，特别是如果相应的位置不在 CPU 缓存中。std::max

根据n_x网格中行（最好t_old在缓存中尽可能多地重用数据。按行处理，您要么根本不重用一个点，t_old直到下一行（for i+1and i-1points），要么只重用一次（对于同一行中的两个邻居）。更好的方法是通过矩形块处理网格，这有助于重用缓存中的热数据。使用 TBB，方法是使用blocked_range2d. 它需要对您的代码进行最少的更改；基本上，更改 lambda 内的范围类型和两个循环：外部和内部循环应该分别迭代range.rows()和range.cols()。

于 2013-05-22T09:32:12.597 回答
0

我最终使用了parallel_for：

parallel_for(range, [&](blocked_range<size_t> range)
        {
            double loc_max = 0.0;
            double tdiff;
            for (size_t j = range.begin(); j<range.end(); j++)
            {
                for (size_t i = 1; i < n_x-1; i++)
                {
                    t_new[j*n_x+i]=0.25*(t_old[j*n_x+i+1]+t_old[j*n_x+i-1]+t_old[(j+1)*n_x+i]+t_old[(j-1)*n_x+i]);
                    tdiff = fabs(t_old[j*n_x+i] - t_new[j*n_x+i]); 
                    loc_max = std::max(loc_max, tdiff); 
                }   
            }
            //reduction function - takes in all the max diffs for each iteration, picks the largest
            {
                max_diff = std::max(max_diff, loc_max);
            }
        }
    );

现在我的代码在 2 秒内运行 8000x8000 网格 :-)

于 2013-05-23T02:09:41.057 回答