Проблемы с многопоточной .NET-очередью

У меня есть своя ошибка в моем коде. Это очень редко (бывает, раз в несколько недель может быть), но там, и я не уверен, почему.

У нас есть 2 потока, 1 поток получает сетевые сообщения и добавляет их в очередь следующим образом:

DataMessages.Enqueue(new DataMessage(client, msg));

Другой поток принимает сообщения из этой очереди и обрабатывает их, например:

while (NetworkingClient.DataMessages.Count > 0)
{
    DataMessage message = NetworkingClient.DataMessages.Dequeue();

    switch (message.messageType)
    {
       ...
    }
}

Однако раз так часто я получаю исключение NullReferenceException в строке switch (message.messageType), и я могу видеть в отладчике, что это сообщение равно null.

Невозможно, чтобы нулевое значение было помещено в очередь (см. первый бит кода), и это единственные 2 вещи, которые используют очередь.

Является ли Queue не потокобезопасным, может быть, что я нахожусь в тот момент, когда другой поток задерживается, и это вызывает сбой?

Ответ 1

    while (NetworkingClient.DataMessages.Count > 0)
    {
        // once every two weeks a context switch happens to be here.
        DataMessage message = NetworkingClient.DataMessages.Dequeue();

        switch (message.messageType)
        {
           ...
        }
    }

... и когда вы получите этот контекстный переключатель в этом месте, результат первого выражения (NetworkingClient.DataMessages.Count > 0) истинно для обоих потоков, а тот, который получает операцию Dequeue(), сначала получает объект, а второй поток получает нуль (вместо InvalidOperationException, потому что внутреннее состояние очереди не было полностью обновлено, чтобы правое исключение).

Теперь у вас есть два варианта:

Используйте .NET 4.0 ConcurrentQueue
Восстановите свой код:

и сделайте так, чтобы это выглядело так:

while(true)
{
  DataMessage message = null;

  lock(NetworkingClient.DataMessages.SyncRoot) {
       if(NetworkingClient.DataMessages.Count > 0) {
          message = NetworkingClient.DataMessages.Dequeue();
       } else {
         break;
       }
    }
    // .. rest of your code
}

Изменить: обновлено, чтобы отразить комментарий Ханделя.

Ответ 2

Очередь не является потокобезопасной, может ли она быть что я нахожусь в нужный момент что другой поток задерживается и это вызывает сбой?

Совершенно верно. Queue не является потокобезопасным. Потоковая безопасность - System.Collections.Concurrent.ConcurrentQueue. Используйте его, чтобы исправить вашу проблему.

Ответ 3

Если вас интересует точная причина:

Enqueue выглядит следующим образом:

this._array[this._tail] = item;
this._tail = (this._tail + 1) % this._array.Length;
this._size++;
this._version++;

И Dequeue вот так:

T result = this._array[this._head];
this._array[this._head] = default(T);
this._head = (this._head + 1) % this._array.Length;
this._size--;
this._version++;

Гонка проходит следующим образом:

В очереди есть 1 элемент (head == tail), так что поток вашего читателя начинает декомпрессироваться, но прерван после первой строки в Dequeue
Затем другой элемент помещается в очередь и помещается в позицию tail, которая в данный момент равна head.
Теперь Dequeue возобновляет и перезаписывает элемент, который был только что вставлен Enqueue с помощью default(T)
В следующий раз, когда вы вызываете dequeue, вы получаете значение по умолчанию (T) (в вашем случае null) вместо фактического значения