我有 200 个大小从 1 到 4000000 的 vector 存储在 vecOfVec 中。我需要将这些 vector 与大小为 9000+ 元素的单个 vector “vecSearched”相交。我尝试使用以下代码来做同样的事情，但是使用 perf 工具我发现我正在做的交集是我代码中的瓶颈。有什么方法可以执行有效的交叉

#include <cstdlib>
#include <iostream>
#include <vector>

using namespace std;

int main(int argc, char** argv) {
  vector<vector<unsigned> > vecOfVec; //contains 120 vectors of size ranging from 1 to 2000000 elements. All vectors in vecOfVec are sorted
  vector<unsigned> vecSearched; //vector searched in contains 9000+ elements. Vectors in vecSearched are sorted
  for(unsigned kbt=0; kbt<vecOfVec.size(); kbt++)
  {                                  
     //first find first 9 values spaced at equi-distant places, use these 9 values for performing comparisons
     vector<unsigned> equiSpacedVec;                             

     if(((vecSearched[0]))>vecOfVec[kbt][(vecOfVec[kbt].size())-1]) //if beginning of searched vector > last value present in individual vectors of vecOfVec then continue
     {
         continue;                             
     }                         

     unsigned elementIndex=0; //used for iterating over equiSpacedVec                             
     unsigned i=0; //used for iterating over individual buckets vecOfVec[kbt].second
     //search for value in bucket and store it in bucketValPos
     bool firstRun=true;             
     for(vector<unsigned>::iterator itValPos=vecSearched.begin();itValPos!=vecSearched.end();++itValPos)
     {
         //construct a summarized vector out of individual vectors of vecOfVec
         if(firstRun)
         {
             firstRun=false;
             unsigned elementIndex1=0; //used for iterating over equiSpacedVec
             while(elementIndex1<(vecOfVec[kbt].size())) //create a small vector for skipping over the remaining vectors
             {
                  if((elementIndex1+(10000))<(vecOfVec[kbt].size()))
                     elementIndex1+=10000;
                  else
                     break;
                 equiSpacedVec.push_back(vecOfVec[kbt][elementIndex1]);
             }          
         }
         //skip individual vectors of vecOfVec using summarized vector constructed above
         while((!(equiSpacedVec.empty()))&&(equiSpacedVec.size()>(elementIndex+1))&&((*itValPos)>equiSpacedVec[elementIndex+1])){
             elementIndex+=1;
             if((i+100)<(vecOfVec[kbt].size()))
                 i+=100;
         }            
         unsigned j=i;
         while(((*itValPos)>vecOfVec[kbt][j])&&(j<vecOfVec[kbt].size())){
             j++;
         }
         if(j>(vecOfVec[kbt].size()-1)) //element not found even at last position.
         {
             break;
         }              

         if((*itValPos)==vecOfVec[kbt][j])
         {                                     
             //store intersection result
         }
     }
  }
    return 0;
}

最佳答案

您的问题很受欢迎。由于您没有相关 vector 相交的数据，它归结为加速两个 vector 之间的交集，基本上有两种方法:

1。无需任何预处理

这通常由三件事来解决:

• 减少比较次数。例如，对于小 vector (大小为 1 到 50)，您应该对每个元素进行二进制搜索，以避免遍历主题 vector 的所有 9000+ 个元素。

• 提高代码质量以减少分支错误预测，例如，观察结果集通常比输入集的大小可以转换这样的代码:

  while (Apos < Aend && Bpos < Bend) {
      if (A[Apos] == B[Bpos]) {
          C[Cpos++] = A[Apos];
          Apos++; Bpos++;
      }
      else if (A[Apos] > B[Bpos]) {
          Bpos++;
      } 
      else {
        Apos++;
      } 
  }
  

  编写“展开”此类比较的代码，虽然更容易预测分支( block 大小 = 2 的示例):

  while (1) {
      Adat0 = A[Apos]; Adat1 = A[Apos + 1]; 
      Bdat0 = B[Bpos]; Bdat1 = B[Bpos + 1];
      if (Adat0 == Bdat0) { 
          C[Cpos++] = Adat0;
      }
      else if (Adat0 == Bdat1) {
          C[Cpos++] = Adat0;
          goto advanceB;
      }
      else if (Adat1 == Bdat0) {
          C[Cpos++] = Adat1;
          goto advanceA;
      }
      if (Adat1 == Bdat1) {
          C[Cpos++] = Adat1;
          goto advanceAB;
      }
      else if (Adat1 > Bdat1) goto advanceB;
      else goto advanceA;
  advanceA:
      Apos+=2;
      if (Apos >= Aend) { break; } else { continue; }
  advanceB:
      Bpos+=2;
      if (Bpos >= Bend) { break; } else { continue; }
  advanceAB:
      Apos+=2;  Bpos+=2;
      if (Apos >= Aend || Bpos >= Bend) { break; }
  }
  //  fall back to naive algorithm for remaining elements
  

• 使用SIMD指令进行 block 操作

这些技术很难在 QA 上下文中描述，但您可以阅读它们(加上相关的优化，如 if 转换)here和 here或查找实现元素 here

2。带预处理

恕我直言，这个方法更适合您的情况，因为您有一个大小为 9000 多个元素的单一主题 vector 。您可以从中制作一棵间隔树，或者简单地找到一种索引它的方法，例如制作一个可以加快搜索速度的结构:

vector<unsigned> subject(9000+); 
vector<range>    index(9000/M); 

其中 range 是一个类似的结构

struct range {
    unsigned min, max; 
}; 

这样就创建了一系列范围

[0, 100], [101, 857], ... [33221, 33500]

这将允许在进行交集时跳过许多比较(例如，如果另一个集合的元素大于子范围的最大值，您可以完全跳过该子范围)

3。并行化

是的，两个元素的列表中总有第三个元素 :P。当您对过程进行了足够的优化时(并且只有到那时)，请将您的工作分解成 block 并并行运行。该问题符合令人尴尬的模式，因此 200 vector 对 1 应该明确地运行为“50 对 1 四次并发”

^{测试、测量、重新设计!!}

关于c++ - 执行 vector c++的交集，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/31065273/