有时候,你需要向后退一步并重新审视它。首先要问的问题显然是,你需要这个吗?是否有另一种算法可以更好地执行?假设出于这个问题的考虑你已经选择了这个算法,我们可以试着推理一下我们实际拥有的东西。
免责声明:我描述的方法受到Tim Peters成功改进传统introsort实现的启发,导致了
TimSort。所以请耐心听我讲解 ;)
1. 提取属性
我能看到的主要问题是迭代之间的依赖关系,这将阻止许多可能的优化和挫败许多并行化尝试。
int64_t v = in[i];
max += v;
if (v > max) max = v;
out[i] = max;
让我们以函数式的方式重新编写此代码:
max = calc(in[i], max)
out[i] = max
地点:
int64_t calc(int64_t const in, int64_t const max) {
int64_t const bumped = max + in;
return in > bumped ? in : bumped;
}
或者,更简化的版本(忽略了溢出,因为它是未定义的):
int64_t calc(int64_t const in, int64_t const max) {
return 0 > max ? in : max + in;
}
你注意到了临界点吗?行为会根据被错误命名的 max
是正数还是负数而改变。
这个临界点使得我们更加关注 in
中的值,特别是它们可能对 max
产生的影响:
max < 0
且 in[i] < 0
,则 out[i] = in[i] < 0
max < 0
且 in[i] > 0
,则 out[i] = in[i] > 0
max > 0
且 in[i] < 0
,则 out[i] = (max + in[i]) ?? 0
max > 0
且 in[i] > 0
,则 out[i] = (max + in[i]) > 0
(*) 这个名字不太恰当,因为它也是一个累加器,这个名字隐藏了这一点。但我没有更好的建议。
2. 优化操作
这使我们发现了一些有趣的情况:
- 如果我们有一个只包含负数值(我们称之为负片)的数组切片
[i,j)
,那么我们可以执行std :: copy(in + i,in + j,out + i)
和max = out [j-1]
- 如果我们有一个只包含正数值的数组切片
[i,j)
,那么它就是一个纯累加代码(可以很容易地展开)
max
在in [i]
为正时变得积极
因此,在实际处理输入之前建立输入文件的配置文件可能是有趣的(但也许不是必要的)。请注意,对于大型输入,可以逐块制作配置文件,例如根据缓存行大小调整块大小。
参考资料,3个例程:
void copy(int64_t const in[], int64_t out[],
size_t const begin, size_t const end)
{
std::copy(in + begin, in + end, out + begin);
}
void accumulate(int64_t const in[], int64_t out[],
size_t const begin, size_t const end)
{
assert(begin != 0);
int64_t max = out[begin-1];
for (size_t i = begin; i != end; ++i) {
max += in[i];
out[i] = max;
}
}
void regular(int64_t const in[], int64_t out[],
size_t const begin, size_t const end)
{
assert(begin != 0);
int64_t max = out[begin - 1];
for (size_t i = begin; i != end; ++i)
{
max = 0 > max ? in[i] : max + in[i];
out[i] = max;
}
}
假设我们可以使用简单的结构来描述输入:
struct Slice {
enum class Type { Negative, Neutral, Positive };
Type type;
size_t begin;
size_t end;
};
typedef void (*Func)(int64_t const[], int64_t[], size_t, size_t);
Func select(Type t) {
switch(t) {
case Type::Negative: return ©
case Type::Neutral: return ®ular;
case Type::Positive: return &accumulate;
}
}
void theLoop(std::vector<Slice> const& slices, int64_t const in[], int64_t out[]) {
for (Slice const& slice: slices) {
Func const f = select(slice.type);
(*f)(in, out, slice.begin, slice.end);
}
}
现在,除非introsort中的工作循环是最小的,因此计算特征可能会太昂贵...但是它非常适合于
并行化。
请注意,特征是输入的纯函数。因此,假设您按块每个块的方式工作,可以并行执行以下操作:
* 切片生产者: 一个characterizer线程,计算Slice::Type值
* 切片消费者: 一个工作线程,实际执行代码
即使输入基本上是随机的,只要切块足够小(例如,CPU L1高速缓存行),就可能有为其工作的块。两个线程之间的同步可以通过具有Slice的简单线程安全队列来完成(生产者/消费者),并添加bool last属性以停止消耗或通过使用原子,创建具有Unknown类型的向量中的Slice,并使消费者阻塞直到已知为止。
注意:由于特征是纯的,因此它可以进行令人尴尬的并行处理。
更多并行化:推测性工作。
记住这句无辜的话:
max
一旦in[i]
为正,就会变成正数。
假设我们可以可靠地猜测
Slice[j-1]
将产生一个负的
max
值,那么对
Slice[j]
的计算与之前的内容无关,我们可以立即开始工作!
当然,这只是一个猜测,所以我们可能会错...但是一旦我们完全描述了所有的Slices,我们就有了空闲的核心,因此我们可以利用它们进行投机性工作!如果我们错了?嗯,消费者线程将简单地轻松擦除我们的错误,并用正确的值替换它。
推测计算
Slice
的启发式方法应该很简单,并且需要进行调整。它也可能是适应性的...但这可能更困难!
结论
分析您的数据集并尝试找出是否有可能打破依赖关系。即使不使用多线程,您也可以利用它。