Skip to content

Handling Drain and Write Backpressure in Asyncio

The memory graph climbs steadily through the afternoon and the process is eventually OOM-killed. There is no leak in the usual sense — no growing dict, no retained objects — and a heap dump shows the memory is in bytes objects held by transport send buffers. Somewhere, a producer is calling writer.write() faster than a peer is reading, and because write() never blocks and never refuses, the difference accumulates inside your process until the kernel intervenes.

await writer.drain() is the fix, and it is one line. Understanding what it waits for is what lets you tune it: drain does not flush the buffer, does not wait for the peer to acknowledge anything, and does nothing at all until the buffer has crossed a threshold you can configure. This guide covers the water marks behind it, where drain belongs in a broadcast loop (which is not where most code puts it), per-peer send queues for when one slow consumer must not throttle everyone, and the policy question that has no default answer: what to do about a peer that simply will not read.

Prerequisites

The write buffer against its water marks Three lanes relating buffer size, write calls, and drain waiting. The write buffer against its water marks buffer size below low mark climbing above high mark draining write() accepts everything, always drain() returns instantly waits here time, one slow peer drain() only waits after the high mark, and only until the low mark.

1. Understand what drain actually waits for

The transport holds an output buffer with a high-water mark (64 KiB by default) and a low-water mark (a quarter of it). write() appends and returns. When the buffer crosses the high mark the transport enters a paused state; drain() returns immediately unless that state is set, and otherwise waits until the buffer falls back below the low mark.

import asyncio


async def send(writer: asyncio.StreamWriter, payload: bytes) -> None:
    writer.write(payload)          # never blocks, never fails, never refuses
    await writer.drain()           # returns instantly unless the buffer is over
                                   # the high-water mark; then waits for the low mark

Two implications people find surprising. drain() is nearly free in the healthy case — it is a state check, not a flush — so there is no performance argument for skipping it. And when it does wait, it does not wait for your bytes to reach the peer, only for the buffer to shrink; delivery confirmation is not something TCP offers your application. Verify: write 1 KiB in a loop to a fast peer and time the drain() calls; they should be sub-microsecond until the peer slows down.

2. Tune the water marks for your traffic shape

The defaults suit request/response traffic. Streaming, broadcasting, or very high connection counts want different numbers.

transport = writer.transport
transport.set_write_buffer_limits(high=256 * 1024, low=64 * 1024)

# Small frames, many connections (10k peers): keep per-peer memory tiny
transport.set_write_buffer_limits(high=16 * 1024, low=4 * 1024)

# Large sequential transfers to few peers: fewer wakeups, more in flight
transport.set_write_buffer_limits(high=1024 * 1024, low=256 * 1024)

The high mark is the per-connection memory you are willing to lend a slow peer; multiply by peak connections for the real exposure — 10,000 peers at the 64 KiB default is 640 MB of potential buffer before anything is "wrong". The low mark controls churn: too close to the high mark and you pause/resume constantly, too far and draining becomes bursty. A quarter of the high mark is a reasonable default. Verify: set a small high mark, write a large payload to a stalled peer, and confirm drain() blocks after roughly that many bytes.

3. Put drain in the right place in a broadcast loop

This is the single most common backpressure bug in async servers: draining inside the fan-out loop makes the slowest subscriber set the pace for everyone.

# WRONG: one slow peer throttles the entire broadcast
async def broadcast_serial(peers: list[asyncio.StreamWriter], msg: bytes) -> None:
    for writer in peers:
        writer.write(msg)
        await writer.drain()          # blocks the loop for every remaining peer


# BETTER: write to all, then drain concurrently with a deadline per peer
async def broadcast(peers: list[asyncio.StreamWriter], msg: bytes) -> None:
    for writer in peers:
        writer.write(msg)
    async def flush(w: asyncio.StreamWriter) -> None:
        try:
            async with asyncio.timeout(5):
                await w.drain()
        except (TimeoutError, ConnectionResetError):
            w.transport.abort()       # this peer cannot keep up: drop it
    await asyncio.gather(*(flush(w) for w in peers), return_exceptions=True)

The second form still bounds memory — each peer's transport enforces its own water marks — but a stalled subscriber only delays itself, and the timeout converts "slow forever" into a disconnect. For a broadcast that must never block at all, step 4's per-peer queue is the next step up. Verify: with one deliberately stalled subscriber among fifty, broadcast latency for the other 49 should stay flat rather than tracking the stalled one.

Where drain goes in a fan-out Two columns contrasting serial drain with concurrent drain in a broadcast. Where drain goes in a fan-out drain inside the loop slowest peer sets the pace Peer 3 stalls Peers 4-50 wait behind it Broadcast latency tracks the worst Looks like a server problem write all, drain concurrently contained All writes issued first Each drain has its own deadline Stalled peer is aborted Healthy peers unaffected Same memory bound either way — very different blast radius.

4. Give each peer a bounded send queue

When producers must not wait on any consumer, decouple them: one bounded queue and one writer task per peer, with an explicit policy when the queue fills.

import asyncio
import logging

log = logging.getLogger("send")


class PeerSender:
    def __init__(self, writer: asyncio.StreamWriter, depth: int = 100) -> None:
        self.writer = writer
        self.queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=depth)
        self.dropped = 0
        self._task = asyncio.create_task(self._run(), name="peer.sender")

    def offer(self, payload: bytes) -> bool:
        """Non-blocking: returns False when the peer is too far behind."""
        try:
            self.queue.put_nowait(payload)
            return True
        except asyncio.QueueFull:
            self.dropped += 1
            return False                       # policy: drop (see step 5)

    async def _run(self) -> None:
        try:
            while True:
                payload = await self.queue.get()
                self.writer.write(payload)
                await self.writer.drain()      # backpressure stops HERE, at the queue
        except (ConnectionResetError, BrokenPipeError):
            pass
        finally:
            self.writer.close()

    async def aclose(self) -> None:
        self._task.cancel()
        try:
            await self._task
        except asyncio.CancelledError:
            pass

The queue's maxsize is now the real per-peer memory bound, and it is explicit rather than implicit in a transport default. The producer never awaits a consumer: it calls offer() and gets an immediate answer. Verify: stall one peer and confirm its queue fills to exactly depth, its drop counter rises, and the producer's throughput is unchanged.

5. Choose a slow-consumer policy deliberately

A full queue is a decision point, and every option is wrong for some workload — which is why the default (grow forever) is wrong for all of them.

# 1. Drop newest — telemetry, metrics, live positions: freshness is cheap to lose
if not sender.offer(payload):
    METRICS.dropped.inc()

# 2. Drop oldest — market data, sensor feeds: the newest value supersedes
if sender.queue.full():
    sender.queue.get_nowait()                  # discard the stale head
sender.queue.put_nowait(payload)

# 3. Disconnect — chat, gameplay, anything order-dependent
if not sender.offer(payload):
    log.warning("peer too slow, disconnecting")
    await sender.aclose()

# 4. Block the producer — durable pipelines where loss is unacceptable
await sender.queue.put(payload)                # propagates backpressure upstream

Pick per stream, not per server: a service may drop position updates while blocking on billing events. The one policy to avoid is silent unbounded buffering, because it converts a peer's problem into your outage. Verify: each policy behaves as documented under a deliberately stalled peer — drops counted, or the peer disconnected, or the producer slowed.

6. Measure drain wait and queue depth

Backpressure is invisible unless you export it.

import time

DRAIN_WAIT = Histogram("stream_drain_seconds", "Time inside writer.drain()")


async def send_measured(writer: asyncio.StreamWriter, payload: bytes) -> None:
    writer.write(payload)
    start = time.monotonic()
    await writer.drain()
    DRAIN_WAIT.observe(time.monotonic() - start)

Three signals tell the whole story: drain wait (p99 above a few milliseconds means peers are falling behind), queue depth per peer (pinned at maxsize identifies exactly which ones), and drops or disconnects per minute (the policy in action). Together they distinguish "one bad client" from "we are over capacity", which need opposite responses. Verify: under normal load p99 drain wait is near zero; when you stall a peer, its queue depth pins and drain wait rises only for that connection.

The queue is full — now what? A decision diamond about message loss tolerance with three policy cards. The queue is full — now what? Can this stream lose messages? yes, freshness matters drop oldest sensor and market feeds yes, volume matters drop newest telemetry and metrics no, order matters disconnect the peer chat, gameplay, state sync The one option that is always wrong is buffering without a bound.

Verification

  • Memory is bounded by configuration. With a peer that never reads, process memory settles at roughly connections × (queue depth × message size + high-water mark), not unbounded growth.
  • One slow peer is contained. Broadcast latency to healthy peers stays flat while a stalled peer's queue fills and its policy fires.
  • The policy is observable. Drops, disconnects, and drain wait are all counted, so "we shed 400 telemetry frames" is a metric rather than a mystery.

Pitfalls and edge cases

  • Skipping drain() because it "never blocks". That is precisely the failure: write() accepting everything is what fills memory.
  • Draining inside a broadcast loop. The slowest subscriber sets the send rate for all subscribers.
  • Two tasks calling drain() on one writer. Concurrent drains on the same stream raise RuntimeError in some versions; serialise writes per connection with a lock or a single writer task.
  • Water marks set per connection but reasoned about per process. Multiply the high mark by peak connections before calling a value "small".
  • Unbounded per-peer queues. Moving the buffer from the transport into an asyncio.Queue(maxsize=0) changes nothing except where the memory shows up.
  • Ignoring ConnectionResetError from drain(). A peer that vanished surfaces the error here; treat it as a normal disconnect path, not an unexpected exception.
  • Using drain() as delivery confirmation. It reports buffer state, not receipt. Application-level acknowledgement is the only proof the peer got it.

Frequently Asked Questions

What does await writer.drain() actually wait for in asyncio?

It waits for the transport's output buffer to fall back below its low-water mark, but only if the buffer previously exceeded the high-water mark; otherwise it returns immediately. It does not flush the buffer, does not force a syscall, and does not confirm the peer received anything — it is purely the flow-control handshake that keeps your producer in step with the socket.

What happens if I never call drain() after write()?

Nothing, until a peer reads more slowly than you write. write() always accepts data and appends it to an in-process buffer, so the difference between your write rate and the peer's read rate accumulates as memory until the process is killed. drain() is what converts that difference into producer backpressure.

How do I set the write buffer high and low water marks?

Call transport.set_write_buffer_limits(high=..., low=...) on the writer's transport, typically right after the connection is made. Defaults are 64 KiB high and a quarter of that low. Lower both for servers holding many connections, raise them for bulk transfers to a few peers, and always multiply the high mark by peak connections to see the memory you have committed.

How should a server handle a client that stops reading?

Decide a policy per stream: drop the newest messages (telemetry), drop the oldest (superseded values), disconnect the peer (order-dependent streams), or block the producer (durable pipelines). Implement it with a bounded per-peer queue and a single writer task, so the choice is explicit and countable rather than an unbounded buffer that fails as an out-of-memory kill.