Монитор: ждать состояние, а не уведомление

Если механизм уже знаком - листай до «Теперь сам».

Как думаешь: что нужно сохранить

Потребитель увидел пустую очередь и решил уснуть. Между проверкой и засыпанием производитель положил элемент и отправил уведомление. Что разбудит потребителя, если уведомление уже прошло? Этот промежуток нужно убрать из протокола. Отдельный флаг «уведомление пришло» обычно создаёт второй протокол синхронизации со своими гонками вместо исправления первого.

На что это похоже

Представь комнату выдачи деталей. Право войти охраняет замок, а табличка «детали есть» описывает состояние склада. Услышав звонок, ты ещё не получаешь деталь: другой посетитель мог забрать её раньше. Вернувшись в комнату, нужно снова проверить полку. Семафор похож на число выданных талонов: разрешение сохраняется, даже если ожидающего в момент выдачи не было. Звонок условной переменной такого талона не создаёт.

Механизм: от определения к алгоритму

Монитор - защищённые данные и операции над ними, исполняемые с взаимным исключением. Условная переменная связана с замком и очередью ожидающих потоков. Вызов wait атомарно помещает поток в ожидание и отпускает замок. До возврата из wait замок захватывается снова. Проверка предиката и вызов ожидания поэтому выполняются в одной критической секции: производитель не может изменить состояние в опасном промежутке.

У очереди ёмкости C есть инвариант 0≤len(queue)≤C. Производитель ждёт, пока очередь полна; потребитель - пока пуста. Изменение состояния выполняется под общим замком, затем уведомляются ожидающие. В модели Mesa уведомивший поток продолжает работу, а разбуженный ещё должен получить замок. Пока он ждёт, предикат может снова стать ложным. Бывают также ложные пробуждения. Поэтому шаблон всегда while not predicate: wait().

Два условия можно реализовать двумя объектами Condition с одним замком, уведомляя потребителей после вставки и производителей после удаления. В учебном варианте ниже одна переменная и notify_all: просыпаются все, но каждый повторно проверяет свой предикат. Это проще, хотя при большом числе потоков создаёт лишние пробуждения.

Семафор хранит неотрицательный счётчик разрешений: захват уменьшает его либо ждёт, освобождение увеличивает. Для буфера обычно нужны семафоры свободных и занятых мест и отдельно защита структуры очереди. Разрешение семафора сохраняется, уведомление условия - нет. Нельзя заменить wait на ожидание события с другим замком и сохранить тот же доказанный протокол без нового анализа.

Поток, ожидающий через условную переменную, отпускает замок. Цикл активного ожидания под этим же замком делает невозможным изменение предиката другим потоком. Если операция может завершиться исключением, освобождение замка должно гарантироваться конструкцией with или finally, иначе все последующие вызовы навсегда остановятся.

Правило

Предикат проверяется и изменяется под одним замком; ожидание отпускает его атомарно, а каждое пробуждение ведёт к повторной проверке в while.

Разбор: проследим вычисление

У буфера два места, производитель отправляет числа от 0 до 99. Один потребитель суммирует их до специальной метки завершения. Порядок пробуждений недетерминирован, поэтому в выводе только число элементов, сумма и остаток очереди. Эти значения определяются контрактом, а не планировщиком. Ограничение размера проверяется при каждой операции.

from collections import deque
from threading import Condition, Thread

queue = deque()
condition = Condition()
capacity = 2
result = []

def put(value):
    with condition:
        while len(queue) == capacity:
            condition.wait()
        queue.append(value)
        assert 0 <= len(queue) <= capacity
        condition.notify_all()

def get():
    with condition:
        while not queue:
            condition.wait()
        value = queue.popleft()
        assert 0 <= len(queue) <= capacity
        condition.notify_all()
        return value

def produce():
    for value in range(100):
        put(value)
    put(None)

def consume():
    while True:
        value = get()
        if value is None:
            return
        result.append(value)

workers = [Thread(target=produce), Thread(target=consume)]
for worker in workers:
    worker.start()
for worker in workers:
    worker.join()
assert result == list(range(100))
print("получено", len(result), "сумма", sum(result))
print("в очереди", len(queue))

получено 100 сумма 4950
в очереди 0

Разбор: где проходит граница

Разбудили двух потребителей, но в очереди один элемент. Первый получил замок и удалил его. Второй проснулся позже. При if not queue: wait() он сразу продолжит и попытается удалить из пустой очереди. При while увидит пустоту и снова уснёт. Даже если конкретная библиотека никогда не даёт ложных пробуждений, конкуренция потребителей уже требует цикла. Это контрпример протоколу, который тест с одним потребителем может никогда не показать.

Что ломается без этого

Разбуженный поток завершит wait лишь после получения замка. Он увидит все изменения, которые уведомляющий сделал до выхода из критической секции, включая изменения после вызова notify. Если изменение выполнено вне замка, возможны потерянные проверки и повреждение структуры. Ожидание заполнения буфера под занятым замком блокирует производителя, который должен его заполнить. А notify одного случайного ожидающего при одной общей очереди разных условий может разбудить поток, который не способен продвинуться: раздели условия или используй notify_all, пока не доказан более точный протокол.

Зачем это в работе

Пул фоновых задач, канал передачи данных и кеш с ожиданием загрузки используют один шаблон: состояние, предикат, атомарное ожидание. Формулируй предикат словами до написания кода. «Ждать уведомления» недостаточно; «ждать, пока очередь непуста или канал закрыт» уже определяет поведение. Для нескольких потребителей отдельно задай завершение: одна метка None закрывает лишь одного, если остальные не знают о флаге закрытия.

Теперь сам

Проснулись три потребителя, в очереди два элемента, новых вставок нет. Сколько потребителей должно снова уснуть? Что изменится при трёх отдельных метках завершения?