并行合并排序性能

Question

我试图直观地了解如果我并行化合并排序，我可以加速多少。

到目前为止我的想法：

如果 N 是要排序的数组中的元素数，那么 log(base 2)N 是我需要的最大内核数。我相信是这种情况，因为合并排序中有 2*log(base 2)N + 1 个级别。首先，你将它一遍又一遍地除以二，然后一遍又一遍地合并两个排序的数组，直到你再次拥有一个包含 N 项的数组（现在它已经排序了）。

我试图弄清楚这实际上会提高多少性能。我认为随着我们向算法中间移动，由于额外内核而导致的性能提升将会增加，因为我们可以使用更多内核。假设我在一个未排序的数组中有 16 个项目。我只需要使用一个核心将其分解为两个 8 项数组，然后我可以使用两个核心将它们分解为四个 4 项数组，等等。

因此，性能会在每个级别的拆分中增加两倍，然后在每个级别的合并中降低两倍......对吗？我在正确的轨道上吗？

另外，为什么我们不能先合并未排序数组中的前两项，然后再合并接下来的两项，以此类推。基本摆脱了算法的前半部分？

想法？

我应该改为在 math.stackexchange.com 上问这个吗？抱歉，如果是这样...我真的不知道

Answer 1

如果您想通过并行化提高 MergeSort 的性能，您应该并行化拆分（在合并结果之前执行的部分）。我假设你有多个 CPU 节点。

拆分： 让当前 CPU 节点保留数组的一半并将另一半交给另一个 CPU 节点。不断重复这个过程。随着您深入到树中，并行度会增加（正如您所提到的）

基本情况： 当数据是一项时，当前 CPU 节点将其发送回其父节点。 父节点将等待子节点传递数据，然后再进行任何合并。

合并： 一旦接收到来自节点子节点的数据，该节点（该子节点的父节点）就可以开始将接收到的数据与其自己的数据合并。合并完成后，将其传递给其父节点，依此类推。由于每个节点都是一个单独的 CPU，因此较低级别的合并是并行完成的。随着我们上树，这种并行性会降低。 （就像你提到的）

这应该会加快合并排序。

然而，wikipedia http://en.wikipedia.org/wiki/Merge_sort#Parallel_processing 上的这篇文章表明，您可以通过并行化和专门化合并步骤（以及在数据大小

我很好奇你为什么不使用快速排序。它非常适合并行化！

编辑：

另外，为什么我们不能从合并前两个项目开始呢？ 未排序的数组，然后是接下来的两个，依此类推。基本上摆脱了 算法的前半部分？

回答你的问题：

这就是合并排序所做的，它合并前 2 个，下一个 2 等等，但是为了得到它们，它使用递归。这使得运行时 O(n*2log(n)) 因为有 2 棵树（一棵在拆分时创建，一棵在合并回一个大列表时创建）。结果为 O(nlog(n))。

根据您的想法，从底部开始，取 2 x 2 的数字并对其进行排序。然后扩大边界以包含 2 个块（每个块有 2 个数字）4 个数字......等等。你正在建造一棵树，从叶子到根。这类似于锦标赛算法（尽管您只能有一个获胜者 - 树的根）。

运行时：起初你有 n 个数字。您循环设置每 2 个数字 O(n/2)、下一级 O(n/4)、下一个 O(n/8) 等的边界。 构建这棵树需要 O(log(n))。但是您仍然必须将其他数字合并到一个列表中。由于您有 n 个数字，因此 n*O(nlogn) 为您提供与合并排序 nlogn 相同的运行时间。

总结： 所以我想说的是，你从底层合并的想法还是很长的。您正在摆脱其中一棵树，因此速度提升并不显着。

Answer 2

1) 并行让每个处理器对数组的 n/p 进行串行排序。

2) 对于 i=1 到 log_2(p) 您必须将两个数组与 2^i 处理器合并。 在 O(log 2^i) 时间内，使用一个处理器使用数组的最大部分（我们将其分成两部分）的中点进行二进制搜索，以找到匹配的位置并在那里形成另一个分区。

例子：

A= 12345 6789

B=1234 567 89

最大的部分是 12345，中点是 3。使用二进制搜索找到这个 3 在另一个数组中的位置并将其拆分。新数组：

A= 12 345 6789 B= 12 34 567 89

您可以使用优先级队列来跟踪哪个数组部分最大。

将 A 和 B 数组拆分为 O(p) 部分后，您可以在每个小块上并行进行串行合并。要获得每个配对的输出位置的偏移量，您可以事先进行并行前缀求和。

O(n/p log n/p) //串行排序，如果可以基数排序则只有O(n/p)

O( log(p) *(log(p) +(n/p) ) = O(log(p)^2 + log(p)(n/p)) // 并行合并

Answer 3

众所周知，Dual Pivot QuickSort 在串行版本中击败了合并排序。我认为 DPQ 的并行形式确实可能是有史以来最快的排序算法。原因是它的常数因子比 MergeSort 低，并且其最坏情况的时间复杂度发生的概率仅为 1/(n!)。如果 N 很大，则更喜欢 DPQ，如果可能的话，可能是多线程。但是并行性有一个支点或限制，由于线程管理，低于限制它的速度很慢。超出限制它会快得多。如果您有兴趣，下面是序列号（升序和降序）

protected static void ASC(int[]a, int left, int right, int div)
{
    int len = 1 + right - left;
    if (len < 27)
    {
        // insertion sort for small array
        int P1 = left + 1;
        int P2 = left;
        while ( P1 <= right )
        {
            div = a[P1];
            while(( P2 >= left )&&( a[P2] > div ))
            {
                a[P2 + 1] = a[P2];
                P2--;
            }
            a[P2 + 1] = div;
            P2 = P1;
            P1++;
        }
        return;
    }
    int third = len / div;
    // "medians"
    int P1 = left + third;
    int P2 = right - third;
    if (P1 <= left)
    {
        P1 = left + 1;
    }
    if (P2 >= right)
    {
        P2 = right - 1;
    }
    int temp;
    if (a[P1] < a[P2])
    {
        temp = a[P1]; a[P1] = a[left]; a[left] = temp;
        temp = a[P2]; a[P2] = a[right]; a[right] = temp;
    }
    else
    {
        temp = a[P1];  a[P1] = a[right];  a[right] = temp;
        temp = a[P2];  a[P2] = a[left];  a[left] = temp;
    }
    // pivots
    int pivot1 = a[left];
    int pivot2 = a[right];
    // pointers
    int less = left + 1;
    int great = right - 1;
    // sorting
    for (int k = less; k <= great; k++)
    {
        if (a[k] < pivot1)
        {
            temp = a[k];  a[k] = a[less];  a[less] = temp;
            less++;
        }
        else if (a[k] > pivot2)
        {
            while (k < great && a[great] > pivot2)
            {
                great--;
            }
            temp = a[k];  a[k] = a[great];  a[great] = temp;
            great--;
            if (a[k] < pivot1)
            {
                temp = a[k];  a[k] = a[less];  a[less] = temp;
                less++;
            }
        }
    }
    int dist = great - less;
    if (dist < 13)
    {
        div++;
    }
    temp = a[less-1];  a[less-1] = a[left];  a[left] = temp;
    temp = a[great+1];  a[great+1] = a[right];  a[right] = temp;
    // subarrays
    ASC(a, left, less - 2, div);
    ASC(a, great + 2, right, div);
    // equal elements
    if (dist > len - 13 && pivot1 != pivot2)
    {
        for (int k = less; k <= great; k++)
        {
            if (a[k] == pivot1)
            {
                temp = a[k];  a[k] = a[less];  a[less] = temp;
                less++;
            }
            else if (a[k] == pivot2)
            {
                temp = a[k];  a[k] = a[great];  a[great] = temp;
                great--;
                if (a[k] == pivot1)
                {
                    temp = a[k];  a[k] = a[less];  a[less] = temp;
                    less++;
                }
            }
        }
    }
    // subarray
    if (pivot1 < pivot2)
    {
        ASC(a, less, great, div);
    }
}

protected static void DSC(int[]a, int left, int right, int div)
{
    int len = 1 + right - left;
    if (len < 27)
    {
        // insertion sort for large array
        int P1 = left + 1;
        int P2 = left;
        while ( P1 <= right )
        {
            div = a[P1];
            while(( P2 >= left )&&( a[P2] < div ))
            {
                a[P2 + 1] = a[P2];
                P2--;
            }
            a[P2 + 1] = div;
            P2 = P1;
            P1++;
        }
        return;
    }
    int third = len / div;
    // "medians"
    int P1 = left + third;
    int P2 = right - third;
    if (P1 >= left)
    {
        P1 = left + 1;
    }
    if (P2 <= right)
    {
        P2 = right - 1;
    }
    int temp;
    if (a[P1] > a[P2])
    {
        temp = a[P1]; a[P1] = a[left]; a[left] = temp;
        temp = a[P2]; a[P2] = a[right]; a[right] = temp;
    }
    else
    {
        temp = a[P1];  a[P1] = a[right];  a[right] = temp;
        temp = a[P2];  a[P2] = a[left];  a[left] = temp;
    }
    // pivots
    int pivot1 = a[left];
    int pivot2 = a[right];
    // pointers
    int less = left + 1;
    int great = right - 1;
    // sorting
    for (int k = less; k <= great; k++)
    {
        if (a[k] > pivot1)
        {
            temp = a[k];  a[k] = a[less];  a[less] = temp;
            less++;
        }
        else if (a[k] < pivot2)
        {
            while (k < great && a[great] < pivot2)
            {
                great--;
            }
            temp = a[k];  a[k] = a[great];  a[great] = temp;
            great--;
            if (a[k] > pivot1)
            {
                temp = a[k];  a[k] = a[less];  a[less] = temp;
                less++;
            }
        }
    }
    int dist = great - less;
    if (dist < 13)
    {
        div++;
    }
    temp = a[less-1];  a[less-1] = a[left];  a[left] = temp;
    temp = a[great+1];  a[great+1] = a[right];  a[right] = temp;
    // subarrays
    DSC(a, left, less - 2, div);
    DSC(a, great + 2, right, div);
    // equal elements
    if (dist > len - 13 && pivot1 != pivot2)
    {
        for (int k = less; k <= great; k++)
        {
            if (a[k] == pivot1)
            {
                temp = a[k];  a[k] = a[less];  a[less] = temp;
                less++;
            }
            else if (a[k] == pivot2)
            {
                temp = a[k];  a[k] = a[great];  a[great] = temp;
                great--;
                if (a[k] == pivot1)
                {
                    temp = a[k];  a[k] = a[less];  a[less] = temp;
                    less++;
                }
            }
        }
    }
    // subarray
    if (pivot1 > pivot2)
    {
        DSC(a, less, great, div);
    }
}