2

私はOpenMPの基本を知っており、その反復を並列化するために、前の反復に依存してはならないことを知っています。また、リダクションを使用することもできますが、サポートされるのは+、-、/、*、&&、||などの基本的な演算子のみです。

これを並列にするにはどうすればよいですか？

for (i = 1; i < n; ++i) {
    for (j = 1; j < n; ++j) {
        // stanga
        if (res[i][j - 1] != res[i][j]) {
            cmin2[i][j][0] = min(cmin2_res[i][j - 1][0] + 1, cmin[i][j][0]);
            cmin2_res[i][j][0] = min(cmin2[i][j - 1][0] + 1, cmin_res[i][j][0]);
         } else {
            cmin2[i][j][0] = min(cmin2[i][j - 1][0] + 1, cmin[i][j][0]);
            cmin2_res[i][j][0] = min(cmin2_res[i][j - 1][0] + 1, cmin_res[i][j][0]);
         }
         // sus
         if (res[i - 1][j] != res[i][j]) {
             cmin2[i][j][0] = min3(cmin2[i][j][0], cmin2_res[i - 1][j][0] + 1, cmin[i][j][1]);
             cmin2_res[i][j][0] = min3(cmin2_res[i][j][0], cmin2[i - 1][j][0] + 1, cmin_res[i][j][1]);
         } else {
             cmin2[i][j][0] = min3(cmin2[i][j][0], cmin2[i - 1][j][0] + 1, cmin[i][j][1]);
             cmin2_res[i][j][0] = min3(cmin2_res[i][j][0], cmin2_res[i - 1][j][0] + 1, cmin_res[i][j][1]);
         }
     }
 }

私の質問は、これをどのように分解して並行して実行できるようにするかです（可能であれば削減を使用することもできます）。

問題は、このようなforのグループがさらに3つあるため、各反復で操作をこの順序で実行する必要があることです。

PSminとmin3はマクロです。

4

1 に答える 1

3

やりたいことを力ずくで行う方法はありますが、並列化を改善するには、ルーチンの内外で何をしたいのかについてもう少し入力する必要があります。

ループ内のデータ依存関係は、ij 空間では次のようになります。

     i →
  ..........
j .....1....
↓ ....12....
  ...123....

ここで、ポイント 3 の値はそれらのポイント 2 に依存し、それらはポイント 1 の値に依存するなどです。この対角構造により、ループを並べ替えて、グリッドを斜めにトラバースすることができます。たとえば、最初の反復は終了します (0 ,1)、(1,0)、(0,2)、(1,1)、(2,0) など。問題の簡略版は次のようになります。

#include <stdio.h>
#include <stdlib.h>
#include <math.h>
#include <sys/time.h>

int **int2darray(int n, int m);
void free2darray(int **array);
void init2darray(int **array, int n, int m);
void tick(struct timeval *timer);
double tock(struct timeval *timer);

int main(int argc, char **argv) {

    const int N=10000;
    int **serialarr, **omparr;

    struct timeval serialtimer, omptimer;
    double serialtime, omptime;

    serialarr = int2darray(N,N);
    omparr    = int2darray(N,N);

    init2darray(serialarr, N, N);
    init2darray(omparr,    N, N);

    /* serial calculation */
    tick(&serialtimer);
    for (int i=1; i<N; i++)
        for (int j=1; j<N; j++)
            serialarr[i][j] = serialarr[i-1][j] + serialarr[i][j-1];
    serialtime = tock(&serialtimer);

    /* omp */
    tick(&omptimer);
    #pragma omp parallel shared(omparr) default(none)
    {
        for (int ipj=1; ipj<=N; ipj++) {
            #pragma omp for
            for (int j=1; j<ipj; j++) {
                int i = ipj - j;
                omparr[i][j] = omparr[i-1][j] + omparr[i][j-1];
            }
        }
        for (int ipj=N+1; ipj<2*N-1; ipj++) {
            #pragma omp for
            for (int j=ipj-N+1; j<N; j++) {
                int i = ipj - j;
                omparr[i][j] = omparr[i-1][j] + omparr[i][j-1];
            }
        }
    }
    omptime = tock(&omptimer);

    /* compare results */
    int abserr = 0;
    for (int i=0; i<N; i++)
        for (int j=0; j<N; j++)
            abserr += abs(omparr[i][j] - serialarr[i][j]);

    printf("Difference between serial and OMP array: %d\n", abserr);
    printf("Serial time = %lf\n", serialtime);
    printf("OMP time    = %lf\n", omptime);

    free2darray(omparr);
    free2darray(serialarr);
    return 0;
}


int **int2darray(int n, int m) {
    int *data = malloc(n*m*sizeof(int));
    int **array = malloc(n*sizeof(int*));
    for (int i=0; i<n; i++)
        array[i] = &(data[i*m]);

    return array;
}

void free2darray(int **array) {
    free(array[0]);
    free(array);
}

void init2darray(int **array, int n, int m) {
    for (int i=0; i<n; i++)
        for (int j=0; j<m; j++)
            array[i][j] = i*m+j;
}

void tick(struct timeval *timer) {
    gettimeofday(timer, NULL);
}

double tock(struct timeval *timer) {
    struct timeval now;
    gettimeofday(&now, NULL);
    return (now.tv_usec-timer->tv_usec)/1.0e6 + (now.tv_sec - timer->tv_sec);
}

実行すると次のようになります。

$ gcc -fopenmp -Wall -O2 loops.c -o loops -std=c99 
$ export OMP_NUM_THREADS=8
$ ./loops
Difference between serial and OMP array: 0
Serial time = 0.246649
OMP time    = 0.174936

反復ごとの計算量が少なく、並列化されるのは内側のループであり、奇妙でキャッシュに適していない順序でメモリを処理しているため、大きな N を使用しても高速化がかなり貧弱であることがわかります。

上記のいくつかはおそらく修正される可能性がありますが、あなたが何をしようとしているのかについてもっと知るのにもう少し役立ちます。たとえば、cmin2_res 配列を気にしますか、それとも単なる中間生成物ですか? つまり、何を計算しようとしているのですか？

于 2012-11-04T18:04:39.330 に答える