andrelouiscaron / a-tale-of-event-loops Goto Github PK

I have been reading various writings on async/await in Python and somehow stumbled onto your write up. It is a nice demo of the basics of an async event loop in Python that I found very helpful. I noticed a couple issues with your final example that I wanted to point out.

The await join(child) statement does not actually work the way one would want it to. At least, I think the idea is that join(child) should wait until the child is done and then allow the calling function to continue. Instead if child is cancelled, join(child) causes the calling function to be suspended forever. A simple way to see this is to add print("Joined") after await join(child) in main(). When you run the example, "Joined" is not printed.

The await cancel(child) statement schedules child to be run again with an exception thrown to it and schedules main() to be continued after that (where it will get to the join() call). On the next tick, CancelledError is thrown to child causing its watch entry to be popped and it to be discarded from tree. So child is not rescheduled in the task queue. Then main() calls join(), adds child back into watch but does not reschedule itself. At this point, there are no tasks left in the queue at the end of the scheduler's tick loop and so run_until_complete() returns without the join() ever resolving.

I think this could be addressed by changing

for task, data in queue:

to

while queue:
    task,data = queue.pop(0)

and then adding

queue = [i for i in queue if i[0] != task]
tasks = [i for i in tasks if i[0] != task]
for key, val in watch.items():
    watch[key] = [i for i in val if val != task]

to the except (StopIteration, CancelledError): block, so that when a task ends or is cancelled all other scheduled calls to it are dropped as well.

I also see a second problem with cancel(). If you used a task that did anything other than suspend(), it would be proceeding through its statements, hitting various yield points. On the tick after its parent called cancel() on it, it likely would have already run its next bit of code, hit its next yield point, and added itself back to the tasks list for the next tick. Then it would be thrown CancelledError which would end the coroutine. On the next tick, the event loop would still try to run the task since it had queued itself before getting CancelledError. This leads to the error message RuntimeError: cannot reuse already awaited coroutine. This can be seen by replacing suspend() with something like:

@coroutine
def suspend():
    while True:
        yield ("spinning",)

I think this could be addressed by adding from weakref import WeakSet to the set of imports, defining completed = WeakSet() where watch is defined, adding completed.add(task) just below tree.discard(task) in the except (StopIteration, CancelledError): block, then changing the code that handles join events to:

elif data and data[0] == 'join':
    if data[1] not in completed:
        watch[data[1]].append(task)
    else:
        tasks.append((task, None))

My suggestion uses a WeakSet here because I don't know of a way to check if a types.coroutine coroutine has completed, so this suggestion just tracks when they are completed by hand.