Параллельная пузырьковая сортировка заблокирована

Question

Моя программа OpenMP блокирует первый «для» l oop следующего кода без видимой причины. Я просто пытаюсь распараллелить Bubble Sort.

Ниже приведен полный код, воспроизводящий проблему:

#include <stdint.h>
#include <stdbool.h>
#include <stdlib.h>
#include <omp.h>

static int N_THREADS;
#define CHUNK_SIZE (size/N_THREADS)

void
parallel_bubble_sort(uint64_t *T, const uint64_t size)
{
    register bool swapped;
    register uint64_t swap;
    register int i,j;

    #pragma omp parallel private(swap,i,j)
    do {
        swapped = false;

        #pragma omp for schedule(static) reduction(||:swapped)
        for (j=0; j<N_THREADS; j++)
        for (i=j*CHUNK_SIZE+1; i<=(j+1)*CHUNK_SIZE-1; i++)
        if (T[i-1] > T[i]) {
            swap = T[i-1];
            T[i-1] = T[i];
            T[i] = swap;
            swapped = true;
        }

        #pragma omp for schedule(static) reduction(||:swapped)
        for (i=CHUNK_SIZE-1; i<size-CHUNK_SIZE; i+=CHUNK_SIZE)
        if (T[i] > T[i+1]) {
            swap = T[i];
            T[i] = T[i+1];
            T[i+1] = swap;
            swapped = true;
        }
    } while(swapped);
}

int main ()
{
    uint64_t i;
    uint64_t N = 1024;
    N_THREADS = omp_get_max_threads();

    uint64_t *X = (uint64_t *) malloc(N * sizeof(uint64_t));
    for (i = 0 ; i < N ; i++) X[i] = N-i;

    parallel_bubble_sort(X, N);
    free(X);
}

Некоторый дополнительный контекст:

• T * указатель на массив типа uint64_t
• размер это размер массива
• CHUNK_SIZE это просто size / NUM_THREADS (что также является значением размера фрагмента по умолчанию, которое OpenMP использует в stati c режим планирования, поэтому я должен получить то же поведение, если я удалю это из предложения)

Относительно логики c за кодом:

• В первом l oop, я делю свой массив на куски и размножаю пузырьки отдельно, не перекрывая нити
• Во втором l oop я проверяю, пузырьки распространяются на границах

Подробнее о проблеме, возникшей у меня во время выполнения:

• Моя программа зависла при первом "for" l oop. Я локализовал, где программа блокируется с помощью #pragma omp single и простого оператора печати. ​​

Answer 1

Причиной тупика является состояние гонки данных в вашем внешнем l oop:

do {
   swapped = false;  // <--- here
   ...
} while(swapped);    // <--- here

Гонка происходит, потому что нет никакой гарантии, что все потоки прибудут к инструкции, реализующей while(swapped) условно одновременно. Представьте, что у вас есть две темы. Поток 0 завершает вторую параллель l oop, видит, что swapped равен true, проходит через условие l oop и затем снова запускает тело l oop, устанавливая swapped в false. Если поток 1 достигает условного до того, как поток 0 сможет установить swapped в false, он также начнет новую итерацию. Но если поступит слишком поздно, swapped будет false, а l oop завершится. В результате поток 1 не присоединится к параллельной l oop, а поток 0 будет всегда ждать у неявного барьера синхронизации.

Решение состоит в том, чтобы убедиться, что все потоки имеют согласованное представление о том, какое значение swapped - это когда они принимают решение, начинать ли новую итерацию или нет. Самое простое решение - вставить барьер непосредственно перед установкой swapped на false:

do {
   #pragma omp barrier
   swapped = false;
   ...
} while(swapped);

Кроме того, сброс всех потоков swapped не является действительно необходимым, и это может быть возможно (не уверен в этом ) go против OpenMP spe c, который запрещает одновременный доступ к исходной переменной до завершения сокращения. Я не уверен, применимо ли это к изменениям до области сокращения (как я не был уверен пару лет go), и был удален абзац из OpenMP 4.5 spe c относительно одновременного доступа, но для безопасности я бы дал обработку single:

do {
   #pragma omp barrier
   #pragma omp single
   swapped = false;
   ...
} while(swapped);

Answer 2

Обратите внимание, что omp_get_max_threads() соответствует максимальному количеству потоков, которое может быть назначено любой команде, выполняющей параллельный регион, но в целом вы не гарантированно получите это количество потоков в данном параллельном регионе. Даже если вы запросите указанное число c потоков с помощью предложения num_threads для директивы OMP, вы все равно можете получить меньше. Хотя в вашей конкретной программе вы должны получить полное число потоков, это плохая форма, чтобы зависеть от этого.

Вместо этого используйте omp_get_num_threads() внутри параллельной области, чтобы определить, сколько потоков на самом деле в команде, выполняющей регион. Я также предлагаю использовать omp_get_thread_num() для получения номера текущего потока в команде, что позволит вам планировать свои итерации l oop вручную, что наиболее удобно, когда алгоритм фактически зависит от того, как они запланированы, как у вас. Кроме того, используйте тот факт, что переменные, объявленные внутри параллельной области, автоматически являются частными по отношению к потокам, выполняющим эту область. В сочетании с объявлением ваших переменных в самой узкой области, это сократит количество необходимых вам разделов данных.

Но все это не решит вашу проблему для меня. То, что делает , разрешает ее (после применения вышеизложенного), перемещая параллельную директиву omp от do к do и соответствующему блоку. Это следует интерпретировать как вызов параллельного выполнения блока, а не самого do. И это не должно быть проблемой для вас, потому что вы все равно хотите барьер в конце каждого выполнения блока. Вам также нужен барьер между вашими двумя внутренними гнездами l oop, чтобы избежать гонок данных.

Объединение всего этого, плюс немного больше реорганизации, дает это, что работает для меня ^* :

void parallel_bubble_sort(uint64_t *T, const uint64_t size) {
    bool swapped;

    do {
        swapped = false;
        #pragma omp parallel
        {
            register uint64_t swap;
            register int i;
            int n_threads = omp_get_num_threads();
            int thread_num = omp_get_thread_num();
            int chunk_size = size / n_threads;

            for (i = thread_num * chunk_size + 1;
                    i < (thread_num + 1) * chunk_size;
                    i++) {
                if (T[i - 1] > T[i]) {
                    swap = T[i - 1];
                    T[i - 1] = T[i];
                    T[i] = swap;
                    swapped = true;
                }
            }
            #pragma omp barrier

            if (i < size && T[i - 1] > T[i]) {
                swap = T[i - 1];
                T[i - 1] = T[i];
                T[i] = swap;
                swapped = true;
            }
        }
    } while(swapped);
}

---

^* Он «работает» до (неполной) степени, в которой алгоритм корректен. Алгоритм как написано не является правильным, если только размер массива не кратен числу потоков, выполняющих параллельную область. У моей машины 12 логических ядер (6 физических), а 1024 не кратно 6. Когда я запускаю программу, указанную выше, я получаю несколько конечных элементов, которые не сортируются. Подобное может произойти на любой машине, потому что, опять же, вы не уверены, что получите полное количество ядер по вашему запросу. Исправление этой проблемы оставлено в качестве упражнения.