c# - 循环枚举器的 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 中的分区程序相关方法中都发生了死锁，但我不确定如何解释这一点。

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

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

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

Answer 1

您实际上没有值的逻辑序列，因此首先尝试创建 IEnumerable 根本没有任何意义。此外，您几乎肯定不应该尝试创建可由多个线程使用的 IEnumerator。这很疯狂，只是因为 IEnumerator 公开的接口(interface)并没有真正公开您想要它公开的内容。您可能会创建一个 IEnumerator，它只会被单个线程使用，该线程根据多个线程使用的基础数据源计算要返回的数据，因为这是相当不同的。

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