循环枚举器的 PLINQ 迭代导致死锁

Question

我有一个简单的程序，它迭代一个作为反馈枚举器实现的无限枚举。我已经在 TPL 和 PLINQ 中实现了这一点。这两个示例都在可预测的迭代次数后锁定：PLINQ 8 次，TPL 3 次。如果代码在不使用 TPL/PLINQ 的情况下执行，它运行良好。我以非线程安全的方式和线程安全的方式实现了枚举器。如果并行度限制为一个（如示例中的情况），则可以使用前者。非线程安全枚举器非常简单，不依赖于任何“花哨”的 .NET 库类。如果我增加并行度，死锁之前执行的迭代次数就会增加，例如对于 PLINQ，迭代次数是 8 * 并行度。

以下是迭代器：
枚举器（非线程安全）

public class SimpleEnumerable<T>: IEnumerable<T>
{
    private T _value;
    private readonly AutoResetEvent _releaseValueEvent = new AutoResetEvent(false);

    IEnumerator IEnumerable.GetEnumerator()
    {
        return GetEnumerator();
    }

    public IEnumerator<T> GetEnumerator()
    {
        while(true)
        {
            _releaseValueEvent.WaitOne();
            yield return _value;
        }
    }

    public void OnNext(T value)
    {
        _value = value;
        _releaseValueEvent.Set();
    }
}

枚举器（线程安全）

public class SimpleEnumerable<T>: IEnumerable<T>
{
    private readonly BlockingCollection<T> _blockingCollection = new BlockingCollection<T>();

    IEnumerator IEnumerable.GetEnumerator()
    {
        return GetEnumerator();
    }

    public IEnumerator<T> GetEnumerator()
    {
        while(true)
        {
            yield return _blockingCollection.Take();
        }
    }

    public void OnNext(T value)
    {
        _blockingCollection.Add(value);
    }
}

PLINQ 示例：

public static void Main(string[] args)
{
    var enumerable = new SimpleEnumerable<int>();
    enumerable.OnNext(0);

    enumerable
        .Do(i => Debug.WriteLine($"{i} {Thread.CurrentThread.ManagedThreadId}"))
        .AsParallel()
        .WithDegreeOfParallelism(1)
        .ForEach
        (
            i =>
            {
                Debug.WriteLine($"{i} {Thread.CurrentThread.ManagedThreadId}");
                enumerable.OnNext(i+1);
            }
        );
}

TPL 示例：

public static void Main(string[] args)
{
    var enumerable = new SimpleEnumerable<int>();
    enumerable.OnNext(0);

    Parallel.ForEach
    (
        enumerable,
        new ParallelOptions { MaxDegreeOfParallelism = 1},
        i =>
        {
            Debug.WriteLine($"{i} {Thread.CurrentThread.ManagedThreadId}");
            enumerable.OnNext(i+1);
        }
    );
}

根据我对调用堆栈的分析，似乎在 PLINQ 和 TPL 中的分区程序相关方法中都发生了死锁，但我不确定如何解释。

通过反复试验，我发现将 PLINQ enumerable 包装在 Partitioner.Create(enumerable, EnumerablePartitionerOptions.NoBuffering) 中可以解决问题，但我不确定为什么会发生死锁。

我很想找出错误的根本原因。

请注意，这是一个人为的示例。我不是在寻找对代码的批评，而是为什么会发生死锁。具体来说，在 PLINQ 示例中，如果 .AsParallel() 和 .WithDegreeOfParallelism(1) 行被注释掉，代码就可以正常工作。

Answer 1

您实际上没有值的逻辑序列，因此首先尝试创建IEnumerable 根本没有任何意义。此外，您几乎肯定不应该尝试创建一个可以被多个线程使用的IEnumerator。那里很疯狂，仅仅是因为IEnumerator 暴露的界面并没有真正暴露你想要它做的事情。您可能会创建一个 IEnumerator，它只会被单个线程使用，该线程根据多个线程使用的底层数据源计算要返回的数据，因为这完全不同。

如果您只是想创建在不同线程中运行的生产者和消费者，请不要围绕BlockingCollection 创建自己的“包装器”，*只需使用BlockingCollection。让生产者添加到它，消费者从中读取。消费者可以使用GetConsumingEnumerable，如果它只是想在获取这些项目时迭代项目（想要做的常见操作）。