eduplan-cpp-eltex/diag/tasks/06_threads/task.md

Задача 06 — потокобезопасная ограниченная очередь (C++)

На собеседовании это не вопрос «знаешь ли ты std::thread», а «умеешь ли ты не сломать счётчик под нагрузкой» — то есть понимаешь ли механику мьютекса и условной переменной настолько, чтобы очередь не зависла и не словила гонку данных под ThreadSanitizer.

Интерфейс и требования

class BlockingQueue {
public:
    explicit BlockingQueue(size_t capacity);
    ~BlockingQueue();
    void   push(int v);          // блокируется, пока очередь полна
    bool   pop(int& out);        // блокируется, пока пуста; false — если закрыта и пуста
    void   close();              // после close: pop() опустошает остаток и отдаёт false
    size_t size() const;
};

Требования:

• push после close() бросает std::runtime_error;
• закрытие разблокирует все ждущие потоки (никакого вечного ожидания и busy-wait);
• размер очереди никогда не превышает capacity;
• ни одной гонки: тест собирается и гоняется с ThreadSanitizer (в том числе size() вызывается из другого потока — он тоже должен быть потокобезопасным).

Факты для карточек

• base | Каким флагом компилятора включается ThreadSanitizer? — -fsanitize=thread
• base | Что бросает push после close()? — std::runtime_error
• core | Должен ли size() быть потокобезопасным в этой задаче? — да, его тоже вызывают из другого потока и он тоже под захватом мьютекса
• core | Что происходит с ждущими потоками при close()? — все разблокируются (никакого вечного wait)

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

Механизм: mutex + condition_variable + predicate-цикл

Общее состояние очереди (буфер, счётчик элементов, флаг closed) защищено одним std::mutex через RAII-обёртку (std::lock_guard/std::unique_lock) — так unlock() происходит автоматически в деструкторе при выходе из области видимости любым путём, включая исключение, и его невозможно забыть вызвать вручную.

condition_variable::wait(lock) атомарно освобождает мьютекс и усыпляет поток — атомарность важна: если бы освобождение и уход в сон были двумя отдельными шагами, между ними мог бы вклиниться другой поток, изменить состояние и отправить notify, которое тогда потеряется, потому что ждущий поток ещё не успел зайти в wait. При пробуждении wait снова захватывает тот же мьютекс перед тем, как вернуть управление — код после wait уже работает под защитой.

Стандарт C++ прямо разрешает ложные пробуждения (spurious wakeup) — wait может вернуться без единого вызова notify_one/notify_all, просто по решению ОС (futex на Linux иногда пробуждается по внутренним причинам платформы). Из-за этого if (!ready) cv.wait(lock); недостаточно: после такого пробуждения предикат мог остаться ложным, и поток пойдёт работать с данными, которые на самом деле не готовы — баг, который не ловится юнит-тестами и стреляет только под конкурентной нагрузкой. Правильный паттерн — цикл с перепроверкой:

std::unique_lock<std::mutex> lock(mtx_);
while (!(size_ > 0 || closed_))          // predicate-цикл, не if
    not_empty_.wait(lock);

Для push и pop нужны разные условия ожидания (не полна и не пуста или закрыта), поэтому один condition_variable на оба события удобен только тогда, когда notify_all будит все ждущие потоки и каждый сам перепроверяет свой предикат в цикле — так ложные пробуждения для «не того» условия просто не проходят проверку и поток снова засыпает. Если использовать notify_one при одном общем condition_variable для двух разных условий, можно разбудить не тот поток (например, разбудить ждущего push, когда освободилось место для pop), а нужный так и останется спать — отсюда правило: либо notify_all, либо два раздельных condition_variable.

Ловушки

• if вместо while вокруг wait → поток продолжает работу до реального выполнения условия → трудновоспроизводимый баг под нагрузкой, не ловится обычными юнит-тестами.
• Изменение состояния без удержания мьютекса (например, size_++ вне lock_guard) → гонка данных, UB → ловится ThreadSanitizer как data race при size() из другого потока.
• notify_one при общем condition_variable на два разных предиката → не тот поток проснулся, нужный продолжает ждать → зависание, видно по таймауту теста, а не по явной ошибке.
• Забыть разбудить всех при close() → часть потоков навсегда в wait → зависание процесса, тест не завершается за отведённое время.

Факты для карточек

• base | Что атомарно делает cv.wait(lock) при входе? — освобождает мьютекс и усыпляет поток одной неделимой операцией
• core | Почему while вокруг wait, а не if? — из-за spurious wakeup: wait может вернуться без вызова notify
• core | Чем опасен один condition_variable на два разных предиката вместе с notify_one? — можно разбудить не тот поток, а нужный останется ждать
• deep | На каком примитиве ОС реализован condition_variable на Linux? — futex

Почему дальше: раз тест явно требует ThreadSanitizer и отсутствия зависаний, логично понять, что именно проверяет grade.py и почему «просто работает на глаз» здесь недостаточно.

Проверка

Запуск: python3 grade.py 06. Критерий: все ok, TSan чистый, нет зависания. TSan инструментирует каждое обращение к памяти и отслеживает happens-before между потоками во время исполнения — то есть ловит гонку по факту конкретного запуска, а не статическим анализом кода, поэтому «прогнать разок и не увидеть краша» не равно «гонки нет»: раскладка потоков в конкретном запуске могла просто не проявить проблему.

Факты для карточек

• base | Какой командой запускается проверка задачи 06? — python3 grade.py 06
• core | Почему TSan может не показать гонку при одном прогоне, даже если она есть в коде? — он ловит гонку по факту конкретной раскладки потоков в этом запуске, а не статическим анализом

Проверь себя

1. Почему cv.wait(lock) должен принимать именно unique_lock, уже захвативший тот же мьютекс, что защищает очередь, а не произвольный лок?

   Ответ

   Потому что `wait` должен атомарно освободить именно этот мьютекс перед сном и захватить его же при пробуждении — иначе разные потоки проверяли бы предикат под разными блокировками, и защита состояния очереди перестала бы работать.

2. Почему после close() pop() не должен блокироваться, даже если очередь ещё не пуста?

   Ответ

   Требование — `close()` опустошает остаток: `pop()` должен продолжать отдавать оставшиеся элементы (`true`) и только когда очередь действительно пуста — отдавать `false`, не уходя в ожидание.

3. Почему size(), который выглядит как «просто чтение одного числа», всё равно требует захвата мьютекса?

   Ответ

   Потому что запись в `size_` из `push`/`pop` без синхронизации с чтением в `size()` — это гонка данных (одна сторона пишет, другая читает без общего порядка) — UB по стандарту C++, а не просто «иногда неверное число».