スレッドを結合する際のパフォーマンスの問題

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に通知する必要があります。

ここでのより大きな問題は、ワーカースレッドの消費が均衡していないことです。

すべてのスレッドに対してよりバランスのとれた負荷を得るには、データ構造をフラット化する必要があります。そのため、異なるワーカースレッドは、比較的類似したサイズのデータ​​チャンクを処理できます。私はあなたのデータがどこから来ているのかは分かりませんが、大規模なデータセットを扱うアプリケーション内のベクトルのベクトルを持つことは素晴らしいアイデアのようには聞こえません。既存のベクトルベクトルを単一のベクトルに処理するか、可能であればそのような方法でデータを読み込みます。処理に行番号が必要な場合は、行番号を見つけるための開始範囲のベクトルを保持することができます。

大きなベクトルが1つあれば、同じサイズのチャンクに分割してワーカースレッドに供給することができます。次に、スレッドの処理中にメモリを割り当てるために問題にぶつかっているので、スタック上にベクタを作成して別のベクタにプッシュする必要はありません。メモリの割り当てはグローバルな状態変更であり、ある程度のレベルのロックが必要です（適切なアドレス分割で回避することもできます）。経験則として、パフォーマンスを求めている場合は、パフォーマンスの重要な部分から動的割り当てを削除する必要があります。

この場合、おそらくあなたのスレッドは満足しているelemsのベクトルを構築するのではなく、むしろ 'mark'要素が満足できる条件です。そしてそれが終わると、何かを押したりコピーしたりすることなく、良いものだけを繰り返すことができます。このような解決策は無駄が少なくなる。

実際、もし私があなただったら、最初にこの問題を1つのスレッドで解決し、上記の提案をしてください。 vector-of-vectors構造体を取り除き、条件付きで要素を反復処理すると（これはC++ 11の標準ライブラリが提供するxxxx_ifアルゴリズムを使用するのと同じくらい簡単かもしれません）、十分なパフォーマンスを得ることができます。その時点で、この作業のチャンクをワーカースレッドに委任するだけの価値があります。コード化された時点で、ワーカースレッドをフィルタリングするために使用する正当な理由はほとんどありません。できるだけ書いたり動いたりすることはほとんどなく、多くのパフォーマンスを得ることができます。パラレル化は特定の状況でのみうまく機能します。