C#. Исключение при многопоточной работе с Queue

202

17 октября 2017, 00:26

Здравствуйте. Сушествует класс очереди последовательного порта, который работает в своем потоке. Данные для очереди могут поступать из других потоков. При разматывании очереди классом последовательным порта возникает исключение:

2017/10/15 08:52:36.907|Error|Ошибка работы с портом: COM4. ОШИБКА: System.InvalidOperationException: Collection was modified after the enumerator was instantiated.
   at System.ThrowHelper.ThrowInvalidOperationException(ExceptionResource resource)
   at System.Collections.Generic.Queue`1.Enumerator.MoveNext()
   at System.Linq.Enumerable.Any[TSource](IEnumerable`1 source)
   at CommunicationDevices.Behavior.ExhangeBehavior.SerialPortBehavior.ChannelManagement.ChannelManagementExchangeBehavior.<OneTimeExchangeService>d__9.MoveNext() in C:\Git\Autodictor\src\CommunicationDevices\Behavior\ExhangeBehavior\SerialPortBehavior\ChannelManagement\ChannelManagementExchangeBehavior.cs:line 46
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at Communication.SerialPort.MasterSerialPort.<RunExchange>d__38.MoveNext() in C:\Git\Autodictor\src\Communication\SerialPort\MasterSerialPort.cs:line 214

Проанализировав стэк вызовов, видно что виновником является InDataQueue.Any(), т.к. Приводится InDataQueue к IEnumerable, а из другого потока я могу добавить данные в InDataQueue, что приведет к исключению. OneTimeExchangeService - ВЫЗЫВАЕТСЯ ИЗ ПОТОКА ПОРТА.

protected override async Task OneTimeExchangeService(MasterSerialPort port, CancellationToken ct)
{
    var inData = (InDataQueue != null && InDataQueue.Any()) ? InDataQueue.Dequeue() : null;  //хранит адресс устройства.
    if (inData != null)
    {
        WriteProvider.InputData = inData;
        DataExchangeSuccess = await Port.DataExchangeAsync(TimeRespone, WriteProvider, ct);
    }
}

Наверно можно InDataQueue.Any() заменить на InDataQueue.Count > 0. А как правильно работать с такими коллекциями, возможно есть специальные многопоточные коллекции, как с ними работать и в чем отличие?

Answer 1

Самое простое решение — используйте внешнюю блокировку. Заведите объект синхронизации и на время работы с очередью лочьте его.

Получится что-то такое:

object queueGuard = new object();
protected override async Task OneTimeExchangeService(
        MasterSerialPort port, CancellationToken ct)
{
    DataT inData; // или какой у вас там тип данных у элементов InDataQueue
    lock (queueGuard)
    {
        inData = (InDataQueue != null && InDataQueue.Any()) ?
                InDataQueue.Dequeue() : null;  //хранит адрес устройства.
    }
    if (inData != null)
    {
        WriteProvider.InputData = inData;
        DataExchangeSuccess = await Port.DataExchangeAsync(TimeRespone, WriteProvider, ct);
    }
}

Не забудьте использовать один и тот же объект queueGuard при всех операциях доступа к InDataQueue.