这是 Alexandru 的基数排序与 Zjarek 的螺纹智能旋转的混合。编译它
g++ -std=c++0x -pthread -O3 -march=native sorter_gaussian_radix.cxx -o sorter_gaussian_radix
您可以通过定义 STEP 来更改基数大小（例如添加 -DSTEP=11）。我发现最适合我的笔记本电脑的是 8（默认）。
默认情况下，它将问题分成 4 部分并在多个线程上运行。您可以通过将深度参数传递给命令行来更改它。所以如果你有两个核心，运行它
sorter_gaussian_radix 50000000 1
如果你有 16 个核心
sorter_gaussian_radix 50000000 4
现在的最大深度是 6（64 个线程）。如果你放太多关卡，你只会拖慢代码。
我还尝试过来自英特尔性能基元 (IPP) 库的基数排序。 Alexandru 的实现完全击败了 IPP，IPP 慢了大约 30%。该变体也包含在此处（已注释掉）。
#include <stdlib.h>
#include <string.h>
#include <algorithm>
#include <ctime>
#include <iostream>
#include <thread>
#include <vector>
#include <boost/cstdint.hpp>
// #include "ipps.h"
#ifndef STEP
#define STEP 8
#endif
const int step = STEP;
const int start_step=24;
const int num_steps=(64-start_step+step-1)/step;
int size;
double *dbuf, *copy;
clock_t c1, c2, c3, c4, c5;
const double distrib[]={-2.15387,
-1.86273,
-1.67594,
-1.53412,
-1.4178,
-1.31801,
-1.22986,
-1.15035,
-1.07752,
-1.00999,
-0.946782,
-0.887147,
-0.830511,
-0.776422,
-0.724514,
-0.67449,
-0.626099,
-0.579132,
-0.53341,
-0.488776,
-0.445096,
-0.40225,
-0.36013,
-0.318639,
-0.27769,
-0.237202,
-0.197099,
-0.157311,
-0.11777,
-0.0784124,
-0.0391761,
0,
0.0391761,
0.0784124,
0.11777,
0.157311,
0.197099,
0.237202,
0.27769,
0.318639,
0.36013,
0.40225,
0.445097,
0.488776,
0.53341,
0.579132,
0.626099,
0.67449,
0.724514,
0.776422,
0.830511,
0.887147,
0.946782,
1.00999,
1.07752,
1.15035,
1.22986,
1.31801,
1.4178,
1.53412,
1.67594,
1.86273,
2.15387};
class Distrib
{
const int value;
public:
Distrib(const double &v): value(v) {}
bool operator()(double a)
{
return a<value;
}
};
void recursive_sort(const int start, const int end,
const int index, const int offset,
const int depth, const int max_depth)
{
if(depth<max_depth)
{
Distrib dist(distrib[index]);
const int middle=std::partition(dbuf+start,dbuf+end,dist) - dbuf;
// const int middle=
// std::partition(dbuf+start,dbuf+end,[&](double a)
// {return a<distrib[index];})
// - dbuf;
std::thread lower(recursive_sort,start,middle,index-offset,offset/2,
depth+1,max_depth);
std::thread upper(recursive_sort,middle,end,index+offset,offset/2,
depth+1,max_depth);
lower.join(), upper.join();
}
else
{
// ippsSortRadixAscend_64f_I(dbuf+start,copy+start,end-start);
c1=clock();
double *dbuf_local(dbuf), *copy_local(copy);
boost::uint64_t mask = (1 << step) - 1;
int cnt[num_steps][mask+1];
boost::uint64_t *ibuf = reinterpret_cast<boost::uint64_t *> (dbuf_local);
for(int i=0;i<num_steps;++i)
for(uint j=0;j<mask+1;++j)
cnt[i][j]=0;
for (int i = start; i < end; i++)
{
for (int w = start_step, v = 0; w < 64; w += step, v++)
{
int p = (~ibuf[i] >> w) & mask;
(cnt[v][p])++;
}
}
c2=clock();
std::vector<int> sum(num_steps,0);
for (uint i = 0; i <= mask; i++)
{
for (int w = start_step, v = 0; w < 64; w += step, v++)
{
int tmp = sum[v] + cnt[v][i];
cnt[v][i] = sum[v];
sum[v] = tmp;
}
}
c3=clock();
for (int w = start_step, v = 0; w < 64; w += step, v++)
{
ibuf = reinterpret_cast<boost::uint64_t *>(dbuf_local);
for (int i = start; i < end; i++)
{
int p = (~ibuf[i] >> w) & mask;
copy_local[start+((cnt[v][p])++)] = dbuf_local[i];
}
std::swap(copy_local,dbuf_local);
}
// Do the last set of reversals
for (int p = start; p < end; p++)
if (dbuf_local[p] >= 0.)
{
std::reverse(dbuf_local+p, dbuf_local + end);
break;
}
c4=clock();
// Insertion sort
for (int i = start+1; i < end; i++) {
double value = dbuf_local[i];
if (value < dbuf_local[i - 1]) {
dbuf_local[i] = dbuf_local[i - 1];
int p = i - 1;
for (; p > 0 && value < dbuf_local[p - 1]; p--)
dbuf_local[p] = dbuf_local[p - 1];
dbuf_local[p] = value;
}
}
c5=clock();
}
}
int main(int argc, char **argv) {
size = atoi(argv[1]);
copy = new double[size];
dbuf = new double[size];
FILE *f = fopen("gaussian.dat", "r");
fread(dbuf, size, sizeof(double), f);
fclose(f);
clock_t c0 = clock();
const int max_depth= (argc > 2) ? atoi(argv[2]) : 2;
// ippsSortRadixAscend_64f_I(dbuf,copy,size);
recursive_sort(0,size,31,16,0,max_depth);
if(num_steps%2==1)
std::swap(dbuf,copy);
// for (int i=0; i<size-1; i++){
// if (dbuf[i]>dbuf[i+1])
// std::cout << "BAD "
// << i << " "
// << dbuf[i] << " "
// << dbuf[i+1] << " "
// << "\n";
// }
std::cout << "Finished after "
<< (double) (c1 - c0) / CLOCKS_PER_SEC << " "
<< (double) (c2 - c1) / CLOCKS_PER_SEC << " "
<< (double) (c3 - c2) / CLOCKS_PER_SEC << " "
<< (double) (c4 - c3) / CLOCKS_PER_SEC << " "
<< (double) (c5 - c4) / CLOCKS_PER_SEC << " "
<< "\n";
// delete [] dbuf;
// delete [] copy;
return 0;
}
编辑：我实现了 Alexandru 的缓存改进，这在我的机器上节省了大约 30% 的时间。
编辑：这实现了递归排序，因此它应该在 Alexandru 的 16 核机器上运行良好。它还使用了 Alexandru 的最后一项改进，并移除了其中一项相反的改进。对我来说，这提高了 20%。
编辑：修复了一个导致超过 2 个核心时效率低下的标志错误。
编辑：删除了 lambda，因此它可以与旧版本的 gcc 一起编译。它包括注释掉的 IPP 代码变体。我还修复了在 16 核上运行的文档。据我所知，这是最快的实现。
编辑：修复了 STEP 不是 8 时的错误。将最大线程数增加到 64。添加了一些计时信息。