Мой GetEnumerator вызывает тупик?

Question

Я начинаю писать свои первые параллельные приложения. Этот разделитель будет перечислять более IDataReader тянущий chunkSize записи за один раз из источника данных.

TLDR; версия

private object _Lock = new object();
public IEnumerator GetEnumerator()
{
    var infoSource = myInforSource.GetEnumerator();
                   //Will this cause a deadlock if two threads 
    lock (_Lock)   //use the enumator at the same time?
    {
        while (infoSource.MoveNext())
        {
            yield return infoSource.Current;
        }
    }
}

полный код

protected class DataSourcePartitioner<object[]> : System.Collections.Concurrent.Partitioner<object[]>
{
    private readonly System.Data.IDataReader _Input;
    private readonly int _ChunkSize;
    public DataSourcePartitioner(System.Data.IDataReader input, int chunkSize = 10000)
        : base()
    {
        if (chunkSize < 1)
            throw new ArgumentOutOfRangeException("chunkSize");
        _Input = input;
        _ChunkSize = chunkSize;
    }

    public override bool SupportsDynamicPartitions { get { return true; } }

    public override IList<IEnumerator<object[]>> GetPartitions(int partitionCount)
    {

        var dynamicPartitions = GetDynamicPartitions();
        var partitions =
            new IEnumerator<object[]>[partitionCount];

        for (int i = 0; i < partitionCount; i++)
        {
            partitions[i] = dynamicPartitions.GetEnumerator();
        }
        return partitions;


    }

    public override IEnumerable<object[]> GetDynamicPartitions()
    {
        return new ListDynamicPartitions(_Input, _ChunkSize);
    }
    private class ListDynamicPartitions : IEnumerable<object[]>
    {
        private System.Data.IDataReader _Input;
        int _ChunkSize;
        private object _ChunkLock = new object();
        public ListDynamicPartitions(System.Data.IDataReader input, int chunkSize)
        {
            _Input = input;
            _ChunkSize = chunkSize;
        }

        public IEnumerator<object[]> GetEnumerator()
        {

            while (true)
            {
                List<object[]> chunk = new List<object[]>(_ChunkSize);
                lock(_Input)
                {
                    for (int i = 0; i < _ChunkSize; ++i)
                    {
                        if (!_Input.Read())
                            break;
                        var values = new object[_Input.FieldCount];
                        _Input.GetValues(values);
                        chunk.Add(values);
                    }
                    if (chunk.Count == 0)
                        yield break;
                }
                var chunkEnumerator = chunk.GetEnumerator();
                lock(_ChunkLock) //Will this cause a deadlock?
                {
                    while (chunkEnumerator.MoveNext())
                    {
                        yield return chunkEnumerator.Current;
                    }
                }
            }
        }

        IEnumerator IEnumerable.GetEnumerator()
        {
            return ((IEnumerable<object[]>)this).GetEnumerator();
        }
    }
}

я хотел IEnumerable объект, который он передал обратно, чтобы быть потокобезопасным ( пример MSDN был, так что я предполагаю, что PLINQ и TPL может понадобиться) заблокирует _ChunkLock в нижней части помогают обеспечить безопасность потока или это может привести к тупику? Из документации я не мог сказать, будет ли снята блокировка на yeld return,

Также, если в.net есть встроенная функциональность, которая будет делать то, что я пытаюсь сделать, я бы скорее использовал это. И если вы обнаружите какие-либо другие проблемы с кодом, я был бы признателен.

Answer 1

Одним словом: возможно *.

Если вы всегда используете этот код в контексте foreach цикл, то вы вряд ли попадете в тупик (если возможно, что ваш myInfoSource бесконечен, или что ваш foreach цикл содержит некоторый код, который никогда не прекратится), хотя вы можете увидеть замедления.

Более вероятная причина потенциальной (фактически гарантированной) тупиковой ситуации будет заключаться в следующем:

var myObject = new YourObject();
var enumerator = myObject.GetEnumerator();

// if you do this, and then forget about it...
enumerator.MoveNext();

// ...your lock will never be released

* Я основываю этот ответ на вашем начальном блоке кода.

Answer 2

Я написал тестовый фреймворк, он не блокируется, но второй поток никогда не получит данные.

static void Main()
{
    En en = new En();
    Task.Factory.StartNew(() =>
        {
            foreach (int i in en)
            {
                Thread.Sleep(100);
                Console.WriteLine("A:" + i.ToString());
            }
        });
    Task.Factory.StartNew(() =>
    {
        foreach (int i in en)
        {
            Thread.Sleep(10);
            Console.WriteLine("B:" +i.ToString());
        }
    });
    Console.ReadLine();
}

public class En : IEnumerable
{
    object _lock = new object();
    static int i = 0;
    public IEnumerator GetEnumerator()
    {
        lock (_lock)
        {
            while (true)
            {
                if (i < 10)
                    yield return i++;
                else
                    yield break;
            }
        }
    }
}

Возвращает

A:0
A:1
A:2
A:3
A:4
A:5
A:6
A:7
A:8
A:9

Вот обновленная версия GetEnumerator это должно вести себя правильно.

public IEnumerator<object[]> GetEnumerator()
{

    while (true)
    {
        List<object[]> chunk = new List<object[]>(_ChunkSize);
        _ChunkPos = 0;
        lock(_Input)
        {
            for (int i = 0; i < _ChunkSize; ++i)
            {
                if (!_Input.Read())
                    break;
                var values = new object[_Input.FieldCount];
                _Input.GetValues(values);
                chunk.Add(values);
            }
            if (chunk.Count == 0)
                yield break;
        }
        var chunkEnumerator = chunk.GetEnumerator();
        while (true)
        {
            object[] retVal;
            lock (_ChunkLock)
            {
                if (chunkEnumerator.MoveNext())
                {
                    retVal = chunkEnumerator.Current;
                }
                else 
                    break; //break out of chunk while loop.
            }
            yield return retVal;
        }
    }
}