Usando threads OpenMP e std: :( ​​experimental: :) simd para calcular o conjunto de Mandelbrot

Question

Estou procurando implementar um plotter de conjunto Mandelbrot simples usando diferentes tipos de paradigmas de HPC, mostrando seus pontos fortes e fracos e como suas implementações são fáceis ou difíceis. Pense em GPGPU (CUDA / OpenACC / OpenMP4.5), threading / OpenMP e MPI. E use esses exemplos para dar aos programadores novos em HPC um apoio para as mãos e ver quais são as possibilidades. Clareza de código é mais importante do que obter o melhor desempenho absoluto do hardware, essa é a segunda etapa;)

Como o problema de paralelização é trivial e as CPUs modernas podem ganhar muito desempenho usando instruções vetoriais, também quero combinar OpenMP e SIMD. Infelizmente, simplesmente adicionar um #pragma omp simdnão produz resultados satisfatórios e usar intrínsecos não é muito amigável ou à prova de futuro. Ou bonita .

Felizmente, o padrão C ++ está sendo feito de forma que seja mais fácil implementar genericamente as instruções vetoriais, conforme mencionado no TS: "Extensões para paralelismo, versão 2" , especificamente a seção 9 sobre tipos de dados paralelos. Uma implementação WIP pode ser encontrada aqui , que é baseada no VC, que pode ser encontrada aqui .

Suponha que eu tenha a seguinte classe (que foi alterada para torná-la um pouco mais simples)

#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 a seguinte implementação de computeMandelbrot()usar 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;
        }   
    }
}

Podemos supor que as resoluções nas direções xey são múltiplos de 2/4/8 / .., dependendo de quais instruções SIMD usamos.

Infelizmente, há muito pouca informação disponível online em std::experimental::simd. Nem quaisquer exemplos não triviais, até onde pude encontrar.

No repositório Vc git, existe uma implementação da calculadora de conjuntos Mandelbrot, mas é bastante complicada e devido à falta de comentários um tanto difícil de seguir.

É claro que devo alterar os tipos de dados das duplas na função computeMandelbrot(), mas não tenho certeza do quê. O TS menciona dois novos tipos de dados principais para algum tipo 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>>;

Usar native_simdfaz mais sentido, já que não sei meus limites em tempo de compilação. Mas então não está claro para mim o que esses tipos representam, é um native_simd<double>único duplo ou uma coleção de duplos em que uma instrução vetorial é executada? E quantas duplas há nesta coleção?

Se alguém pudesse me apontar para exemplos onde esses conceitos são usados, ou me dar algumas dicas sobre como implementar instruções vetoriais usando std :: experimental :: simd, eu ficaria muito grato.

Answer 1

Aqui está uma implementação muito básica, que funciona (pelo que posso dizer). O teste de quais elementos do vetor têm valor absoluto maior que 2 é feito de uma forma muito complicada e ineficiente. Deve haver uma maneira melhor de fazer isso, mas ainda não a encontrei.

Obtenho um aumento de desempenho de cerca de 72% em um AMD Ryzen 5 3600 e dando a opção g ++ -march=znver2, que é menor do que o esperado.

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

Todo o código de teste (começando de auto test = (...)) compila até

  .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