迭代FFT性能劣于递归FFT,请求排查代码性能瓶颈
迭代FFT比递归版慢的原因排查
我用Python实现了递归FFT和迭代FFT函数,两者均通过所有单元测试,但绘制复杂度曲线时发现迭代版本反而比递归版本慢——虽然理论复杂度都是O(n lg(n)),但迭代版的曲线明显向上偏移。请帮忙排查代码里拖慢迭代FFT的原因,是不是我误用了ThreadPoolExecutor?
def iterative_fft(a: np.array, inverse=False) -> np.array: """ :param a: polynomial in coefficient form :param inverse: direct or inverse DFT flag :return: DFT(a) in point form """ n = len(a) a = bit_reverse_copy(a) for s in range(1, int(np.log2(n)) + 1): m = 2 ** s omega_m = np.exp(2 * np.pi * 1j / m) if inverse: omega_m **= -1 def butterfly(k): omega = 1 for j in range(m // 2): t = omega * a[k + j + m // 2] u = a[k + j] a[k + j] = u + t a[k + j + m // 2] = u - t omega = omega * omega_m with concurrent.futures.ThreadPoolExecutor() as executor: executor.map(butterfly, range(0, n, m)) if inverse: a = a / n return a def bit_reverse_copy(a: np.array) -> np.array: """ :param a: polynomial :return: reordered array based on index bits reverse """ n = len(a) bit_size = int(np.log2(n)) reversed_a = np.zeros(n, dtype=complex) for i in range(n): reversed_index = reverse_bits(i, bit_size) reversed_a[reversed_index] = a[i] return reversed_a def reverse_bits(n, bit_size): """ :param n: integer :param bit_size: bits used for integer representation :return: reversed bit integer """ reverse_n = 0 for i in range(bit_size): reverse_n = (reverse_n << 1) | (n & 1) n >>= 1 return reverse_n
核心问题分析
1. ThreadPoolExecutor的误用是主要拖慢因素
- 重复创建线程池开销极大:在每一轮
s循环中,你都新建一个ThreadPoolExecutor,线程的创建、初始化、销毁都会产生显著开销,而循环次数是log2(n)次,累积下来会吃掉大量性能。 - 任务粒度太小,线程调度得不偿失:当
s较小时(比如s=1,m=2),每个butterfly(k)只处理1次蝴蝶操作,任务的计算量远小于线程调度的开销。加上Python的GIL限制,CPU密集型任务的多线程不仅无法并行加速,反而会因为线程切换增加额外消耗。
2. 位反转拷贝的效率低下
bit_reverse_copy和reverse_bits用纯Python循环实现,而Python循环的速度远低于numpy的向量化操作。当n较大时,这部分的耗时会非常明显。
3. 蝴蝶操作的细节优化空间
- 每次蝴蝶操作中,
omega通过循环累积乘法计算,可预计算所有需要的ω幂次,避免重复计算。 - 直接修改numpy数组虽然没问题,但结合多线程的共享访问,即使没有竞争,也会有缓存一致性的额外开销。
优化建议
移除多线程,改用单线程处理
先把ThreadPoolExecutor相关代码删掉,直接用循环遍历range(0, n, m)调用butterfly(k),这几乎肯定会提升性能——Python的GIL决定了CPU密集型任务单线程反而更快。用numpy向量化实现位反转
替换掉纯Python循环的位反转逻辑,改用numpy的向量化操作:def bit_reverse_copy(a: np.array) -> np.array: n = len(a) bit_size = int(np.log2(n)) indices = np.arange(n, dtype=np.uint32) reversed_indices = np.zeros(n, dtype=np.uint32) for i in range(bit_size): reversed_indices = (reversed_indices << 1) | (indices & 1) indices >>= 1 return a[reversed_indices]这样能把位反转的速度提升几个数量级。
预计算ω幂次
在每轮s循环中,预计算所有需要的ω值,避免蝴蝶操作里的循环累积:m = 2 ** s omega_m = np.exp(2 * np.pi * 1j / m) if inverse: omega_m = omega_m.conj() # 用共轭替代幂运算,更高效 omegas = np.power(omega_m, np.arange(m//2))然后在蝴蝶操作中直接使用预计算的
omegas[j]。如果非要用多线程,复用线程池
把ThreadPoolExecutor的创建移到s循环外面,避免重复创建销毁线程,但依然要注意GIL的限制——这种优化的收益可能非常有限,甚至没有收益。
内容的提问来源于stack exchange,提问作者Gio Formichella
相关产品推荐
相关产品推荐

