Race conditions in async code

Concurrency in Python services Mid-level Python 3.13+ PostgreSQL 17+ Reviewed 31 Oct 2026 By Best Latest Apps

Predict the output

A hundred tasks each 'increment' a counter. What's printed?

Python 3.13

import asyncio

counter = 0


async def increment() -> None:
    global counter
    seen = counter
    await asyncio.sleep(0)
    counter = seen + 1


async def main() -> None:
    async with asyncio.TaskGroup() as tg:
        for _ in range(100):
            tg.create_task(increment())
    print(counter)


asyncio.run(main())

Options1100A random number between 1 and 1000

Hint 1 of 1

What happens at the await?

Choose an answer, then open Show answer below to check it.

Show answer

Answer: A 1

Correct. Every task reads 0, yields at the await, then writes 0 + 1.

An await is a point where other tasks can run. All 100 tasks read 0 before any of them resumes and writes, so each writes 0 + 1 and the updates overwrite each other.

Why not the others?

B 100. That needs each read and write to happen without a pause between them.
C A random number between 1 and 100. Scheduling here is deterministic: all reads run before any write.
D 0. Every task writes 1.

30-second answer

Yes. A coroutine can be paused at any await, and another can change shared state before it resumes. A read, an await, then a write based on the old read loses updates exactly like threads do. Keep read-modify-write sequences free of awaits, protect them with asyncio.Lock, or push the operation into the database or Redis as one atomic statement.

Practise saying it · 30 s. No recording, just a timer

Races in asyncio?

Say your answer out loud, then check which of these points you covered:

Pauses at every await
Read-await-write loses updates
asyncio.Lock
Atomic DB/Redis ops across processes

Deep dive

Between two awaits, code isn't interrupted by other coroutines; that's the only safe window.

asyncio.Lock is for coroutines in one loop; it does nothing across processes.

Across processes or machines, use database row locks, atomic UPDATEs or Redis scripts.

Check-then-act ('if not exists, create') is the classic form.

The same counter with asyncio.Lock Ran on Python 3.13.5 · mypy --strict

Python 3.13

import asyncio

counter = 0
lock = asyncio.Lock()


async def increment() -> None:
    global counter
    async with lock:
        seen = counter
        await asyncio.sleep(0)
        counter = seen + 1


async def main() -> None:
    async with asyncio.TaskGroup() as tg:
        for _ in range(100):
            tg.create_task(increment())
    print(counter)


asyncio.run(main())

Common mistakes

Assuming single-threaded means race-free.
Using threading.Lock in async code (it blocks the loop).
Relying on an in-process lock when the app runs several workers.

Senior take & follow-ups

Show the await gap, then three fixes: no await in the critical section, asyncio.Lock, atomic storage operations.

Follow-up questions

Think of your answer first, then open the hint.

Why not threading.Lock?
Show hint
Waiting on it blocks the whole event loop.
What about several worker processes?
Show hint
In-process locks don't coordinate them; use the database.

Similar on Android: Threads, @Volatile and synchronized On Android the race is between real threads; here one thread interleaves coroutines at each await.

Suggest a fix if something here is wrong or out of date.