排序向量的交集答案

【问题标题】：Intersection of sorted vectors排序向量的交集
【发布时间】：2015-08-06 10:08:55
【问题描述】：

我知道两个已排序向量或集合之间的交集可以使用 std::set_intersection() 来执行。是否可以使用 openMP 4.0 SIMD 执行相同的集合交集。我需要在我的代码中多次执行两个排序向量之间的集合交集，因此 c++ set_intersection() 在这里成为瓶颈（由 perf 工具识别）。是否可以使用 SIMD 执行 set_intersection 以加快大向量之间的集合交集。如果是，那么如何？

我正在使用 gcc-4.9.2

我需要在两个排序后的向量之间进行交集——其中第一个向量的大小最多为 1,000，第二个向量的大小最多为 10,000 个元素。

如果无法使用 openMP 4.0 SIMD 执行 set_intersection，那么是否可以使用 Boost SIMD 执行 set_intersection。如果是，那么如何。

提前非常感谢

【问题讨论】：

抱歉，我在提交标量逻辑的答案后才注意到 SIMD 要求。鉴于这种串行 I/O 特性，我不确定您是否可以使用 SIMD 来完成，而且因为 std::set_intersection 是做繁重工作的人。如果有任何可能对此进行矢量化，那么 OMP 或 Boost 可能不会自动解决它，而是通过使用（或可能创建）实际上设计为矢量化的 set_intersection 例程（不是 std::set_intersection）来解决。如果可能的话，OMP 或 Boost SIMD 甚至自己编写内部函数都应该可以工作。
找到这个 -- : highlyscalable.wordpress.com/2012/06/05/…
@lke 你能分享你的标量逻辑吗？实际上，我想将标量逻辑的性能与 SIMD 进行比较。实际上，在您发表评论之后，我也认为标量逻辑应该比 SIMD 执行得更好——所以让我们实际尝试一下
您有兴趣使用 SSE（或 AVX）进行此操作吗？如果是这样，那么添加 SSE 标记，我会看看我是否有时间使用 SSE 和 OpenMP 组合一个结果。这是什么类型的 (int32_t, int64_t, float, double, ...)？
@StegVerner 哦，抱歉回复晚了，虽然Z boson 有一个很好的答案。

标签： c++ openmp simd

【解决方案1】：

这里有一些 OpenMP 代码，我将它们放在一起来查找两个排序集的交集。合并结果时可以在没有关键部分的情况下执行此操作（以及将它们并行排序），但我在这里没有这样做。

使用 SIMD 也可能（有效地）做到这一点。我会明确地使用内在或 SIMD 向量类来做到这一点。

size_t intersect_scalar_omp(int *A, int *B, size_t s_a, size_t s_b, int *C) {
    size_t counter = 0;
    #pragma omp parallel
    {
        size_t i_a = 0, i_b = 0;
        size_t aend = s_a, bend = s_b;
        size_t counter_private = 0;
        int ithread = omp_get_thread_num();
        int nthreads = omp_get_num_threads();   

        if (s_a > s_b) {
            i_a = ithread*s_a / nthreads;
            aend = (ithread + 1)*s_a / nthreads;            
        }
        else {
            i_b = ithread*s_b / nthreads;
            bend = (ithread + 1)*s_b / nthreads;
        }
        int *C_private = new int[aend > bend ? aend : bend];
        while (i_a < aend && i_b < bend) {
            if (A[i_a] < B[i_b]) {
                i_a++;
            }
            else if (B[i_b] < A[i_a]) {
                i_b++;
            }
            else {
                C_private[counter_private++] = A[i_a];
                i_a++; i_b++;
            }
        }
        #pragma omp critical
        {
            memcpy(&C[counter], C_private, sizeof(int)*counter_private);
            counter += counter_private;
        }
        delete[] C_private;
    }
    std::sort(C, C + counter);
    return counter;
}

【讨论】：