Asyncio streams - what is guaranteed?

Async-SIG
1

Imagine a simple TCP chat server such as this (cribbed from the docs example of an echo server and tweaked):

import asyncio

writers = set()
async def handle_echo(reader, writer):
  try:
    writers.add(writer)
    # Identify users by port number for simplicity
    addr = str(writer.get_extra_info('peername')[1])
    while True:
        message = (await reader.read(100)).decode().strip()
        if message == "quit": break
        for w in writers:
            w.write((addr + ": " + message + "\n").encode())
            await w.drain()

    print("Close the connection")
    writer.close()
    await writer.wait_closed()
  finally:
    writers.discard(writer)

async def main():
    server = await asyncio.start_server(
        handle_echo, '127.0.0.1', 8888)

    addrs = ', '.join(str(sock.getsockname()) for sock in server.sockets)
    print(f'Serving on {addrs}')

    async with server:
        await server.serve_forever()

asyncio.run(main())

Connect to it using a telnet client, you can chat, everyone sees everyone’s messages. Very simple. And it works, at least in simple cases. What I’m interested in is what is actually guaranteed, and what simply “happens to work” (especially with a toy example like this).

  1. Are individual write() calls atomic or could they potentially be interleaved?
  2. Suppose client #33603 is lagging badly. The server fills up its buffer, and drain() blocks. Then another client sends a message. It then also wants to send that to #33603, which means that it will call drain() on that socket reentrantly. Is it safe to do that? What is supposed to happen?
  3. Omitting the drain() call seems to work. The docs for write() say that it “should” be used alongside drain(). What guarantees become weakened when not explicitly draining?

I suspect that, for my actual use cases, none of these will be immediate problems; but I want to deploy something and then be able to forget about it for years. So I want to follow best practices… which I’m a bit confused about!

1 Like
2

Calling write only places data into a buffer.

Each drain() call creates its own Future object internally.

If you do not await the drain method before writing again, the buffer may grow indefinitely. Multiple small write() calls may not cause drain() to block.

1 Like
3

But if I comment out the drain call, the content DOES get sent. So it must, at least in some circumstances, actually write it.

So I guess the question is: How do I ask for something to be written, actually written and not just put into a buffer, but without blocking if the buffer’s full? Maybe that’s write() without drain(), but the docs recommend against that. Ideally, I’d like to be able to specify a moderately large buffer, and then say “if the buffer fills, throw”.

1 Like
4

You mean sent to the client. We do not know when it is actually written to the wire. The kernel does not notify the application of physical transmission. send() only means the data has entered the kernel buffer, not that it has been sent or received. For confirmation, we need an application-level protocol.

That cannot be done using the high-level API, as far as I know. You have to go to a lower level and use set_write_buffer_limits.

Another option is to manually track how much data you write in the client and then await drain when needed. That would feel like reinventing a low-level API.

1 Like
5

Not sure what you mean by “client” and “wire” here, but when I say that the content WAS sent even when I didn’t drain() it, I mean that the other end of the TCP socket did receive the data. IOW I could simply omit the drain() call and not notice any problem. Which isn’t what the docs imply, so I’m trying to figure out what the issue is.

Fair enough. Though I don’t really need to know “that was sent”, and much more interesting to me would be “hey, that couldn’t be sent, the other end might be gone”.

Hmm. I tried writer.transport.set_write_buffer_limits(1, 1) and writer.transport.set_write_buffer_limits(0), and both ways, data still got sent perfectly normally. So I’m not sure what exactly would block. And this is without draining.

How would I know when it’s “needed” though? I tried printing out w.transport.get_write_buffer_size() after each non-drained write, and it was always zero.

It’s looking like, despite the warnings in the docs, Python’s doing the simple and obvious thing of “write as much as can be written, right now, and buffer the rest”. And if the docs said that drain() was “wait until all queued data has been written”, then it’d be easy - I would simply never call it. So I guess the only real question is, what do the docs mean when they say that the write method “should be used along with the drain() method”? Does “should” imply RFC 2119 semantics, where violating it requires that you understand the full implications? Or is it a more colloquial “hey, you should probably do this”, which is somewhat weaker?

1 Like
6

Note that the drain() documentation explains in detail what it does, so “should” feels more like “if you want to engage flow control.”

You are right, “should” makes it sound like it is preventing undefined behavior. Mentioning flow control would be clearer.

7

Yeah, IF that’s how it is. So this could be as simple as a docs change, but it would be a change of guarantee.

8

In practical terms, in answer to the original question, I don’t think all that much is guaranteed. Quite a bit of the internal behavior is currently implementation specifics, not documented properties of the design.

Implementation detail, but I can’t imagine this one ever being changed. FIFO, sync call. May require more thought if mixing threading with async. This is probably worth turning into an actual guarantee as documented behavior.