4.19 mpi: коммуникации «точка-точка». «пинг-понг».

В этом примере рассматривается работа в неблокирующем режиме. Процессы посылают друг другу целое число, всякий раз увеличивая его на 1. Когда число достигнет некоего максимума, переданного в качестве параметра командной строки, все ветви завершаются. Для того, чтобы указать всем последующим ветвям на то, что пора завершаться, процесс, обнаруживший достижение максимума, обнуляет число.

#include <mpi.h>

#include <stdio.h>

int main(int argc, char **argv)

{

          int size, rank, max, i = 1, j, prev, next;

          MPI_Request recv_request, send_request;

          MPI_Status status;

MPI_Init(&argc, &argv); /* Инициализируем библиотеку */

          MPI_Comm_size(MPI_COMM_WORLD, &size);

/* Узнаем количество задач в запущенном приложении... */

MPI_Comm_rank (MPI_COMM_WORLD, &rank);   

/* ...и свой собственный номер: от 0 до (size-1) */

/* определяем максимум */

if ((argc < 2) || ((max = atoi(argv[1])) <= 0)) {

                   printf("Bad arguments! ");

                   MPI_Finalize();

                   return 1;

          }

/* каждая ветвь определяет, кому посылать и от кого принимать сообщения */

next = (rank + 1) \% size;

          prev = (rank == 0) ? (size - 1) : (rank - 1);

          if (rank == 0) {

/* ветвь с номером 0 начинает «пинг-понг» */

MPI_Isend(&i, 1, MPI_INT, 1, next, MPI_COMM_WORLD, &send_request);

printf("process \%d sent value \"\%d\" to process \%d ", rank, i, next);

                   MPI_Wait(&send_request, NULL);

          }

          while ((i > 0) && (i < max)) {

MPI_Irecv(&i, 1, MPI_INT, prev, MPI_ANY_TAG, MPI_COMM_WORLD, &recv_request);

                   MPI_Wait(&recv_request, &status);

                   if (i > 0) {

printf("process \%d received value \"\%d\" " + "from process \%d ", rank, i++, prev);

                             if (i == max) {

                                      i = 0;

                             }

MPI_Isend(&i, 1, MPI_INT, next, 0, MPI_COMM_WORLD, &send_request);

printf("process \%d sent value \"\%d\" " + " to process \%d ", rank, i, next);

                             MPI_Wait(&send_request, NULL);

                   } else if (i == 0) {

MPI_Isend(&i, 1, MPI_INT, next, 0, MPI_COMM_WORLD, &send_request);

                             MPI_Wait(&send_request, NULL);

                             break;

                   }

          }

          MPI_Finalize();

          return 0;

}

Коллективные коммуникации.

Как уже говорилось, помимо коммуникаций «точка-точка», MPI предоставляет различные возможности для осуществления коллективных операций, в которых участвуют все ветви приложения, входящие в определенный коммуникационный контекст. К таким операциям относятся рассылка данных от выделенной ветви всем остальным, сбор данных от всех ветвей на одной выделенной ветви, рассылка данных от всех ветвей ко всем, а также коллективная пересылка данных, совмещенная с их одновременной обработкой.

Отметим, что хотя все те же самые действия можно запрограммировать с использованием коммуникаций «точка-точка», однако более целесообразным является  применение коллективных функций, и тому есть несколько причин. Во-первых, во многих случаях использование коллективной функции может быть эффективнее, нежели моделирование ее поведения при помощи функций обмена «точка-точка», так как реализация коллективных функций может содержать некоторую оптимизацию (например, при рассылке всем ветвям одинаковых данных может использоваться не последовательная посылка всем ветвям сообщений от одной выделенной ветви,  а распространение сообщения между ветвями по принципу двоичного дерева). Кроме того, при использовании коллективных функций сводится к нулю риск программной ошибки, связанной с тем, что сообщение из коллективного потока данных будет перепутано с другим, индивидуальным сообщением (например, получено вместо него при использовании констант MPI_ANY_SOURCE, MPI_ANY_TAG в качестве параметров функции приема сообщения). При передаче сообщений, формируемых коллективными функциями, используется временный дубликат заданного коммуникатора, который неявно создается библиотекой на время выполнения данной функции и уничтожается автоматически после ее завершения. Таким образом, сообщения, переданные коллективной функцией, могут быть получены в других или в той же самой ветви только при помощи той же самой коллективной функции.

Важной особенностью любой коллективной функции является то, что ее поведение будет корректным только в том случае, если она была вызвана во всех ветвях, входящих в заданный коммуникационный контекст. В некоторых (или во всех) ветвях при помощи этой функции данные будут приниматься, в некоторых (или во всех) ветвях – отсылаться. Иногда с помощью одного вызова происходит одновременно и отправка, и прием данных. Как правило, все ветви должны вызвать коллективную функцию с одними и теми же значениями фактических параметров (за исключением параметров, содержащих адреса буферов – эти значения, разумеется, в каждой ветви могут быть своими). Все коллективные функции имеют параметр-коммуникатор, описывающий коммуникационный контекст, в котором происходит коллективный обмен.

Семантика буферизации для коллективных функций аналогична стандартному режиму буферизации для одиночных коммуникаций в блокирующем режиме: возврат управления из коллективной функции означает, что вызвавшая ветвь может повторно использовать буфер, в котором содержались данные для отправки (либо гарантирует, что в буфере для приема находятся принятые данные). При этом, отнюдь не гарантируется, что в остальных ветвях к этому моменту данная коллективная операция также завершилась (и даже не гарантируется, что она в них уже началась). Таким образом, хотя в некоторых случаях возврат из коллективных функций действительно происходит во всех ветвях одновременно, нельзя полагаться на этот факт и использовать его для целей синхронизации ветвей. Для явной синхронизации ветвей может использоваться лишь одна специально предназначенная коллективная функция – это описанная выше функция MPI_Barrier().

Коллективный обмен данными.

Для отправки сообщений от одной выделенной ветви всем ветвям, входящим в данный коммуникационный контекст, служат функции MPI_Bcast() и MPI_Scatter(). Функция MPI_Bcast() служит для отправки всем ветвям одних и тех же данных:

#include <mpi.h>

int MPI_Bcast(void* buffer, int count, MPI_Datatype datatype, int root, MPI_Comm comm);

Параметр comm задает коммуникатор, в рамках которого осуществляется коллективная операция. Параметр root задает номер ветви, которая выступает в роли отправителя данных. Получателями выступают все ветви из данного коммуникационного контекста (включая и ветвь-отправитель). При этом параметр buffer во всех ветвях задает адрес буфера, но ветвь-отправитель передает в этом буфере данные для отправки, а в ветвях-получателях данный буфер используется для приема данных. После возврата из данной функции в буфере каждой ветви будут записаны данные, переданные ветвью с номером root.

Параметры datatype и count задают соответственно тип данных, составляющих тело сообщения, и количество элементов этого типа в теле сообщения. Их значения должны совпадать во всех ветвях. Разумеется, для корректного выполнения функции необходимо, чтобы размер буфера, переданного каждой ветвью в качестве параметра buffer, был достаточен для приема необходимого количества данных.

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

#include <mpi.h>

int MPI_Scatter(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm);

Параметр root здесь опять задает номер выделенной ветви, являющейся отправителем данных, параметр comm – коммуникатор, в рамках которого осуществляется обмен. Параметры sendbuf, sendtype, sendcount имеют смысл только для ветви-отправителя и задают соответственно адрес буфера с данными для рассылки, их тип и количество элементов заданного типа, которое нужно отправить каждой из ветвей. Для остальных ветвей эти параметры игнорируются.  В результате действия функции массив данных в sendbuf делится на N равных частей (где N – количество ветвей в коммуникаторе), и каждой ветви посылается i-я часть этого массива, где i – уникальный номер данной ветви в этом коммуникаторе.  Отметим, что для того, чтобы вызов был корректным, буфер sendbuf должен, очевидно, содержать N*sendcount элементов (ответственность за это возлагается на программиста).

Параметры recvbuf, recvtype, recvcount имеют значение для всех ветвей (в том числе и ветви-отправителя) и задают адрес буфера для приема данных, тип принимаемых данных и их количество. Формально типы отправляемых и принимаемых данных могут не совпадать, однако жестко задается ограничение, в соответствии с которым общий размер данных, отправляемых ветви, должен точно совпадать с размером данных, которые ею принимаются.

Операцию, обратную MPI_Scatter(), – сбор порций данных от всех ветвей на одной выделенной ветви – осуществляет функция MPI_Gather():

#include <mpi.h>

int MPI_Gather(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm);

Параметр root задает номер ветви-получателя данных; отправителями являются все ветви из данного коммуникационного контекста (включая и ветвь с номером root). В результате выполнения этой функции в буфере recvbuf у ветви с номером root формируется массив, составленный из N равных частей, где i-я часть представляет собой содержимое буфера sendbuf у ветви с номером i (т.е. порции данных от всех ветвей располагаются в порядке их номеров). Параметры sendbuf, sendtype, sendcount должны задаваться всеми ветвями и описывают отправляемые данные; параметры recvbuf, recvtype, recvcount описывают буфер для приема данных, а также количество и тип принимаемых данных и имеют значение только для ветви с номером root, а у остальных ветвей игнорируются. Для корректной работы функции необходимо, чтобы буфер recvbuf имел достаточную емкость, чтобы вместить данные от всех ветвей.

Работу функций MPI_Scatter() и MPI_Gather() для случая 3х ветвей и root=0 наглядно иллюстрирует Рис. 24.

Рис. 24 Работа MPI_scatter() и MPI_gather()

Существует также возможность осуществить сбор данных от всех ветвей в единый массив, но так, чтобы доступ к этому результирующему массиву имела не одна выделенная ветвь, а все ветви, принимающие участие в операции. Для этого служит функция MPI_Allgather():

#include <mpi.h>

int MPI_Allgather(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, MPI_Comm comm);

Работа этой функции проиллюстрирована на Рис. 25. Эта функция отличается от предыдущей лишь тем, что у нее отсутствует параметр root, а параметры recvbuf, recvtype, recvcount имеют смысл для всех ветвей. В результате работы этой функции на каждой из ветвей в буфере recvbuf формируется результирующий массив, аналогично тому, как описано для MPI_Gather(). Ответственность за то, чтобы приемные буфера имели достаточную емкость, возлагается на программиста.

Рис. 25 Работа MPI_Allgather()

Функция MPI_Alltoall() представляет собой расширение MPI_Allgather(), заключающееся в том, что каждая ветвь-отправитель посылает каждой конкретной ветви-получателю свою отдельную порцию данных, подобно тому, как это происходит в MPI_Scatter(). Другими словами, i-я часть данных, посланных ветвью с номером j, будет получена ветвью с номером i и размещена в j-м блоке ее результирующего буфера.

#include <mpi.h>

int MPI_Alltoall(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, MPI_Comm comm);

Параметры этой функции аналогичны параметрам MPI_Allgather().  

Рис. 26 Работа функции MPI_Alltoall()

Помимо рассмотренных выше функций, библиотека MPI предоставляет так называемые «векторные» аналоги функций MPI_Scatter(), MPI_Gather(), MPI_Allgather() и MPI_Alltoall(), позволяющие разным ветвям пересылать или получать части массива разных длин, однако их подробное рассмотрение выходит за рамки данного пособия.