我正在尝试并行化C中的卷积函数。这是原始函数,它会卷积两个64位浮点数组:
void convolve(const Float64 *in1,
UInt32 in1Len,
const Float64 *in2,
UInt32 in2Len,
Float64 *results)
{
UInt32 i, j;
for (i = 0; i < in1Len; i++) {
for (j = 0; j < in2Len; j++) {
results[i+j] += in1[i] * in2[j];
}
}
}
为了允许并发(没有信号量),我创建了一个函数来计算特定位置的结果。 results
数组:
void convolveHelper(const Float64 *in1,
UInt32 in1Len,
const Float64 *in2,
UInt32 in2Len,
Float64 *result,
UInt32 outPosition)
{
UInt32 i, j;
for (i = 0; i < in1Len; i++) {
if (i > outPosition)
break;
j = outPosition - i;
if (j >= in2Len)
continue;
*result += in1[i] * in2[j];
}
}
问题是,使用 convolveHelper
将代码减慢约3.5倍(在单个线程上运行时)。
关于如何加速的任何想法 convolveHelper
,同时保持线程安全?
时域中的卷积在傅立叶域中成为乘法。我建议你抓一个快速的FFT库(比如 FFTW并使用它。你将从O(n ^ 2)到O(n log n)。
算法优化几乎总是优于微优化。
可能有帮助的最明显的事情是预先计算循环的起始和结束索引,并删除额外的测试 i
和 j
(及其相关的跳跃)。这个:
for (i = 0; i < in1Len; i++) {
if (i > outPosition)
break;
j = outPosition - i;
if (j >= in2Len)
continue;
*result += in1[i] * in2[j];
}
可以改写为:
UInt32 start_i = (in2Len < outPosition) ? outPosition - in2Len + 1 : 0;
UInt32 end_i = (in1Len < outPosition) ? in1Len : outPosition + 1;
for (i = start_i; i < end_i; i++) {
j = outPosition - i;
*result += in1[i] * in2[j];
}
这样,条件 j >= in2Len
从来都不是,循环测试基本上是测试的组合 i < in1Len
和 i < outPosition
。
从理论上讲,你也可以摆脱任务 j
然后转 i++
成 ++i
,但编译器可能已经为您做了那些优化。
除非您的数组非常大,否则使用线程实际上不太可能有用,因为启动线程的开销将大于循环的开销。但是,让我们假设您的阵列很大,并且线程是一个净胜利。在那种情况下,我会做以下事情:
进入自己的功能 i
作为参数以及其他一切。
- 有身体
convolve
只需启动线程。为了获得最大效率,请使用信号量以确保永远不会创建比核心更多的线程。
答案在于简单数学而非多线程(更新)
这就是为什么......
考虑 一个b + aC
你可以优化它 的a *(B + C) (一
多次繁殖
在你的情况下有 in2Len 不必要的乘法 内环。哪个可以消除。
因此,如下修改代码应该给我们reqd卷积:
(注意: 以下代码返回 圆形卷积 必须展开才能获得 线性卷积 结果。
void convolve(const Float64 *in1,
UInt32 in1Len,
const Float64 *in2,
UInt32 in2Len,
Float64 *results)
{
UInt32 i, j;
for (i = 0; i < in1Len; i++) {
for (j = 0; j < in2Len; j++) {
results[i+j] += in2[j];
}
results[i] = results[i] * in1[i];
}
}
这应该给U带来最大的性能跳跃。试试吧,看看!!
祝你好运!!
CVS @ 2600Hertz
我终于想出了如何正确预先计算开始/结束索引(两者给出的建议) 泰勒麦克亨利 和 interjay):
if (in1Len > in2Len) {
if (outPosition < in2Len - 1) {
start = 0;
end = outPosition + 1;
} else if (outPosition >= in1Len) {
start = 1 + outPosition - in2Len;
end = in1Len;
} else {
start = 1 + outPosition - in2Len;
end = outPosition + 1;
}
} else {
if (outPosition < in1Len - 1) {
start = 0;
end = outPosition + 1;
} else if (outPosition >= in2Len) {
start = 1 + outPosition - in2Len;
end = in1Len;
} else {
start = 0;
end = in1Len;
}
}
for (i = start; i < end; i++) {
*result = in1[i] * in2[outPosition - i];
}
不幸的是,预先计算索引会产生 执行时间没有明显减少 :(
让convolve助手在更大的集上工作,使用短外循环计算多个结果。
并行化的关键是在线程之间的工作分配之间找到一个很好的平衡点。不要使用比CPU核心数更多的线程。
在所有线程之间平均分配工作。有了这种问题,每个线程工作的复杂性应该是相同的。