尝试在长时间运行的生成器上使用 PLINQ 的陷阱？答案

【问题标题】：Pitfalls of trying to use PLINQ over long-running generators?尝试在长时间运行的生成器上使用 PLINQ 的陷阱？
【发布时间】：2012-01-25 06:40:04
【问题描述】：

我有一些无限生成器方法，包括一些长时间运行和无限长时间运行的生成器。

IEnumerable<T> ExampleOne() { 
    while(true) // this one blocks for a few seconds at a time
        yield return LongRunningFunction();
}
IEnumerable<T> ExampleTwo() { 
    while(true) //this one blocks for a really long time
        yield return OtherLongRunningFunction();
}

我的目标是拥有一个无限序列，将两个示例中的项目组合在一起。这是我尝试过的，使用 PLINQ：

 IEnumerable<T> combined =  new[] { ExampleOne(), ExampleTwo() }
           .AsParallel()
           .WithMergeOptions(ParallelMergeOptions.NotBuffered)
           .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
           .SelectMany(source => source.GetRequests());

这似乎恰当地将两个 IEnumerable 组合成一个新的，IEnumerable #1 和 #2 中的项目只要出现在两个源 IEnumerables 中的任何一个中都可用：

//assuming ExampleTwo yields TWO but happens roughly 5 times 
//less often then ExampleOne
Example output:  one one one one one TWO one one one one one one TWO

但是，似乎有时（通常在运行许多小时之后）OtherLongRunningFunction() 会运行很长时间而不会返回，并且在难以重现的条件下，@987654327 @sequence 将阻止它，而不是继续从第一个 LongRunningFunction 返回结果。看起来虽然组合并行查询开始使用两个线程，但它决定稍后切换到一个线程。

我的第一个想法是“这可能是 RX Observable.Merge 的工作，而不是 PLINQ。”但我会很感激这两个答案，它们显示了处理这种情况的正确替代方法，以及关于 PLINQ 如何在查询开始数小时后改变并行度的机制的解释。

【问题讨论】：

我不能说我对 PLINQ 有很多经验，所以这主要是不合格的猜测：看起来好像你假设结合 ExampleOne() 和 ExampleTwo() 并运行 AsParallel() on生成的IEnumerable 将通过从第一个返回一个结果和从第二个返回一个结果来严格交替。难道这个假设是错误的？如果是这样，您最终可能会遇到正在处理的序列看起来像 1st 2nd 1st 2nd 1st 1st 1st 2nd 1st 2nd 1st 1st 2nd 1st 2nd 2nd 2nd 2nd... 这可以解释为什么它看起来好像你被困在第二个问题上。
@Nailuj 这实际上更像我所期望的 1 1 1 1 1 1 1 2 1 1 1 1 1 1 1 2 等，2s 是相当罕见的，并且可以有连续 2 秒之间的长时间。我希望组合序列继续返回 1，事实上，这实际上是大多数情况下发生的情况。但有时它也会停止返回 1。
另一个想法：可能是 PLINQ 以并行处理相同数量的 1 和 2 开始，但由于 1 完成得更快，它将显示为您描述的序列。但是，每当 1 完成时，该“点”将分别由 1 和 2 交替填充。结果是，一开始，看起来好像你得到了很多 1，只有很少的 2，但从长远来看，正在处理的任务的“队列”会被 2 填满，因此似乎卡住了?如果不完全像这样，它可能有点相关吗？只是想大声思考:-)
您是否真的验证过这种行为（在两个长时间运行的函数执行期间添加一些诊断日志记录，包括当前线程 ID）

标签： c# system.reactive plinq

【解决方案1】：

这是执行此操作的 Rx 方法，实际上，它确实使用了 Merge：

IObservable<T> LongRunningFunction()
{
    return Observable.Start(() => {
        // Calculate some stuff
        return blah;
    }, Scheduler.TaskPoolScheduler);
}

Observable.Merge(
    Observable.Defer(LongRunningFunction).Repeat(),
    Observable.Defer(OtherLongRunningFunction).Repeat(),
).Subscribe(x => {
    Console.WriteLine("An item: {0}", x);
});

【讨论】：

【解决方案2】：

如果您想要 TPL 的好处，特别是对于负载变化的任务（当您的订阅阻塞，并且已经生成了许多项目时会发生什么 - 您应该停止生成项目吗？），我推荐 TPL DataFlow。

如果你想用 Rx 来做，对于真正长时间运行的计算任务，最好不要阻塞线程池：

var stream = Observable.Merge(ExampleTwo().ToObservable(Scheduler.NewThread), ExampleOne().ToObservable(Scheduler.NewThread));

stream.Subscribe(...);

【讨论】：

如果你在使用 Rx 时阻塞，你做错了(tm)。垄断任务池并不像人们想象的那么糟糕，因为它是一种任务窃取实现。话虽如此，如果您知道要将一堆线程停放在 CPU 密集型工作流中，那么您可以做 NewThread 的事情。
@PaulBetts 说得好。你写的是TaskPool版本，所以我放了thread版本。了解选项总是好的。

【解决方案3】：

关于 PLINQ 的机制：

我遇到了同样的问题：我有一个序列，其中的项目需要不均匀的处理时间，其中一些时间要长几个数量级。我经历了线程饥饿，在 8 核处理器上比在 4 核上更容易重现，尽管在处理数小时后它也可能在 4 核上发生。一些线程可能会在一段时间后再次开始工作。请注意，使用了动态分块，如示例中所示。

观察：饥饿更可能发生在完成连续的非常长时间运行的工作项时。

MSDN 主题Parallel Loops 提供了一些启示：

如果您使用并行循环的各个步骤需要几秒钟或更长时间，请务必小心。这可能发生在 I/O 密集型工作负载以及冗长的计算中。如果循环需要很长时间，您可能会遇到工作线程的无限增长，这是由于 .NET ThreadPool 类的线程注入逻辑使用的用于防止线程饥饿的启发式方法。当当前池的工作项长时间运行时，这种启发式方法会稳步增加工作线程的数量。动机是在线程池中的所有内容都被阻塞的情况下添加更多线程。不幸的是，如果工作确实在进行，更多的线程可能不一定是你想要的。 .NET Framework 无法区分这两种情况。

我仍然不知道细节，但我认为底层 ThreadPool 的启发式不能很好地解释运行时间很长的工作项，由于某些上限没有适当调整而无法为下一次迭代提供线程，因此有迭代排队。我没有 Visual Studio 访问 8 核机器的权限，在该机器上问题更容易重现。我还无法在 4 核机器上的 Visual Studio 调试下重现该问题。调查仍在继续。

更多详情，"Does the Task Parallel Library (or PLINQ) take other processes into account?" 主题高度相关。

【讨论】：