什么是 MPI

MPI 全名叫 Message Passing Interface，即信息传递接口，作用是可以通过 MPI 可以在不同进程间传递消息，从而可以并行地处理任务，即进行并行计算。需要注意的是，尽管我们偶尔会说使用 MPI 编写了某某可执行程序，但是 MPI 其实只是一个库，而不是一种语言，其可以被 Fortran、C、C++、Python 调用

说到并行，为了避免在概念上混淆，那我们又不得不再提一下串行、并发、同步、异步这几个容易混淆的概念

  • 串行：指的是在执行多个任务时，各个任务按顺序执行，完成一个后才能进行下一个
  • 并行：指的是多个任务可以同时执行
  • 并发：并发编程又叫多线程编程，指的是在执行多个任务时，因为资源不够所以采用一个 CPU 轮换着进行这些任务，从而提高任务效率，因为切换任务的速度很快，所以看上去是多个任务一起进行
  • 同步：同步就是顺序执行，执行完一个再执行下一个，需要等待、协调运行。例如消息发送方必须等待消息接收方接受完才能继续执行后续任务
  • 异步：异步就是彼此独立，在等待某件事的过程中继续做自己的事。例如消息发送方再发送完一条消息后不必等待接收方的接收即可继续执行后续任务

在开始 MPI 之前还需要再了解一下有关 MPI 的一些基本概念，从而帮助我们在宏观上理解 MPI 通信

首先是通信域(communicating domain)，通信域定义了一组能够互相发消息的进程。在这组进程中，每个进程会被分配一个序号，称为 rank，进程间显性地通过指定 rank 作为标识来进行通信，一个进程 rank 可以指定另一个进程的 rank 以及独一无二的消息标签 tag 来发送消息。接收者也可以发送一个特定标签标记的消息的请求。类似于这样的涉及一个发送者以及一个接收者的通信被称为点对点(point-to-point)通信

当然在很多情况下，某个进程可能需要跟其他所有进程进行通信，比如主进程想发一个广播给所有的从进程。在这种情况下，手动去写一个个点对点通信就显得很笨拙，而且这样会导致网络利用率很低。MPI 有专门的接口帮我们处理这里所有进程间的集体性(collective)通信

MPI 的程序流程

学习记录

L1

MPI_Init()

  • 初始化进程环境

MPI_Comm_rank()

  • 获取当前通信域内的进程号

MPI_Comm_size()

  • 获取当前通信域内的进程数

MPI_Finalize()

  • 释放进程环境
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
#include <iostream>
#include <mpi.h>

using namespace std;

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

	int world_rank;
	MPI_Comm_rank(MPI_COMM_WORLD, &world_rank);
	int world_size;
	MPI_Comm_size(MPI_COMM_WORLD, &world_size);

	printf("Process %d of %d: Hello world!\n", world_rank, world_size);

	MPI_Finalize();
	return 0;
}

L2

MPI_Send()

  • 发送数据

MPI_Bsend()

  • 先设立缓存区，再发送数据

MPI_Rsend()

  • 同步通信，确认数据接收后再进行下一步

MPI_Recv()

  • 接受数据

MPI_Buffer_attach()

  • 绑定 MPI 缓存区

MPI_Buffer_detach()

  • 释放 MPI 缓存区
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
#include <iostream>
#include <mpi.h>

using namespace std;

int send() {
	int buf = 114514;
	printf("Send: %d\n", buf);
	return MPI_Send(&buf, 1, MPI_INT, 1, 1, MPI_COMM_WORLD);
}

int Bsend() {
	int data = 114514, buf_size = 1;
	int* buf = (int*)malloc((buf_size + MPI_BSEND_OVERHEAD) * sizeof(int));
	MPI_Buffer_attach(buf, buf_size + MPI_BSEND_OVERHEAD);
	printf("Send: %d\n", data);
	MPI_Bsend(&data, buf_size, MPI_INT, 1, 1, MPI_COMM_WORLD);
	return MPI_Buffer_detach(&buf, &buf_size);
}

int Rsend() {
	int buf = 114514;
	printf("Send: %d\n", buf);
	return MPI_Rsend(&buf, 1, MPI_INT, 1, 1, MPI_COMM_WORLD);
}

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

	int idx, size;
	MPI_Comm_rank(MPI_COMM_WORLD, &idx);
	MPI_Comm_size(MPI_COMM_WORLD, &size);

	if (size != 2) {
		printf("Error Process Number.\n");
		MPI_Abort(MPI_COMM_WORLD, 1);
		return 0;
	}

	if (idx == 0) {
		Rsend();
	}
	else if (idx == 1) {
		int buf;
		MPI_Status st;
		MPI_Recv(&buf, 1, MPI_INT, 0, 1, MPI_COMM_WORLD, &st);
		printf("Recv: %d\n", buf);
	}

	MPI_Finalize();
	return 0;
}

L3

MPI_Bcast()

  • 数据广播
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
#include <iostream>
#include <mpi.h>

using namespace std;

int main_3(int argc, char* argv[]) {
	int rank, value;
	MPI_Init(&argc, &argv);

	MPI_Comm_rank(MPI_COMM_WORLD, &rank);
	do {
		if (rank == 0)
			scanf("%d", &value);
		MPI_Bcast(&value, 1, MPI_INT, 0, MPI_COMM_WORLD);
		printf("Process %d got %d\n", rank, value);
	} while (value > 0);

	MPI_Finalize();

	return 0;
}

L4

MPI_Gather()

  • 数据收集

MPI_Gatherv()

  • 更高自由度的数据收集

MPI_Scatter()

  • 数据扩散

MPI_Scatterv()

  • 更高自由度的数据扩散

MPI_Allgather()

  • 数据批量收集

MPI_Allgatherv()

  • 更高自由度的数据批量收集
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
#include <iostream>
#include <mpi.h>
#include <cstring>

using namespace std;

void Gather() {
	int g_size;
	int data;

	int rank;
	MPI_Comm_rank(MPI_COMM_WORLD, &rank);
	MPI_Comm_size(MPI_COMM_WORLD, &g_size);

	data = rank;
	int* rebuf = (int*)malloc(g_size * 1 * sizeof(int));
	MPI_Gather(&data, 1, MPI_INT, rebuf, 1, MPI_INT, 0, MPI_COMM_WORLD);

	if (rank == 0) {
		for (int i = 0; i < 100; i++)
			cout << rebuf[i] << endl;
	}

	free(rebuf);

	return;
}

void Gatherv() {
	int g_size;
	int rank;
	MPI_Comm_rank(MPI_COMM_WORLD, &rank);
	MPI_Comm_size(MPI_COMM_WORLD, &g_size);

	int* data = (int*)malloc(g_size * sizeof(int));
	int* recvcounts = (int*)malloc(g_size * sizeof(int));
	int* displs = (int*)malloc(g_size * sizeof(int));
	int* recvbuf = (int*)malloc(g_size * sizeof(int));

	data[rank] = rank;
	for (int i = 0; i < g_size; i++) {
		recvcounts[i] = i;
		displs[i] = i;
	}

	MPI_Gatherv(data, 1, MPI_INT, recvbuf, recvcounts, displs, MPI_INT, 0, MPI_COMM_WORLD);

	if (rank == 0) {
		for (int i = 0; i < g_size; i++)
			cout << recvcounts[i] << endl;
	}

	free(data);
	free(recvcounts);
	free(displs);
	free(recvbuf);

	return;
}

void Scatter() {
	int g_size;
	int rank;
	MPI_Comm_rank(MPI_COMM_WORLD, &rank);
	MPI_Comm_size(MPI_COMM_WORLD, &g_size);

	int* sendbuf = new int[g_size];
	for (int i = 0; i < g_size; i++)
		sendbuf[i] = i;
	int recvbuf;

	MPI_Scatter(sendbuf, 1, MPI_INT, &recvbuf, 1, MPI_INT, 0, MPI_COMM_WORLD);

	printf("%d got %d\n", rank, recvbuf);

	delete[] sendbuf;

	return;
}

void Scatterv() {
	int g_size;
	int rank;
	MPI_Comm_rank(MPI_COMM_WORLD, &rank);
	MPI_Comm_size(MPI_COMM_WORLD, &g_size);

	int* sendbuf = new int[2 * g_size];
	int* sendcounts = new int[g_size];
	int* displs = new int[g_size];
	for (int i = 0; i < g_size; i++) {
		sendbuf[2 * i] = 2 * i;
		sendbuf[2 * i + 1] = 2 * i + 1;
		sendcounts[i] = 2;
		displs[i] = 2 * i;
	}
	int recvbuf[2];

	MPI_Scatterv(sendbuf, sendcounts, displs, MPI_INT, &recvbuf, 2, MPI_INT, 0, MPI_COMM_WORLD);

	printf("%d got %d %d\n", rank, recvbuf[0], recvbuf[1]);

	delete[] sendbuf;
	delete[] sendcounts;
	delete[] displs;

	return;
}

void AllGather() {
	int g_size;
	int rank;
	MPI_Comm_rank(MPI_COMM_WORLD, &rank);
	MPI_Comm_size(MPI_COMM_WORLD, &g_size);

	int data = rank;
	int* recvbuf = new int[g_size];

	MPI_Allgather(&data, 1, MPI_INT, recvbuf, 1, MPI_INT, MPI_COMM_WORLD);

	printf("Process %d:", rank);
	for (int i = 0; i < g_size; i++)
		printf(" %d", recvbuf[i]);

	delete[] recvbuf;

	return;
}

void AllGatherv() {
	int g_size;
	int rank;
	MPI_Comm_rank(MPI_COMM_WORLD, &rank);
	MPI_Comm_size(MPI_COMM_WORLD, &g_size);

	int data = rank;
	int* recvbuf = new int[g_size];
	int* recvcount = new int[g_size];
	int* displs = new int[g_size];

	for (int i = 0; i < g_size; i++) {
		recvcount[i] = 1;
		displs[i] = i;
	}

	MPI_Allgatherv(&data, 1, MPI_INT, recvbuf, recvcount, displs, MPI_INT, MPI_COMM_WORLD);

	printf("Process %d:", rank);
	for (int i = 0; i < g_size; i++)
		printf(" %d", recvbuf[i]);

	delete[] recvbuf;
	delete[] recvcount;
	delete[] displs;

	return;
}

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

	AllGatherv();

	MPI_Finalize();
	return 0;
}

L5

MPI_Reduce()

  • 数据汇聚

MPI_Reduce_scatter()

  • 数据汇聚后再广播
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
#include <iostream>
#include <mpi.h>

using namespace std;

void Reduce() {
	int size, rank;
	MPI_Comm_size(MPI_COMM_WORLD, &size);
	MPI_Comm_rank(MPI_COMM_WORLD, &rank);

	int data = rank;
	int recvbuf;

	MPI_Reduce(&data, &recvbuf, 1, MPI_INT, MPI_MAX, 0, MPI_COMM_WORLD);

	if (rank == 0)
		cout << recvbuf << endl;

	return;
}

void ReduceScatter() {
	int size, rank;
	MPI_Comm_size(MPI_COMM_WORLD, &size);
	MPI_Comm_rank(MPI_COMM_WORLD, &rank);

	int* data = new int[size];
	int recvbuf;
	int* recvcounts = new int[size];

	for (int i = 0; i < size; i++) {
		data[i] = (rank + i) % 10;
		recvcounts[i] = 1;
	}

	MPI_Reduce_scatter(data, &recvbuf, recvcounts, MPI_INT, MPI_MAX, MPI_COMM_WORLD);

	printf("%d got %d\n", rank, recvbuf);

	delete[] data;
	delete[] recvcounts;

	return;
}

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

	ReduceScatter();

	MPI_Finalize();
	return 0;
}

L6

一个用多进程计算圆周率的案例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
#include <iostream>
#include <mpi.h>
#include <ctime>

using namespace std;

double f(double x) {
	return 4 / (1 + x * x);
}

double loop(int l, int r, int n) {
	double res = 0;
	for (int i = l; i < r; i++)
		res += f((i - 0.5) / n);
	return res;
}

int main(int argc, char argv[]) {
	int n = 1;
	double res;
	MPI_Init(NULL, NULL);

	int rank, size;
	MPI_Comm_size(MPI_COMM_WORLD, &size);
	MPI_Comm_rank(MPI_COMM_WORLD, &rank);

	if (rank == 0) {
		cout << "Enter n : ";
		cin >> n;
	}

	MPI_Bcast(&n, 1, MPI_INT, 0, MPI_COMM_WORLD);

	clock_t start = clock();
	double data = loop((n / size) * rank, (n / size) * (rank + 1), n);
	MPI_Reduce(&data, &res, 1, MPI_DOUBLE, MPI_SUM, 0, MPI_COMM_WORLD);
	clock_t end = clock();

	if (rank == 0) {
		cout << "花费了" << (double)(end - start) / CLOCKS_PER_SEC << "秒" << endl;
		printf("%.18lf", res / n);
	}

	MPI_Finalize();

	return 0;
}