c - MPI 奇偶排序如何工作？

Question

只是在面试前查看了一些笔记，并且正在努力理解如何 Odd-Even sort在并行架构中工作。

int MPI_OddEven_Sort(int n, double *a, int root, MPI_Comm comm)
{
    int rank, size, i, sorted_result;
    double *local_a;

// get rank and size of comm
    MPI_Comm_rank(comm, &rank); //&rank = address of rank
    MPI_Comm_size(comm, &size);

    local_a = (double *) calloc(n / size, sizeof(double));


// scatter the array a to local_a
    MPI_Scatter(a, n / size, MPI_DOUBLE, local_a, n / size, MPI_DOUBLE,
        root, comm);

// sort local_a
    merge_sort(n / size, local_a);

//odd-even part
    for (i = 0; i < size; i++) {

        if ((i + rank) % 2 == 0) {  // means i and rank have same nature
            if (rank < size - 1) {
                MPI_Compare(n / size, local_a, rank, rank + 1, comm);
            }
        } else if (rank > 0) {
            MPI_Compare(n / size, local_a, rank - 1, rank, comm);
        }

        MPI_Barrier(comm);
        // test if array is sorted
        MPI_Is_Sorted(n / size, local_a, root, comm, &sorted_result);

        // is sorted gives integer 0 or 1, if 0 => array is sorted
        if (sorted_result == 0) {
            break;
        }           // check for iterations
    }

// gather local_a to a
    MPI_Gather(local_a, n / size, MPI_DOUBLE, a, n / size, MPI_DOUBLE,
           root, comm)
    return MPI_SUCCESS;
}

是我为此功能编写的一些代码(不是今天也不是昨天!)。有人可以分解一下它是如何工作的吗？

我将我的数组 a 分散到每个处理器，它正在获取 local_a 的副本(大小为 n/size)

正在对每个 local_a 调用合并排序。

这之后发生了什么？ (假设到目前为止我是正确的!)

Answer 1

看到这些 PRAM 类型的分拣网络在这些年后再次出现，有点有趣。这些东西的并行计算的原始心智模型是微型处理器的大规模并行阵列作为“比较器”，例如Connection Machines。 - 在网络比 CPU/RAM 便宜的那一天。当然，这最终看起来与 80 年代中后期及以后的 super 计算机非常不同，甚至与 90 年代后期的 x86 集群不同；但现在它们开始流行起来 with GPUs和其他加速器，如果你眯着眼睛，它们实际上看起来有点像过去的 future 。

看起来你上面的更像是一个 Baudet-Stevenson odd-even sort ，它已经开始朝着假设处理器将在本地存储多个项目的方向发展，并且您可以通过在通信步骤之间对这些本地列表进行排序来充分利用处理器。

充实你的代码并稍微简化它，我们有这样的东西:

#include <stdio.h>
#include <stdlib.h>
#include <mpi.h>

int merge(double *ina, int lena, double *inb, int lenb, double *out) {
    int i,j;
    int outcount=0;

    for (i=0,j=0; i<lena; i++) {
        while ((inb[j] < ina[i]) && j < lenb) {
            out[outcount++] = inb[j++];
        }
        out[outcount++] = ina[i];
    }
    while (j<lenb)
        out[outcount++] = inb[j++];

    return 0;
}

int domerge_sort(double *a, int start, int end, double *b) {
    if ((end - start) <= 1) return 0;

    int mid = (end+start)/2;
    domerge_sort(a, start, mid, b);
    domerge_sort(a, mid,   end, b);
    merge(&(a[start]), mid-start, &(a[mid]), end-mid, &(b[start]));
    for (int i=start; i<end; i++)
        a[i] = b[i];

    return 0;
}

int merge_sort(int n, double *a) {
    double b[n];
    domerge_sort(a, 0, n, b);
    return 0;
}

void printstat(int rank, int iter, char *txt, double *la, int n) {
    printf("[%d] %s iter %d: <", rank, txt, iter);
    for (int j=0; j<n-1; j++)
        printf("%6.3lf,",la[j]);
    printf("%6.3lf>\n", la[n-1]);
}

void MPI_Pairwise_Exchange(int localn, double *locala, int sendrank, int recvrank,
                           MPI_Comm comm) {

    /*
     * the sending rank just sends the data and waits for the results;
     * the receiving rank receives it, sorts the combined data, and returns
     * the correct half of the data.
     */
    int rank;
    double remote[localn];
    double all[2*localn];
    const int mergetag = 1;
    const int sortedtag = 2;

    MPI_Comm_rank(comm, &rank);
    if (rank == sendrank) {
        MPI_Send(locala, localn, MPI_DOUBLE, recvrank, mergetag, MPI_COMM_WORLD);
        MPI_Recv(locala, localn, MPI_DOUBLE, recvrank, sortedtag, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
    } else {
        MPI_Recv(remote, localn, MPI_DOUBLE, sendrank, mergetag, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
        merge(locala, localn, remote, localn, all);

        int theirstart = 0, mystart = localn;
        if (sendrank > rank) {
            theirstart = localn;
            mystart = 0;
        }
        MPI_Send(&(all[theirstart]), localn, MPI_DOUBLE, sendrank, sortedtag, MPI_COMM_WORLD);
        for (int i=mystart; i<mystart+localn; i++)
            locala[i-mystart] = all[i];
    }
}

int MPI_OddEven_Sort(int n, double *a, int root, MPI_Comm comm)
{
    int rank, size, i;
    double *local_a;

// get rank and size of comm
    MPI_Comm_rank(comm, &rank); //&rank = address of rank
    MPI_Comm_size(comm, &size);

    local_a = (double *) calloc(n / size, sizeof(double));


// scatter the array a to local_a
    MPI_Scatter(a, n / size, MPI_DOUBLE, local_a, n / size, MPI_DOUBLE,
        root, comm);
// sort local_a
    merge_sort(n / size, local_a);

//odd-even part
    for (i = 1; i <= size; i++) {

        printstat(rank, i, "before", local_a, n/size);

        if ((i + rank) % 2 == 0) {  // means i and rank have same nature
            if (rank < size - 1) {
                MPI_Pairwise_Exchange(n / size, local_a, rank, rank + 1, comm);
            }
        } else if (rank > 0) {
            MPI_Pairwise_Exchange(n / size, local_a, rank - 1, rank, comm);
        }

    }

    printstat(rank, i-1, "after", local_a, n/size);

// gather local_a to a
    MPI_Gather(local_a, n / size, MPI_DOUBLE, a, n / size, MPI_DOUBLE,
           root, comm);

    if (rank == root)
        printstat(rank, i, " all done ", a, n);

    return MPI_SUCCESS;
}

int main(int argc, char **argv) {

    MPI_Init(&argc, &argv);

    int n = argc-1;
    double a[n];
    for (int i=0; i<n; i++)
        a[i] = atof(argv[i+1]);

    MPI_OddEven_Sort(n, a, 0, MPI_COMM_WORLD);

    MPI_Finalize();

    return 0;
}

所以它的工作方式是列表在处理器之间平均分配(不相等的分配也很容易处理，但它需要大量额外的簿记工作，对本次讨论没有太大帮助)。

我们首先对本地列表进行排序(时间复杂度为 O(n/P ln n/P))。当然，没有理由必须是合并排序，除了在这里我们可以在以下步骤中重新使用该合并代码。然后我们进行 P 个邻居交换步骤，每个方向各一半。这里的模型是有一个线性网络，我们可以在其中直接快速地与近邻交流，也许根本无法与更远的邻居交流。

original odd-even sorting network是每个处理器都有一个 key 的情况，在这种情况下，通信很容易——你将你的项目与你的邻居进行比较，并在必要时交换(因此这基本上是一个并行的冒泡排序)。在这种情况下，我们在进程对之间进行简单的并行排序——在这里，每一对只将所有数据发送到其中一个，该对合并已经在本地排序的列表 O(N/P)，然后给出适当的一半的数据返回到另一个处理器。我拿出了你的支票，如果完成了；可以证明是在P次邻居交换中完成的。您当然可以将其添加回去，以防提前终止；然而，所有的处理器都必须在一切完成时达成一致，这需要类似于 all reduce 的东西。 ，这在一定程度上打破了原始模型。

所以我们每个链接有 O(n) 次数据传输，(发送和接收 n/P 项，每次 P 次)，每个处理器执行 (n/P ln n/P) + (2 n/P - 1) *P/2 = O(n/P ln n/P + N) 次比较；在这种情况下，还需要考虑分散和聚集，但通常这种排序是在数据就位的情况下完成的。

运行上面的代码——为清楚起见，使用相同的示例 in that document linked给出(重新排序输出以使其更易于阅读):

$ mpirun -np 4 ./baudet-stevenson 43 54 63 28 79 81 32 47 84 17 25 49
[0] before iter 1: <43.000,54.000,63.000>
[1] before iter 1: <28.000,79.000,81.000>
[2] before iter 1: <32.000,47.000,84.000>
[3] before iter 1: <17.000,25.000,49.000>

[0] before iter 2: <43.000,54.000,63.000>
[1] before iter 2: <28.000,32.000,47.000>
[2] before iter 2: <79.000,81.000,84.000>
[3] before iter 2: <17.000,25.000,49.000>

[0] before iter 3: <28.000,32.000,43.000>
[1] before iter 3: <47.000,54.000,63.000>
[2] before iter 3: <17.000,25.000,49.000>
[3] before iter 3: <79.000,81.000,84.000>

[0] before iter 4: <28.000,32.000,43.000>
[1] before iter 4: <17.000,25.000,47.000>
[2] before iter 4: <49.000,54.000,63.000>
[3] before iter 4: <79.000,81.000,84.000>

[0] after iter 4: <17.000,25.000,28.000>
[1] after iter 4: <32.000,43.000,47.000>
[2] after iter 4: <49.000,54.000,63.000>
[3] after iter 4: <79.000,81.000,84.000>

[0]  all done  iter 5: <17.000,25.000,28.000,32.000,43.000,47.000,49.000,54.000,63.000,79.000,81.000,84.000>