2 Mutex.Wait()
3 buffer.add(event)
4 items.signal()
5 mutex.signal()
Листинг. 4 Решение производителя
Производитель не должен получить эксклюзивный доступ к буферу, пока не произойдет событие. Несколько потоков могут выполнить waitForEvent одновременно. Items семафоры отслеживают число элементов в буфере. Каждый раз, когда производитель добавляет элемент, он сигнализирует items, постепенно увеличивая его на один.
Код потребителя аналогичен.
1 items.wait()
2 mutex.wait()
3 event = buffer.get()
4 mutex.signal()
5 event.process()
Листинг. 5 Решение потребителя
Буферные операции защищены мьютексами, но прежде чем потребитель доберется до него, он должен уменьшить items. Если items равен нулю или отрицательный, потребитель блокируется, пока производитель не подаст сигнал.
Хотя это решение правильное, есть возможность внести в него небольшое улучшение. Представим себе, что по крайне мере один потребитель стоит в очереди, когда производитель сигнализирует items. Если диспетчер позволит работать потребителю, что произойдет дальше? Произойдет немедленная блокировка мьютексом, который пока еще захвачен производителем.
Блокировка и активация являются средне затратными операциями, их излишнее выполнение может ухудшить качество программы. Таким образом, вероятно, было бы лучше перестроить производителя вот так:
1 event = waitForEvent()
2 Mutex.Wait()
3 Buffer.Add(event)
4 Mutex.Signal()
5 Items.Signal()
Листинг. 6 Улучшенное решение производителя
Теперь мы можем не беспокоиться разблокировать потребителя, пока не будем знать, что работа может быть продолжена (кроме того редкого случая, когда другой производитель захватит мьютекс).
Однако в этом решении есть проблема. Семафор items отслеживает число элементов в очереди. Взглянув на код потребителя, мы видимо, что есть ситуация, в которой несколько потребителей постепенно могут уменьшить items, прежде чем один из них захватит мьютекс и удалит переменную из буфера. По крайне мере, на некоторое время items была бы неточной.
Мы могли бы попытаться решить эту проблему по средствам проверки буфера внутри мьютекса.
1 Mutex.Wait()
2 items.wait()
3 event = buffer.get()
4 mutex.signal()
5 event.process()
Листинг. 7 Нарушенное решение потребителя
Однако, это решение плохое.
Deadlock #4
Если потребитель начнет выполнение кода, это может привести к блокировке.
1 mutex.wait()
2 items.wait()
3 event = buffer.get()
4 mutex.signal()
5
6 event.process()
Листинг. 8 Нарушенное решение потребителя
Представим себе, что буфер пуст. Потребитель приходит, захватывает мьютекс, а затем блокируется на items. Когда придет производитель, он заблокируется на мьютексе, что приведет к остановке системы.
Это общая ошибка в коде синхронизации: если всегда ожидать захвата семафора, когда удерживается мьютекс, то существует опасность наступления deadlock. При решении проблемы синхронизации нужно обязательно учитывать этот нюанс.
Производитель – потребитель с конечным буфером
В примере выше описан метод обработки событий потоками с использованием бесконечного буфера (размер ограничен только физическими ресурсами).
Однако в ядре операционной системы существует лимит на свободное место. Для таких вещей, как запросы к дискам, сетевые пакеты, буфер как правило имеет фиксированный размер. В подобных ситуациях у нас есть дополнительное ограничение синхронизации.
• Если производитель приходит, когда буфер полон, он блокируется, пока потребитель не удаляет элемент.
Предположим мы знаем размер буфера. Назовем его BufferSize. Теперь мы имеем семафор, который отслеживает число элементов. Это можно попробовать представить так:
1 if items >= bufferSize: