Coroutines example with python's asyncio module

coroutines_example.py

import asyncio

@asyncio.coroutine
def open_file(name):
    print("opening {}".format(name))
    return open(name)

@asyncio.coroutine
def close_file(file):
    print("closing {}".format(file.name))
    file.close()

@asyncio.coroutine
def read_data(file):
    print("reading {}".format(file.name))
    return file.read()

@asyncio.coroutine
def process_data(filename):
    # I want the result from open_file(filename)
    # untill it's done don't bother calling me
    file = yield from asyncio.async(open_file(filename))
    print('opened {}'.format(filename))

    # I want the result from read_data(file)
    # untill it's done don't bother calling me
    data = yield from asyncio.async(read_data(file))

    print('read {}'.format(filename))

    yield from close_file(file)


@asyncio.coroutine
def main_coro(loop):
    # start our tasks asynchronously in futures
    tasks = [
        asyncio.async(process_data('/etc/passwd')),
        asyncio.async(process_data('/etc/group')),
        asyncio.async(process_data('/var/log/Xorg.0.log')),
    ]

    # untill all futures are done
    while not all(task.done() for task in tasks):
        # take a short nap
        yield from asyncio.sleep(0.01)

    # we're done, so stop the event loop
    loop.stop()


# get event loop
loop = asyncio.get_event_loop()

# schedule the main coroutine to start as soon as possible
loop.call_soon(asyncio.async, main_coro(loop))
# run untill explicitly stopped
loop.run_forever()

# instead of the above two lines we can also run
# loop.run_until_complete(main_coro()) and remove
# the loop parameter for main_coro and the call
# to loop.stop() at the end of it

loop.close()

This does not work. The IO operations are blocking IO operations. https://groups.google.com/forum/#!topic/python-tulip/MvpkQeetWZA Running with PYTHONASYNCIODEBUG=1 python3 showed me that the read_data coroutine was taking 0.1s to run as was blocking.

Hi @kunev, nice Gist idea! but as @drt24 comments there are blocking IO operations that block the loop. You can use loop.run_in_executor() to detach the blocking operations from the loop thread. Something like:

@asyncio.coroutine
def read_data(file):
  print("reading {}".format(file.name))
  loop = asyncio.get_event_loop()
  data = yield from loop.run_in_executor(None, file.read) 
  return data

And you could replace lines such as

 data = yield from asyncio.async(read_data(file))

by

 data = yield from read_data(file)

You could elaborate a bit more the gist by reading the file line by line:

@asyncio.coroutine
def read_line(file, loop):
  line = yield from loop.run_in_executor(None, file.readline)
  if len(line) == 0:
    return None
  return line.decode("utf-8").rstrip("\n")

  while True:
      line = yield from read_line(file, loop)
      if line is None:
        break
      # do something with line

I've created an AsyncIteratorExecutor class, which can be a handy way to execute I/O via a thread. I release this to the public domain:

import asyncio


class AsyncIteratorExecutor:
    """
    Converts a regular iterator into an asynchronous
    iterator, by executing the iterator in a thread.
    """
    def __init__(self, iterator, loop=None, executor=None):
        self.__iterator = iterator
        self.__loop = loop or asyncio.get_event_loop()
        self.__executor = executor

    def __aiter__(self):
        return self

    async def __anext__(self):
        value = await self.__loop.run_in_executor(
            self.__executor, next, self.__iterator, self)
        if value is self:
            raise StopAsyncIteration
        return value


async def cat_file_async(filename):
    with open(filename, 'rt') as f:
        async for line in AsyncIteratorExecutor(f):
            print(line.rstrip())


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(
            cat_file_async('/path/of/file.txt'))
    finally:
        loop.close()