在C ++中实现并行Copy_If

在有关十二种过滤元素的博客文章中，我只提到了代码的串行版本。但是如何利用并发呢？也许我们可以抛出更多线程和异步任务并更快地完成复制？

例如，我的机器上有6个内核，因此很高兴看到，例如顺序复制速度提高了5倍？

在C ++ 17中，我们有并行算法，因此让我们尝试使用std :: execution :: par调用std :: copy_if。

如果在MSVC库中转到std :: copy_if的实现，则并行版本可以看到以下内容：

// VS 2019 16.8 //目前尚未并行化，预计将来的发行版中并行性将是可行的

免责声明：这些只是我的实验（主要是为了学习一些东西）；如果您想在项目中使用它，请进行测量，测量和测量:)

基本而言，C ++ 17的并行算法很容易实现。只需传递一个std :: execution :: par，您就完成了！例如：

该代码将调用一堆线程（可能利用一些现有的线程池），并将在多个线程上批量启动较小的任务。

我们应该记住，这种调用总是比顺序调用产生更多的工作！以及准备，设置批处理，启动线程池，同步的成本-这增加了整个处理的可见成本。

理想情况下，并行运行事物最适合许多对象，也适用于单独执行小任务的情况。一个完美的例子：

C ++ 17并行算法的惊人性能，可能吗？ -C ++故事在文章中，我展示了一些使用Fresnel和3D Vector的“真实”用例，并且加速几乎与系统中内核的数量成线性关系。

您会看到一些加速（当您有大量对象时），但是它与核的数量不是线性的。

这是因为sort需要重新整理容器中的内容，并且为了安全地进行操作，该算法必须执行一些同步操作，以便其他线程看到正确的结果。

对于我们的测试（除了简单的调试输出），我将使用以下代码。

const size_t VEC_SIZE = argc＆gt; 1个atoi（argv [1]）：10； std :: vector＆lt; std :: pair＆lt; double，double＆gt; testVec（VEC_SIZE）; std :: ranges :: generate（testVec.begin（），testVec.end（），[]（）可变{return std :: pair {GenRandom（-10.0，10.0），GenRandom（-10.0，10.0）};} ）; auto test = []（auto＆amp; elem）{auto sn = sin（elem.first）* cos（elem.second + 10.0）;返回sn＆gt; 0.0;};

总的来说，我想比elem％2 == 0拥有更多的计算能力。此外，每个元素都是16字节，因此对象也不是很小。

我们必须在所有元素上运行谓词函数-在大多数情况下，它不依赖于其他元素，因此最好在许多线程上执行

但是我们必须将匹配的元素放入新容器中。这是一个可变的步骤，需要线程之间进行一些同步。

首先，最好采用蛮力方法并从中学习：

模板＆lt;类型名称T，类型名称Pred＆gt;自动FilterCopyIfParNaive（const std :: vector＆lt; vec，Pred p）{std :: vector＆lt; T出去; std :: mutex mut; std :: for_each（std :: execution :: par，begin（vec），end（vec），[＆amp; out，＆amp; mut，p]（auto＆amp;＆amp; elem）{if（p（elem））{ std :: unique_lock lock（mut）; out.push_back（elem）;}}）;返回；}

由于std :: for_each和std :: execution :: par，我们并行运行所有步骤，但是当我们要将元素放入输出容器时，我们需要进行同步。

可以注意到，要保护修改容器状态的所有操作。

// 4核心/ 8线索屏蔽VEC尺寸：100000Transform仅SEQ：2.5878 MS，RET：100000Transform仅适用：1.3734 MS，RET：100000FilterCopyIF：5.3675 ​​MS，RET：50203FopyIFParnaive：9.1836 MS，RET：50203

// 6核心/ 12线程屏蔽VEC大小：100000Transform仅SEQ：2.223 MS，RET：100000Transform仅适用：0.5507 MS，RET：100000FilterCopyIF：3.851 MS，RET：50203FILterCopyIFParnaive：10.1295 MS，RET：50203

对于比较，我还包括仅转换SEQ并仅转换仅PAR，这只是一个简单的变换在集合中运行：

STD :: Vector＆lt; uint8_t＆gt;缓冲区（testvec.size（））; rundandmeasure（＆＃34;只转换SEQ＆＃34; [＆amp; testvec，＆amp; buffer，＆amp; test]（）{std :: transform（begin（testvec），结束（testvec），开始（缓冲区），测试）; return buffer.size（）;}）; rundandmeasure（＆＃34;只转换为＆＃34; [＆amp; testvec，＆amp; buffer，＆amp; test]（ ）{std :: transform（std ::执行:: par，begin（testvec），结束（testvec），begin（缓冲区），测试）; return buffer.size（）;}）;

请注意，缓冲区是在变换Lambda之外创建的，因此我们不支付其初始化的价格。看看它是如何用许多核心划衡的。

在所有输入元素上运行std :: transform以计算谓词函数，将布尔结果存储在临时容器中。

然后我们需要计算匹配元素的最终位置 - 这可以通过调用std :: exlusive_scan来完成

模板＆lt; typeName t，typename pred＆gt;自动筛选copyifparcompose（const std :: vertent＆lt; t＆gt;＆amp; vec，pred p）{std :: vector＆lt; uint8_t＆gt;缓冲区（vec.size（））; STD :: Vector＆lt; uint32_t＆gt; idx（vec.size（））; std :: transform（std ::执行:: par，begin（vec），结束（vec），begin（缓冲区），[＆amp; p]（const t＆amp; elem）{return p（elem）;}）; std :: fellusive_scan（std ::执行:: par，begin（beaffer），结束（缓冲区），begin（idx），0）; std :: vector＆lt; t＆gt; out（idx.back（）+ 1）; std :: vector＆lt; size_t＆gt;索引（vec.size（））; std :: iota（indexes.begin（），indexes.end（），0）; std :: for_each（std ::执行:: par，begine（索引），结束（索引），[＆amp; beaffer，＆amp; vec;＆amp; idx，＆amp; out]（size_t i）{if（buffer [i ]）出局[Idx [i]] = VEC [i];}）;返回;}

输入：0,1,1,0,1，1,0,1,1，1，1，1,0,1，1，1，1，1，1，1，1，2， 2,3,4,4,5，6out：1,2,4,5,7,8,9

所以...是的，它有效，在某些情况下，它将比顺序版本更快。

让我们有一个测试运行。这可以比顺序版本快吗？

// 4核心/ 8个线索博音号VEC大小：100000Transform仅SEQ：2.5878 MS，RET：100000Transform仅适用：1.3734 MS，RET：100000FilterCopyIF：5.3675 ​​MS，RET：50203FopyIFParCAPE：9.1836 MS，RET：3.03 MS，RET：50203FILTERCOPYIFPARCOMPOSEQ ：2.3454 MS，RET：50203FilterCopyIFPartransformPush：2.5735 MS，RET：50203

// 6核心/ 12线索屏蔽VEC大小：100000Transform仅SEQ：2.3379 MS，RET：100000Transform仅适用：0.5979 MS，RET：100000FilterCopyIF：3.675 MS，RET：50203毫米，RET：50203毫米，RET：50203FILTERCOPYIFPARCOMPOSEQ ：1.0754 MS，RET：50203FilterCopyIFPARTRANSFORMPUSH：2.0103 MS，RET：50203

FilterCopyIfParComposeSeq-是FilterCopyIfParCompose的一个版本，带有一个简单的循环来复制结果：

FilterCopyIfParTransformPush是另一个变体，其中只有std :: transform可以并行运行，然后我们使用常规的push_back。

模板＆lt; typename T，typename Pred＆gt;自动FilterCopyIfParTransformPush（const std :: vector＆lt; T＆gt;＆vec，Pred p）{std :: vector＆lt; uint8_t＆gt;缓冲区（vec.size（））; std :: transform（std :: execution :: par，begin（vec），end（vec），begin（buffer），[＆amp; p]（const T＆amp; elem）{return p（elem）;}）; std :: vector＆lt; T＆gt;出去; for（size_t i = 0; i＆lt; vec.size（）; ++ i）if（buffer [i]）out.push_back（vec [i]）;返回；}

但是我们可以看到该版本比顺序版本快2倍！ （4核），而6核则快3倍！因此，这是一种很有前途的方法。

这次，我们将工作分成较小的块，然后分别调用copy_if：

模板＆lt; typename T，typename Pred＆gt;自动FilterCopyIfParChunks（const std :: vector＆lt; T＆gt;＆amp; vec，Pred p）{const auto chunks = std :: thread :: hardware_concurrency（）; const auto chunkLen = vec.size（）/块; std :: vector＆lt; size_t＆gt;指数（块）; std :: iota（indexes.begin（），index.end（），0）; std :: vector＆lt; std :: vector＆lt; T＆gt;复制的块（块）; std :: for_each（std :: execution :: par，begin（indexes），end（indexes），[＆amp;]（size_t i）{auto startIt = std :: next（std :: begin（vec），i * chunkLen）; auto endIt = std :: next（startIt，chunkLen）; std :: copy_if（startIt，endIt，std :: back_inserter（copiedChunks [i]），p）;}）; std :: vector＆lt; T＆gt;出去;对于（const auto＆amp; part：copyedChunks）out.insert（out.end（），part.begin（），part.end（））; if（vec.size（）％chunks！= 0）{auto startIt = std :: next（std :: begin（vec），chunks * chunkLen）; std :: copy_if（startIt，end（vec），std :: back_inserter（out），p）; }返回；}

// 4核/ 8线程基准vec大小：100000仅转换seq：2.5878 ms，ret：100000仅转换par：1.3734 ms，ret：100000FilterCopyIf：5.3675 ​​ms，ret：50203FilterCopyIfParNaive：9.1836 ms，ret：50203FilterCopyIfParCompose：3.03 ms，ret：50203eqFilterCopyIfComp ：2.3454 ms，ret：50203FilterCopyIfParTransformPush：2.5735 ms，ret：50203FilterCopyIfParChunks：2.9718 ms，ret：50203

// 6核/ 12线程基准vec大小：100000仅转换seq：2.3379 ms，ret：100000仅转换par：0.5979 ms，ret：100000FilterCopyIf：3.675 ms，ret：50203FilterCopyIfParNaive：10.0073 ms，ret：50203FilterCopyIfParCompose：1.2936 ms，ret：50203FilterCopyIfParComp ：1.0754 ms，ret：50203FilterCopyIfParTransformPush：2.0103 ms，ret：50203FilterCopyIfParChunks：2.0974 ms，ret：50203

此版本的实现较为简单，但速度不高。不过，它比顺序版本要快。

先前的版本很有前途，但是我们还有另一种方式来划分任务。与其依赖std :: execution :: par，我们可以踢几个std :: future对象，然后等待它们完成。

我在名为“ C ++高性能”的书中发现了类似的想法免责声明：我没有这本书，但其Github Repo似乎已经公开可用：Cpp-High-Performance / copy_if_split_into_two_parts.cpp

模板＆lt; typename T，typename Pred＆gt;自动FilterCopyIfParChunksFuture（const std :: vector＆lt; T＆gt;＆vec，Pred p）{const auto chunks = std :: thread :: hardware_concurrency（）; const auto chunkLen = vec.size（）/块; std :: vector＆lt; std :: future＆lt; std :: vector＆lt; T＆gt;＆gt;任务（块）； for（size_t i = 0; i＆lt; chunks; ++ i）{auto startIt = std :: next（std :: begin（vec），i * chunkLen）;自动endIt = std :: next（startIt，chunkLen）;任务[i] = std :: async（std :: launch :: async，[=，＆amp; p] {std :: vector＆lt; T＆gt; chunkOut; std :: copy_if（startIt，endIt，std :: back_inserter（chunkOut ），p）; return chunkOut;}）; } std :: vector＆lt; T＆gt;出去;对于（auto＆amp; ft：任务）{auto part = ft.get（）; out.insert（out.end（），part.begin（），part.end（））; } //其余部分：if（vec.size（）％chunks！= 0）{auto startIt = std :: next（std :: begin（vec），chunks * chunkLen）; std :: copy_if（startIt，end（vec），std :: back_inserter（out），p）; }返回；}

// 4核/ 8线程基准vec大小：100000仅转换seq：2.5878 ms，ret：100000仅转换par：1.3734 ms，ret：100000FilterCopyIf：5.3675 ​​ms，ret：50203FilterCopyIfParNaive：9.1836 ms，ret：50203FilterCopyIfParCompose：3.03 ms，ret：50203eqFilterCopyIfComp ：2.3454 ms，ret：50203FilterCopyIfParTransformPush：2.5735 ms，ret：50203FilterCopyIfParChunks：2.9718 ms，ret：50203FilterCopyIfParChunksFuture：2.5091 ms，ret：50203

// 6核/ 12线程基准vec大小：100000仅转换seq：2.3379 ms，ret：100000仅转换par：0.5979 ms，ret：100000FilterCopyIf：3.675 ms，ret：50203FilterCopyIfParNaive：10.0073 ms，ret：50203FilterCopyIfParCompose：1.2936 ms，ret：50203FilterCopyIfParComp ：1.0754 ms，ret：50203FilterCopyIfParTransformPush：2.0103 ms，ret：50203FilterCopyIfParChunks：2.0974 ms，ret：50203FilterCopyIfParChunksFuture：1.9456 ms，ret：50203

// 4 CoreS / 8 ThreadsBenchmark Vec大小：1000000Transform仅SEQ：24.7069 MS，RET：1000000Transform仅适用：5.9799 MS，RET：1000000FilterCopyIF：45.2647 MS，RET：499950FilterCopyIFParnaive：499950FilterCopyIFParCompose：17.1237 MS，RET：499950FilterCopyIFPARCOMPOSEQ ：16.7736 MS，RET：499950FILTERCOPYIFPARTRANSFORMPUSH：21.2285 MS，RET：499950FILTERCOPYIFPARCHUSKS：22.1941 MS，RET：499950FILTERCOPYIFPARCHUNKSFUTURE：22.4486 MS，RET：499950

// 6核心/ 12线索屏蔽VEC尺寸：1000000Transform仅限SEQ：24.7731 MS，RET：1000000Transform仅适用：2.8692 MS，RET：1000000FilterCopyIF：35.6397 MS，RET：499950 //基础线FilterCopyIFParnaive：499950FilterCopyIFParCompose：99953 MS ，RET：499950FilterCopyIFPARCOMPOSEQ：9.9909 MS，RET：499950FILTERCOPYIFPARTRANSFORMPUSH：13.9003 MS，RET：499950FILTERCOPYIFPARCHUNKS：499950FILTERCOPYIFPARCHUNKSFUTURE：12.6284 MS，RET：499950

正如您所看到的，我们可以更快地使代码更快，但仍然，您需要大量的元素来处理（我猜50K ...... 100k至少），以及您必须支付额外设置甚至内存的价格。

像往常一样，这取决于您的环境和要求。但是，如果您使用多个线程，它可能是明智的，依赖顺序拷贝_IF并保持其他线程忙。例如，您可以启动一些小的“副本”线程，在等待副本完成时同时执行其他工作。这里有各种各样的场景和方法。

本文旨在不创建最好的并行算法，而是实验并学习一些东西。我希望这里提出的想法为您提供一些提示。

我完全依赖于基于标准库的实心多线程框架。尽管如此，使用英特尔TBB或其他高度复杂的计算框架时，还有更多选项。

如您所见，我敦促动态创建输出矢量。这会产生一些额外的开销，所以在解决方案中，您可能会限制为此。为什么不分配与输入容器相同的大小？也许可以在解决方案中工作？也许我们可以在以后推载载体？或者可能利用一些智能分配者？

我还应该提到，在最近的超负荷杂志ACCU中有一篇文章：危害C ++并行算法的盲目的案例，所以你也可以看看它。 文章列出了您可能希望在跳跃和放置std :: execution :: dellower之前考虑的五个问题。 我展示了4种不同的技术，但也许你有更多的东西？ 你会建议什么？ 我在MSVC上测试了我的代码，但在GCC中，通过Intel TBB获得并行算法。 你试过他们吗？ 很高兴看到该编译器/库的结果。 如果您＆＃39; ve＆＃39; vere＆＃39; vere＆＃39;对现代c ++感兴趣！ 了解最近C ++标准的所有主要功能！ 检查一下：