在 C++ 中并行查找向量的第一个

Posted 2023-02-21

【中文标题】在 C++ 中并行查找向量的第一个

【英文标题】：find_first of a vector in parallel in C++

【发布时间】：2016-10-27 12:50:33

【问题描述】：

我有一个很大的向量。一些向量成员并行匹配某个条件。我想找到与条件匹配的第一个元素。

我的问题与这个问题 (tbb: parallel find first element) 非常相似，但我没有 tbb。检查条件非常繁琐（所以我不能按顺序为所有这些检查）。这就是为什么我想并行运行它。不得不提的是，我想找到第一个元素（所以元素的索引位置对我来说很重要）。

例如，如果我有 4 个线程。

ThreadNr   Index      condition
1            0         Not Meet
2            1         Not Meet
3            2         Not Meet
4            3         Not Meet

ThreadNr   Index      condition
1            4         Not Meet
2            5          Meet
3            6         Not Meet
4            7          Meet

函数必须重新调整索引号 5。 线程必须分布并处理顺序迭代块（块大小可以大于 1。例如线程 1 处理前 4 个元素，线程 2 处理后 4 个元素，依此类推）。

对于上面的例子，如果线程 4（索引 7）在线程 2（索引 5）之前找到成员，它必须等待所有线程完成工作。正如我之前所说，最低的索引号是目标。

如果您有更好的算法，请纠正我。

注意：我可以使用外部库，例如 boost 1.62、OpenMP 2.0

【问题讨论】：

请注意，最佳情况复杂度从 0 增加到 N/K，其中 N 是元素总数，K 是线程数。

这很大程度上取决于元素的大小以及检查条件需要多长时间。如果你对std::vector<int> 做一些简单的事情，比如==，你不会想要这样交错，因为它会对缓存造成严重破坏。在这种情况下，您可能希望线程一次检查整数块。如果您的矢量对象是 1mb 一块，并且需要 1-2 秒来执行检查，那么交错可能更有意义。

如果 tbb 对您有吸引力，请使用它。在 Openmp 中，如果我理解这个问题，您会像对内部搜索循环所做的那样对数据进行分区，将内部结果组合成外部 omp 减少 firstprivate lastprivate 或critical。

【参考方案1】：

由于 OpenMP 2.0 没有取消构造，您必须自己实现一个，例如，通过使用共享变量。这也意味着您不能使用 for 工作共享结构，因为不允许中断并行循环（这就是 OpenMP 4.0 引入取消结构的原因）。如果您在每个元素的评估之间实施取消检查，则可能会发生两个或多个线程找到匹配标准的元素。因此，您应该对索引执行最小缩减：

int found = 0;
int first_index = INVALID_VALUE;
int iteration = 0;

#pragma omp parallel

   int my_index = INVALID_VALUE;
   int i;

   do
   
      // Later versions of OpenMP allow for "atomic capture"
      // but OpenMP 2.0 requires a critical directive instead
      #pragma omp critical(iteration)
      
         i = iteration++;
      

      if (i < N && check(i))
      
         found = 1;
         my_index = i;
      
    while (!found && i < N);

   #pragma omp critical(reduction)
   if (my_index != INVALID_VALUE)
   
      if (first_index == INVALID_VALUE || my_index < first_index)
         first_index = my_index;
   

   // Only needed if more code follows before the end of the region
   #pragma omp barrier

   ...

此代码假定检查第 i 个元素 (check(i)) 的条件不会改变元素的状态，因此，可能发生的最坏情况是找到匹配元素的线程可能等待所有其他线程完成检查它们当前正在处理的元素，并且等待时间将是所有处理时间的最大值。

do-loop 中使用的critical 结构很昂贵。如果check() 不需要那么多时间，那么您可以考虑使用块而不是迭代：

do

   #pragma omp critical(chunk)
   
       my_chunk = chunk++;
   

   if (my_chunk >= N_chunks)
      break;

   for (i = my_chunk * chunk_size; !found && i < (my_chunk+1)*chunk_size; i++)
   
      if (check(i))
      
         found = 1;
         my_index = i;
         break;
      
   
 while (!found && my_chunk < N_chunks);

---

当元素的数量不是那么大并且检查每个元素的成本很高时，另一种效果相当不错的解决方案：

#pragma omp parallel

   #pragma omp for schedule(dynamic,x)
   for (i = 0; i < N; i++)
   
      if (!found && check(i))
      
         my_index = i;
         found = 1;
      
   

   // Min reduction from the solution above
   ...

一旦found 变为真，其余的循环迭代将运行“空”主体，因为&& 的快捷方式属性。

【讨论】：

只是一个问题：对于最后一个解决方案，您不需要在某处刷新found 吗？

除了我之前的评论之外，前两个解决方案不需要flush，因为循环中的critical 部分暗示了一些解决方案。但是对于最后一个解决方案，found 没有隐式或显式刷新，因此没有找到解决方案的线程可以在缓存中保持不变。至少，我是这么看的。我错了吗？

found 最初是 volatile，然后我将其删除。事实上，宽松的内存模型允许共享变量在不同的线程中具有临时不同的值，但是正确使用flush 是很棘手的，我不想在这里介绍这个构造。此外，x86 上的大多数 OpenMP 编译器倾向于实现更严格的内存模型（分配后共享变量仅保留在寄存器中）并且flush 被编译为内存栅栏。如果没有刷新，它可能会导致额外的迭代，所以没什么大不了的。但为了正确起见，是的，应该有一个。

【参考方案2】：

使用 OpenMP，您可以尝试使用 #pragma omp for schedule(dynamic) 构建一个 for 循环。每个线程将按照与向量相同的顺序执行一次迭代。 如果你想通过线程检查4个元素，试试#pragma omp for schedule(dynamic, 4)

【讨论】：

谢谢，但它不尊重索引。例如，如果线程号 4 找到线程号 2 之前的元素。我可能会得到 wring index

我觉得如果条件ok的话，可以在写入索引的值退出循环之前加上#pragma omp ordered。