Summary

Adds asyncio.open_channel(max_buffer_size) — a split send/receive channel primitive that improves on asyncio.Queue with:

• Bounded by default — backpressure via max_buffer_size (supports 0 for rendezvous, math.inf for unbounded)
• Split ownership — separate SendChannel / ReceiveChannel endpoints enforce correct resource management
• Clone + ref-counting — clone() enables multi-producer / multi-consumer (fan-in / fan-out) with clean shutdown
• Async iteration — async for item in receive_channel
• Deterministic close signaling — EndOfChannel when all senders close, BrokenResourceError when all receivers close
• Context manager support — both sync (with) and async (async with)

Part of the proposal to adopt proven anyio/Trio patterns natively into asyncio (Tier 2, item 2.2).

API

send, recv = asyncio.open_channel(max_buffer_size=10)

# Producer
await send.send(item)
send.close()

# Consumer
async for item in recv:
    process(item)
# StopAsyncIteration raised automatically when all senders close

New public names

• asyncio.open_channel() — factory returning (SendChannel, ReceiveChannel)
• asyncio.SendChannel — send endpoint with send(), send_nowait(), clone(), close()
• asyncio.ReceiveChannel — receive endpoint with receive(), receive_nowait(), clone(), async iteration
• asyncio.ChannelStatistics — frozen dataclass for introspection
• asyncio.EndOfChannel — raised when all senders are closed
• asyncio.ClosedResourceError — raised when using a closed endpoint
• asyncio.BrokenResourceError — raised when sending with no open receivers
• asyncio.WouldBlock — raised by _nowait methods when operation would block

Files

Test plan

• ./python -m test test_asyncio.test_channels -v — all 40 tests pass
• Verify no regressions in existing asyncio tests
• Review for edge cases in cancellation and close ordering

🤖 Generated with Claude Code