Stanford-CS149-并行计算-Assignment1-指南

CS的公开课鸽了好久, 打算开CS149这个新坑, 之前因为实习和秋招导致6.5840lab4B一直没写完, 回来后发现很多思路忘了, 有时间再补吧。

课程官网: https://gfxcourses.stanford.edu/cs149/fall24
作业: https://github.com/stanford-cs149/asst1
我的实现: https://github.com/ToniXWD/CS149-asst1

这个作业主要是对既有的程序进行分析和优化,写的代码倒不是特别多

0 环境配置

我使用Wsl Ubuntu22.04作为开发环境, wsl的使用可以参考: WSL入门到入土

基于配置好的wsl, 还需要进行的配置包括

  1. gcc, git吧啦吧啦, 这些应该不在话下了
  2. ispc的安装, 这个是intelispc编译器, 安装方式包括:
    1. https://ispc.github.io/downloads.html 处下载编译好的ispc编译器, 然后解压, 将ispc路径添加到环境变量中。
    2. 使用snap安装, sudo snap install ispc即可。
  3. ppm图像查看工具, 不建议使用官方推荐的工具, 这里推荐使用feh:
    1. sudo apt install feh
    2. 使用feh查看ppm图像: feh <image_path>

1 多线程加速Mandelbrot set图像迭代绘制

1.1 实现多线程版本

开胃菜, 很简单, 不需要懂Mandelbrot set是什么, 只需要将操作分配到多个线程中即可:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
//
// MandelbrotThread --
//
// Multi-threaded implementation of mandelbrot set image generation.
// Threads of execution are created by spawning std::threads.
void mandelbrotThread(
int numThreads,
float x0, float y0, float x1, float y1,
int width, int height,
int maxIterations, int output[])
{
static constexpr int MAX_THREADS = 32;

if (numThreads > MAX_THREADS)
{
fprintf(stderr, "Error: Max allowed threads is %d\n", MAX_THREADS);
exit(1);
}

// Creates thread objects that do not yet represent a thread.
std::thread workers[MAX_THREADS];
WorkerArgs args[MAX_THREADS];

for (int i=0; i<numThreads; i++) {

// TODO FOR CS149 STUDENTS: You may or may not wish to modify
// the per-thread arguments here. The code below copies the
// same arguments for each thread
args[i].x0 = x0;
args[i].y0 = y0;
args[i].x1 = x1;
args[i].y1 = y1;
args[i].width = width;
args[i].height = height;
args[i].maxIterations = maxIterations;
args[i].numThreads = numThreads;
args[i].output = output;

args[i].threadId = i;
}

// Spawn the worker threads. Note that only numThreads-1 std::threads
// are created and the main application thread is used as a worker
// as well.
for (int i=1; i<numThreads; i++) {
workers[i] = std::thread(workerThreadStart, &args[i]);
}

workerThreadStart(&args[0]);

// join worker threads
for (int i=1; i<numThreads; i++) {
workers[i].join();
}
}

这里的workerThreadStart最可能的实现是这样的:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
//
// workerThreadStart --
//
// Thread entrypoint.
void workerThreadStart(WorkerArgs * const args) {

// TODO FOR CS149 STUDENTS: Implement the body of the worker
// thread here. Each thread should make a call to mandelbrotSerial()
// to compute a part of the output image. For example, in a
// program that uses two threads, thread 0 could compute the top
// half of the image and thread 1 could compute the bottom half.

int row_num_per_t = args->height / args->numThreads;
int start_row = args->threadId * row_num_per_t;
int end_row = start_row + row_num_per_t;

// 行的范围:[start_row, end_row)

if (args->threadId==args->numThreads-1) {
end_row = args->height;
}

double t_startTime = CycleTimer::currentSeconds();

mandelbrotSerial(args->x0, args->y0, args->x1, args->y1, args->width, args->height, start_row, end_row-start_row, args->maxIterations, args->output);

double t_endTime = CycleTimer::currentSeconds();
printf("Thread %d: [%.3f] ms\n", args->threadId, (t_endTime - t_startTime) * 1000);
}

1.2 && 1.3 加速效果分析

根据之前的初版代码测试, 实际上加速效果如下图, 当线程数为3时, 加速效果反而不如线程数为2时:
prog1 Accelerate
根据提示:

Hint: take a careful look at the three-thread datapoint.

以 t=3 为例,输出如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
$ ./mandelbrot -t 3
[mandelbrot serial]: [282.544] ms
Wrote image file mandelbrot-serial.ppm
Thread 0: [57.570] ms
Thread 2: [59.159] ms
Thread 1: [176.854] ms
Thread 2: [56.701] ms
Thread 0: [56.825] ms
Thread 1: [175.035] ms
Thread 0: [56.324] ms
Thread 2: [57.591] ms
Thread 1: [176.025] ms
Thread 0: [59.012] ms
Thread 2: [59.654] ms
Thread 1: [180.582] ms
Thread 0: [56.932] ms
Thread 2: [57.607] ms
Thread 1: [176.718] ms
[mandelbrot thread]: [175.223] ms
Wrote image file mandelbrot-thread.ppm
(1.61x speedup from 3 threads)

利用率不足的原因是不同线程负责的图像区域迭代的计算量不同,导致计算时间不同,出现了短板效应。3 线程中中间部分的计算时间明细大于上下两边,导致时间增加。

1.4 优化方案

因此, 解决方案就是, 每个线程计算的粒度进行细分, 从上到下按照一定间隔计算多块区域,每块区域不应该太小而导致无法利用局部性, 也不应该太大导致不同线程计算量差异过大, 这样每个线程的计算量被平均了,不会出现短板效应:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
//
// workerThreadStart --
//
// Thread entrypoint.
void workerThreadStart(WorkerArgs * const args) {

double t_startTime = CycleTimer::currentSeconds();

const unsigned int CHUNK_SIZE = 16; // 或 8、32 等,需要测试最优值

// 每次处理多行,减少函数调用开销
for (unsigned int cur_row = args->threadId * CHUNK_SIZE;
cur_row < args->height;
cur_row += args->numThreads * CHUNK_SIZE) {

// 确保不会越界
int numRows = std::min(CHUNK_SIZE,
args->height - cur_row);

mandelbrotSerial(args->x0, args->y0, args->x1, args->y1,
args->width, args->height,
cur_row, numRows,
args->maxIterations, args->output);
}

double t_endTime = CycleTimer::currentSeconds();
printf("Thread %d: [%.3f] ms\n", args->threadId,
(t_endTime - t_startTime) * 1000);
}

1.5 最大化利用率

就是将线程数量设置为cpu核心数, 这样多个线程可以实现真正的并行计算, 而减少非必要的上下文切换, 从而最大化利用率, 我自己的机器是因为我的 CPU 是 10 核 16 线程的 12600k, t=16 时,利用率为 10.51, 但当 t=32 时,利用率为 10.50,没有提升。

2 使用SIMD加速指数计算

这里的SIMD是官方提供的模拟器, 因为这样方便数据统计和调试, 因此开始这个实验前, 需要认真阅读CS149intrin.h每个函数的声明。

2.1 实现SIMD

直接给出代码吧, 有些细节要注意, 但没什么特别难的点需要解释, 每个步骤我批注了注释。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
void clampedExpVector(float* values, int* exponents, float* output, int N) {

//
// CS149 STUDENTS TODO: Implement your vectorized version of
// clampedExpSerial() here.
//
// Your solution should work for any value of
// N and VECTOR_WIDTH, not just when VECTOR_WIDTH divides N
//
__cs149_vec_int zero_i = _cs149_vset_int(0);
__cs149_vec_int one_i = _cs149_vset_int(1);
__cs149_vec_float nine_999 = _cs149_vset_float(9.999999f);
__cs149_mask mask_one = _cs149_init_ones(VECTOR_WIDTH); // 1 的 mask

__cs149_vec_float s_x;
__cs149_vec_int s_e;
__cs149_vec_float results;

__cs149_mask mask_lt_N; // 索引小于 N 的 mask
__cs149_mask mask_unfinished; // 计算还没有结束的 mask
__cs149_mask mask_zero_exp; // 指数为 0 的 mask
__cs149_mask greater_than_9_999; // 大于 9.999 的 mask

for (int i=0; i<N; i+=VECTOR_WIDTH) {
mask_lt_N = _cs149_init_ones(min(N-i, VECTOR_WIDTH)); // 标记索引小于 N 的元素

_cs149_vload_float(s_x, values+i, mask_lt_N); // 从 values[i] 开始,加载 VECTOR_WIDTH 个元素到 s_x
_cs149_vload_int(s_e, exponents+i, mask_lt_N); // 从 exponents[i] 开始,加载 VECTOR_WIDTH 个元素到 s_e
_cs149_vset_float(results, 1.f, mask_lt_N); // 将结果初始化为 1

mask_unfinished = mask_lt_N; // 初始情况时,只要索引有效,则表示计算未结束
greater_than_9_999 = _cs149_init_ones(0); // 初始化为 0

_cs149_veq_int(mask_zero_exp, s_e, zero_i, mask_one); // 如果 exponent == 0, 则标记为 1 表示计算结束,否则标记为 0
mask_zero_exp = _cs149_mask_not(mask_zero_exp); // 反转,如果 exp 为 0, 则表示计算结束,用 0 表示
mask_unfinished = _cs149_mask_and(mask_unfinished, mask_zero_exp); // 如果指数为 0, 则表示计算结束
mask_unfinished = _cs149_mask_and(mask_unfinished, mask_lt_N); // 保险起见,再和 mask_lt_N 取交集,保证索引无效的部分为 0

while (_cs149_cntbits(mask_unfinished) > 0) {
_cs149_vmult_float(results, results, s_x, mask_unfinished); // 计算一次乘方
_cs149_vsub_int(s_e, s_e, one_i, mask_unfinished); // 指数自减

_cs149_vgt_float(greater_than_9_999, results, nine_999, mask_unfinished); // 如果结果大于 9.999, 则标记为 1
_cs149_vset_float(results, 9.999999f, greater_than_9_999); // 如果结果大于 9.999, 则将结果设置为 9.999

// 更新 mask_unfinished
mask_zero_exp = _cs149_init_ones(0); // 重新初始化 mask_zero_exp
_cs149_veq_int(mask_zero_exp, s_e, zero_i, mask_one); // 如果 exponent == 0, 则标记为 1 表示计算结束,否则标记为 0

__cs149_mask new_finished = _cs149_mask_or(mask_zero_exp, greater_than_9_999); // 如果指数为 0 或结果大于 9.999, 则表示计算结束,用 1 表示

mask_zero_exp = _cs149_mask_not(new_finished); // 反转,计算结束的位置用 0 表示,计算未结束的位置用 1 表示
mask_unfinished = _cs149_mask_and(mask_unfinished, mask_zero_exp); // 如果指数为 0, 则表示计算结束
mask_unfinished = _cs149_mask_and(mask_unfinished, mask_lt_N); // 保险起见,再和 mask_lt_N 取交集,保证索引无效的部分为 0
}

_cs149_vstore_float(output+i, results, mask_lt_N); // 将结果存储到 output 中

addUserLog("clampedExpVector");
}
}

2.2 测试加速效果

VECTOR_WIDTH 利用率
2 85.3%
4 81.3%
8 79.0%
16 77.7%

利用率肯定是随着 VECTOR_WIDTH 增长而减少的,因为随着 VECTOR_WIDTH 增长,每次计算的元素数量增加,导致迭代次数最多的那一个元素相较于平均的差异越大,导致利用率越低。

2.3 数组求和的SIMD实现

有了之前的经验, 这里应该手到擒来吧

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
// returns the sum of all elements in values
// You can assume N is a multiple of VECTOR_WIDTH
// You can assume VECTOR_WIDTH is a power of 2
float arraySumVector(float* values, int N) {

//
// CS149 STUDENTS TODO: Implement your vectorized version of arraySumSerial here
//

__cs149_vec_float v_sum;
__cs149_mask mask_one = _cs149_init_ones(VECTOR_WIDTH); // 1 的 mask
_cs149_vset_float(v_sum, 0.f, mask_one);

__cs149_vec_float v_elems;

for (int i=0; i<N; i+=VECTOR_WIDTH) {
_cs149_vload_float(v_elems, values+i, mask_one); // 从 values[i] 开始,加载 VECTOR_WIDTH 个元素到 v_elems
_cs149_vadd_float(v_sum, v_sum, v_elems, mask_one); // 将 v_elems 加到 v_sum 上
}

float sum = 0.f;
for (int i=0; i<VECTOR_WIDTH; i++) {
sum += v_sum.value[i];
}

return sum;
}

3 使用ispc加速Mandelbrot set图像绘制

3.1 ispc入门

参考官方文档: https://ispc.github.io/example.html, 这里就不赘述了。

3.2 分析ispc加速效果

3.2.1 既有程序加速效果

我的结果:

1
2
3
4
5
6
7
8
9
./mandelbrot_ispc --tasks
[mandelbrot serial]: [142.173] ms
Wrote image file mandelbrot-serial.ppm
[mandelbrot ispc]: [40.064] ms
Wrote image file mandelbrot-ispc.ppm
[mandelbrot multicore ispc]: [5.939] ms
Wrote image file mandelbrot-task-ispc.ppm
(3.55x speedup from ISPC)
(23.94x speedup from task ISPC)

3.2.2 优化ispc代码

task数量设置为本机最大硬件线程数,可以获得最佳利用率。

3.2.3 ispc和第一个任务重的线程任务的区别是什么?

task实际上会尽量均匀分配给硬件线程,所以当任务数量小于硬件线程数量时,随着任务数量增加,利用率会逐渐增加,当任务数量大于硬件线程数量时,反而会因为上下文切换导致利用率降低

4 牛顿迭代法实现sqrt

这个实验还挺激动的, 以前数值分析课学过牛顿迭代法, 但都是用matlab调包, 这次终于可以自己实现了, 而且还可以用ispc加速, 虽然这个实验的代码量也不多, 但确实让我对ispc有了更深的理解, 同时有一点科研的成就感, 哈哈。

4.1 实现ispc版本

结果:

1
2
3
4
5
6
(base) toni@DESKTOP-59EELP2:prog4_sqrt$ ./sqrt
[sqrt serial]: [633.737] ms
[sqrt ispc]: [170.368] ms
[sqrt task ispc]: [14.496] ms
(3.72x speedup from ISPC)
(43.72x speedup from task ISPC)

4.2 ispc加速

这里的思路其实和前一个实验的思路是一样的, 就是尽量将SIMD一次计算的每个通道元素的迭代次数的差异最小化, 因此直接分析官方提供的这张图就够了:
sqrt_ispc
如果要加速, 就让元素尽量迭代次数相近, 这里我选择初始化在0.5-2.5之间:

1
values[i] = .5f + 2.0f * static_cast<float>(rand()) / RAND_MAX; // Q2

4.3 ispc减速

思路和sqrt的加速是一样的, 就是尽量让每个通道元素的迭代次数相差更大, 这里我如下实现最大化差异:

1
2
3
4
5
if (i % 8 == 0) {
values[i] = 2; // Q3
} else {
values[i] = 1; // Q3
}

4.4 (附加题)使用avx2自己实现sqrt

这应该是最难的一个实验了, 需要对avx2的指令集有一定的了解, 这里我就不赘述了, 直接看代码吧, 每一行基本上都有注释:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
#include <cmath>
#include <cstdio>
#include <immintrin.h>
#include <omp.h>

// 使用 AVX 指令集实现平方根计算(牛顿迭代法)
// N: 数组长度
// initialGuess: 迭代初始值
// values: 输入数组,存储需要计算平方根的数
// output: 输出数组,存储计算结果
// 方程:1 / guess^2 = x
// !!!: x 不是迭代变量,guess 才是,且这里的 x 是数组的元素
// !!!: sqrt(x) = 1 / guess = x * guess
void sqrt_avx(int N, float initialGuess, float *values, float *output) {
// 输入参数有效性检查
if (!values || !output || N <= 0) return;

// 设置收敛阈值
static const float kThreshold = 0.00001f;

// 计算能被 8 整除的最大长度(AVX2 可以同时处理 8 个 float)
int aligned_n = (N / 8) * 8;

// 使用 AVX 指令并行处理每 8 个元素
for (int i = 0; i < aligned_n; i += 8) {
// 将阈值广播到 256 位向量中(8 个 float)
__m256 threshold = _mm256_set1_ps(kThreshold);
// 从内存加载 8 个连续的 float 到向量寄存器
__m256 x_vec = _mm256_loadu_ps(&values[i]);
// 将初始猜测值广播到向量寄存器
__m256 guess = _mm256_set1_ps(initialGuess);

// 计算初始误差:|guess² * x - 1|
__m256 pred = _mm256_sub_ps(
_mm256_mul_ps(_mm256_mul_ps(guess, guess), x_vec),
_mm256_set1_ps(1.0f)
);
// 计算误差的绝对值
pred = _mm256_andnot_ps(_mm256_set1_ps(-0.0f), pred);

/*
1. 在浮点数的二进制表示中,最高位(符号位)为:
- 0:表示正数
- 1:表示负数

2. `-0.0f` 的二进制表示是 `1000 0000 0000 0000 0000 0000 0000 0000`
- 即只有符号位为 1,指数和尾数位都是 0

3. `_mm256_andnot_ps(a, b)` 的操作相当于 `~a & b`,即:
- 先对第一个操作数取位反(NOT)
- 然后与第二个操作数进行位与(AND)

4. 所以 `_mm256_andnot_ps(_mm256_set1_ps(-0.0f), x)` 的过程是:
    x:              [sign][exponent][mantissa] (原始数)
    -0.0f:          1000...000      (全 0,仅符号位为 1)
    ~(-0.0f):       0111...111      (全 1,仅符号位为 0)
    ~(-0.0f) & x:   0[exponent][mantissa]     (保持数值位不变,符号位强制为 0)
    
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

这样就巧妙地将任何负数转换为对应的正数,而正数保持不变,从而实现了绝对值的计算。
*/


// 当任意一个元素的误差大于阈值时继续迭代
while (_mm256_movemask_ps(_mm256_cmp_ps(pred, threshold, _CMP_GT_OQ)) != 0) {
// _mm256_movemask_ps 是 AVX 指令集中的一个函数,用于从 256 位浮点向量中提取每个元素的符号位,并将这些符号位组合成一个 8 位的整数。
// 牛顿迭代公式:guess = (3 * guess - x * guess³) * 0.5
guess = _mm256_mul_ps(
_mm256_set1_ps(0.5f),
_mm256_sub_ps(
_mm256_mul_ps(_mm256_set1_ps(3.0f), guess),
_mm256_mul_ps(x_vec, _mm256_mul_ps(guess, _mm256_mul_ps(guess, guess)))
)
);

// 重新计算误差
pred = _mm256_sub_ps(
_mm256_mul_ps(_mm256_mul_ps(guess, guess), x_vec),
_mm256_set1_ps(1.0f)
);
pred = _mm256_andnot_ps(_mm256_set1_ps(-0.0f), pred);
}

// 计算最终结果 sqrt(x) = x * guess,并存储到输出数组
_mm256_storeu_ps(&output[i], _mm256_mul_ps(x_vec, guess));
}

// 使用标准算法处理剩余的元素(不足 8 个的部分)
for (int i = aligned_n; i < N; i++) {
float x = values[i];
float guess = initialGuess;
float error = fabs(guess * guess * x - 1.f);

while (error > kThreshold) {
guess = (3.f * guess - x * guess * guess * guess) * 0.5f;
error = fabs(guess * guess * x - 1.f);
}

output[i] = x * guess;
}
}

// 使用 AVX 指令集实现平方根计算(牛顿迭代法)
// 使用掩码,当满足阈值时不继续迭代 (这个案例其实没有必要)
// N: 数组长度
// initialGuess: 迭代初始值
// values: 输入数组,存储需要计算平方根的数
// output: 输出数组,存储计算结果
// 方程:1 / guess^2 = x
// !!!: x 不是迭代变量,guess 才是,且这里的 x 是数组的元素
// !!!: sqrt(x) = 1 / guess = x * guess
void sqrt_avx_mask(int N, float initialGuess, float *values, float *output) {
// 输入参数有效性检查
if (!values || !output || N <= 0) return;

// 设置收敛阈值
static const float kThreshold = 0.00001f;

// 计算能被 8 整除的最大长度(AVX2 可以同时处理 8 个 float)
int aligned_n = (N / 8) * 8;

// 使用 AVX 指令并行处理每 8 个元素
for (int i = 0; i < aligned_n; i += 8) {
// 将阈值广播到 256 位向量中(8 个 float)
__m256 threshold = _mm256_set1_ps(kThreshold);
// 从内存加载 8 个连续的 float 到向量寄存器
__m256 x_vec = _mm256_loadu_ps(&values[i]);
// 将初始猜测值广播到向量寄存器
__m256 guess = _mm256_set1_ps(initialGuess);

// 计算初始误差:|guess² * x - 1|
__m256 pred = _mm256_sub_ps(
_mm256_mul_ps(_mm256_mul_ps(guess, guess), x_vec),
_mm256_set1_ps(1.0f)
);
// 计算误差的绝对值
pred = _mm256_andnot_ps(_mm256_set1_ps(-0.0f), pred);

/*
1. 在浮点数的二进制表示中,最高位(符号位)为:
- 0:表示正数
- 1:表示负数

2. `-0.0f` 的二进制表示是 `1000 0000 0000 0000 0000 0000 0000 0000`
- 即只有符号位为 1,指数和尾数位都是 0

3. `_mm256_andnot_ps(a, b)` 的操作相当于 `~a & b`,即:
- 先对第一个操作数取位反(NOT)
- 然后与第二个操作数进行位与(AND)

4. 所以 `_mm256_andnot_ps(_mm256_set1_ps(-0.0f), x)` 的过程是:
x: [sign][exponent][mantissa] (原始数) -0.0f: 1000...000 (全 0,仅符号位为 1) ~(-0.0f): 0111...111 (全 1,仅符号位为 0) ~(-0.0f) & x: 0[exponent][mantissa] (保持数值位不变,符号位强制为 0)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

这样就巧妙地将任何负数转换为对应的正数,而正数保持不变,从而实现了绝对值的计算。
*/

// 初始化掩码为全 1
__m256 mask = _mm256_cmp_ps(pred, threshold, _CMP_GT_OQ);
/*
常用的比较操作符(imm8 的值):
_CMP_EQ_OQ:相等
_CMP_LT_OQ:小于
_CMP_LE_OQ:小于或等于
_CMP_GT_OQ:大于
_CMP_GE_OQ:大于或等于
_CMP_NEQ_OQ:不等于
*/

// 当任意一个元素的误差大于阈值时继续迭代
while (_mm256_movemask_ps(mask) != 0) {
// _mm256_movemask_ps 是 AVX 指令集中的一个函数,用于从 256 位浮点向量中提取每个元素的符号位,并将这些符号位组合成一个 8 位的整数。
// 牛顿迭代公式:guess = (3 * guess - x * guess³) * 0.5
__m256 new_guess = _mm256_mul_ps(
_mm256_set1_ps(0.5f),
_mm256_sub_ps(
_mm256_mul_ps(_mm256_set1_ps(3.0f), guess),
_mm256_mul_ps(x_vec, _mm256_mul_ps(guess, _mm256_mul_ps(guess, guess)))
)
);

// 仅更新掩码为 1 的元素
guess = _mm256_blendv_ps(guess, new_guess, mask);
/*
_mm256_blendv_ps 用于根据掩码在两个 256 位浮点向量之间进行条件选择。它允许在两个向量中选择性地从一个向量复制元素到结果向量中,具体选择由掩码决定。
参数:
a:第一个 256 位浮点向量。
b:第二个 256 位浮点向量。
mask:一个 256 位浮点向量,用于控制选择。
返回值:
返回一个新的 256 位浮点向量。对于每个元素,如果 mask 中对应位置的符号位为 1,则从 b 中选择该元素;否则,从 a 中选择该元素。
*/

// 重新计算误差
pred = _mm256_sub_ps(
_mm256_mul_ps(_mm256_mul_ps(guess, guess), x_vec),
_mm256_set1_ps(1.0f)
);
pred = _mm256_andnot_ps(_mm256_set1_ps(-0.0f), pred);

// 更新掩码
mask = _mm256_cmp_ps(pred, threshold, _CMP_GT_OQ);
}

// 计算最终结果 sqrt(x) = x * guess,并存储到输出数组
_mm256_storeu_ps(&output[i], _mm256_mul_ps(x_vec, guess));
}

// 使用标准算法处理剩余的元素(不足 8 个的部分)
for (int i = aligned_n; i < N; i++) {
float x = values[i];
float guess = initialGuess;
float error = fabs(guess * guess * x - 1.f);

while (error > kThreshold) {
guess = (3.f * guess - x * guess * guess * guess) * 0.5f;
error = fabs(guess * guess * x - 1.f);
}

output[i] = x * guess;
}
}

5 BLAS中saxpy实现

这一部分都是对代码进行分析, 没有自己需要写代码的地方

5.1 分析saxpy为什么使用task`加速不明显

因为给定的数组太大了, 这是主要瓶颈是内存带宽, 而不是计算资源, 所以使用task加速不明显。

5.2 分析对内存访问次数的计算

读取和写入result需要 2 次内存访问,而读取X和Y各需要 1 次内存访问,所以总内存访问次数为 4N。

5.3 (附加题)优化方案

尴尬, 这道题没做出来…

6 k-means优化

这道题的数据文件没有开源, 所以没法做…