多线程时，您可以做些什么来停止耗尽堆栈空间？

Question

我已经在 C++ 中实现了一个有效的多线程合并排序，但我碰壁了。

在我的实现中，我递归地将一个输入向量分成两部分，然后将这两部分线程化：

void MergeSort(vector<int> *in)
{
if(in->size() < 2)
    return;

vector<int>::iterator ite = in->begin();
vector<int> left = vector<int> (ite, ite + in->size()/2);
vector<int> right = vector<int> (ite + in->size()/2, in->end() );

//current thread spawns 2 threads HERE
thread t1 = thread(MergeSort, &left);
thread t2 = thread(MergeSort, &right);

t1.join();
t2.join();

vector<int> ret;
ret.reserve(in->size() );

ret = MergeSortMerge(left, right);

in->clear();
in->insert(in->begin(), ret.begin(), ret.end() );

return;
}

代码看起来很漂亮，但它是我写过的最恶毒的代码之一。尝试对超过 1000 个 int 值的数组进行排序会导致产生如此多的线程，以至于我的堆栈空间不足，并且我的计算机蓝屏 :( 始终如一。

我很清楚这段代码产生这么多线程的原因，这不是很好，但从技术上（如果不是理论上），这不是一个正确的实现吗？

根据一些谷歌搜索，我似乎发现需要一个线程池。使用线程池会解决我遇到的基本问题，即我试图产生太多线程的事实吗？如果有，您对图书馆有什么建议吗？

感谢您的建议和帮助！

Answer 1

我认为线程池不会帮助您。由于您的算法是递归的，您将到达池中的所有线程都被消耗并且池不想再创建任何线程并且您的算法将阻塞的地步。

您可能只需将线程创建递归深度限制为 2 或 3（除非您有很多 CPU，否则不会对性能产生任何影响）。

Answer 2

正如 zdan 解释的那样，您应该限制线程数。有两件事要考虑确定什么是限制，

1. CPU 内核数。在 C++11 中，您可以使用std::thread::hardware_concurrency() 来确定硬件内核。但是，这个函数可能返回 0，表示程序不知道有多少核，在这种情况下，你可以假设这个值是 2 或 4。

2. 受要处理的数据数量的限制。您可以将要由线程处理的数据划分为每个线程 1 个数据，但仅 1 个数据会花费太多，而且成本效率不高。例如，您可能会说，当数据数量小于 50 时，您不想再划分了。因此，您可以根据total_data_number / 50 + 1 之类的内容确定所需的最大线程数。

然后，您在案例 1 和案例 2 之间选择一个最小数量来确定限制。

在您的情况下，由于您是通过递归生成线程，因此您可以尝试以类似的方式确定递归深度。

Answer 3

您可以设置堆栈空间的限制，但这是徒劳的。太多的线程，即使有一个池，也会以 log2(N)*cost per thread 吃掉它。采用迭代方法并减少开销。头顶是杀手。 就性能而言，您会发现使用 N 线程的某种程度的过度提交，硬件并发可能会产生最佳结果。开销和每个核心的工作量之间会有一个很好的平衡。如果 N get 非常大，例如在 GPU 上，则存在其他选项（双音），它们进行不同的权衡以减少通信（等待/加入）开销。

假设您有一个任务管理器和一个为 N 构造的信号量，在允许等待任务通过之前通知， `

#include <algorithm>
#include <array>
#include <cstdint>
#include <vector>
#include <sometaskmanager.h>

void parallel_merge( size_t N ) {
    std::array<int, 1000> ary {0};
    // fill array...
    intmax_t stride_size = ary.size( )/N;  //TODO: Put a MIN size here
    auto semaphore = make_semaphore( N );
    using iterator = typename std::array<int, 1000>::iterator;
    std::vector<std::pair<iterator, iterator>> ranges;
    auto last_it = ary.begin( );
    for( intmax_t n=stride_size; n<N; n +=stride_size  ) {
      ranges.emplace_back( last_it, std::next(last_it, std::min(std::distance(last_it, ary.end()), stride_size)));
      semaphore.notify( );
    }
    for( auto const & rng: ranges ) {
      add_task( [&semaphore,rng]( ) {
        std::sort( rng.first, rng.second );
      });
    }
    semaphore.wait( );
    std::vector<std::pair<iterator, iterator>> new_rng;
    while( ranges.size( ) > 1 ) {
        semaphore = make_semaphore( ranges.size( )/2 );
        for( size_t n=0; n<ranges.size( ); n+=2 ) {
            auto first=ranges[n].first;
            auto last=ranges[n+1].second;
            add_task( [&semaphore, first, mid=ranges[n].second, last]( ) {
                std::inplace_merge( first, mid, last );
                semaphore.notify( );
            });
            new_rng.emplace_back( first, last );
        }
        if( ranges.size( ) % 2 != 0 ) {
            new_rng.push_back( ranges.back( ) );
        }
        ranges = new_rng;
        semaphore.wait( );
    }
}

如您所见，瓶颈在合并阶段，因为必须完成很多协调。 Sean Parent 在他的演示 Better Code: Concurrency, http://sean-parent.stlab.cc/presentations/2016-11-16-concurrency/2016-11-16-concurrency.pdf 中很好地介绍了如何构建任务管理器（如果您没有任务管理器）以及它的比较以及相对性能分析。 TBB 和 PPL 都有任务管理器。