MPI SEND和RECV

send和recv是MPI里面两个基础的概念，MPI里面几乎所有单个的方法都可以使用基础的发送和接受API实现，本节会介绍如何使用MPI的同步或异步send和recv，以及另外的使用MPI进行数据传输的基础概念。

本节代码在tutorials/mpi-send-and-receive/code

MPI的发送和接收简介

MPI的发送和接收按照如下方式进行：开始时，A进程决定发送一些消息给B进程，A进程会把需要发送的数据全部分配好，放到一个服务器里面，因为所有的数据会被备份到一个大的信息里面，所以共享存储会被比作信封，数据备份进存储之后，通信设备就需要负责将信息传递到正确的位置。这个正确的位置按照顺序确定的那个进程。

尽管数据已经发送到B，但是进程B仍然需要确认它将要接收A的数据，一旦确定了这一点，数据就被传输成功了，进程A会接收到数据传递成功的信息，然后去干其他事情。

有时A需要传递很多不同的消息给B，让B能够比较方便的区分不同的消息，MPI运行发送者和接收者另外指定一些信息ID（正式名称是标签、标签）。当B只要求接收某种特定标签的信息时，其他不是这个标签的信息会被先缓存起来，等到B需要时才给B。

如下是MPI send和recv的定义

MPI_Send(
    void* data,
    int count,
    MPI_Datatype datatype,
    int destination,
    int tag,
    MPI_Comm communicator)

MPI_Recv(
    void* data,
    int count,
    MPI_Datatype datatype,
    int source,
    int tag,
    MPI_Comm communicator,
    MPI_Status* status)

尽管一开始看起来参数有点多，慢慢地你会发现其实这些参数还是很好记忆的，因为大多数 MPI 方法定义都是类似的。第一个参数是数据存储。第二个和第三个参数分别描述了数据的数量和类型。MPI_send会准确地发送计数指定的数量个元素，MPI_Recv会最多接受计数个元素（之后会详细讲）。第四个和第五个参数指定了发送方/接受个方进程的排名以及信息的标签。第六个参数指定了使用的通信器。MPI_Recv方法的最后一个参数提供了接受到的信息的状态。

基础MPI数据结构

MPI_send和MPI_Recv方法使用了 MPI 的数据结构作为一种在更高层次指定消息结构的方法。举个例子，如果一个进程想要发送一个整数给另一个进程，它会指定 count 为 1，数据结构为MPI_INT。其他的MPI数据结构以及它们在C语言里的对应结构如下：

MPI	C
MPI_SHORT	short int
MPI_INT	int
MPI_LONG	long int
MPI_LONG_LONG	long long int
MPI_UNSIGNED_CHAR	unsigned char
MPI_UNSIGNED_SHORT	unsigned short int
MPI_UNSIGNED	unsigned int
MPI_UNSIGNED_LONG	unsigned long int
MPI_UNSIGNED_LONG_LONG	unsigned long long int
MPI_FLOAT	float
MPI_DOUBLE	double
MPI_LONG_DOUBLE	long double
MPI_BYTE	char

此处只展示基础类型，后续会学习如何创建自己的MPI数据类型来构建更复杂的消息类型

MPI SEND/RECV example

第一个example在send_recv.c

// 得到当前进程的 rank 以及整个 communicator 的大小
int world_rank;
MPI_Comm_rank(MPI_COMM_WORLD, &world_rank);
int world_size;
MPI_Comm_size(MPI_COMM_WORLD, &world_size);

int number;
if (world_rank == 0) {
    number = -1;
    MPI_Send(&number, 1, MPI_INT, 1, 0, MPI_COMM_WORLD);
} else if (world_rank == 1) {
    MPI_Recv(&number, 1, MPI_INT, 0, 0, MPI_COMM_WORLD,
             MPI_STATUS_IGNORE);
    printf("Process 1 received number %d from process 0\\n",
           number);
}

MPI_Comm_rank 和 MPI_Comm_size 一开始是用来得到整个 communicator 空间的大小（也就是所有进程的数量）以及当前进程的秩。然后如果当前进程是 0 进程，那么我们就初始化一个数字 -1 然后把它发送给 1 进程。然后你可以看到 else if 条件语句里的话题，进程 1 会调用 MPI_Recv 去接受这个数字。然后会将接收到的数字打印出来。由于我们明确地发送接收了一个整数，因此 MPI_INT 数据类型被使用了。每个进程还使用了 0 作为消息标签来指定消息。由于我们这里只有一种类型的信息被传递了，因此进程也可以使用预先定义好的常量 MPI_ANY_TAG 来作为标签数字。

你可以把代码从GitHub下载下来并运行 run.py 脚本.

>>> git clonehttps://github.com/mpitutorial/mpitutorial
>>> cd mpitutorial/tutorials
>>> ./run.py send_recv
mpirun -n 2 ./send_recv
Process 1 received number -1 from process 0

MPI PINGPONG example

在pingpong example中，两个进程会一直使用MPI_SEND和MPI_RECV方法阻挡消息，直到两者决定不玩了，主要代码逻辑如下：

int ping_pong_count = 0;
int partner_rank = (world_rank + 1) % 2;
while (ping_pong_count < PING_PONG_LIMIT) {
    if (world_rank == ping_pong_count % 2) {
        // Increment the ping pong count before you send it
        ping_pong_count++;
        MPI_Send(&ping_pong_count, 1, MPI_INT, partner_rank, 0, MPI_COMM_WORLD);
        printf("%d sent and incremented ping_pong_count %d to %d\\n",
               world_rank, ping_pong_count,
               partner_rank);
    } else {
        MPI_Recv(&ping_pong_count, 1, MPI_INT, partner_rank, 0,
                 MPI_COMM_WORLD, MPI_STATUS_IGNORE);
        printf("%d received ping_pong_count %d from %d\\n",
               world_rank, ping_pong_count, partner_rank);
    }
}

这个程序是为两个进程执行而设计的，进程会根据求与算法来获取各自对手。ping_pong_count 一开始被初始化为0，然后每次发送消息之后会递增1。随着 ping_pong_count 的递增，两个进程会轮流成为发送者和接受者。最后，当我们设定的 limit 被触发的时候（我的代码里设定为10），进程就停止了发送和接收。程序的输出如下。

>>> ./run.py ping_pong
0 sent and incremented ping_pong_count 1 to 1
0 received ping_pong_count 2 from 1
0 sent and incremented ping_pong_count 3 to 1
0 received ping_pong_count 4 from 1
0 sent and incremented ping_pong_count 5 to 1
0 received ping_pong_count 6 from 1
0 sent and incremented ping_pong_count 7 to 1
0 received ping_pong_count 8 from 1
0 sent and incremented ping_pong_count 9 to 1
0 received ping_pong_count 10 from 1
1 received ping_pong_count 1 from 0
1 sent and incremented ping_pong_count 2 to 0
1 received ping_pong_count 3 from 0
1 sent and incremented ping_pong_count 4 to 0
1 received ping_pong_count 5 from 0
1 sent and incremented ping_pong_count 6 to 0
1 received ping_pong_count 7 from 0
1 sent and incremented ping_pong_count 8 to 0
1 received ping_pong_count 9 from 0
1 sent and incremented ping_pong_count 10 to 0

Cycle example

另一个example中，一个值会在各个进程间以一个环的形式传递

int token;
if (world_rank != 0) {
    MPI_Recv(&token, 1, MPI_INT, world_rank - 1, 0,
             MPI_COMM_WORLD, MPI_STATUS_IGNORE);
    printf("Process %d received token %d from process %d\\n",
           world_rank, token, world_rank - 1);
} else {
    // Set the token's value if you are process 0
    token = -1;
}
MPI_Send(&token, 1, MPI_INT, (world_rank + 1) % world_size,
         0, MPI_COMM_WORLD);

// Now process 0 can receive from the last process.
if (world_rank == 0) {
    MPI_Recv(&token, 1, MPI_INT, world_size - 1, 0,
             MPI_COMM_WORLD, MPI_STATUS_IGNORE);
    printf("Process %d received token %d from process %d\\n",
           world_rank, token, world_size - 1);
}

这种写法避免了死锁的发生

利用MPI进行动态接收

本节讨论如何发送事先知道消息长度的消息。

MPI_Status结构体

MPI_Recv 将 MPI_Status 结构体的地址作为参数（可以使用 MPI_STATUS_IGNORE 忽略）。如果我们将 MPI_Status 结构体传递给 MPI_Recv 函数，则操作完成后将在该结构体中填充有关接收操作的其他信息。三个主要的信息包括：

发送端秩. 发送端的秩存储在结构体的 MPI_SOURCE 元素中。也就是说，如果我们声明一个 MPI_Status stat 变量，则可以通过 stat.MPI_SOURCE 访问秩。
消息的标签. 消息的标签可以通过结构体的 MPI_TAG 元素访问（类似于 MPI_SOURCE）。
消息的长度. 消息的长度在结构体中没有预定义的元素。相反，我们必须使用 MPI_Get_count 找出消息的长度。

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

在 MPI_Get_count 函数中，使用者需要传递 MPI_Status 结构体，消息的 datatype（数据类型），并返回 count。变量 count 是已接收的 datatype 元素的数目。

为什么需要这些信息？事实证明，MPI_Recv 可以将 MPI_ANY_SOURCE 用作发送端的秩，将 MPI_ANY_TAG 用作消息的标签。在这种情况下，MPI_Status 结构体是找出消息的实际发送端和标签的唯一方法。 此外，并不能保证 MPI_Recv 能够接收函数调用参数的全部元素。相反，它只接收已发送给它的元素数量（如果发送的元素多于所需的接收数量，则返回错误。） MPI_Get_count 函数用于确定实际的接收量。

`MPI_Status` 结构体查询的示例

查询 MPI_Status 结构体的程序在 check_status.c 中。程序将随机数量的数字发送给接收端，然后接收端找出发送了多少个数字。代码的主要部分如下所示。

const int MAX_NUMBERS = 100;
int numbers[MAX_NUMBERS];
int number_amount;
if (world_rank == 0) {
    // Pick a random amount of integers to send to process one
    srand(time(NULL));
    number_amount = (rand() / (float)RAND_MAX) * MAX_NUMBERS;

    // Send the amount of integers to process one
    MPI_Send(numbers, number_amount, MPI_INT, 1, 0, MPI_COMM_WORLD);
    printf("0 sent %d numbers to 1\\n", number_amount);
} else if (world_rank == 1) {
    MPI_Status status;
    // Receive at most MAX_NUMBERS from process zero
    MPI_Recv(numbers, MAX_NUMBERS, MPI_INT, 0, 0, MPI_COMM_WORLD,
             &status);

    // After receiving the message, check the status to determine
    // how many numbers were actually received
    MPI_Get_count(&status, MPI_INT, &number_amount);

    // Print off the amount of numbers, and also print additional
    // information in the status object
    printf("1 received %d numbers from 0. Message source = %d, "
           "tag = %d\\n",
           number_amount, status.MPI_SOURCE, status.MPI_TAG);
}

使用MPI_Probe找出消息大小

现在您了解了 MPI_Status 的工作原理，现在我们可以使用它来发挥更高级的优势。除了传递接收消息并简易地配备一个很大的缓冲区来为所有可能的大小的消息提供处理（就像我们在上一个示例中所做的那样），您可以使用 MPI_Probe 在实际接收消息之前查询消息大小。函数原型看起来像这样：

MPI_Probe(
    int source,
    int tag,
    MPI_Comm comm,
    MPI_Status* status)

MPI_Probe 看起来与 MPI_Recv 非常相似。实际上，您可以将 MPI_Probe 视为 MPI_Recv，除了不接收消息外，它们执行相同的功能。与 MPI_Recv 类似，MPI_Probe 将阻塞具有匹配标签和发送端的消息。当消息可用时，它将填充 status 结构体。然后，用户可以使用 MPI_Recv 接收实际的消息。

int number_amount;
if (world_rank == 0) {
    const int MAX_NUMBERS = 100;
    int numbers[MAX_NUMBERS];
    // Pick a random amount of integers to send to process one
    srand(time(NULL));
    number_amount = (rand() / (float)RAND_MAX) * MAX_NUMBERS;

    // Send the random amount of integers to process one
    MPI_Send(numbers, number_amount, MPI_INT, 1, 0, MPI_COMM_WORLD);
    printf("0 sent %d numbers to 1\\n", number_amount);
} else if (world_rank == 1) {
    MPI_Status status;
    // Probe for an incoming message from process zero
    MPI_Probe(0, 0, MPI_COMM_WORLD, &status);

    // When probe returns, the status object has the size and other
    // attributes of the incoming message. Get the message size
    MPI_Get_count(&status, MPI_INT, &number_amount);

    // Allocate a buffer to hold the incoming numbers
    int* number_buf = (int*)malloc(sizeof(int) * number_amount);

    // Now receive the message with the allocated buffer
    MPI_Recv(number_buf, number_amount, MPI_INT, 0, 0,
             MPI_COMM_WORLD, MPI_STATUS_IGNORE);
    printf("1 dynamically received %d numbers from 0.\\n",
           number_amount);
    free(number_buf);
}

MPI_Probe 构成了许多动态 MPI 应用程序的基础。例如，控制端/执行子程序在交换变量大小的消息时通常会大量使用 MPI_Probe。作为练习，对 MPI_Recv 进行包装，将 MPI_Probe 用于您可能编写的任何动态应用程序。

点对点通信应用-随机游走

本节利用之前所介绍的MPI相关教程来研究具体的应用程序实例。本文应用程序模拟了一个被称为“随机游走”的过程。

本文的代码位于此 tutorials/point-to-point-communication-application-random-walk/code 目录下。

随机游走的基本问题定义如下：给定 Min，Max 和随机游走器 W，让游走器 W 向右以任意长度的 S 随机移动。如果该过程越过边界，它就会绕回。 W 一次只能左右移动一个单位。

并行化的随机游走可以模拟各种并行程序的行为。

随机游走问题的并行化

在许多并行程序的应用中，首要任务是在各个进程之间划分域。随机行走问题的一维域大小为 Max - Min + 1（因为游走器包含 Max 和 Min）。假设游走器只能采取整数大小的步长，我们可以轻松地将域在每个进程中划分为大小近乎相等的块。例如，如果 Min 为 0，Max 为 20，并且我们有四个进程，则将像这样拆分域。

前三个进程拥有域的五个单元，而最后一个进程则拥有最后五个单元并且再加上一个剩余的单元。一旦对域进行了分区，应用程序将初始化游走器。如前所述，游走器将以步长 S 进行总步数随机的游走。例如，如果游走器在进程 0（使用先前的分解域）上进行了移动总数为 6 的游走，则游走器的执行将如下所示：

游走器的步行长度开始增加。但是，当它的值达到 4 时，它已到达进程 0 的边界。因此，进程 0 必须与进程 1 交流游走器的信息。
进程 1 接收游走器，并继续移动，直到达到移动总数 6。然后，游走器可以继续进行新的随机移动。

在此示例中，W 仅需从进程 0 到进程 1 进行一次通信。但是，如果 W 必须移动更长的距离，则可能需要沿其通过域的路径将其传递给更多的进程。

使用MPI_SEND和MPI_RECV组织代码

分割进程域

void decompose_domain(int domain_size, int world_rank,
                      int world_size, int* subdomain_start,
                      int* subdomain_size) {
    if (world_size > domain_size) {
        // Don't worry about this special case. Assume the domain
        // size is greater than the world size.
        MPI_Abort(MPI_COMM_WORLD, 1);
    }
    *subdomain_start = domain_size / world_size * world_rank;
    *subdomain_size = domain_size / world_size;
    if (world_rank == world_size - 1) {
        // Give remainder to last process
        *subdomain_size += domain_size % world_size;
    }
  }

定义walker结构体

typedef struct {
    int location;
    int num_steps_left_in_walk;
} Walker;

初始化walker

void initialize_walkers(int num_walkers_per_proc, int max_walk_size,
                        int subdomain_start, int subdomain_size,
                        vector<Walker>* incoming_walkers) {
    Walker walker;
    for (int i = 0; i < num_walkers_per_proc; i++) {
        // Initialize walkers in the middle of the subdomain
        walker.location = subdomain_start;
        walker.num_steps_left_in_walk =
            (rand() / (float)RAND_MAX) * max_walk_size;
        incoming_walkers->push_back(walker);
    }
}

使walkers前进

void walk(Walker* walker, int subdomain_start, int subdomain_size,
          int domain_size, vector<Walker>* outgoing_walkers) {
    while (walker->num_steps_left_in_walk > 0) {
        if (walker->location == subdomain_start + subdomain_size) {
            // Take care of the case when the walker is at the end
            // of the domain by wrapping it around to the beginning
            if (walker->location == domain_size) {
                walker->location = 0;
            }
            outgoing_walkers->push_back(*walker);
            break;
        } else {
            walker->num_steps_left_in_walk--;
            walker->location++;
        }
    }
}

仅在需要两个函数：发送待传出的walker的函数和接收待传入的walker的函数。发送功能如下所示：

void send_outgoing_walkers(vector<Walker>* outgoing_walkers,
                           int world_rank, int world_size) {
    // Send the data as an array of MPI_BYTEs to the next process.
    // The last process sends to process zero.
    MPI_Send((void*)outgoing_walkers->data(),
             outgoing_walkers->size() * sizeof(Walker), MPI_BYTE,
             (world_rank + 1) % world_size, 0, MPI_COMM_WORLD);

    // Clear the outgoing walkers
    outgoing_walkers->clear();
}

接收传入的walkers的函数应该使用MPI_PROBE，大致如下所示：

void receive_incoming_walkers(vector<Walker>* incoming_walkers,
                              int world_rank, int world_size) {
    MPI_Status status;

    // Receive from the process before you. If you are process zero,
    // receive from the last process
    int incoming_rank =
        (world_rank == 0) ? world_size - 1 : world_rank - 1;
    MPI_Probe(incoming_rank, 0, MPI_COMM_WORLD, &status);

    // Resize your incoming walker buffer based on how much data is
    // being received
    int incoming_walkers_size;
    MPI_Get_count(&status, MPI_BYTE, &incoming_walkers_size);
    incoming_walkers->resize(
        incoming_walkers_size / sizeof(Walker));
    MPI_Recv((void*)incoming_walkers->data(), incoming_walkers_size,
             MPI_BYTE, incoming_rank, 0, MPI_COMM_WORLD,
             MPI_STATUS_IGNORE);
}

现在我们已经建立了程序的主要功能。我们必须将所有这些功能集成在一起，如下所示：

初始化 walkers.
使用 walk 函数使 walkers 前进。
发出 outgoing_walkers 向量中的所有的 walkers。
将新接收的 walkers 放入 incoming_walkers 向量中。
重复步骤 2 到 4，直到所有 walkers 完成。

下面是完成此程序的第一次尝试。此刻，我们不必担心如何确定所有 walkers 完成的时间。但在查看代码之前，我必须警告您-该代码不正确！知晓这个问题以后，让我们看一下代码，希望您能发现它可能有什么问题。

// Find your part of the domain
decompose_domain(domain_size, world_rank, world_size,
                 &subdomain_start, &subdomain_size);

// Initialize walkers in your subdomain
initialize_walkers(num_walkers_per_proc, max_walk_size,
                   subdomain_start, subdomain_size,
                   &incoming_walkers);

while (!all_walkers_finished) { // Determine walker completion later
    // Process all incoming walkers
    for (int i = 0; i < incoming_walkers.size(); i++) {
        walk(&incoming_walkers[i], subdomain_start, subdomain_size,
             domain_size, &outgoing_walkers);
    }

    // Send all outgoing walkers to the next process.
    send_outgoing_walkers(&outgoing_walkers, world_rank,
                          world_size);

    // Receive all the new incoming walkers
    receive_incoming_walkers(&incoming_walkers, world_rank,
                             world_size);
}

但是该种写法会导致死锁诞生，

死锁及预防

// Find your part of the domain
decompose_domain(domain_size, world_rank, world_size,
                 &subdomain_start, &subdomain_size);

// Initialize walkers in your subdomain
initialize_walkers(num_walkers_per_proc, max_walk_size,
                  subdomain_start, subdomain_size,
                  &incoming_walkers);

// Determine the maximum amount of sends and receives needed to
// complete all walkers
int maximum_sends_recvs =
    max_walk_size / (domain_size / world_size) + 1;
for (int m = 0; m < maximum_sends_recvs; m++) {
    // Process all incoming walkers
    for (int i = 0; i < incoming_walkers.size(); i++) {
        walk(&incoming_walkers[i], subdomain_start, subdomain_size,
             domain_size, &outgoing_walkers);
    }

    // Send and receive if you are even and vice versa for odd
    if (world_rank % 2 == 0) {
        send_outgoing_walkers(&outgoing_walkers, world_rank,
                              world_size);
        receive_incoming_walkers(&incoming_walkers, world_rank,
                                 world_size);
    } else {
        receive_incoming_walkers(&incoming_walkers, world_rank,
                                 world_size);
        send_outgoing_walkers(&outgoing_walkers, world_rank,
                              world_size);
    }
}

Quartz 5

Explorer

阻塞式点对点通信

MPI SEND和RECV

MPI的发送和接收简介

基础MPI数据结构

MPI SEND/RECV example

MPI PINGPONG example

Cycle example

利用MPI进行动态接收

MPI_Status结构体

`MPI_Status` 结构体查询的示例

使用MPI_Probe找出消息大小

点对点通信应用-随机游走

随机游走问题的并行化

使用MPI_SEND和MPI_RECV组织代码

死锁及预防

Graph View

Table of Contents

Quartz 5

Explorer

阻塞式点对点通信

MPI SEND和RECV

MPI的发送和接收简介

基础MPI数据结构

MPI SEND/RECV example

MPI PINGPONG example

Cycle example

利用MPI进行动态接收

MPI_Status结构体

MPI_Status 结构体查询的示例

使用MPI_Probe找出消息大小

点对点通信应用-随机游走

随机游走问题的并行化

使用MPI_SEND和MPI_RECV组织代码

死锁及预防

Graph View

Table of Contents

`MPI_Status` 结构体查询的示例