连接线程时的性能问题

Question

我写了下面的并行代码来检查向量中的所有元素。我只存储vector<vector<int> >中满足给定条件的元素。然而，我的问题是，vector<vector<int> >中的一些向量相当大，而另一些却很小。由于这个原因，我的代码需要很长时间来执行thread.join()。有人能建议我如何提高我的代码的性能吗？

void check_if_condition(vector<int>& a, vector<int>& satisfyingElements)
{
    for(vector<int>::iterator i1=a.begin(), l1=a.end(); i1!=l1; ++i1)
        if(some_check_condition(*i1))
            satisfyingElements.push_back(*i1);

}

void doWork(std::vector<vector<int> >& myVec, std::vector<vector<int> >& results, size_t current, size_t end)
{
    end = std::min(end, myVec.size());
    int numPassed = 0;
    for(; current < end; ++current) {
        vector<int> satisfyingElements;
        check_if_condition(myVec[current], satisfyingElements); 
        if(!satisfyingElements.empty()){
            results[current] = satisfyingElements;            
        }
    }    
}

int main()
{
    std::vector<std::vector<int> > myVec(1000000);
    std::vector<std::vector<int> > results(myVec.size());   
    unsigned numparallelThreads = std::thread::hardware_concurrency();

    std::vector<std::thread> parallelThreads;
    auto blockSize = myVec.size() / numparallelThreads;
    for(size_t i = 0; i < numparallelThreads - 1; ++i) {
        parallelThreads.emplace_back(doWork, std::ref(myVec), std::ref(results), i * blockSize, (i+1) * blockSize);
    }

    //also do work in this thread
    doWork(myVec, results, (numparallelThreads-1) * blockSize, myVec.size());

    for(auto& thread : parallelThreads)
        thread.join();

    std::vector<int> storage;
    storage.reserve(numPassed.load());

    auto itRes = results.begin();
    auto itmyVec = myVec.begin();
    auto endRes = results.end();
    for(; itRes != endRes; ++itRes, ++itmyVec) {
        if(!(*itRes).empty())
            storage.insert(storage.begin(),(*itRes).begin(), (*itRes).end());
    }

    std::cout << "Done" << std::endl;
}

Answer 1

如果你能给出这些“大”内部向量的一些尺度，看看问题有多糟糕，那就太好了。

然而，我认为你的问题是：

for(auto& thread : parallelThreads)
    thread.join();

这个位使顺序地遍历所有线程，并等待它们完成，然后才会查看下一个线程。对于线程池，您需要等待直到每个线程都完成。这可以通过对每个线程使用condition_variable来完成。在它们完成之前，它们必须通知condition_variable，您可以等待。

看看你的实现，这里更大的问题是你的工作线程在消耗上不平衡。

为了在所有线程上获得更均衡的负载，您需要扁平化数据结构，以便不同的工作线程可以处理相对相似大小的数据块。我不确定你的数据是从哪里来的，但是在处理大型数据集的应用程序中使用向量的向量听起来不是一个好主意。或者将现有的向量处理成一个单独的向量，或者如果可能的话，像这样读取数据。如果您需要处理行号，您可以保留一个起始-结束范围向量，从中可以找到您的行号。

一旦你有了一个大的向量，你就可以将它分解成大小相等的块，以馈送到工作线程中。其次，您不希望在堆栈上构建向量来处理并将它们推入另一个向量中，因为在线程工作期间，您很可能会遇到分配内存的问题。分配内存是一种全局状态更改，因此需要一定级别的锁定(通过适当的地址分区，可以避免这种情况)。作为一条经验法则，每当您在寻找性能时，都应该从性能关键部分中删除动态分配。

在这种情况下，也许你的线程宁愿‘标记’满足条件的元素，而不是构建令人满意的元素的向量。一旦这样做了，你就可以只迭代那些好的，而不需要推送和复制任何东西。这样的解决方案会减少浪费。

事实上，如果我是你，我会先试着在一个线程上解决这个问题，做上面的建议。如果您摆脱了向量的向量结构，并有条件地遍历元素(这可能与使用C++11标准库提供的xxxx_if算法一样简单)，那么您最终可能会获得足够好的性能。只有在这一点上，才值得考虑将这项工作的大块委托给工作线程。在代码中的这一点上，几乎没有理由使用工作线程，只是为了过滤它们。尽可能少地写字和移动，你就会获得更多的性能。并行化只有在某些情况下才能很好地工作。