这个问题通过一个简化的例子更容易解释(因为我的真实情况远非“最小”):给定...
template <typename T>
void post_in_thread_pool(T&& f)
...函数模板,我想创建一个具有树状递归结构的并行异步算法。我将使用 std::count_if
编写以下结构的示例作为占位符。我将要使用的策略如下:
如果我检查的范围长度小于
64
, 我将回到顺序std::count_if
功能。 (0)如果它大于或等于
64
,我将在线程池中生成一个作业,该作业在范围的左半部分递归,并在当前线程上计算范围的右半部分。 (1)我将使用原子共享
int
“等待”计算两半。 (2)我将使用原子共享
int
累积部分结果。 (3)
简化代码:
auto async_count_if(auto begin, auto end, auto predicate, auto continuation)
{
// (0) Base case:
if(end - begin < 64)
{
continuation(std::count_if(begin, end, predicate));
return;
}
// (1) Recursive case:
auto counter = make_shared<atomic<int>>(2); // (2)
auto cleanup = [=, accumulator = make_shared<atomic<int>>(0) /*(3)*/]
(int partial_result)
{
*accumulator += partial_result;
if(--*counter == 0)
{
continuation(*accumulator);
}
};
const auto mid = std::next(i_begin, sz / 2);
post_in_thread_pool([=]
{
async_count_if(i_begin, mid, predicate, cleanup);
});
async_count_if(mid, i_end, predicate, cleanup);
}
然后可以按如下方式使用代码:
std::vector<int> v(512);
std::iota(std::begin(v), std::end(v), 0);
async_count_if{}(std::begin(v), std::end(v),
/* predicate */ [](auto x){ return x < 256; },
/* continuation */ [](auto res){ std::cout << res << std::endl; });
上面代码中的问题是auto cleanup
.自 auto
cleanup
的每个实例都将被推导为唯一类型lambda,并且自 cleanup
捕获 cont
按值...由于递归,将在编译时计算无限大的嵌套 lambda 类型,导致以下错误:
fatal error: recursive template instantiation exceeded maximum depth of 1024
从概念上讲,您可以大致像这样想像正在构建的类型:
cont // user-provided continuation
cleanup0<cont> // recursive step 0
cleanup1<cleanup0<cont>> // recursive step 1
cleanup2<cleanup1<cleanup0<cont>>> // recursive step 2
// ...
(!):记住 async_count_if
只是一个例子,展示我真实情况的“树状”递归结构。我知道异步 count_if
可以使用单个原子计数器和 sz / 64
轻松实现任务。
我想避免错误,最小化任何可能的运行时或内存开销。
一个可能的解决方案是使用
std::function<void(int)> cleanup
,它允许代码正确编译和运行,但会产生次优汇编并引入额外的动态分配。 wandbox example- 另一种可能的解决方案是使用
std::size_t
模板参数+特化人为限制async_count_if::operator()
的递归深度——不幸的是,这会使二进制大小膨胀并且非常不优雅。
- 另一种可能的解决方案是使用
令我困扰的是,当我调用 async_count_if
时,我知道范围的大小。 :是std::distance(i_begin, i_end)
.如果我知道范围的大小,我还可以推断出所需计数器和延续的数量:(2^k - 1)
, 其中k
是递归树的深度。
因此,我认为在 async_count_if
的第一次调用中应该有一种预先计算“控制结构”的方法并通过引用将其传递给递归调用。这个“控制结构”可以包含足够的空间用于(2^k - 1)
原子计数器和 (2^k - 1)
清理/延续功能。
不幸的是,我找不到一个干净的方法来实现这个,并决定在这里发布一个问题,因为在开发异步并行递归算法时这个问题应该很常见。
在不引入不必要的开销的情况下处理这个问题的优雅方法是什么?
最佳答案
我肯定遗漏了一些非常明显的东西,但为什么你需要多个计数器和结构?你可以预先计算迭代的总数(如果你知道你的基本情况)并在所有迭代中与累加器一起共享它,a la(不得不稍微修改你的简化代码):
#include <algorithm>
#include <memory>
#include <vector>
#include <iostream>
#include <numeric>
#include <future>
using namespace std;
template <class T>
auto post_in_thread_pool(T&& work)
{
std::async(std::launch::async, work);
}
template <class It, class Pred, class Cont>
auto async_count_if(It begin, It end, Pred predicate, Cont continuation)
{
// (0) Base case:
if(end - begin <= 64)
{
continuation(std::count_if(begin, end, predicate));
return;
}
const auto sz = std::distance(begin, end);
const auto mid = std::next(begin, sz / 2);
post_in_thread_pool([=]
{
async_count_if(begin, mid, predicate, continuation);
});
async_count_if(mid, end, predicate, continuation);
}
template <class It, class Pred, class Cont>
auto async_count_if_facade(It begin, It end, Pred predicate, Cont continuation)
{
// (1) Recursive case:
const auto sz = std::distance(begin, end);
auto counter = make_shared<atomic<int>>(sz / 64); // (fix this for mod 64 !=0 cases)
auto cleanup = [=, accumulator = make_shared<atomic<int>>(0) /*(3)*/]
(int partial_result)
{
*accumulator += partial_result;
if(--*counter == 0)
{
continuation(*accumulator);
}
};
return async_count_if(begin, end, predicate, cleanup);
}
int main ()
{
std::vector<int> v(1024);
std::iota(std::begin(v), std::end(v), 0);
async_count_if_facade(std::begin(v), std::end(v),
/* predicate */ [](auto x){ return x > 1000; },
/* continuation */ [](const auto& res){ std::cout << res << std::endl; });
}
一些 demo
关于c++ - 避免并行递归异步算法中的递归模板实例化溢出,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/41572863/