在 tbb parallel_pipeline 中调整粒度

Question

管道任务如下：

1. 顺序读取大量（10-15k）~100-200 Mb 压缩文件
2. 并行解压每个文件
3. 并行反序列化每个解压文件
4. 处理结果反序列化对象并根据所有对象（平均值、中位数、分组等）获取一些值

当我得到解压缩的文件内存缓冲区时，序列化块一个接一个，所以我想以相同的方式将它们传递给下一个过滤器，或者至少通过将序列化块打包成一些组来调整这个过程编号，然后通过。但是（据我了解） tbb_pipeline 使我将指针传递给带有所有序列化块的缓冲区，因为每个过滤器都必须获取指针并返回指针。

据我了解，使用并发队列来累积序列化对象包会扼杀使用 tbb_pipeline 的问题。此外，过滤器中 operator() 的常量不允许有我自己的中间“任务池”（但是如果每个线程都有自己的“任务”存储本地副本并且只是正确切割从它的碎片，它会很棒）

主要问题： 在这种情况下有没有办法“调整”粒度？ （即某些过滤器获取指向所有序列化对象的指针并传递给下一个过滤器小包对象）

重新格式化（分割等）输入文件几乎是不可能的。

次要问题： 当我积累处理结果时，我并不关心任何一种顺序，我只需要汇总统计信息。我可以使用 parallel 过滤器而不是 serial_out_of_order 并在某处为每个线程累积处理结果，然后合并它们吗？

Answer 1

但是（据我了解）tbb_pipeline 让我将指针传递给带有所有序列化块的缓冲区，因为每个过滤器都必须获取指针并返回指针。

首先我认为，最好使用更现代、类型安全的管道形式：parallel_pipeline。它没有规定您传递任何特定数据的任何特定指针。您只需指定下一阶段需要哪种类型的数据才能处理它。因此，问题在于您的第一个过滤器如何对要由以下过滤器处理的数据进行分区。

主要问题：在这种情况下有没有办法“调整”粒度？ （即一些过滤器获取指向所有序列化对象的指针并传递给下一个过滤器小包对象）

您可以安全地将一种并行算法嵌入到另一种算法中，以更改某些阶段的粒度，例如在顶层，第一个管道通过文件列表；第二个管道在嵌套级别读取文件的大块；最后，对于某些 2 级阶段，最里面的管道将大块分解为较小的块。请参阅下面的一般嵌套示例。

第二个问题：我可以使用并行过滤器代替serial_out_of_order，并在某处为每个线程累积处理结果，然后合并它们吗？

是的，如果并行过滤器不修改共享数据，您始终可以使用它。例如，您可以使用tbb::combinable 来收集特定于线程的部分和，然后将它们组合起来。

但是，如果每个线程都有自己的本地“任务”存储副本并从中切出正确的部分，那就太好了

是的，他们有。每个线程都有自己的本地任务池。

---

嵌套parallel_pipelines的一般示例

parallel_pipeline( 2/*only two files at once*/,
    make_filter<void,std::string>(
        filter::serial,
        [&](flow_control& fc)-> std::string {
            if( !files.empty() ) {
                std::string filename = files.front();
                files.pop();
                return filename;
             } else {
                fc.stop();
                return "stop";
            }
        }    
    ) &
    make_filter<std::string,void>(
        filter::parallel,
        [](std::string s) {

            // a nested pipeline
            parallel_pipeline( 1024/*only two files at once*/,
                make_filter<void,char>(
                    filter::serial,
                    [&s](flow_control& fc)-> char {
                        if( !s.empty() ) {
                            char c = s.back();
                            s.pop_back();
                            return c;
                         } else {
                            fc.stop();
                            return 0;
                        }
                    }    
                ) &
                make_filter<char,void>(
                    filter::parallel,
                    [](char c) {
                        putc(c, stdout);
                    } 
                )
            );
        } 
    )
);