logo标签列表

缓存友好的离线随机读取

algorithmperformanceoptimizationx86cpu-cache
4

考虑下面这个C++函数：

void foo(uint32_t *a1, uint32_t *a2, uint32_t *b1, uint32_t *b2, uint32_t *o) {
    while (b1 != b2) {
        // assert(0 <= *b1 && *b1 < a2 - a1)
        *o++ = a1[*b1++];
    }
}

它的目的应该足够清楚。不幸的是，b1 包含 随机 数据并清除缓存，使得 foo 成为我程序的瓶颈。有没有办法可以优化它？

这是一个类似于我的实际代码的 SSCCE：

#include <iostream>
#include <chrono>
#include <algorithm>
#include <numeric>

namespace {
    void foo(uint32_t *a1, uint32_t *a2, uint32_t *b1, uint32_t *b2, uint32_t *o) {
        while (b1 != b2) {
            // assert(0 <= *b1 && *b1 < a2 - a1)
            *o++ = a1[*b1++];
        }
    }

    constexpr unsigned max_n = 1 << 24, max_q = 1 << 24;
    uint32_t data[max_n], index[max_q], result[max_q];
}

int main() {
    uint32_t seed = 0;
    auto rng = [&seed]() { return seed = seed * 9301 + 49297; };
    std::generate_n(data, max_n, rng);
    std::generate_n(index, max_q, [rng]() { return rng() % max_n; });

    auto t1 = std::chrono::high_resolution_clock::now();
    foo(data, data + max_n, index, index + max_q, result);
    auto t2 = std::chrono::high_resolution_clock::now();
    std::cout << std::chrono::duration<double>(t2 - t1).count() << std::endl;

    uint32_t hash = 0;
    for (unsigned i = 0; i < max_q; i++)
        hash += result[i] ^ (i << 8) ^ i;
    std::cout << hash << std::endl;
}

这不是关于随机写入并假设b为置换的已知索引、收集、散布数组的缓存友好复制，有关内容请参考此处。

1
在一个快速测试中，我可以通过应用一些预取（虽然模式是随机的，但很容易提前查看）稍微加速它，但只有勉强足够的差异才能被测量为合法的差异。也许了解更多这种内存优化的人可以从中获得更多收益。 - harold
@johnchen902 我不确定它是否会有影响，但你知道它已经排序了，所以当你将其用作索引时，它的随机访问会更少。 - NoSenseEtAl
2
@harold - 是的，预取可能在这里并没有太大作用，因为每次迭代的索引不依赖于先前的读取，所以即使是普通的OoO芯片也可以在这里获得接近完整的MLP。预取有时仍然可以帮助一些，特别是提前触发页面行走。 - BeeOnRope
@johnchen902 - 你有特定的硬件平台吗？你能使用除了C/C++之外的其他语言吗？ - BeeOnRope
1
@BeeOnRope（1）假设它是一个带有扩展和缓存大小的x86_64，您可以选择自己的配置。（我的笔记本电脑是i7-4712MQ。L1d缓存32K，L2 256K，L3 6144K。）（2）是的。 - johnchen902
显示剩余3条评论
2个回答
5

首先，让我们来看一下上面代码的实际性能：

$ sudo perf stat ./offline-read
0.123023
1451229184

 Performance counter stats for './offline-read':

        184.661547      task-clock (msec)         #    0.997 CPUs utilized          
                 3      context-switches          #    0.016 K/sec                  
                 0      cpu-migrations            #    0.000 K/sec                  
               717      page-faults               #    0.004 M/sec                  
       623,638,834      cycles                    #    3.377 GHz                    
       419,309,952      instructions              #    0.67  insn per cycle         
        70,803,672      branches                  #  383.424 M/sec                  
            16,895      branch-misses             #    0.02% of all branches        

       0.185129552 seconds time elapsed

我们的IPC很低，只有0.67，可能主要是由于对DRAM5的负载未命中引起的。让我们确认一下：
sudo ../pmu-tools/ocperf.py stat -e cycles,LLC-load-misses,cycle_activity.stalls_l3_miss ./offline-read
perf stat -e cycles,LLC-load-misses,cpu/event=0xa3,umask=0x6,cmask=6,name=cycle_activity_stalls_l3_miss/ ./offline-read
0.123979
1451229184

 Performance counter stats for './offline-read':

       622,661,371      cycles                                                      
        16,114,063      LLC-load-misses                                             
       368,395,404      cycle_activity_stalls_l3_miss                                   

       0.184045411 seconds time elapsed

我想你知道，620,000个周期中有大约370,000个周期因未完成的缺失而被暂停。实际上，在foo()中以这种方式暂停的周期比例要高得多，接近90％，因为perf还测量了大约三分之一的运行时间（但没有重要的L3缺失）的初始化和accumulate代码。

这并不出乎意料，因为我们知道随机读取模式a1 [* b1 ++]几乎没有局部性。实际上，LLC-load-misses的数量为16百万1，几乎完全对应于a1的16百万次随机读取。2

如果我们假设100％的foo()花费在等待内存访问上，我们就可以大致了解每个缺失的总成本：0.123秒/ 16,114,063个缺失 == 7.63 ns/miss。在我的计算机上，最佳情况下的内存延迟约为60 ns，因此每个缺失不到8 ns意味着我们已经提取了很多内存级并行性（MLP）：平均需要重叠和正在进行处理的8个缺失才能实现这一点（甚至完全忽略来自b1的流式加载和o的流式写入所需的额外流量）。
因此，我认为对于简单循环，您不能应用太多优化。但仍有两种可能性：
  • 如果您的平台支持，可以使用非时间性存储来写入o。这将消除普通存储所隐含的RFO读取。这应该是一个直接的胜利，因为o在定时部分内不会再被读取！

  • 软件预取。精心调整a1或b1的预取可能有助于一点点。然而，影响将相当有限，因为我们已经接近上述MLP的极限。此外，我们预计b1的线性读取几乎可以完美地由硬件预取器预取。a1的随机读取似乎可以容易地进行预取，但在实践中，循环中的ILP足以通过乱序处理（至少在像最近的x86这样的大OoO处理器上）实现足够的MLP。

    在评论中，用户哈罗德已经提到他尝试过预取，但效果很小。

由于简单的调整不太可能有很大的效果，所以您需要转换循环。一种“明显”的转换是对索引b1（以及索引元素的原始位置）进行排序，然后按排序顺序从a1中读取。这将把对a1的读取从完全随机变成几乎3线性，但现在写入是全部随机的，这并没有改善情况。

排序和取消排序

关键问题在于由b1控制的a1的读取是随机的，而且a1很大，每次读取都会导致缺失到DRAM。我们可以通过对b1进行排序，然后按顺序读取a1来获取一个排列结果来解决这个问题。现在您需要“取消排列”结果a1以获得最终顺序的结果，这只是另一种排序，这次是在“输出索引”上进行。

这是一个使用给定的输入数组a，索引数组b和输出数组o的工作示例，以及i表示每个元素的（隐式）位置：
  i =   0   1   2   3
  a = [00, 10, 20, 30]
  b = [ 3,  1,  0,  1]
  o = [30, 10, 00, 10] (desired result)

首先，对数组b进行排序，将原始数组位置i作为次要数据（或者您可以将其视为对元组进行排序(b[0], 0), (b[1], 1), ...），这将为您提供已排序的b数组b'和已排序的索引列表i'，如下所示：
  i' = [ 2, 1, 3, 0]
  b' = [ 0, 1, 1, 3]

现在你可以在b'的控制下从a中读取排列结果数组o'。这个读取过程按照严格递增顺序进行，应该能够接近memcpy速度运行。实际上，你可能可以利用广泛连续的SIMD读取和一些洗牌来做几个读取并一次移动4字节元素到正确位置（复制一些元素并跳过其他元素）。
  a  = [00, 10, 20, 30]
  b' = [ 0,  1,  1,  3]
  o' = [00, 10, 10, 30]

最后，你需要对 o' 进行去排列以得到 o，概念上只需按照排列后的索引 i' 对 o' 进行排序即可。
  i' = [ 2,  1,  3,  0]
  o' = [00, 10, 10, 30]
  i  = [ 0,  1,  2,  3]
  o  = [30, 10, 00, 10]

完成！

现在这是最简单的技术想法，不太适合缓存（每个迭代概念上都会遍历一个或多个2^26字节的数组），但至少完全使用了它读取的每一条缓存线（与原始循环不同，原始循环只从缓存行中读取单个元素，这就是为什么即使数据仅占用100万个缓存行，你也有1600万次缺失！）。所有的读取都是更或多少线性的，所以硬件预取将有很大帮助。

你获得多少加速可能取决于你如何实现排序：它们需要快速且对缓存敏感。几乎肯定某种类型的缓存感知基数排序效果最好。

以下是改进此内容的一些注释：

优化排序量

你实际上不需要完全对b进行排序。你只需要将其排序“足够”，以使在b'的控制下对a的后续读取更加线性。例如，16个元素适合于缓存行，因此您不需要根据最后4位进行排序：无论如何都将读取相同的缓存行的线性序列。您还可以更少地排序：例如，如果您忽略了5个最低有效位，则会以“几乎线性”的方式读取缓存行，有时从完全线性模式中交换两个缓存行，如：0、1、3、2、5、4、6、7。在这里，您仍将获得L1缓存的全部好处（对缓存行的后续读取将始终命中），我认为这样的模式仍然会被良好地预取，如果没有，您总是可以通过软件预取来帮助它。
您可以在系统上测试忽略的最佳位数。忽略位有两个好处：
  • 在基数搜索中需要做的工作更少，可以通过减少所需的遍数或在一个或多个遍历中需要更少的桶（有助于缓存）来实现。
  • 在最后一步中，"撤销"排列可能需要的工作量也可能更少：如果通过检查原始索引数组b来进行撤销并忽略位，则意味着撤销搜索时可以获得相同的节省。

缓存块处理

上述描述将所有内容按顺序分为几个不连续的遍历，每个遍历都在整个数据集上进行。在实践中，您可能希望交错它们以获得更好的缓存行为。例如，假设您使用 MSD 基数-256 排序，则可以进行第一次遍历，将数据排序为大约 256K 元素的 256 个桶。

如果不进行完整的第二遍排序，您可以完成仅对第一个（或前几个）桶进行排序，并根据生成的 b' 块执行对 a 的读取。您保证此块是连续的（即最终排序序列的后缀），因此您不会放弃读取中的任何局部性，并且您的读取通常会被缓存。您还可以进行去排列的第一遍处理 o'，因为 o' 的块也在高速缓存中（也许您可以将后两个阶段合并为一个循环）。

智能去排列

优化的一个方面是如何实现对o'的去排列。在上面的描述中，我们假设一些索引数组i最初具有值[0,1,2，...，max_q]，并与b一起排序。这是它的工作方式的概念，但您可能不需要立即实现i并将其排序为辅助数据。例如，在基数排序的第一遍中，可以隐式地知道i的值（因为您正在迭代数据），因此可以免费计算并在第一遍中写出，而无需按排序顺序出现。
还可能有更有效的方法来执行“取消排序”操作，而不是维护完整的索引。例如，原始未排序的b数组在概念上具有执行取消排序所需的所有信息，但我不清楚如何使用它来高效地取消排序。
它会更快吗？
这样做是否比朴素方法更快？这在很大程度上取决于实现细节，特别是所实现排序的效率。在我的硬件上，朴素方法每秒处理约140万个元素。在线缓存感知基数排序的描述似乎从200到600百万个元素/秒不等，而且由于需要两个排序，如果你相信这些数字，那么大幅加速的机会似乎有限。另一方面，这些数字来自较旧的硬件，并且对于稍微更一般的搜索（例如，对于密钥的所有32位，而我们可能只能使用16位），这些数字也适用。

只有仔细的实现才能确定其可行性，而可行性还取决于硬件。例如，在无法维持太多MLP的硬件上，排序-取消排序方法变得相对更有利。

最佳方法还取决于max_n和max_q的相对值。例如，如果max_n >> max_q，即使使用最佳排序，读取也会很“稀疏”，因此朴素的方法会更好。另一方面，如果max_n << max_q，则通常会多次读取相同的索引，因此排序方法将具有良好的读取局部性，排序步骤本身将具有更好的局部性，并且进一步优化可以明确处理重复读取。

多核

从问题中并不清楚您是否有兴趣并行化。对于foo()的朴素解决方案已经允许“直接”并行化，其中您只需将a和b数组分成大小相等的块，每个线程一个，这似乎提供了完美的加速。不幸的是，您可能会发现，由于内存控制器和相关的uncore/offcore资源在套接字上的所有核心之间共享，因此您将遇到资源争用，因此您将得到远低于线性比例的扩展。因此，当您添加更多核心时，纯并行随机读取负载对内存的吞吐量将不清楚会增加多少。
对于基数排序版本，大多数瓶颈（存储吞吐量、总指令吞吐量）在核心部分，因此我预计它将随着额外核心的增加而合理扩展。正如彼得在评论中提到的那样，如果您正在使用超线程，则排序可能具有良好的局部性能，在核心本地L1和L2缓存中，有效地让每个兄弟线程使用整个缓存，而不是将有效容量减半。当然，这涉及仔细管理线程亲和力，以使兄弟线程实际上使用附近的数据，而不仅仅是让调度程序执行其操作。

1 你可能会问为什么 LLC-load-misses 不是3200万或4800万，因为我们还要读取b1的1600万个元素，然后accumulate()调用会读取整个result。答案是，LLC-load-misses只计算在L3中实际缺失的需求缺失。其他提到的读取模式是完全线性的，因此预取器将始终在需要之前将行带入L3中。按perf使用的定义，这些不算作“LLC缺失”。

2 你可能想知道我如何确定缺失加载全部来自于foo中对a1的读取：我只是使用perf record和perf mem确认缺失来自于预期的汇编指令。

3 线性近似，因为b1不是所有索引的排列，所以原则上可能会跳过和重复的索引。但在缓存行级别上，每个元素有约63％的概率被包含在内，因此高度可能按顺序读取每个缓存行，并且缓存行具有16个4字节元素，因此任何给定缓存没有零元素的机会只有大约1000万分之1。因此，在缓存行级别上运行的预取将很好地工作。

4 这里我指的是计算值的成本几乎为零，但写入仍然需要成本。然而，这仍然比“预先实例化”方法要好得多，该方法首先创建 i 数组[0, 1, 2, ...]，需要max_q次写入，然后再需要另外max_q次写入才能在第一个基数排序中对其进行排序。隐式实例化仅产生第二个写入。

5 实际定时部分 foo() 的IPC要低得多：根据我的计算，大约为0.15。整个进程的报告IPC是定时部分IPC和初始化累加代码之前和之后的平均值，这些代码具有更高的IPC。

6 值得注意的是，这与依赖加载延迟限制工作流程的缩放方式不同：一个正在进行随机读取但只能有一个加载正在进行的加载（因为每个加载都依赖于上一个结果）非常适合多个核心，因为加载的串行性质不使用许多下游资源（但是这样的加载在单个核心上也可以概念上加速，方法是更改核心循环以处理多个依赖加载流并行处理）。

1
对于OP提到的特定情况(hash += result[i] ^ (i << 8) ^ i;)，您可以优化掉将o[]放入正确顺序的步骤。uint32_t加法是可结合的，因此您可以在获取方便的任何顺序的o[i]时即时进行操作。具体来说，在x86上使用AVX2，对o'和i'进行SIMD循环，可以为您提供所需的数据以进行VPSLLD乘以8和VPXOR，然后VPADDD到向量累加器中。 - Peter Cordes
1
@peter - 呃，原始帖子中最初有一个std::accumulate调用来验证结果，但我认为它已经被更改了。无论如何，我认为想法只是优化foo函数：这就是正在计时的内容，也是SSCCE潜在问题的关键所在。据我所知，哈希值用于验证结果（并防止编译器优化），但本身并不真正有趣。 - BeeOnRope
是的，显然答案中的“将其放回顺序”部分很有用，但通常在任何特定用例中寻找避免这样做的方法是一个非常好的想法（正如这个用例所展示的那样）。或者至少要延迟它，直到您可以将其与慢计算相结合，因为这可能更好（如果内存延迟+ ALU延迟大于OOO执行可以隐藏的话，则更糟）。 - Peter Cordes
1
@PeterCordes 确实，实际底层问题中可能存在各种变化形式，可以采用各种优化措施，但是在哪里划定界限呢？我很确定实际问题中的索引不是通过随机数生成器生成的，我们也可能没有 max_p 和 max_q 的实际值。虽然问题的提出方式非常直接：使 foo() 更快，并且这本身就是一个有趣的问题——它很可能是现实世界中的问题（例如，结果 o 可能直接输出为原样或其他什么）。 - BeeOnRope
1
@PeterCordes - 你提出了一个很好的观点，关于排序版本的问题易于添加核心：基数排序通常似乎受到核心限制（例如，存储器限制，或总uop吞吐量限制，或者可能是由于各种内存/缓存延迟和TLB行为的混合），而原始问题几乎完全受到内存延迟的限制。因此，基数排序问题应该能够很好地扩展到更多的核心。 - BeeOnRope
显示剩余5条评论
0
您可以将索引划分为桶，其中索引的高位相同。请注意，如果索引不是随机的，则桶将溢出。
#include <iostream>
#include <chrono>
#include <cassert>
#include <algorithm>
#include <numeric>
#include <vector>

namespace {
constexpr unsigned max_n = 1 << 24, max_q = 1 << 24;

void foo(uint32_t *a1, uint32_t *a2, uint32_t *b1, uint32_t *b2, uint32_t *o) {
    while (b1 != b2) {
        // assert(0 <= *b1 && *b1 < a2 - a1)
        *o++ = a1[*b1++];
    }
}

uint32_t* foo_fx(uint32_t *a1, uint32_t *a2, uint32_t *b1, uint32_t *b2, const uint32_t b_offset, uint32_t *o) {
    while (b1 != b2) {
        // assert(0 <= *b1 && *b1 < a2 - a1)
        *o++ = a1[b_offset+(*b1++)];
    }
    return o;
}

uint32_t data[max_n], index[max_q], result[max_q];
std::pair<uint32_t, uint32_t[max_q / 8]>index_fx[16];

}

int main() {
    uint32_t seed = 0;
    auto rng = [&seed]() { return seed = seed * 9301 + 49297; };
    std::generate_n(data, max_n, rng);
    //std::generate_n(index, max_q, [rng]() { return rng() % max_n; });
    for (size_t i = 0; i < max_q;++i) {
        const uint32_t idx = rng() % max_n;
        const uint32_t bucket = idx >> 20;
        assert(bucket < 16);
        index_fx[bucket].second[index_fx[bucket].first] = idx % (1 << 20);
        index_fx[bucket].first++;
        assert((1 << 20)*bucket + index_fx[bucket].second[index_fx[bucket].first - 1] == idx);
    }
    auto t1 = std::chrono::high_resolution_clock::now();
    //foo(data, data + max_n, index, index + max_q, result);

    uint32_t* result_begin = result;
    for (int i = 0; i < 16; ++i) {
        result_begin = foo_fx(data, data + max_n, index_fx[i].second, index_fx[i].second + index_fx[i].first, (1<<20)*i, result_begin);
    }
    auto t2 = std::chrono::high_resolution_clock::now();
    std::cout << std::chrono::duration<double>(t2 - t1).count() << std::endl;
    std::cout << std::accumulate(result, result + max_q, 0ull) << std::endl;
}
1
你正在定时区域之外进行分区，这与OP的问题并不完全相同：你应该保持std::generate_n(index, ...)调用不变，并将任何对index的操作放入定时区域中以使其等效。否则，我可以在定时区域之外“排序”索引，这种方法看起来确实非常快！ - BeeOnRope
抱歉，我搞混了角色 - 当我说“如果你尝试过好的排序和分区实现，我不抱太大希望”时 - 我以为你是原帖作者！你的回答确实是一种分区方法，我认为总体来说这是可行的方法。当然，有些类型的分区（例如你的分区）每个元素只进行不到10次写操作。即使使用基数排序（基数为2^12），每个元素的排序也只需要约2次写操作。 - BeeOnRope
1
不幸的是，输出顺序很重要。你的解决方案没有尊重这一点。（在我看来，我的要求很明确，就是写一个等同于“foo”的函数）也许我不应该使用如此简单的std::accumulate，这样那些****就会更快地发现问题了。 - johnchen902
1
我的代码应该是正确的，因为校验和是相同的。这正是“也许我不应该使用如此简单的std::accumulate”的关键所在。显然重新排列元素不会改变std::accumulate的结果。 - johnchen902
1
如果您知道索引是均匀分布的，并且它们非常多，那么概率实际上就等于0。如果您感到无聊，可以打印出1000次不同运行中的桶的最大尺寸并查看。 - NoSenseEtAl
显示剩余9条评论
网页内容由stack overflow 提供, 点击上面的
可以查看英文原文，
原文链接