C++ 并行计算 MPI Tutorial-8

reduce 介绍

规约 Reduce 是函数式编程中的经典概念。数据规约 涉及通过函数将一组数字简化为较小的一组数字。例如，假设我们有一个数字列表 [1、2、3、4、5] 。用 sum 函数 此数字列表规约 将产生 sum（[1、2、3、4、5]）=15。类似地，乘法规约将产生 multiply([1、2、3、4、5]) = 120。

就像您想象的那样，将归约函数应用于一组分布式数字可能非常麻烦。随之而来的是，难以有效地编程以设定顺序发生的规约。幸运的是，MPI 有一个方便的函数，称为MPI_Reduce，它将处理程序员在并行应用程序中需要执行的常见的规约操作。

MPI_Reduce

与 MPI_Gather 相似，MPI_Reduce 在每个进程上获取一个输入元素数组，并将输出元素数组返回给根进程。输出元素包含规约的结果。 MPI_Reduce 的原型如下所示：

MPI_Reduce(void* send_data,void* recv_data,int count,MPI_Datatype datatype,MPI_Op op,int root,MPI_Comm communicator)

send_data 参数是每个进程要规约的数据数组。 recv_data 仅与具有根级别的进程相关。 recv_data 数组包含规约的结果，大小为 sizeof（datatype）* count。 op 参数是您希望应用于数据的操作。MPI 包含一组可以使用的常见归约运算。尽管可以定义自定义归约操作，但这超出了本课程的范围。 MPI 定义的规约操作包括：

• MPI_MAX - 返回最大元素
• MPI_MIN - 返回最小元素
• MPI_SUM - 返回元素和
• MPI_PROD - 返回元素累乘
• MPI_LAND - 跨元素执行逻辑 AND
• MPI_LOR - 跨元素执行逻辑 OR
• MPI_BAND - 跨元素按位执行逻辑 AND
• MPI_BOR - 跨元素按位执行逻辑 OR
• MPI_MAXLOC - 返回元素最大值及其进程秩
• MPI_MINLOC - 返回元素最小值及其进程秩

下图说明了 MPI_Reduce 的通讯方式：

在上面，每个进程包含一个整数。以0 的根进程调用 MPI_Reduce，并使用 MPI_SUM作为规约操作。这四个数字相加后得出结果，并存储在根进程中。

查看进程包含多个元素时会发生什么也很有用。下图显示了每个进程规约多个数字的情况。

上图中的过程每个都有两个元素。结果求和基于每个元素进行。换句话说，不是将所有数组中的所有元素求和成一个元素，而是将每个数组中的第 i 个元素求和成过程0 的结果数组中的第 i 个元素。

使用 MPI_Reduce 计算数字平均值

在上一课中展示了如何使用 MPI_Scatter 和 MPI_Gather 计算平均值。使用 MPI_Reduce 可以简化上一课的代码

int main(int argc, char** argv) {
    if (argc != 2) {
    fprintf(stderr, "Usage: avg num_elements_per_proc\n");exit(1);}int num_elements_per_proc = atoi(argv[1]);MPI_Init(NULL, NULL);int world_rank;MPI_Comm_rank(MPI_COMM_WORLD, &world_rank);int world_size;MPI_Comm_size(MPI_COMM_WORLD, &world_size);// 所有进程都各自创建一个随机数组srand(time(NULL)*world_rank);   float *rand_nums = NULL;rand_nums = create_rand_nums(num_elements_per_proc);// 本地数字求和float local_sum = 0;int i;for (i = 0; i < num_elements_per_proc; i++) {
    local_sum += rand_nums[i];}// 在每个进程上打印随机数相关信息printf("Local sum for process %d - %f, avg = %f\n",world_rank, local_sum, local_sum / num_elements_per_proc);// 将所有本地和规约为总和float global_sum;MPI_Reduce(&local_sum, &global_sum, 1, MPI_FLOAT, MPI_SUM, 0,MPI_COMM_WORLD);// 打印结果if (world_rank == 0) {
    printf("Total sum = %f, avg = %f\n", global_sum,global_sum / (world_size * num_elements_per_proc));}// 清理free(rand_nums);MPI_Barrier(MPI_COMM_WORLD);MPI_Finalize();
}

>>> mpirun -n 4  ./reduce_avg 100
Local sum for process 0 - 51.385098, avg = 0.513851
Local sum for process 1 - 51.842468, avg = 0.518425
Local sum for process 2 - 49.684948, avg = 0.496849
Local sum for process 3 - 47.527420, avg = 0.475274
Total sum = 200.439941, avg = 0.501100

MPI_Allreduce

许多并行应用程序将需要在所有进程而不是根进程中访问规约的结果。 MPI_Allreduce以 MPI_Allgather 与 MPI_Gather 类似的互补方式，将规约值并将结果分配给所有进程。函数原型如下：

MPI_Allreduce(void* send_data,void* recv_data,int count,MPI_Datatype datatype,MPI_Op op,MPI_Comm communicator)

您可能已经注意到，MPI_Allreduce 与 MPI_Reduce 相同，不同之处在于它不需要根进程ID（因为结果分配到所有进程）。下面说明了 MPI_Allreduce 的通信模式：

MPI_Allreduce 等效于先执行 MPI_Reduce，再执行 MPI_Bcast

使用MPI_Allreduce 计算标准差

许多计算问题需要进行多次归约来解决问题。这样的问题之一是找到一组分布式数字的标准差。标准差是对数字与其均值的离散程度的度量。较低的标准差表示数字靠得更近，较高的标准偏差则相反。

要找到标准差，必须首先计算所有数字的平均值。计算完平均值后，便会计算出与平均值的平方差之和。求和平均值的平方根是最终结果。给定问题描述，我们知道所有数字至少会有两个和，转化为两个规约。

int main(int argc, char** argv) {
    if (argc != 2) {
    fprintf(stderr, "Usage: avg num_elements_per_proc\n");exit(1);}int num_elements_per_proc = atoi(argv[1]);MPI_Init(NULL, NULL);int world_rank;MPI_Comm_rank(MPI_COMM_WORLD, &world_rank);int world_size;MPI_Comm_size(MPI_COMM_WORLD, &world_size);// 所有进程都各自创建一个随机数组srand(time(NULL)*world_rank);float *rand_nums = NULL;rand_nums = create_rand_nums(num_elements_per_proc);// 本地数字求和float local_sum = 0;int i;for (i = 0; i < num_elements_per_proc; i++) {
    local_sum += rand_nums[i];}// 将所有本地总和规约为全局总和，以便计算平均值float global_sum;MPI_Allreduce(&local_sum, &global_sum, 1, MPI_FLOAT, MPI_SUM,MPI_COMM_WORLD);float mean = global_sum / (num_elements_per_proc * world_size);// 计算均值平方差的局部和float local_sq_diff = 0;for (i = 0; i < num_elements_per_proc; i++) {
    local_sq_diff += (rand_nums[i] - mean) * (rand_nums[i] - mean);}// 将平方差的全局和规约到根进程并打印出答案float global_sq_diff;MPI_Reduce(&local_sq_diff, &global_sq_diff, 1, MPI_FLOAT, MPI_SUM, 0,MPI_COMM_WORLD);// 标准差是均方根的平方根if (world_rank == 0) {
    float stddev = sqrt(global_sq_diff /(num_elements_per_proc * world_size));printf("Mean - %f, Standard deviation = %f\n", mean, stddev);}// 清理free(rand_nums);MPI_Barrier(MPI_COMM_WORLD);MPI_Finalize();
}

在上面的代码中，每个进程计算元素的 local_sum 并使用 MPI_Allreduce 对它们求和。在所有进程上都有全局和之后，可以计算平均值，以便可以计算 local_sq_diff。一旦计算出所有局部平方差，便可以使用 MPI_Reduce 找到 global_sq_diff。然后，根进程可以通过取全局平方差的平均值的平方根来计算标准偏差。

使用运行脚本运行示例代码将产生如下输出：

>>> mpirun -n 4  ./reduce_stddev 100
Mean - 0.501100, Standard deviation = 0.301126

参考

MPI 教程