Неблокирующие обмены

Презентацию к данной лекции Вы можете скачать здесь.

Общая характеристика неблокирующих обменов

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

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

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

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

Неблокирующий обмен выполняется в два этапа:

Разделение этих шагов делает необходимым маркировку каждой операции обмена, которая позволяет целенаправленно выполнять проверки завершения соответствующих операций.

Для маркировки в неблокирующих операциях используются идентификаторы операций обмена

Неблокирующие передача и приём

Инициализация неблокирующей стандартной передачи выполняется подпрограммами MPI_I[S, B, R]send. Стандартная неблокирующая передача выполняется подпрограммой:

int MPI_Isend(void *buf, int count, MPI_Datatype datatype, int dest, int tag, MPI_Comm comm, MPI_Request *request)

MPI_Isend(buf, count, datatype, dest, tag, comm, request, ierr)

Входные параметры этой подпрограммы аналогичны аргументам подпрограммы MPI_Send.

Выходной параметр request - идентификатор операции

Инициализация неблокирующего приема выполняется при вызове подпрограммы:

int MPI_Irecv(void *buf, int count, MPI_Datatype datatype, int source, int tag, MPI_Comm comm, MPI_Request *request)

MPI_Irecv(buf, count, datatype, source, tag, comm, request, ierr)

Назначение аргументов здесь такое же, как и в ранее рассмотренных подпрограммах, за исключением того, что указывается ранг не адресата, а источника сообщения (source).

Вызовы подпрограмм неблокирующего обмена формируют запрос на выполнение операции обмена и связывают его с идентификатором операции request. Запрос идентифицирует свойства операции обмена:

Запрос содержит информацию о состоянии ожидающих обработки операций обмена и может быть использован для получения информации о состоянии обмена или для ожидания его завершения.

Проверка выполнения неблокирующих обменов

Проверка выполнения обмена

Проверка фактического выполнения передачи или приема в неблокирующем режиме осуществляется с помощью вызова подпрограмм ожидания, блокирующих работу процесса до завершения операции или неблокирующих подпрограмм проверки, возвращающих логическое значение "истина", если операция выполнена

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

Есть три типа таких проверок:

Каждая из этих проверок имеет две разновидности:

Блокирующие операции проверки

Подпрограмма MPI_Wait блокирует работу процесса до завершения приема или передачи сообщения:

int MPI_Wait(MPI_Request *request, MPI_Status *status)

MPI_Wait(request, status, ierr)

Входной параметр request - идентификатор операции обмена, выходной - статус (status).

Успешное выполнение подпрограммы MPI_Wait после вызова MPI_Ibsend подразумевает, что буфер передачи можно использовать вновь, то есть пересылаемые данные отправлены или скопированы в буфер, выделенный при вызове подпрограммы MPI_Buffer_attach.

В этот момент уже нельзя отменить передачу. Если не будет зарегистрирован соответствующий прием, буфер нельзя будет освободить. В этом случае можно применить подпрограмму MPI_Cancel, которая освобождает память, выделенную подсистеме коммуникаций.

Проверка завершения всех обменов

Проверка завершения всех обменов выполняется подпрограммой:

int MPI_Waitall(int count, MPI_Request requests[], MPI_Status statuses[])

MPI_Waitall(count, requests, statuses, ierr)

При вызове этой подпрограммы выполнение процесса блокируется до тех пор, пока все операции обмена, связанные с активными запросами в массиве requests, не будут выполнены. Возвращается статус этих операций. Статус обменов содержится в массиве statuses. count - количество запросов на обмен (размер массивов requests и statuses).

В результате выполнения подпрограммы MPI_Waitall запросы, сформированные неблокирующими операциями обмена, аннулируются, а соответствующим элементам массива присваивается значение MPI_REQUEST_NULL.

В случае неуспешного выполнения одной или более операций обмена подпрограмма MPI_Waitall возвращает код ошибки MPI_ERR_IN_STATUS и присваивает полю ошибки статуса значение кода ошибки соответствующей операции.

Если операция выполнена успешно, полю присваивается значение MPI_SUCCESS, а если не выполнена, но и не было ошибки - значение MPI_ERR_PENDING. Это соответствует наличию запросов на выполнение операции обмена, ожидающих обработки.

Проверка завершения любого числа обменов

Проверка завершения любого числа обменов выполняется подпрограммой:

int MPI_Waitany(int count, MPI_Request requests[], int *index, MPI_Status *status)

MPI_Waitany(count, requests, index, status, ierr)

Выполнение процесса блокируется до тех пор, пока, по крайней мере, один обмен из массива запросов (requests) не будет завершен.

Входные параметры:

Выходные параметры:

Неблокирующие процедуры проверки

Подпрограмма MPI_Test выполняет неблокирующую проверку завершения приема или передачи сообщения:

int MPI_Test(MPI_Request *request, int *flag, MPI_Status *status)

MPI_Test(request, flag, status, ierr)

Входной параметр: идентификатор операции обмена request.

Выходные параметры:

Неблокирующая проверка завершения всех обменов

Подпрограмма MPI_Testall выполняет неблокирующую проверку завершения приема или передачи всех сообщений:

int MPI_Testall(int count, MPI_Request requests[], int *flag, MPI_Status statuses[])

MPI_Testall(count, requests, flag, statuses, ierr)

При вызове возвращается значение флага (flag) "истина", если все обмены, связанные с активными запросами в массиве requests, выполнены. Если завершены не все обмены, флагу присваивается значение "ложь", а массив statuses не определен.

Параметр count - количество запросов.

Каждому статусу, соответствующему активному запросу, присваивается значение статуса соответствующего обмена.

Неблокирующая проверка любого числа обменов

Подпрограмма MPI_Testany выполняет неблокирующую проверку завершения приема или передачи сообщения:

int MPI_Testany(int count, MPI_Request requests[], int *index, int *flag, MPI_Status *status)

MPI_Testany(count, requests, index, flag, status, ierr)

Смысл и назначение параметров этой подпрограммы те же, что и для подпрограммы MPI_Waitany. Дополнительный аргумент flag, принимает значение "истина", если одна из операций завершена.

Блокирующая подпрограмма MPI_Waitany и неблокирующая MPI_Testany взаимозаменяемы, как и другие аналогичные пары.

Примеры использования неблокирующих двухточечных обменов

Пример

program main_mpi
include 'mpif.h'
integer rank, tag, cnt, ierr, status(MPI_STATUS_SIZE)
integer request
real sndbuf(5) /1., 2., 3., 4., 5./
real rcvbuf(5)
cnt = 5
tag = 0
call MPI_Init(ierr)
call MPI_Comm_rank(MPI_COMM_WORLD, rank, ierr)
…
if(rank.eq.0) then
call MPI_Isend(sndbuf(1), cnt, MPI_REAL, 1, tag, MPI_COMM_WORLD, request, ierr)
print *, "process ", rank, " send before Wait", sndbuf
call MPI_Wait(request, status, ierr)
print *, "process ", rank, " send after Wait", sndbuf
else
call MPI_Irecv(rcvbuf(1), cnt, MPI_REAL, 0, tag, MPI_COMM_WORLD, request, ierr)
print *, "process ", rank, " received before Wait", rcvbuf
call MPI_Wait(request, status, ierr)
print *, "process ", rank, " received after Wait", rcvbuf
end if
call MPI_Finalize(ierr)
stop
end

Пример

program main_mpi
include 'mpif.h'
integer rank, tag1, tag2, cnt, ierr, status(MPI_STATUS_SIZE)
integer request
real sndbuf1, sndbuf2, rcvbuf1, rcvbuf2
cnt = 1
tag = 0
sndbuf1 = 3.14159
sndbuf2 = 2.71828
call MPI_Init(ierr)
call MPI_Comm_rank(MPI_COMM_WORLD, rank, ierr)
…
if (rank.eq.0) then
call MPI_Send(sndbuf1, cnt, MPI_REAL, 1, tag1, MPI_COMM_WORLD, ierr)
print *, "process ", rank, " send ", sndbuf1
call MPI_Send(sndbuf2, cnt, MPI_REAL, 1, tag2, MPI_COMM_WORLD, ierr)
print *, "process ", rank, " send ", sndbuf2
else
call MPI_Irecv(rcvbuf1, cnt, MPI_REAL, 0, tag1, MPI_COMM_WORLD, request, ierr)
call MPI_Recv(rcvbuf2, cnt, MPI_REAL, 0, tag2, MPI_COMM_WORLD, status, ierr)
print *, "process ", rank, " received before Wait", rcvbuf1
print *, "process ", rank, " received before Wait", rcvbuf2
call MPI_Wait(request, status, ierr)
print *, "process ", rank, " received after Wait", rcvbuf1
print *, "process ", rank, " received after Wait", rcvbuf2
end if
call MPI_Finalize(ierr)
end

Подпрограммы-пробники

Неблокирующая проверка сообщения

Неблокирующая проверка сообщения выполняется подпрограммой:

int MPI_Iprobe(int source, int tag, MPI_Comm comm, int *flag, MPI_Status *status)

MPI_Iprobe(source, tag, comm, flag, status, ierr)

Входные параметры этой подпрограммы те же, что и у подпрограммы MPI_Probe. Выходные параметры:

Если сообщение уже поступило и может быть принято, возвращается значение флага "истина".

Размер полученного сообщения (count) можно определить с помощью вызова подпрограммы

int MPI_Get_count(MPI_Status *status, MPI_Datatype datatype, int *count)

MPI_Get_count(status, datatype, count, ierr)

Параметры

Аргумент datatype должен соответствовать типу данных, указанному в операции обмена.

Пример

program main_mpi
include 'mpif.h'
integer rank, i, k, ierr, tag, dest, status(MPI_status_size)
real x
tag = 0
dest = 2
call MPI_Init(ierr)
call MPI_Comm_rank(MPI_COMM_WORLD, rank, ierr)
if (rank.eq.0) then
 i = 2002
 call MPI_Send(i, 1, MPI_INTEGER, dest, tag, MPI_COMM_WORLD, ierr)
else if(rank.eq.1) then
x = 3.14159
call MPI_Send(x, 1, MPI_REAL, dest, tag, MPI_COMM_WORLD, ierr)
…
do k = 1, 2
 call MPI_Probe(MPI_ANY_SOURCE, tag, MPI_COMM_WORLD, status, ierr)
 if (status(MPI_source).eq.0) then
 call MPI_Recv(i, 1, MPI_INTEGER, 0, tag, MPI_COMM_WORLD, status, ierr)
 print *, "received ", i, " from 0"
else
 call MPI_Recv(x, 1, MPI_REAL, 1, tag, MPI_COMM_WORLD, status, ierr)
 print *, "received ", x, " from 1"
 end if
end do
end if
call MPI_Finalize(ierr)
stop
end