基本集群通信

MPI broadcast及集群通信

集群通信指的是一个涉及communicator里面所有进程的一个方法。本节会介绍集群通信以及一个标准的方法-broadcasting（广播）

集群通信及同步点

集群通信在进程间引入了同步点的概念。这意味着所有进程在执行代码时必须都到达一个同步点才能继续执行后面代码。

MPI有一个特殊的函数来做同步进程操作

MPI_Barrier(MPI_Comm communicator)

函数名十分贴切（Barrier，屏障），这个方法会构建一个屏障，任何进程都没法跨越屏障，直到所有的进程都能到达屏障。水平轴代表程序执行，小圆圈代表不同的进程。

进程0在时间点（T1)首先调用MPI_BARRIER。然后进程0就一直等在屏障之前，之后进程1和进程3在（T2）时间点到达屏障，当进程2最终在时间点（T3)到达屏障时，其他进程就可以在（T4）时间点再次运行。

MPI_Barrier很多时候很有用，其中一个用途是来同步一个程序，使得分布式代码中的某一部分可以被精确计时。

想知道MPI_Barrier是怎么实现的么？之前的令牌环进程是最简单的一种实现屏障的方式，因为令牌只有在所有程序都完成之后才能被传递回第一个进程。

如果没法让所有进程都完成MPI_Barrier，那么也没法完成任何集体调用。

使用MPI Broadcast进行广播

broadcast是标准集群通信技术之一。一个广播发生时，一个进程会把同样一份数据传递给一个communicator中的其他所有进程。广播的主要用途之一是把用户输入传递给一个分布式程序，或者把一些配置参数传递给所有的进程。

广播的通信模式类似如下图：

在MPI里面，广播可以通过使用MPI_Bcast来做到，函数类似如下：

MPI_Bcast(
    void* data,
    int count,
    MPI_Datatype datatype,
    int root,
    MPI_Comm communicator)

尽管根节点和接收节点做不同的事情，他们都是调用同样的这个函数来实现广播，当根节点调用MPI_Bcast函数时，data变量里的值会被发送到其他节点上。当其他节点调用MPI_Bcast时，data变量会被赋值成从根节点接收到的数据。

使用MPI_Send和MPI_Recv来做广播

粗略来看，MPI_Bcast 仅仅是在 MPI_Send 和 MPI_Recv 基础上进行了一层包装。事实上，我们现在就可以自己来做这层封装。我们的函数叫做 my_bcast

void my_bcast(void* data, int count, MPI_Datatype datatype, int root,
              MPI_Comm communicator) {
  int world_rank;
  MPI_Comm_rank(communicator, &world_rank);
  int world_size;
  MPI_Comm_size(communicator, &world_size);

  if (world_rank == root) {
    // If we are the root process, send our data to everyone
    int i;
    for (i = 0; i < world_size; i++) {
      if (i != world_rank) {
        MPI_Send(data, count, datatype, i, 0, communicator);
      }
    }
  } else {
    // If we are a receiver process, receive the data from the root
    MPI_Recv(data, count, datatype, root, 0, communicator,
             MPI_STATUS_IGNORE);
  }
}

根节点把数据传递给其他所有节点，其他节点接收根节点的数据。类似的输出可能如下：

>>> cd tutorials
>>> ./run.py my_bcast
mpirun -n 4 ./my_bcast
Process 0 broadcasting data 100
Process 2 received data 100 from root process
Process 3 received data 100 from root process
Process 1 received data 100 from root process

基于树的沟通算法是最高效的

MPI_Bcast和MPI_Send以及MPI_Recv的比较

MPI_Bcast实际上就是实现了一个类似树形的广播算法来获得比较好的网络利用率。

MPI Scatter、Gather和Allgather

本节将介绍额外的两个机制来补充集体通信的知识-MPI_Scatter以及MPI_Gather，还会介绍MPI_Gather变体，MPI_Allgather

本节教程代码在tutorials/mpi-scatter-gather-and-allgather/code

MPI_Scatter介绍

MPI_Scatter是一个跟MPI_Bcast类似的集群通信机制，MPI_Scatter操作会指定一个指定的根进程，根进程会将数据发送到communicator里面的所有进程。MPI_Bcast和MPI_Scatter的主要区别很小但是很重要。MPI_Bcast给每个进程发送的是同样的数据，然而MPI_Scatter给每个进程发送的是一个数组的一部分数据，下图展示了区别。

在图中我们可以看到，MPI_Bcast 在根进程上接收一个单独的数据元素（红色的方块），然后把它复制到所有其他的进程。MPI_Scatter 接收一个数组，并把元素按进程的秩分发出去。第一个元素（红色方块）发往进程0，第二个元素（绿色方块）发往进程1，以此类推。尽管根进程（进程0）拥有整个数组的所有元素，MPI_Scatter 还是会把正确的属于进程0的元素放到这个进程的接收缓存中。下面的 MPI_Scatter 函数的原型。

MPI_Scatter(
    void* send_data,
    int send_count,
    MPI_Datatype send_datatype,
    void* recv_data,
    int recv_count,
    MPI_Datatype recv_datatype,
    int root,
    MPI_Comm communicator)

MPI_Gather介绍

MPI_Gather 跟 MPI_Scatter 是相反的。MPI_Gather 从好多进程里面收集数据到一个进程上面而不是从一个进程分发数据到多个进程。这个机制对很多平行算法很有用，比如并行的排序和搜索。下图是这个算法的一个示例。

跟MPI_Scatter类似，MPI_Gather从其他进程收集元素到根进程上面。元素是根据接收到的进程的秩排序的。MPI_Gather的函数原型跟MPI_Scatter长的一样。

MPI_Gather(
    void* send_data,
    int send_count,
    MPI_Datatype send_datatype,
    void* recv_data,
    int recv_count,
    MPI_Datatype recv_datatype,
    int root,
    MPI_Comm communicator)

使用MPI_Scatter和MPI_Gather来计算平均数

本节提供一个用来计算数组里面所有数字的平均数的样例程序（avg.c）。尽管这个程序十分简单，但是它展示了我们如何使用MPI来把工作拆分到不同的进程上，每个进程对一部分数据进行计算，然后再把每个部分计算出来的结果汇集成最终的答案。这个程序有以下几个步骤：

1. 在根进程（进程0）上生成一个充满随机数字的数组。
2. 把所有数字用MPI_Scatter分发给每个进程，每个进程得到的同样多的数字。
3. 每个进程计算它们各自得到的数字的平均数。
4. 根进程收集所有的平均数，然后计算这个平均数的平均数，得出最后结果。

代码里面有 MPI 调用的主要部分如下所示：

if (world_rank == 0) {
  rand_nums = create_rand_nums(elements_per_proc * world_size);
}

// Create a buffer that will hold a subset of the random numbers
float *sub_rand_nums = malloc(sizeof(float) * elements_per_proc);

// Scatter the random numbers to all processes
MPI_Scatter(rand_nums, elements_per_proc, MPI_FLOAT, sub_rand_nums,
            elements_per_proc, MPI_FLOAT, 0, MPI_COMM_WORLD);

// Compute the average of your subset
float sub_avg = compute_avg(sub_rand_nums, elements_per_proc);
// Gather all partial averages down to the root process
float *sub_avgs = NULL;
if (world_rank == 0) {
  sub_avgs = malloc(sizeof(float) * world_size);
}
MPI_Gather(&sub_avg, 1, MPI_FLOAT, sub_avgs, 1, MPI_FLOAT, 0,
           MPI_COMM_WORLD);

// Compute the total average of all numbers.
if (world_rank == 0) {
  float avg = compute_avg(sub_avgs, world_size);
}

代码开头根进程创建里一个随机数的数组。当MPI_Scatter被调用的时候，每个进程现在都持有elements_per_proc个原始数据里面的元素。每个进程计算子数组的平均数，然后根进程收集这些平均数。然后总的平均数就可以在这个小的多的平均数数组里面被计算出来。

可能的运行结果如下图：

>>> cd tutorials
>>> ./run.py avg
/home/kendall/bin/mpirun -n 4 ./avg 100
Avg of all elements is 0.478699
Avg computed across original data is 0.478699

MPI_Allgather以及修改后的平均程序

除了多对一场景以及一对多场景，发送多个元素到多个进程也很有用（也就是多对多通信模式）。MPI_Allgather就是这个作用。

对于分发在所有进程上的一组数据来说，MPI_Allgather会收集所有数据到所有进程上，从最基础的角度来看，MPI_Allgather相当于一个MPI_Gather操作之后跟着一个MPI_Bcast操作。下面的示意图显示了MPI_Allgather调用之后数据是如何分布的。

就跟MPI_Gather一样，每个进程上的元素是根据他们的秩为顺序被收集起来的，只不过这次是收集到了所有进程上面。很简单吧？MPI_Allgather的方法定义跟MPI_Gather几乎一样，只不过MPI_Allgather不需要root这个参数来指定根节点。

MPI_Allgather(
    void* send_data,
    int send_count,
    MPI_Datatype send_datatype,
    void* recv_data,
    int recv_count,
    MPI_Datatype recv_datatype,
    MPI_Comm communicator)

把计算平均数的代码修改为了Allgather

// Gather all partial averages down to all the processes
float *sub_avgs = (float *)malloc(sizeof(float) * world_size);
MPI_Allgather(&sub_avg, 1, MPI_FLOAT, sub_avgs, 1, MPI_FLOAT,
              MPI_COMM_WORLD);

// Compute the total average of all numbers.
float avg = compute_avg(sub_avgs, world_size);

可能的输出结果为：

>>> ./run.py all_avg
/home/kendall/bin/mpirun -n 4 ./all_avg 100
Avg of all elements from proc 1 is 0.479736
Avg of all elements from proc 3 is 0.479736
Avg of all elements from proc 0 is 0.479736
Avg of all elements from proc 2 is 0.479736

利用MPI计算并行排名

本节将通过MPI工具包编写一个实用的功能-并行排名来拓展基本的集群通信

并行排名-问题综述

当每个进程在其本地内存中存储了一个数，所有进程中存储的数字构成了一个数字集合，了解该数相对于整个数字集合的顺序是有用的。例如，用户可能正在对MPI群集中的处理器进行基准测试，并想知道每个处理器相对于其他处理器有多快。这个信息可用于安排、调度任务等。可以想象，如果所有其他数字分散在各个进程中，那么很难找出一个数字相对于所有其他数字的顺序。这个并行排名问题是我们在本课中要解决的问题。

下图展示了可能的输入和输出结果

并行排名-函数定义

TMPI_Rank(
    void *send_data,
    void *recv_data,
    MPI_Datatype datatype,
    MPI_Comm comm)

TMPI_Rank 把 send_data 作为缓冲区，其中包含一个类型为 datatype 的数字。 recv_data 在每个进程中只接收一个整数，即 send_data 的排名。comm 变量是进行排名的通讯器。

问题解决

gather_numbers_to_root 函数负责将所有数字收集到根进程（root process）

// 为进程0的TMPI_Rank收集数字。为MPI的数据类型分配空间
// 对进程0返回 void * 指向的缓冲区
// 对所有其他进程返回NULL
void *gather_numbers_to_root(void *number, MPI_Datatype datatype,
                             MPI_Comm comm) {
  int comm_rank, comm_size;
  MPI_Comm_rank(comm, &comm_rank);
  MPI_Comm_size(comm, &comm_size);

  // 在根进程上分配一个数组
  // 数组大小取决于所用的MPI数据类型
  int datatype_size;
  MPI_Type_size(datatype, &datatype_size);
  void *gathered_numbers;
  if (comm_rank == 0) {
    gathered_numbers = malloc(datatype_size * comm_size);
  }

  // 在根进程上收集所有数字
  MPI_Gather(number, 1, datatype, gathered_numbers, 1,
             datatype, 0, comm);

  return gathered_numbers;
}

通过使用新的MPI函数- MPI_Type_size 来收集datatype_size变量。尽管我们的代码仅支持将 MPI_INT 和 MPI_FLOAT 作为数据类型，但可将其扩展以支持不同大小的数据类型。 在使用 MPI_Gather 在根进程上收集了数字之后，必须在根进程上对数字进行排序，以确定它们排名。

排序数字

为了便于将所属进程附到对应数字上，我们在代码中创建了一个结构体（struct）来保存此信息。 我们的结构定义如下：

// 保存进程在通讯器中的次序（rank）和对应数字
// 该结构体用于数组排序，
// 并同时完整保留所属进程信息

typedef struct {
  int comm_rank;
  union {
    float f;
    int i;
  } number;
} CommRankNumber;

CommRankNumber 结构体保存了我们要排序的数字（记住它可以是浮点数或整数，因此我们使用联合体union），并且它拥有该数字所属进程在通讯器中的次序（rank）。 代码的下一部分，即 get_ranks 函数，负责创建这些结构体并对它们进行排序。

// 这个函数在根进程上对收集到的数字排序
// 返回一个数组，数组按进程在通讯器中的次序排序
// 注意 - 该函数只在根进程上运行

int *get_ranks(void *gathered_numbers, int gathered_number_count,
               MPI_Datatype datatype) {
  int datatype_size;
  MPI_Type_size(datatype, &datatype_size);

  // 将收集到的数字数组转换为CommRankNumbers数组
  // 这允许我们在排序的同时，完整保留数字所属进程的信息

  CommRankNumber *comm_rank_numbers = malloc(
    gathered_number_count * sizeof(CommRankNumber));
  int i;
  for (i = 0; i < gathered_number_count; i++) {
    comm_rank_numbers[i].comm_rank = i;
    memcpy(&(comm_rank_numbers[i].number),
           gathered_numbers + (i * datatype_size),
           datatype_size);
  }

  // 根据数据类型对comm_rank_numbers排序
  if (datatype == MPI_FLOAT) {
    qsort(comm_rank_numbers, gathered_number_count,
          sizeof(CommRankNumber), &compare_float_comm_rank_number);
  } else {
    qsort(comm_rank_numbers, gathered_number_count,
          sizeof(CommRankNumber), &compare_int_comm_rank_number);
  }

  // 现在comm_rank_numbers是排好序的，下面生成一个数组，
  // 包含每个进程的排名，数组第i个元素是进程i的数字的排名

  int *ranks = (int *)malloc(sizeof(int) * gathered_number_count);
  for (i = 0; i < gathered_number_count; i++) {
    ranks[comm_rank_numbers[i].comm_rank] = i;
  }

  // 清理并返回排名数组
  free(comm_rank_numbers);
  return ranks;
}

其中qsort需要使用特殊的排序函数

整合

现在将上面两个函数整合到一起，此函数将数字收集到根进程，并对数字进行排序以确定排名，然后将排名分散回请求的进程。

// 获取send_data的排名, 类型为datatype
// 排名用recv_data返回，类型为datatype
int TMPI_Rank(void *send_data, void *recv_data, MPI_Datatype datatype,
             MPI_Comm comm) {
  // 首先检查基本情况 - 此函数只支持MPI_INT和MPI_FLOAT

  if (datatype != MPI_INT && datatype != MPI_FLOAT) {
    return MPI_ERR_TYPE;
  }

  int comm_size, comm_rank;
  MPI_Comm_size(comm, &comm_size);
  MPI_Comm_rank(comm, &comm_rank);

  // 为了计算排名，必须将数字收集到一个进程中
  // 对数字排序, 然后将排名结果分散传回
  // 首先在进程0上收集数字
  void *gathered_numbers = gather_numbers_to_root(send_data, datatype,
                                                  comm);

  // 获取每个进程的次序（rank)
  int *ranks = NULL;
  if (comm_rank == 0) {
    ranks = get_ranks(gathered_numbers, comm_size, datatype);
  }

  // 分散发回排名结果
  MPI_Scatter(ranks, 1, MPI_INT, recv_data, 1, MPI_INT, 0, comm);

  // 清理
  if (comm_rank == 0) {
    free(gathered_numbers);
    free(ranks);
  }
}

结果

可能的运行结果如下：

>>> cd tutorials
>>> ./run.py random_rank
mpirun -n 4  ./random_rank 100
Rank for 0.242578 on process 0 - 0
Rank for 0.894732 on process 1 - 3
Rank for 0.789463 on process 2 - 2
Rank for 0.684195 on process 3 - 1