跳转至

指针访存优化

1.__restrict优化

// 定义一个函数 func,该函数接受三个参数:
// 1. 指向整数数组的指针 a
// 2. 指向整数的指针 b
// 3. 整数 n,表示数组 a 的元素数量
// 函数的功能是将指针 b 所指向的值赋给数组 a 的每个元素
void func(int* a, int* b, int n) {
  for (int i = 0; i < n; ++i) {
    a[i] = *b;
  }
}

// 定义一个函数 func_restrict,该函数与 func 功能相同,但使用了 __restrict
// 关键字
// __restrict
// 关键字是一个类型限定符,用于告诉编译器,指针所指向的内存区域不会被其他指针访问
// 这可以帮助编译器和cpu消除冗余加载、循环重排序与并行化、简化存储转发、减少依赖检查的额外代码
void func_restrict(int* __restrict a, int* __restrict b, int n) {
  for (int i = 0; i < n; ++i) {
    a[i] = *b;
  }
}

int main() {
  int a[10];
  int b;

  func(a, &b, 10);
  func_restrict(a, &b, 10);

  return 0;
}

2.指针访存优化

整数索引遍历

const int N = 1024;  // 数组大小
const int K = 1000;  // 测试轮次
int p[N], q[N];

// 1. 生成随机排列p
std::iota(p, p + N, 0);
std::random_shuffle(p, p + N);

// 2. 构造单循环排列q
int k = p[N - 1];
for (int i = 0; i < N; i++) {
  k = q[k] = p[i];  // 构建循环链
}

// 3. 索引遍历(核心操作)
for (int t = 0; t < K; t++) {
  for (int i = 0; i < N; i++) {
    k = q[k];  // 需计算地址:q基地址 + k*4
  }
}
  • 特点:4字节索引,内存占用小,地址计算需乘法(k*4)

64位指针遍历

struct Node {
  Node* ptr;  // 8字节指针
};
Node q[N];

// 1. 初始化指针链
Node* k = &q[p[N - 1]];
for (int i = 0; i < N; i++) {
  k->ptr = &q[p[i]];
  k = k->ptr;
}

// 2. 指针遍历(核心操作)
for (int t = 0; t < K; t++) {
  for (int i = 0; i < N; i++) {
    k = k->ptr;  // 直接解引用,无地址计算
  }
}
  • 特点:8字节指针,内存占用翻倍,省去地址计算
  • 优势:L1命中时利用CPU零延迟转发,当 CPU 发现一个操作(比如 mov)的目的地是另一个操作的源地址时,它可以走一条“绿色通道”,直接将数据转发过去,省去了写回寄存器再读出的开销。
  • 劣势:数据量超L1后缓存命中率骤降,性能恶化

32位指针兼容方案

// 编译为32位程序或使用uint32_t模拟
struct Node32 {
  uint32_t ptr;  // 32位指针(实际为偏移量)
};
Node32 q[N];
uint32_t k = (uint32_t)&q[p[N - 1]];  // 基地址偏移

// 遍历核心操作
for (int t = 0; t < K; t++) {
  for (int i = 0; i < N; i++) {
    k = ((Node32*)k)->ptr;  // 32位地址直接跳转
  }
}
  • 特点:4字节指针,内存占用与整数索引相同
  • 性能:L1命中时接近64位指针,L2/L3命中时优于64位指针

24位紧凑索引(手动位运算)

const int MASK_24 = (1 << 24) - 1;  // 24位掩码
uint8_t q[N * 3];                   // 3字节/元素(紧凑存储)

// 1. 初始化24位索引链(每个元素占3字节)
int k = p[N - 1] * 3;  // 初始偏移(3字节步长)
for (int i = 0; i < N; i++) {
  int next = p[i] * 3;
  // 写入24位值(分3字节存储)
  q[k] = next & 0xFF;
  q[k + 1] = (next >> 8) & 0xFF;
  q[k + 2] = (next >> 16) & 0xFF;
  k = next;
}

// 2. 遍历核心操作(手动解析24位)
k = p[N - 1] * 3;  // 重置起始偏移
for (int t = 0; t < K; t++) {
  for (int i = 0; i < N; i++) {
    // 手动读取3字节并拼接为24位值
    int next_val = q[k] | (q[k + 1] << 8) | (q[k + 2] << 16);
    k = next_val & MASK_24;  // 确保24位范围
  }
}
  • 特点:3字节索引,内存占用比4字节方案减少25%
  • 优势:L2/L3缓存中可容纳更多元素,命中率提升
  • 限制:最大支持2^24(约1600万)元素

64位复合索引(交易所级系统)

// 64位复合索引:客户(32位) + 会员(12位) + 合约(20位)
struct CompositeIndex {
  uint64_t client : 32;    // 客户ID(0~2^32-1)
  uint64_t member : 12;    // 会员ID(0~4095)
  uint64_t contract : 20;  // 合约ID(0~100万)
};

// 转换为整数存储
uint64_t encode(CompositeIndex idx) {
  return (idx.client) | (idx.member << 32) | (idx.contract << 44);
}

// 从整数解析
CompositeIndex decode(uint64_t val) {
  return {.client = val & 0xFFFFFFFF,
          .member = (val >> 32) & 0xFFF,
          .contract = (val >> 44) & 0xFFFFF};
}

// 索引使用示例
uint64_t idx = encode({12345, 67, 890});
CompositeIndex info = decode(idx);
  • 特点:64位存储多维度信息,无额外内存开销
  • 适用场景:百万级热点数据(如持仓索引),需多维度查询

3.指针压缩

/*
 * 指针压缩库
 * 提供指针到偏移量的压缩和解压缩功能,减少内存使用
 */
#ifndef PTR_COMPRESS_H
#define PTR_COMPRESS_H

#include <cstddef>
#include <cstdint>
#include <cstring>
#include <type_traits>

namespace compress {

// 错误码定义
enum class CompressError {
  SUCCESS = 0,                // 成功
  INVALID_PARAMETER = -1,     // 无效参数
  OUT_OF_RANGE = -2,          // 超出范围
  COMPRESSION_ERROR = -3,     // 压缩错误
  DECOMPRESSION_ERROR = -4,   // 解压缩错误
  UNSUPPORTED_OPERATION = -5  // 不支持的操作
};

// 计算前导零数量(64位)
inline uint32_t count_leading_zeros_64(uint64_t value) {
  if (value == 0) return 64;

  uint32_t count = 0;
  while ((value & (1ULL << 63)) == 0) {
    value <<= 1;
    ++count;
  }
  return count;
}

// 计算尾随零数量(64位)
inline uint32_t count_trailing_zeros_64(uint64_t value) {
  if (value == 0) return 64;

  uint32_t count = 0;
  while ((value & 1) == 0) {
    value >>= 1;
    ++count;
  }
  return count;
}

// 计算表示范围内指针所需的位数
constexpr uint32_t bits_needed_for_range(size_t memory_length) {
  if (memory_length < 2) return 1;
  // 计算 (memory_length - 1) 的最高有效位位置
  uint64_t val = memory_length - 1;
  uint32_t bits = 0;
  while (val > 0) {
    val >>= 1;
    ++bits;
  }
  return bits;
}

// 从对齐值计算位移
constexpr uint32_t bit_shift_from_alignment(size_t alignment) {
  if (alignment == 0) return 0;
  return count_trailing_zeros_64(alignment);
}

// 检查是否可以使用16位压缩
constexpr bool can_compress_16_shift(size_t memory_length, size_t alignment) {
  uint32_t needed_bits = bits_needed_for_range(memory_length);
  uint32_t shift_bits = bit_shift_from_alignment(alignment);
  return (needed_bits > shift_bits) ? (needed_bits - shift_bits) <= 16
                                    : needed_bits <= 16;
}

// 检查是否可以使用32位压缩
constexpr bool can_compress_32_shift(size_t memory_length, size_t alignment) {
  uint32_t needed_bits = bits_needed_for_range(memory_length);
  uint32_t shift_bits = bit_shift_from_alignment(alignment);
  return (needed_bits > shift_bits) ? (needed_bits - shift_bits) <= 32
                                    : needed_bits <= 32;
}

// 32位指针压缩函数
template <typename T>
CompressError compress_ptr_32_shift(T* base_ptr, const T* const* src_table,
                                    uint32_t* dest_table, size_t count,
                                    uint8_t bit_shift) {
  if (!base_ptr || !src_table || !dest_table || count == 0) {
    return CompressError::INVALID_PARAMETER;  // 错误:参数无效
  }

  uintptr_t base_addr = reinterpret_cast<uintptr_t>(base_ptr);

  for (size_t i = 0; i < count; ++i) {
    if (!src_table[i]) {
      dest_table[i] = 0;  // 空指针压缩为0
      continue;
    }

    uintptr_t ptr_addr = reinterpret_cast<uintptr_t>(src_table[i]);
    if (ptr_addr < base_addr) {
      return CompressError::OUT_OF_RANGE;  // 错误:指针地址在基地址之前
    }

    uintptr_t offset = ptr_addr - base_addr;
    offset >>= bit_shift;  // 右移对齐位

    if (offset > UINT32_MAX) {
      return CompressError::COMPRESSION_ERROR;  // 错误:偏移量超出32位范围
    }

    dest_table[i] = static_cast<uint32_t>(offset);
  }

  return CompressError::SUCCESS;
}

// 32位指针解压缩函数
template <typename T>
CompressError decompress_ptr_32_shift(T* base_ptr, const uint32_t* src_table,
                                      T** dest_table, size_t count,
                                      uint8_t bit_shift) {
  if (!base_ptr || !src_table || !dest_table || count == 0) {
    return CompressError::INVALID_PARAMETER;  // 错误:参数无效
  }

  uintptr_t base_addr = reinterpret_cast<uintptr_t>(base_ptr);

  for (size_t i = 0; i < count; ++i) {
    if (src_table[i] == 0) {
      dest_table[i] = nullptr;  // 0解压缩为空指针
      continue;
    }

    uintptr_t offset = static_cast<uintptr_t>(src_table[i]);
    offset <<= bit_shift;  // 左移恢复对齐位

    uintptr_t ptr_addr = base_addr + offset;
    dest_table[i] = reinterpret_cast<T*>(ptr_addr);
  }

  return CompressError::SUCCESS;
}

// 16位指针压缩函数
template <typename T>
CompressError compress_ptr_16_shift(T* base_ptr, const T* const* src_table,
                                    uint16_t* dest_table, size_t count,
                                    uint8_t bit_shift) {
  if (!base_ptr || !src_table || !dest_table || count == 0) {
    return CompressError::INVALID_PARAMETER;  // 错误:参数无效
  }

  uintptr_t base_addr = reinterpret_cast<uintptr_t>(base_ptr);

  for (size_t i = 0; i < count; ++i) {
    if (!src_table[i]) {
      dest_table[i] = 0;  // 空指针压缩为0
      continue;
    }

    uintptr_t ptr_addr = reinterpret_cast<uintptr_t>(src_table[i]);
    if (ptr_addr < base_addr) {
      return CompressError::OUT_OF_RANGE;  // 错误:指针地址在基地址之前
    }

    uintptr_t offset = ptr_addr - base_addr;
    offset >>= bit_shift;  // 右移对齐位

    if (offset > UINT16_MAX) {
      return CompressError::COMPRESSION_ERROR;  // 错误:偏移量超出16位范围
    }

    dest_table[i] = static_cast<uint16_t>(offset);
  }

  return CompressError::SUCCESS;
}

// 16位指针解压缩函数
template <typename T>
CompressError decompress_ptr_16_shift(T* base_ptr, const uint16_t* src_table,
                                      T** dest_table, size_t count,
                                      uint8_t bit_shift) {
  if (!base_ptr || !src_table || !dest_table || count == 0) {
    return CompressError::INVALID_PARAMETER;  // 错误:参数无效
  }

  uintptr_t base_addr = reinterpret_cast<uintptr_t>(base_ptr);

  for (size_t i = 0; i < count; ++i) {
    if (src_table[i] == 0) {
      dest_table[i] = nullptr;  // 0解压缩为空指针
      continue;
    }

    uintptr_t offset = static_cast<uintptr_t>(src_table[i]);
    offset <<= bit_shift;  // 左移恢复对齐位

    uintptr_t ptr_addr = base_addr + offset;
    dest_table[i] = reinterpret_cast<T*>(ptr_addr);
  }

  return CompressError::SUCCESS;
}

// 指针压缩管理器类
template <typename T>
class PtrCompressor {
 private:
  T* base_ptr_;
  size_t memory_size_;
  size_t alignment_;
  uint8_t bit_shift_;
  bool use_16bit_;

 public:
  PtrCompressor()
      : base_ptr_(nullptr),
        memory_size_(0),
        alignment_(1),
        bit_shift_(0),
        use_16bit_(false) {}

  // 初始化压缩器
  CompressError initialize(T* base_ptr, size_t memory_size,
                           size_t alignment = 1) {
    if (!base_ptr || memory_size == 0) {
      return CompressError::INVALID_PARAMETER;  // 错误:参数无效
    }

    base_ptr_ = base_ptr;
    memory_size_ = memory_size;
    alignment_ = alignment;
    bit_shift_ = bit_shift_from_alignment(alignment);

    // 决定使用16位还是32位压缩
    use_16bit_ = can_compress_16_shift(memory_size, alignment);

    return CompressError::SUCCESS;
  }

  // 压缩指针数组
  CompressError compress(const T* const* src_ptrs, void* dest_buffer,
                         size_t count) {
    if (!src_ptrs || !dest_buffer || count == 0 || !base_ptr_) {
      return CompressError::INVALID_PARAMETER;  // 错误:参数无效
    }

    if (use_16bit_) {
      uint16_t* dest_16 = static_cast<uint16_t*>(dest_buffer);
      return compress_ptr_16_shift(base_ptr_, src_ptrs, dest_16, count,
                                   bit_shift_);
    } else {
      uint32_t* dest_32 = static_cast<uint32_t*>(dest_buffer);
      return compress_ptr_32_shift(base_ptr_, src_ptrs, dest_32, count,
                                   bit_shift_);
    }
  }

  // 解压缩指针数组
  CompressError decompress(const void* src_buffer, T** dest_ptrs,
                           size_t count) {
    if (!src_buffer || !dest_ptrs || count == 0 || !base_ptr_) {
      return CompressError::INVALID_PARAMETER;  // 错误:参数无效
    }

    if (use_16bit_) {
      const uint16_t* src_16 = static_cast<const uint16_t*>(src_buffer);
      return decompress_ptr_16_shift(base_ptr_, src_16, dest_ptrs, count,
                                     bit_shift_);
    } else {
      const uint32_t* src_32 = static_cast<const uint32_t*>(src_buffer);
      return decompress_ptr_32_shift(base_ptr_, src_32, dest_ptrs, count,
                                     bit_shift_);
    }
  }

  // 获取压缩后的字节大小
  size_t get_compressed_size(size_t count) const {
    return use_16bit_ ? count * sizeof(uint16_t) : count * sizeof(uint32_t);
  }

  // 检查是否使用16位压缩
  bool uses_16bit_compression() const { return use_16bit_; }

  // 获取位移值
  uint8_t get_bit_shift() const { return bit_shift_; }

  // 获取基地址
  T* get_base_ptr() const { return base_ptr_; }

  // 获取内存大小
  size_t get_memory_size() const { return memory_size_; }

  // 获取对齐值
  size_t get_alignment() const { return alignment_; }

  // 获取错误信息
  static const char* error_to_string(CompressError error) {
    switch (error) {
      case CompressError::SUCCESS:
        return "Success";
      case CompressError::INVALID_PARAMETER:
        return "Invalid parameter";
      case CompressError::OUT_OF_RANGE:
        return "Out of range";
      case CompressError::COMPRESSION_ERROR:
        return "Compression error";
      case CompressError::DECOMPRESSION_ERROR:
        return "Decompression error";
      case CompressError::UNSUPPORTED_OPERATION:
        return "Unsupported operation";
      default:
        return "Unknown error";
    }
  }
};

}  // namespace compress

#endif  // PTR_COMPRESS_H

4.Non-Temporal写大数据块

使用CPU提供的特殊指令,而不是依赖memset或普通的赋值语句。

#include <immintrin.h>
// 假设我们要用AVX2指令清零一个数组 (每次处理256位/32字节)

const __m256i zeros = _mm256_set1_epi32(0);  // 全零向量
for (int i = 0; i + 7 < N; i += 8) {         // 每次处理8个int (32字节)
  // 注意这里的函数名:_mm256_stream_si256
  _mm256_stream_si256((__m256i*)&a[i], zeros);
}
// 处理末尾不足8个的部分 (省略)

// 关键在于 _mm256_stream_si256
// 这个intrinsic函数。它告诉CPU执行一次Non-Temporal写操作。
// 它允许我们执行“流式”写操作(Streaming
// Stores),数据直接写入内存,尽量不干扰缓存。