为什么排序组的分组求和比未排序组慢？
我有2列制表符分隔的整数，其中第一列是随机整数，第二列是标识组的整数，可以由此程序生成。（generate_groups）
#include <cstdlib>
#include <iostream>
#include <ctime>
int main(int argc, char* argv[]) {
int num_values = atoi(argv[1]);
int num_groups = atoi(argv[2]);
int group_size = num_values / num_groups;
int group = -1;
std::srand(42);
for (int i = 0; i < num_values; ++i) {
if (i % group_size == 0) {
++group;
}
std::cout << std::rand() << '\t' << group << '\n';
}
return 0;
}
然后，我使用第二个程序（sum_groups.cc）计算每个组的和。
#include <iostream>
#include <chrono>
#include <vector>
// This is the function whose performance I am interested in
void grouped_sum(int* p_x, int *p_g, int n, int* p_out) {
for (size_t i = 0; i < n; ++i) {
p_out[p_g[i]] += p_x[i];
}
}
int main() {
std::vector<int> values;
std::vector<int> groups;
std::vector<int> sums;
int n_groups = 0;
// Read in the values and calculate the max number of groups
while(std::cin) {
int value, group;
std::cin >> value >> group;
values.push_back(value);
groups.push_back(group);
if (group > n_groups) {
n_groups = group;
}
}
sums.resize(n_groups);
// Time grouped sums
std::chrono::system_clock::time_point start = std::chrono::system_clock::now();
for (int i = 0; i < 10; ++i) {
grouped_sum(values.data(), groups.data(), values.size(), sums.data());
}
std::chrono::system_clock::time_point end = std::chrono::system_clock::now();
std::cout << (end - start).count() << std::endl;
return 0;
}
如果我在给定大小的数据集上运行这些程序，然后打乱相同数据集的行的顺序，打乱的数据计算总和的速度比有序数据快2倍或更多。
g++ -O3 generate_groups.cc -o generate_groups
g++ -O3 sum_groups.cc -o sum_groups
generate_groups 1000000 100 > groups
shuf groups > groups2
sum_groups < groups
sum_groups < groups2
sum_groups < groups2
sum_groups < groups
20784
8854
8220
21006
我本来希望按组排序的原始数据具有更好的数据局部性并且速度更快，但我观察到相反的行为。我想知道是否有人可以假设原因？
共有2个答案
这就是为什么排序组比未排序组慢的原因；
首先，这里是求和循环的汇编代码：
008512C3 mov ecx,dword ptr [eax+ebx]
008512C6 lea eax,[eax+4]
008512C9 lea edx,[esi+ecx*4] // &sums[groups[i]]
008512CC mov ecx,dword ptr [eax-4] // values[i]
008512CF add dword ptr [edx],ecx // sums[groups[i]]+=values[i]
008512D1 sub edi,1
008512D4 jne main+163h (08512C3h)
让我们看看add指令，这是这个问题的主要原因；
008512CF add dword ptr [edx],ecx // sums[groups[i]]+=values[i]
当处理器执行此指令时，它将首先向edx中的地址发出存储器读取(加载)请求，然后添加ecx的值，然后向同一地址发出写入(存储)请求。
处理器调用者内存重新排序中有一个html" target="_blank">功能
为了优化指令执行的性能，IA-32体系结构允许在奔腾4、英特尔至强和P6系列处理器中偏离称为处理器排序的强排序模型。这些processorordering变体（这里称为内存排序模型）允许性能增强操作，例如允许读操作先于缓冲写操作。任何这些变化的目标都是提高指令执行速度，同时保持内存一致性，即使在多处理器系统中也是如此。
还有一条规则
读取可以使用对不同位置的较旧写入重新排序，但不使用对同一位置的较老写入重新排序。
因此，如果下一次迭代在写入请求完成之前到达add指令，如果edx地址与前一个值不同，它将不会等待，并发出读取请求，并在旧的写入请求上重新排序，add指令继续。但如果地址相同，add指令将等到旧的写入完成。
请注意，循环很短，处理器执行循环的速度比存储器控制器完成写入存储器请求的速度快。
因此，对于已排序的组，您将从同一地址连续多次读取和写入，因此它将失去使用内存重新排序的性能增强；同时，如果使用随机组，则每次迭代可能会有不同的地址，因此读操作不会等待更早的写操作并在它之前重新排序；加法指令不会等待前一条指令执行。
首先，程序在几乎相同的时间运行，不管:
sumspeed$ time ./sum_groups < groups_shuffled
11558358
real 0m0.705s
user 0m0.692s
sys 0m0.013s
sumspeed$ time ./sum_groups < groups_sorted
24986825
real 0m0.722s
user 0m0.711s
sys 0m0.012s
大部分时间都花在输入循环中。但是，由于我们对grouped_sum（）感兴趣，让我们忽略它。
将基准测试循环从 10 次迭代更改为 1000 次迭代，grouped_sum（） 开始主导运行时间：
sumspeed$ time ./sum_groups < groups_shuffled
1131838420
real 0m1.828s
user 0m1.811s
sys 0m0.016s
sumspeed$ time ./sum_groups < groups_sorted
2494032110
real 0m3.189s
user 0m3.169s
sys 0m0.016s
现在我们可以使用perf来查找我们程序中的热点。
sumspeed$ perf record ./sum_groups < groups_shuffled
1166805982
[ perf record: Woken up 1 times to write data ]
[kernel.kallsyms] with build id 3a2171019937a2070663f3b6419330223bd64e96 not found, continuing without symbols
Warning:
Processed 4636 samples and lost 6.95% samples!
[ perf record: Captured and wrote 0.176 MB perf.data (4314 samples) ]
sumspeed$ perf record ./sum_groups < groups_sorted
2571547832
[ perf record: Woken up 2 times to write data ]
[kernel.kallsyms] with build id 3a2171019937a2070663f3b6419330223bd64e96 not found, continuing without symbols
[ perf record: Captured and wrote 0.420 MB perf.data (10775 samples) ]
以及它们之间的区别：
sumspeed$ perf diff
[...]
# Event 'cycles:uppp'
#
# Baseline Delta Abs Shared Object Symbol
# ........ ......... ................... ........................................................................
#
57.99% +26.33% sum_groups [.] main
12.10% -7.41% libc-2.23.so [.] _IO_getc
9.82% -6.40% libstdc++.so.6.0.21 [.] std::num_get<char, std::istreambuf_iterator<char, std::char_traits<c
6.45% -4.00% libc-2.23.so [.] _IO_ungetc
2.40% -1.32% libc-2.23.so [.] _IO_sputbackc
1.65% -1.21% libstdc++.so.6.0.21 [.] 0x00000000000dc4a4
1.57% -1.20% libc-2.23.so [.] _IO_fflush
1.71% -1.07% libstdc++.so.6.0.21 [.] std::istream::sentry::sentry
1.22% -0.77% libstdc++.so.6.0.21 [.] std::istream::operator>>
0.79% -0.47% libstdc++.so.6.0.21 [.] __gnu_cxx::stdio_sync_filebuf<char, std::char_traits<char> >::uflow
[...]
在 main（） 中需要更多的时间，这可能已经内联了grouped_sum（）。太好了，非常感谢，性能。
在 main（） 中花费的时间有区别吗？
洗牌：
sumspeed$ perf annotate -i perf.data.old
[...]
│ // This is the function whose performance I am interested in
│ void grouped_sum(int* p_x, int *p_g, int n, int* p_out) {
│ for (size_t i = 0; i < n; ++i) {
│180: xor %eax,%eax
│ test %rdi,%rdi
│ ↓ je 1a4
│ nop
│ p_out[p_g[i]] += p_x[i];
6,88 │190: movslq (%r9,%rax,4),%rdx
58,54 │ mov (%r8,%rax,4),%esi
│ #include <chrono>
│ #include <vector>
│
│ // This is the function whose performance I am interested in
│ void grouped_sum(int* p_x, int *p_g, int n, int* p_out) {
│ for (size_t i = 0; i < n; ++i) {
3,86 │ add $0x1,%rax
│ p_out[p_g[i]] += p_x[i];
29,61 │ add %esi,(%rcx,%rdx,4)
[...]
已排序:
sumspeed$ perf annotate -i perf.data
[...]
│ // This is the function whose performance I am interested in
│ void grouped_sum(int* p_x, int *p_g, int n, int* p_out) {
│ for (size_t i = 0; i < n; ++i) {
│180: xor %eax,%eax
│ test %rdi,%rdi
│ ↓ je 1a4
│ nop
│ p_out[p_g[i]] += p_x[i];
1,00 │190: movslq (%r9,%rax,4),%rdx
55,12 │ mov (%r8,%rax,4),%esi
│ #include <chrono>
│ #include <vector>
│
│ // This is the function whose performance I am interested in
│ void grouped_sum(int* p_x, int *p_g, int n, int* p_out) {
│ for (size_t i = 0; i < n; ++i) {
0,07 │ add $0x1,%rax
│ p_out[p_g[i]] += p_x[i];
43,28 │ add %esi,(%rcx,%rdx,4)
[...]
不，是同样的两条指令支配着。因此，在这两种情况下，它们都需要很长时间，但在对数据进行排序时，情况会更糟。
好的。但是我们应该运行它们相同的次数，所以每条指令肯定会因为某种原因变得更慢。让我们看看perf stat说什么。
sumspeed$ perf stat ./sum_groups < groups_shuffled
1138880176
Performance counter stats for './sum_groups':
1826,232278 task-clock (msec) # 0,999 CPUs utilized
72 context-switches # 0,039 K/sec
1 cpu-migrations # 0,001 K/sec
4 076 page-faults # 0,002 M/sec
5 403 949 695 cycles # 2,959 GHz
930 473 671 stalled-cycles-frontend # 17,22% frontend cycles idle
9 827 685 690 instructions # 1,82 insn per cycle
# 0,09 stalled cycles per insn
2 086 725 079 branches # 1142,639 M/sec
2 069 655 branch-misses # 0,10% of all branches
1,828334373 seconds time elapsed
sumspeed$ perf stat ./sum_groups < groups_sorted
2496546045
Performance counter stats for './sum_groups':
3186,100661 task-clock (msec) # 1,000 CPUs utilized
5 context-switches # 0,002 K/sec
0 cpu-migrations # 0,000 K/sec
4 079 page-faults # 0,001 M/sec
9 424 565 623 cycles # 2,958 GHz
4 955 937 177 stalled-cycles-frontend # 52,59% frontend cycles idle
9 829 009 511 instructions # 1,04 insn per cycle
# 0,50 stalled cycles per insn
2 086 942 109 branches # 655,014 M/sec
2 078 204 branch-misses # 0,10% of all branches
3,186768174 seconds time elapsed
只有一件事很突出：前端停滞的周期。
好的，指令流水线停止了。在前台。确切的含义可能因微架构而异。
不过，我有个猜测。如果你很慷慨，你甚至可以称之为假设。
通过对输入进行排序，您增加了写入的局部性。事实上，它们将非常本地化；您所做的几乎所有添加都将写入与前一个相同的位置。
这对缓存来说很好，但对管道来说不太好。您正在引入数据依赖项，阻止下一个添加指令继续进行，直到上一个添加完成（或以其他方式使结果可用于后续指令）
那是你的问题。
我认为。
实际上，让我们尝试一些东西。如果我们使用多个和向量，为每个加法在它们之间切换，然后在最后对它们求和，会怎么样？这会使我们损失一点局部性，但应该消除数据依赖关系。
(代码不好看；不要评价我，互联网！！)
#include <iostream>
#include <chrono>
#include <vector>
#ifndef NSUMS
#define NSUMS (4) // must be power of 2 (for masking to work)
#endif
// This is the function whose performance I am interested in
void grouped_sum(int* p_x, int *p_g, int n, int** p_out) {
for (size_t i = 0; i < n; ++i) {
p_out[i & (NSUMS-1)][p_g[i]] += p_x[i];
}
}
int main() {
std::vector<int> values;
std::vector<int> groups;
std::vector<int> sums[NSUMS];
int n_groups = 0;
// Read in the values and calculate the max number of groups
while(std::cin) {
int value, group;
std::cin >> value >> group;
values.push_back(value);
groups.push_back(group);
if (group >= n_groups) {
n_groups = group+1;
}
}
for (int i=0; i<NSUMS; ++i) {
sums[i].resize(n_groups);
}
// Time grouped sums
std::chrono::system_clock::time_point start = std::chrono::system_clock::now();
int* sumdata[NSUMS];
for (int i = 0; i < NSUMS; ++i) {
sumdata[i] = sums[i].data();
}
for (int i = 0; i < 1000; ++i) {
grouped_sum(values.data(), groups.data(), values.size(), sumdata);
}
for (int i = 1; i < NSUMS; ++i) {
for (int j = 0; j < n_groups; ++j) {
sumdata[0][j] += sumdata[i][j];
}
}
std::chrono::system_clock::time_point end = std::chrono::system_clock::now();
std::cout << (end - start).count() << " with NSUMS=" << NSUMS << std::endl;
return 0;
}
（哦，我还修复了n_groups计算；它偏离了一。）
在配置我的makefile以向编译器提供-DNSUMS=…arg之后，我可以这样做：
sumspeed$ for n in 1 2 4 8 128; do make -s clean && make -s NSUMS=$n && (perf stat ./sum_groups < groups_shuffled && perf stat ./sum_groups < groups_sorted) 2>&1 | egrep '^[0-9]|frontend'; done
1134557008 with NSUMS=1
924 611 882 stalled-cycles-frontend # 17,13% frontend cycles idle
2513696351 with NSUMS=1
4 998 203 130 stalled-cycles-frontend # 52,79% frontend cycles idle
1116188582 with NSUMS=2
899 339 154 stalled-cycles-frontend # 16,83% frontend cycles idle
1365673326 with NSUMS=2
1 845 914 269 stalled-cycles-frontend # 29,97% frontend cycles idle
1127172852 with NSUMS=4
902 964 410 stalled-cycles-frontend # 16,79% frontend cycles idle
1171849032 with NSUMS=4
1 007 807 580 stalled-cycles-frontend # 18,29% frontend cycles idle
1118732934 with NSUMS=8
881 371 176 stalled-cycles-frontend # 16,46% frontend cycles idle
1129842892 with NSUMS=8
905 473 182 stalled-cycles-frontend # 16,80% frontend cycles idle
1497803734 with NSUMS=128
1 982 652 954 stalled-cycles-frontend # 30,63% frontend cycles idle
1180742299 with NSUMS=128
1 075 507 514 stalled-cycles-frontend # 19,39% frontend cycles idle
和向量的最佳数量可能取决于CPU的流水线深度。我7年前的ultrabook CPU可能比一个新的台式机CPU所需的向量更少，可以最大限度地利用管道。
显然，更多不一定更好；当我疯狂地使用128个和向量时，我们开始遭受更多的缓存未命中的痛苦——正如您最初预期的那样，混洗的输入变得比排序慢就证明了这一点。我们已经走了一个完整的圈子！：）
（这是在编辑中添加的）
啊，书呆子狙击！如果您知道您的输入将被排序并正在寻找更高的性能，那么以下重写函数（没有额外的和数组）甚至更快，至少在我的计算机上是这样。
// This is the function whose performance I am interested in
void grouped_sum(int* p_x, int *p_g, int n, int* p_out) {
int i = n-1;
while (i >= 0) {
int g = p_g[i];
int gsum = 0;
do {
gsum += p_x[i--];
} while (i >= 0 && p_g[i] == g);
p_out[g] += gsum;
}
}
这个诀窍在于它允许编译器将gsum变量（组的总和）保存在寄存器中。我猜（但可能非常错误）这更快，因为管道中的反馈循环在这里可以更短，和/或更少的内存访问。一个好的分支预测器将使组相等性的额外检查变得便宜。
对于洗牌输入来说，这是可怕的...
sumspeed$ time ./sum_groups < groups_shuffled
2236354315
real 0m2.932s
user 0m2.923s
sys 0m0.009s
…但比排序输入的“多和”解决方案快40%左右。
sumspeed$ time ./sum_groups < groups_sorted
809694018
real 0m1.501s
user 0m1.496s
sys 0m0.005s
许多小团体将比几个小组慢，因此这是否是更快的实现将真正取决于您在此处的数据。而且，一如既往，在您的CPU型号上。
Sopel建议使用四种展开的添加来替代我的位屏蔽方法。我已经实现了他们建议的通用版本，它可以处理不同的nsum。我指望编译器为我们展开内部循环（至少对于NSUMS=4，它做到了）。
#include <iostream>
#include <chrono>
#include <vector>
#ifndef NSUMS
#define NSUMS (4) // must be power of 2 (for masking to work)
#endif
#ifndef INNER
#define INNER (0)
#endif
#if INNER
// This is the function whose performance I am interested in
void grouped_sum(int* p_x, int *p_g, int n, int** p_out) {
size_t i = 0;
int quadend = n & ~(NSUMS-1);
for (; i < quadend; i += NSUMS) {
for (int k=0; k<NSUMS; ++k) {
p_out[k][p_g[i+k]] += p_x[i+k];
}
}
for (; i < n; ++i) {
p_out[0][p_g[i]] += p_x[i];
}
}
#else
// This is the function whose performance I am interested in
void grouped_sum(int* p_x, int *p_g, int n, int** p_out) {
for (size_t i = 0; i < n; ++i) {
p_out[i & (NSUMS-1)][p_g[i]] += p_x[i];
}
}
#endif
int main() {
std::vector<int> values;
std::vector<int> groups;
std::vector<int> sums[NSUMS];
int n_groups = 0;
// Read in the values and calculate the max number of groups
while(std::cin) {
int value, group;
std::cin >> value >> group;
values.push_back(value);
groups.push_back(group);
if (group >= n_groups) {
n_groups = group+1;
}
}
for (int i=0; i<NSUMS; ++i) {
sums[i].resize(n_groups);
}
// Time grouped sums
std::chrono::system_clock::time_point start = std::chrono::system_clock::now();
int* sumdata[NSUMS];
for (int i = 0; i < NSUMS; ++i) {
sumdata[i] = sums[i].data();
}
for (int i = 0; i < 1000; ++i) {
grouped_sum(values.data(), groups.data(), values.size(), sumdata);
}
for (int i = 1; i < NSUMS; ++i) {
for (int j = 0; j < n_groups; ++j) {
sumdata[0][j] += sumdata[i][j];
}
}
std::chrono::system_clock::time_point end = std::chrono::system_clock::now();
std::cout << (end - start).count() << " with NSUMS=" << NSUMS << ", INNER=" << INNER << std::endl;
return 0;
}
时间来衡量。注意，由于我昨天在/tmp中工作，我没有完全相同的输入数据。因此，这些结果不能与之前的结果直接比较(但可能足够接近)。
sumspeed$ for n in 2 4 8 16; do for inner in 0 1; do make -s clean && make -s NSUMS=$n INNER=$inner && (perf stat ./sum_groups < groups_shuffled && perf stat ./sum_groups < groups_sorted) 2>&1 | egrep '^[0-9]|frontend'; done; done1130558787 with NSUMS=2, INNER=0
915 158 411 stalled-cycles-frontend # 16,96% frontend cycles idle
1351420957 with NSUMS=2, INNER=0
1 589 408 901 stalled-cycles-frontend # 26,21% frontend cycles idle
840071512 with NSUMS=2, INNER=1
1 053 982 259 stalled-cycles-frontend # 23,26% frontend cycles idle
1391591981 with NSUMS=2, INNER=1
2 830 348 854 stalled-cycles-frontend # 45,35% frontend cycles idle
1110302654 with NSUMS=4, INNER=0
890 869 892 stalled-cycles-frontend # 16,68% frontend cycles idle
1145175062 with NSUMS=4, INNER=0
948 879 882 stalled-cycles-frontend # 17,40% frontend cycles idle
822954895 with NSUMS=4, INNER=1
1 253 110 503 stalled-cycles-frontend # 28,01% frontend cycles idle
929548505 with NSUMS=4, INNER=1
1 422 753 793 stalled-cycles-frontend # 30,32% frontend cycles idle
1128735412 with NSUMS=8, INNER=0
921 158 397 stalled-cycles-frontend # 17,13% frontend cycles idle
1120606464 with NSUMS=8, INNER=0
891 960 711 stalled-cycles-frontend # 16,59% frontend cycles idle
800789776 with NSUMS=8, INNER=1
1 204 516 303 stalled-cycles-frontend # 27,25% frontend cycles idle
805223528 with NSUMS=8, INNER=1
1 222 383 317 stalled-cycles-frontend # 27,52% frontend cycles idle
1121644613 with NSUMS=16, INNER=0
886 781 824 stalled-cycles-frontend # 16,54% frontend cycles idle
1108977946 with NSUMS=16, INNER=0
860 600 975 stalled-cycles-frontend # 16,13% frontend cycles idle
911365998 with NSUMS=16, INNER=1
1 494 671 476 stalled-cycles-frontend # 31,54% frontend cycles idle
898729229 with NSUMS=16, INNER=1
1 474 745 548 stalled-cycles-frontend # 31,24% frontend cycles idle
是的，NSUMS=8的内部循环在我的计算机上是最快的。与我的“本地gsum”方法相比，它还有一个额外的好处，那就是对于混洗的输入来说不会变得很糟糕。
有趣的是:< code>NSUMS=16变得比< code>NSUMS=8更差。这可能是因为我们开始看到更多的缓存未命中，或者因为我们没有足够的寄存器来正确展开内部循环。
-
问题内容： 这是一段C ++代码，显示了一些非常特殊的行为。由于某些奇怪的原因，奇迹般地对数据进行排序使代码快了将近六倍： 不使用std::sort(data, data + arraySize);，代码将在11.54秒内运行。 使用排序的数据，代码将在1.93秒内运行。 最初，我认为这可能只是语言或编译器异常，所以我尝试了Java： 具有类似但不太极端的结果。 我首先想到的是排序将数据带入缓存，
-
在这篇文章中，为什么处理排序数组比处理随机数组更快，它说分支预测是排序数组性能提升的原因。 但是我刚刚使用Python尝试了这个例子；我认为排序数组和随机数组没有区别（我尝试了字节数组和数组；并使用line_profile来分析计算）。 我遗漏了什么吗？ 这是我的代码：
-
我有一个多维numpy数组，其中包含代表一个框的四个点（坐标）： [开始X，开始Y]，[结束X，开始Y]，[结尾X，结束Y]，[开始X，结束Y] 一个例子: 因为每个框都代表一个单词在图像中的位置，所以我想按照阅读的方式对这些框进行排序。从左上角开始，到右下角结束。我正在考虑按框的起始值排序： 这很好。现在，我必须将每个高度上与另一个重叠的框分组为一行。我用以下方法进行检查： 将分组框放在一行上，
-
我一直试图这样做了一段时间，但我不能这样做。 如何将默认的升序更改为降序？另外，还有什么简单的方法可以对数组进行排序吗？
-
问题内容： 希望有人知道此Java认证问题的答案： 哪两个结果可能？（选择两个。） A）7 0 B）7 1 C）7 3 D）-1 0 E）-1 1 F）-1 3 唯一的正确答案是E）-1 1，因为如果您执行二进制搜索算法，这是唯一可能的输出。但是他们要我选择两个…所以第二个必须是B）7 1然后，因为排序数组中的第二个binarySearch总是会返回。 所以我的问题是，为什么B）7 1是可能的结果
-
问题内容： 如果binarySearch方法要求您先对数组进行排序，然后再将其作为参数传递给方法调用，那么为什么不对binarySearch方法进行排序呢？ 问题答案： 二进制搜索的工作原理是假设数组的中间包含数组中的中值。如果未排序，则此假设就没有意义，因为中位数可以在任何地方，并且将数组减半可能意味着您削减了要搜索的数字。 二进制搜索不进行排序本身的原因是因为它不需要…该数组已排序。