AsyncIO for the working PyGame programmer (part I)

As I explained in the introductory post, asyncio is not a replacement for threads, and it definitely isn't a magical way around the Global Interpreter Lock (GIL) in Python. The threading and multiprocessing modules still have their place in Python 3.6.

It's not called "async_compute" for a reason. Like threading, asyncio can't help you utilise multiple cores to do multiple computation-heavy things at the same time. Like "node.js" the asyncio module lets you run multiple i/o-bound tasks, or "green threads" if you like that terminology, inside a single OS thread.

That is especially useful if you want to write a server that handles tens of thousands of client connections at once, like an IRC server. Most of the time, most IRC users are just idling, so running a dedicated thread for each client would be really wasteful.

It would use up a lot of RAM, put strain on the scheduler of your OS, and take time to context-switch. asyncio is a way to run lots of tasks that are not doing much anyway, except waiting for input. So don't throw out threading and multiprocessing just yet!

There are two major reasons: The first is that you might want to use a library that is based on asyncio in your game. If you just call that library with loop.run_until_complete() from your game loop, you lose all the benefits of asynchronous, event-based code. You have to understand coroutines and event loops to use asyncio effectively.

The other reason is responsive game feel. Event-based i/o and coroutines let you write code that looks a lot like blocking code, but actually executes only when input is available. You can write long-running functions that read files, wait for network input, or upload savegames to a server over a slow connection without blocking, which would be bad for your frame rate, and without cluttering your game loop with i/o logic.

Sure, you could put the logic for piecemeal uploading for savegame files to cloud services into your game loop, and stream a couple of kilobytes between every frame, but this kind of thing is what asyncio was made for.

This is a simple PyGame-based game. When you press the up-arrow on your keyboard, the black box goes up. If you press it while the box is in the air, it will increase the speed, making it go up faster or fall slower. If the box collides with the top of the window, it will bounce off.

If you keep the box in the air for ten seconds, it prints "ACHIEVEMENT UNLOCKED" to stdout.

import pygame

pygame.init()

blob_yposition=30
blob_yspeed=0
achievement=False

gravity=1

screen_size=640,480
screen=pygame.display.set_mode(screen_size)

clock=pygame.time.Clock()
running=True
flying_frames=0
best=0
color=(50,50,50)
font=pygame.font.SysFont("Helvetica Neue,Helvetica,Ubuntu Sans,Bitstream Vera Sans,DejaVu Sans,Latin Modern Sans,Liberation Sans,Nimbus Sans L,Noto Sans,Calibri,Futura,Beteckna,Arial", 16)

while running:
    clock.tick(30)

    events=pygame.event.get()
    for e in events:
        if e.type==pygame.QUIT:
            running=False
        if e.type==pygame.KEYDOWN and e.key==pygame.K_UP:
            blob_yspeed+=10

    # ...
    # move sprites around, collision detection, etc

    blob_yposition+=blob_yspeed
    blob_yspeed-=gravity

    if blob_yposition<=30:
        blob_yspeed=0
        blob_yposition=30
        flying_frames=0
    else:
        flying_frames+=1
        if flying_frames>best:
            best=flying_frames
        if not achievement and best>300:
            # 10 seconds
            print("ACHIEVEMENT UNLOCKED")
            achievement=True
            color=(100,0,0)

    if blob_yposition>480:
        blob_yposition=480
        blob_yspeed=-1*abs(blob_yspeed)

    # ...
    # draw 

    screen.fill((255,255,255))

    pygame.draw.rect(screen,color,
                        pygame.Rect(screen_size[0]/2,
                                    screen_size[1]-blob_yposition,
                                    18,25))
    fps=clock.get_fps()
    message=f"current:{flying_frames//30},   best:{best//30},   fps:{fps}"
    surf=font.render(message, True, (0,0,0))
    screen.blit(surf,(0,0)) 
    pygame.display.update()
print("Thank you for playing!")

What if instead of printing to stdout, we want to send this achievement to Steam? What if that takes some time? We can simulate this delay by adding in pygame.time.wait(500) before we print the message. Try doing that, play until you get the achievement, and feel the frames drop!

Python 3.5 has introduced syntax for coroutines based on the keywords async and await. To define a coroutine function, you have to write async def instead of def, and to call a coroutine, you have to await it.

import asyncio

async def short_coroutine():
    print("ALPHA")
    await asyncio.sleep(0.1)
    print("BRAVO")
    await asyncio.sleep(0.1)
    print("CHARLIE")
    return None

If you call a coroutine function, like short_coroutine(), you will get a coroutine object back. You can try this out in the interactive Python shell. Calling short_coroutine() will not return the return value None or cause any side effects like printing.

This is similar to generators which we already know from Python 2. Calling a generator function like enumerate will return a generator object. Internally, coroutines in Python 3 are based on generators, but you can't iterate over them. It's just a nice analogy to understand why short_coroutine() doesn't print anything right away.

That is very important to understand, because if you want to call another coroutine from your coroutine, you must await it. If you just call asyncio.sleep(0.1) in your coroutine without awaiting it, you get a coroutine object and then you don't do anything with it. The other way around, calling a non-coroutine function from a coroutine, it works as you would expect it.

If you are not already inside a coroutine, you can't use await. You have to execute coroutines as tasks in an event loop.

loop = asyncio.get_event_loop()

loop.run_until_complete(short_coroutine())

The event loop is a scheduler-like object that keeps track of tasks based on coroutines, and runs them whenever there is work available. Whenever any of your coroutines awaits something, the control flow goes back to the event loop, and the event loop can decide which coroutine to run next. If one coroutine is currently sleeping or waiting for input that has not arrived yet, the event loop will either wait, or run another coroutine until the next await.

Here are some more coroutines that call each other, executed all at once with asyncio.gather().

async def long_running_coroutine():
    await short_coroutine()
    print("ONE")
    await asyncio.sleep(10)
    print("TWO")
    await asyncio.sleep(10)
    print("THREE")
    await asyncio.sleep(10)
    print("FOUR")

async def third_coroutine():
    print("GROUCHO")
    await asyncio.sleep(0.1)
    print("HARPO")
    await asyncio.sleep(0.1)
    print("ZEPPO")
    await asyncio.sleep(0.1)
    print("KARL")
    await asyncio.sleep(0.1)

async def fourth_coroutine():
    print("ONE FISH")
    await asyncio.sleep(0.1)
    print("TWO FISH")
    await asyncio.sleep(0.1)
    print("RED FISH")
    await asyncio.sleep(0.1)
    print("BLUE FISH")
    await asyncio.sleep(0.1)

loop.run_until_complete(asyncio.gather(
    long_running_coroutine(),
    third_coroutine(),
    fourth_coroutine()))

This time the event loop executes three tasks in parallel, but each of them still runs in sequential order. Our long_running_coroutine() will await short_coroutine(), so "alpha", "bravo" and "charlie" are always printed before "one". We can't make any guarantees between the coroutines though. Maybe third_coroutine() finishes after fourth_coroutine() for some reason. It never happened on my machine, but there are no guarantees that the scheduling of the event loop is deterministic.

This is still running in a single thread with cooperative multi-tasking, so if any coroutine hangs, the whole event loop will hang. You can try this out by inserting while True:pass somewhere in a coroutine. Don't try this in production!

Running tasks/coroutines in an event loop with run_until_complete() can be appropriate for networked servers, but in our games, we don't want to leave the frame rate up to some opaque scheduling logic². We want to stay in control of the game loop. Putting our game in a coroutine and running the input-update-render loop from the event loop is right out.

Calling run_until_complete() from inside our game loop will block until the task is done, so it is as bad as blocking i/o inside the game loop, with the added negative that there are useless await statements peppered throughout our code.

Instead, we can use loop.create_task() to create individual tasks from coroutine objects, and use this small function run_once() which looks like a hack but is the officially endorsed way to do this kind of thing according to the asyncio developers:

task = loop.create_task(long_running_coroutine())
task2 = loop.create_task(fourth_coroutine())

def run_once(loop):
    loop.call_soon(loop.stop)
    loop.run_forever()

while True:
    run_once(loop)
    if input("Press RETURN >")=="exit":
        break
loop.close()

With loop.call_soon() we schedule the loop to run loop.stop() right after the next piece of a task has been executed. So loop.run_forever() will poll for input events, futures or sleep times that are over, and either run a task or not if there are none available, and then call the function scheduled in loop.call_soon(). It looks wonky, but you can rely on this behaviour, because it is the intended way to let a coroutine task stop its own event loop.

Based on this, we can write a game loop that wakes up the event loop once per frame, and runs a task if there is work available.

We are going to put it all together in part II.