chaining coroutines in asyncio (and observer pattern)

Question

I'm having trouble wrapping my head around how coroutines are chained together. In a slightly less trivial example than hello world or factorials, I'd like to have a loop which continually watches file modification times, and then prints out the time whenever the file is touched:

#!/usr/bin/env python3
import os
import asyncio

@asyncio.coroutine
def pathmonitor(path):
    modtime = os.path.getmtime(path)
    while True:
        new_time = os.path.getmtime(path)
        if new_time != modtime:
            modtime = new_time
            yield modtime
        yield from asyncio.sleep(1)

@asyncio.coroutine
def printer():
    while True:
        modtime = yield from pathmonitor('/home/users/gnr/tempfile')
        print(modtime)

loop = asyncio.get_event_loop()
loop.run_until_complete(printer())
loop.run_forever()

I would expect this to work - however, when I run it I get a:

RuntimeError: Task got bad yield: 1426449327.2590399

What am i doing wrong here?

UPDATE: see my answer below for an example of the observer pattern (i.e. efficiently allow multiple registrants to get updates when a file gets touched) without using callbacks (you have to use Tasks).

UPDATE2: there is a better fix for this: 3.5's async for (asynchronous iterators): https://www.python.org/dev/peps/pep-0492/

score 6 · Accepted Answer · answered Mar 15 '15 at 22:09

6

I got your code working by using return instead of yield in the chained coroutine, just like the chained coroutines example:

#!/usr/bin/env python3
import os
import asyncio2

@asyncio.coroutine
def pathmonitor(path):
    modtime = os.path.getmtime(path)
    while True:
        new_time = os.path.getmtime(path)
        if new_time != modtime:
            modtime = new_time
            return modtime
        yield from asyncio.sleep(1)


@asyncio.coroutine
def printer():
    while True:
        modtime = yield from pathmonitor('/tmp/foo.txt')
        print(modtime)


loop = asyncio.get_event_loop()
loop.run_until_complete(printer())
loop.run_forever()

Note that printer()'s loop will create a new pathmonitor generator for each iteration. Not sure if this is what you had in mind but it might be a start.

I find the coroutines API and syntax a bit confusing myself. Here's some reading that I have found helpful:

answered Mar 15 '15 at 22:09

André Laszlo

15,169
3
63
81

1

Thanks for the helpful answer- after looking at the docs a bit more I noticed that they state that a coroutine can only return or yield from - a simple yield is not allowed. I wonder why coroutines aren't able to use yield like a normal generator. – gnr Mar 16 '15 at 00:08
Yeah, it's curious. You could rewrite the `pathmonitor` as a regular generator and move the sleep to the `printer`. But I guess you wanted a chained coroutine. – André Laszlo Mar 16 '15 at 00:20
1

Ah that's an interesting idea - I'll play around a bit to see what I can come up with. I have library code which uses callbacks to notify when a file gets touched and I'd like to see how to make it work with coroutines (or without callbacks). I would have a printer function, or a logger function, or a socket function which all might do something differently when the file gets touched. – gnr Mar 16 '15 at 00:31
4

@gnr You don't want `pathmonitor` to be a generator; coroutines and generators are not the same thing, and aren't interchangeable. While generators can potentially use both `yield` (normal case) or `yield from` (if it needs to yield all the values from a subgenerator), `asyncio` coroutines are only designed to use `yield from`, because `yield from` can be used to fetch a value actually *returned* from the sub-generator. a`yield subgenerator()` call will always return a generator object. – dano Mar 16 '15 at 14:59

gnr · Answer 2 · 2015-04-06T13:05:45.043

As others pointed out, my mistake was that I was trying to use a coroutine like a generator. Instead of relying on a generator for iteration, I needed to create multiple coroutines. Also, I needed to use tasks to implement the observer pattern without callbacks since multiple registrants can yield from the same task. My pathmonitor looks something like this:

import os
import asyncio

class PathInfo:

    def __init__(self, path):
        self.path = path
        self.modtime = os.path.getmtime(path)
        self.startTask()

    def startTask(self):
        self.task = asyncio.async(self._checkIfTouched())

    def _checkIfTouched(self):
        while True:
            yield from asyncio.sleep(1)
            newtime = os.path.getmtime(self.path)
            if self.modtime != newtime:
                self.modtime = newtime
                return newtime

class PathMonitor:

    def __init__(self):
        self._info = {}

    @asyncio.coroutine
    def wasTouched(self, path):
        try:
            info = self._info[path]
        except KeyError:
            self._info[path] = info = PathInfo(path)
        if info.task.done():
            info.startTask()
        modtime = yield from info.task
        return modtime

def printer():
    while True:
        modtime = yield from mon.wasTouched('/tmp/myfile')
        print(modtime)

mon = PathMonitor()

loop = asyncio.get_event_loop()
asyncio.async(printer())
loop.run_forever()

chaining coroutines in asyncio (and observer pattern)

2 Answers2

Linked