【问题标题】:How to get performance boost from AVX intrinsics for calculating basic statistics?如何从 AVX 内在函数中获得性能提升以计算基本统计信息?
【发布时间】:2017-11-29 20:16:54
【问题描述】:

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

我从我的 AVX 方法中得到的答案与我从幼稚的方法中得到的答案相同,而且是正确的,但是使用 AVX 指令得到的答案需要稍长一些,所以我想知道我做错了什么/使用矢量化代码效率低下。

这个问题有点太复杂,无法提供一个独立的可编译代码单元,对此我很抱歉。但是,我在下面有功能性代码 sn-ps,我希望它们相当简单且风格得体,希望它们很容易理解以处理手头的问题。

一些环境细节:

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

我有一个程序,其中用户提供了一个多行文本文件,每行包含一个逗号分隔的数字字符串,即 n 维向量列表,其中 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

我从这些向量的比较中生成一些统计分数,例如 Pearson 相关性,但分数函数可以是任何东西,比如说,简单的东西,比如算术平均值。

天真的方法

这些向量中的每一个都被放入一个指向名为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 矢量字符串的文件上运行这种方法时,我得到了 6.58 秒的运行时间。

AVX 方法

我有一个修改后的 signal_t 结构,名为 signal_avx_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

这是我解析向量字符串并将其存储在__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 矢量字符串的文件上运行基于 AVX 的方法时,我得到了 6.86 秒的运行时间,大约慢了 5%。无论输入的大小如何,这种差异大致是恒定的。

总结

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

代码 sn-ps 中是否有任何内容表明滥用 __m256 数据类型和相关的内在函数以计算基本汇总统计?

主要是,我想弄清楚我在这里做错了什么,然后再在更大的数据集之间进行更复杂的评分函数。感谢您的任何建设性建议!

【问题讨论】:

  • 我知道这很难做到,但是如果你准备了一个包含main() 的实际可编译的 C 代码,我很乐意运行一些分析,看看时间在哪里实际花费。
  • (不过,明天要出去睡一晚)
  • 嗨,Marcus,感谢您提供审查代码。我在这里发布了一个可以在 CentOS 7 (gcc 4.8.5) 和 Mac OS X 10.12 (clang via Xcode) 下编译的 Github 项目。你可以克隆它,然后make all 运行测试:github.com/alexpreynolds/simd-pearson-test
  • Clang 可能会自动矢量化您的标量版本,至少如果您使用 -ffast-math-O2 或更高版本。它可能比您手动矢量化的代码做得更好。 (例如,clang 通常使用多个向量累加器展开循环,在这种情况下隐藏了 FP add 的延迟)。查看两个版本的热循环的 asm。

标签: c clang sse simd avx


【解决方案1】:

首先,我希望我们同意将文本解析为浮点数可能比算术平均值更占用 CPU,更不用说从物理存储上的文件中读取数据了。如果您要进行基准测试,则绝对应该省略读取和解析。

这里的主要问题似乎是您在阅读时试图成为和矢量化。你在现实中所做的是从signal_bufs 的不必要的数据副本。

您必须意识到 __mm256_* 并不是真正的内存数据类型。它只是一个宏,可确保您使用的内存地址和寄存器支持 256 位值。

因此,只需将您的 signal_buf__mm256_load_ps 加载到 SIMD 寄存器中,然后对其执行 AVX 魔术,或者直接使用 sscanf 顺序填充 s,然后执行相同的操作 @987654331 @魔术。

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

同样,您的浮点数学工作,特别是如果您编写的代码甚至可以让编译器自动矢量化,这并不是您需要花费时间的事情。就是解析字符串。

VOLK(内核向量优化库)有很多手写的 SIMD 内核,包括一个累积浮点数组的内核:

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

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

【讨论】:

  • 相关性或其他参数成对评分函数可以依赖于正确顺序的元素以获得正确的分值,因此使用 setr 对浮点数进行正确排序。
  • @AlexReynolds 但同样,在这种情况下,请在执行非对齐/缓存密集的工作(解析然后将值存储在数组中)时进行洗牌,而不是在无关的洗牌中。诀窍是在内存中尽可能保持线性——如果不可避免的话,将所有的洗牌都放在一个地方。
  • @AlexReynolds:您似乎正在使用_mm256_setr_ps(d[0], d[1], ...) 创建一个向量,其中元素的顺序与它们在内存中的顺序相同。这正是您从_mm256_load_ps(d) 得到的。如果幸运的话,编译器会看穿setr 并仍然发出一条加载指令。 (Marcus:很多人使用setr,因为它以“小端”顺序获取参数,第一个参数到元素0,等等,就像数组初始化器一样。_mm_set* 将参数从高到低元素,这匹配左/右移位如何在向量内移动数据。)
  • 如果 Alex 的数据真的是逗号分隔的有限长度整数,则可以使用 SIMD 而不是 scanfatof() 函数调用来解析它,并且可以大大提高速度。看到我回答的结尾,但这真的很复杂,对于 SIMD 新手来说绝对不是一个好的第一个项目。
  • 此外,如果您使用随机播放而不是存储到数组中,答案的水平总和部分会更有效。我的 SSE horizontal-sum answer 有一个 AVX 部分。
【解决方案2】:

在矢量化部分之外有很多低效问题(@Marcus 在他的回答中提到了这一点)。

不要为signal_t* s 动态分配空间。这是一个非常小的固定大小的结构,你只需要其中一个,所以你应该只使用signal_t s(自动存储)并删除一个间接级别。


在进行任何转换之前最好不要扫描整个字符串以查找,,因为如果字符串不适合 L1 缓存 (32k),那么您在转换时就会失去数据重用它。

如果你不能即时求和(比如平均值),那么分配一个大缓冲区来放入转换后的数据。如果你在没有填充缓冲区的情况下到达字符串的末尾,那很好。 realloc 缩小尺寸。 (您分配但从未接触过的内存页在大多数操作系统上基本上是空闲的。)如果您在到达字符串末尾之前填满了初始缓冲区大小,请使用realloc 将其增加两倍。 (指数大小增加了附加元素的平均 O(1) 成本,这就是 C++ 的 std::vector 以这种方式工作的原因。哈希表也是如此。)

如果您知道字符串的完整长度(例如,根据文件大小),您可以使用它来估算您需要多大的缓冲区。 (例如,假设每个浮点数是 2 个字节长,包括 ',',因为它们至少会那么长,并且在合理范围内过度分配是可以的。)


如果你真的想在转换之前计算逗号,你可以向量化它_mm_cmpeq_epi8 以在字符串数据向量中查找逗号,_mm_add_epi8 将这些向量求和为 0/-1。至少每 255 个向量使用 _mm_sad_epu8 将 8 位元素的水平总和转换为 64 位元素,以避免溢出。


如果您的数据类似于您的简单示例,其中每个数字实际上都是一个 1 位整数,那么您可以比使用 scanf 将它们转换为 float 做得更好很多。例如您可以使用整数 SIMD 将 ASCII 数字转换为 0 到 9 之间的整数。

// if we don't know the string length ahead of time,
// we could look for a '\0' on the fly with _mm256_cmpeq_epi8 / _mm256_movemask_epi
uint64_t digitstring_sum(const char*p, size_t len)
{
    const char *endp = p+len - 31;  // up to the last full-vector of string data

    __m256i sum = _mm256_setzero_si256();
    for ( ; p < endp ; p+=32 ) {
        __m256i stringdata = _mm256_loadu_si256((const __m256i*)p);
        __m256i integers = _mm256_sub_epi16(stringdata, _mm256_set1_epi16( '0'+(','<<8) ));  // turn "1,2,3,..." into 0x0100 0200 0300...

        // horizontal sum the 8-bit elements into 64-bit elements
        // There are various ways to optimize this by doing this part less frequently, but still often enough to avoid overflow.  Or doing it a different way.
        __m256i hsum = _mm256_sad_epu8(integers, _mm256_setzero_si256());  // sum(x[0] - 0, x[1] - 0, ...) = sum (x[...])

        sum = _mm256_add_epi64(sum, hsum);
    }
    // sum holds 4x 64-bit accumulators.  Horizontal sum that:
    // (this is probably more efficient than storing to memory and looping, but just barely for a vector of only 4 elements)
    __m128i hi = _mm256_extract_si128(sum, 1);
    __m128i lo = _mm256_castsi256_si128(sum);
    __m128i t1 = _mm_add_epi64(lo, hi);
    __m128i t2 = _mm_unpackhi_epi64(t1,t1);  // copy high 64 bit element to low
    __m128i t3 = _mm_add_epi64(t1, t2);
    uint64_t scalar_sum = _mm_cvtsi128_si32(t3);

    // Then a cleanup loop to handle the last partial-vector of string data.
    // or do it with an unaligned vector and some masking...

    return scalar_sum;
}

您的一些示例具有多位数字,但仍然只有整数。您可以通过使用向量比较来查找逗号的位置,将它们解析为 4 个一组的整数向量,并将该位图用作随机控制向量查找表的整数索引。

这变得非常复杂,但请参阅 @stgatilov's answer about parsing a dotted-quad IPv4 address string into an 32-bit integer using this technique. 由于 pshufb (_mm_shuffle_epi8) 在两个单独的通道中运行,因此您最好只使用 128 位向量。

您希望在循环中执行此操作,并一次在字符串中移动 4 个整数。给定一个整数形式的比较掩码结果,您可以通过剥离前 4 个设置位,然后使用位扫描指令/内在函数来找到第 5 个逗号的位置。可以使用BMI1 _blsr_u32 4 次来剥离前 4 个设置位。 (它通过一条指令执行dst = (a - 1) &amp; a)。

或者因为无论如何您都需要一个随机控制向量的 LUT,所以您可以使用 LUT 条目中的一些无关位来保存字节数。

【讨论】:

    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 2016-11-29
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2019-02-27
    • 1970-01-01
    • 2022-07-07
    相关资源
    最近更新 更多