如何将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);

谢谢你。

- Mike H.

---

当您打印消息时，是否仍希望接收过程有一定顺序？ - haraldkl

@haraldkl 直接使用OP的代码是不能保证消息按任何顺序接收，因此我认为不行。 - NoseKnowsAll

不，顺序无关紧要。 - Mike H.

@NoseKnowsAll 是的，在OP的代码中有一个强制顺序，因为阻塞接收必须按照循环的顺序执行。 - haraldkl

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)));
    }

- haraldkl

---

非常感谢。这解释得很清楚。 - Mike H.

@MikeH，很高兴听到这个消息。 - haraldkl