2
votes

Comment compter les occurrences de caractères à l'aide de SIMD

On me donne un tableau de caractères minuscules (jusqu'à 1,5 Go) et un caractère c. Et je veux trouver le nombre d'occurrences du caractère c en utilisant les instructions AVX.

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;

}


8 commentaires

Quel est votre format de caractère? ASCII ou une sorte d'Unicode?


Le format est ASCII


AVX1 ou AVX2? Qu'avez-vous essayé? Astuce: vérifiez _mm256_cmpeq_epi8 et _mm256_sub_epi8 pour la boucle la plus interne. Après 255 itérations, vous devez commencer à combiner deux octets en un seul uint16 , et ainsi de suite


AVX2, jusqu'à présent, j'ai un tableau de 4 variables __m256i et je pousse le débordement de l'index 0 à 3.


Pouvez-vous montrer un code de ce que vous avez fait? Pour ajouter des entiers 8 bits à des entiers plus grands, quelle est cette question: stackoverflow.com/questions/54541127


_mm256_cmpeq_epi8 vous donnera un -1 dans chaque octet. Si vous soustrayez cela d'un compteur (en utilisant _mm256_sub_epi8 ), vous pouvez compter directement jusqu'à 255 ou 128, c'est-à-dire que votre boucle la plus interne devrait simplement contenir ces deux intrinsèques.


@ Adamos2468: J'ai ajouté un code de somme horizontale efficace à la réponse de chtz, en utilisant vpsadbw contre zéro pour octet à qword. Et des mélanges efficaces de 256 bits à 64 bits comme Le moyen le plus rapide de faire une somme vectorielle flottante horizontale sur x86 .


Un cœur ne peut généralement pas saturer la bande passante DRAM, donc pour les entrées grandes , il peut être intéressant d'utiliser plusieurs threads (surtout si vous avez déjà un thread de travail démarré et que vous pouvez simplement lui envoyer un pointeur de fonction et des arguments). Vous avez tagué ce traitement parallèle , demandez-vous OpenMP ou quelque chose d'autre?


3 Réponses :


2
votes

Si vous n'insistez pas pour n'utiliser que les instructions SIMD, vous pouvez utiliser
de l'instruction VPMOVMSKB en combinaison avec l'instruction Instruction POPCNT . Le premier combine les bits les plus élevés de chaque octet dans un masque d'entier de 32 bits et le dernier compte les bits 1 dans cet entier (= le nombre de correspondances de caractères).

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) ))
      ) 
    );
  ...
}    

Je n'ai pas testé cette solution, mais vous devriez en comprendre l'essentiel.


4 commentaires

J'ai eu la même idée dans l'autre question d'OP, maintenant supprimée. GMTA.


Faire _mm256_movemask_epi8 et _mm_popcnt_u32 dans la boucle interne est beaucoup moins efficace que _mm256_sub_epi8


Je suppose. Mais c'est une alternative qui mérite d'être mentionnée pour sa simplicité.


Peut-être utile dans le cadre d'une boucle de nettoyage, ou pour un début / fin non aligné où vous décalez certains des bits avant de sauter, en utilisant un décompte de décalage calculé à partir du chevauchement . Sinon, la version "simple" la plus raisonnable consiste à mettre le psadbw epu8-> epu64 hsum dans la boucle interne et à utiliser _mm256_add_epi64 . C'est seulement 1 instruction supplémentaire par vecteur par rapport à la méthode efficace, contre 2 ( vpcmpeqb + vpmovmskb + popcnt + add vs vpcmpeqb (+ vpsadbw ) + vpsubb / q ).



3
votes

Je laisse intentionnellement de côté certaines parties, que vous devez déterminer vous-même (par exemple, gérer des longueurs qui ne sont pas un multiple de 4 * 255 * 32 octets), mais votre boucle la plus interne devrait ressembler à celui commençant par for (int i ...) :

_mm256_cmpeq_epi8 vous donnera un -1 dans chaque octet, ce que vous peut être utilisé comme un entier . Si vous soustrayez cela d'un compteur (en utilisant _mm256_sub_epi8 ), vous pouvez directement compter jusqu'à 255 ou 128. La boucle interne contient juste ces deux éléments intrinsèques. Vous devez arrêter et

#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;
}

Lien Godbolt: https: //godbolt.org/z/do5e3- (clang est légèrement meilleur que gcc pour dérouler la boucle la plus interne: gcc inclut des instructions vmovdqa inutiles qui gouleront le front-end si les données sont chaudes dans le cache L1d, ce qui nous empêche d'exécuter près de 2x charges de 32 octets par horloge)


5 commentaires

L'élargissement à epu64 peut et doit être fait avec _mm256_sad_epu8 (counter, _mm256_setzero_si256 ()) , puis _mm256_add_epi64 dans un vecteur que vous hsumez à la toute fin.


J'ai ajouté du code qui fait cela hsum, et une limite de taille de boucle externe. Notez que clang utilise des modes d'adressage indexés, il n'est donc pas plus proche que gcc de fonctionner à 2 charges par horloge sur Haswell / Skylake. :( Ils seront libérés de vpcmpeqb dans des uops séparés au cours de la phase d'émission. L'écrire avec les limites de la boucle comme comparaison de pointeur pourrait être un meilleur pari, et se faire entendre pour faire des incréments de pointeur purs au lieu de indexation idiote, par exemple const char * endp = min (buf + size, buf + 4 * 255 * 32) ou quelque chose.


Merci @PeterCordes d'avoir amélioré cela! Je suppose que pour gcc, il serait préférable de dérouler manuellement la boucle interne (c'est-à-dire de créer 4 variables au lieu d'un tableau). Belle astuce avec vpsadbw .


Ouais, cela pourrait aider GCC à éviter les instructions stupides de vmovdqa . Ça vaut le coup d'essayer si vous êtes curieux. Ou déposez un bug d'optimisation manquée; il est déjà optimisé pour tout stockage / rechargement du tableau de 4 vecteurs, et c'est clairement quelque chose qu'il devrait pouvoir optimiser. Quoi qu'il en soit, les -funroll-loops de gcc ne sont activés que manuellement ou dans le cadre de -fprofile-use ; pour les grandes bases de code, le déroulement de chaque boucle fait plus mal que cela n'aide, mais l'utilisation du profil identifiera les boucles chaudes et les déroulera. Je suppose que le déroulement permettrait également d'éviter des movdqa supplémentaires.


L'astuce vpsadbw est relativement bien connue pour hsumming des données 8 bits, et vaut la peine d'être utilisée même pour signé par XORing pour déplacer la plage, puis soustraire le biais 16 ou 32 * 128 à la fin. Je pense que le guide d'optimisation d'Agner Fog le mentionne, ou du moins sa bibliothèque VectorClass l'utilise.



1
votes

Probablement le plus rapide: memcount_avx2 et 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;
}

Benchmark skylake i7-6700 - 3,4 GHz - gcc 8.3:

memcount_avx2: 28 Go / s
memcount_sse: 23 Go / s
char_count_AVX2: 23 Go / s (à partir de post )


4 commentaires

Vous pouvez utiliser _mm256_sub_epi8 pour accumuler les résultats cmpeq au lieu de gaspiller des instructions dans la boucle externe. En outre, ce code est trop compact pour son propre bien et n'est pas correctement mis en retrait. (Peut-être un problème de tabulation vs espace dans la démarque SO?) Il m'a fallu un certain temps pour trouver où vous étiez à zéro acr0..3 entre les itérations de la boucle interne; serait plus logique de les déclarer à l'intérieur de la boucle externe. Je ne pense pas qu'il existe des compilateurs prenant en charge AVX2 mais pas C99. Je ferais également les calculs du pointeur de fin sur des lignes source distinctes.


Je ne vois pas ce que tu veux dire, en gaspillant des instructions


Si vous utilisez acr0 = _mm256_sub_epi8 (acr0, cmp (...)) alors la boucle externe peut simplement utiliser acr0 au lieu de _mm256_sub_epi8 (zv, acr0) < / code>. Utilisez x - = -1 au lieu de x + = -1 à l'intérieur de la boucle interne. Ce sous est une instruction gaspillée dans votre version.


Merci Peter, j'ai fait les changements et un nouveau benchmark