Проблемы с производительностью при соединении потоков

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, для которого вы можете подождать.

Глядя на вашу реализацию, большая проблема заключается в том, что ваши рабочие потоки не сбалансированы в их потреблении.

Чтобы получить более сбалансированную нагрузку на все ваши потоки, вам необходимо сгладить структуру данных, чтобы различные рабочие потоки могли обрабатывать фрагменты данных с аналогичным размером. Я не уверен, откуда берутся ваши данные, но вектор вектора в приложении, который имеет дело с большими наборами данных, не похож на отличную идею. Либо обрабатывайте существующий вектор векторов в один, либо читайте данные, как это возможно. Если вам нужен номер строки для вашей обработки, вы можете сохранить вектор диапазона начала, из которого вы можете найти номер строки.

Как только у вас есть один большой вектор, вы можете разбить его на равные по размеру куски для подачи в рабочие потоки. Во-вторых, вы не хотите создавать векторы при передаче стека и подталкивать их к другому вектору, потому что, скорее всего, вы сталкиваетесь с проблемами для выделения памяти во время работы ваших потоков. Выделение памяти - это глобальное изменение состояния и, как таковое, требует некоторого уровня блокировки (при правильном разбиении адресов его можно было бы избежать). Как правило, всякий раз, когда вы ищете производительность, вы должны удалить динамическое распределение из критически важных компонентов.

В этом случае, возможно, ваши потоки будут скорее «отмечать» элементы, удовлетворяющие условиям, а не строят векторы удовлетворяющих элементов. И как только это будет сделано, вы можете перебирать только хорошие, не нажимая и не копируя что-либо. Такое решение было бы менее бесполезным.

Фактически, если бы я был вами, я бы попытался решить эту проблему сначала в одной теме, выполнив приведенные выше предложения. Если вы избавитесь от структуры вектор-векторов и условно перейдете по элементам (это может быть так же просто, как и использование алгоритмов xxxx_if, предоставленных стандартной библиотекой C++ 11), вы можете получить достаточно приличную производительность. И только в этот момент стоит рассмотреть делегирование кусков этой работы рабочим потокам. На данный момент в вашем коде есть очень мало оснований использовать рабочие потоки, просто чтобы их отфильтровать. Делайте как можно меньше писем и движений, и вы получаете много работы. Параллелизация работает только при определенных обстоятельствах.