如何用SIMD快速掩码分隔符后的字节？求高效实现方案

我有如下格式的字符串数据：

uint8_t data[] = "mykeyxyz:1234\nky:123\n...";

每行都是key:value格式，且确定len(key) <= 16。我需要把mykeyxyz这类Key加载到__m128i中，并将高位填充为0。

目前最简单的方法是用一个由255或0组成的掩码数组，但这需要额外的内存加载。有没有更快的实现方式？

已采纳方案的效果 #

已采纳的方案使整体程序耗时降低约2%，可对比测试1brc_valid13.cpp与1brc_valid14.cpp（后者采用该方案）。硬件环境：AMD 2950X，Ubuntu 18.04，g++ 11.4，编译命令：

g++ -o main 1brc_final_valid.cpp -O3 -std=c++17 -march=native -m64 -lpthread

补充说明 #

• 补充说明1：最好不使用AVX512
• 补充说明2：我需要变量len来开始解析value部分
• 补充说明3：该函数会在循环中调用（例如解析100万行文本），但strcmp_mask基本会一直处于L1缓存中
• 补充说明4：我通过解析10亿行(key,value)数据并处理来对函数进行基准测试
• 补充说明5：我测试了maskafterc256，发现它使我的代码运行速度慢了50倍！！！如果将_mm256_set_epi8替换为_mm256_setr_epi8，速度会慢500倍以上（耗时太长我直接按Ctrl-C终止了）。我不确定_mm256_set_epi8的作用，因为它被翻译成了一系列指令而非单条指令。

性能测试结果 #

maskafterc的perf stat -d ./main结果 #

14,470.46 msec task-clock                #   20.785 CPUs utilized
             3,032      context-switches          #    0.210 K/sec
                 5      cpu-migrations            #    0.000 K/sec
           341,894      page-faults               #    0.024 M/sec
    55,073,525,723      cycles                    #    3.806 GHz                      (37.19%)
     1,714,679,575      stalled-cycles-frontend   #    3.11% frontend cycles idle     (36.71%)
    11,442,758,125      stalled-cycles-backend    #   20.78% backend cycles idle      (36.92%)
    80,739,874,133      instructions              #    1.47  insn per cycle
                                                  #    0.14  stalled cycles per insn  (37.39%)
     8,661,529,181      branches                  #  598.566 M/sec                    (38.22%)
        39,299,214      branch-misses             #    0.45% of all branches          (38.13%)
    17,784,400,757      L1-dcache-loads           # 1229.015 M/sec                    (37.93%)
       827,509,870      L1-dcache-load-misses     #    4.65% of all L1-dcache hits    (37.52%)
   <not supported>      LLC-loads
   <not supported>      LLC-load-misses

       0.696207306 seconds time elapsed

      12.918590000 seconds user
       1.546535000 seconds sys

maskafterc256的perf stat -d ./main结果 #

Performance counter stats for './main':

      1,047,414.73 msec task-clock                #   29.982 CPUs utilized
           125,296      context-switches          #    0.120 K/sec
               211      cpu-migrations            #    0.000 K/sec
           341,889      page-faults               #    0.326 K/sec
 4,229,832,527,830      cycles                    #    4.038 GHz                      (37.50%)
    10,965,796,240      stalled-cycles-frontend   #    0.26% frontend cycles idle     (37.50%)
   167,711,051,408      stalled-cycles-backend    #    3.96% backend cycles idle      (37.49%)
   296,573,918,148      instructions              #    0.07  insn per cycle
                                                  #    0.57  stalled cycles per insn  (37.50%)
    44,843,867,352      branches                  #   42.814 M/sec                    (37.50%)
        56,509,334      branch-misses             #    0.13% of all branches          (37.51%)
    91,621,829,978      L1-dcache-loads           #   87.474 M/sec                    (37.50%)
    18,776,996,709      L1-dcache-load-misses     #   20.49% of all L1-dcache hits    (37.51%)
   <not supported>      LLC-loads
   <not supported>      LLC-load-misses

      34.935225940 seconds time elapsed

    1039.609651000 seconds user
       6.774492000 seconds sys

示例代码 #

#include <iostream>
#include <immintrin.h>
#include <string>
#include <cstring>
using namespace std;

alignas(4096) const uint8_t strcmp_mask[32] = {
  255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
  0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
};

int main()
{
  uint8_t data[] = "mykeyxyz:1234\naaaaaaaaaaa";
  __m128i chars = _mm_loadu_si128((__m128i*)data);
  __m128i separators = _mm_set1_epi8(':');
  __m128i compared = _mm_cmpeq_epi8(chars, separators);
  uint32_t separator_mask = _mm_movemask_epi8(compared);
  uint32_t len = __builtin_ctz(separator_mask);
  
  cout << "len = " << len << "\n";
  __m128i mask = _mm_loadu_si128((__m128i*)(strcmp_mask + 16 - len));
  __m128i key_chars = _mm_and_si128(chars, mask);
  
  uint8_t res[16];
  memcpy(res, (char*)&key_chars, 16);
  for (int i = 0; i < 16; i++) cout << int(res[i]) << " ";
  cout << "\n";
}
// len = 8
// 109 121 107 101 121 120 121 122 0 0 0 0 0 0 0 0

内容的提问来源于stack exchange，提问作者Huy Le