Skip to content

Bridging AnyIO Memory Object Streams and asyncio Queues

asyncio.Queue is a fine data structure and an incomplete channel: it has no notion of being closed, so every producer–consumer pipeline built on it reinvents sentinel values, and it is unbounded by default, so back-pressure is opt-in. AnyIO's memory object streams are the same idea with those two gaps filled — the buffer size is a constructor argument rather than an option, and closing the send side ends the consumer's async for with no sentinel at all. Verified on both backends: 100 items through a 10-slot stream in 0.11 s, and a consumer loop that exited cleanly the moment the producer's handle closed.

Prerequisites

Memory object streams against asyncio.Queue 2 columns contrasting memory object stream, asyncio.Queue. Memory object streams against asyncio.Queue memory object stream closing is part of it closing the sender ends async for clone() for extra producers or consumers statistics() reports buffer use bounded by construction asyncio.Queue you build the protocol a sentinel per consumer to stop no clone; share the object qsize() only unbounded unless you pass maxsize Verified on both backends: 100 items through a 10-slot stream, consumer exiting cleanly on close.

1. Create a bounded stream and let it push back

send, receive = anyio.create_memory_object_stream(max_buffer_size=10)


async def producer(send_stream) -> None:
    async with send_stream:                            # closes the handle on exit
        for item in items:
            await send_stream.send(item)               # blocks when the buffer is full


async def consumer(receive_stream, results: list) -> None:
    async with receive_stream:
        async for item in receive_stream:              # ends when all senders close
            results.append(await handle(item))

The buffer size is the first positional argument and there is no unbounded default — max_buffer_size=0 means a rendezvous where each send waits for a receive, and math.inf is available but must be asked for explicitly. That is the opposite of asyncio.Queue, where unbounded is what you get unless you remember maxsize, and it is the single best reason to prefer streams for pipelines.

Measured: 100 items through a 10-slot stream with a consumer sleeping 1 ms per item took 0.11 s — the producer was paced by the consumer throughout, which is what back-pressure means.

Verify: with a slow consumer, the producer's progress tracks it rather than racing ahead.

2. Close instead of sending sentinels

This is the difference that removes code. With asyncio.Queue, a pipeline must send one sentinel per consumer and every consumer must recognise it:

for _ in range(worker_count):
    await queue.put(None)                              # one per consumer, count must be right

With streams, exiting the producer's async with closes its handle; when every send handle is closed, the stream is done and every consumer's async for ends:

producer closed its end -> consumer loop exited cleanly with [0, 1, 2, 3, 4]

No sentinel, no count, and no risk of a consumer that outlives the pipeline because it missed its marker. Note the ownership rule that makes this work: whoever will close a handle should own it, which usually means passing send into the producer task and receive into the consumer, rather than keeping both in the parent.

Verify: consumers exit on their own when producers finish, with no sentinel in the code.

Why closing matters more than it looks 5 stages from producer finishes to consumers exit. Why closing matters more than it looks producer finishes async with send the handle closes automatically all senders closed the stream ends async for stops no sentinel needed consumers exit the group completes With a queue you send one sentinel per consumer and hope the count is right.

3. Fan out with clones

clone() produces another handle to the same stream, which is how you get several producers or several consumers without extra machinery:

async with anyio.create_task_group() as tg:
    for i in range(3):
        tg.start_soon(consumer, receive.clone(), results[i])
    await receive.aclose()                             # the parent's handle: close it
    tg.start_soon(producer, send, 30)

Verified: three cloned receivers shared 30 items 10/10/10 — each item goes to exactly one receiver, so this is a work queue, not a broadcast. For broadcast you need one stream per subscriber and a fan-out task, as in fanning out a queue to consumer groups.

The await receive.aclose() on the parent's own handle is essential and easy to miss: the stream ends only when every receive handle is closed, and a forgotten parent handle keeps it open forever, so producers never see the end and the task group never completes.

Verify: the item counts across consumers sum to the number produced, and the group exits.

4. Shed instead of blocking when you need to

send_nowait is the non-blocking path, and it raises rather than growing the buffer:

try:
    send.send_nowait(item)
except anyio.WouldBlock:
    DROPPED.inc()                                      # or return 503

Verified: send_nowait on a full one-slot stream raised anyio.WouldBlock. That is the equivalent of asyncio.QueueFull, and it is the hook for load shedding — a request handler that cannot enqueue should fail fast rather than wait.

statistics() gives the numbers to export alongside it:

stats = send.statistics()
# current_buffer_used=4, max_buffer_size=10, open_send_streams=1, open_receive_streams=1

The handle counts are the diagnostic for the hang above: an open_receive_streams that never reaches zero is the forgotten clone.

Verify: buffer usage and handle counts are exported, and handle counts return to zero at shutdown.

The parts of the API you will use A grid of 5 rows by 2 columns. The parts of the API you will use call does note create_memory_object_stream(n) returns (send, receive) n is the buffer size await send.send(item) blocks when the buffer is full this is the back-pressure send.send_nowait(item) raises WouldBlock if full the shedding path stream.clone() another handle to the same stream for extra producers or consumers await stream.aclose() closes this handle the stream ends when all are closed Each item goes to exactly one receiver: three clones shared 30 items 10/10/10.

5. Map between the two worlds

Porting in either direction is mechanical:

asyncio.Queue memory object stream
asyncio.Queue(maxsize=n) create_memory_object_stream(max_buffer_size=n)
await q.put(x) await send.send(x)
q.put_nowait(x) / QueueFull send.send_nowait(x) / WouldBlock
await q.get() await receive.receive()
sentinel per consumer async with send: — closing ends it
q.qsize() stream.statistics().current_buffer_used
await q.join() / task_done() no equivalent — use a task group

The missing row is worth noting: streams have no join()/task_done() pair. The AnyIO idiom is a task group, which does not complete until every task in it has finished — which is the same guarantee expressed structurally rather than by counting.

In asyncio-only code, the choice is genuinely a matter of taste; asyncio.Queue with an explicit maxsize and a disciplined sentinel protocol does the same job. In code that must run on trio, or in a library, streams are the only portable option.

Verify: a ported pipeline has no sentinels and no unbounded buffers.

Which channel does this pipeline need? A decision on What are you connecting with 3 outcomes. Which channel does this pipeline need? What are you connecting? stages inside one process a memory object stream closing and fan-out included asyncio-only code, simple asyncio.Queue already there, one less dependency separate processes a broker or a table in-process channels do not survive Both in-process options lose everything on restart; that is a durability decision, not a style one.

Verification

An in-process channel is well built when:

  • The buffer is bounded deliberately, with a size you chose.
  • Producers close their handles, and consumers end on close rather than on a sentinel.
  • Every clone is closed, including the parent's original handle.
  • Overflow has a policy: block, or send_nowait and shed.
  • Buffer usage and handle counts are exported.
  • Nothing relies on join() — the task group provides completion.

Pitfalls & edge cases

  • Forgetting to close the parent's handle. The stream never ends and the task group hangs.
  • max_buffer_size=math.inf. Legal, and it reintroduces the unbounded-queue memory problem.
  • Expecting broadcast from clones. Each item goes to exactly one receiver; broadcast needs a stream per subscriber.
  • Sharing one handle across tasks that each close it. Closing is per handle; clone one per task.
  • Blocking in the consumer loop. A slow consumer back-pressures the producer, which is intended — but a blocking call stalls the whole loop.
  • Using streams across processes. They are in-process only; a restart loses everything buffered.

Frequently Asked Questions

What is an AnyIO memory object stream?

A bounded in-process channel with separate send and receive handles. Closing every send handle ends the consumers' async for loops, the buffer size is a required constructor argument, and clone() adds producers or consumers. It is asyncio.Queue with closing and bounding built in.

How is a memory object stream better than asyncio.Queue?

Two things: it is bounded by construction, so back-pressure is the default rather than an option, and it has a real closed state, so consumers exit without sentinel values. Verified, a consumer's async for ended cleanly the moment the producer's handle closed.

Do cloned receivers each get every item?

No — each item goes to exactly one receiver. Three cloned receivers shared 30 items 10/10/10 in testing, making it a work queue rather than a broadcast. For broadcast, create one stream per subscriber and have a fan-out task send to each.

Why does my task group hang when using memory object streams?

Almost always an unclosed handle. The stream ends only when every send or receive handle is closed, so the parent's original handle must be closed after cloning it for the tasks. statistics().open_receive_streams shows the count that is not reaching zero.

What replaces queue.join() and task_done() with streams?

A task group. It does not exit until every task inside it has finished, which is the same completion guarantee expressed structurally instead of by counting outstanding items. There is no join()/task_done() pair on streams.