如何修改MPI阻止发送和接收到非阻止

Question

我试图了解使用MPI进行并行处理时阻塞和非阻塞消息传递机制之间的区别。假设我们有以下阻止代码：

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

int main (int argc, char* argv[]) {
    const int maximum_message_length = 100;
    const int rank_0= 0;
    char message[maximum_message_length+1]; 
    MPI_Status status; /* Info about receive status */ 
    int my_rank; /* This process ID */
    int num_procs; /* Number of processes in run */ 
    int source; /* Process ID to receive from */
    int destination; /* Process ID to send to */
    int tag = 0; /* Message ID */

    MPI_Init(&argc, &argv);
    MPI_Comm_rank(MPI_COMM_WORLD, &my_rank); 
    MPI_Comm_size(MPI_COMM_WORLD, &num_procs);

    /* clients processes */
    if (my_rank != server_rank) {
        sprintf(message, "Hello world from process# %d", my_rank);
        MPI_Send(message, strlen(message) + 1, MPI_CHAR, rank_0, tag, MPI_COMM_WORLD);
    } else {    
    /* rank 0 process */ 
        for (source = 0; source < num_procs; source++) { 
            if (source != rank_0) {
                MPI_Recv(message, maximum_message_length + 1, MPI_CHAR, source, tag, 
                MPI_COMM_WORLD,&status);
                fprintf(stderr, "%s\n", message); 
            } 
        } 
    } 
         MPI_Finalize();
}

每个处理器执行其任务，然后将其发送回rank_0（接收方）。rank_0将运行一个从1到n-1个进程的循环并按顺序打印它们（如果当前客户端尚未发送任务，则循环中的i步骤可能不会继续进行）。如何使用MPI_Isend和修改此代码以实现非阻塞机制MPI_Irecv？我是否需要删除接收器部分（rank_0）中的循环，并为每个客户端明确声明MPI_Irecv（..），即

MPI_Irecv(message, maximum_message_length + 1, MPI_CHAR, source, tag, 
                    MPI_COMM_WORLD,&status);

谢谢。

Answer 1

对非阻塞通信的处理方式是发布通信，然后立即继续执行程序来做其他事情，这又可能是发布更多通信。特别是，您可以一次过帐所有收货，然后等待它们稍后再完成。这就是您通常在此处的方案中要执行的操作。

但是请注意，此特定设置是一个不好的示例，因为它基本上只是重新实现了MPI_Gather！

这是您通常在设置中进行无阻塞通信的方式。首先，您需要一些存储空间来存储所有消息，并且还需要一个请求句柄列表来跟踪非阻塞通信请求，因此需要相应地更改代码的第一部分：

#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include "mpi.h"

int main (int argc, char* argv[]) {
    const int maximum_message_length = 100;
    const int server_rank = 0;
    char message[maximum_message_length+1];
    char *allmessages;
    MPI_Status *status; /* Info about receive status */
    MPI_Request *req; /* Non-Blocking Requests */
    int my_rank; /* This process ID */
    int num_procs; /* Number of processes in run */
    int source; /* Process ID to receive from */
    int tag = 0; /* Message ID */

    MPI_Init(&argc, &argv);
    MPI_Comm_rank(MPI_COMM_WORLD, &my_rank);
    MPI_Comm_size(MPI_COMM_WORLD, &num_procs);

    /* clients processes */
    if (my_rank != server_rank) {
        sprintf(message, "Hello world from process# %d", my_rank);
        MPI_Send(message, maximum_message_length + 1, MPI_CHAR, server_rank,
                 tag, MPI_COMM_WORLD);
    } else {

此处无需阻塞发送。现在，我们继续在server_rank上接收所有这些消息。我们需要遍历所有它们，并为它们每个存储一个请求句柄：

    /* rank 0 process */
        allmessages = malloc((maximum_message_length+1)*num_procs);
        status = malloc(sizeof(MPI_Status)*num_procs);
        req = malloc(sizeof(MPI_Request)*num_procs);

        for (source = 0; source < num_procs; source++) {
            req[source] = MPI_REQUEST_NULL;
            if (source != server_rank) {
                /* Post non-blocking receive for source */
                MPI_Irecv(allmessages+(source*(maximum_message_length+1)),
                          maximum_message_length + 1, MPI_CHAR, source, tag,
                          MPI_COMM_WORLD, req+source);
                /* Proceed without waiting on the receive */
                /* (posting further receives */
            }
        }
        /* Wait on all communications to complete */
        MPI_Waitall(num_procs, req, status);
        /* Print the messages in order to the screen */
        for (source = 0; source < num_procs; source++) {
            if (source != server_rank) {
                fprintf(stderr, "%s\n",
                        allmessages+(source*(maximum_message_length+1)));
            }
        }
    }
    MPI_Finalize();
}

发布非阻塞接收后，我们需要等待所有接收完成，以正确的顺序打印消息。为此，使用了MPI_Waitall，它使我们可以阻塞直到满足所有请求句柄。请注意，为简单起见，我在此处包括server_rank，但最初将其请求设置为MPI_REQUEST_NULL，因此将被忽略。如果您不关心订单，则可以通过遍历请求并使用MPI_Waitany尽快处理可用的通信。一旦任何通信完成，该操作就会返回，您可以对相应的数据进行操作。

使用MPI_Gather，该代码将如下所示：

#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include "mpi.h"

int main (int argc, char* argv[]) {
    const int maximum_message_length = 100;
    const int server_rank = 0;
    char message[maximum_message_length+1];
    char *allmessages;
    int my_rank; /* This process ID */
    int num_procs; /* Number of processes in run */
    int source; /* Process ID to receive from */
    int tag = 0; /* Message ID */

    MPI_Init(&argc, &argv);
    MPI_Comm_rank(MPI_COMM_WORLD, &my_rank);
    MPI_Comm_size(MPI_COMM_WORLD, &num_procs);

    if (my_rank == server_rank) {
        allmessages = malloc((maximum_message_length+1)*num_procs);
    }
    sprintf(message, "Hello world from process# %d", my_rank);
    MPI_Gather(message, (maximum_message_length+1), MPI_CHAR,
               allmessages, (maximum_message_length+1), MPI_CHAR,
               server_rank, MPI_COMM_WORLD);

    if (my_rank == server_rank) {
        /* Print the messages in order to the screen */
        for (source = 0; source < num_procs; source++) {
            if (source != server_rank) {
                fprintf(stderr, "%s\n",
                        allmessages+(source*(maximum_message_length+1)));
            }
        }
    }
    MPI_Finalize();
}

借助MPI-3，您甚至可以使用非阻塞MPI_Igather。

如果您不关心顺序，那么最后一部分（以MPI_Waitall开头）可以通过MPI_Waitany完成，如下所示：

    for (i = 0; i < num_procs-1; i++) {
        /* Wait on any next communication to complete */
        MPI_Waitany(num_procs, req, &source, status);
        fprintf(stderr, "%s\n",
                allmessages+(source*(maximum_message_length+1)));
    }