Многопоточность на C++ через практику и замеры

— добавите в shardkv EXPIRE/TTL с фоновым потоком, который спит ровно до ближайшего дедлайна.

3.1. Как ждать правильно

Condition variable — это механизм «усни, пока тебя не разбудят». Сама по себе она ничего не хранит и ничего не знает о состоянии программы. Это ключевая мысль, из которой следует всё остальное: состояние хранится в обычных переменных под мьютексом, а condition variable лишь будит тех, кто этого состояния ждёт.

Минимальный правильный шаблон выглядит так:

Листинг 3.1 — «дождаться события» тремя способами (ex01_event.cpp)

Три обязательных элемента:

1. Состояние (здесь ready_) — обычная переменная, которую читают и меняют только под мьютексом. Даже если это bool.

2. Мьютекс — один и тот же для изменения состояния и для ожидания.

3. Предикат в wait — функция, которая проверяет состояние.

Что происходит внутри cv.wait(lock, pred):

Листинг 3.2 — что делает cv.wait(lock, pred)

Слово «атомарно» здесь критично: между освобождением мьютекса и засыпанием нет окна, в которое могло бы проскочить уведомление. Поэтому wait принимает именно unique_lock: ему нужно отпускать и снова захватывать мьютекс, а lock_guard так не умеет.

3.2. Три способа ошибиться

Ошибка 1: wait без предиката и spurious wakeup

У condition variable есть официально разрешённое право проснуться без всякого notify. Это называется spurious wakeup (ложное пробуждение). Причины — детали реализации ОС: так дешевле реализовать примитив. Стандарт явно разрешает такое поведение.

Если код выглядит так:

Листинг 3.3 — if вместо while

то после ложного пробуждения он пойдёт дальше, считая, что событие произошло. Лечится просто: всегда используйте перегрузку wait с предикатом (или явный цикл while). Предикат перепроверит состояние после любого пробуждения.

Есть и вторая причина для цикла, совсем не экзотическая. Даже при честном пробуждении к моменту, когда поток снова захватил мьютекс, состояние могло измениться: другой потребитель успел забрать элемент из очереди. Проснулся — не значит, что условие выполняется.

Ошибка 2: lost wakeup — уведомление, которое никто не услышал

Condition variable не запоминает уведомления. Если notify случился, когда никто не ждал, он просто пропал. Поэтому ждущий поток обязан проверить состояние до засыпания, и проверка с засыпанием должны быть атомарны относительно изменения состояния. Это обеспечивает мьютекс — но только если и изменение состояния идёт под ним.

Вот как это ломается:

Листинг 3.4 — потерянное пробуждение: два потока по времени

Поток B поменял флаг и уведомил в окне между проверкой и засыпанием потока A. Если бы B менял ready под тем же мьютексом, он не смог бы попасть в это окно: A держит мьютекс от проверки до момента, когда wait атомарно отпускает его и засыпает.

Самое неприятное: эта ошибка не лечится тем, что ready объявлен как std::atomic. Атомарность флага убирает data race, но не закрывает окно между проверкой и засыпанием. В задаче 3.1 вас ждёт именно такой код.

Правило: изменение состояния, которого ждут через condition variable, выполняется под тем же мьютексом, что и ожидание. Без исключений, даже для атомиков.

Ошибка 3: разные мьютексы или состояние вне мьютекса

Вариации предыдущей ошибки: ждут под одним мьютексом, а состояние меняют под другим; или предикат читает данные, которые защищены совсем другим локом. Результат тот же — окно для потерянного уведомления, плюс data race в предикате. Одна condition variable — один мьютекс — одно защищаемое состояние.

3.3. notify_one, notify_all и где их вызывать

notify_one будит одного ждущего (какого — не определено). Используйте, когда изменение состояния может обработать ровно один поток: в очередь положили один элемент.

notify_all будит всех. Используйте, когда изменение важно для всех: очередь закрыли, конфиг загружен, наступил момент «всем стоп». Не используйте его «на всякий случай» в горячем пути: при 32 ждущих потоках один элемент в очереди разбудит всех 32, один заберёт элемент, 31 проверит предикат и снова уснёт. Это называется thundering herd (паническое стадо), и на нагруженных системах оно стоит заметных процентов CPU.

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

Уведомлять под локом или после? Оба варианта корректны, если состояние менялось под локом. Но если вызвать notify под локом, проснувшийся поток сразу упрётся в мьютекс, который уведомляющий ещё держит, и уснёт снова, теперь уже на мьютексе. Поэтому обычно делают так:

Листинг 3.5 — изменить состояние под локом, уведомить после

Есть одно исключение. Если сразу после уведомления condition variable может быть уничтожена (например, ждущий поток проснётся, увидит флаг и удалит объект, содержащий cv), то notify после освобождения лока может обратиться к уже уничтоженной cv. В таких сценариях уведомляют под локом. Это редкость, но это ровно тот тип бага, который находят в продакшене.

3.4. Ожидание с таймаутом: wait_for и wait_until

Ждать вечно можно не всегда. Для ожидания с ограничением есть две функции:

— wait_for(lock, duration, pred) — ждать не дольше заданной длительности;

— wait_until(lock, time_point, pred) — ждать до заданного момента.

Обе с предикатом возвращают bool: true, если предикат стал истинным, false, если время вышло, а предикат так и остался ложным.

Листинг 3.6 — ожидание с дедлайном (ex03_deadline.cpp)

Почему дедлайн лучше длительности. Представьте функцию, которой дано 100 мс на всё, а она ждёт несколько событий подряд. Если каждое ожидание делать через wait_for(100ms), общая длительность может стать 300 мс. Правильно вычислить дедлайн один раз в начале и ждать до него во всех вызовах. Так и сделано в примере: один дедлайн на оба ожидания.

Та же логика внутри одного ожидания: если вы реализуете цикл ожидания сами и после ложного пробуждения снова зовёте wait_for с той же длительностью, отсчёт начинается заново, и под постоянными ложными пробуждениями поток может не выйти никогда. wait_until с фиксированным моментом от этого защищён. (Перегрузки с предикатом из стандартной библиотеки сделаны правильно, но если пишете цикл руками — помните.)

И, как всегда для таймаутов, используйте steady_clock (глава 2, раздел 2.8).

3.5. condition_variable_any и stop_token

В главе 1 мы увидели проблему: поток, заснувший в блокирующем вызове, не узнает о запросе остановки, пока его не разбудят. Для condition variable в C++20 есть готовое решение. std::condition_variable_any — более общая версия condition_variable, которая работает с любым типом лока и, главное, умеет ждать с stop_token:

Листинг 3.7 — ожидание с stop_token

Такое ожидание просыпается, когда предикат стал истинным или запрошена остановка. Возвращаемое значение — результат предиката: если вернулось false, значит, вас разбудила остановка (или дедлайн), а не событие.

Внутри это сделано через stop_callback, который будит condition variable при request_stop() — ровно тот приём, который мы писали руками для сокетов в главе 1. Плата за удобство — condition_variable_any чуть тяжелее обычной condition_variable (у неё свой внутренний мьютекс). В большинстве случаев это несущественно, но в очень горячих путях имейте в виду.

3.6. Блокирующая очередь

Очередь «производитель — потребитель» — самый распространённый многопоточный паттерн: одни потоки создают задачи, другие их выполняют. Соберём её из всего, что знаем:

Листинг 3.8 — производитель-потребитель на неограниченной очереди с закрытием (ex02_blocking_queue.cpp)

На что обратить внимание:

— pop возвращает std::optional. Проверка «не пусто ли» и извлечение — одна операция под одним локом (урок главы 2).

— push уведомляет одного, close — всех. Новый элемент может забрать один потребитель; о закрытии должны узнать все.

— После close элементы можно дочитать. Предикат просыпается, если есть элементы ИЛИ очередь закрыта, а pop возвращает nullopt, только когда очередь закрыта и пуста. Ни одна принятая задача не теряется.

— pop принимает stop_token. Потребителя можно остановить, даже если очередь не закрыта и пуста.

— Каждый потребитель пишет результат в свою ячейку. Общий счётчик потребовал бы ещё одного мьютекса; отдельные ячейки, которые читаются после join, — нет.

У этой очереди один серьёзный недостаток: она неограниченная. Если производитель быстрее потребителей, очередь растёт, пока не закончится память. В продакшене это классический сценарий падения под нагрузкой: сервис не тормозит, а копит задачи в памяти, пока его не убьёт OOM killer. Ограниченную очередь, которая заставляет производителя ждать, вы напишете в задаче 3.2, а о стратегиях работы с переполнением (backpressure) поговорим в главе 12.

3.7. Graceful shutdown: как остановить систему потоков

Корректная остановка — одна из самых недооценённых задач. Сервис, который при деплое теряет запросы, зависает на остановке (и его убивают по таймауту через SIGKILL) или падает в деструкторах, — это сервис, который боятся деплоить. А сервис, который боятся деплоить, деплоят в пятницу вечером, когда «ну наконец-то всё готово».

Общая схема для системы «производители → очередь → потребители»:

1. Перестать принимать новую работу. Производители останавливаются (или закрываются входные соединения).

2. Закрыть очередь. close() будит всех потребителей.

3. Дочитать остаток. Потребители обрабатывают то, что уже в очереди, и выходят, когда pop вернул nullopt.

4. Дождаться потребителей. join (в нашем случае — деструкторы jthread).

5. Только после этого разрушать общие ресурсы, которыми пользовались потоки.

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

Порядок шагов важен. Закрыть очередь до остановки производителей — значит, часть задач отвергнется (push вернёт false), и производители должны уметь это обработать. Разрушить ресурсы до join потребителей — use-after-free, о котором мы говорили в главе 1.

3.8. Грабли с ревью

— wait без предиката (или через if, а не while).

— Изменение состояния без мьютекса, «потому что оно атомарное».

— notify_one при закрытии или остановке, когда ждущих может быть несколько.

— notify_all на каждый элемент в горячем пути.

— Предикат, который читает данные под другим мьютексом или вообще без него.

— wait_for в цикле с той же длительностью вместо дедлайна.

— Ожидание, которое нельзя прервать при остановке сервиса.

— Неограниченная очередь между компонентами с разной скоростью.

— sleep_for как синхронизация: «подождём, пока поток успеет».

3.9. Что спросят на собеседовании

Зачем condition variable нужен мьютекс? (middle) Чтобы проверка состояния и засыпание были атомарны относительно изменения состояния. Без этого возможен lost wakeup. Мьютекс защищает состояние, cv только будит.

Что такое spurious wakeup и как с ним бороться? (middle) Пробуждение без notify, разрешённое стандартом. Бороться — всегда проверять предикат в цикле (перегрузка wait с предикатом).

Что такое lost wakeup? (middle/senior) Уведомление, отправленное, когда ещё никто не ждал. Возникает, если состояние меняется без того мьютекса, под которым ждут. Хороший ответ: «атомарный флаг не спасает» — и объяснение, почему.

notify_one или notify_all? (middle) notify_one, когда событие может обработать один поток; notify_all, когда событие важно всем (закрытие, остановка). Бонус: thundering herd.

Уведомлять под локом или после? (senior) Оба корректны; после — эффективнее (проснувшийся не упрётся в мьютекс); под локом — когда cv может быть уничтожена сразу после пробуждения ждущего.

Почему wait_until лучше wait_for? (senior) Дедлайн не сдвигается при повторных ожиданиях и ложных пробуждениях. Общий бюджет времени на операцию соблюдается.

Как корректно остановить пул потребителей? (senior) Перестать принимать работу, закрыть очередь, разбудить всех, дочитать остаток (или выйти по stop_token), дождаться потоков, потом освобождать ресурсы. Сильный ответ обсуждает, какие задачи допустимо потерять.

3.10. Сквозной проект shardkv, шаг 3: TTL и поток, который спит ровно сколько нужно

Что делаем:

— команды EXPIRE и PEXPIRE (время жизни в секундах и миллисекундах), TTL и PTTL (сколько осталось) с семантикой Redis: −2 — ключа нет, −1 — ключ без времени жизни;

— SET снимает время жизни, INCR сохраняет;

— ленивое истечение: любая операция с ключом сначала проверяет, не истёк ли он;

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

Изначально в плане книги этот шаг включал и ограниченную очередь между приёмом соединений и обработчиками. Я перенёс её в главу 7, где появится пул потоков: сейчас она потребовала бы переделать модель «поток на соединение», а заодно выдала бы решение задачи 3.2.

Как устроен фоновый поток

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

— дедлайны хранятся в мин-куче (std::priority_queue), на вершине — ближайший;

— поток спит через wait_until ровно до ближайшего дедлайна;

— если появился дедлайн раньше текущего ближайшего, expire() будит поток через notify, и тот пересчитывает время сна;

— если дедлайн стал не ближайшим, будить поток не нужно: он и так проснётся вовремя;

— остановка через stop_token работает, даже если поток спит до дедлайна через час.

Тонкость: запись в куче может устареть. Ключу назначили TTL, а потом его перезаписали через SET (TTL снят) или назначили новый TTL. Удалять запись из середины priority_queue нельзя, поэтому мы оставляем её и при извлечении сверяем: совпадает ли дедлайн в куче с текущим expires_at ключа. Не совпадает — запись просто выбрасывается.

Код хранилища

Листинг 3.9 — хранилище shardkv, шаг 3 (storage.hpp)

Листинг 3.10 — реализация (storage.cpp)

Что стоит разобрать подробно:

— run_expirer держит мьютекс почти всё время, кроме сна. wait_until отпускает его на время сна, поэтому клиенты не блокируются. Но очистка (purge_expired_locked) идёт под мьютексом, и если одновременно истечёт миллион ключей, все клиенты подождут. В Redis эту проблему решают, ограничивая работу за один проход; для нашего масштаба это преждевременная оптимизация, но держите её в голове.

— expire будит поток только при новом ближайшем дедлайне и делает notify после освобождения лока.

— Предикат в wait_until — «появился дедлайн раньше того, до которого я сплю». Если поток проснулся ложно или по таймауту, цикл просто пересчитает всё заново: purge, затем новое ожидание.

— get теперь не const. Ленивое истечение удаляет ключ, то есть get может менять хранилище. Делать map_ mutable ради сохранения const было бы обманом: const-метод, который меняет видимое состояние, сбивает с толку и тех, кто читает код, и тех, кто рассуждает о потокобезопасности.

Протокол и запуск

В протокол добавлены команды EXPIRE, PEXPIRE, TTL, PTTL (полный код — в архиве главы). В main.cpp появился фоновый поток:

Листинг 3.11 — порядок объявления storage и expirer (main.cpp)

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

Тесты

Листинг 3.12 — тесты (test_ttl.cpp)

Обратите внимание на функцию eventually. Тесты, завязанные на время, — главный источник flaky-тестов. Плохой вариант: sleep_for(100ms), потом проверка. На медленной CI-машине 100 мс не хватит, и тест упадёт. Хороший вариант: проверять условие в цикле до щедрого дедлайна. Когда всё работает, тест заканчивается за миллисекунды; когда машина медленная, тест подождёт, а не упадёт зря.

Тест EarlierDeadlineWakesExpirer проверяет главную многопоточную логику: поток заснул на час, а потом появился ключ, который истекает через 10 мс. Если expire не разбудит поток, тест не дождётся удаления. А ExpirerStopsPromptly проверяет, что stop_token будит поток, спящий до далёкого дедлайна.

Все тесты проходят под TSan и ASan 10 раз подряд.

Замеры

Добавили ли TTL накладные расходы на обычные GET и SET? Сравним шаг 2 и шаг 3 тем же kvbench (4 клиента, режим mix, три прогона подряд):

Таблица 3.1 — накладные расходы TTL: шаг 2 и шаг 3, ops/s

Вывод: разница между шагами меньше, чем разброс между прогонами одной и той же версии (от 102 до 141 тысячи). Значит, по этим данным мы не можем сказать, что TTL что-то замедлил. Это нормальный и важный результат. Типичная ошибка — запустить бенчмарк по разу, увидеть 141 против 106 и написать в отчёте «оптимизация дала +33%». Без нескольких прогонов и оценки разброса любые цифры бенчмарка — гадание на кофейной гуще.

Задачки для разминки мозгов

Задача 3.1 (лёгкая). Потерянное пробуждение

Листинг 3.13 — код к задаче 3.1 (task3_1_lost_wakeup.cpp)

Эта программа зависает. У меня — каждый запуск, но на вашей машине может понадобиться несколько. Флаг атомарный, notify вызывается — почему поток спит вечно?

Нарисуйте трассу событий двух потоков (как в разделе 3.2), которая приводит к зависанию. Найдите все проблемы в коде (их больше одной) и исправьте.

Проверка: исправленная программа проходит 100 000 итераций без зависаний, 10 запусков подряд, под TSan; трасса показывает конкретное окно, в которое попадает notify.

Задача 3.2 (средняя). BoundedQueue

Напишите шаблон ограниченной блокирующей очереди BoundedQueue с ёмкостью, заданной в конструкторе:

Листинг 3.14 — интерфейс BoundedQueue

Требования:

— push блокируется, когда очередь заполнена, и просыпается, когда место освободилось;

— после close() оставшиеся элементы можно дочитать, затем pop возвращает nullopt;

— производители, ждущие свободного места, при close() просыпаются и получают false;