9

この論文で並列マージソートのアルゴリズムを見てきました。これはコードです：

void mergesort_parallel_omp (int a[], int size, int temp[], int threads) 
{  
    if ( threads == 1)       { mergesort_serial(a, size, temp); }
    else if (threads > 1) 
    {
         #pragma omp parallel sections
         {
             #pragma omp section
             mergesort_parallel_omp(a, size/2, temp, threads/2);
             #pragma omp section
             mergesort_parallel_omp(a + size/2, size - size/2, temp + size/2, threads - threads/2);
         }
         merge(a, size, temp); 
    } // threads > 1
}

マルチコアで実行しています。何が起こるかというと、ツリーの葉で 2 つのスレッドが並行して実行されます。彼らが仕事を終えた後、他の2つのスレッドが開始されます。すべてのリーフ ノードに空きコアがある場合でも。

その理由は、この OpenMP コードが並列領域内に並列領域を作成しないためだと思います。私は正しいですか？

4

2 に答える 2

15

その理由は、OpenMP が並列領域内に並列領域を作成できないためだと思います

並列領域の並列領域を持つことができます。

OpenMP の並列領域は、互いに入れ子にすることができます。ネストされた並列処理が無効になっている場合、並列領域内で並列構造に遭遇したスレッドによって作成された新しいチーム は、遭遇したスレッドのみで構成されます。ネストされた並列処理が有効になっている場合、新しいチームは複数のスレッド( source ) で構成される場合があります。

コードを正しく実行するには、 と を呼び出す必要がありomp_set_nested(1)ますomp_set_num_threads(2)。

OMP_NESTED 環境変数を設定するか、omp_set_nested() 関数を呼び出すことで、ネストされた並列処理を有効または無効にすることができます。

セクションの代わりにパフォーマンスを向上させるには、次のように OpenMP タスクを使用できます (詳細な情報と例については、こちらを参照してください)。

void merge(int * X, int n, int * tmp) {
   ...
} 

void mergeSort(int *X, int n, int *tmp)
{  
   if (n < 2) return;
   
   #pragma omp task shared(X) if (n > TASK_SIZE)
   mergeSort(X, n/2, tmp);
   
   #pragma omp task shared(X) if (n > TASK_SIZE)
   mergeSort(X+(n/2), n-(n/2), tmp + n/2);
   
   #pragma omp taskwait
   mergeSortAux(X, n, tmp);
}



int main()
{
   ...
   #pragma omp parallel
   {
      #pragma omp single
      mergesort(data, n, tmp);
   }
}

マージ アルゴリズムのシーケンシャル コードは、Dr. Johnnie W. Baker の Web ページに由来します。. ただし、この回答で提供しているコードには、いくつかの修正とパフォーマンスの改善が含まれています。

完全な実行例:

#include <assert.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#include <omp.h>

#define TASK_SIZE 100

unsigned int rand_interval(unsigned int min, unsigned int max)
{
    // https://stackoverflow.com/questions/2509679/
    int r;
    const unsigned int range = 1 + max - min;
    const unsigned int buckets = RAND_MAX / range;
    const unsigned int limit = buckets * range;

    do
    {
        r = rand();
    } 
    while (r >= limit);

    return min + (r / buckets);
}

void fillupRandomly (int *m, int size, unsigned int min, unsigned int max){
    for (int i = 0; i < size; i++)
    m[i] = rand_interval(min, max);
} 

void mergeSortAux(int *X, int n, int *tmp) {
   int i = 0;
   int j = n/2;
   int ti = 0;

   while (i<n/2 && j<n) {
      if (X[i] < X[j]) {
         tmp[ti] = X[i];
         ti++; i++;
      } else {
         tmp[ti] = X[j];
         ti++; j++;
      }
   }
   while (i<n/2) { /* finish up lower half */
      tmp[ti] = X[i];
      ti++; i++;
   }
   while (j<n) { /* finish up upper half */
      tmp[ti] = X[j];
      ti++; j++;
   }
   memcpy(X, tmp, n*sizeof(int));
} 

void mergeSort(int *X, int n, int *tmp)
{
   if (n < 2) return;

   #pragma omp task shared(X) if (n > TASK_SIZE)
   mergeSort(X, n/2, tmp);

   #pragma omp task shared(X) if (n > TASK_SIZE)
   mergeSort(X+(n/2), n-(n/2), tmp + n/2);

   #pragma omp taskwait
   mergeSortAux(X, n, tmp);
}

void init(int *a, int size){
   for(int i = 0; i < size; i++)
       a[i] = 0;
}

void printArray(int *a, int size){
   for(int i = 0; i < size; i++)
       printf("%d ", a[i]);
   printf("\n");
}

int isSorted(int *a, int size){
   for(int i = 0; i < size - 1; i++)
      if(a[i] > a[i + 1])
        return 0;
   return 1;
}

int main(int argc, char *argv[]) {
        srand(123456);
        int N  = (argc > 1) ? atoi(argv[1]) : 10;
        int print = (argc > 2) ? atoi(argv[2]) : 0;
        int numThreads = (argc > 3) ? atoi(argv[3]) : 2;
        int *X = malloc(N * sizeof(int));
        int *tmp = malloc(N * sizeof(int));

        omp_set_dynamic(0);              /** Explicitly disable dynamic teams **/
        omp_set_num_threads(numThreads); /** Use N threads for all parallel regions **/

         // Dealing with fail memory allocation
        if(!X || !tmp)
        { 
           if(X) free(X);
           if(tmp) free(tmp);
           return (EXIT_FAILURE);
        }

        fillupRandomly (X, N, 0, 5);

        double begin = omp_get_wtime();
        #pragma omp parallel
        {
            #pragma omp single
            mergeSort(X, N, tmp);
        }   
        double end = omp_get_wtime();
        printf("Time: %f (s) \n",end-begin);
    
        assert(1 == isSorted(X, N));

        if(print){
           printArray(X, N);
        }

        free(X);
        free(tmp);
        return (EXIT_SUCCESS);
}

4 コア マシンでの had-doc ベンチマークでは、次の結果が得られます。

100000000 elements 
1 thread : Time: 11.052081 (s)
2 threads: Time: 5.907508  (s)
4 threads: Time: 4.984839  (s)

A overall Speed up of 2.21x

今後の改善点はGitHubで入手できます。

並列バージョンの高度な C++ バージョンは、ここにあります。最終的なアルゴリズムは次のようになります。

void mergeSortRecursive(vector<double>& v, unsigned long left, unsigned long right) {
   if (left < right) {
      if (right-left >= 32) {
         unsigned long mid = (left+right)/2; 
         #pragma omp taskgroup
         {
            #pragma omp task shared(v) untied if(right-left >= (1<<14))
            mergeSortRecursive(v, left, mid);
            #pragma omp task shared(v) untied if(right-left >= (1<<14))
            mergeSortRecursive(v, mid+1, right);
            #pragma omp taskyield
         }
         inplace_merge(v.begin()+left, v.begin()+mid+1, v.begin()+right+1);
      }else{
         sort(v.begin()+left, v.begin()+right+1);
     }
    }
  }
}


void mergeSort(vector<double>& v) { 
     #pragma omp parallel
     #pragma omp single
     mergeSortRecursive(v, 0, v.size()-1); 
}

6.61x48 スレッドのスピードアップが報告されています。

于 2012-12-10T23:51:52.283 に答える