初探并行编程技术之消息传递接口(Message Passing Interface, MPI)

初探消息传递接口

Message Passing Interface, MPI

之前的文章中介绍了天河二号的架构，我们大致了解到了天河二号是一个由很多计算节点组成的具有强大运算能力的超级计算机。

天河二号深度解密，你值得拥有

如果我们要使用天河二号这个强大的超级计算机，那么首先要了解到，天河二号的使用方式有两种。一种方式是云平台使用，即是将天河二号看成是一个虚拟机，这个虚拟机是从单个计算节点上虚拟出来的，使用起来就跟普通的云虚拟机没什么差别。另一种方式是高性能使用，这种使用方式是直接面向计算节点的，对用户来说，计算节点是可见的，用户通过ssh登录到计算节点（系统为Red Hat Linux），申请节点资源，运行作业。

如果我们只是需要一个普普通通的云虚拟机，在上面跑跑小作业，那么云平台方式就足够了（当然，如果只是这样可能上机申请不被批准）。如果我们要跑大型的作业，调度多个计算节点进行运算，充分利用天河二号的能力，那么就必须使用高性能方式。

高性能使用方式即是通过编程，使得作业具有多节点协作的能力，而多个节点之间的协作是通过MPI(Message Passing Interface)来实现的。MPI是一种通过提供并行库来实现并行化的方法，我们通过调用MPI并行库实现的方法来实现并行化，只需要安装MPI库，那么就可以用原来的编译器来进行编译运行，当前的MPI允许拥有MPI实现库的语言进行调用，支持C,C++,Fortran,C#,java,python等语言直接调用。

现如今MPI的实现主要有：

MPICH:http://www-unix.mcs.anl.gov/mpi/mpich

CHIMP:ftp://ftp.epcc.ed.ac.uk/pub/packages/chimp/release/

LAM:http://www.mpi.nd.edu/lam/download/

有关MPI环境的配置在这里就不赘述，有兴趣的读者可以在以下网址内学习：

windows配置方法：https://www.cnblogs.com/hantan2008/p/5390375.html

Linux配置方法：

https://blog.csdn.net/qq_30239975/article/details/77703321

MacOS配置方法：

https://blog.csdn.net/bemf168/article/details/52187140

而天河二号平台上有自实现的MPI库，对其为天河二号内部的高速互联网络进行了优化，速度较快。不同的MPI库虽然实现不一样，但提供的接口都是一样的，因此我们下面就以C++语言为例子介绍标准的MPI第一版。

MPI编程准备

使用MPI库方法之前我们需要先包含其头文件mpi.h，并且在调用任何MPI库方法之前我们首先通过MPI_Init需要初始化MPI运行环境，在结束完成所有MPI方法的调用之后需要用MPI_Finalize来释放资源。亦即我们的MPI程序格式为：

#include “mpi.h”
    …

Int main(int argc, char** argv)
{
   MPI_Init(&argc, &argv);

  …

   MPI_Finalize();
}

由于MPI编程是相同的代码运行在多个节点上，所以每个节点运行的进程都拥有自己的进程号，可以通过MPI_Comm_rank()获取当前进程的进程号，通过MPI_Comm_size()获取总的进程数。这两个函数的定义如下:

简单的点对点通信

MPI的点对点通信包括发送和接收两种操作，发送操作为一个进程发送一段数据到指定的一个进程，而接受操作则是一个进程接收指定进程(也可指定为任意进程)的数据。在这里我们介绍最简单的点对点通信：MPI_Send和MPI_Recv。

MPI_Send的函数：

datatype表示发送数据类型，MPI通信需要指定通信数据类型，自带的类型如下：

dest用来指定数据发往的进程号。

tag是一个标志，用来匹配发送与接收操作的，tag相同的操作才会进行通信。

comm为通信域，一般直接使用MPI_COMM_WORLD即可。

MPI_Recv函数：

Source用来指定接收数据的来源进程，可以用MPI_ANY_SOURCE 来表示可以接收任意进程的数据。

Tag为消息标志，只有与发送操作的tag匹配才会进行通信，可以用MPI_ANY_TAG来表示可以接受任意tag的数据。

Status为通信结果，是一个数据类型结构，在C语言中，status结构包括MPI_SOURCE 、MPI_TAG和MPI_ERROR这三个域，分别表示接收操作的来源进程、来源标识以及可能的错误代码。可以用MPI_STATUS_IGNORE表示丢弃通信结果。

接下来我们看一个简单的通信例子：

…
Int main(int argc, char** argv)
{
    // MPI初始化
    MPI_Init(&argc, &argv);

    Int rank;
    Char msg[40] = {0}; 

    // 获取当前进程的进程号
    MPI_Comm_rank( MPI_COMM_WORLD, & rank);

    // 进程号为0的执行代码
    If (rank == 0)
    {
        // 拷贝发送的信息到缓冲区
       Strcpy(msg,”Hello, greeting from process 0!”);
    // 指定要发送的大小为缓冲区内容大小，发往进程号为1的进程，标识为0
       MPI_Send(msg, strlen(msg), MPI_CHAR, 1, 0, MPI_COMM_WORLD);
    }
// 进程号为1的执行代码
    Else if (rank == 1)
   {
    // 指定最多可以接收20个char大小，接受内容放在msg中，来源进程号为0，来源标识为0，忽略接收的结果status。
       MPI_Recv(msg, 40, MPI_CHAR, 0, 0, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
       Printf(“Process 1 receive: %s\n”, msg);
   }
    // 释放MPI资源
    MPI_Finalize();
}

将上述代码进行编译，生成可执行文件，然后我们可以用 yhru –n N 指令运行，其中参数 –n 表示指定生成N个进程执行代码，亦即N为多少就有多少个进程共同执行上述代码。

在本例中，我们指定两个进程执行以上代码，进程号为0的进程向进程号为1的进程发送一条消息，进程1接收到该消息之后则打印该消息。

运行结果如下

上述程序中的Send和Recv方法都是标准通信，Send方法在Recv方法成功接收全部数据之前不会返回，Recv方法在成功接收全部数据之前不会返回。

相对应的，还有另外三种通信方式缓存通信、同步通信、就绪通信。

（1）缓存通信：用户提供通信缓冲区，避免了系统内存拷贝，提高了通信效率，但是缓冲区需用户自己管理。

（2）同步通信：发送进程只有当接受进程开始接收（不需要全部接收）的时候才返回。

（3）就绪通信：发送进程的发送操作只有当接受进程已经开启了接收操作的时候才能够成功调用，否则发送操作将会出错。

以上的三种通信方式在本篇中不予介绍，感兴趣的小伙伴可以参考 都志辉《高性能计算并行编程技术-MPI并行程序设计》 第九章

简单的组通信

MPI中的组通信表示涉及多个进程之间的通信，可以分为三种：一对多通信、多对一通信、多对多通信。

一般来说，交换相同大小的数据，组通信是比点对点通信要快的。

例如我们进程0需要发送数据到其他进程，如果单纯使用点对点通信，那么就需要n-1次串行MPI发送操作，亦即是n-1倍的通信时间。

而在组通信中，有专门操作MPI_Bcast方法来完成这一操作。简单来说就是进程0需要发送数据到进程1~n，那么进程0就会先发送数据到另外一个进程，现在我们就有两个进程拥有数据的副本了，接下来两个进程分别发送数据到另外一个进程，我们就拥有四个进程拥有数据的副本，接下来就是以类似的方法进行指数级数据的扩展，因此仅需耗费log2n次通信时间就能够完成该操作。

由于MPI中的组通信方法很多，所以这里在一对多、多对一、多对多通信方法中各挑选一个例子进行阐述。

1）上述提到的MPI_Bcast是MPI通信中典型的一对多通信，其功能为将一个进程的数据发往通信域里其他的进程。函数原型为：

其中root为需要发送数据的进程，在comm通信域中除root以外的其他进程都会接收到该数据。在root进程中，缓冲区buffer中存放有count个数据类型为datatype的数据，而在comm通信域的其他进程中，缓冲区buffer则用来接收count个数据类型为datatype的数据。

需要注意的是，MPI中的组通信会自动进行同步，也就是所有进程中的组通信操作只有当组通信操作完成之后才能继续往下执行（拥有类似同步功能的有MPI_Barrier函数），下面所描述的组通信函数也一样。

一个简单的MPI_Bcast代码：

…
…
int main(int argc, char** argv)
{
    // MPI初始化
    MPI_Init(&argc, &argv);

    Int rank, len;
    Char msg[50] {0};   

    // 获取当前进程的进程号
    MPI_Comm_rank( MPI_COMM_WORLD, & rank);

    // 进程号为0的执行代码
    If (rank == 0)
    {
        // 根进程初始化缓冲区
       Strcpy(msg, “Hello, this is greeting from prcess 0!”);
    // 获取广播信息长度
       Len = strlen(msg);
    }
// 在MPI_COMM_WORLD的通信域中，进程号为0的进程向其他进程发送msg信息的长度
    MPI_Bcast(&len, 1, MPI_INT, 0, MPI_COMM_WORLD);
// 在MPI_COMM_WORLD的通信域中，进程号为0的进程向其他进程发送msg的内容
    MPI_Bcast(msg, len, MPI_CHAR, 0, MPI_COMM_WORLD);
    Printf(“I am process %d, receiving message: %s!”, rank, msg);    

    // 释放MPI资源
    MPI_Finalize();
}

上述例子中进程0广播了一条消息到comm域里面的所有进程。Comm域中的进程在接收到该广播信息后将其打印出来。运行结果如下：

（2）如果我们需要将各个进程的数据收集到一个进程，那么就需要用到MPI中多对一的组通信方法MPI_Gather。其函数原型为：

其中带有send前缀的参数只有在非root进程有意义，带有recv前缀的参数只有在root进程有意义。进程root会将comm通信域中所有进程的sendbuf中的数据按顺序收集到recvbuf中，如下图所示：

图中每一个圆角矩形以及里面的编号i代表的是进程i的发送缓冲区sendbuf，下方的矩形表示进程root的接收缓冲区recvbuf，MPI_Gather操作会将每一个进程（包括root进程）的发送缓冲区的内容以进程号为偏移按顺序放置在接收缓冲区上，当MPI_Gather操作完成后进程root的recvbuf里面就放置有comm通信域里所有进程的数据了。

需要注意的是参数中的sendcount表示发送缓冲区的数据个数，而recvcount表示进程root从每一个进程接收数据的个数，因此sendcount和recvcount应该是一致的。

…
…
Int main(int argc, char** argv)
{
    // 初始化
    MPI_Init(&argc, &argv);

    Int rank, size;
    // 获取进程号
    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
    // 获取进程数量
    MPI_Comm_size(MPI_COMM_WORLD, &size);

    // 每个进程声明一个发送缓冲区，缓冲区里面只有一个进程号整数
    Int sendbuf[1];
    Sendbuf[0] = rank;

    Int recvbuf[20];
    // 进程0从各进程收集数据
    MPI_Gather(sendbuf, 1, MPI_INT, recvbuf, 1, MPI_INT, 0, MPI_COMM_WORLD);

    If (rank == 0)
    {
        Printf(“Process 0: ”);
For (int i=0;i<size;++i) 
            Printf(“%d “, recvbuf[i]);
        Printf(“\n”);
    }

    // 释放资源
    MPI_Finalize();
}

上述示例中，每一个进程的缓冲区中存放有一个独特的整数（在这里为进程号），MPI_Gather操作将这些整数收集到进程0，进程0接下来将这些整数打印出来。运行结果如下：

（3）假设我们需要交换每一个进程的数据，亦即每个进程都需要获取其他进程的数据，那么就可以使用MPI中多对多的通信操作MPI_Allgather，其函数原型为：

Int main(int argc, char** argv)
{
    // 初始化
    MPI_Init(&argc, &argv);

    Int rank, size;
    // 获取进程号
    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
    // 获取进程数量
    MPI_Comm_size(MPI_COMM_WORLD, &size);

    // 每个进程声明一个发送缓冲区，缓冲区里面只有一个进程号整数
    Int sendbuf[1];
    Sendbuf[0] = rank;

    Int recvbuf[20];
    // 进程0从各进程收集数据
    MPI_Allgather(sendbuf, 1, MPI_INT, recvbuf, 1, MPI_INT, MPI_COMM_WORLD);

    Printf(“Process %d:”, rank);
For (int i=0;i<size;++i) 
        Printf(“%d “, recvbuf[i]);
    Printf(“\n”);

    // 释放资源
    MPI_Finalize();
}

上述示例中，每一个进程的缓冲区中存放有一个独特的整数（在这里为进程号），MPI_Allgather操作将这些整数收集到comm域中每一个进程， 接下来每一个进程都能够把这些整数打印出来。运行结果如下：

以上介绍了MPI中部分组通信，实际上MPI支持的组通信操作非常多样，不仅支持灵活的数据交换操作，而且支持用户自定义的组通信操作，这些内容在本篇中不予介绍，感兴趣的小伙伴可以参考 都志辉《高性能计算并行编程技术-MPI并行程序设计》 第十三章。

---The End---

文案 && 代码：邹海枫

审稿 && 排版：贺兴

指导老师： 张子臻（中山大学数据科学与计算机学院副教授，中山大学ACM集训队教练）