如何使用 SIMD 计算字符出现次数

Question

我得到一个小写字符数组（最大 1.5Gb）和一个字符 c。我想使用 AVX 指令找出字符 c 出现了多少次。

unsigned long long char_count_AVX2(char * vector, int size, char c){
unsigned long long sum =0;
int i, j;
const int con=3;
__m256i ans[con];
for(i=0; i<con; i++)
    ans[i]=_mm256_setzero_si256();

__m256i Zer=_mm256_setzero_si256();
__m256i C=_mm256_set1_epi8(c);
__m256i Assos=_mm256_set1_epi8(0x01);
__m256i FF=_mm256_set1_epi8(0xFF);
__m256i shield=_mm256_set1_epi8(0xFF);
__m256i temp;
int couter=0;
for(i=0; i<size; i+=32){
    couter++;
    shield=_mm256_xor_si256(_mm256_cmpeq_epi8(ans[0], Zer), FF);
    temp=_mm256_cmpeq_epi8(C, *((__m256i*)(vector+i)));
    temp=_mm256_xor_si256(temp, FF);
    temp=_mm256_add_epi8(temp, Assos);
    ans[0]=_mm256_add_epi8(temp, ans[0]);
    for(j=1; j<con; j++){
        temp=_mm256_cmpeq_epi8(ans[j-1], Zer);
        shield=_mm256_and_si256(shield, temp);
        temp=_mm256_xor_si256(shield, FF);
        temp=_mm256_add_epi8(temp, Assos);
        ans[j]=_mm256_add_epi8(temp, ans[j]);
    }
}
for(j=con-1; j>=0; j--){
    sum<<=8;
    unsigned char *ptr = (unsigned char*)&(ans[j]);
    for(i=0; i<32; i++){
        sum+=*(ptr+i);
    }
}
return sum;

}

Answer 1

如果你不坚持只使用 SIMD 指令，你可以使用
VPMOVMSKB 指令与 POPCNT 指令的组合。前者将每个字节的最高位组合成一个 32 位整数掩码，后者计算此整数中的 1 位（=char 匹配的计数）。

int couter=0;
for(i=0; i<size; i+=32) {
  ...
  couter += 
    _mm_popcnt_u32( 
      (unsigned int)_mm256_movemask_epi8( 
        _mm256_cmpeq_epi8( C, *((__m256i*)(vector+i) ))
      ) 
    );
  ...
}    

我没有测试过这个解决方案，但你应该明白要点。

Answer 2

我故意省略了一些你需要自己弄清楚的部分（例如处理不是4*255*32字节倍数的长度），但你最内部的循环应该看起来像以@开头的循环987654323@:

_mm256_cmpeq_epi8 将在每个字节中为您提供 -1，您可以将其用作整数。如果你从计数器中减去它（使用_mm256_sub_epi8），你可以直接数到 255 或 128。内部循环只包含这两个内在函数。你必须停下来，

#include <immintrin.h>
#include <stdint.h>

static inline
__m256i hsum_epu8_epu64(__m256i v) {
    return _mm256_sad_epu8(v, _mm256_setzero_si256());  // SAD against zero is a handy trick
}

static inline
uint64_t hsum_epu64_scalar(__m256i v) {
    __m128i lo = _mm256_castsi256_si128(v);
    __m128i hi = _mm256_extracti128_si256(v, 1);
    __m128i sum2x64 = _mm_add_epi64(lo, hi);   // narrow to 128

    hi = _mm_unpackhi_epi64(sum2x64, sum2x64);
    __m128i sum = _mm_add_epi64(hi, sum2x64);  // narrow to 64
    return _mm_cvtsi128_si64(sum);
}


unsigned long long char_count_AVX2(char const* vector, size_t size, char c)
{
    __m256i C=_mm256_set1_epi8(c);

    // todo: count elements and increment `vector` until it is aligned to 256bits (=32 bytes)
    __m256i const * simd_vector = (__m256i const *) vector;
     // *simd_vector is an alignment-required load, unlike _mm256_loadu_si256()

    __m256i sum64 = _mm256_setzero_si256();
    size_t unrolled_size_limit = size - 4*255*32 + 1;
    for(size_t k=0; k<unrolled_size_limit ; k+=4*255*32) // outer loop: TODO
    {
        __m256i counter[4]; // multiple counter registers to hide latencies
        for(int j=0; j<4; j++)
            counter[j]=_mm256_setzero_si256();
        // inner loop: make sure that you don't go beyond the data you can read
        for(int i=0; i<255; ++i)
        {   // or limit this inner loop to ~22 to avoid branch mispredicts
            for(int j=0; j<4; ++j)
            {
                counter[j]=_mm256_sub_epi8(counter[j],           // count -= 0 or -1
                                           _mm256_cmpeq_epi8(*simd_vector, C));
                ++simd_vector;
            }
        }

        // only need one outer accumulator: OoO exec hides the latency of adding into it
        sum64 = _mm256_add_epi64(sum64, hsum_epu8_epu64(counter[0]));
        sum64 = _mm256_add_epi64(sum64, hsum_epu8_epu64(counter[1]));
        sum64 = _mm256_add_epi64(sum64, hsum_epu8_epu64(counter[2]));
        sum64 = _mm256_add_epi64(sum64, hsum_epu8_epu64(counter[3]));
    }

    uint64_t sum = hsum_epu64_scalar(sum64);

    // TODO add up remaining bytes with sum.
    // Including a rolled-up vector loop before going scalar
    //  because we're potentially a *long* way from the end

    // Maybe put some logic into the main loop to shorten the 255 inner iterations
    // if we're close to the end.  A little bit of scalar work there shouldn't hurt every 255 iters.

    return sum;
}

Godbolt 链接：https://godbolt.org/z/do5e3-（clang 在展开最内层循环方面略胜于 gcc：gcc 包含一些无用的 vmovdqa 指令，如果数据在 L1d 缓存中很热，这些指令将成为前端的瓶颈，阻止我们每个时钟运行接近 2 次 32 字节负载）

Answer 3

可能是最快的：memcount_avx2 和 memcount_sse2

size_t memcount_avx2(const void *s, int c, size_t n) 
{    
  __m256i cv = _mm256_set1_epi8(c), 
          zv = _mm256_setzero_si256(), 
         sum = zv, acr0,acr1,acr2,acr3;
  const char *p,*pe;    

  for(p = s; p != (char *)s+(n- (n % (252*32)));) 
  { 
    for(acr0 = acr1 = acr2 = acr3 = zv, pe = p+252*32; p != pe; p += 128) 
    {
      acr0 = _mm256_sub_epi8(acr0, _mm256_cmpeq_epi8(cv, _mm256_lddqu_si256((const __m256i *)p))); 
      acr1 = _mm256_sub_epi8(acr1, _mm256_cmpeq_epi8(cv, _mm256_lddqu_si256((const __m256i *)(p+32)))); 
      acr2 = _mm256_sub_epi8(acr2, _mm256_cmpeq_epi8(cv, _mm256_lddqu_si256((const __m256i *)(p+64)))); 
      acr3 = _mm256_sub_epi8(acr3, _mm256_cmpeq_epi8(cv, _mm256_lddqu_si256((const __m256i *)(p+96)))); 
      __builtin_prefetch(p+1024);
    }
    sum = _mm256_add_epi64(sum, _mm256_sad_epu8(acr0, zv));
    sum = _mm256_add_epi64(sum, _mm256_sad_epu8(acr1, zv));
    sum = _mm256_add_epi64(sum, _mm256_sad_epu8(acr2, zv));
    sum = _mm256_add_epi64(sum, _mm256_sad_epu8(acr3, zv));
  } 

  for(acr0 = zv; p+32 < (char *)s + n; p += 32)  
    acr0 = _mm256_sub_epi8(acr0, _mm256_cmpeq_epi8(cv, _mm256_lddqu_si256((const __m256i *)p))); 
  sum = _mm256_add_epi64(sum, _mm256_sad_epu8(acr0, zv));

  size_t count = _mm256_extract_epi64(sum, 0) 
               + _mm256_extract_epi64(sum, 1) 
               + _mm256_extract_epi64(sum, 2) 
               + _mm256_extract_epi64(sum, 3);  

  while(p != (char *)s + n) 
      count += *p++ == c;
  return count;
}

基准 skylake i7-6700 - 3.4GHz - gcc 8.3：

memcount_avx2：28 GB/s
memcount_sse：23 GB/秒
char_count_AVX2：23 GB/s（来自post）