Async stdio with asyncio

gistfile1.py

import os
import asyncio
import sys

from asyncio.streams import StreamWriter, FlowControlMixin

reader, writer = None, None

@asyncio.coroutine
def stdio(loop=None):
    if loop is None:
        loop = asyncio.get_event_loop()

    reader = asyncio.StreamReader()
    reader_protocol = asyncio.StreamReaderProtocol(reader)

    writer_transport, writer_protocol = yield from loop.connect_write_pipe(FlowControlMixin, os.fdopen(1, 'wb'))
    writer = StreamWriter(writer_transport, writer_protocol, None, loop)

    yield from loop.connect_read_pipe(lambda: reader_protocol, sys.stdin)

    return reader, writer

@asyncio.coroutine
def async_input(message):
    if isinstance(message, str):
        message = message.encode('utf8')

    global reader, writer
    if (reader, writer) == (None, None):
        reader, writer = yield from stdio()

    writer.write(message)
    yield from writer.drain()

    line = yield from reader.readline()
    return line.decode('utf8').replace('\r', '').replace('\n', '')

@asyncio.coroutine
def main():
    name = yield from async_input("What's your name? ")

    print("Hello, {}!".format(name))

asyncio.get_event_loop().run_until_complete(main())

You shouldn't really use sys.stdout in place of os.fdopen(0, 'wb'). connect_write_pipe is going to make the given file object non-blocking, and sys.stdout non-blocking for things that don't expect it is dangerous (e.g. print).

How could this have ever worked? fd 0 is stdin, not stdout. Works fine with os.fdopen(1, 'wb') though, thanks!

@minus7, https://www.mail-archive.com/[email protected]/msg00427.html is worth reading - if your process is attached to a pseudoterminal then 0 and 1 seem to be the same file, which explains why it worked for me.

Python asyncio documentation uses print() from a coroutine, but I must suggest this recipe instead.

Once the size of the data printed reaches a few hundred bytes the BlockingIOError exception is raised:

  File "/Users/jquast/.pyenv/versions/3.5.1/lib/python3.5/asyncio/tasks.py", line 239, in _step
    result = coro.send(None)
  File "/Users/jquast/Code/telnetlib3/telnetlib3/client_shell.py", line 43, in telnet_client_shell
    print(ucs, end='')
BlockingIOError: [Errno 35] write could not complete without blocking

The given writer recipe by @nathan-hoad is required to work around such BlockingIOError. Thank you!

This doesn't seem to work on windows 10, python 3.5:

    data = await async_input(">>> ")
  File "C:\Users\wiese\Documents\Courses\Project\embedded\tools\asyncstdio.py", line 29, in async_input
    reader, writer = await stdio()
  File "C:\Users\wiese\Documents\Courses\Project\embedded\tools\asyncstdio.py", line 16, in stdio
    writer_transport, writer_protocol = await loop.connect_write_pipe(FlowControlMixin, os.fdopen(sys.stdout.fileno(), 'wb'))
  File "C:\Program Files\Python 3.5\lib\asyncio\base_events.py", line 894, in connect_write_pipe
    transport = self._make_write_pipe_transport(pipe, protocol, waiter)
  File "C:\Program Files\Python 3.5\lib\asyncio\base_events.py", line 265, in _make_write_pipe_transport
    raise NotImplementedError
NotImplementedError

FWIW, there are a couple of issues with this, e.g. python/asyncio#147, and as noted earlier, the fact that print() suddenly becomes dangerously non-blocking. I'd recommend implementing a coroutine that reads from stdin asynchronously, and then makes it synchronous again once done with it. Beware that this will still make stdout (and thus print()) temporarily asynchronous for interactive programs, because of how TTYs are implemented on Linux (and possibly in general). See python/asyncio#213 for some discussion on this.

My earlier statement about "you shouldn't really use sys.stdout use os.fdopen(1, 'wb') blah blah blah" was naive and incorrect, they're exactly equivalent, because that's how file descriptors work, which is why print() becomes dangerous with this snippet.

First idea off the top of my head if you want something more robust, sadly I'd recommend you use a thread for stdio interaction, using a mixture of threading.Event, asyncio.Queue() and loop.call_soon_threadsafe() to activate a read in the thread, and then get the data back to the event loop.

FWIW, if you're writing an application that doesn't interact with the user, or use print(), then you could use the code in the gist just fine.

I hadn't really intended for people to use this gist, so I'm not going to update it or make it any more robust, I just want people to be aware of the issues it has. (this is part where I reveal the snippet has an incredibly restrictive license and wield my lawyer, j/k)

In the spirit of making sure there's no room for ambiguity, I'm releasing this code under The Unlicense. Enjoy the free and subtly broken code folks!