Нити для межпроцессного взаимодействия в OpenMP

Question

У меня есть OpenMP распараллелить программу, которая выглядит следующим образом:

[...] 
#pragma omp parallel 
{ 
//initialize threads 

#pragma omp for 
for(...) 
    { 
    //Work is done here 

    } 

} 

Теперь я добавляю поддержка MPI. То, что мне понадобится, это поток, который обрабатывает сообщение, в моем случае, вызывает GatherAll все время и заполняет/освобождает связанный список для приема/отправки данных из других процессов. Этот поток должен отправлять/получать, пока не будет установлен флаг. Итак, прямо сейчас в примере нет файлов MPI, мой вопрос касается реализации этой процедуры в OpenMP. Как реализовать такой поток? Например, я попытался ввести единую директиву здесь:

[...] 
int kill=0 
#pragma omp parallel shared(kill) 
{ 
//initialize threads 
#pragma omp single nowait 
{ 
    while(!kill) 
    send_receive(); 
} 
#pragma omp for 
for(...) 
    { 
    //Work is done here 

    } 
kill=1 

} 

, но в этом случае программа застревает, потому что неявный барьер после того, как ждет-петли для резьбы в то время как петли выше.

спасибо, rugermini.

Answer 1

Вы можете попробовать добавить пункт nowait к вашей single конструкции:

EDIT: отвечая на первый комментарий

Если включить вложенный параллелизм OpenMP, вы могли бы добиться того, что вы хотите делая два уровня параллелизма. На верхнем уровне у вас есть два параллельных параллельных раздела: один для MPI-связи, другой для локальных вычислений. Этот последний раздел сам по себе может быть распараллелен, что дает вам второй уровень параллелизации. Только потоки, выполняющие этот уровень, будут затронуты барьерами в нем.

#include <iostream> 
#include <omp.h> 

int main() 
{ 
    int kill = 0; 
#pragma omp parallel sections 
    { 
#pragma omp section 
    { 
     while (kill == 0){ 
     /* manage MPI communications */ 
     } 
    } 

#pragma omp section 
    { 
#pragma omp parallel 
#pragma omp for 
     for (int i = 0; i < 10000 ; ++i) { 
     /* your workload */ 
     } 
     kill = 1; 
    } 
    } 
} 

Однако, вы должны знать, что ваш код будет перерыв, если вы не имеете по крайней мере, два потока, а это значит, вы нарушаете предположение, что последовательные и Параллельные версии коды должны сделать тоже самое.

Было бы намного чище обернуть ваше ядро ​​OpenMP внутри более глобальной схемы обмена MPI (потенциально используя асинхронную связь для перекрытия связи с вычислениями).

Answer 2

Хммм. Если вы действительно добавляете поддержку MPI в свою программу, то вы должны использовать mpi_allgather, так как mpi_gatherall не существует. Обратите внимание, что mpi_allgather - это коллективная операция, то есть все процессы в коммуникаторе называют ее. У вас не может быть процесс сбора данных, в то время как другие процессы делают то, что они делают. Что вы можете сделать, так это использование MPI односторонней связи для реализации вашей идеи; это будет немного сложнее, но не более того, если один процесс только считывает память других процессов.

Я озадачен вашим использованием термина «поток» по MPI. Я боюсь, что вы запутываете OpenMP и MPI, один из вариантов которого называется OpenMPI. Несмотря на это имя, он отличается от OpenMP как мел от сыра. Программы MPI записываются в терминах процессов, а не потоков. Типичная реализация OpenMP действительно использует потоки, хотя детали, как правило, хорошо скрыты от программиста.

Я серьезно впечатлен тем, что вы пытаетесь или пытаетесь использовать MPI внутри своего кода OpenMP. Это как раз противоположность работы, которую я делаю, и видеть, как другие делают на некоторых серьезно больших компьютерах. Стандартный режим для такой «гибридной» параллелизации - это писать программы MPI, которые вызывают код OpenMP. Многие из сегодняшних очень крупных компьютеров включают в себя коллекции, которые, по сути, многоядерные коробки.Типичным подходом к программированию одного из них является наличие одного процесса MPI на каждом ящике и для каждого из этих процессов использовать один поток OpenMP для каждого ядра в поле.

Answer 3

Вы должны быть осторожны, потому что вы не можете просто использовать поток вызовов MPI «пропустить» цикл omp; все потоки в команде потока должны пройти цикл for.

Там есть пара способов, вы могли бы сделать это: с вложенной parallism и задачами, вы можете запустить одну задачу, чтобы сделать передачи сообщений и пыльники, чтобы вызвать режим работы, который имеет OMP параллель в ней:

#include <mpi.h> 
#include <omp.h> 
#include <stdio.h> 

void work(int rank) { 
    const int n=14; 
    #pragma omp parallel for 
    for (int i=0; i<n; i++) { 
     int tid = omp_get_thread_num(); 
     printf("%d:%d working on item %d\n", rank, tid, i); 
    } 
} 

void sendrecv(int rank, int sneighbour, int rneighbour, int *data) { 
    const int tag=1; 
    MPI_Sendrecv(&rank, 1, MPI_INT, sneighbour, tag, 
        data, 1, MPI_INT, rneighbour, tag, 
        MPI_COMM_WORLD, MPI_STATUS_IGNORE); 
} 

int main(int argc, char **argv) { 
    int rank, size; 
    int sneighbour; 
    int rneighbour; 
    int data; 
    int got; 

    MPI_Init_thread(&argc, &argv, MPI_THREAD_FUNNELED, &got); 
    MPI_Comm_size(MPI_COMM_WORLD,&size); 
    MPI_Comm_rank(MPI_COMM_WORLD,&rank); 

    omp_set_nested(1); 
    sneighbour = rank + 1; 
    if (sneighbour >= size) sneighbour = 0; 
    rneighbour = rank - 1; 
    if (rneighbour <0) rneighbour = size-1; 

    #pragma omp parallel 
    { 
     #pragma omp single 
     { 
      #pragma omp task 
      { 
       sendrecv(rank, sneighbour, rneighbour, &data); 
       printf("Got data from %d\n", data); 
      } 

      #pragma omp task 
      work(rank); 
     } 
    } 


    MPI_Finalize(); 
    return 0; 
} 

в качестве альтернативы, вы можете сделать свой OMP для цикла schedule(dynamic) так что другие потоки могут забрать некоторые из слабины от в то время как основной поток посылает, а главный поток может подобрать какую-то работу, когда это делается:

#include <mpi.h> 
#include <omp.h> 
#include <stdio.h> 

void sendrecv(int rank, int sneighbour, int rneighbour, int *data) { 
    const int tag=1; 
    MPI_Sendrecv(&rank, 1, MPI_INT, sneighbour, tag, 
        data, 1, MPI_INT, rneighbour, tag, 
        MPI_COMM_WORLD, MPI_STATUS_IGNORE); 
} 

int main(int argc, char **argv) { 
    int rank, size; 
    int sneighbour; 
    int rneighbour; 
    int data; 
    int got; 
    const int n=14; 

    MPI_Init_thread(&argc, &argv, MPI_THREAD_FUNNELED, &got); 
    MPI_Comm_size(MPI_COMM_WORLD,&size); 
    MPI_Comm_rank(MPI_COMM_WORLD,&rank); 

    omp_set_nested(1); 
    sneighbour = rank + 1; 
    if (sneighbour >= size) sneighbour = 0; 
    rneighbour = rank - 1; 
    if (rneighbour <0) rneighbour = size-1; 

    #pragma omp parallel 
    { 
      #pragma omp master 
      { 
       sendrecv(rank, sneighbour, rneighbour, &data); 
       printf("Got data from %d\n", data); 
      } 

      #pragma omp for schedule(dynamic) 
      for (int i=0; i<n; i++) { 
       int tid = omp_get_thread_num(); 
       printf("%d:%d working on item %d\n", rank, tid, i); 
      } 
    } 


    MPI_Finalize(); 
    return 0; 
}