在 C++ 中执行向量交集答案

【问题标题】：performing vector intersection in C++在 C++ 中执行向量交集
【发布时间】：2015-07-20 09:13:55
【问题描述】：

我有一个无符号向量的向量。我需要找到所有这些无符号向量的交集，为此我编写了以下代码：

int func()
{
   vector<vector<unsigned> > t;
   vector<unsigned> intersectedValues;
   bool firstIntersection=true;
   for(int i=0;i<(t).size();i++)
   {
       if(firstIntersection)
       {
           intersectedValues=t[0];
           firstIntersection=false;
       }else{
           vector<unsigned> tempIntersectedSubjects;                                                              
           set_intersection(t[i].begin(),
                  t[i].end(), intersectedValues.begin(),
                  intersectedValues.end(),
                  std::inserter(tempIntersectedSubjects, tempIntersectedSubjects.begin()));
           intersectedValues=tempIntersectedSubjects;
       }         
       if(intersectedValues.size()==0)
           break;
   }               
}

每个单独的向量有 9000 个元素，“t”中有很多这样的向量。当我分析我的代码时，我发现 set_intersection 花费的时间最多，因此在调用 func() 时会导致代码变慢。有人可以建议我如何使代码更有效率。

我正在使用：gcc (GCC) 4.8.2 20140120 (Red Hat 4.8.2-15)

编辑：对向量“t”中的各个向量进行排序。

【问题讨论】：

你的向量排序了吗？您也可以尝试在末尾而不是开头插入找到的元素。
@Galik 是的，向量已排序。
你知道交叉口的大小吗？如果交集预计“小”，那可能会改变算法。
@StegVerner 您列出了编译器，但没有发布使用的编译器选项。如果您正在分析未优化的代码，那么在您构建程序的优化版本之前，您的结果将毫无意义。
@PaulMcKenzie 我使用的编译器选项是 O3

标签： c++ c++11 gcc vector

【解决方案1】：

您可以通过在编译期间定义_GLIBCXX_PARALLEL 使std::set_intersection 以及一堆其他标准库算法并行运行。这可能具有最佳的工作增益比。有关文档，请参阅 this。

强制性陷阱警告：

请注意，_GLIBCXX_PARALLEL 定义可能会改变标准类模板（例如std::search）的大小和行为，因此如果没有容器的实例化，则只能链接以并行模式编译的代码和非并行模式编译的代码在两个翻译单元之间传递。并行模式功能具有明显的联系，不能与普通模式符号混淆。

来自here。

另一个简单的优化是reserving 在填充向量之前有足够的空间，虽然可能很小。

另外，尝试找出在后面而不是前面插入值然后反转向量是否有帮助。（尽管我什至认为您的代码现在是错误的，并且您的 intersectedValues 排序方式错误。如果我没记错的话，您应该使用 std::back_inserter 而不是 std::inserter(...,begin) 然后不反向。）虽然通过内存转移内容非常快，但不转移应该更快。

【讨论】：

我在哪里定义 _GLIBCXX_PARALLEL 。我是新手，你能帮我解决一下吗
@StegVerner 与-D_GLIBCXX_PARALLEL 编译。
我正在使用 netbeans，如何指示 netbeans 使用此选项
@StegVerner 我不知道，我从未使用过它。阅读文档或谷歌它，如果它是一个不错的 IDE，应该有类似“构建选项”的东西，你可以在其中添加定义。
+1 用于保留向量内存。你知道tempIntersectedSubjects 将等于或小于intersectedValues

【解决方案2】：

我无法对此进行测试，但也许这样的东西会更快？

int func()
{
   vector<vector<unsigned> > t;
   vector<unsigned> intersectedValues;

   // remove if() branching from loop

   if(t.empty())
       return -1;

   intersectedValues = t[0];

   // now start from 1
   for(size_t i = 1; i < t.size(); ++i)
   {
       vector<unsigned> tempIntersectedSubjects;
       tempIntersectedSubjects.reserve(intersectedValues.size()); // pre-allocate

       // insert at end() not begin()
       set_intersection(t[i].begin(),
              t[i].end(), intersectedValues.begin(),
              intersectedValues.end(),
              std::inserter(tempIntersectedSubjects, tempIntersectedSubjects.end()));

       // as these are not used again you can move them rather than copy
       intersectedValues = std::move(tempIntersectedSubjects);

       if(intersectedValues.empty())
           break;
   }
   return 0;
}

另一种可能性：

考虑使用swap() 可以优化数据交换并消除重新分配的需要。然后临时构造函数也可以移出循环。

int func()
{
   vector<vector<unsigned> > t;
   vector<unsigned> intersectedValues;

   // remove if() branching from loop

   if(t.empty())
       return -1;

   intersectedValues = t[0];

   // no need to construct this every loop
   vector<unsigned> tempIntersectedSubjects;

   // now start from 1
   for(size_t i = 1; i < t.size(); ++i)
   {
       // should already be the correct size from previous loop
       // but just in case this should be cheep
       // (profile removing this line)
       tempIntersectedSubjects.reserve(intersectedValues.size());

       // insert at end() not begin()
       set_intersection(t[i].begin(),
              t[i].end(), intersectedValues.begin(),
              intersectedValues.end(),
              std::inserter(tempIntersectedSubjects, tempIntersectedSubjects.end()));

       // swap should leave tempIntersectedSubjects preallocated to the
       // correct size
       intersectedValues.swap(tempIntersectedSubjects);
       tempIntersectedSubjects.clear(); // will not deallocate

       if(intersectedValues.empty())
           break;
   }
   return 0;
}

【讨论】：

不 std::vector 默认调用移动赋值运算符吗？我们需要调用 std::move() 吗？
@Steephen 否，因为tempIntersectedSubjects 是一个左值（它是一个命名变量）。只有编译器知道永远不会再次使用的未命名临时对象才会自动移动。
@Steephen 我有了一个新想法——使用swap() 可能更有效（添加到答案中）
带有 emplce_back () 的简单 for 循环比交换更好，我在代码中观察到了！！
@Steephen 真的吗？我以为swap() 是O(1) - 数据指针的简单交换？

【解决方案3】：

我没有分析操作的框架，但我肯定会更改代码以重用易于分配的向量。此外，我会将初始交叉点提升到循环之外。此外，std::back_inserter() 应确保将元素添加到正确的位置而不是开头：

int func()
{
    vector<vector<unsigned> > t = some_initialization();
    if (t.empty()) {
        return;
    }
    vector<unsigned> intersectedValues(t[0]);
    vector<unsigned> tempIntersectedSubjects;
    for (std::vector<std::vector<unsigned>>::size_type i(1u);
         i < t.size() && !intersectedValues.empty(); ++i) {
        std::set_intersection(t[i].begin(), t[i].end(),
                              intersectedValues.begin(), intersectedValues.end(),
                             std::back_inserter(tempIntersectedSubjects);
        std::swap(intersectedValues, tempIntersectedSubjects);
        tempIntersectedSubjects.clear();
    }
}

我认为这段代码很有可能更快。将不同的集合相交也可能是合理的：您可以为成对的相邻集合创建一个新的交集，然后将第一个集合与其相关的相邻集合相交，而不是保留一个集合并与之相交：

std::vector<std::vector<unsigned>> intersections(
    std::vector<std::vector<unsigned>> const& t) {
    std::vector<std::vector<unsigned>> r;
    std::vector<std::vector<unsignned>>::size_type i(0);
    for (; i + 1 < t.size(); i += 2) {
        r.push_back(intersect(t[i], t[i + 1]));
    }
    if (i < t.size()) {
        r.push_back(t[i]);
    }
    return r;
}

std::vector<unsigned> func(std::vector<std::vector<unsigned>> const& t) {
    if (t.empty()) { /* deal with t being empty... */ }
    std::vector<std::vector<unsigned>> r(intersections(t))
    return r.size() == 1? r[0]: func(r);
}

当然，您不会真的像这样实现它：您会使用 Stepanov 的 二进制计数器 来保留中间集。这种方法假设结果很可能是非空的。如果期望结果是空的，那可能不是改进。

【讨论】：

根据 OP，for 循环内的变量 intersectedValues 似乎需要重新初始化

【解决方案4】：

使用 emplace_back() 从向量 for 循环中复制向量中的元素可能会节省您的时间。如果更改 for 循环的迭代器索引，则不需要标志。所以可以优化for循环，每次迭代都可以去掉条件检查。

void func()
{
   vector<vector<unsigned > > t;
   vector<unsigned int > intersectedValues;

   for(unsigned int i=1;i<(t).size();i++)
   {
        intersectedValues=t[0];

        vector<unsigned > tempIntersectedSubjects;                                                              
        set_intersection(t[i].begin(),
                  t[i].end(), intersectedValues.begin(),
                  intersectedValues.end(),
                   std::back_inserter(tempIntersectedSubjects);
        for(auto &ele: tempIntersectedSubjects)
                intersectedValues.emplace_back(ele);

        if( intersectedValues.empty())
          break;
   }               
}

【讨论】：

【解决方案5】：

set::set_intersection 对于大型向量可能会相当慢。可以使用创建一个使用lower_bound 的类似函数。像这样的：

template<typename Iterator1, typename Iterator2, typename Function>
void lower_bound_intersection(Iterator1 begin_1, Iterator1 end_1, Iterator2 begin_2, Iterator2 end_2, Function func)
{
    for (; begin_1 != end_1 && begin_2 != end_2;)
    {
        if (*begin_1 < *begin_2)
        {
            begin_1 = begin_1.lower_bound(*begin_2);
            //++begin_1;
        }
        else if (*begin_2 < *begin_1)
        {
            begin_2 = begin_2.lower_bound(*begin_1);
            //++begin_2;
        }
        else // equivalent
        {
            func(*begin_1);
            ++begin_1;
            ++begin_2;
        }
    }
}

【讨论】：