第四讲 MPI并行程序设计 课程网站：CourseGrading http://judge. buaa.edu.cn 主讲教师： 赵长海 主讲教师： 赵长海 办公室： 新主楼G910 Email: zch@buaa.edu.cn Spring 2013

4.1 认识MPI 本章内容 4.2 MPI编程基础 4.3 MPI的集合通信 4.4 MPI与pthread混合编程

用于分布式共享内存编程 4.1 认识MPI MPI 一.什么是MPI Message Passing Interface，消息传递接口。是很多公司和组织共同制定的标准，用于进程间通信。是集群系统最流行的编程模型。有名的免费实现有：MPICH、 LAM/MPI 。 MPI 用于分布式共享内存编程

二.为什么用MPI 如何扩展硬件计算能力？ 更多 CPU(核) 更多 计算节点 MPI提供了计算节点的进程间通信机制

一个典型的高性能计算集群配置 I/O节点 所有计算节点通过交换机互联； 所有计算节点都通过I/O节点（作为NFS Server）访问一个或者多个磁盘阵列；所有的计算节点共享磁盘阵列存储空间 节点间配置rsh和ssh无密码登录 用户目录挂载到共享盘上，所有计算节点可以看到同样的用户目录 I/O节点

Google，yahoo等数据中心的典型网络配置 几千到几万个节点

三.消息通信模型 消息传递特点： 通过网络传递数据（消息）； send和receive必须配对使用； Message passing model

消息传递模型（伪代码表示） Process 1 (P1) Process 2 (P2) compute; compute; send( P2, info ); compute; compute; Receive( P1,info); idle; compute; idle; Send( P1, reply ); receive( P2, reply ); Communicate Waiting Synchronize 两个进程间互相通信

Hello world (1) 例 4.2 MPI编程基础 头文件 初始化 结束 #include <stdio.h> #include “mpi.h” int main(int argc, char *argv[] ) { MPI_Init( &argc, &argv ); printf( "Hello, world!\n" ); MPI_Finalize(); return 0; } 头文件 初始化 结束

int MPI_Init(int *argc, char ** argv); #include “mpi.h” int MPI_Init(int *argc, char ** argv); MPI_INIT必须在其它MPI函数之前调用，它完成MPI程序所有的初始化工作 main函数必须带参数运行,否则出错

int MPI_Finalize( ); MPI结束 MPI_FINALIZE是MPI程序的最后一个调用 #include “mpi.h” int MPI_Finalize( ); MPI_FINALIZE是MPI程序的最后一个调用 标志并行代码的结束,结束除主进程外其它进程

MPI程序的的编译与运行 编译 运行 mpicc –o helloworld1 helloworld1 .c 基本执行命令： 启动的进程数量 可执行文件的绝对或者相对路径 基本执行命令： mpiexec –n 5 ./helloworld1 运行 在指定的节点上运行： mpiexec -machinefile hostfile -n 5 ./helloworld1 在hostfile文件列出的节点上启动进程 hostfile格式： #主机名:该节点上最多允许启动的进程数 hosta hostb:2 2018年11月 MPI并行程序设计

运行MPI注意事项 mpiexec –n 4 helloworld1 (，除非该可执行文件路径在PATH或者通过-path执行搜索路径) 1.首先确认mpd守护进程已经启动，如果没有启动，在终端执行：mpd& 最新版本MPICH不再使用mpd进程管理器： http://www.mcs.anl.gov/research/projects/mpich2/ 【配置过程中，注意在/etc/hosts里面配置机器名和IP的对应关系】 2.错误的命令： ./helloworld1 () mpiexec –n 4 helloworld1 (，除非该可执行文件路径在PATH或者通过-path执行搜索路径) 3.正确的命令： mpiexec –n 4 ./helloworld1 mpiexec –n 4 ~/PP/lecture04 /helloworld1 2018年11月 MPI并行程序设计

MPI程序的执行流程 SPMD: Single Program Multiple Data(SPMD) MPMD: Multiple Program Multiple Data(MPMD) :::: #include "mpi.h" #include <stdio.h> main( int argc, char *argv[] ) { MPI_Init( &argc, &argv ); printf( "Hello, world!\n" ); MPI_Finalize(); } #include "mpi.h" #include <stdio.h> main( int argc, char *argv[] ) { MPI_Init( &argc, &argv ); printf( "Hello, world!\n" ); MPI_Finalize(); } #include "mpi.h" #include <stdio.h> main( int argc, char *argv[] ) { MPI_Init( &argc, &argv ); printf( "Hello, world!\n" ); MPI_Finalize(); } #include "mpi.h" #include <stdio.h> main( int argc, char *argv[] ) { MPI_Init( &argc, &argv ); printf( "Hello, world!\n" ); MPI_Finalize(); } #include "mpi.h" #include <stdio.h> main( int argc, char *argv[] ) { MPI_Init( &argc, &argv ); printf( "Hello, world!\n" ); MPI_Finalize(); } Hello World! 2018年11月 MPI并行程序设计

Hello world (2) 例 演示:helloworld2 当前进程ID 总的进程数 #include <stdio.h> #include “mpi.h” int main( int argc, char *argv[] ) { int myid, numprocs; MPI_Init( &argc, &argv ); MPI_Comm_rank( MPI_COMM_WORLD, &myid ); MPI_Comm_size( MPI_COMM_WORLD, &numprocs ); printf(“%d of %d：Hello world!\n”, myid, numprocs); MPI_Finalize(); return 0; } 演示:helloworld2 当前进程ID 总的进程数 2018年11月 MPI并行程序设计

int MPI_Comm_rank(MPI_Comm comm, int *rank); 当前进程ID 通信域 #include “mpi.h” int MPI_Comm_rank(MPI_Comm comm, int *rank); 返回调用进程在给定的通信域中的进程ID（整型，从0开始编号） 根据ID，不同的进程就可以将自身和其它的进程区别开来，实现各进程的并行和协作

int MPI_Comm_size(MPI_Comm comm, int *size) 通信域内的进程数 通信域 #include “mpi.h” int MPI_Comm_size(MPI_Comm comm, int *size) 返回给定的通信域中所包括的进程的个数

通信域(Communicator) MPI_COMM_WORLD是MPI预定义的全局通信域 10个节点的集群

Communicators & Groups 一个MPI程序可以有多个通信域，每一个进程组只能与组内的成员通信 进程组：进程的集合，可根据进程组创建通信域 Communicators & Groups

helloworld2执行流程 helloworld2 结束 SPMD(Simple Program Multiple Data) 进程0 MPI_Init MPI_Comm_Rank myid = 0 printf MPI_Finalize 进程1 myid = 1 进程2 myid = 2 结束

例 Hello world (3) MPI的进程间通信 2018年11月 MPI并行程序设计

Hello world (3) 例 当前进程ID 总的进程数 int main( int argc, char *argv[] ) { int numprocs, myid, source; MPI_Status status; char message[100]; MPI_Init( &argc, &argv ); MPI_Comm_rank( MPI_COMM_WORLD, &myid ); MPI_Comm_size( MPI_COMM_WORLD, &numprocs ); 当前进程ID 总的进程数 2018年11月 MPI并行程序设计

Hello world (3)续 非0号进程 0号进程 根据进程ID区分各个进程的执行逻辑 发送消息 接收消息 演示:helloworld3 if (myid != 0) { strcpy(message, "Hello World!"); MPI_Send(message, strlen(message)+1, MPI_CHAR, 0, 99, MPI_COMM_WORLD); } else { for (source = 1; source < numprocs; source++) { MPI_Recv(message, 100, MPI_CHAR, source, 99, MPI_COMM_WORLD, &status); printf("%s\n", message); } MPI_Finalize(); return 0; } /* end main */ 0号进程 接收消息 根据进程ID区分各个进程的执行逻辑 演示:helloworld3 MPI并行程序设计

MPI消息定义 信封 数据 ▲ 源/目的 ▲ 标识 ▲ 通信域 ▲ 起始地址 ▲ 数据个数 ▲ 数据类型

MPI_Datatype type, //发送数据元素的类型 int dest, //目标进程的ID int tag, //消息标签 发送消息 #include “mpi.h” int MPI_Send( void * buf, //发送缓冲区的地址 int count, //发送数据元素的个数 MPI_Datatype type, //发送数据元素的类型 int dest, //目标进程的ID int tag, //消息标签 MPI_Comm comm //通信域 ); 将该消息与发送到同一个进程内的其它消息区分开 将消息发送至另外一个进程 阻塞发送：直到消息被发送出去，该函数才返回

void * buf, //接收缓冲区的起始地址 int count, //接收数据元素的个数 接收消息 对应发送操作的tag，如果匹配任意标签，使用MPI_ANY_TAG #include “mpi.h” int MPI_Recv( void * buf, //接收缓冲区的起始地址 int count, //接收数据元素的个数 MPI_Datatype type, //接收数据元素的类型 int source, //发送进程的ID或MPI_ANY_SOURCE int tag, //消息标签 MPI_Comm comm, //通信域 MPI_Status *status //接收操作的状态 ); 从其它进程接收消息，如果从其它任意进程接收消息，source参数使用MPI_ANY_SOURCE 阻塞接收：直到接收到消息后，该函数才返回

最常用的MPI函数汇总 6个常用函数： MPI_Init(…); MPI_Comm_size(…); MPI_Comm_rank(…); MPI_Send(…); MPI_Recv(…); MPI_Finalize();

MPI程序结构 #include “mpi.h” MPI_Init(…) MPI_Comm_size(…) MPI_Comm_rank(…) 并行代码 计算任务+消息传递 MPI_Finalize();

死锁的产生 进程0 从进程1接收消息A 向进程1发送消息B 进程1 从进程0接收消息B 向进程0发送消息A 产生死锁的通信调用次序 1

死锁的产生 进程0 向进程1发送消息A 从进程1接收消息B 进程1 向进程0发送消息B 从进程0接收消息A 产生死锁的通信调用次序2

避免死锁 两个进程需要相互交换数据时，一定要将它们的发送和接收操作，按照次序进行匹配 进程0 向进程1发送消息A 从进程1接收消息B 进程1

4.3 MPI集合通信 1对多 多对1

Hello world (4) 例 0号线程 演示:helloworld4 向其它进程广播 main( int argc, char *argv[] ) { int numprocs, myid; char message[64]; MPI_Init( &argc, &argv ); MPI_Comm_rank( MPI_COMM_WORLD, &myid ); MPI_Comm_size( MPI_COMM_WORLD, &numprocs ); if (myid == 0) strcpy(message, "Hello World!"); MPI_Bcast(message, 64, MPI_CHAR, 0, MPI_COMM_WORLD); printf("%s\n", message); MPI_Finalize(); } 演示:helloworld4 0号线程 向其它进程广播 2018年11月 MPI并行程序设计

广播前后各进程缓冲区数据变化 root(0号进程) HelloWorld!\0 message[64] 广播前 广播后

void *buffer, /*发送/接收缓冲区*/ int count, /*广播或者接收的元素个数*/ 广播消息 #include “mpi.h” int MPI_Bcast( void *buffer, /*发送/接收缓冲区*/ int count, /*广播或者接收的元素个数*/ MPI_Datatype type, /*广播/接收数据的数据类型*/ int root, /*广播数据的根进程的ID*/ MPI_Comm comm /*通信域*/ ); 从root进程广播消息至所有其它的进程（同一个通信域内）

广播注意事项 通信域内的所有进程必须调用MPI_Bcast 所有的进程收到消息之后，MPI_Bcast才返回 data . MPI_Bcast(); Process 0 myrank = 0 Process 1 myrank = 1 Process p-1 myrank = p-1 通信域内的所有进程必须调用MPI_Bcast 所有的进程收到消息之后，MPI_Bcast才返回 不管是广播消息的根进程，还是从根接收消息的其它进程在调用形式上完全一致，即指明相同的根、相同的元素个数、以及相同的数据类型

思考 MPI_Bcast是用MPI_Send和MPI_Recv 实现的，你认为如何实现性能会比较高

还有没有 更好的实现 MPI_Bcast的可能实现 d: 广播的数据大小 n: 总的进程数 串行发送的时间： d * n / 125 千兆网卡，理论传输速率：125MB/s 还有没有 更好的实现 1 2 3 4 5 6 7 8 9 10 11 12 13 14 串行发送的时间： d * n / 125 64 * 128/125 = 66s 1 2 6 5 13 14 3 4 12 11 7 8 9 10 64 * log2128/125 = 8s 二叉树形式发送的时间： 2d * log2n / 125

MPI_Bcast的实现(小数据) 小数据：二项式树 (binomial tree） log2n 6 7 1

MPI_Bcast的实现(大数据) scatter t1= d / v t2= d / v gather O(1)

int MPI_Barrier(MPI_Comm comm); 本讲最后一个重要的函数 同步(栅栏) #include “mpi.h” int MPI_Barrier(MPI_Comm comm); 当每个进程都到达MPI_Barrier调用后，程序才接着往下执行

例 阅 读 下面的代码，分析程序的输出结果 int main( int argc, char **argv ) { int myid, size, i; int *table; int errors=0; MPI_Init( &argc, &argv ); MPI_Comm_rank( MPI_COMM_WORLD, &myid); MPI_Comm_size( MPI_COMM_WORLD, &size );

0号进程打印 table = (int *) malloc (size * sizeof(int)); table[myid] = myid + 1; /*准备要广播的数据*/ MPI_Barrier ( MPI_COMM_WORLD ); for ( i=0; i < size; i++ ) MPI_Bcast( &table[i], 1, MPI_INT, i, MPI_COMM_WORLD ); /* 检查接收到的数据的正确性 */ for ( i=0; i<size; i++ ) if (table[i] != i+1) errors++; /*检查完毕后执行一次同步*/ if(!myid) printf(“%d\n”, table[i]); MPI_Finalize(); } 演示:example_bcast 0号进程打印

如何发挥多核 的性能 4.4 MPI与Pthread混合编程 解决方案 混合编程：当前高性能计算领域流行的编程方式

例 若A为m×n矩阵，B为n×p矩阵，计算它们的乘积C=A× B. 要求使用MPI+pthread实现 并行方法

解题思路 主-从并行模式 1. 0号进程（主进程）将矩阵B广播给所有MPI进程（从进程）； 2. 主进程将矩阵A的各行依次发送给从进程； 主从模式（master-slave或master-worker或manager-worker）一个进程作为主进程负责分派任务，从进程执行任务 1. 0号进程（主进程）将矩阵B广播给所有MPI进程（从进程）； 2. 主进程将矩阵A的各行依次发送给从进程； 3. 从进程内启动多个线程； 4. 每一个线程分配B矩阵中的一列，与A中的一行相乘； 5. 计算结果发给主进程汇总； 6. 主进程搜集完所有的结果，结束！

MPI进程内 线 程 0 线 程 1 线 程 2 B A MPI进程 C 主进程分配A的一行给从进程

MPI进程内 MPI进程 B A C 从进程内：平均分配B矩阵的所有列 线 程 0 线 程 1 线 程 2 线 程 0 线 程 1

实现 广播矩阵B int main( int argc, char *argv[] ) { …… float A[M][N],B[N][P],C[M][P]; //变量声明 MPI_Init( &argc, &argv ); MPI_Comm_rank( MPI_COMM_WORLD, &myid ); MPI_Comm_size( MPI_COMM_WORLD, &numprocs ); if(!myid) { for(i = 0; i < M; i++) for(j = 0; j < N; j++) A[i][j] = i * j + 1; for(i = 0; i < N; i++) for(j = 0; j < P; j++) B[i][j] = i * j + 1; } MPI_Bcast(B[0], N * P, MPI_FLOAT, 0, MPI_COMM_WORLD); 在主进程内初始化矩阵A和B 广播矩阵B

主进程 给从进程依次分配A的一行 依次回收计算结果 如果A还没有分配完,向返回计算结果的从进程再分配A中的一行 if( !myid ) { /*主进程：分配任务和回收结果*/ j = (numprocs -1) < M ? (numprocs -1) : M; for(i = 1; i <= j; i++) { MPI_Send(A[i-1], N, MPI_FLOAT, i, 99, MPI_COMM_WORLD); } numsend = j; for(i = 1; i <= M; i++) { sender = (i - 1) % (numprocs - 1) + 1; MPI_Recv(C[i-1], P, MPI_FLOAT, sender, 100, MPI_COMM_WORLD, &status); if(numsend < M) { MPI_Send(A[i-1], N, MPI_FLOAT, sender, 99, MPI_COMM_WORLD); numsend++; }else { MPI_Send(&j, 0, MPI_INT, sender, 0, MPI_COMM_WORLD); } else {//从进程 给从进程依次分配A的一行 依次回收计算结果 如果A还没有分配完,向返回计算结果的从进程再分配A中的一行

从进程 else { numthreads = get_nprocs(); /*从进程(myid > 0)：接收主进程发来的任务， 计算完毕发回主进程*/ else { numthreads = get_nprocs(); tids = (pthread_t*)malloc(numthreads * sizeof(pthread_t)); A_row = (float*)malloc(N * sizeof(float)); C_row = (float*)malloc(P * sizeof(float)); targs = (struct threadArg*)malloc(numthreads * sizeof(struct threadArg)); for( i = 0; i < numthreads; i++) { targs[i].tid = i; targs[i].B = B; targs[i].A_row = A_row; targs[i].C_row = C_row; targs[i].numthreads = numthreads; } 从进程 从进程内的准备工作 从进程所在节点的CPU数 pthread_t * tids;tids数组用来存放线程ID 存放主进程分配的A中一行 存放计算结果C中的一行 线程参数，由于传入参数比较多，采用结果传递

从进程 从进程接收任务、计算、返回计算结果 接收主进程发送来A的一行 while(1) { MPI_Recv(A_row, N, MPI_FLOAT, 0, MPI_ANY_TAG, MPI_COMM_WORLD, &status); if(status.MPI_TAG == 0) break; for( i = 0; i < numthreads; i++) { pthread_create(&tids[i], NULL, worker, &targs[i]); } pthread_join(tids[i], NULL); MPI_Send(C_row, P, MPI_FLOAT, 0, 100, MPI_COMM_WORLD); }/*从进程结束*/ 若接收到标识为0的消息则退出执行 创建线程执行计算 等待线程内的计算完成

计算线程 void* worker(void* arg) { int i, j; struct threadArg* myarg = (struct threadArg*)arg; for(i = myarg->tid; i < P; i += myarg->numthreads) { myarg->C_row[i] = 0.0; for(j = 0; j < N; j++){ myarg->C_row[i] += myarg->A_row[j] * myarg->B[j][i] ; } return NULL; 平均分配B的所有列 B中的一列与A行相乘，计算结果存入C一行中的对应位置

思考1 例子代码的矩阵相乘的实现，可能存在 哪些性能问题

思考2 上述代码在某些情况下会死锁，找出死锁 的原因，并解决，作为第三次作业！

混合编程优点 缺点 每一个节点只需要一个线程参与通信，其它线程共享数据，具备较高的性能，例如广播 (2) 较快的上下文切换 (2) 较快的上下文切换 缺点 (1) 编程难度大！ (2) 多个线程同时调用MPI函数(注意看使用的MPI版本是否支持MPI_THREAD_MULTIPLE‎)，可能崩溃

本 章 内 容 小 结

2. MPI编程基础 3. 集合通信 4. MPI+Pthread混合编程 MPI 编 程 1. 认识MPI 用于分布式共享内存编程 基于TCP Socket的进程间通信 消息传递 2. MPI编程基础 MPI程序结构 MPI 编 程 通信域 编译与执行 mpicc –o helloworld helloworld.c mpiexec –n 5 ./helloworld 6个最常用的函数 3. 集合通信 MPI_Bcast：广播 MPI_Barrier：栅栏 4. MPI+Pthread混合编程 一个节点启动一个进程； 进程内一个线程处理通信，其它执行计算任务