_mm256_store_ps() 函数是原子的吗？在与 openmp 一起使用时

Question

我正在尝试创建一个使用英特尔 AVX 技术并执行向量乘法和加法的简单程序。在这里，我正在使用 Open MP。但由于函数调用 _mm256_store_ps() 导致分段错误。

我已经尝试过使用 OpenMP 的原子特性，如 atomic、critical 等，如果这个函数本质上是原子的，并且多个内核试图同时执行，但它不起作用。

#include<stdio.h>
#include<time.h>
#include<stdlib.h>
#include<immintrin.h>
#include<omp.h>
#define N 64

__m256 multiply_and_add_intel(__m256 a, __m256 b, __m256 c) {
  return _mm256_add_ps(_mm256_mul_ps(a, b),c);
}

void multiply_and_add_intel_total_omp(const float* a, const float* b, const float* c, float* d)
{
  __m256 a_intel, b_intel, c_intel, d_intel;
  #pragma omp parallel for private(a_intel,b_intel,c_intel,d_intel)
  for(long i=0; i<N; i=i+8) {
    a_intel = _mm256_loadu_ps(&a[i]);
    b_intel = _mm256_loadu_ps(&b[i]);
    c_intel = _mm256_loadu_ps(&c[i]);
    d_intel = multiply_and_add_intel(a_intel, b_intel, c_intel);
    _mm256_store_ps(&d[i],d_intel);
  }
}
int main()
{
    srand(time(NULL));
    float * a = (float *) malloc(sizeof(float) * N);
    float * b = (float *) malloc(sizeof(float) * N);
    float * c = (float *) malloc(sizeof(float) * N);
    float * d_intel_avx_omp = (float *)malloc(sizeof(float) * N);
    int i;
    for(i=0;i<N;i++)
    {
        a[i] = (float)(rand()%10);
        b[i] = (float)(rand()%10);
        c[i] = (float)(rand()%10);
    }
    double time_t = omp_get_wtime();
    multiply_and_add_intel_total_omp(a,b,c,d_intel_avx_omp);
    time_t = omp_get_wtime() - time_t;
    printf("\nTime taken to calculate with AVX2 and OMP : %0.5lf\n",time_t);
  }

  free(a);
  free(b);
  free(c);
  free(d_intel_avx_omp);
    return 0;
}

我希望我会得到 d = a * b + c 但它显示分段错误。我试图在没有 OpenMP 的情况下执行相同的任务，并且它可以正常工作。如果有任何兼容性问题或我缺少任何部分，请告诉我。

• gcc 版本 7.3.0
• 英特尔® 酷睿™ i3-3110M 处理器
• 操作系统 Ubuntu 18.04
• 打开 MP 4.5，我执行了命令$ echo |cpp -fopenmp -dM |grep -i open，它显示#define _OPENMP 201511
• 编译命令，gcc first_int.c -mavx -fopenmp

** 更新 **

根据讨论和建议，新代码是，

 float * a = (float *) aligned_alloc(N, sizeof(float) * N);
 float * b = (float *) aligned_alloc(N, sizeof(float) * N);
 float * c = (float *) aligned_alloc(N, sizeof(float) * N);
 float * d_intel_avx_omp = (float *)aligned_alloc(N, sizeof(float) * N);

这工作不完美。

请注意，我试图比较一般计算、avx 计算和 avx+openmp 计算。这是我得到的结果，

• 不使用 AVX 的计算时间：0.00037
• 用 AVX 计算所用时间：0.00024
• 使用 AVX 和 OMP 计算所需的时间：0.00019

N = 50000

Answer 1

_mm256_store_ps 的文档说：

将 a 中的 256 位（由 8 个打包的单精度（32 位）浮点元素组成）存储到内存中。 mem_addr 必须在 32 字节边界上对齐，否则可能会生成一般保护异常。

您可以使用_mm256_storeu_si256 代替未对齐的商店。

---

更好的选择是在 32 字节边界上对齐所有数组（对于 256 位 avx 寄存器）并使用对齐的加载和存储来获得最佳性能，因为未对齐的加载/存储跨越缓存线边界会导致性能损失。

使用std::aligned_alloc（或C11 aligned_alloc、memalign、posix_memalign，无论你有什么可用的）而不是malloc(size)，例如：

float* allocate_aligned(size_t n) {
    constexpr size_t alignment = alignof(__m256);
    return static_cast<float*>(aligned_alloc(alignment, sizeof(float) * n));
}
// ...
float* a = allocate_aligned(N);
float* b = allocate_aligned(N);
float* c = allocate_aligned(N);
float* d_intel_avx_omp = allocate_aligned(N);

在 C++-17 中 new 可以对齐分配：

float* allocate_aligned(size_t n) {
    constexpr auto alignment = std::align_val_t{alignof(__m256)};
    return new(alignment) float[n];
}

---

或者，使用 Vc: portable, zero-overhead C++ types for explicitly data-parallel programming 为您对齐堆分配的 SIMD 向量：

#include <cstdio>
#include <memory>
#include <chrono>
#include <Vc/Vc>

Vc::float_v random_float_v() {
    alignas(Vc::VectorAlignment) float t[Vc::float_v::Size];
    for(unsigned i = 0; i < Vc::float_v::Size; ++i)
        t[i] = std::rand() % 10;
    return Vc::float_v(t, Vc::Aligned);
}

unsigned reverse_crc32(void const* vbegin, void const* vend) {
    unsigned const* begin = reinterpret_cast<unsigned const*>(vbegin);
    unsigned const* end = reinterpret_cast<unsigned const*>(vend);
    unsigned r = 0;
    while(begin != end)
        r = __builtin_ia32_crc32si(r, *--end);
    return r;
}

int main() {
    constexpr size_t N = 65536;
    constexpr size_t M = N / Vc::float_v::Size;

    std::unique_ptr<Vc::float_v[]> a(new Vc::float_v[M]);
    std::unique_ptr<Vc::float_v[]> b(new Vc::float_v[M]);
    std::unique_ptr<Vc::float_v[]> c(new Vc::float_v[M]);
    std::unique_ptr<Vc::float_v[]> d_intel_avx_omp(new Vc::float_v[M]);

    for(unsigned i = 0; i < M; ++i) {
        a[i] = random_float_v();
        b[i] = random_float_v();
        c[i] = random_float_v();
    }

    auto t0 = std::chrono::high_resolution_clock::now();
    for(unsigned i = 0; i < M; ++i)
        d_intel_avx_omp[i] = a[i] * b[i] + c[i];
    auto t1 = std::chrono::high_resolution_clock::now();

    double seconds = std::chrono::duration_cast<std::chrono::duration<double>>(t1 - t0).count();
    unsigned crc = reverse_crc32(d_intel_avx_omp.get(), d_intel_avx_omp.get() + M); // Make sure d_intel_avx_omp isn't optimized out.
    std::printf("crc: %u, time: %.09f seconds\n", crc, seconds);
}

并行版本：

#include <tbb/parallel_for.h>
// ...
    auto t0 = std::chrono::high_resolution_clock::now();
    tbb::parallel_for(size_t{0}, M, [&](unsigned i) {
        d_intel_avx_omp[i] = a[i] * b[i] + c[i];
    });
    auto t1 = std::chrono::high_resolution_clock::now();

Answer 2

您必须为这些内在函数使用对齐的内存。将您的 malloc(...) 更改为 aligned_alloc(sizeof(float) * 8, ...) (C11)。

这与原子完全无关。您正在处理完全独立的数据片段（即使在不同的缓存行上），因此不需要任何保护。