Skip to content

Coordinating Producers and Consumers with asyncio.Condition

Some coordination problems are not "wait for a signal" but "wait until the shared state looks like this": until the in-flight count drops below a limit, until a cache entry is populated, until a batch has twenty items or the stream has ended. Engineers reach for asyncio.Event first, and the result is a thicket of events that must be set and cleared in exactly the right order, with a wakeup lost whenever a task checks the state, finds it unsatisfied, and only then starts waiting — just after another task changed it. asyncio.Condition is the primitive designed for this. It pairs a lock that protects the state with a waiting room: tasks check a predicate while holding the lock and, if it is false, release the lock and sleep atomically until another task changes the state and notifies them. This guide builds a bounded buffer on it, explains when notify() is enough and when it causes stalls, and covers what happens to the lock when a waiting task is cancelled or times out.

Prerequisites

One wait on a Condition 5 stages from acquire lock to re-test. One wait on a Condition acquire lock async with cond test predicate state ready? release + sleep atomic wait() notified reacquire lock re-test then proceed Release-and-sleep is atomic, so no change can slip in between check and wait.

1. Wait for a predicate, not for a signal

The core loop is: acquire the condition's lock, test the state, and if it is not ready, wait() — which releases the lock, suspends, and reacquires the lock before returning. Because other tasks may change the state between a notification and the waiter's turn to run, the test must be repeated after every wakeup. wait_for(predicate) does exactly that loop for you.

import asyncio


class InflightLimiter:
    """Allow new work only while fewer than `limit` operations are in flight."""

    def __init__(self, limit: int) -> None:
        self.limit = limit
        self.inflight = 0
        self._cond = asyncio.Condition()

    async def acquire(self) -> None:
        async with self._cond:
            await self._cond.wait_for(lambda: self.inflight < self.limit)   # re-checked on every wakeup
            self.inflight += 1

    async def release(self) -> None:
        async with self._cond:
            self.inflight -= 1
            self._cond.notify()                 # one slot freed: one waiter can proceed


async def worker(limiter: InflightLimiter, peak: list[int], n: int) -> None:
    await limiter.acquire()
    try:
        peak[0] = max(peak[0], limiter.inflight)
        await asyncio.sleep(0.01)
    finally:
        await limiter.release()


async def main() -> None:
    limiter = InflightLimiter(limit=3)
    peak = [0]
    await asyncio.gather(*(worker(limiter, peak, n) for n in range(20)))
    print("peak in flight:", peak[0], "| finally in flight:", limiter.inflight)   # 3 | 0


asyncio.run(main())

This is what a semaphore does, which makes it a good first example: the predicate is simple enough to check by eye. Condition earns its place when the predicate involves more than a counter — as in the next step.

Verify: twenty workers never exceed three in flight, and the count returns to zero.

2. Build a bounded buffer with two predicates

A bounded buffer has producers that must wait while it is full and consumers that must wait while it is empty. Two conditions sharing one lock keep the two groups of waiters separate, so a producer adding an item wakes consumers only, and a consumer removing an item wakes producers only.

import asyncio
from collections import deque


class BoundedBuffer:
    def __init__(self, capacity: int) -> None:
        self.capacity = capacity
        self.items: deque = deque()
        self.closed = False
        lock = asyncio.Lock()
        self._not_full = asyncio.Condition(lock)     # producers wait here
        self._not_empty = asyncio.Condition(lock)    # consumers wait here

    async def put(self, item) -> None:
        async with self._not_full:
            await self._not_full.wait_for(lambda: len(self.items) < self.capacity or self.closed)
            if self.closed:
                raise RuntimeError("buffer closed")
            self.items.append(item)
            self._not_empty.notify()                 # exactly one consumer can take it

    async def get(self):
        async with self._not_empty:
            await self._not_empty.wait_for(lambda: self.items or self.closed)
            if not self.items:
                raise EOFError("buffer closed and drained")
            item = self.items.popleft()
            self._not_full.notify()                  # exactly one producer can add
            return item

    async def close(self) -> None:
        async with self._not_empty:
            self.closed = True
            self._not_empty.notify_all()             # every consumer must see the change
            self._not_full.notify_all()


async def main() -> None:
    buffer = BoundedBuffer(capacity=2)
    received: list[int] = []

    async def produce() -> None:
        for i in range(10):
            await buffer.put(i)
        await buffer.close()

    async def consume() -> None:
        try:
            while True:
                received.append(await buffer.get())
        except EOFError:
            pass

    await asyncio.gather(produce(), consume(), consume())
    print(sorted(received))                          # 0..9, each exactly once


asyncio.run(main())

The closed flag is part of both predicates, which is what makes shutdown work: a consumer asleep on an empty buffer wakes, sees closed, and exits once the buffer is drained. Every state change that can make a predicate true is followed by a notify on the condition whose waiters test that predicate.

Verify: two consumers receive all ten items exactly once and both exit cleanly after close().

One condition or two? 2 columns contrasting one shared condition, two conditions, one lock. One condition or two? one shared condition mixed waiters producers and consumers notify() may wake wrong group needs notify_all() extra wakeups two conditions, one lock not_full / not_empty each holds one kind notify() wakes a useful task no wasted wakeups same mutual exclusion Split conditions by predicate so a single notify is always correct.

3. Choose notify() or notify_all() deliberately

notify(n) wakes up to n waiters; notify_all() wakes every waiter. notify() is efficient but correct only when any single waiter can make progress on the change. When waiters test different predicates on the same condition, waking one that cannot proceed while another that could stays asleep is a stall.

import asyncio


class Resource:
    def __init__(self) -> None:
        self.version = 0
        self.cond = asyncio.Condition()

    async def wait_for_version(self, wanted: int, woke: list[int]) -> None:
        async with self.cond:
            await self.cond.wait_for(lambda: self.version >= wanted)
            woke.append(wanted)

    async def bump(self, all_waiters: bool) -> None:
        async with self.cond:
            self.version += 1
            if all_waiters:
                self.cond.notify_all()
            else:
                self.cond.notify()                   # wakes one waiter, maybe the wrong one


async def run(all_waiters: bool) -> list[int]:
    resource = Resource()
    woke: list[int] = []
    waiters = [asyncio.create_task(resource.wait_for_version(v, woke)) for v in (3, 1)]
    await asyncio.sleep(0)
    await resource.bump(all_waiters)                 # version 1: only the second waiter is satisfied
    await asyncio.sleep(0.01)
    for task in waiters:
        task.cancel()
    await asyncio.gather(*waiters, return_exceptions=True)
    return woke


async def main() -> None:
    print("notify():    ", await run(all_waiters=False))   # [] : woke the version-3 waiter, which slept again
    print("notify_all():", await run(all_waiters=True))    # [1]


asyncio.run(main())

With notify(), the first waiter in line wanted version 3; it woke, re-checked its predicate, and went back to sleep, while the waiter that wanted version 1 was never woken. notify_all() wakes both and lets each predicate decide. Use notify() only with a single kind of waiter per condition — which is exactly why the bounded buffer used two conditions.

Verify: the notify() run reports no waiter satisfied, and the notify_all() run reports the version-1 waiter.

4. Handle cancellation and timeouts while waiting

A waiting task can be cancelled or time out. In both cases wait() reacquires the lock before the exception propagates, so the async with block exits normally and releases it. The waiter must not assume the predicate is true after an exception, and code that notified it loses nothing — the next notification goes to someone else.

import asyncio


async def main() -> None:
    cond = asyncio.Condition()

    async def waiter() -> None:
        async with cond:
            try:
                await cond.wait()
            except asyncio.CancelledError:
                print("cancelled while waiting; lock held again:", cond.locked())
                raise

    task = asyncio.create_task(waiter())
    await asyncio.sleep(0.01)
    task.cancel()
    await asyncio.gather(task, return_exceptions=True)
    print("lock released after the block:", not cond.locked())

    async with cond:
        try:
            async with asyncio.timeout(0.02):
                await cond.wait_for(lambda: False)   # a state that never arrives
        except TimeoutError:
            print("timed out; lock held inside the block:", cond.locked())


asyncio.run(main())

Put the timeout inside the async with cond: block, as shown, so the lock is still held when the timeout is handled and any recovery can inspect the state safely. A timeout wrapped around the whole block also works, but its handler runs after the lock was released.

Verify: the output shows the lock held again inside the cancelled waiter, released after its block, and held inside the timed-out block.

Which primitive for this wait? A decision on What are tasks waiting for with 3 outcomes. Which primitive for this wait? What are tasks waiting for? an item to hand off asyncio.Queue already bounded, FIFO a one-time signal asyncio.Event set once, stays set state matching a rule asyncio.Condition predicate + notify Condition is the general tool; reach for it when the simpler ones do not fit.

5. Test the coordination under load

Condition-based code fails through lost or missed wakeups that only appear with many tasks and unlucky timing. A stress test with many producers and consumers, randomised delays, and a hard deadline turns a potential hang into a failing test.

import asyncio
import random


async def stress(buffer_capacity: int = 3, producers: int = 8, consumers: int = 5,
                 items_each: int = 200, seed: int = 7) -> None:
    rng = random.Random(seed)
    buffer = BoundedBuffer(buffer_capacity)
    produced = [f"p{p}-{i}" for p in range(producers) for i in range(items_each)]
    received: list[str] = []

    async def produce(p: int) -> None:
        for i in range(items_each):
            if rng.random() < 0.1:
                await asyncio.sleep(0)
            await buffer.put(f"p{p}-{i}")

    async def consume() -> None:
        try:
            while True:
                received.append(await buffer.get())
                if rng.random() < 0.1:
                    await asyncio.sleep(0)
        except EOFError:
            pass

    async with asyncio.timeout(10):                  # a hang fails the test instead of CI
        consumer_tasks = [asyncio.create_task(consume()) for _ in range(consumers)]
        await asyncio.gather(*(produce(p) for p in range(producers)))
        await buffer.close()
        await asyncio.gather(*consumer_tasks)

    assert sorted(received) == sorted(produced), "lost or duplicated items"
    print(f"{len(received)} items through a buffer of {buffer_capacity}: ok")


asyncio.run(stress())

Change one notify() in BoundedBuffer to a missing call, and the test hangs until the deadline and fails — which is the point. Deterministic interleavings for specific races are covered in Testing Async Code.

Verify: 1,600 items pass through the buffer with none lost or duplicated, well within the deadline.

Verification

Condition-based coordination is correct when:

  • Waiting uses predicates: every wait is wait_for(predicate) or a loop that re-checks state after waking.
  • Every state change notifies: each change that can make a predicate true is followed by a notify on the matching condition.
  • notify() is used only with uniform waiters: mixed predicates on one condition use notify_all(), or are split into separate conditions.
  • Shutdown is part of the state: a closed flag in the predicates wakes and releases every waiter.
  • A stress test with a deadline passes, proving no lost wakeups under load.

Pitfalls & edge cases

  • Waiting without the lock. wait() and notify() outside async with cond: raise RuntimeError: cannot wait on un-acquired lock and cannot notify on un-acquired lock.
  • Long work while holding the lock. The lock serialises all state access; awaiting I/O inside the block blocks every producer and consumer. Hold it only to read and change state.
  • Using Condition where a Queue fits. For a plain FIFO handoff, asyncio.Queue is simpler and already correct. Reach for Condition when the readiness condition is richer than "an item exists".
  • Sharing a condition across event loops. Like every asyncio primitive, a condition belongs to one loop; threads need threading.Condition or a thread-safe bridge.
  • Notifying before changing state. Waiters woken before the change see the old state and go back to sleep; with notify(), the wakeup is simply lost.

Frequently Asked Questions

When should I use asyncio.Condition instead of asyncio.Event?

Use Condition when tasks must wait until shared state satisfies a predicate that can change back and forth, such as a buffer not being full or a counter being below a limit. Event is a single flag that stays set until cleared, and coordinating changing state with events tends to lose wakeups between checking and waiting.

What is the difference between notify and notify_all in asyncio?

notify wakes up to one waiter, or n waiters when given a count, while notify_all wakes every waiter. notify is correct only when any single waiter can make progress on the change. When waiters test different predicates on the same condition, notify can wake one that cannot proceed and leave the right one asleep.

Why use wait_for instead of wait on an asyncio Condition?

A waiter woken by a notification may find the state changed again by another task before it runs, so it must re-check the state. wait_for(predicate) repeats wait until the predicate is true, holding the lock whenever it evaluates the predicate, which avoids writing that loop by hand.

Does a cancelled asyncio Condition wait release the lock?

wait reacquires the condition's lock before raising CancelledError or a timeout, so the exception propagates from inside the async with block while the lock is held, and the block's exit then releases it. The waiter should not assume its predicate is true after the exception.