CPU受限任务的并行化与IO受限任务的并行化

Question

我正试图找到一种很好的并行化代码的方法，对大数据集进行处理，然后将结果数据导入RavenDb。

数据处理是CPU绑定和数据库导入IO绑定。

我正在寻找一种在环境上并行处理的解决方案。ProcessorCount线程数。然后应将生成的数据导入到与上述进程并行的x（假设10）池化线程上的RavenDb中。

这里的主要问题是我希望在导入完成的数据时继续处理，以便在等待导入完成时继续处理下一个数据集。

另一个问题是每个批次的内存需要在成功导入后丢弃，因为私有工作内存很容易到达

下面的代码是我目前掌握的代码。请注意，它没有满足上述并行化要求。

datasupplier.GetDataItems()
    .Partition(batchSize)
    .AsParallel()
    .WithDegreeOfParallelism(Environment.ProcessorCount)
    .ForAll(batch =>
    {
        Task.Run(() =>
        {
            ...
        }
    }

GetDataItem生成可枚举的数据项，这些数据项被划分到批处理数据集中。GetDataItem将产生约2000000个项目，每个项目平均处理时间约为0.3ms。

该项目在最新的. NET 4.5 RC上运行。x64平台。

使现代化

我当前的代码（如上所示）将获取项目并将其分批分区。每个批在八个线程上并行处理（i7上的Environment.ProcessorCount）。处理速度慢，cpu受限，内存密集。当单个批处理完成时，将启动一个任务，将结果数据异步导入RavenDb。批量导入作业本身是同步的，看起来像：

using (var session = Store.OpenSession())
{
    foreach (var data in batch)
    {
        session.Store(data);
    }
    session.SaveChanges();
}

这种方法存在一些问题：

>

每次批处理完成时，都会启动一个任务来运行导入作业。我想限制并行运行的任务数量（例如最大10个）。此外，即使许多任务已经启动，它们似乎永远不会并行运行。

内存分配是一个巨大的问题。处理/导入批次后，它似乎仍保留在内存中。

我正在寻找解决上述问题的方法。理想情况下，我想要：

• 每个逻辑处理器一个线程，处理大量数据
• 大约十个并行线程等待完成的批导入RavenDb
• 将内存分配保持在最小值，这意味着在导入任务完成后取消批分配
• 不在其中一个线程上运行导入作业以进行批处理。已完成批次的导入应与正在处理的下一批并行运行

解决方案

var batchSize = 10000;
var bc = new BlockingCollection<List<Data>>();
var importTask = Task.Run(() =>
{
    bc.GetConsumingEnumerable()
        .AsParallel()
        .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
        .WithMergeOptions(ParallelMergeOptions.NotBuffered)
        .ForAll(batch =>
        {
            using (var session = Store.OpenSession())
            {
                foreach (var i in batch) session.Store(i);
                session.SaveChanges();
            }
        });
});
var processTask = Task.Run(() =>
{
    datasupplier.GetDataItems()
        .Partition(batchSize)
        .AsParallel()
        .WithDegreeOfParallelism(Environment.ProcessorCount)
        .ForAll(batch =>
        {
            bc.Add(batch.Select(i => new Data()
            {
                ...
            }).ToList());
        });
});

processTask.Wait();
bc.CompleteAdding();
importTask.Wait();

Answer 1

我最近构建了类似的东西，我在Parallel.Foreach中使用了Queue类vs List。我发现太多的线程实际上减慢了速度，这是一个最佳点。

Answer 2

对于每个批次，您都要开始一项任务。这意味着您的循环完成得非常快。它会留下（批次数）任务，而这些任务不是您想要的。您需要（CPU数量）。

解决方案：不要为每个批次启动新任务。for循环已经并行。

为了回应您的评论，这里有一个改进的版本：

//this runs in parallel
var processedBatches = datasupplier.GetDataItems()
    .Partition(batchSize)
    .AsParallel()
    .WithDegreeOfParallelism(Environment.ProcessorCount)
    .Select(x => ProcessCpuBound(x));

foreach (var batch in processedBatches) {
 PerformIOIntensiveWorkSingleThreadedly(batch); //this runs sequentially
}

Answer 3

总体而言，您的任务听起来像是生产者-消费者工作流。批处理程序是生产者，RavenDB数据“导入”是生产者输出的消费者。

考虑使用BlockingCollection

批处理程序生产者可以全速运行，并且始终与处理先前完成的批的db导入器任务并行运行。如果您担心批处理程序可能远远领先于数据库导入程序（b/c数据库导入比处理每个批要长得多），您可以设置阻塞集合的上限，以便生产者在添加超过该限制时阻塞，从而给消费者一个追上的机会。

不过，你的一些评论令人担忧。启动任务实例以异步执行db导入到批处理并没有什么特别的错误。任务！=线创建新任务实例的开销与创建新线程的开销不同。

不要过于精确地控制线程。即使您指定您想要的存储桶数量与您拥有的核心数量完全相同，您也不能独占使用这些核心。来自其他进程的数百个其他线程仍将安排在您的时间片之间。使用任务指定逻辑工作单元，并让TPL管理线程池。避免因错误的控制感而受挫；

在你的评论中，你指出你的任务似乎没有异步运行（你是如何确定的？）并且在每批完成后内存似乎不会释放。我建议放弃所有内容，直到你能先弄清楚这两个问题的原因。你是不是忘记在某个地方调用Dispose（）了？你是否保留了一个引用，该引用不必要地保持了一整棵对象树的活动？你测量的是正确的事情吗？并行任务是否由阻塞数据库或网络I/O序列化？在这两个问题解决之前，你的并行计划是什么并不重要。