高效利用CPU Cache

2022-12-17 本文已影响0人谭英智

本文动机

提高程序性能，一般是通过算法复杂度来提升性能；

如果在复杂度无法再提升后，可以通过更底层次的优化来提升性能。

例如利用CPU Cache的硬件特性，来加快代码运行的速度

CPU Cache知识背景

CPU在访问内存的时候，会以cache line的形式（64bytes）加载数据到CPU cache。

当CPU访问的数据是连贯的

这样的取数形式，会达到加速的效果

否则，性能会下降

Tips

Tip：遍历数据，该使用数组还是链表？

如果是遍历一个集合的数据，是使用数组更优还是链表更优？

通过构造一个数组，并遍历其中的元素；构造一个相同大小的链表，并遍历其中的元素

对比他们之间的性能差别

遍历数组：

void processVector(benchmark::State& state) {
  for (auto _ : state) {
    std::vector<int> v;
    for(int i=0; i< size; ++i){
      v.push_back(i);
    }
    for(int i=0; i< size; i++) {
      for(auto& item: v){
        benchmark::DoNotOptimize(v);
      }
      state.SetBytesProcessed(size);
    }
    
  }
}
// Register the function as a benchmark
BENCHMARK(processVector);

遍历链表：

void processList(benchmark::State& state) {
  for (auto _ : state) {
    std::list<int> v;
    for(int i=0; i< size; ++i){
      v.push_back(i);
    }
    for(int i=0; i< size; i++) {
      for(auto& item: v){
        benchmark::DoNotOptimize(v);
      }
      state.SetBytesProcessed(size);
    }
  }
}
// Register the function as a benchmark
BENCHMARK(processList);

结果如下，遍历数组的性能是遍历链表的3倍

cache-vector-or-list
分析：

由于数组的内存是连续的，当CPU处理第一个元素的时候，会把后续的7个元素一同加载到CPU cache中，当CPU需要处理第二个元素时，则不需要再次访存，直接从cache中就能拿到数据
而链表，由于分配的内存不连续的，导致无法通过cache的预取机制来加速，反而每次获取下一个元素的时候，都很大可能发生cache miss

Tip：经常被一起访问的数据，应该尽量在内存中靠近彼此

定义一个类的时候，一般会有多个成员变量，成员变量跟成员变量之间的亲疏关系是不一样的。

通过构造两个亲密关系的成员变量，一个类彼此靠近，一个类彼此远离。

查看他们共同访问时，性能有什么差别

#include <vector>

class A {
public:
  int a;
  int b;
  int padding[16];
};
class B {
public:
  int a;
  int padding[16];
  int b;
};
#define kArrSize 100

彼此靠近的类：

static void closeMemory(benchmark::State& state) {
  for (auto _ : state) {
    A arr[kArrSize];
    for(auto& item: arr) {
      item.a = 10;
      item.b = 20;
      benchmark::DoNotOptimize(item);
    }
  }
}
BENCHMARK(closeMemory);

彼此远离的类：

static void farMemory(benchmark::State& state) {
  for (auto _ : state) {
    B arr[kArrSize];
    for(auto& item: arr) {
      item.a = 10;
      item.b = 20;
      benchmark::DoNotOptimize(item);
    }
  }
}
BENCHMARK(farMemory);

彼此靠近的类，访问速度是彼此远离的类的1.6倍

cache-close-or-far
分析：

成员变量彼此远离的类，故意制造了变量a和变量b不位于同一个cache line。那么当加载a时，由于b不位于同一个cache line，这时就不会加载到b，直到cpu要处理b时，发生cache miss，才把b加载到cache中。
成员变量彼此靠近的类，成员变量a和成员变量b的地址是连续的，那么当CPU加载a的时候，很大几率会同时加载b，达到加速的效果

Tip：使用对象数组而不是对象指针数组

当程序组织对象时，有两个选择，一是以指针方式放入容器中，一是把对象直接放入容器中。

那么这两种方式，在性能上，哪个更高呢？

#include <vector>
#include <random>
#include <algorithm>
class A {
public:
  int a;
  int b;
};

#define kArrSize 100000

构造元素是对象的数组，并进行遍历

static void arrOfObjectMemory(benchmark::State& state) {
  for (auto _ : state) {
    A arr[kArrSize];
    std::array<A*, kArrSize> point_arr;
    for(int i=0; i<kArrSize; ++i){
      point_arr[i] = &arr[i];
    }
    for(auto& item: arr) {
      item.a = 10;
      item.b = 20;
      benchmark::DoNotOptimize(item);
    }
  }
}
BENCHMARK(arrOfObjectMemory);

构造元素是对象指针的数组，并进行遍历

static void arrOfPointMemory(benchmark::State& state) {
  for (auto _ : state) {
    A arr[kArrSize];
    std::array<A*, kArrSize> point_arr;
    for(int i=0; i<kArrSize; ++i){
      point_arr[i] = &arr[i];
    }
    for(auto& item: point_arr) {
      item->a = 10;
      item->b = 20;
      benchmark::DoNotOptimize(item);
    }
  }
}
BENCHMARK(arrOfPointMemory);

遍历对象数组是对象指针数组的2.4倍

cache-array-of-object
分析：

虽然指针对象相邻的元素是内存连续的，但是由于多了指针内存的overhead，访问的内存就会增多，cache冲突也会增大，导致cache miss增大，性能不如对象数组

Tip：通过cache line大小来优化对象大小

如果有一个对象数组，对象的大小是48bytes（不是cache line大小64bytes），那么随机访问数组中的元素，性能会不会比以64 bytes大小对齐的对象性能下降呢？

#include <benchmark/benchmark.h>

#include <vector>
#include <random>
#include <algorithm>
class A {
public:
  int a;
  int number[4];
  int b;
};

class  alignas(64) B {
public:
  int a;
  int number[4];
  int b;
};

#define kArrSize 10000

构造48 bytes的对象数组，并遍历

static void _48bytesClass(benchmark::State& state) {
  for (auto _ : state) {
    A arr[kArrSize];
    int i = 0;
    srand(1);
    for(int j=0; j<10000; ++j){
      for(int i=0; i< kArrSize; ++i) {
        int index = rand()%(kArrSize-3);
        arr[index].a = i;
        arr[index].b = i*2;
        benchmark::DoNotOptimize(arr);
      }
    }
    
  }
}
BENCHMARK(_48bytesClass);

构造64 bytes的对象数组，并遍历

static void _64bytesClass(benchmark::State& state) {
  for (auto _ : state) {
    B arr[kArrSize];
    int i = 0;
    srand(1);
    for(int j=0; j<10000; ++j){
      for(int i=0; i< kArrSize; ++i) {
        int index = rand()%(kArrSize-3);
        arr[index].a = i;
        arr[index].b = i*2;
        benchmark::DoNotOptimize(arr);
      }
    }
  }
}
BENCHMARK(_64bytesClass);

结果64 bytes的对象随机访问比48 bytes的对象性能快1.2倍

cache-align-random
分析：

以cache line对齐的对象，对象都会占用一个cache line，那么加载时，只需要访问一个cache line就可以了
48 bytes的对象，因为不是cache line对齐的，那么第一个对象占用48bytes，剩余16 bytes 就会给第二个元素占用，那么第二个元素就会占用一个cache line 的尾16 bytes和另一个cache line的头32 bytes。当CPU需要访问跨两个cache的元素，则需要多次访问cache，才能加载完对象，导致性能不如以cache line对齐的对象

Tip：优化多维数组访问顺序

CPU cache是行优先的，对于列优先的数据结构，如果更好的利用cache呢？

使用列优先访问二维数组：

static void multipy(benchmark::State& state) {
  // Code inside this loop is measured repeatedly
  for (auto _ : state) {
    int in_matrix1[size][size] = {0};
    int in_matrix2[size][size] = {0};

    for(int i=0; i<size; ++i){
      for(int j=0; j<size; ++j) {
        in_matrix1[i][j] = i+j;
        in_matrix2[i][j] = i+j;
      }
    }
    
    int result[size][size] = {0};
    for(int m=0; m<loop_size; ++m){
      int i, j, k; 
      for (i = 0; i < size; i++) { 
          for (j = 0; j < size; j++) { 
              result[i][j] = 0; 
              for (k = 0; k < size; k++) {
                 result[i][j] += in_matrix1[i][k] *  
                               in_matrix2[k][j];
              benchmark::DoNotOptimize(result[i][j]); 
      
            } 
        } 
      } 
      
    }
    state.SetBytesProcessed(loop_size);
    
    
  }
}
// Register the function as a benchmark
BENCHMARK(multipy);

使用行优先访问二维数组

static void opt_multipy(benchmark::State& state) {
  // Code inside this loop is measured repeatedly
  for (auto _ : state) {
    int in_matrix1[size][size] = {0};
    int in_matrix2[size][size] = {0};

    for(int i=0; i<size; ++i){
      for(int j=0; j<size; ++j) {
        in_matrix1[i][j] = i+j;
        in_matrix2[i][j] = i+j;
      }
    }
    int result[size][size] = {0};
    for(int m=0; m<loop_size; ++m){
      int i, j, k; 
      for (i = 0; i < size; i++) { 
          for (j = 0; j < size; j++) { 
              result[i][j] = 0; 
          }
          for (k = 0; k < size; k++) {
              for (j = 0; j < size; j++) {
                  result[i][j] += in_matrix1[i][k] *  
                              in_matrix2[k][j]; 
                  benchmark::DoNotOptimize(result[i][j]);
      
            } 
        } 
      } 
      // Make sure the variable is not optimized away by compiler
      
      
    }
    state.SetBytesProcessed(loop_size);
    
  }
}
// Register the function as a benchmark
BENCHMARK(opt_multipy);

行优先的算法是列优先的性能的2倍

cache-metrics
分析：

由于CPU加载数据是以cache line加载的，cache line加载的内存是连续的，因为它更亲和行优先的算法

Tip：避免在类中发生默认padding

一个类中会有不同类型的成员变量，如果成员变量没有什么亲疏关系，如果调整他们的布局，会不会发生微妙的性能差别？

#include <vector>
#include <random>
#include <algorithm>
class A {
public:
  int my_int;
  double my_double;
  int my_second_int;
};

class B {
public:
  double my_double;
  int my_int;
  int my_second_int;
};

#define kArrSize 10000

构造顺序为 int double int的类

static void paddingClass(benchmark::State& state) {
  for (auto _ : state) {
    A arr[kArrSize];
    for(int j=0; j<10000; ++j){
      for(int i=0; i< kArrSize; ++i) {
        auto& item = arr[i];
        item.my_int = i;
        item.my_second_int = i+2;
        benchmark::DoNotOptimize(item);
      }
    }
  }
}
BENCHMARK(paddingClass);

构造顺序为double int int的类

static void noPaddingClass(benchmark::State& state) {
  for (auto _ : state) {
    B arr[kArrSize];
    for(int j=0; j<10000; ++j){
      for(int i=0; i< kArrSize; ++i) {
        auto& item = arr[i];
        item.my_int = i;
        item.my_second_int = i+2;
        benchmark::DoNotOptimize(item);
      }
    }
  }
}
BENCHMARK(noPaddingClass);

结果顺序为double int int的类的遍历快1.4倍

cache-padding
分析：

这是因为int double int的类，发生了padding，导致类占用的内存更大，那么加载到cache line中，占用的空间也会更大，而其中padding部分，是不必要的内存开销，导致cache miss增大

Tip：使用栈分配内存而不是堆分配内存

构造使用堆内存和使用栈内存，观察哪种方式性能更优

使用堆内存

#include <iostream>
#include <string.h>
#define size  100
#define loop_size 10000
static void heap(benchmark::State& state) {
  // Code inside this loop is measured repeatedly
  for (auto _ : state) {
    for(int i=0; i<loop_size; ++i){
      void* p = malloc(size);
      memset(p, 1, size);
      benchmark::DoNotOptimize(p);
      free(p);
    }
    state.SetBytesProcessed(loop_size);
  }
}
// Register the function as a benchmark
BENCHMARK(heap);

使用栈内存

static void stack(benchmark::State& state) {
  // Code inside this loop is measured repeatedly
  for (auto _ : state) {
    for(int i=0; i<loop_size; ++i){
      char p[size];
      memset(p, 1, size);
      benchmark::DoNotOptimize(p);
    }
    state.SetBytesProcessed(loop_size);
  }
}
// Register the function as a benchmark
BENCHMARK(stack);

使用栈内存性能是使用堆内存的6倍

cache-stack
分析：

堆内存分配需要调用malloc，有可能发生缺页中断，另外malloc在管理内存上，也会有一定的开销。
栈的内存是直接从当前基地址取一片地址，没有锁开销，没有上下文切换开销，而且cache line很可能就已经存在与CPU cache中，因此访问栈会比访问堆更cache亲和

Tip：当数据还在cache中，尽量重复使用，而不是多次加载

在循环一个数组时，是遍历数组一次做一件事？还是遍历一次数组做多件事效率高？

我们来看下面的实验

遍历两次数组，一次获取最大值，一次获取最小值

#include <iostream>
#include <string.h>
#include <algorithm> 
#include <vector>
#define size  10000
#define loop_size 10000
static void noshare(benchmark::State& state) {
  // Code inside this loop is measured repeatedly
  std::vector<int> a;
  a.resize(size);
  for(int i=0; i<size; ++i) {
    a[i] = i;
  }
  std::random_shuffle(a.begin(), a.end());
  int min = a[0];
  int max = a[0];
  for (auto _ : state) {
    for(int i=0; i<loop_size; ++i){
      for (int i = 0; i < size; i++) {
        max = std::max(a[i], max);
        benchmark::DoNotOptimize(max);
      }
      for (int i = 0; i < size; i++) {
        min = std::min(a[i], min);
        benchmark::DoNotOptimize(min);
      } 
    }
    state.SetBytesProcessed(loop_size);
  }
}
// Register the function as a benchmark
BENCHMARK(noshare);

遍历一次数组，同时获取最大值和最小值

static void share(benchmark::State& state) {
  // Code inside this loop is measured repeatedly
  std::vector<int> a;
  a.resize(size);
  for(int i=0; i<size; ++i) {
    a[i] = i;
  }
  std::random_shuffle(a.begin(), a.end());
  int min = a[0];
  int max = a[0];
  for (auto _ : state) {
    for(int i=0; i<loop_size; ++i){
      for (int i = 0; i < size; i++) {
        max = std::max(a[i], max);
        benchmark::DoNotOptimize(max);
        min = std::min(a[i], min);
        benchmark::DoNotOptimize(min);
      }
    }
    state.SetBytesProcessed(loop_size);
  }
}
// Register the function as a benchmark
BENCHMARK(share);

结果只遍历数组一次的性能是遍历两次的性能的2倍

cache-share
分析：

从时间复杂度来说，两片代码都是2n
但是从cache的角度去思考，却不一样，循环两次数组，会把数组的数组加载到cache两次，才能完成计算
而遍历数组一次，只需要加载数组到cache一次就完成两个操作
所以它的性能会更高

Tip：尽量少去写内存

写内存的次数会不会影响性能？

构造一个冒泡排序，每次比较都交换元素

#include <iostream>
#include <string.h>
#include <algorithm> 
#include <vector>
#define size  100
#define loop_size 1000
static void sort_slow(benchmark::State& state) {
  // Code inside this loop is measured repeatedly
  std::vector<int> b;
  b.resize(size);
  for(int i=0; i<size; ++i) {
    b[i] = i;
  }
  std::random_shuffle(b.begin(), b.end());
  for (auto _ : state) {
    for(int i=0; i<loop_size; ++i){
       std::vector<int> a = b;
       for (int i = 0; i < size; i++) {
        for (int j = i+1; j < size; j++) {
            if (a[j] < a[i]) {
                std::swap(a[j], a[i]);
            }
        }
      }
    }
    state.SetBytesProcessed(loop_size);
  }
}
// Register the function as a benchmark
BENCHMARK(sort_slow);

冒泡排序，找到最终位置后，才交换一次元素

static void sort_fast(benchmark::State& state) {
  // Code inside this loop is measured repeatedly
  std::vector<int> b;
  b.resize(size);
  for(int i=0; i<size; ++i) {
    b[i] = i;
  }
  std::random_shuffle(b.begin(), b.end());
  for (auto _ : state) {
    for(int i=0; i<loop_size; ++i){
       std::vector<int> a = b;
       for (int i = 0; i < size; i++) {
        int min = a[i];
        int min_index = i;
        for (int j = i+1; j < size; j++) {
            if (a[j] < min) {
                min = a[j];
                min_index = j;
            }
        }
        std::swap(a[i], a[min_index]);
     }
    }
    state.SetBytesProcessed(loop_size);
  }
}
// Register the function as a benchmark
BENCHMARK(sort_fast);

交换少的性能是频繁交换的性能的1.4倍

cache-write-memory
分析：

CPU写内存是消耗性能的，因为CPU写完cache，内存在cache中会标记为dirty，是需要刷新到主存才算完成

Tip：使用软件预取

当程序随机访问内存时，是否可以通过软件预取，来加快访问内存的速度呢？

朴素的二分查找

#include <iostream>
#include <string.h>
#include <algorithm> 
#include <vector>
#define size  100000000
#define loop_size 100000
#define index_size 100
int binarySearch(std::vector<int>& array, int number_of_elements, int key) {
  int low = 0, high = number_of_elements-1, mid;
  while(low <= high) {
    mid = (low + high)/2;
    if(array[mid] < key)
        low = mid + 1; 
    else if(array[mid] == key)
          return mid;
    else if(array[mid] > key)
          high = mid-1;
  }
}
static void no_prefetch(benchmark::State& state) {
  // Code inside this loop is measured repeatedly
  srand(1);
  std::vector<int> a;
  a.resize(size);
  for(unsigned int i=0; i<size; ++i) {
    a[i] = i;
  }
  for (auto _ : state) {
    for(unsigned int i=0; i<size; ++i){
      int ret = binarySearch(a, size, rand()%size);
      benchmark::DoNotOptimize(ret);
    }
    state.SetBytesProcessed(size);
  }
}
// Register the function as a benchmark
BENCHMARK(no_prefetch);

预取功能的二分查找

int prefetch_binarySearch(std::vector<int>& array, int number_of_elements, int key) {
  int low = 0, high = number_of_elements-1, mid;
    while(low <= high) {
        mid = (low + high)/2;
        // low path
        __builtin_prefetch (&array[(mid + 1 + high)/2], 0, 1);
        // high path
        __builtin_prefetch (&array[(low + mid - 1)/2], 0, 1);
        if(array[mid] < key)
            low = mid + 1; 
        else if(array[mid] == key)
             return mid;
        else if(array[mid] > key)
             high = mid-1;
        }
}
static void preFetch(benchmark::State& state) {
  // Code inside this loop is measured repeatedly
  srand(1);
  std::vector<int> a;
  a.resize(size);
  for(unsigned int i=0; i<size; ++i) {
    a[i] = i;
  }

  for (auto _ : state) {
    for(unsigned int i=0; i<size; ++i){
      int ret = prefetch_binarySearch(a, size, rand()%size);
      benchmark::DoNotOptimize(ret);
    }
    state.SetBytesProcessed(size);
  }
}
// Register the function as a benchmark
BENCHMARK(preFetch);

预取的二分查找是朴素的性能的1.4倍

cache-prefetch
分析：

由于访存是跳跃的，程序通过预先加载下一次循环需要用的内存到cache中，达到提速的效果