从文件并发处理答案

【问题标题】：Concurrent Processing From File从文件并发处理
【发布时间】：2012-08-09 21:30:05
【问题描述】：

考虑以下代码：

std::vector<int> indices = /* Non overlapping ranges. */;
std::istream& in = /*...*/;

for(std::size_t i= 0; i< indices.size()-1; ++i)
{
    in.seekg(indices[i]);

    std::vector<int> data(indices[i+1] - indices[i]);

    in.read(reinterpret_cast<char*>(data.data()), data.size()*sizeof(int));

    process_data(data);
}

我想让这段代码尽可能快地并行。

使用PPL 将其并行化的一种方法是：

std::vector<int> indices = /* Non overlapping ranges. */;
std::istream& in = /*...*/;
std::vector<concurrency::task<void>> tasks;    

for(std::size_t i= 0; i< indices.size()-1; ++i)
{
    in.seekg(indices[i]);

    std::vector<int> data(indices[i+1] - indices[i]);

    in.read(reinterpret_cast<char*>(data.data()), data.size()*sizeof(int));

    tasks.emplace_back(std::bind(&process_data, std::move(data)));
}
concurrency::when_all(tasks.begin(), tasks.end()).wait();

这种方法的问题是我想在同一个线程中处理数据（适合 CPU 缓存），因为它被读入内存（数据在缓存中是热的），这里不是这种情况，简直是在浪费使用热点数据的机会。

我有两个想法如何改进它，但是我也没有能够实现。

在单独的任务上开始下一次迭代。

/* ??? */
{
     in.seekg(indices[i]);

     std::vector<int> data(indices[i+1] - indices[i]); // data size will fit into CPU cache.

     in.read(reinterpret_cast<char*>(data.data()), data.size()*sizeof(int));

     /* Start a task that begins the next iteration? */

     process_data(data);
}

使用内存映射文件并映射文件的所需区域，而不是仅仅从具有正确偏移量的指针中读取。使用parallel_for_each 处理数据范围。但是，我不了解内存映射文件在何时读取到内存和缓存注意事项方面的性能影响。也许我什至不必考虑缓存，因为文件只是 DMA:d 到系统内存，从不通过 CPU？

有什么建议吗？

【问题讨论】：

为什么不尝试将文件拆分成更小的文件，然后在单独的线程中读取和处理它们呢？对于文件，您必须考虑 CPU 绑定和 I/O 绑定
+1。似乎将内存保留在缓存中所获得的任何好处都将与等待文件 I/O 相比相形见绌。
@BenFulton：你是对的，假设文件不在操作系统缓存中和/或我没有执行此函数的多个并发执行。

标签： c++ windows file memory-mapped-files ppl

【解决方案1】：

您很可能追求错误的目标。如前所述，“热数据”的任何优势都将与磁盘速度相形见绌。否则，有些重要的细节你没有告诉。
1) 文件是否“大”
2) 单条记录是否“大”
3) 处理是否“慢”

如果文件“大”，您的首要任务是确保按顺序读取文件。你的“指数”让我不这么认为。根据我自己的经验，最近的示例是 6 秒 vs 20 分钟，具体取决于随机读取还是顺序读取。不开玩笑。

如果文件“小”并且您确定它已被完全缓存，您只需要一个同步队列将任务交付给您的线程，那么在同一个线程中处理不会有问题。

另一种方法是将“索引”分成两半，每个线程一个。

【讨论】：