排序10个数字的最快方法？(数字是32位)

Question

我正在解决一个问题,它涉及非常快速地排序10个数字(int32).我的应用程序需要尽可能快地对10个数字进行数百万次排序.我正在对数十亿个元素的数据集进行采样,每次我需要从中挑选10个数字(简化)并对它们进行排序(并从排序的10个元素列表中得出结论).

目前我正在使用插入排序,但我想我可以实现一个非常快速的自定义排序算法,针对10个数字的特定问题,这将超过插入排序.

有没有人知道如何处理这个问题？

Answer 1

(跟进HelloWorld的建议,研究排序网络.)

似乎29比较/交换网络是进行10输入排序的最快方法.我使用了Waksman在1969年发现的网络,用于Javascript中的这个例子,它应该直接转换为C,因为它只是一个if语句,比较和交换的列表.

function sortNet10(data) {	// ten-input sorting network by Waksman, 1969
    var swap;
    if (data[0] > data[5]) { swap = data[0]; data[0] = data[5]; data[5] = swap; }
    if (data[1] > data[6]) { swap = data[1]; data[1] = data[6]; data[6] = swap; }
    if (data[2] > data[7]) { swap = data[2]; data[2] = data[7]; data[7] = swap; }
    if (data[3] > data[8]) { swap = data[3]; data[3] = data[8]; data[8] = swap; }
    if (data[4] > data[9]) { swap = data[4]; data[4] = data[9]; data[9] = swap; }
    if (data[0] > data[3]) { swap = data[0]; data[0] = data[3]; data[3] = swap; }
    if (data[5] > data[8]) { swap = data[5]; data[5] = data[8]; data[8] = swap; }
    if (data[1] > data[4]) { swap = data[1]; data[1] = data[4]; data[4] = swap; }
    if (data[6] > data[9]) { swap = data[6]; data[6] = data[9]; data[9] = swap; }
    if (data[0] > data[2]) { swap = data[0]; data[0] = data[2]; data[2] = swap; }
    if (data[3] > data[6]) { swap = data[3]; data[3] = data[6]; data[6] = swap; }
    if (data[7] > data[9]) { swap = data[7]; data[7] = data[9]; data[9] = swap; }
    if (data[0] > data[1]) { swap = data[0]; data[0] = data[1]; data[1] = swap; }
    if (data[2] > data[4]) { swap = data[2]; data[2] = data[4]; data[4] = swap; }
    if (data[5] > data[7]) { swap = data[5]; data[5] = data[7]; data[7] = swap; }
    if (data[8] > data[9]) { swap = data[8]; data[8] = data[9]; data[9] = swap; }
    if (data[1] > data[2]) { swap = data[1]; data[1] = data[2]; data[2] = swap; }
    if (data[3] > data[5]) { swap = data[3]; data[3] = data[5]; data[5] = swap; }
    if (data[4] > data[6]) { swap = data[4]; data[4] = data[6]; data[6] = swap; }
    if (data[7] > data[8]) { swap = data[7]; data[7] = data[8]; data[8] = swap; }
    if (data[1] > data[3]) { swap = data[1]; data[1] = data[3]; data[3] = swap; }
    if (data[4] > data[7]) { swap = data[4]; data[4] = data[7]; data[7] = swap; }
    if (data[2] > data[5]) { swap = data[2]; data[2] = data[5]; data[5] = swap; }
    if (data[6] > data[8]) { swap = data[6]; data[6] = data[8]; data[8] = swap; }
    if (data[2] > data[3]) { swap = data[2]; data[2] = data[3]; data[3] = swap; }
    if (data[4] > data[5]) { swap = data[4]; data[4] = data[5]; data[5] = swap; }
    if (data[6] > data[7]) { swap = data[6]; data[6] = data[7]; data[7] = swap; }
    if (data[3] > data[4]) { swap = data[3]; data[3] = data[4]; data[4] = swap; }
    if (data[5] > data[6]) { swap = data[5]; data[5] = data[6]; data[6] = swap; }
    return(data);
}

alert(sortNet10([5,7,1,8,4,3,6,9,2,0]));

这是网络的图形表示,分为独立的阶段.

为了利用并行处理,可以将5-4-3-4-4-4-3-2分组改为4-4-4-4-4-4-3-2分组.

Answer 2

处理此固定大小时,请查看排序网络.这些算法具有固定的运行时间,并且与其输入无关.对于您的用例,您没有某些排序算法所具有的开销.

Bitonic排序是这种网络的实现.这个最适合在CPU上使用len(n)<= 32.在更大的输入上,您可以考虑转移到GPU. https://en.wikipedia.org/wiki/Sorting_network

顺便说一句,这是一个比较排序算法的好页面(尽管它缺少了bitonic sort.

http://www.sorting-algorithms.com

Answer 3

使用具有4个组的比较的排序网络,因此您可以在SIMD寄存器中执行此操作.一对压缩的最小/最大指令实现了压缩比较器功能.抱歉,我现在没有时间寻找一个我记得看过的页面,但希望在SIMD或SSE排序网络上搜索会有所改变.

对于四个32位整数的向量,x86 SSE确实具有打包的32位整数最小和最大指令.AVX2(Haswell和更高版本)具有相同的但是对于具有8个整数的256b向量.还有有效的随机指示.

如果你有很多独立的小分类,可能可以使用向量并行进行4或8种分类.ESP.如果你随机选择元素(所以要分类的数据无论如何都不会在内存中连续),你可以避免随机播放,只需按照你需要的顺序进行比较.10个寄存器用于保存4个(AVX2:8)10个整数列表中的所有数据,但仍有6个reg用于暂存空间.

如果您还需要对关联数据进行排序,则矢量排序网络效率较低.在这种情况下,最有效的方法似乎是使用打包比较来获取哪些元素被更改的掩码,并使用该掩码来混合(引用)关联数据的向量.

Answer 4

那么展开的无分支选择排序怎么样？

#include <iostream>
#include <algorithm>
#include <random>

//return the index of the minimum element in array a
int min(const int * const a) {
  int m = a[0];
  int indx = 0;
  #define TEST(i) (m > a[i]) && (m = a[i], indx = i ); 
  //see http://stackoverflow.com/a/7074042/2140449
  TEST(1);
  TEST(2);
  TEST(3);
  TEST(4);
  TEST(5);
  TEST(6);
  TEST(7);
  TEST(8);
  TEST(9);
  #undef TEST
  return indx;
}

void sort( int * const a ){
  int work[10];
  int indx;
  #define GET(i) indx = min(a); work[i] = a[indx]; a[indx] = 2147483647; 
  //get the minimum, copy it to work and set it at max_int in a
  GET(0);
  GET(1);
  GET(2);
  GET(3);
  GET(4);
  GET(5);
  GET(6);
  GET(7);
  GET(8);
  GET(9);
  #undef GET
  #define COPY(i) a[i] = work[i];
  //copy back to a
  COPY(0);
  COPY(1);
  COPY(2);
  COPY(3);
  COPY(4);
  COPY(5);
  COPY(6);
  COPY(7);
  COPY(8);
  COPY(9);
  #undef COPY
}

int main() {
  //generating and printing a random array
  int a[10] = { 1,2,3,4,5,6,7,8,9,10 };
  std::random_device rd;
  std::mt19937 g(rd());
  std::shuffle( a, a+10, g);
  for (int i = 0; i < 10; i++) {
    std::cout << a[i] << ' ';
  }
  std::cout << std::endl;

  //sorting and printing again
  sort(a);
  for (int i = 0; i < 10; i++) {
    std::cout << a[i] << ' ';
  } 

  return 0;
}

http://coliru.stacked-crooked.com/a/71e18bc4f7fa18c6

唯一相关的线是前两个#define.

它使用两个列表并完全重新检查第一个十次,这将是一个严格实现的选择排序,但它避免了分支和可变长度循环,这可以补偿现代处理器和如此小的数据集.

基准

我对排序网络进行了基准测试,我的代码似乎更慢.但是我试图删除展开和副本.运行此代码:

#include <iostream>
#include <algorithm>
#include <random>
#include <chrono>

int min(const int * const a, int i) {
  int m = a[i];
  int indx = i++;
  for ( ; i<10; i++) 
    //see http://stackoverflow.com/a/7074042/2140449
    (m > a[i]) && (m = a[i], indx = i ); 
  return indx;
}

void sort( int * const a ){
  for (int i = 0; i<9; i++)
    std::swap(a[i], a[min(a,i)]); //search only forward
}


void sortNet10(int * const data) {  // ten-input sorting network by Waksman, 1969
    int swap;
    if (data[0] > data[5]) { swap = data[0]; data[0] = data[5]; data[5] = swap; }
    if (data[1] > data[6]) { swap = data[1]; data[1] = data[6]; data[6] = swap; }
    if (data[2] > data[7]) { swap = data[2]; data[2] = data[7]; data[7] = swap; }
    if (data[3] > data[8]) { swap = data[3]; data[3] = data[8]; data[8] = swap; }
    if (data[4] > data[9]) { swap = data[4]; data[4] = data[9]; data[9] = swap; }
    if (data[0] > data[3]) { swap = data[0]; data[0] = data[3]; data[3] = swap; }
    if (data[5] > data[8]) { swap = data[5]; data[5] = data[8]; data[8] = swap; }
    if (data[1] > data[4]) { swap = data[1]; data[1] = data[4]; data[4] = swap; }
    if (data[6] > data[9]) { swap = data[6]; data[6] = data[9]; data[9] = swap; }
    if (data[0] > data[2]) { swap = data[0]; data[0] = data[2]; data[2] = swap; }
    if (data[3] > data[6]) { swap = data[3]; data[3] = data[6]; data[6] = swap; }
    if (data[7] > data[9]) { swap = data[7]; data[7] = data[9]; data[9] = swap; }
    if (data[0] > data[1]) { swap = data[0]; data[0] = data[1]; data[1] = swap; }
    if (data[2] > data[4]) { swap = data[2]; data[2] = data[4]; data[4] = swap; }
    if (data[5] > data[7]) { swap = data[5]; data[5] = data[7]; data[7] = swap; }
    if (data[8] > data[9]) { swap = data[8]; data[8] = data[9]; data[9] = swap; }
    if (data[1] > data[2]) { swap = data[1]; data[1] = data[2]; data[2] = swap; }
    if (data[3] > data[5]) { swap = data[3]; data[3] = data[5]; data[5] = swap; }
    if (data[4] > data[6]) { swap = data[4]; data[4] = data[6]; data[6] = swap; }
    if (data[7] > data[8]) { swap = data[7]; data[7] = data[8]; data[8] = swap; }
    if (data[1] > data[3]) { swap = data[1]; data[1] = data[3]; data[3] = swap; }
    if (data[4] > data[7]) { swap = data[4]; data[4] = data[7]; data[7] = swap; }
    if (data[2] > data[5]) { swap = data[2]; data[2] = data[5]; data[5] = swap; }
    if (data[6] > data[8]) { swap = data[6]; data[6] = data[8]; data[8] = swap; }
    if (data[2] > data[3]) { swap = data[2]; data[2] = data[3]; data[3] = swap; }
    if (data[4] > data[5]) { swap = data[4]; data[4] = data[5]; data[5] = swap; }
    if (data[6] > data[7]) { swap = data[6]; data[6] = data[7]; data[7] = swap; }
    if (data[3] > data[4]) { swap = data[3]; data[3] = data[4]; data[4] = swap; }
    if (data[5] > data[6]) { swap = data[5]; data[5] = data[6]; data[6] = swap; }
}


std::chrono::duration<double> benchmark( void(*func)(int * const), const int seed ) {
  std::mt19937 g(seed);
  int a[10] = {10,11,12,13,14,15,16,17,18,19};
  std::chrono::high_resolution_clock::time_point t1, t2; 
  t1 = std::chrono::high_resolution_clock::now();
  for (long i = 0; i < 1e7; i++) {
    std::shuffle( a, a+10, g);
    func(a);
  }
  t2 = std::chrono::high_resolution_clock::now();
  return std::chrono::duration_cast<std::chrono::duration<double>>(t2 - t1);
}

int main() {
  std::random_device rd;
  for (int i = 0; i < 10; i++) {
    const int seed = rd();
    std::cout << "seed = " << seed << std::endl;
    std::cout << "sortNet10: " << benchmark(sortNet10, seed).count() << std::endl;
    std::cout << "sort:      " << benchmark(sort,      seed).count() << std::endl;
  }
  return 0;
}

与分拣网络相比,我一直在为无分支选择排序获得更好的结果.

$ gcc -v
gcc version 5.2.0 (GCC) 
$ g++ -std=c++11 -Ofast sort.cpp && ./a.out
seed = -1727396418
sortNet10: 2.24137
sort:      2.21828
seed = 2003959850
sortNet10: 2.23914
sort:      2.21641
seed = 1994540383
sortNet10: 2.23782
sort:      2.21778
seed = 1258259982
sortNet10: 2.25199
sort:      2.21801
seed = 1821086932
sortNet10: 2.25535
sort:      2.2173
seed = 412262735
sortNet10: 2.24489
sort:      2.21776
seed = 1059795817
sortNet10: 2.29226
sort:      2.21777
seed = -188551272
sortNet10: 2.23803
sort:      2.22996
seed = 1043757247
sortNet10: 2.2503
sort:      2.23604
seed = -268332483
sortNet10: 2.24455
sort:      2.24304

Answer 5

问题并不是说这是某种基于Web的应用程序.引起我注意的一件事是:

我正在对数十亿个元素的数据集进行采样,每次我需要从中挑选10个数字(简化)并对它们进行排序(并从排序的10个元素列表中得出结论).

作为一名软件和硬件工程师,这绝对让我感到尖叫"FPGA".我不知道你需要什么样的结论,从排序的一组数字或其中的数据来自画,但我知道这将是几乎微不足道的介于处理亿和十亿这些"排序和-的分析" 每秒操作次数.我以前做过FPGA辅助的DNA测序工作.当问题非常适合这种类型的解决方案时,几乎不可能击败FPGA的大量处理能力.

在某种程度上,唯一的限制因素是您可以多快地将数据转移到FPGA中,以及您能够以多快的速度获取数据.

作为参考,我设计了一个高性能的实时图像处理器,以每秒约3亿像素的速率接收32位RGB图像数据.在从另一端出来之前,数据通过FIR滤波器,矩阵乘法器,查找表,空间边缘检测块和许多其他操作进行流式传输.所有这一切都在一个相对较小的Xilinx Virtex2 FPGA上,内部时钟范围从大约33MHz到如果我没记错,400MHz.哦,是的,它还有DDR2控制器实现并运行了两组DDR2内存.

FPGA可以在每个时钟转换时输出一个10位32位数,同时工作在数百MHz.当数据填满处理管道时,在操作开始时会有短暂的延迟.之后,您应该能够每个时钟获得一个结果.或者更多,如果可以通过复制排序和分析管道来并行化处理.原则上,解决方案几乎是微不足道的.

重点是:如果应用程序不受PC限制,并且数据流和处理与FPGA解决方案"兼容"(独立或作为机器中的协处理器卡),那么你无法继续无论算法如何,用任何语言编写的软件都能够达到可达到的性能水平.

编辑:

只是快速搜索并找到了一篇可能对你有用的论文.看起来它可以追溯到2012年.今天你可以做得更好(甚至当时).这里是:

在FPGA上对网络进行排序

Answer 6

我最近写了一个小类,它使用Bose-Nelson算法在编译时生成一个排序网络.

它可用于为10个数字创建一个非常快速的排序.

/**
 * A Functor class to create a sort for fixed sized arrays/containers with a
 * compile time generated Bose-Nelson sorting network.
 * \tparam NumElements  The number of elements in the array or container to sort.
 * \tparam T            The element type.
 * \tparam Compare      A comparator functor class that returns true if lhs < rhs.
 */
template <unsigned NumElements, class Compare = void> class StaticSort
{
    template <class A, class C> struct Swap
    {
        template <class T> inline void s(T &v0, T &v1)
        {
            T t = Compare()(v0, v1) ? v0 : v1; // Min
            v1 = Compare()(v0, v1) ? v1 : v0; // Max
            v0 = t;
        }

        inline Swap(A &a, const int &i0, const int &i1) { s(a[i0], a[i1]); }
    };

    template <class A> struct Swap <A, void>
    {
        template <class T> inline void s(T &v0, T &v1)
        {
            // Explicitly code out the Min and Max to nudge the compiler
            // to generate branchless code.
            T t = v0 < v1 ? v0 : v1; // Min
            v1 = v0 < v1 ? v1 : v0; // Max
            v0 = t;
        }

        inline Swap(A &a, const int &i0, const int &i1) { s(a[i0], a[i1]); }
    };

    template <class A, class C, int I, int J, int X, int Y> struct PB
    {
        inline PB(A &a)
        {
            enum { L = X >> 1, M = (X & 1 ? Y : Y + 1) >> 1, IAddL = I + L, XSubL = X - L };
            PB<A, C, I, J, L, M> p0(a);
            PB<A, C, IAddL, J + M, XSubL, Y - M> p1(a);
            PB<A, C, IAddL, J, XSubL, M> p2(a);
        }
    };

    template <class A, class C, int I, int J> struct PB <A, C, I, J, 1, 1>
    {
        inline PB(A &a) { Swap<A, C> s(a, I - 1, J - 1); }
    };

    template <class A, class C, int I, int J> struct PB <A, C, I, J, 1, 2>
    {
        inline PB(A &a) { Swap<A, C> s0(a, I - 1, J); Swap<A, C> s1(a, I - 1, J - 1); }
    };

    template <class A, class C, int I, int J> struct PB <A, C, I, J, 2, 1>
    {
        inline PB(A &a) { Swap<A, C> s0(a, I - 1, J - 1); Swap<A, C> s1(a, I, J - 1); }
    };

    template <class A, class C, int I, int M, bool Stop = false> struct PS
    {
        inline PS(A &a)
        {
            enum { L = M >> 1, IAddL = I + L, MSubL = M - L};
            PS<A, C, I, L, (L <= 1)> ps0(a);
            PS<A, C, IAddL, MSubL, (MSubL <= 1)> ps1(a);
            PB<A, C, I, IAddL, L, MSubL> pb(a);
        }
    };

    template <class A, class C, int I, int M> struct PS <A, C, I, M, true>
    {
        inline PS(A &a) {}
    };

public:
    /**
     * Sorts the array/container arr.
     * \param  arr  The array/container to be sorted.
     */
    template <class Container> inline void operator() (Container &arr) const
    {
        PS<Container, Compare, 1, NumElements, (NumElements <= 1)> ps(arr);
    };

    /**
     * Sorts the array arr.
     * \param  arr  The array to be sorted.
     */
    template <class T> inline void operator() (T *arr) const
    {
        PS<T*, Compare, 1, NumElements, (NumElements <= 1)> ps(arr);
    };
};

#include <iostream>
#include <vector>

int main(int argc, const char * argv[])
{
    enum { NumValues = 10 };

    // Arrays
    {
        int rands[NumValues];
        for (int i = 0; i < NumValues; ++i) rands[i] = rand() % 100;
        std::cout << "Before Sort: \t";
        for (int i = 0; i < NumValues; ++i) std::cout << rands[i] << " ";
        std::cout << "\n";
        StaticSort<NumValues> staticSort;
        staticSort(rands);
        std::cout << "After Sort: \t";
        for (int i = 0; i < NumValues; ++i) std::cout << rands[i] << " ";
        std::cout << "\n";
    }

    std::cout << "\n";

    // STL Vector
    {
        std::vector<int> rands(NumValues);
        for (int i = 0; i < NumValues; ++i) rands[i] = rand() % 100;
        std::cout << "Before Sort: \t";
        for (int i = 0; i < NumValues; ++i) std::cout << rands[i] << " ";
        std::cout << "\n";
        StaticSort<NumValues> staticSort;
        staticSort(rands);
        std::cout << "After Sort: \t";
        for (int i = 0; i < NumValues; ++i) std::cout << rands[i] << " ";
        std::cout << "\n";
    }

    return 0;
}

请注意if (compare) swap,我们为min和max明确编写了三元运算符,而不是语句.这是为了帮助推动编译器使用无分支代码.

基准

下面的基准测试是用clang -O3编写的,并在我2012年中期的macbook air上运行.

对随机数据进行排序

将它与DarioP的代码进行比较,下面是对100万个大小为10的32位int数组进行排序所需的毫秒数:

Hardcoded Sort Net 10: 88.774 ms
模板化Bose-Nelson排序10: 27.815 ms

使用这种模板化方法,我们还可以在编译时为其他数量的元素生成排序网络.

分类100万个不同大小的数组的时间(以毫秒为单位).
大小为2,4,8的数组的毫秒数分别为1.943,8.655,20.246.

Glenn Teitelbaum对展开的插入排序的信用.

以下是6个元素的小数组的每种平均时钟数.基准代码和示例可以在这个问题上找到:
最快的固定长度6 int数组

Direct call to qsort library function       : 326.81
Naive implementation (insertion sort)       : 132.98
Insertion Sort (Daniel Stutzbach)           : 104.04
Insertion Sort Unrolled                     : 99.64
Insertion Sort Unrolled (Glenn Teitelbaum)  : 81.55
Rank Order                                  : 44.01
Rank Order with registers                   : 42.40
Sorting Networks (Daniel Stutzbach)         : 88.06
Sorting Networks (Paul R)                   : 31.64
Sorting Networks 12 with Fast Swap          : 29.68
Sorting Networks 12 reordered Swap          : 28.61
Reordered Sorting Network w/ fast swap      : 24.63
Templated Sorting Network (this class)      : 25.37

它的性能与6个元素的问题中最快的例子一样快.

排序分类数据的性能

通常,输入数组可能已经排序或大部分排序.
在这种情况下,插入排序可能是更好的选择.

您可能希望根据数据选择合适的排序算法.

用于基准测试的代码可以在这里找到.

Answer 7

虽然网络排序在小数组上具有很快的快速几率,但如果经过适当优化,有时您无法击败插入排序.例如,具有2个元素的批量插入:

{
    final int a=in[0]<in[1]?in[0]:in[1];
    final int b=in[0]<in[1]?in[1]:in[0];
    in[0]=a;
    in[1]=b;
}
for(int x=2;x<10;x+=2)
{
    final int a=in[x]<in[x+1]?in[x]:in[x+1];
    final int b=in[x]<in[x+1]?in[x+1]:in[x];
    int y= x-1;

    while(y>=0&&in[y]>b)
    {
        in[y+2]= in[y];
        --y;
    }
    in[y+2]=b;
    while(y>=0&&in[y]>a)
    {
        in[y+1]= in[y];
        --y;
    }
    in[y+1]=a;
}