避免并行递归异步算法中的递归模板实例化溢出

Question

这个问题通过一个简化的例子更容易解释（因为我的实际情况远非“最小”）：给定一个...

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 类型递归，导致如下错误：

致命错误：递归模板实例化超出最大深度 1024

wandbox example

从概念上讲，您可以大致认为构建的类型是这样的：

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) 清理/延续函数包含足够的空间。

不幸的是，我找不到一个干净的方法来实现这一点，并决定在这里发布一个问题，因为在开发异步并行递归算法时，这个问题似乎应该很常见。

在不引入不必要开销的情况下，有什么优雅的方式来处理这个问题？

Answer 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

Answer 2

您使用原子整数进行同步是共享可变状态。共享可变状态会降低并行算法的性能。您的共享状态在每个线程上共享。

不要那样做。

template<class T>
auto sink_into_pointer( T* target ) {
  return [target](T x){*target=x;};
}
template<class T>
auto sink_into_promise( std::promise<T>& p ) {
  return [&p](T x){p.set_value(x);};
}
void async_count_if(auto begin, auto end, auto predicate, auto continuation) {
  // (0) Base case:  
  if(end - begin < 64)
  {
    continuation(std::count_if(begin, end, std::move(predicate)));
    return;
  }

  std::promise< int > sub_count;
  auto sub_count_value = sub_count.get_future();

  auto sub_count_task = sink_into_promise(sub_count);
  // (1) Recursive case:
  const auto mid = std::next(i_begin, sz / 2);        

  post_in_thread_pool(
    [sub_count_task]()mutable
    {
      async_count_if(i_begin, mid, predicate, sub_count_task);
    }
  );

  int second_half = 0;
  auto second_sub_count = sink_into_pointer(&second_half);

  async_count_if(mid, i_end, predicate, second_sub_count);

  continuation( second_half + sub_count_value.get() );
}

在这种情况下，线程之间唯一共享的状态是通过packaged_tasks 和线程池管理器返回的值。

在编写并行代码时，您的目标应该是最大化并行性，而不是最大化给定线程内的速度。与每个线程跟踪一次函数指针相比​​，共享资源的争用等会导致更严重的缩放问题。

Answer 3

您可以通过以下方式解决模板递归问题：

#include <algorithm>
#include <future>
#include <iostream>
#include <memory>
#include <numeric>
#include <vector>

using namespace std;

template <class T> auto post_in_thread_pool(T &&work) {
  std::async(std::launch::async, work);
}

template <class Terminal_T> struct Accumulator {
  std::shared_ptr<atomic<int>> counter;
  std::shared_ptr<atomic<int>> accumulator;
  Terminal_T func;
  std::shared_ptr<Accumulator> parent;

  void operator()(int value) {
    *accumulator += value;
    if (--*counter == 0) {
      if (parent)
        (*parent)(*accumulator);
      else
        func(*accumulator);
    }
  }
};

template <class T>
auto make_shared_accumulator(T func, int nb_leaves,
                             std::shared_ptr<Accumulator<T>> parent = nullptr) {
  return make_shared<Accumulator<T>>(
      Accumulator<T>{make_shared<atomic<int>>(nb_leaves),
                     make_shared<atomic<int>>(0), func, parent});
}

template <class Begin_T, class End_T, class Predicate_T, class Continuation_T>
auto async_count_if(Begin_T begin, End_T end, Predicate_T predicate,
                    Continuation_T continuation) {
  auto sz = end - begin;

  // (0) Base case:
  if (sz < 64) {
    (*continuation)(std::count_if(begin, end, predicate));
    return;
  }

  // (1) Recursive case:
  auto counter = make_shared<atomic<int>>(2); // (2)
  auto cleanup = make_shared_accumulator(continuation->func, 2, continuation);
  const auto mid = std::next(begin, sz / 2);

  post_in_thread_pool([=] { async_count_if(begin, mid, predicate, cleanup); });

  async_count_if(mid, end, predicate, cleanup);
}

int main() {
  std::vector<int> v(512);
  std::iota(std::begin(v), std::end(v), 0);

  std::vector<std::future<size_t>> results;

  auto res_func = [](int res) { std::cout << res << std::endl; };
  async_count_if(std::begin(v), std::end(v),
                 /*    predicate */ [](auto x) { return x < 256; },
                 /* continuation */
                 make_shared_accumulator(res_func, 1));
}

On Coliru。它并不完美，通过使用引用包装器可以避免很多无用的副本（可能还有其他优化可以 完成），但我试图保持示例的解释性超过优化。

问题在于，要适应具有多个不同累加器的更复杂的数据流并不容易，我想这是你的真实情况。

您正在尝试实现并行化数据计算管道。这不是一个可以通过语法技巧解决的简单问题。您需要一种线程安全的方式来在您的任务之间进行通信，这种方式既不是递归的，也不是线程阻塞的。

单靠标准库是不够的。您能做的最好的事情就是基于期货的不稳定实现。

要摆脱这个陷阱，您需要更多工具。您可以考虑使用TensorFlow 来实现您的计算模型。您还可以使用实验性框架，例如 Boson 或 RaftLib（多线程尚未在此实现）。或者实现你自己的，但要注意，要做到这一点需要做很多工作。