Jak uzyskać wymierną korzyść z samoobsługi wstępnej?

Question

Używanie gcc 4.4.5 (tak ... wiem, że to stare) na x86_64. Ograniczone do instrukcji SSE2 (lub wcześniejszych) ze względu na kompatybilność.

Mam to, co moim zdaniem powinno być podręcznikiem dla uzyskania dużych korzyści z pobierania wstępnego. Mam tablicę ("A") elementów 32-bitowych, które nie są (i nie mogą być) w porządku sekwencyjnym. Te 32-bitowe elementy są indeksami w większej tablicy danych ("D") danych __m128i. Dla każdego elementu "A", muszę pobrać dane __m128i z odpowiedniej lokalizacji w "D", wykonać na nim operację i zapisać ją z powrotem w tej samej lokalizacji w "D". W rzeczywistości każdy "wpis" w D jest duży "SOME_CONST" __m128i. Jeśli więc wartość w A wynosi "1", indeksem do D jest D [1 * SOME_CONST].

Ponieważ kolejne elementy w "A" prawie nigdy nie wskazują na kolejne lokalizacje w "D", mam tendencję do myślenia, że ​​sprzętowy prefetcher będzie zmagał się lub nie osiągnął niczego użytecznego.

Mogę jednak łatwo przewidzieć, do których lokalizacji będę mieć dostęp, po prostu spoglądając przed siebie w "A". Dosyć słówka ... oto jakiś kod. Operacja wykonywana na danych polega na wzięciu niższych 64 bitów __m128i i sklonowaniu ich do wyższych 64 bitów tego samego. Pierwszy podstawowy pętli, bez fanaberii ...

// SOME_CONST is either 3 or 4, but this "operation" only needs to happen for 3 

for (i=0; i<arraySize; ++i) 
{ 
    register __m128i *dPtr = D + (A[i] * SOME_CONST); 
    dPtr[0] = _mm_shuffle_epi32(dPtr[0], 0 | (1<<2) | (0<<4) | (1<<6)); 
    dPtr[1] = _mm_shuffle_epi32(dPtr[1], 0 | (1<<2) | (0<<4) | (1<<6)); 
    dPtr[2] = _mm_shuffle_epi32(dPtr[2], 0 | (1<<2) | (0<<4) | (1<<6)); 

    // The immediate operand selects: 
    // Bits 0-31 = bits 0-31 
    // Bits 32-63 = bits 32-63 
    // Bits 64-95 = bits 0-31 
    // Bits 96-127 = bits 32-63 

    // If anyone is more clever than me and knows of a better way to do this in SSE2, 
    // bonus points. ;-) 
} 

Próbowałem wiele różnych sposobów, aby posypać instrinsics prefetch tam, i żaden z nich nie spowodowała jakichkolwiek SpeedUp jeszcze. Na przykład, próbowałem rozwinięciem pętli mieć kroku 2 lub nawet 4 elementów, ale to nie pomogło ...

// Assume the "A" array size is appropriately padded so that overruns don't 
// result in SIGSEGV accessing uninitialized memory in D. 

register __m128i *dPtr0, *dPtr1, *dPtr2, *dPtr3, *dPtr4, *dPtr5, *dPtr6, *dPtr7; 
dPtr4 = D + (A[0] * SOME_CONST); 
dPtr5 = D + (A[1] * SOME_CONST); 
dPtr6 = D + (A[2] * SOME_CONST); 
dPtr7 = D + (A[3] * SOME_CONST); 

for (i=0; i<arraySize; i+=4) 
{ 
    dPtr0 = dPtr4; 
    dPtr1 = dPtr5; 
    dPtr2 = dPtr6; 
    dPtr3 = dPtr7; 

    dPtr4 = D + (A[i+4] * SOME_CONST); 
    _mm_prefetch(dPtr4, _MM_HINT_NTA); 
    _mm_prefetch(dPtr4+1, _MM_HINT_NTA); // Is it needed? Tried both ways 

    dPtr5 = D + (A[i+5] * SOME_CONST); 
    _mm_prefetch(dPtr5, _MM_HINT_NTA); 
    _mm_prefetch(dPtr5+1, _MM_HINT_NTA); // Is it needed? Tried both ways 

    dPtr6 = D + (A[i+6] * SOME_CONST); 
    _mm_prefetch(dPtr6, _MM_HINT_NTA); 
    _mm_prefetch(dPtr6+1, _MM_HINT_NTA); // Is it needed? Tried both ways 

    dPtr7 = D + (A[i+7] * SOME_CONST); 
    _mm_prefetch(dPtr7, _MM_HINT_NTA); 
    _mm_prefetch(dPtr7+1, _MM_HINT_NTA); // Is it needed? Tried both ways 

    dPtr0[0] = _mm_shuffle_epi32(dPtr0[0], 0 | (1<<2) | (0<<4) | (1<<6)); 
    dPtr0[1] = _mm_shuffle_epi32(dPtr0[1], 0 | (1<<2) | (0<<4) | (1<<6)); 
    dPtr0[2] = _mm_shuffle_epi32(dPtr0[2], 0 | (1<<2) | (0<<4) | (1<<6)); 
    dPtr1[0] = _mm_shuffle_epi32(dPtr1[0], 0 | (1<<2) | (0<<4) | (1<<6)); 
    dPtr1[1] = _mm_shuffle_epi32(dPtr1[1], 0 | (1<<2) | (0<<4) | (1<<6)); 
    dPtr1[2] = _mm_shuffle_epi32(dPtr1[2], 0 | (1<<2) | (0<<4) | (1<<6)); 
    dPtr2[0] = _mm_shuffle_epi32(dPtr2[0], 0 | (1<<2) | (0<<4) | (1<<6)); 
    dPtr2[1] = _mm_shuffle_epi32(dPtr2[1], 0 | (1<<2) | (0<<4) | (1<<6)); 
    dPtr2[2] = _mm_shuffle_epi32(dPtr2[2], 0 | (1<<2) | (0<<4) | (1<<6)); 
    dPtr3[0] = _mm_shuffle_epi32(dPtr3[0], 0 | (1<<2) | (0<<4) | (1<<6)); 
    dPtr3[1] = _mm_shuffle_epi32(dPtr3[1], 0 | (1<<2) | (0<<4) | (1<<6)); 
    dPtr3[2] = _mm_shuffle_epi32(dPtr3[2], 0 | (1<<2) | (0<<4) | (1<<6)); 
} 

To jest wersja 4 elementów, ale Próbowałem też z tylko 2 w Sprawa, która zawierała za dużo danych. Próbowałem również używać _MM_HINT_NTA i _MM_HINT_T0. Jakoś nie ma żadnej zauważalnej różnicy.

Próbowałem też prostszy wariant, który właśnie próbuje się umieścić jak najwięcej miejsca wydawały się rozsądne pomiędzy pobierania wstępnego i kiedy go stosować:

#define PREFETCH_DISTANCE 10 
// trying 5 overnight, will see results tomorrow... 

for (i=0; i<arraySize; ++i) 
{ 
    register __m128i *dPtrFuture, *dPtr; 
    dPtrFuture = D + (A[i + PREFETCH_DISTANCE] * SOME_CONST); 
    _mm_prefetch(dPtrFuture, _MM_HINT_NTA);  // tried _MM_HINT_T0 too 
    _mm_prefetch(dPtrFuture + 1, _MM_HINT_NTA); // tried _MM_HINT_T0 too 

    dPtr = D + (A[i] * SOME_CONST); 
    dPtr[0] = _mm_shuffle_epi32(dPtr[0], 0 | (1<<2) | (0<<4) | (1<<6)); 
    dPtr[1] = _mm_shuffle_epi32(dPtr[1], 0 | (1<<2) | (0<<4) | (1<<6)); 
    dPtr[2] = _mm_shuffle_epi32(dPtr[2], 0 | (1<<2) | (0<<4) | (1<<6)); 
} 

Początkowo spodziewać ten kod do budy, ale po nim dostaje "PREFETCH_DISTANCE" w pętlę, miałem nadzieję, że zauważy wystarczająco dobre zwiększenie prędkości. Większość z tych wariantów powoduje nie więcej niż 0,2 sekundy różnicy czasu wykonywania w stosunku do milionów iteracji, które łączą czas procesora 4m: 30 s na tej konkretnej maszynie (która jest kamienna bezczynność inna niż ja). Różnice wydają się być nieodróżnialne od "szumu" w danych.

Czy mam rację, zakładając, że pobieranie wstępne powinno pomóc mi tutaj? Jeśli tak, co robię źle?

Wszystkie przydatne i interesujące myśli są mile widziane.

EDIT:

stworzyłem contrived przykład, że naprawdę losuje dane w A. Grałem z wielkości buforów z 64MB aż do 6400MB. Zauważyłem, że uzyskuję ogromne korzyści z rozwijania pętli i wstępnego obliczania adresów kolejnych 4 elementów podczas wykonywania operacji na bieżącym 4. Ale moje zbiorniki uruchomieniowe mają współczynnik> 10x, jeśli próbuję dokonać wstępnego pobrania którykolwiek z adresów, które wcześniej przeliczyłem. Naprawdę drapię się po tym. Moja samodzielna kod wymyślony jest:

#include <xmmintrin.h> 
#include <stdlib.h> 
#include <time.h> 
#include <stdio.h> 

#define QUEUE_ELEMENTS 1048576 
#define DATA_ELEMENT_SIZE 4 * sizeof(__m128i) 
#define DATA_ELEMENTS  QUEUE_ELEMENTS 

#define QUEUE_ITERATIONS 100000 
#define LOOP_UNROLL_4 
#define LOOP_UNROLL_2 

#ifdef LOOP_UNROLL_4 
    #define UNROLL_CONST 4 
#else 
    #ifdef LOOP_UNROLL_2 
    #define UNROLL_CONST 2 
    #else 
    #define UNROLL_CONST 1 
    #endif 
#endif 

int main(void) 
{ 
    unsigned long long randTemp; 
    unsigned long i, outerLoop; 
    unsigned long *workQueue; 
    __m128i *data, *dataOrig; 
    clock_t timeStamp; 

    workQueue = malloc(QUEUE_ELEMENTS * sizeof(unsigned long)); 

    dataOrig = malloc((DATA_ELEMENTS * DATA_ELEMENT_SIZE) + 2); 
    if ((unsigned long long) dataOrig & 0xf) 
    { 
    data = (__m128i *) (((unsigned long long) dataOrig & ~0xf) + 0x10); 
    // force 16-byte (128-bit) alignment 
    } else data = dataOrig; 

    // Not initializing data, because its contents isn't important. 

    for (i=0; i<QUEUE_ELEMENTS; ++i) 
    { 
    randTemp = (unsigned long long)rand() * 
    (unsigned long long) QUEUE_ELEMENTS/(unsigned long long) RAND_MAX; 
    workQueue[i] = (unsigned long) randTemp; 
    } 

    printf("Starting work...\n"); 
    // Actual work happening below... start counting. 
    timeStamp = clock(); 

    for (outerLoop = 0; outerLoop < QUEUE_ITERATIONS; ++outerLoop) 
    { 
    register __m128i *dataPtr0, *dataPtr1, *dataPtr2, *dataPtr3; 
    register __m128i *dataPtr4, *dataPtr5, *dataPtr6, *dataPtr7; 

    #ifdef LOOP_UNROLL_2 
     dataPtr4 = data + (workQueue[0] * DATA_ELEMENT_SIZE); 
     dataPtr5 = data + (workQueue[1] * DATA_ELEMENT_SIZE); 
    #endif 
    #ifdef LOOP_UNROLL_4 
     dataPtr6 = data + (workQueue[2] * DATA_ELEMENT_SIZE); 
     dataPtr7 = data + (workQueue[3] * DATA_ELEMENT_SIZE); 
    #endif 

    for (i=0; i<QUEUE_ELEMENTS; i+=UNROLL_CONST) 
    { 
     #ifdef LOOP_UNROLL_2 
     dataPtr0 = dataPtr4; 
     dataPtr4 = data + (workQueue[i+4] * DATA_ELEMENT_SIZE); 
     // _mm_prefetch(dataPtr4, _MM_HINT_T0); 
     dataPtr1 = dataPtr5; 
     dataPtr5 = data + (workQueue[i+5] * DATA_ELEMENT_SIZE); 
     // _mm_prefetch(dataPtr5, _MM_HINT_T0); 
     #endif 
     #ifdef LOOP_UNROLL_4 
     dataPtr2 = dataPtr6; 
     dataPtr6 = data + (workQueue[i+6] * DATA_ELEMENT_SIZE); 
     // _mm_prefetch(dataPtr6, _MM_HINT_T0); 
     dataPtr3 = dataPtr7; 
     dataPtr7 = data + (workQueue[i+7] * DATA_ELEMENT_SIZE); 
     // _mm_prefetch(dataPtr7, _MM_HINT_T0); 
     #endif 
     #if !defined(LOOP_UNROLL_2) && !defined(LOOP_UNROLL_4) 
     dataPtr0 = data + (workQueue[i] * DATA_ELEMENT_SIZE); 
     #endif 

     _mm_shuffle_epi32(dataPtr0[0], 0 | (1<<2) | (0<<4) | (1<<6)); 
     _mm_shuffle_epi32(dataPtr0[1], 0 | (1<<2) | (0<<4) | (1<<6)); 
     _mm_shuffle_epi32(dataPtr0[2], 0 | (1<<2) | (0<<4) | (1<<6)); 
     // Per original code, no need to perform operation on dataPtrx[3] 

     #ifdef LOOP_UNROLL_2 
     _mm_shuffle_epi32(dataPtr1[0], 0 | (1<<2) | (0<<4) | (1<<6)); 
     _mm_shuffle_epi32(dataPtr1[1], 0 | (1<<2) | (0<<4) | (1<<6)); 
     _mm_shuffle_epi32(dataPtr1[2], 0 | (1<<2) | (0<<4) | (1<<6)); 
     #endif 
     #ifdef LOOP_UNROLL_4 
     _mm_shuffle_epi32(dataPtr2[0], 0 | (1<<2) | (0<<4) | (1<<6)); 
     _mm_shuffle_epi32(dataPtr2[1], 0 | (1<<2) | (0<<4) | (1<<6)); 
     _mm_shuffle_epi32(dataPtr2[2], 0 | (1<<2) | (0<<4) | (1<<6)); 
     _mm_shuffle_epi32(dataPtr3[0], 0 | (1<<2) | (0<<4) | (1<<6)); 
     _mm_shuffle_epi32(dataPtr3[1], 0 | (1<<2) | (0<<4) | (1<<6)); 
     _mm_shuffle_epi32(dataPtr3[2], 0 | (1<<2) | (0<<4) | (1<<6)); 
     #endif 
    } 
    if ((outerLoop % 1000) == 0) { putchar('.'); fflush(stdout); } 
    } 

    timeStamp = clock() - timeStamp; 
    printf("\nRun was %lu seconds.\n", timeStamp/CLOCKS_PER_SEC); 

    free(dataOrig); 
    free(workQueue); 

    return 0; 
} 

• mój czas pracy z LOOP_UNROLL_2 wynosi 40 sekund.
• Moje środowisko wykonawcze z LOOP_UNROLL_4 wynosi 20 sekund.
• Moje środowisko uruchomieniowe bez rozwijania wynosi 80 sekund.
• Po włączeniu funkcji pobierania wstępnego środowisko wykonawcze jest tak długie, że po prostu zabijam proces, zamiast czekać na niego.

Napisałem nawet 8x rozwijaną pętlę i nadal skalowałem ją idealnie do 10 sekund czasu pracy. Jestem zdumiony, że nie nasycił się tam, ponieważ w tym momencie na pewno skończyłbym się rejestrom, mając 16 unikalnych wskaźników. Czego mogę się z tego nauczyć? Że mój wewnętrzny kod pętli jest tak ciasny, że jest mocno przyćmiony przez narzut w konstrukcie pętli? Czy są jakieś błędy w tym kodzie, których nie widzę? Wszystkie kompilacje były z gcc -O2.

Answer 1

Jeśli dane znajdują się w pamięci, nie należy oczekiwać zbyt dużej prędkości; wstępne pobieranie z pamięci ma niewiele do poprawienia.

Z czasem cyklu 150 ns, 64 bajtowe linie pamięci podręcznej i szybkość transmisji danych 4 GB/s , system pobiera 320 MB użytecznych danych na sekundę. Prefetching przybliża tempo do maksimum 4000 MB/s. Całkowite oszczędności dostępne dla wstępnego pobierania wynoszą 0,92 sekundy na każde 320 MB odczytu. Przy 320 MB/s, 270 sekund (4m 30s) to czas transferu pamięci wynoszący 840 GB; progran prawdopodobnie nie wydaje więcej niż niewielki ułamek tego (< 1%, 8GB) na faktyczne czytanie pamięci. Całkowite wyeliminowanie pamięci i/o pozwoli zaoszczędzić ... 1% czasu pracy.

Wadą zbyt dużego wstępnego pobierania jest to, że dane z preselekcją wypierają przydatne rzeczy z naprawdę szybkiej, ale naprawdę małej pamięci, blisko procesora (pamięci podręczne poziomu 1 i poziomu 2, kilobajty, a nie megabajty). Może to odpowiadać za pesymalną wydajność niektórych testów.

To rozwinięcie pętli przyspieszyło program, ale wstępne pobieranie nie sugerowało również, że samo przetwarzanie jest podstawowym wąskim gardłem, a nie przepływem danych.