并行程序设计 学习笔记

intro

并行计算:

• 表示一个系统具有多个处理器, 所有处理器可以访问共享存储器以交换信息，通信开销低

分布式计算:

• 每个系统具有自己的存储器，并通过网络将多个系统连通，系统间通过消息传递的方式交换信息，通信开销高

Amdahl定律

• 加速比 $S =\frac{1}{(1 – a + a/n)}$
• $a$：可并行计算部分的占比
• $n$：并行处理节点个数

并行开销

寻找到足够多的并行任务，是实现预期加速的最大障碍
并行开销包括：

• 启动线程或进程的成本
• 共享数据的通信成本
• 同步的成本
• 额外的(冗余)计算

在某些系统上，每一种开销都可能是毫秒级(数百万次浮点运算)。
权衡：算法需要足够大的工作集(即大粒度)以快速的并行运行，但又不能太大以至于没有足够的并行任务

FP

并行计算机体系结构

• 单指令多数据流机SIMD（Single-Instruction Multiple-Data）
• 并行向量处理机PVP（Parallel Vector Processor）
• 对称多处理机SMP（Symmetric Multiprocessor）
• 大规模并行处理机MPP (Massively Parallel Processor)
• 工作站机群COW (Cluster of Workstation)
• 分布式共享存储DSM (Distributed Shared Memory) 多处理机

• 均匀存储访问模型－ UMA
• 非均匀存储访问模型－ NUMA
• 全高速缓存访问模型－COMA
• 高速缓存一致性非均匀存储访问模型－CC-NUMA
• 非远程存储访问模型－NORMA

UMA访存模型

UMA（Uniform Memory Access）模型是均匀存储访问模型的简称。其特点是：

• 物理存储器被所有处理器均匀共享
• 所有处理器访问任何存储字取相同的时间
• 每台处理器可带私有高速缓存
• 外围设备也可以一定形式共享

NUMA访存模型

NUMA(Nonuniform Memory Access)模型是非均匀存储访问模型的简称。特点是：

• 被共享的存储器在物理上是分布在所有的处理器中的，其所有本地存储器的集合就组成了全局地址空间；
• 处理器访问存储器的时间是不一样的；访问本地存储器LM或群内共享存储器CSM较快，而访问外地的存储器或全局共享存储器GSM较慢(此即非均匀存储访问名称的由来)；
• 每台处理器可拥有私有高速缓存，外设也可以某种形式共享

COMA访存模型

COMA(Cache-Only Memory Access)模型是全高速缓存存储访问的简称。其特点是：

• 各处理器节点中没有存储层次结构，全部高速缓存组成了全局地址空间
• 利用分布的高速缓存目录D进行远程高速缓存的访问
• COMA中的高速缓存容量一般都大于二级高速缓存容量
• 使用COMA时，数据开始时可任意分配，因为在运行时它最终会被迁移到要用到它们的地方。

CC-NUMA访存模型

CC-NUMA（Coherent-Cache Nonuniform Memory Access）模型是高速缓存一致性非均匀存储访问模型的简称。其特点是：

• 大多数使用基于目录的高速缓存一致性协议
• 保留SMP结构易于编程的优点，也改善常规SMP的可扩放性
• CC-NUMA实际上是一个分布共享存储的DSM多处理机系统
• 它最显著的优点是程序员无需明确地在节点上分配数据，系统的硬件和软件开始时自动在各节点分配数据，在运行期间，高速缓存一致性硬件会自动地将数据迁移至要用到它的地方

NORMA访存模型

NORMA（No-Remote Memory Access）模型是非远程存储访问模型的简称。NORMA的特点是：

• 所有存储器是私有的
• 绝大数NUMA都不支持远程存储器的访问
• 在DSM中，NORMA就消失了

并行程序开发方法

• 主-从式（Master-Slave）
• 单程序多数据流（Single Program Multiple Data ）
• 数据流水线（Data Pipelining）
• 分治策略（Divide and Conquer）

主-从式（Master-Slave）

基本思想是：将一个待求解的任务分成一个主任务（主进程）和一些从任务（子进程）

• 主进程负责将任务的分解、派发和收集诸各子任务的求解结果并最后汇总得到问题的最终解
• 各子进程接收主进程发来的消息；并行进行各自计算；向主进程发回各自的计算结果

单程序多数据流（SPMD）

基本思想是：并行运行的进程均执行相同的代码段，但处理的数据不同

• 首先将应用程序的数据预先分配给各个计算进程（处理器）
• 然后计算进程并行的完成各自的计算任务，包括计算过程中各进程间的数据交换（进行通信同步）
• 最后将计算结果汇集起来

数据流水线（Data Pipelining）

基本思想是：将计算进程组织成一条流水线，每个进程执行特定的计算任务（相当于流水线的一个阶段）

• 将任务在功能上划分成一些子任务（进程），这些子任务完成某种特定功能的计算工作
• 一旦前一个子任务完成，后继的子任务就可立即开始
• 整个计算过程中各进程之间的通信仅发生在相邻的阶段之间，且通信可以完全异步地进行

分治策略（Divide and Conquer）

基本思想是：将一个大而复杂的问题分解成若干个特性相同的子问题分而治之

• 若所得的子问题规模仍嫌过大，则可反复使用分治策略，直至很容易求解诸子问题为止
• 问题求解可分为三步：①将输入分解成若干个规模近似相等的子问题；②同时递归地求解诸子问题；③归并各子问题的解成为原问题的解

PCAM

设计并行应用的四个阶段：

• 划分(Partitioning)
• 通信(Communication)
• 组合(Agglomeration)
• 映射(Mapping)

划分：分解成小的任务，开拓并发性
通信：确定诸任务间的数据交换，监测划分的合理性
组合：依据任务的局部性，组合成更大的任务
映射：将每个任务分配到处理器上，提高并行性能

域分解

• 划分的对象是数据，可以是程序中的输入数据、中间处理数据和输出数据
• 将数据分解成大致相等的小数据片
• 划分时考虑数据上的相应操作
• 如果一个任务需要别的任务中的数据，则会产生任务间的通信

功能分解

• 划分的对象是计算（亦称为任务分解或计算划分），将计算划分为不同的任务，其出发点不同于域分解
• 划分后，研究不同任务所需的数据。如果这些数据不相交的，则划分是成功的；如果数据有相当的重叠， 意味着存 在大量的通信开销，要重新进行域分解和功能分解
• 功能分解是一种更深层次的分解

通信分析

通信是PCAM设计过程的重要阶段；
划分产生的各任务，一般不能完全独立执行，需要在任务间进行数据交流，从而产生了通信；
功能分解确定了各任务间的数据流；
各任务是并发执行的，通信则限制了这种并发性；

四种通信模式

• 局部/全局通信
• 结构化/非结构化通信
• 静态/动态通信
• 同步/异步通信

局部通信

通信限制在一个邻域内，即局部内通信

全局通信

通信是全局的，例如：

• All to All
• Master-Worker

结构化通信

每个任务的通信模式是相同的

非结构化通信

没有一个统一的通信模式，例如：无结构化网格

静态/动态通信

静态通信中，通信伙伴的身份不随时间改变
动态通信中，通信伙伴的身份则可能由运行时所计算的数据决定且是可变的

同步/异步通信

同步通信时，接收方和发送方协同操作
异步通信中，接收方获取数据无需与发送方协同

SMP

共享内存系统

• 每个处理器拥有私有存储（缓存）
• 所有处理器共享内存空间
• 处理器可以并行进行计算

共享数据访问

• 竞争条件（Race Condition）
  • 当执行的结果取决于两个或多个事件的执行时间时，就存在竞争条件
• 临界区（Critical Section）
  • 对共享存储区域进行更新的代码段，或会造成竞争条件的代码段
• 原子性（Atomicity）
  • 指事务的不可分割性，一个事务的所有操作要么不间断地全部被执行，要么一个也没有执行
• 互斥锁（Mutual Exclusion）
  • 在编程中，引入了对象互斥锁的概念，来保证共享数据操作的完整性。某段代码被标记了互斥锁，那么在任何情况下，最多只能有一个线程可以执行该段代码
• 路障（Barrier）
  • 同步点又称为路障(barrier)，只有所有线程都抵达此路障，线程才能继续运行下去，否则会阻塞在路障处

PThread

Pthreads 不是编程语言，而是POSIX 线程库，也经常称为Pthreads 线程库

• 定义了一套多线程编程的应用程序编程接口（API）
• 在支持POSIX 的系统( Linux 、Mac OS X 、Solaris 、HPUX 等)上才有效

Pthreads线程库支持：

• 创建并行环境
• 同步
• 隐式通信

启动线程

函数原型：

1

int pthread_create(pthread_t *, const pthread_attr_t *, void * (*)(void *), void *);

调用示例：

1

errcode = pthread_create(&thread_id, &thread_attribute, &thread_fun, &fun_arg);

• thread_id是线程id或句柄
• thread_attribute属性，通过传递NULL指针获得标准默认值
• thread_fun要运行的函数(获取并返回void*)
• fun_arg可以传递给thread_fun的参数
• 如果创建操作失败，错误代码将被设置为非零

1

pthread_create(&thread_handles[thread],NULL,Hello, (void*) thread);

停止线程

函数原型：

1

int pthread_join(pthread_t,void **);

调用示例：

1

errcode = pthread_create(thread_id,return);

• thread_id是线程id，函数等待thread_id对象关联的线程结束
• return参数可以接收该线程产生的返回值

1

pthread_join(thread_handles[thread], NULL);

互斥锁

• Pthreads线程库定义了互斥锁变量类型：pthread_mutex_t
• 特殊类型的变量，通过某些特殊类型的函数，互斥锁可以用来限制每次只有一个线程能进入临界区
• 使用pthread_mutex_t类型的变量前，必须初始化

1

int pthread_mutex_init(pthread_mutex_t* mutex_p, const pthread _mutexattr_t* attr_p);

• 使用结束后回收空间

1

int pthread_mutex_destroy(pthread_mutex_t* mutex_p) ;    

• 线程调用pthread_mutex_lock函数获得互斥锁

1

int pthread_mutex_lock(pthread_mutex_t* mutex_p);

• 线程执行完临界区代码后，需要调用pthread_mutex_unlock函数释放互斥锁的使用权

1

int pthread_mutex_unlock(pthread_mutex_t* mutex_p);

1
2
3
4
5
6
7
8
9

pthread_mutex_t mutex;/*全局变量*/
/*主函数*/
pthread_mutex_init(&mutex, NULL); /*互斥锁初始化*/
/*多线程并行执行函数部分*/        
pthread_mutex_lock(&mutex);
sum+=my_sum;
pthread_mutex_unlock(&mutex);
/*主函数*/
pthread_mutex_destroy(&mutex);/*回收互斥锁空间*/

信号量

信号量可以认为是一种特殊类型的unsigned int变量，可以赋值为0 、1 、2 ……，

• 只有0 和1 值的信号量称为二元信号量

互斥量最大的区别在于信号量没有个体拥有权，主线程信号量初始化，所有线程都可以通过调用sem_post和sem_wait函数更新信号量的值
需要链接信号量函数库

1

#include <semaphore.h>

• 使用信号量前，需要初始化

1

int sem_init(sem_t* semaphore_p, int shared, unsigned initial_val);

• 结束后回收

1

int sem_destroy(sem_t* semaphore_p);

• 线程再临界区前调用函数sem_wait函数：

1

int sem_wait(sem_t* semaphore_p);

• 如果信号量值为0，则线程阻塞
• 如果信号量值为非0，则将信号量值减1
• 线程执行完临界区后调用函数sem_wait函数：

1

int sem_post(sem_t* semaphore_p);

• 信号量的值加1

1
2
3
4
5
6
7
8
9

sem_t sem;/*全局变量*/
/*主函数*/
 sem_init(&sem, 0, 1);/*信号量初始值为1*/
/*多线程并行执行函数部分*/        
sem_wait(&sem);
sum+=my_sum;
sem_post(&sem);
/*主函数*/
sem_destroy(&sem);/*回收信号量空间*/

条件变量

条件变量是一种数据对象，允许线程在某个条件或事件发生前都处于挂起状态
另一个线程可以通过信号来唤醒挂起的线程
条件变量为共享变量，需要与互斥锁一起使用

• 使用pthread_cond_t类型的变量前，必须初始化

1

int pthread_cond_init(pthread_cond_t* cond_p, const pthread _condattr_t* cond_attr_p);

• 使用结束后回收空间

1

int pthread_cond_destroy(pthread_cond_t* cond_p);    

• 唤醒一个阻塞的线程

1

int pthread_cond_signal(pthread_cond_t* cond_p);  

• 唤醒所有阻塞的线程

1

int pthread_cond_broadcast(pthread_cond_t* cond_p);  

• 使用互斥锁阻塞线程

1

int pthread_cond_wait(pthread_cond_t* cond_p, pthread_mutex_t* mutex_p);

1
2
3
4

/*调用pthread_cond_wait等同于执行如下语句*/
pthread_mutex_unlock(&mutex_p);
wait_on_signal(&cond_p); 
pthread_mutex_lock(&mutex_p);

同步

忙等待和互斥锁实现路障

1
2
3
4
5

/*路障*/
pthread_mutex_lock(&barrier_mutex);
counter++;
pthread_mutex_unlock(&barrier_mutex);
while(counter < thread_count);

信号量实现路障

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

Sem_t count_sem; /*初始值为1*/
Sem_t barrier_sem; /*初始值为0*/
/*多线程并行执行函数部分*/ 
/*路障*/
Sem_wait(&count_sem);
if(counter == thread_count-1){
	counter = 0;
	sem_post(&count_sem);
	for(j=0;j<thread_count-1;j++)
		sem_post(&barrier_sem); /*最后一个线程唤醒前面阻塞的thread_count-1个线程*/
}else{
	counter++;
	sem_post(&count_sem);
	sem_wait(&barrier_sem);
}

条件变量实现路障

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

/*共享变量*/
int counter = 0;
pthread_mutex_t mutex_p;
pthread_cond_t cond_p;
/*多线程并行执行函数部分*/
/*路障*/
pthread_mutex_lock(&mutex_p);
counter++;
if (counter == thread_count){
	counter = 0;
	pthread_cond_broadcast(&cond_p);
}else{
	while(pthread_cond_wait(&cond_p,&mutex_p) != 0);
	/*除了pthread_cond_broadcast，可能有其他事件唤醒线程，因此需要将pthread_cond_wait放进while循环中*/
}
pthread_mutex_unlock(&mutex_p);

OpenMP

由三个主要API组件构成：

• 编译器指令
• 运行时库函数
• 环境变量

支持以增量方式并行化(Incremental Parallelization)串行程序
基于线程的并行编程模型
需要编译器的支持

准备工作和编译制导

• #include <omp.h>
  • 包含OpenMP的库函数头文件
• #pragma omp parallel private(nthreads, tid)
  • 并行区域的编译制导语句
  • 大小写敏感
  • 指令最多应用于一个后继语句，语句必须是一个结构块（structured block）

并行区域结构（Parallel Construct）

基本的OpenMP并行结构，Fork-Join模型

1
2
3
4
5
6
7
8
9
10
11

#pragma omp parallel [[clause[[,]clause]…] 
	structured-block
clause:
	if(scalar-expression)
	num_threads(integer-expression)
	default(shared | none)
	private(list)
	firstprivate(list)
	shared(list)
	copyin(list)
	reduction(reduction-identifier: list)

• 并行区域是多线程执行的代码块，是基本的OpenMP并行结构
• 一个线程执行到一个并行指令时，Fork一组线程并成为该线程组的主线程（线程号为0）
• 从并行区域开始，代码被复制，一组线程都执行该代码
• 并行区域出口隐含barrier，只有主线程继续执行

用于显示定义变量范围的子句

• private(list)：将list中的变量声明为每个线程的私有变量
  • 为线程组中每个线程声明一个相同类型的变量
  • 作用域只在并行区域内
• shared(list)：将list中的变量声明为线程组中线程之间的共享变量
• default(shared | none)：指定并行区域内变量的属性是shared，none作为默认值要求程序员必须显式地限定所有变量的作用域
• firstprivate(list)：在进入并行区域之前进行一次初始化，让并行区域的list中变量的值初始化为同名共享变量的值
• lastprivate(list)：在退出并行区域时，将并行区域的list中变量的值赋值给同名的共享变量
  • 循环迭代：将最后一次循环迭代中的值赋给对应的共享变量
  • section结构：将语法上最后一个section语句中的值赋给对应的共享变量
• copyin(list)：将主线程中threadprivate变量的值拷贝到执行并行区域的各个线程的threadprivate变量中，list包含要复制的变量的名称
  • threadprivate是指令，指定全局变量被所有线程各自产生一个私有的副本，对于不同并行区域之间的同一个线程，该副本变量是共享的
• copyprivate(list)：将单个线程私有list变量的值广播到其他线程的私有list变量
  • 只用于single指令，在一个single块的结尾处完成广播操作
• reduction(reduction-identifier: list) ：对list中的变量进行约简操作
  • 为每个线程创建并初始化list中变量的私有副本（list中变量为共享变量）
  • 对所有线程的私有副本进行约简操作，并将最终结果写入共享变量

if子句

• if(scalar-expression)：如果有if子句，那么只有表达式为真（非0）才会创建一个线程组，否则该区域由主线程串行执行

num_threads子句

• num_threads(integer-expression)：用于设置运行并行区域的线程数量，integer-expression表示线程数量
• 并行区域的线程数量由以下因素决定，优先级从高到低：
  • if子句的记过
  • num_threads子句的设置
  • omp_set_num_threads()库函数的设置
  • OMP_NUM_THREADS环境变量的设置
  • 编译器默认实现（一般默认总线程数等于处理器的核心数）

使用限制

• 并行区域不能是跨越多个程序或代码文件的结构化块
• 从一个并行区域只能由一个入口和一个出口，任何转入和转出都是非法的
  • 不能包含break语句
• 只允许一个if子句和一个num_threads子句
• 程序运行结果不能依赖于子句的顺序

工作共享结构(Worksharing Constructs)

• 工作共享结构不会启动新线程，为了使指令能够并行执行，必须将工作共享结构封装在一个并行区域中
• 工作共享结构将所包含的代码划分给线程组的成员来执行
• 进入工作共享结构没有barrier，但在出口隐含barrier
  • 并行do/for loop：线程组并行执行代码，实现数据并行
  • 并行sections：计算任务分成单独的、不连续的部分，每个线程执行一部分，可以实现函数并行
  • single：串行执行代码

for指令

for指令指定循环语句必须由线程组并行执行，假设已经启动了并行区域，否则将串行执行

1
2
3
4
5
6
7
8
9
10
11

#pragma omp for [[clause[[,]clause]…] 
	for-loops
clause:
	schedule(kind[,chunk_size])
	ordered[(n)]
	private(list)
	firstprivate(list)
	lastprivate(list)
	reduction(reduction-identifier : list)
	collapse(n)
	nowait

schedule子句

• schedule(kind[,chunk_size])：描述循环迭代如何划分给多个线程
• kind可以为static（静态）、dynamic（动态）、guided（引导）、runtime（运行时）、auto（自动）五种模式
• static：循环迭代被划分成小块，静态的分配给线程。
  • chunk_size为块大小，如果没有指定则迭代均匀连续划分给线程
• dynamic：循环迭代被划分成小块，在线程间动态调度。
  • 当线程完成一块时，动态分配另一块，块大小默认为1
• guided：当线程请求任务时，迭代块被动态分配给线程，直到所有迭代块被分配完为止。与dynamic类似，只是每次分配的迭代块的大小会减少。
  • 初始块大小与num_iterations/num_threads成正比
  • 后续分配的块大小与num_iterations_remain/num_threads成正比
  • chunk_size定义最小块大小，默认为1
• runtime：运行时根据环境变量omp_schedule在确定调度类型，最终使用的是上述三种之一。
• auto：由编译器或者运行时系统决定调度类型

ordered子句和nowait子句

• ordered[(n)]：指定区域的循环迭代将按串行顺序执行，与单个处理器处理结果顺序一致
  • ordered子句只能用在for或parallel for中
• nowait：忽略并行区域隐含barrier的同步等待，

collapse子句

• collapse(n)：指定嵌套循环中的n个循环折叠到一个大的迭代空间中，并根据调度子句划分并行执行
  • 所有相关循环的顺序执行决定了折叠迭代空间中迭代的顺序
  • 能够解决线程间负载均衡或线程负载太小的问题

使用限制

• 不能是while循环，或者任何循环迭代次数不确定的循环
• 循环迭代变量（i）必须是整数，并且对于所有线程，循环 控制参数（i++）必须相同
• 程序的正确性不能依赖于某个线程执行的特定迭代
• 跳转或跳出循环是非法的
• 块大小必须为整数次迭代
• schedule、ordered和collapse子句只可以出现一次

sections指令

• sections指令指定所包含的代码段被分配给各个线程执行
• 不同section部分可以由不同线程执行，如果一个线程运行的块，也可以执行多个部分

1
2
3
4
5
6
7
8
9
10
11
12
13
14

#pragma omp sections [clause[ [, ] clause] ...]
{
	#pragma omp section
		structured-block
	#pragma omp section
		structured-block
	...
}
clause:
	private(list)
	firstprivate(list)
	lastprivate(list)
	reduction(reduction-identifier: list)
	nowait

使用限制

• sections指令的末尾有隐含的barrier，可以使用nowait子句忽略
• 不能跳转或跳出section代码块
• section指令必须在封闭的sections指令的词法范围内

single指令

single指令指定所包含的代码仅由一个线程执行，通常用于处理非线程安全的代码段，例如I/O

1
2
3
4
5
6
7

#pragma omp single [clause[ [, ]clause] ...]
	structured-block
clause:
	private(list)
	firstprivate(list)
	copyprivate(list)
	nowait

使用限制

不能跳转或跳出一个single代码块

合并结构(Combined Constructs)

• 合并了并行区域结构与共享任务结构的指令
• parallel for指令：合并parallel和for两个指令
  • 除了nowait子句外，所有parallel和for适用的子句和规范也都适用于parallel for指令
• parallel sections指令：合并parallel和sections两个指令
  • 除了nowait子句外，所有parallel和for适用的子句和规范也都适用于parallel for指令

同步结构

实现多线程间互斥访问和同步的指令

• master指令：指定一个代码区域由主线程执行，其他线程跳过这个区域
• critical指令：指定一个代码区域每次只能由一个线程执行（互斥访问），可以实现临界区访问
• barrier指令：指定线程组所有的线程在此指令处同步
• atomic指令：指定以原子方式访问特定的存储位置，该指令仅适用于其后的单个语句，可以实现一个最小临界区的访问
• flush指令：标识一个数据同步点，将线程的变量写回内存，实现内存数据更新
• ordered指令：指定循环迭代以串行执行顺序执行

master指令

• master指令没有隐含barrier
• 跳转或跳出master代码块是非法的

critical指令

1
2

#pragma omp critical [(name)]
	structured-block

• 如果一条线程正在一个critical区域执行而另一个线程到达这个区域，并企图执行，那么它将会被阻塞，直到第一个线程离开这个区域
• name是可选项，使不同的cirtical区域共存，具有相同命名的不同的critical区域被当作同一个区域，所有未命名critical区域被当作同一个区域

atomic指令

1
2
3

#pragma omp atomic [atomic-clause]
	expression-stmt
atomic-clause: read, write, update, or capture

flush指令

• 明确的表明程序点处需要进行内存更新
• 指令隐含flush指令，不过如果由nowait子句，则flush指令失效：
  • barrier指令；
  • parallel指令——进入和退出
  • critical指令——进入和退出
  • ordered指令——进入和退出
  • for指令——退出
  • sections指令——退出
  • single指令——退出

ordered指令

• 指定循环迭代以串行执行顺序执行，如果前面的迭代没有完成，则执行后面迭代的线程需要等待
• ordered指令只能出现在出现在for或者parallel for的动态范围内

其他指令

• threadprivate指令：指定全局变量被所有线程各自产生一个私有的副本，对于不同并行区域之间的同一个线程，该副本变量是共享的

运行时库函数

MPI

• 使用消息传递的实现称为消息传递接口（Message-Passing Interface，MPI）
• MPI是一个消息传递接口标准，而不是编程语言
• MPI标准定义了一组具有可移植性的编程接口
• MPI以语言独立的形式存在，可运行在不同的操作系统和硬件平台上
• 共有上百个函数调用接口，C/C++和Fortran语言中可以直接对这些函数进行调用

MPI基本函数

• MPI初始化：通过MPI_Init函数进入MPI环境并完成所有的初始化工作

1

int MPI_Init( int *argc, char ***argv )

• MPI结束：通过MPI_Finalize函数从MPI环境中退出

1

int MPI_Finalize(void)

• 获取进程的编号：调用MPI_Comm_rank函数获得当前进程在指定通信域中的编号，将自身与其他程序区分

1

int MPI_Comm_rank(MPI_Comm comm, int *rank)

• 获取指定通信域的进程数：调用MPI_Comm_size函数获取指定通信域的进程个数，确定自身完成任务比例

1

int MPI_Comm_size(MPI_Comm comm, int *size)

• 消息发送：MPI_Send函数用于发送一个消息到目标进程

1

int MPI_Send(void *buf, int count, MPI_Datatype dataytpe, int dest, int tag, MPI_Comm comm) 

• 消息接收：MPI_Recv函数用于从指定进程接收一个消息

1

int MPI_Recv(void *buf, int count, MPI_Datatype datatyepe,int source, int tag, MPI_Comm comm, MPI_Status *status)

MPI消息

• 消息的内容即信的内容，在MPI中称为消息缓冲(Message Buffer)
• 消息缓冲由三元组<起始地址，数据个数，数据类型>标识
• 消息的接收/发送者即信的地址，在MPI中称为消息信封(Message Envelop)
• 消息信封由三元组<源/目标进程，消息标签，通信域>标识

数据类型

• MPI的消息类型分为两种：预定义类型和派生数据类型(Derived Data Type)
• 预定义数据类型：MPI支持异构计算(Heterogeneous Computing)
  • 异构计算指在不同计算机系统上运行程序，每台计算可能有不同生产厂商，不同操作系统。
  • MPI通过提供预定义数据类型来解决异构计算中的互操作性问题，建立它与具体语言的对应关系
• 派生数据类型：MPI引入派生数据类型来定义由数据类型不同且地址空间不连续的数据项组成的消息

MPI_PACKED

1
2
3
4
5
6
7
8

double A[100];
MPI_Pack_size (50,MPI_DOUBLE,comm,&BufferSize);
TempBuffer = malloc(BufferSize);
j = sizeof(MPI_DOUBLE);
Position = 0;
for (i=0;i<50;i++)
	MPI_Pack(A+i*j,1,MPI_DOUBLE,TempBuffer,BufferSize,&Position,comm);
MPI_Send(TempBuffer,Position,MPI_PACKED,destination,tag,comm); 

• MPI_Pack_size函数来决定用于存放50个MPI_DOUBLE数据项的临时缓冲区的大小
• 调用malloc函数为这个临时缓冲区分配内存
• for循环中将数组A的50个偶序数元素打包成一个消息并存放在临时缓冲区
• 将临时缓冲区的数据类型为MPI_PACKED的消息发送

消息打包，然后发送

1
2
3
4
5

MPI_Pack(buf, count, dtype, 
			//以上为待打包消息描述
			packbuf, packsize, packpos, 
			//以上为打包缓冲区描述
			communicator)

消息接收，然后拆包

1
2
3
4
5

MPI_Unpack(packbuf, packsize, packpos,
			//以上为拆包缓冲区描述
			buf, count, dtype,
			// 以上为拆包消息描述
			communicatior)

派生数据类型

• 派生数据类型可以用类型图来描述，这是一种通用的类型描述方法，它是一系列二元组<基类型，偏移>的集合，可以表示成如下格式：

  • {<基类型0,偏移0>，···，<基类型n-1,偏移n-1>}

• 在派生数据类型中，基类型可以是任何MPI预定义数据类型，也可以是其它的派生数据类型，即支持数据类型的嵌套定义

• MPI提供了全面而强大的构造函数(Constructor Function)来定义派生数据类型

MPI_Type_vector

• MPI_Type_vector(count, blocklength, stride, oldtype, &newtype)
  • count：派生数据类型newtype由count个数据块构成。
  • blocklength和oldtype：每个数据块由blocklength个oldtype类型的连续数据项组成。
  • stride：两个连续数据块的起始位置之间的oldtype类型元素的个数。因此，两个块之间的间隔可以由(stride-blocklength)来表示

MPI_Type_struct

1
2
3
4
5
6
7

MPI_Type_struct(
    count, //成员数 
    array_of_blocklengths, //成员块长度数组
    array_of_displacements,//成员偏移数组     
    array_of_types, //成员类型数组
    newtype // 新类型
) 

通信域

MPI提供丰富的函数用于管理通信域

组间通信域

• 组间通信域是一种特殊的通信域，该通信域包括了两个进程组，分属于两个进程组的进程之间通过组间通信域实现通信
• 一般把调用进程所在的进程组称为本地进程组，而把另外一个称为远程进程组

点对点通信

• 共有下面四种通信模式：
  • 同步(synchronous)通信模式
  • 缓冲(buffered)通信模式
  • 标准(standard)通信模式
  • 就绪(ready)通信模式
• 同时也提供了阻塞和非阻塞两种通信机制
• 不同通信模式和不同通信机制的结合，便产生了非常丰富的点对点通信函数

同步(synchronous)通信模式

• 同步通信模式：只有相应的接收过程已经启动，发送过程才正确返回
• 同步发送返回后，表示发送缓冲区中的数据已经全部被系统缓冲区缓存，并且已经开始发送
• 同步发送返回后，发送缓冲区可以被释放或者重新使用

缓冲(buffered)通信模式

• 缓冲通信模式：缓冲通信模式的发送不管接收操作是否已经启动都可以执行
• 但是需要用户程序事先申请一块足够大的缓冲区，通过MPI_Buffer_attach实现，通过MPI_Buffer_detach来回收申请的缓冲区

标准(standard)通信模式

• 标准通信模式：是否对发送的数据进行缓冲由MPI的实现来决定，而不是由用户程序来控制
• 发送可以是同步的或缓冲的，取决于实现

就绪(ready)通信模式

• 就绪通信模式：发送操作只有在接收进程相应的接收操作已经开始才进行发送
• 当发送操作启动而相应的接收还没有启动，发送操作将出错。就绪通信模式的特殊之处就是接收操作必须先于发送操作启动

通信机制

• 阻塞和非阻塞通信的主要区别在于返回后的资源可用性
• 阻塞通信返回的条件：
  • 通信操作已经完成，即消息已经发送或接收。
  • 调用的缓冲区可用。若是发送操作，则该缓冲区可以被其它的操作更新；若是接收操作，该缓冲区的数据已经完整，可以被正确引用
• 非阻塞通信返回后并不意味着通信操作的完成，MPI还提供了对非阻塞通信完成的检测，主要的有两种MPI_Wait函数和MPI_Test函数
• MPI的发送操作支持四种通信模式，它们与阻塞属性一起产生了MPI中的8种发送操作
• MPI的接收操作只有两种：阻塞接收和非阻塞接收

Send-Recv函数

• 给一个进程发送消息，从另一个进程接收消息；
• 特别适用于在进程链（环）中进行“移位”操作，而避免在通讯为阻塞方式时出现死锁

1
2
3
4
5

MPI_Sendrecv( sendbuf, sendcount, sendtype, dest, sendtag,
			//以上为消息发送的描述
			recvbuf, recvcount, recvtype, source, recvtag, 
			// 以上为消息接收的描述
			comm, status)

集合通信

• 集合通信一般实现三个功能：通信、聚集和同步
  • 通信功能主要完成组内数据的传输
  • 聚集功能在通信的基础上对给定的数据完成一定的操作
  • 同步功能实现组内所有进程在执行进度上取得一致
• 按照通信方向的不同，又可以分为三种：
  • 一对多通信：一个进程向其它所有的进程发送消息，这个负责发送消息的进程叫做Root进程
  • 多对一通信：一个进程从其它所有的进程接收消息，这个接收的进程也叫做Root进程
  • 多对多通信：每个进程都向其它所有的进程发送或者接收消息

广播MPI_Bcast

• Root进程发送相同的消息给通信域Comm中的所有进程
• 对Root进程来说，这个三元组既定义了发送缓冲也定义了接收缓冲。对其它进程来说，这个三元组只定义了接收缓冲

1

MPI_Bcast(Address, Count, Datatype, Root, Comm)

收集MPI_Gather

• Root进程从进程域Comm的所有进程(包括它自已)接收消息
• 消息按照进程的标识rank排序并进行拼接，然后存放在Root进程的接收缓冲中
• 所有非Root进程忽略接收缓冲

1

MPI_Gather(SendAddress, SendCount, SendDatatype, RecvAddress, RecvCount, RecvDatatype, Root, Comm)

散播MPI_Scatter

• Scatter执行与Gather相反的操作
• Root进程给所有进程(包括它自已)发送不同的消息，这n (n为进程域comm包括的进程个数)个消息在Root进程的发送缓冲区中按进程标识的顺序有序地存放

1

MPI_Scatter(SendAddress, SendCount, SendDatatype, RecvAddress, RecvCount, RecvDatatype, Root, Comm)

全局收集MPI_Allgather

• Allgather操作相当于每个进程都作为Root进程执行了一次Gather调用，即每一个进程都按照Gather的方式收集来自所有进程(包括自己)的数据

1

MPI_Allgather(SendAddress, SendCount, SendDatatype, RecvAddress, RecvCount, RecvDatatype, Comm)

全局交换MPI_Alltoall

• 每个进程发送一个消息给所有进程(包括它自已)
• n (n为进程域comm包括的进程个数)个消息在发送缓冲中以进程标识的顺序有序地存放。从接收角度看，每个进程都从所有进程接收一个消息，这n个消息以标号的顺序被连接起来，存放在接收缓冲中
• 全局交换等价于每个进程作为Root进程执行了一次散播操作

1

MPI_Alltoall(SendAddress, SendCount, SendDatatype, RecvAddress, RecvCount, RecvDatatype, Comm)

路障MPI_Barrier

• 在路障同步操作MPI_Barrier(Comm)中，通信域Comm中的所有进程相互同步

1

MPI_Barrier(Comm)

归约MPI_Reduce

• 归约操作对每个进程的发送缓冲区(SendAddress)中的数据按给定的操作进行运算，并将最终结果存放在Root进程的接收缓冲区(RecvAddress)中
• 参与计算操作的数据项的数据类型在Datatype中定义，归约操作由Op定义
• 归约操作可以是MPI预定义的,也可以是用户自定义的
• 归约操作允许每个进程提供向量值，而不只是标量值，向量的长度由Count定义

1

MPI_Reduce(SendAddress, RecvAddress, Count, Datatype, Op, Root, Comm)

扫描MPI_scan

• 可以把扫描操作看作是一种特殊的归约，即每一个进程都对排在它前面的进程进行归约操作
• MPI_SCAN调用的结果是：对于每一个进程i，它对进程0,1,…,i的发送缓冲区的数据进行了指定的归约操作
• 扫描操作也允许每个进程贡献向量值，而不只是标量值。向量的长度由Count定义

1

MPI_scan(SendAddress, RecvAddress, Count, Datatype, Op, Comm)