4

与 MPICH 相比，我试图用 OpenMP 来证明一个观点，我编写了以下示例来演示在 OpenMP 中实现高性能是多么容易。

Gauss-Seidel 迭代被分成两个独立的运行，这样在每次扫描中，每个操作都可以按任何顺序执行，并且每个任务之间应该没有依赖关系。所以理论上每个处理器都不应该等待另一个进程执行任何类型的同步。

我遇到的问题是，与问题大小无关，我发现只有 2 个处理器的加速很弱，如果有 2 个以上的处理器，它甚至可能会更慢。许多其他线性并行例程我可以获得非常好的缩放，但这一个很棘手。

我担心我无法向编译器“解释”我在数组上执行的操作是线程安全的，因此它无法真正有效。

请参见下面的示例。

任何人都知道如何使用 OpenMP 使其更有效？

void redBlackSmooth(std::vector<double> const & b,
                    std::vector<double> & x,
                    double h)
{
    // Setup relevant constants.
    double const invh2 = 1.0/(h*h);
    double const h2 = (h*h);
    int const N = static_cast<int>(x.size());
    double sigma = 0;

    // Setup some boundary conditions.
    x[0] = 0.0;
    x[N-1] = 0.0;

    // Red sweep.
    #pragma omp parallel for shared(b, x) private(sigma)
    for (int i = 1; i < N-1; i+=2)
    {
        sigma = -invh2*(x[i-1] + x[i+1]);
        x[i] = (h2/2.0)*(b[i] - sigma);
    }

    // Black sweep.
    #pragma omp parallel for shared(b, x) private(sigma)
    for (int i = 2; i < N-1; i+=2)
    {
        sigma = -invh2*(x[i-1] + x[i+1]);
        x[i] = (h2/2.0)*(b[i] - sigma);
    }
}

补充：我现在也尝试使用原始指针实现，它与使用 STL 容器具有相同的行为，因此可以排除它是来自 STL 的一些伪关键行为。

4

2 回答 2

1

首先，确保x向量与缓存边界对齐。我做了一些测试，如果我强制对齐内存，我的机器（核心二重奏）上的代码会得到 100% 的改进：

double * x;
const size_t CACHE_LINE_SIZE = 256;
posix_memalign( reinterpret_cast<void**>(&x), CACHE_LINE_SIZE, sizeof(double) * N);

其次，您可以尝试为每个线程分配更多计算（通过这种方式，您可以保持缓存行分离），但我怀疑 openmp 已经在后台做了类似的事情，所以它可能对大 N 毫无价值。

x在我的情况下，当没有缓存对齐时，这个实现要快得多。

const int workGroupSize = CACHE_LINE_SIZE / sizeof(double);
assert(N % workGroupSize == 0); //Need to tweak the code a bit to let it work with any N
const int workgroups = N / workGroupSize;
int j, base , k, i;

#pragma omp parallel for shared(b, x) private(sigma, j, base, k, i)
for ( j = 0; j < workgroups; j++ ) {
    base = j * workGroupSize;
    for (int k = 0; k < workGroupSize; k+=2)
    {
        i = base + k + (redSweep ? 1 : 0);
        if ( i == 0 || i+1 == N) continue;
        sigma = -invh2* ( x[i-1] + x[i+1] );
        x[i] = ( h2/2.0 ) * ( b[i] - sigma );
    }
}

总之，你肯定有缓存冲突的问题，但考虑到 openmp 的工作方式（遗憾的是我不熟悉它），它应该足以使用正确分配的缓冲区。

于 2013-03-30T21:31:27.977 回答
0

我认为主要问题在于您使用的数组结构的类型。让我们尝试将结果与向量和数组进行比较。（数组 = 使用 new 运算符的 c 数组）。

矢量和数组大小为 N = 10000000。我强制平滑函数重复以保持运行时间 > 0.1 秒。

Vector Time:    0.121007        Repeat: 1       MLUPS:  82.6399

Array Time:     0.164009        Repeat: 2       MLUPS:  121.945

MLUPS = ((N-2)*repeat/runtime)/1000000 (Million Lattice Points Update per second)

MFLOPS 在网格计算方面具有误导性。基本等式中的一些变化可以导致考虑相同运行时的高性能。

修改后的代码：

double my_redBlackSmooth(double *b, double* x, double h, int N)
{
    // Setup relevant constants.
    double const invh2 = 1.0/(h*h);
    double const h2 = (h*h);
    double sigma = 0;

    // Setup some boundary conditions.
    x[0] = 0.0;
    x[N-1] = 0.0;

    double runtime(0.0), wcs, wce;
    int repeat = 1;
    timing(&wcs);
    for(; runtime < 0.1; repeat*=2)
    {
        for(int r = 0; r < repeat; ++r)
        {
            // Red sweep.
            #pragma omp parallel for shared(b, x) private(sigma)
            for (int i = 1; i < N-1; i+=2)
            {
                sigma = -invh2*(x[i-1] + x[i+1]);
                x[i] = (h2*0.5)*(b[i] - sigma);
            }
            // Black sweep.
            #pragma omp parallel for shared(b, x) private(sigma)
            for (int i = 2; i < N-1; i+=2)
            {
                sigma = -invh2*(x[i-1] + x[i+1]);
                x[i] = (h2*0.5)*(b[i] - sigma);
            }
            //  cout << "In Array:      " << r << endl;
        }
        if(x[0] != 0) dummy(x[0]);
        timing(&wce);
        runtime = (wce-wcs);
    }
    //  cout << "Before division:   " << repeat << endl;
    repeat /= 2;
    cout << "Array Time:\t" << runtime << "\t" << "Repeat:\t" << repeat
         << "\tMLUPS:\t" << ((N-2)*repeat/runtime)/1000000.0 << endl;

    return runtime;
}

除了数组类型之外，我没有更改代码中的任何内容。为了更好的缓存访问和阻塞，您应该查看数据对齐 (_mm_malloc)。

于 2013-04-04T08:23:44.940 回答