Deadlock, race conditions, async/await, thread safety
Конкурентный баг не воспроизводится на вашей машине, проходит все тесты и ждёт пятничного вечера с высокой нагрузкой. Ревью — почти единственное место, где его можно поймать.
Вы освоите приём, который заменяет интуицию в конкурентном коде: явно проставить точки, где выполнение может прерваться, и посмотреть, что успеет вклиниться. Научитесь находить в диффе гонки на чтение-изменение-запись, блокирующие вызовы в событийном цикле, потерянные исключения и неидемпотентные обработчики.
Читая асинхронную функцию, отметьте каждый await — это место, где выполнение может остановиться, и между чтением и записью успеет пройти другой запрос. В многопоточном коде точка прерывания — почти любая инструкция, поэтому там правило жёстче: любое изменение общего состояния подозрительно.
# ❌ гонка: между чтением и записью есть await
async def increment_views(post_id: int):
post = await db.fetch_one("SELECT views FROM posts WHERE id = $1", post_id)
# ↑ здесь может вклиниться другой запрос
await db.execute("UPDATE posts SET views = $1 WHERE id = $2", post.views + 1, post_id)Два одновременных запроса прочитают 100, оба запишут 101. Один просмотр потерян. На счётчике просмотров это терпимо; на балансе, остатке товара или лимите попыток — инцидент.
# ✅ изменение вычисляется базой атомарно
await db.execute("UPDATE posts SET views = views + 1 WHERE id = $1", post_id)Общее правило для любого хранилища: не читайте, чтобы записать вычисленное значение. Пусть значение вычисляет тот, кто владеет данными: UPDATE ... SET x = x + 1, INCR в Redis, атомарный счётчик. Если так нельзя — нужна блокировка или условное обновление:
# ✅ оптимистическая блокировка: обновится только если версия не изменилась
rows = await db.execute(
"UPDATE orders SET status = $1, version = version + 1 "
"WHERE id = $2 AND version = $3",
new_status, order_id, expected_version,
)
if rows == 0:
raise ConcurrentModificationПроверьте себя. Найдите в проекте место, где значение читается, изменяется в коде и записывается обратно. Что произойдёт при двух одновременных вызовах?
Частая ошибка. Считать, что однопоточный событийный цикл защищает от гонок. Он защищает от гонок в памяти между инструкциями, но не между await — а бизнес-логика почти вся между ними.
В асинхронном приложении один синхронный вызов останавливает обработку всех запросов, а не только своего.
# ❌ каждый из этих вызовов «замораживает» весь процесс
async def handler(request):
time.sleep(1) # не asyncio.sleep
data = requests.get(url).json() # не httpx.AsyncClient
rows = psycopg2.connect(...) # синхронный драйвер
text = open("big.csv").read() # блокирующий файловый ввод-вывод
hash = bcrypt.hashpw(pw, salt) # CPU-bound, 100 мс на вызовКоварство в том, что при одном пользователе всё работает идеально. Проблема появляется под нагрузкой и выглядит как «сервис тормозит целиком», без указания на конкретный эндпоинт.
# ✅ асинхронные аналоги для ввода-вывода
async with httpx.AsyncClient(timeout=10) as client:
data = (await client.get(url)).json()
# ✅ CPU-bound — в отдельный поток или процесс
hash = await asyncio.to_thread(bcrypt.hashpw, pw, salt)Что искать в диффе асинхронного кода: любой импорт синхронной библиотеки ввода-вывода, time.sleep, вызовы криптографии и обработки изображений, чтение файлов без aiofiles.
Проверьте себя. Поищите в асинхронных модулях проекта import requests и time.sleep.
Частая ошибка. Оборачивать блокирующий вызов в async def и считать проблему решённой. async не делает код неблокирующим — он лишь разрешает await внутри.
# ❌ корутина создана и не запущена — письмо не отправится, ошибки не будет
async def register(user):
save(user)
send_welcome_email(user) # забыт awaitОшибка тихая: Python лишь предупредит в логе, и то не всегда. Ловится статической проверкой — включите её в CI, чтобы не искать глазами.
Более тонкий случай — задача, на которую никто не смотрит:
# ❌ ссылку на задачу не сохранили: сборщик мусора может её уничтожить,
# а исключение внутри — исчезнет бесследно
asyncio.create_task(process_payment(order))
# ✅ ссылка сохранена, исключения не теряются
task = asyncio.create_task(process_payment(order))
background_tasks.add(task)
task.add_done_callback(background_tasks.discard)И самое частое в реальном коде — gather без обработки отказов:
# ❌ первое же исключение прерывает ожидание, остальные задачи продолжают
# выполняться, а их исключения будут потеряны
results = await asyncio.gather(*tasks)
# ✅ отказы приходят как результаты и обрабатываются явно
results = await asyncio.gather(*tasks, return_exceptions=True)
for task_result in results:
if isinstance(task_result, Exception):
logger.error("partial failure", exc_info=task_result)Проверьте себя. Найдите в проекте create_task и проверьте, что происходит с исключением внутри такой задачи.
Частая ошибка. Ожидать, что gather отменит остальные задачи при отказе одной. Он лишь перестаёт их ждать.
Параллельность без предела — способ уронить чужой сервис и получить блокировку.
# ❌ десять тысяч одновременных запросов
await asyncio.gather(*(fetch(uid) for uid in user_ids))
# ✅ ограничение одновременности
sem = asyncio.Semaphore(10)
async def fetch_limited(uid):
async with sem:
return await fetch(uid)
await asyncio.gather(*(fetch_limited(uid) for uid in user_ids))То же относится к пулу соединений БД: сто параллельных запросов при пуле в десять соединений дают не ускорение, а очередь и таймауты. Вопрос ревьюера: чем ограничено число одновременных операций и совпадает ли этот предел с размером пула.
Проверьте себя. Посмотрите, чем ограничено количество одновременных исходящих запросов в вашем самом «широком» сценарии.
Частая ошибка. Ставить таймаут на отдельный запрос, но не ограничивать их количество. Тысяча запросов с таймаутом в 30 секунд исчерпает всё что угодно.
Классический deadlock — два замка, захватываемые в разном порядке:
# поток A: lock_a → lock_b
# поток B: lock_b → lock_a ← встретились посередине, оба ждут навсегдаЛечение простое и механическое: единый порядок захвата во всей кодовой базе (например, по имени ресурса) плюс таймаут на ожидание. То же справедливо для блокировок на уровне БД: две транзакции, обновляющие те же строки в разном порядке, дают ту же картину.
Второе, что стоит искать, — ввод-вывод под блокировкой:
# ❌ замок удерживается на всё время сетевого запроса
async with lock:
rate = await http.get(rates_url) # 2 секунды все ждут
cache[key] = rate
# ✅ под блокировкой только изменение состояния
rate = await http.get(rates_url)
async with lock:
cache[key] = rateФормально корректно и в первом случае — но пропускная способность падает до одной операции в две секунды.
Проверьте себя. Найдите в проекте блок под блокировкой и проверьте, есть ли внутри обращения к сети или диску.
Частая ошибка. Расширять критическую секцию «чтобы точно было безопасно». Это превращает конкурентный сервис в последовательный.
Самое частое конкурентное требование в продуктовом коде — не блокировки, а идемпотентность. Любой обработчик, который может быть вызван дважды, обязан это выдерживать: вебхуки доставляются «минимум один раз», очереди повторяют сообщения, пользователь нажимает кнопку два раза, балансировщик повторяет запрос по таймауту.
# ❌ повторная доставка вебхука начислит бонус второй раз
async def on_payment_succeeded(event):
await add_bonus(event.user_id, event.amount * 0.05)
# ✅ ключ идемпотентности и уникальный индекс в БД
async def on_payment_succeeded(event):
try:
await db.execute(
"INSERT INTO processed_events (event_id) VALUES ($1)", event.id
)
except UniqueViolation:
return # уже обработали
await add_bonus(event.user_id, event.amount * 0.05)Ключевая деталь: гарантию даёт уникальный индекс в базе, а не проверка «если нет — вставить». Проверка перед вставкой — это то самое чтение-изменение-запись, между которыми успеет вклиниться второй обработчик.
Проверьте себя. Перечислите обработчики в проекте, которые вызываются внешними системами. У скольких есть защита от повторного вызова?
Частая ошибка. Реализовывать идемпотентность через SELECT ... IF NOT EXISTS ... INSERT. Это гонка, просто более редкая.
Дифф на 30 строк, тесты есть, нагрузочных нет.
+@app.post("/orders")
+async def create_order(items: list[ItemIn], user=Depends(current_user)):
+ total = sum(i.price * i.qty for i in items)
+
+ balance = await db.fetch_val(
+ "SELECT balance FROM wallets WHERE user_id = $1", user.id
+ )
+ if balance < total:
+ raise HTTPException(400, "insufficient funds")
+
+ order = await db.fetch_one(
+ "INSERT INTO orders (user_id, total) VALUES ($1, $2) RETURNING *",
+ user.id, total,
+ )
+
+ await db.execute(
+ "UPDATE wallets SET balance = $1 WHERE user_id = $2",
+ balance - total, user.id,
+ )
+
+ await asyncio.gather(*[reserve_stock(i.sku, i.qty) for i in items])
+ send_order_email(user.email, order.id)
+
+ return orderРасставим точки прерывания. Их шесть, и между второй и четвёртой находится вся логика списания.
Гонка на балансе. Между SELECT balance и UPDATE ... SET balance = $1 есть три await. Два одновременных заказа прочитают баланс 1000, каждый проверит, что хватает, и каждый запишет 1000 - total. Итог: две покупки по цене одной, баланс уменьшился на сумму одной. Это не теоретический сценарий — двойной клик по кнопке воспроизводит его надёжно.
Отсутствие транзакции. Заказ вставляется одним запросом, баланс списывается другим. Сбой между ними оставляет заказ, за который не заплатили. Порядок тоже неудачен: сначала создаём заказ, потом списываем.
Неограниченный gather. Заказ на двести позиций — двести одновременных резервирований, при пуле соединений в двадцать. И если резервирование одной позиции упадёт, исключение прервёт ожидание, а остальные резервирования продолжат выполняться — часть товара останется зарезервированной под заказ, который завершился ошибкой.
Забытое ожидание. send_order_email без await — либо корутина, которая не выполнится, либо синхронная функция, которая заблокирует событийный цикл на время работы SMTP. Оба варианта плохи, и по диффу непонятно, какой из них.
Идемпотентность. Повторный запрос создаёт второй заказ и списывает второй раз.
Комментарии в PR:
orders.py:5· blocker Гонка на балансе: междуSELECT balanceиUPDATEтри точки прерывания. Двойной клик — и два заказа спишут сумму одного. Списание должно быть условным и атомарным:UPDATE wallets SET balance = balance - $1 WHERE user_id = $2 AND balance >= $1Если обновлено 0 строк — средств не хватило, и это единственная надёжная проверка. Отдельный
SELECTдля проверки не нужен.
orders.py:10· blocker Создание заказа и списание денег не в одной транзакции. Сбой между ними оставит неоплаченный заказ. Оберните в транзакцию и списывайте до вставки заказа.
orders.py:2· blocker Нет защиты от повторного вызова. Клиент повторит запрос по таймауту — получится два заказа и два списания. Нужен ключ идемпотентности из заголовка запроса и уникальный индекс по нему.
orders.py:21· majorgatherбез ограничения: на большом заказе это сотни одновременных запросов при пуле в 20 соединений. И при отказе одного резервирования остальные останутся зарезервированными — нуженreturn_exceptions=Trueи явный откат уже зарезервированного.
orders.py:22· majorsend_order_emailбезawait. Если это корутина — письмо не отправится вообще; если синхронная функция — заблокирует событийный цикл на время работы SMTP. И отправлять письмо стоит после фиксации транзакции, иначе можно уведомить о заказе, которого нет.
orders.py:3· questiontotalсчитается из цен, присланных клиентом. Это точно так задумано? Клиент может прислать любую цену.
Чем закончилось. Последний вопрос оказался самым дорогим — цены действительно брались из тела запроса, и заказ можно было оформить по цене 1 копейка за товар. Это не конкурентная проблема, но нашлась в том же ревью.
Итоговый код:
@app.post("/orders")
async def create_order(
items: list[ItemIn],
idempotency_key: str = Header(...),
user=Depends(current_user),
):
prices = await products.get_prices([i.sku for i in items]) # цены с сервера
total = sum(prices[i.sku] * i.qty for i in items)
async with db.transaction():
try:
await db.execute(
"INSERT INTO order_requests (key, user_id) VALUES ($1, $2)",
idempotency_key, user.id,
)
except UniqueViolation:
return await orders.get_by_key(idempotency_key) # тот же ответ
charged = await db.execute(
"UPDATE wallets SET balance = balance - $1 "
"WHERE user_id = $2 AND balance >= $1",
total, user.id,
)
if charged == 0:
raise HTTPException(400, "insufficient funds")
order = await orders.create(user.id, total, idempotency_key)
await reserve_stock_all(items, max_concurrency=10) # с откатом
await mailer.send_order_confirmation(user.email, order.id) # после commit
return orderОбратите внимание: проверка баланса исчезла как отдельный шаг. Условие balance >= $1 внутри UPDATE одновременно проверяет и списывает — гонки не остаётся вообще. Это типичный результат правильного разбора конкурентности: не добавляется блокировка, а убирается сама возможность гонки.
ПРЕРЫВАНИЯ отмечены все await; между чтением и записью их нет
АТОМАРНОСТЬ изменение вычисляет хранилище (SET x = x + 1), а не код
ТРАНЗАКЦИИ связанные изменения в одной транзакции; отправка писем — после commit
ИДЕМПОТЕНТНОСТЬ повторный вызов безопасен; гарантию даёт уникальный индекс
БЛОКИРОВКИ единый порядок захвата; ввод-вывод вне критической секции
ЦИКЛ нет синхронного ввода-вывода и CPU-bound в async-функциях
ЗАДАЧИ await не забыт; ссылки на задачи сохранены; исключения не теряются
ПРЕДЕЛ число одновременных операций ограничено и согласовано с пуломГраницы транзакций и уровни изоляции — Работа с БД. Очереди, повторные доставки и backpressure — Масштабируемость. Тренировка: Code Review Python → Ошибки async, Code Review React → Асинхронность в эффектах.
Ключевая мысль: лучший результат разбора конкурентности — не добавленная блокировка, а исчезнувшая возможность гонки.
Далее: Рецензирование работы с БД