Utilizzo di thread OpenMP e std: :( ​​sperimentale: :) ​​simd per calcolare l'insieme di Mandelbrot

Aug 19 2020

Sto cercando di implementare un semplice plotter di Mandelbrot utilizzando diversi tipi di paradigmi HPC, mostrando i loro punti di forza e di debolezza e quanto siano facili o difficili le loro implementazioni. Pensa a GPGPU (CUDA / OpenACC / OpenMP4.5), threading / OpenMP e MPI. E usa questi esempi per dare ai programmatori nuovi di HPC un appiglio e per vedere quali sono le possibilità. La chiarezza del codice è più importante che ottenere le massime prestazioni dall'hardware, questo è il secondo passaggio;)

Poiché il problema è banale da parallelizzare e le moderne CPU possono ottenere un'enorme quantità di prestazioni utilizzando le istruzioni vettoriali, desidero anche combinare OpenMP e SIMD. Sfortunatamente, la semplice aggiunta di a #pragma omp simdnon produce risultati soddisfacenti e l'utilizzo di elementi intrinseci non è molto facile da usare o a prova di futuro. O carino .

Fortunatamente, si sta lavorando allo standard C ++ in modo tale che dovrebbe essere più facile implementare genericamente istruzioni vettoriali, come menzionato nel TS: "Estensioni per parallelismo, versione 2" , in particolare sezione 9 sui tipi paralleli di dati. Un'implementazione WIP può essere trovata qui , basata su VC che può essere trovata qui .

Supponi di avere la seguente classe (che è stata modificata per renderla un po 'più semplice)

#include <stddef.h>

using Range = std::pair<double, double>;
using Resolution = std::pair<std::size_t, std::size_t>;

class Mandelbrot
{
    double* d_iters;
    Range d_xrange;
    Range d_yrange;
    Resolution d_res;
    std::size_t d_maxIter;
    
public:
    Mandelbrot(Range xrange, Range yrange, Resolution res, std::size_t maxIter);
    ~Mandelbrot();

    void writeImage(std::string const& fileName);
    void computeMandelbrot();
private:
    void calculateColors();
}; 

E la seguente implementazione computeMandelbrot()dell'utilizzo di OpenMP

void Mandelbrot::computeMandelbrot()
{
    double dx = (d_xrange.second - d_xrange.first) / d_res.first;
    double dy = (d_yrange.second - d_yrange.first) / d_res.second;

    #pragma omp parallel for schedule(dynamic)
    for (std::size_t row = 0; row != d_res.second; ++row)
    {
        double c_imag = d_yrange.first + row * dy;
        for (std::size_t col = 0; col != d_res.first; ++col)
        {
            double real = 0.0;
            double imag = 0.0;
            double realSquared = 0.0;
            double imagSquared = 0.0;
            double c_real = d_xrange.first + col * dx;

            std::size_t iter = 0;
            while (iter < d_maxIter && realSquared + imagSquared < 4.0)
            {
                realSquared = real * real;
                imagSquared = imag * imag;
                imag = 2 * real * imag + c_imag;
                real = realSquared - imagSquared + c_real;
                ++iter;
            }
            d_iters[row * d_res.first + col] = iter;
        }   
    }
}

Possiamo presumere che le risoluzioni in entrambe le direzioni xey siano multipli di 2/4/8 / .., a seconda delle istruzioni SIMD che usiamo.

Sfortunatamente, le informazioni disponibili online su std::experimental::simd. Né esempi non banali per quanto ho potuto trovare.

Nel repository git di Vc, c'è un'implementazione del calcolatore di set di Mandelbrot, ma è piuttosto complicato e per la mancanza di commenti è piuttosto difficile da seguire.

È chiaro che dovrei cambiare i tipi di dati dei doppi nella funzione computeMandelbrot(), ma non sono sicuro di cosa. Il TS menziona due principali nuovi tipi di dati per alcuni tipi T,

native_simd = std::experimental::simd<T, std::experimental::simd_abi::native>;

e

fixed_size_simd = std::experimental::simd<T, std::experimental::simd_abi::fixed_size<N>>;

L'uso native_simdha più senso, dal momento che non conosco i miei limiti in fase di compilazione. Ma poi non mi è chiaro cosa rappresentino questi tipi, è native_simd<double>un singolo doppio o è una raccolta di doppi su cui viene eseguita un'istruzione vettoriale? E poi quanti doppi ci sono in questa collezione?

Se qualcuno potesse indicarmi esempi in cui vengono utilizzati questi concetti, o darmi alcuni suggerimenti su come implementare le istruzioni vettoriali usando std :: experiment :: simd, sarei molto grato.

Risposte

NigelOvermars Aug 24 2020 at 19:45

Ecco un'implementazione molto semplice, che funziona (per quanto ne so). La verifica di quali elementi del vettore hanno un valore assoluto maggiore di 2 viene eseguita in modo molto macchinoso e inefficiente. Deve esserci un modo migliore per farlo, ma non l'ho ancora trovato.

Ottengo un aumento delle prestazioni di circa il 72% su un AMD Ryzen 5 3600 e do a g ++ l'opzione -march=znver2, che è inferiore al previsto.

template <class T>
void mandelbrot(T xstart, T xend,
                T ystart, T yend)
{
    namespace stdx = std::experimental;

    constexpr auto simdSize = stdx::native_simd<T>().size();
    constexpr unsigned size = 4096;
    constexpr unsigned maxIter = 250;

    assert(size % simdSize == 0);
    unsigned* res = new unsigned[size * size];

    T dx = (xend - xstart) / size;
    T dy = (yend - ystart) / size;

    for (std::size_t row = 0; row != size; ++row)
    {
        T c_imag = ystart + row * dy;
        for (std::size_t col = 0; col != size; col += simdSize)
        {
            stdx::native_simd<T> real{0};
            stdx::native_simd<T> imag{0};
            stdx::native_simd<T> realSquared{0};
            stdx::native_simd<T> imagSquared{0};
            stdx::fixed_size_simd<unsigned, simdSize> iters{0};

            stdx::native_simd<T> c_real;
            for (int idx = 0; idx != simdSize; ++idx)
            {
                c_real[idx] = xstart + (col + idx) * dx;
            }

            for (unsigned iter = 0; iter != maxIter; ++iter)
            {
                realSquared = real * real;
                imagSquared = imag * imag;
                auto isInside = realSquared + imagSquared > stdx::native_simd<T>{4};
                for (int idx = 0; idx != simdSize; ++idx)
                {
                    // if not bigger than 4, increase iters
                    if (!isInside[idx])
                    {
                        iters[idx] += 1;
                    }
                    else
                    {
                        // prevent that they become inf/nan
                        real[idx] = static_cast<T>(4);
                        imag[idx] = static_cast<T>(4);
                    }
                }

                if (stdx::all_of(isInside) )
                {
                    break;
                }        

                imag = static_cast<T>(2.0) * real * imag + c_imag;
                real = realSquared - imagSquared + c_real;
            }
            iters.copy_to(res + row * size + col, stdx::element_aligned);
        }

    }
    delete[] res;
}

L'intero codice di test (a partire da auto test = (...)) viene compilato fino a

  .L9:
  vmulps ymm1, ymm1, ymm1
  vmulps ymm13, ymm2, ymm2
  xor eax, eax
  vaddps ymm2, ymm13, ymm1
  vcmpltps ymm2, ymm5, ymm2
  vmovaps YMMWORD PTR [rsp+160], ymm2
  jmp .L6
.L3:
  vmovss DWORD PTR [rsp+32+rax], xmm0
  vmovss DWORD PTR [rsp+64+rax], xmm0
  add rax, 4
  cmp rax, 32
  je .L22
.L6:
  vucomiss xmm3, DWORD PTR [rsp+160+rax]
  jp .L3
  jne .L3
  inc DWORD PTR [rsp+96+rax]
  add rax, 4
  cmp rax, 32
  jne .L6