我正在解决一个问题，它涉及到快速排序10个数字(int32)。我的应用程序需要以尽可能快的速度对10个数字进行百万次排序。我正在对一个包含数十亿个元素的数据集进行采样，每次我都需要从中挑选10个数字(简化)并对它们进行排序(并从排序的10个元素列表中得出结论)。
目前我正在使用插入排序，但我想我可以为我的10个数字的特定问题实现一个非常快速的自定义排序算法，这将击败插入排序。
我该如何处理这个问题呢?
当前回答
插入排序平均需要29,6次与排序10个输入的比较，最佳情况为9，最差情况为45(给定输入的顺序是相反的)。
{9,6,1} shellsort平均需要25.5次比较来排序10个输入。最好的情况是14次比较，最坏的情况是34次，对反向输入排序需要22次。
因此，使用shellsort而不是插入排序可以减少14%的平均情况。尽管最佳情况增加了56%，但最坏情况减少了24%，这对于控制最坏情况性能非常重要的应用程序来说意义重大。反之则减少51%。
因为你似乎对插入排序很熟悉，你可以将算法实现为{9,6}的排序网络，然后在那之后附加插入排序({1}):
i[0] with i[9] // {9}
i[0] with i[6] // {6}
i[1] with i[7] // {6}
i[2] with i[8] // {6}
i[3] with i[9] // {6}
i[0 ... 9] // insertion sort
其他回答
以下是运行在10个CUDA线程上的CUDA内核(秩排序算法)，在42毫秒内对1000个数组排序1000次，每次排序42纳秒或每次排序~70个周期(1.7 GHz):
inline
__device__ int findOrder(const int mask, const int data0)
{
const int order1 = data0>__shfl_sync(mask,data0,0);
const int order2 = data0>__shfl_sync(mask,data0,1);
const int order3 = data0>__shfl_sync(mask,data0,2);
const int order4 = data0>__shfl_sync(mask,data0,3);
const int order5 = data0>__shfl_sync(mask,data0,4);
const int order6 = data0>__shfl_sync(mask,data0,5);
const int order7 = data0>__shfl_sync(mask,data0,6);
const int order8 = data0>__shfl_sync(mask,data0,7);
const int order9 = data0>__shfl_sync(mask,data0,8);
const int order10 = data0>__shfl_sync(mask,data0,9);
return order1 + order2 + order3 + order4 + order5 + order6 + order7 + order8 + order9 + order10;
}
// launch this with 10 CUDA threads in 1 block in 1 grid
// sorts 10 elements using only SIMT registers
__global__ void rankSort(int * __restrict__ buffer)
{
const int id = threadIdx.x;
// enable first 10 lanes of warp for shuffling
const int mask = __activemask();
__shared__ int data[10000];
for(int i=0;i<1000;i++)
{
data[id+i*10] = buffer[id+i*10];
}
__syncwarp();
// benchmark 1000 times
for(int k=0;k<1000;k++)
{
// sort 1000 arrays
for(int j=0;j<1000;j+=5)
{
// sort 5 arrays at once to hide latency
const int data1 = data[id+j*10];
const int data2 = data[id+(j+1)*10];
const int data3 = data[id+(j+2)*10];
const int data4 = data[id+(j+3)*10];
const int data5 = data[id+(j+4)*10];
const int order1 = findOrder(mask,data1);
const int order2 = findOrder(mask,data2);
const int order3 = findOrder(mask,data3);
const int order4 = findOrder(mask,data4);
const int order5 = findOrder(mask,data5);
data[order1+j*10]=data1;
data[order2+(j+1)*10]=data2;
data[order3+(j+2)*10]=data3;
data[order4+(j+3)*10]=data4;
data[order5+(j+4)*10]=data5;
}
}
__syncwarp();
for(int i=0;i<1000;i++)
{
buffer[id+i*10] = data[id+i*10];
}
}
由于所有10个线程都在同一个SIMT单元上处理，它类似于运行在矢量寄存器上的AVX512优化版本，但除了更多的寄存器空间以隐藏更多的延迟之外。此外，SIMT单元是32宽的，因此它可以运行线性时间复杂度直到32个元素。
该算法假设元素是唯一的。对于重复的数据，需要一个额外的缩减步骤来将重复的顺序值解压缩为10个元素。首先，它计算每个元素的秩，然后直接将它们复制到数组中作为索引的秩。顺序值需要蛮力(O(N x N))次比较，为了减少延迟，使用warp-shuffles从(向量寄存器的)不同的warp-lanes收集数据。
尽管网络排序在小数组上有很好的快速几率，但如果适当优化，有时您无法击败插入排序。例如，有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;
}
(根据@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); } document.write(sortNet10([5,7,1,8,4,3,6,9,2,0]));
这里是网络的图形表示，分为独立的阶段。
为了利用并行处理的优势，可以将5-4-3-3 - 4-4-2 -3-2分组改为4-4-4-2 -4-4-3-2分组。
出于类似于我在这里描述的原因，以下排序函数sort6_iterator()和sort10_iterator_local()应该能很好地执行，其中排序网络是从这里取的:
template<class IterType>
inline void sort10_iterator(IterType it)
{
#define SORT2(x,y) {if(data##x>data##y)std::swap(data##x,data##y);}
#define DD1(a) auto data##a=*(data+a);
#define DD2(a,b) auto data##a=*(data+a), data##b=*(data+b);
#define CB1(a) *(data+a)=data##a;
#define CB2(a,b) *(data+a)=data##a;*(data+b)=data##b;
DD2(1,4) SORT2(1,4) DD2(7,8) SORT2(7,8) DD2(2,3) SORT2(2,3) DD2(5,6) SORT2(5,6) DD2(0,9) SORT2(0,9)
SORT2(2,5) SORT2(0,7) SORT2(8,9) SORT2(3,6)
SORT2(4,9) SORT2(0,1)
SORT2(0,2) CB1(0) SORT2(6,9) CB1(9) SORT2(3,5) SORT2(4,7) SORT2(1,8)
SORT2(3,4) SORT2(5,8) SORT2(6,7) SORT2(1,2)
SORT2(7,8) CB1(8) SORT2(1,3) CB1(1) SORT2(2,5) SORT2(4,6)
SORT2(2,3) CB1(2) SORT2(6,7) CB1(7) SORT2(4,5)
SORT2(3,4) CB2(3,4) SORT2(5,6) CB2(5,6)
#undef CB1
#undef CB2
#undef DD1
#undef DD2
#undef SORT2
}
为了调用这个函数，我给它传递了一个std::vector迭代器。
这个问题并没有说这是某种基于web的应用程序。有一件事引起了我的注意:
我正在对一个包含数十亿个元素的数据集进行采样，每次我都需要从中挑选10个数字(简化)并对它们进行排序(并从排序的10个元素列表中得出结论)。
As a software and hardware engineer this absolutely screams FPGA to me. I don't know what kind of conclusions you need to draw from the sorted set of numbers or where the data comes from, but I know it would be almost trivial to process somewhere between one hundred million and a billion of these "sort-and-analyze" operations per second. I've done FPGA-assisted DNA sequencing work in the past. It is nearly impossible to beat the massive processing power of FPGAs when the problem is well suited for that type of a solution.
在某种程度上，唯一的限制因素是将数据铲入FPGA的速度有多快，以及取出数据的速度有多快。
As a point of reference, I designed a high performance real-time image processor that received 32 bit RGB image data at a rate of about 300 million pixels per second. The data streamed through FIR filters, matrix multipliers, lookup tables, spatial edge detection blocks and a number of other operations before coming out the other end. All of this on a relatively small Xilinx Virtex2 FPGA with internal clocking spanning from about 33 MHz to, if I remember correctly, 400 MHz. Oh, yes, it also had a DDR2 controller implementation and ran two banks of DDR2 memory.
当工作在数百MHz时，FPGA可以在每次时钟转换中输出10个32位数字。当数据填满处理管道时，操作开始时会有短暂的延迟。在此之后，您应该能够在每个时钟获得一个结果。如果可以通过复制排序和分析管道使处理并行化，则会更多。原则上，解决方案几乎是微不足道的。
关键在于:如果应用程序不是pc绑定的，并且数据流和处理与FPGA解决方案“兼容”(无论是独立的还是作为机器中的协处理器卡)，那么无论使用哪种算法，用任何语言编写的软件都无法击败可达到的性能水平。
我只是快速搜索了一下，找到了一篇可能对你有用的论文。看起来要追溯到2012年。在今天(甚至在过去)，您可以在性能上做得更好。下面就是:
fpga上的排序网络
