Feed Workers Through a Bounded Queue

Your script reads a million lines from a log file and sends each one to an API, with four worker tasks making the calls. Reading is fast and the API is slow. If the reader puts every line into an unlimited queue, the queue grows until most of the file sits in memory.

Give the queue a size limit. Then await queue.put(item) waits while the queue is full, so the reader can run only a few items ahead of the workers. This is called backpressure. When the workers fall behind, the reader slows down to match.

The reader is the producer, the code that makes work. The tasks that take work from the queue are the workers.

import asyncio


async def main() -> None:
    queue: asyncio.Queue[int | None] = asyncio.Queue(maxsize=2)
    results: list[int] = []

    async def worker() -> None:
        while True:
            item = await queue.get()
            try:
                if item is None:  # the stop message
                    return
                await asyncio.sleep(0)  # stands in for the API call
                results.append(item * 10)
            finally:
                queue.task_done()

    async with asyncio.TaskGroup() as group:
        group.create_task(worker())
        for item in range(3):
            await queue.put(item)  # waits while the queue holds 2 items
        await queue.put(None)
        await queue.join()

    print(results)


asyncio.run(main())

[0, 10, 20]

Size the queue for the work you can hold

asyncio.Queue(maxsize=n) holds at most n waiting items. A maxsize of zero, the default, means there is no limit. get() waits while the queue is empty, and put() waits while it is full.

An item leaves the queue as soon as a worker takes it. So three things limit the items in progress at once. The queue holds up to maxsize, each worker holds one, and the producer holds one more while it waits to put it. With maxsize=2 and four workers, the producer is at most 2 + 4 + 1 = 7 items ahead of the finished work.

Mark every item done exactly once

The queue counts unfinished items. Each put() adds one, and each task_done() subtracts one. await queue.join() waits until the count is zero. get() leaves the count alone, so a worker must call task_done() for every item it gets, the stop message included.

Call it in a finally, as the example does. If a worker logs an error and skips task_done(), join() waits forever. If a worker calls it twice for one item, join() can return before the workers have handled every item. A call with the count already at zero raises ValueError.

Stop each worker with a stop message

A worker loops on get(), so it needs to be told when the work is over. After the last real item, put one stop message in the queue per worker. A worker that gets one returns. The queue is first in, first out, so every real item comes out before the stop messages.

A stop message like this is called a sentinel. None works when None is never real data. Otherwise make a class for it, such as class Stop: pass, and check with isinstance(item, Stop). The type checker then knows that anything else is real data.

Run the producer and the workers in one TaskGroup

Suppose every worker crashes. The producer then waits on put() forever, because nothing empties the queue. Run the producer and the workers under one asyncio.TaskGroup, as the example does. When a worker raises, the group cancels the other workers and the producer's waiting put(). Then it raises an ExceptionGroup that holds the error, with nothing left running.

The producer can be the body of the async with, as above, or a task of its own in the group.

Going further

• asyncio.Queue is for tasks on one event loop. To hand items over from another thread, see the article on bridging blocking code.
• Python 3.13 added queue.shutdown(). After it, put() raises asyncio.QueueShutDown, and get() returns the items still waiting, then raises QueueShutDown. Workers can stop on that exception instead of a sentinel. shutdown(immediate=True) also throws away the waiting items. The exercise supports Python 3.12, so it uses sentinels.

See the asyncio queue documentation.