Асинхронная трансляция MPI из неизвестного источника

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

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

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

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

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

• ранг 0 имеет два открытых MPI_Ibcasts с корнем 1 и 2
• ранг 1 имеет два открытых MPI_Ibcasts с корнем 0 и 2
• ранг 2 имеет два открытых MPI_Ibcasts с корнем 0 и 1

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

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

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

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

int main()
{
    int size, rank;
    MPI_Init(NULL, NULL);
    MPI_Comm_size(MPI_COMM_WORLD, &size);
    MPI_Comm_rank(MPI_COMM_WORLD, &rank);

    const int root = 0;
    const int found_tag = 42;
    const long long chunk = 100000ll;
    const long long target = 134861523ll;

    MPI_Request found_request;
    long long found_item;
    if (rank == root) {
        MPI_Irecv(&found_item, 1, MPI_LONG_LONG, MPI_ANY_SOURCE, found_tag, MPI_COMM_WORLD, &found_request);
    } else {
        MPI_Ibcast(&found_item, 1, MPI_LONG_LONG, root, MPI_COMM_WORLD, &found_request);
    }

    for (long long my_number = rank;; my_number += size) {
        if (my_number == target) {
            found_item = my_number;
            printf("I found it: %d, %lld\n", rank, found_item);
            // We don't stop here, we will continue until the root tells us to stop
            // This also works if we are the root
            MPI_Send(&my_number, 1, MPI_LONG_LONG, root, found_tag, MPI_COMM_WORLD);
        }
        // Avoid checking MPI_Test too often.
        if ((1 + (my_number / size)) % chunk == 0) {
            if (rank == root) {
                int found = 0;
                MPI_Test(&found_request, &found, MPI_STATUS_IGNORE);
                if (found) {
                    printf("Someone told me that he found it: %d, %lld\n", rank, found_item);
                    MPI_Request bcast_request;
                    MPI_Ibcast(&found_item, 1, MPI_LONG_LONG, root, MPI_COMM_WORLD, &bcast_request);
                    MPI_Wait(&bcast_request, MPI_STATUS_IGNORE);
                    break;
                }
            } else {
                int found = 0;
                MPI_Test(&found_request, &found, MPI_STATUS_IGNORE);
                if (found) {
                    break;
                }
            }
        }
    }
    printf("%d, %lld\n", rank, found_item);
    MPI_Finalize();
}

Этот код предполагает, что вы найдете ровно одно число — один раз. Для поиска нескольких номеров потребуется дополнительный код.

Как вы заметили, вы не можете размещать трансляции с неизвестным отправителем. Теперь можно было опубликовать MPI_Allreduce (или даже MPI_Allgather) - после чего все ранги узнают найденное значение. Однако вы не можете просто опубликовать асинхронный один раз, потому что вы не можете изменить значение после его публикации.

Прежде всего, как и другие коллективные примитивы, широковещательная операция MPI требует участия всех процессов, а это означает, что если вы хотите использовать коллективную операцию, вам нужно, чтобы все процессы вводили оператор. Это связано с тем, что широковещательный примитив MPI выполняет как отправку, так и получение в одном и том же коллективном операторе: >Соответствующая процедура приема MPI_Bcast

Эта абстракция обычно допускает ненаивные реализации коллективов, все процессы могут реально участвовать в трансляции. Это хорошо объяснено здесь: http://mpitutorial.com/tutorials/mpi-broadcast-and-collective-communication/

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

В вашем вопросе вы, кажется, обеспокоены только использованием цикла для прослушивания сообщений. Обратите внимание, что вы можете избежать цикла получения, используя MPI_ANY_SOURCE в качестве исходного параметра для проверки и получения сообщений. Это позволяет избежать цикла, но проверяет/получает только одно сообщение, поэтому в зависимости от того, что вы хотите сделать, вы можете зацикливаться до тех пор, пока в очереди больше не будет сообщений, или иметь что-то более сложное для защиты от большого количества сообщений, препятствующих продвижению обработать.

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

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

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

Другим вариантом может быть изучение RMA (операций с удаленной памятью) MPI-3. Их также называют «односторонними», потому что получателю не нужно вызывать соответствующий прием. Таким образом, вам не нужно знать корень, чтобы получать широковещательные данные. Возможно, вы сможете построить интеллектуальную схему пут/получить, используя эти операции.

Я не нашел хорошего решения для себя, все еще имею аналогичную проблему. Моя лучшая идея на данный момент: запустить прослушиватель широковещательной рассылки (=поток) для каждого возможного корня широковещательной рассылки в каждом процессе. Конечно, это создает большие накладные расходы: вам нужен поток, и они должны использовать собственный MPI_Communicator для каждого источника bcast. Слушатели бросают полученные данные в общую очередь сообщений, и все готово.