Блог

Я снова застрял при использовании OpenMP в C . На этот раз я пытаюсь реализовать параллельную быструю сортировку.

Код:

 #include <iostream>
#include <vector>
#include <stack>
#include <utility>
#include <omp.h>
#include <stdio.h>

#define SWITCH_LIMIT 1000

using namespace std;

template <typename T>
void insertionSort(std::vector<T> amp;v, int q, int r)
{
    int key, i;
    for(int j = q   1; j <= r;   j)
    {
        key = v[j];
        i = j - 1;
        while( i >= q amp;amp; v[i] > key )
        {
            v[i 1] = v[i];
            --i;
        }
        v[i 1] = key;
    }
}

stack<pair<int,int> > s;

template <typename T>
void qs(vector<T> amp;v, int q, int r)
{
    T pivot;
    int i = q - 1, j = r;
    //switch to insertion sort for small data
    if(r - q < SWITCH_LIMIT) 
    {
        insertionSort(v, q, r);
        return;
    }

    pivot = v[r];
    while(true)
    {
        while(v[  i] < pivot);
        while(v[--j] > pivot);
        if(i >= j) break;
        std::swap(v[i], v[j]); 
    }
    std::swap(v[i], v[r]);

    #pragma omp critical
    {
        s.push(make_pair(q, i - 1));
        s.push(make_pair(i   1, r));        
    }
}

int main()
{
    int n, x;
    int numThreads = 4, numBusyThreads = 0;
    bool *idle = new bool[numThreads];
    for(int i = 0; i < numThreads;   i)
        idle[i] = true;
    pair<int, int> p;
    vector<int> v;
    cin >> n;
    for(int i = 0; i < n;   i)
    {
        cin >> x;
        v.push_back(x);
    }
    cout << v.size() << endl;
    s.push(make_pair(0, v.size()));

    #pragma omp parallel shared(s, v, idle, numThreads, numBusyThreads, p) 
    {
        bool done = false;
        while(!done) 
        {
            int id = omp_get_thread_num();
            #pragma omp critical
            {
                if(s.empty() == false amp;amp; numBusyThreads < numThreads) 
                {
                      numBusyThreads;
                    //the current thread is not idle anymore
                    //it will get the interval [q, r] from stack
                    //and run qs on it
                    idle[id] = false;
                    p = s.top();                    
                    s.pop();
                }
                if(numBusyThreads == 0)
                {
                    done = true;
                }
            }
            if(idle[id] == false)
            {

                qs(v, p.first, p.second);
                idle[id] = true;
                #pragma omp critical 
                --numBusyThreads;
            }

        }
    }
    return 0;
}
  

Алгоритм:

Чтобы использовать OpenMP для рекурсивной функции, я использовал стек для отслеживания следующих интервалов, на которых должна выполняться функция qs. Я вручную добавляю 1-й интервал [0, размер], а затем позволяю потокам приступить к работе, когда в стек добавляется новый интервал.

Проблема:

Программа завершается слишком рано, не сортируя массив после создания 1-го набора интервалов ([q, i — 1], [i 1, r], если вы посмотрите на код. Я предполагаю, что потоки, которые получают работу, рассматривают локальные переменные функции быстрой сортировки (qs в коде), совместно используемые по умолчанию, поэтому они путают их и не добавляют интервал в стек.

Как я компилирую:

 g   -o qs qs.cc -Wall -fopenmp
  

Как я запускаю:

./qs < in_100000 > out_100000

где in_100000 — это файл, содержащий 100000 в 1-й строке, за которым следуют 100 тыс. промежуточных символов в следующей строке, разделенных пробелами.

Я использую gcc 4.5.2 в Linux

Спасибо за вашу помощь,

Dan

На самом деле я не запускал ваш код, но я вижу немедленную ошибку p , которой не должно быть private shared . При параллельном вызове qs : qs(v, p.first, p.second); будут запущены гонки p , что приведет к непредсказуемому поведению. Локальные переменные в qs должны быть в порядке, потому что все потоки имеют свой собственный стек. Тем не менее, общий подход хорош. Вы на правильном пути.

Вот мои общие комментарии по реализации параллельной быстрой сортировки. Быстрая сортировка сама по себе смущающе параллельна, что означает, что синхронизация не требуется. Рекурсивные вызовы qs в секционированном массиве ошеломляюще параллельны.

Однако параллелизм представлен в рекурсивной форме. Если вы просто используете вложенный параллелизм в OpenMP, в итоге у вас будет тысяча потоков в секунду. Никакого ускорения получено не будет. Итак, в основном вам нужно превратить рекурсивный алгоритм в интерактивный. Затем вам нужно реализовать своего рода рабочую очередь. Это ваш подход. И это непросто.

Для вашего подхода есть хороший тест: OmpSCR. Вы можете скачать наhttp://sourceforge.net/projects/ompscr /

В тесте есть несколько версий быстрой сортировки на основе OpenMP. Большинство из них похожи на ваши. Однако, чтобы увеличить параллелизм, необходимо свести к минимуму конфликт в глобальной очереди (в вашем коде это s ). Итак, может быть несколько оптимизаций, таких как наличие локальных очередей. Хотя сам алгоритм является чисто параллельным, для реализации могут потребоваться артефакты синхронизации. И, прежде всего, очень сложно добиться ускорения.

Тем не менее, вы по-прежнему напрямую используете рекурсивный параллелизм в OpenMP двумя способами: (1) Регулируя общее количество потоков и (2) используя OpenMP 3.0 task .

Вот псевдокод для первого подхода (он основан только на тесте OmpSCR):

 void qsort_omp_recursive(int* begin, int* end)
{
  if (begin != end) {
    // Partition ...

    // Throttling
    if (...)  {
      qsort_omp_recursive(begin, middle);
      qsort_omp_recursive(  middle,   end);
    } else {

#pragma omp parallel sections nowait
      {
#pragma omp section
        qsort_omp_recursive(begin, middle);
#pragma omp section
        qsort_omp_recursive(  middle,   end);
      }
    }
  }
}
  

Для того, чтобы запустить этот код, вам нужно вызвать omp_set_nested(1) и omp_set_num_threads(2) . Код действительно прост. Мы просто создаем два потока для разделения работы. Однако мы вставляем простую логику регулирования, чтобы предотвратить избыточные потоки. Обратите внимание, что мои эксперименты показали приличное ускорение для этого подхода.

Наконец, вы можете использовать OpenMP 3.0 task , где задача является логически параллельной работой. Во всех описанных выше подходах OpenMP каждая параллельная конструкция порождает два физических потока. Вы можете сказать, что существует жесткое сопоставление 1 к 1 между задачей и рабочим потоком. Однако task разделяет логические задачи и рабочих.

Поскольку OpenMP 3.0 еще не популярен, я буду использовать Cilk Plus, который отлично подходит для выражения такого рода вложенных и рекурсивных параллелизмов. В Cilk Plus распараллеливание чрезвычайно просто:

 void qsort(int* begin, int* end)
{
  if (begin != end) {
    --end;
    int* middle = std::partition(begin, end,
      std::bind2nd(std::less<int>(), *end));
    std::swap(*end, *middle);

    cilk_spawn qsort(begin, middle);
    qsort(  middle,   end);
    // cilk_sync; Only necessay at the final stage.
  }
}
  

Я скопировал этот код из примера кода Cilk Plus. Вы увидите, что единственное ключевое слово cilk_spawn — это все для распараллеливания быстрой сортировки. Я пропускаю объяснения ключевого слова Cilk Plus и spawn. Однако это легко понять: два рекурсивных вызова объявлены как логически параллельные задачи. Всякий раз, когда происходит рекурсия, создаются логические задачи. Но среда выполнения Cilk Plus (которая реализует эффективный планировщик кражи работы) справится со всеми видами грязной работы. Это оптимально ставит параллельные задачи в очередь и сопоставляет их с рабочими потоками.

Обратите внимание, что OpenMP 3.0 task по сути аналогичен подходу Cilk Plus. Мои эксперименты показывают, что было возможно довольно хорошее ускорение. Я получил ускорение в 3-4 раза на 8-ядерном компьютере. И ускорение было масштабируемым. Абсолютное ускорение Cilk Plus больше, чем у OpenMP 3.0.

Подход Cilk Plus (и OpenMP 3.0) и ваш подход по существу одинаковы: разделение параллельной задачи и назначения рабочей нагрузки. Однако это очень сложно эффективно реализовать. Например, необходимо уменьшить конкуренцию и использовать структуры данных без блокировки.