Реализация очереди блокировки в C #

Question

Я использовал приведенный ниже код для реализации и проверки очереди блокировки.Я тестирую очередь, запуская 5 параллельных потоков (съемников), чтобы извлечь элементы из очереди, блокируя, если очередь пуста, и 1 параллельный поток (сумматор), чтобы добавлять элементы в очередь с перерывами.Однако, если я оставлю его работать достаточно долго, я получу исключение, потому что один из потоков удаления выходит из состояния ожидания, даже когда очередь пуста.

Кто-нибудь знает, почему я получаю исключение?Обратите внимание, мне интересно знать, почему это не работает, в отличие от рабочего решения (так как я могу только Google).

Буду очень признателен за вашу помощь.

using System;
using System.Threading;
using System.Collections.Generic;

namespace Code
{
    class Queue<T>
    {
        private List<T> q = new List<T>();

        public void Add(T item)
        {
            lock (q)
            {
                q.Add(item);
                if (q.Count == 1)
                {
                    Monitor.Pulse(q);
                }
            }
        }

        public T Remove()
        {
            lock (q)
            {
                if (q.Count == 0)
                {
                    Monitor.Wait(q);
                }
                T item = q[q.Count - 1];
                q.RemoveAt(q.Count - 1);
                return item;
            }
        }
    }

    class Program
    {
        static Random r = new Random();
        static Queue<int> q = new Queue<int>();
        static int count = 1;
        static void Adder()
        {
            while (true)
            {
                Thread.Sleep(1000 * ((r.Next() % 5) + 1));
                Console.WriteLine("Will try to add");
                q.Add(count++);
            }
        }

        static void Remover()
        {
            while (true)
            {
                Thread.Sleep(1000 * ((r.Next() % 5) + 1));
                Console.WriteLine("Will try to remove");
                int item = q.Remove();
                Console.WriteLine("Removed " + item);
            }
        }

        static void Main(string[] args)
        {
            Console.WriteLine("Test");

            for (int i = 0; i < 5; i++)
            {
                Thread remover = new Thread(Remover);
                remover.Start();
            }

            Thread adder = new Thread(Adder);
            adder.Start();
        }
    }
}

Answer 1

если я оставлю его включенным достаточно долго, я получу исключение, потому что один из потоков удаления выходит из состояния ожидания, даже когда очередь пуста. Кто-нибудь знает, почему я получаю исключение?

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

Чтобы решить эту проблему, вам нужно использовать цикл вместо «если». Правильный код:

while(q.Count == 0) Monitor.Wait(q);

не

if(q.Count == 0) Monitor.Wait(q);

UPDATE:

Комментатор указывает, что, возможно, ваш вопрос был задуман так: «при каких обстоятельствах потребительский поток может получить монитор, когда очередь пуста?»

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

• Потребительская нить 1: ожидание
• Поток клиента 2: готов
• Producer Thread 3: владеет монитором
• В очереди один элемент.
• Резьба 3 импульса.
• Поток 1 переходит в состояние готовности.
• Тема 3 покидает монитор.
• Резьба 2 входит в монитор.
• Поток 2 потребляет элемент в очереди
• Поток 2 покидает монитор.
• Поток 1 входит в монитор.

А теперь поток 1 находится на мониторе с пустой очередью.

Вообще говоря, рассуждая о подобных проблемах, вы должны думать о «Пульсе» как о голубе с прикрепленной к нему запиской. После освобождения он не имеет связи с отправителем, и если он не может найти свой дом, он умирает в пустыне со своим недоставленным сообщением. Все, что вы знаете, когда используете Pulse, это то, что , если какой-либо поток ожидает, тогда один поток перейдет в состояние готовности в будущем; Вы больше ничего не знаете об относительном времени выполнения операций над потоками.

Answer 2

Ваш код работал бы, если бы был 1 потребитель, но когда их больше, этот механизм не работает, и он должен быть while(q.Count == 0) Monitor.Wait(q)

Следующий сценарий показывает, когда if(q.Count == 0) Monitor.Wait(q) потерпит неудачу (это отличается от Эрика):

• потребитель 1 ожидает
• производитель вставил предмет и пульсирует
• потребитель 1 готов
• производитель снимает блокировку
• потребитель 2 только что вошел Удалить, ему повезло и он получает блокировку
• потребитель 2 видит 1 предмет, не ждет и вынимает предмет
• потребитель 2 снимает блокировку
• потребитель 1 повторно получает блокировку, но очередь пуста

Это происходит точно так же, как документация говорит, что это может произойти:

Когда вызвавший потокPulse снимает блокировку, следующий поток в очереди готовности (который не обязательно является потоком, который был импульсным) получает блокировку.

Answer 3

Эрик, конечно, прав;дело в том, что пока код охватывает все основы;тот факт, что возникает исключение, показывает, что вы этого не сделали.

Условие состязания заключается в том, что между Monitor.Wait на съемнике и Monitor.Pulse на сумматоре (который снимает блокировку; но необязательно немедленно вызвать поток, ожидающий его пробуждения и повторного получения);последующий поток удаления может получить блокировку и сразу же перейти к оператору

if (q.Count == 0) 
{ 
  Monitor.Wait(q); 
} 

и сразу перейти к удалению элемента.Затем поток Pulse d просыпается и предполагает, что там еще есть элемент;но это не так.

Способ исправить это, каким бы ни было проявление состояния расы, как сказал Эрик.

В равной степени, если вы читаете пример на Monitor.Pulse вы увидите настройку, подобную той, что вы сделали здесь, но слегка отличающийся способ сделать это.