Kodlo-chan
124 lines
11 KiB
Markdown
124 lines
11 KiB
Markdown
# Задача 06 — потокобезопасная ограниченная очередь (C++)
|
||
|
||
На собеседовании это не вопрос «знаешь ли ты `std::thread`», а «умеешь ли ты не сломать
|
||
счётчик под нагрузкой» — то есть понимаешь ли механику мьютекса и условной переменной
|
||
настолько, чтобы очередь не зависла и не словила гонку данных под ThreadSanitizer.
|
||
|
||
## Интерфейс и требования
|
||
|
||
```c++
|
||
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);`
|
||
недостаточно: после такого пробуждения предикат мог остаться ложным, и поток пойдёт работать
|
||
с данными, которые на самом деле не готовы — баг, который не ловится юнит-тестами и стреляет
|
||
только под конкурентной нагрузкой. Правильный паттерн — цикл с перепроверкой:
|
||
|
||
```c++
|
||
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 может не показать гонку при одном прогоне, даже если она есть в коде? — он ловит гонку по факту конкретной раскладки потоков в этом запуске, а не статическим анализом
|
||
|
||
<details>
|
||
<summary>Проверь себя</summary>
|
||
|
||
1. Почему `cv.wait(lock)` должен принимать именно `unique_lock`, уже захвативший тот же
|
||
мьютекс, что защищает очередь, а не произвольный лок?
|
||
<details><summary>Ответ</summary>Потому что `wait` должен атомарно освободить именно этот
|
||
мьютекс перед сном и захватить его же при пробуждении — иначе разные потоки проверяли бы
|
||
предикат под разными блокировками, и защита состояния очереди перестала бы работать.</details>
|
||
|
||
2. Почему после `close()` `pop()` не должен блокироваться, даже если очередь ещё не пуста?
|
||
<details><summary>Ответ</summary>Требование — `close()` опустошает остаток: `pop()` должен
|
||
продолжать отдавать оставшиеся элементы (`true`) и только когда очередь действительно
|
||
пуста — отдавать `false`, не уходя в ожидание.</details>
|
||
|
||
3. Почему `size()`, который выглядит как «просто чтение одного числа», всё равно требует
|
||
захвата мьютекса?
|
||
<details><summary>Ответ</summary>Потому что запись в `size_` из `push`/`pop` без
|
||
синхронизации с чтением в `size()` — это гонка данных (одна сторона пишет, другая читает
|
||
без общего порядка) — UB по стандарту C++, а не просто «иногда неверное число».</details>
|
||
|
||
</details>
|