如何優化檢查在載體元件在C++

Question

vector<unsigned> listNumbers; //listNumbers vector contains about 1000 million elements 
vector<unsigned> storageVec; 
for(vector<unsigned>::iterator it=listNumbers.begin(), l=listNumbers.end(); it!=l; ++it) 
{ 
      if(check_if_condition_satisfied(*it)) 
        storageVec.push_back(*it); 
} 

載體「listNumbers」包含有關一個十億元件和我需要檢查在「listNumbers」的元素是否滿足一定條件使用check_if_condition_satisfied。不幸的是，一次檢查check_if_condition_satisfied元素非常耗時。是否有某種方式可以使C++中的檢查並行化11

Answer 1

您可以使用std::future和std::async輕鬆並行計算。

未測試（僞）代碼：

// You need to keep track of all futures you spawn. 
std::vector<std::future<bool>> futs; 

// Spawn a future for every "block" that you want to check.  
futs.emplace_back(std::async(std::launch::async, 
    [&]{ return check_condition_range(listNumbers, 0, 10000); })); 

futs.emplace_back(std::async(std::launch::async, 
    [&]{ return check_condition_range(listNumbers, 10000, 20000); })); 

// (Ideally you would divide your range evenly depending on the number of threads you 
// can spawn, and generate the futures in a loop.) 

// ... 

// "Accumulate" the results of every future. 
// This blocks until all futures are done. 
bool satisfied = true; 
for(auto& f : futs) 
{ 
    if(!f.get()) 
    { 
     satisfied = false; 
    } 
} 

Here's a simple example on wandbox。

---

如果你有機會訪問C++ 17，你可以簡單地使用：

std::all_of(std::execution::par, 
    std::begin(listNumbers), std::end(listNumbers),   
    check_if_condition_satisfied); 

---

對不起，我還需要將結果存儲在一個共同的載體。

你的未來型只需更改爲std::future<decltype(listNumbers)::iterator>，讓每一位將來會返回一個迭代器，而不是一個bool的。

之後，你可以做一些與此類似：

std::vector<decltype(listNumbers)::iterator> result; 
for(auto& f : futs) result.emplace_back(f.get()); 
// use `result` 

Answer 2

你可以控制你創建到底有多少線程計算並行你的任務精確。下面是一個明確的解決方案，併爲每個線程分配連續的輸入向量區域，以將結果保存到二進制向量中。主線程本身也參與工作量。最後，所有傳遞的結果都將被複制到存儲向量中，以單獨傳遞（以避免直接從檢查線程直接寫入它的爭用開銷）。

住址代碼here。

#include <thread> 
#include <vector> 
#include <iostream> 
#include <atomic> 

bool check_if_condition(int a) 
{ 
    return true; 
} 

void doWork(std::vector<int>& input, std::vector<bool>& results, size_t current, size_t end, std::atomic_int& totalPassed) 
{ 
    end = std::min(end, input.size()); 
    int numPassed = 0; 
    for(; current < end; ++current) { 
     if(check_if_condition(input[current])) { 
      results[current] = true; 
      ++numPassed; 
     } 
    } 

    totalPassed.fetch_add(numPassed); 
} 

int main() 
{ 
    std::vector<int> input(1000000); 
    std::vector<bool> results(input.size()); 
    std::atomic_int numPassed(0); 

    auto numThreads = std::thread::hardware_concurrency(); 

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

    //also do work in this thread 
    doWork(input, results, (numThreads-1) * blockSize, numThreads*blockSize, numPassed); 

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

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

    auto itRes = results.begin(); 
    auto itInput = input.begin(); 
    auto endRes = results.end(); 
    for(; itRes != endRes; ++itRes, ++itInput) { 
     if(*itRes) 
      storage.emplace_back(*itInput); 
    } 

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