c - 如何从 AVX 内在函数中获得用于计算基本统计数据的性能提升？

Question

我的问题是关于使用 AVX 指令与朴素方法的性能。

我从我的 AVX 方法中得到的答案与我从天真的方法中得到的答案相同且正确，但使用 AVX 指令得到答案所需的时间稍长，所以我想知道我做错了什么/向量化代码效率低下。

这个问题有点太复杂了，无法提供独立的可编译代码单元，对此我深表歉意。但是，我在下面有一些功能性代码片段，我希望它们相当简单明了且样式得体，希望它们足够容易理解以处理手头的问题。

一些环境细节:

• 这两种方法都是使用 Clang(Apple LLVM 版本 8.1.0 (clang-802.0.42))编译的。
• 我正在使用 -mavx 标志进行编译。
• 我的硬件(配备 Intel Core i7 处理器的 MacBook Pro)声称支持 AVX 指令。

我有一个程序，用户提供一个多行文本文件，每行包含一个逗号分隔的数字字符串，即 n 维 vector 列表，其中 n 对于文件是任意的，但是(除非输入错误)对于每一行都是相同的值 n。

例如:

0,4,6,1,2,22,0,2,30,...,39,14,0,3,3,3,1,3,0,3,2,1
0,0,1,1,0,0,0,8,0,1,...,6,0,0,4,0,0,0,0,7,0,8,2,0
...
1,0,1,0,1,0,0,2,0,1,...,2,0,0,0,0,0,2,1,1,0,2,0,0

我通过比较这些 vector 生成一些统计分数，例如 Pearson 相关性，但分数函数可以是任何东西，比如简单的东西，例如算术平均值。

朴素的方法

这些 vector 中的每一个都被放入一个指向名为 signal_t 的结构的指针中:

typedef struct signal {
    uint32_t n;
    score_t* data;
    score_t mean;
} signal_t;

score_t 类型只是 float 的类型定义:

typedef float score_t;

首先，我将字符串解析为 float (score_t) 值并计算算术平均值:

signal_t* s = NULL;
s = malloc(sizeof(signal_t));
if (!s) {
    fprintf(stderr, "Error: Could not allocate space for signal pointer!\n");
    exit(EXIT_FAILURE);
}
s->n = 1;
s->data = NULL;
s->mean = NAN;

for (uint32_t idx = 0; idx < strlen(vector_string); idx++) {
    if (vector_string[idx] == ',') {
        s->n++;
    }
}

s->data = malloc(sizeof(*s->data) * s->n);
if (!s->data) {
    fprintf(stderr, "Error: Could not allocate space for signal data pointer!\n");
    exit(EXIT_FAILURE);
}
char* start = vector_string;
char* end = vector_string;
char entry_buf[ENTRY_MAX_LEN];
uint32_t entry_idx = 0;
bool finished_parsing = false;
bool data_contains_nan = false;
do {
    end = strchr(start, ',');
    if (!end) {
        end = vector_string + strlen(vector_string);
        finished_parsing = true;
    }
    memcpy(entry_buf, start, end - start);
    entry_buf[end - start] = '\0';
    sscanf(entry_buf, "%f", &s->data[entry_idx++]);
    if (isnan(s->data[entry_idx - 1])) {
        data_contains_nan = true;
    }
    start = end + 1;
} while (!finished_parsing);

if (!data_contains_nan) {
    s->mean = pt_mean_signal(s->data, s->n);
}

算术平均值非常简单:

score_t pt_mean_signal(score_t* d, uint32_t len)
{
    score_t s = 0.0f;
    for (uint32_t idx = 0; idx < len; idx++) {
        s += d[idx];
    }
    return s / len;
}

朴素的表现

在 10k vector 字符串的文件上运行这种方法，我得到了 6.58 秒的运行时间。

AVX 方法

我有一个名为 signal_avx_t 的修改过的 signal_t 结构:

typedef struct signal_avx {
    uint32_t n_raw;
    uint32_t n;
    __m256* data;
    score_t mean;
} signal_avx_t;

这存储指向 __m256 地址的指针。每个 __m256 存储八个单精度 float 值。为了方便起见，我定义了一个名为 AVX_FLOAT_N 的常量来存储这个倍数，例如:

#define AVX_FLOAT_N 8

下面是我如何解析 vector 字符串并将其存储在 __m256 中。这与原始方法非常相似，除了现在我一次将八个值读入缓冲区，将缓冲区写入 __m256，然后重复，直到没有更多值可写。然后我计算平均值:

signal_avx_t* s = NULL;
s = malloc(sizeof(signal_avx_t));
if (!s) {
fprintf(stderr, "Error: Could not allocate space for signal_avx pointer!\n");
    exit(EXIT_FAILURE);
}
s->n_raw = 1;
s->n = 0;
s->data = NULL;
s->mean = NAN;

for (uint32_t idx = 0; idx < strlen(vector_string); idx++) {
    if (vector_string[idx] == ',') {
        s->n_raw++;
    }
}

score_t signal_buf[AVX_FLOAT_N];

s->n = (uint32_t) ceil((float)(s->n_raw) / AVX_FLOAT_N);
s->data = malloc(sizeof(*s->data) * s->n);
if (!s->data) {
    fprintf(stderr, "Error: Could not allocate space for signal_avx data pointer!\n");
    exit(EXIT_FAILURE);
}
char* start = id;
char* end = id;
char entry_buf[ENTRY_MAX_LEN];
uint32_t entry_idx = 0;
uint32_t data_idx = 0;
bool finished_parsing = false;
bool data_contains_nan = false;

do {
    end = strchr(start, ',');
    if (!end) {
        end = vector_string + strlen(vector_string);
        finished_parsing = true;
    }
    memcpy(entry_buf, start, end - start);
    entry_buf[end - start] = '\0';
    sscanf(entry_buf, "%f", &signal_buf[entry_idx++ % AVX_FLOAT_N]);
    if (isnan(signal_buf[(entry_idx - 1) % AVX_FLOAT_N])) {
        data_contains_nan = true;
    }
    start = end + 1;

    /* I write every eight floats to an __m256 chunk of memory */
    if (entry_idx % AVX_FLOAT_N == 0) {
        s->data[data_idx++] = _mm256_setr_ps(signal_buf[0],
                                             signal_buf[1],
                                             signal_buf[2],
                                             signal_buf[3],
                                             signal_buf[4],
                                             signal_buf[5],
                                             signal_buf[6],
                                             signal_buf[7]);
    }
} while (!finished_parsing);

if (!data_contains_nan) {
    /* write any leftover floats to the last `__m256` */
    if (entry_idx % AVX_FLOAT_N != 0) {
        for (uint32_t idx = entry_idx % AVX_FLOAT_N; idx < AVX_FLOAT_N; idx++) {
            signal_buf[idx] = 0;
        }
        s->data[data_idx++] = _mm256_setr_ps(signal_buf[0],
                                             signal_buf[1],
                                             signal_buf[2],
                                             signal_buf[3],
                                             signal_buf[4],
                                             signal_buf[5],
                                             signal_buf[6],
                                             signal_buf[7]);
    }
    s->mean = pt_mean_signal_avx(s->data, s->n, s->n_raw);
}

AVX 均值函数

这是我编写的用于生成算术平均值的函数:

score_t pt_mean_signal_avx(__m256* d, uint32_t len, uint32_t len_raw)
{
    score_t s = 0.0f;
    /* initialize a zero-value vector to collect summed value */
    __m256 v_sum = _mm256_setzero_ps();
    /* add data to collector */
    for (uint32_t idx = 0; idx < len; idx++) {
        v_sum = _mm256_add_ps(v_sum, d[idx]);
    }
    /* sum the collector values */
    score_t* res = (score_t*)&v_sum;
    for (uint32_t idx = 0; idx < AVX_FLOAT_N; idx++) {
        s += res[idx];
    }
    return s / len_raw;
}

AVX 性能

在 10k vector 字符串文件上运行基于 AVX 的方法时，我得到了 6.86 秒的运行时间，大约慢了 5%。无论输入的大小如何，这种差异大致恒定。

总结

我的预期是，通过使用 AVX 指令和矢量化循环，我会得到一个减速带，而不是性能会略微变差。

代码片段中是否有任何暗示滥用 __m256 数据类型和相关内部函数以计算基本汇总统计信息的内容？

主要是，在处理较大数据集之间更复杂的评分函数之前，我想弄清楚我在这里做错了什么。感谢您提供任何建设性建议!

Answer 1

首先，我希望我们同意，将文本解析为 float 可能比算术平均更占用 CPU，更不用说从物理存储上的文件中读取数据了。如果你打算做一个基准测试，你绝对应该省略读取和解析。

这里的主要问题似乎是您试图在阅读时进行矢量化。实际上，您所做的是将数据从 signal_buf 复制到 s。

您必须意识到 __mm256_* 并不是真正的内存数据类型。它只是一个宏，用于确保您使用的内存地址和寄存器具有 256 位值的能力。

所以，只需将您的 signal_buf 和 __mm256_load_ps 加载到 SIMD 寄存器中，然后在其上执行您的 AVX 魔法，或者依次填充 s 直接使用 sscanf 然后执行相同的 __mm256_load_ps 魔术。

我真的不明白你为什么要使用 setr .为什么需要颠倒算术平均值的元素顺序？或者这是您的“穷人的负载指令”？

同样，您在 float 学方面的努力，尤其是如果您编写编译器甚至可能能够自动矢量化的代码，并不是在这里花费时间的原因。就是字符串的解析。

VOLK ( vector 优化内核库)有很多手写的 SIMD 内核，包括一个累积 float 数组的内核:

https://github.com/gnuradio/volk/blob/master/kernels/volk/volk_32f_accumulator_s32f.h

AVX 代码如下所示:

static inline void
volk_32f_accumulator_s32f_a_avx(float* result, const float* inputBuffer, unsigned int num_points)
{
  float returnValue = 0;
  unsigned int number = 0;
  const unsigned int eighthPoints = num_points / 8;

  const float* aPtr = inputBuffer;
  __VOLK_ATTR_ALIGNED(32) float tempBuffer[8];

  __m256 accumulator = _mm256_setzero_ps();
  __m256 aVal = _mm256_setzero_ps();

  for(;number < eighthPoints; number++){
    aVal = _mm256_load_ps(aPtr);
    accumulator = _mm256_add_ps(accumulator, aVal);
    aPtr += 8;
  }

  _mm256_store_ps(tempBuffer, accumulator);

  returnValue = tempBuffer[0];
  returnValue += tempBuffer[1];
  returnValue += tempBuffer[2];
  returnValue += tempBuffer[3];
  returnValue += tempBuffer[4];
  returnValue += tempBuffer[5];
  returnValue += tempBuffer[6];
  returnValue += tempBuffer[7];

  number = eighthPoints * 8;
  for(;number < num_points; number++){
    returnValue += (*aPtr++);
  }
  *result = returnValue;
}

它的作用是拥有一个八个元素累加器，它不断地向其中添加八个新元素的集合(分别)，然后最后返回这八个累加器的总和。