Quartz 5

进阶集群通信

7 min read

MPI Reduce和Allreduce

本节将通过MPI_Reduce和MPI_Allreduce进一步扩展集体通信例程。

本节代码在tutorials/mpi-reduce-and-allreduce/code 下。

归约简介

归约是函数式编程中的经典概念。数据归约包括通过函数将一组数字规约为较小的一组数字。例如，假设我们有一个数字列表 [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 参数是每个进程都希望归约的 datatype 类型元素的数组。 recv_data 仅与具有 root 秩的进程相关。 recv_data 数组包含归约的结果，大小为sizeof（datatype）* count。 op 参数是您希望应用于数据的操作。 MPI 包含一组可以使用的常见归约运算。 尽管可以定义自定义归约操作，但这超出了本教程的范围。 MPI 定义的归约操作包括：

  • MPI_MAX 返回最大元素。
  • MPI_MIN 返回最小元素。
  • MPI_SUM 对元素求和。
  • MPI_PROD 将所有元素相乘。
  • MPI_LAND 对元素执行逻辑_与_运算。
  • MPI_LOR 对元素执行逻辑_或_运算。
  • MPI_BAND 对元素的各个位按位_与_执行。
  • MPI_BOR 对元素的位执行按位_或_运算。
  • MPI_MAXLOC 返回最大值和所在的进程的秩。
  • MPI_MINLOC 返回最小值和所在的进程的秩。

下面是 MPI_Reduce 通信模式的说明。

在上图中，每个进程包含一个整数。 调用 MPI_Reduce 的根进程为 0，并使用 MPI_SUM 作为归约运算。 这四个数字相加后将结果存储在根进程中。

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

不是将所有数组中的所有元素累加到一个元素中，而是将每个数组中第i个元素累加到进程0结果数组中的第i个元素中。

使用MPI_Reduce计算均值

上一节中利用MPI_Scatter和MPI_Gather计算平均值，使用MPI_Reduce可以简化代码

float *rand_nums = NULL;
rand_nums = create_rand_nums(num_elements_per_proc);

// Sum the numbers locally
float local_sum = 0;
int i;
for (i = 0; i < num_elements_per_proc; i++) {
  local_sum += rand_nums[i];
}

// Print the random numbers on each process
printf("Local sum for process %d - %f, avg = %f\\n",
       world_rank, local_sum, local_sum / num_elements_per_proc);

// Reduce all of the local sums into the global sum
float global_sum;
MPI_Reduce(&local_sum, &global_sum, 1, MPI_FLOAT, MPI_SUM, 0,
           MPI_COMM_WORLD);

// Print the result
if (world_rank == 0) {
  printf("Total sum = %f, avg = %f\\n", global_sum,
         global_sum / (world_size * num_elements_per_proc));
}

在上面的代码中，每个进程都会创建随机数并计算和保存在 local_sum 中。 然后使用 MPI_SUM 将 local_sum 归约至根进程。 然后，全局平均值为 global_sum / (world_size * num_elements_per_proc)。 如果您从 repo 的 tutorials 目录中运行 reduce_avg 程序，则输出应与此类似。

MPI_Allreduce

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

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计算标准差

许多问题需要多次归约来解决，一个这样的问题是找到一组分布式数字的标准差。

本问题可以转化为两个归约：

rand_nums = create_rand_nums(num_elements_per_proc);

// Sum the numbers locally
float local_sum = 0;
int i;
for (i = 0; i < num_elements_per_proc; i++) {
  local_sum += rand_nums[i];
}

// Reduce all of the local sums into the global sum in order to
// calculate the mean
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);

// Compute the local sum of the squared differences from the mean
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);
}

// Reduce the global sum of the squared differences to the root
// process and print off the answer
float global_sq_diff;
MPI_Reduce(&local_sq_diff, &global_sq_diff, 1, MPI_FLOAT, MPI_SUM, 0,
           MPI_COMM_WORLD);

// The standard deviation is the square root of the mean of the
// squared differences.
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);
}

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

可能的输出为：

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

Graph View