【发布时间】: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。