Non-blocking I/O with asyncio

Question

I'm trying to write a networked game with Pygame and asyncio, but I can't work out how to avoid hanging on reads. Here is my code for the client:

@asyncio.coroutine
def handle_client():
    print("Connected!")
    reader, writer = yield from asyncio.open_connection('localhost', 8000)
    while True:
        mouse_up = False
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                pygame.quit()
                sys.exit()                
            elif event.type == pygame.MOUSEBUTTONUP:
                mouse_up = True

        if mouse_up:
            print("Writing")
            writer.write(b"Mouse up")
        print("Waiting to read")
        line = yield from reader.read(2**12)
        print(line.decode())

    writer.close()

This hangs on the line line = yield from reader.read(2**12). I previously thought that the point of asyncio was that it was non-blocking, and so if there wasn't any data to read it would just continue executing. I see now that this isn't the case.

How do I integrate the asyncio networking code with the Pygame drawing and event code?

jfs · Accepted Answer · 2014-12-29T11:50:18.513

14

The point of yield from is to switch the execution to the asyncio's event loop and to block the current coroutine until the result is available. To schedule a task without blocking the current coroutine, you could use asyncio.async().

To print read-so-far data without blocking the pygame loop:

@asyncio.coroutine
def read(reader, callback):
    while True:
        data = yield from reader.read(2**12)
        if not data: # EOF
            break
        callback(data)

@asyncio.coroutine
def echo_client():
    reader, ...
    chunks = []
    asyncio.async(read(reader, chunks.append))
    while True:
        pygame.event.pump() # advance pygame event loop
        ...
        if chunks: # print read-so-far data
            print(b''.join(chunks).decode())
            del chunks[:]
        yield from asyncio.sleep(0.016) # advance asyncio loop

There should be no blocking calls inside the while loop.

read() and sleep() coroutines run concurrently in the same thread (obviously you could run other coroutines concurrently too).

edited Dec 29 '14 at 11:50

answered Dec 29 '14 at 00:49

jfs

399,953
195
994
1,670

Thanks. Would I put the `clock.tick()` call at the usual place in the loop, or would I need to write my own asynchrous one to avoid blocking the networking code? – rlms Dec 29 '14 at 15:04
1

@sweeneyrod: the latter. The `sleep()` call above emulates async `tick()` – jfs Dec 29 '14 at 15:43
Wouldn't this cause the pygame code (the event loop, and sleeping) to be called for each client, rather than only once? – rlms Dec 29 '14 at 16:29
@sweeneyrod: the answer is about *"how to avoid hanging on reads."* There should be only one pygame loop. – jfs Dec 29 '14 at 17:34
What is `async`? `asyncio.async()` gives me `SyntaxError: invalid syntax`. – Anna May 22 '20 at 15:21
3

@Anna: it is a Python 3.4 code. `async()` function is deprecated since Python 3.5. You could use `create_task()` or if it is not available, `ensure_future()` instead. https://docs.python.org/3/library/asyncio-task.html#asyncio.create_task – jfs May 22 '20 at 15:43

score 11 · Answer 2 · edited Sep 03 '15 at 17:25

You can "transform" a blocking task into a non-blocking one.

I suggest this: https://docs.python.org/3/library/asyncio-eventloop.html#executor.

I have a function that listens to a twitter feed, function "mention", and I run it in an executor, so if it hangs, it doesn't block the other tasks.

@asyncio.coroutine
def boucle_deux():
#faire attendre la boucle si pas bcp de mots
    while True:
        print("debut du deux")
        value = t.next()
        future2 = loop.run_in_executor(None, mention, "LQNyL2xvt9OQMvje7jryaHkN8",
                                       "IRJX6S17K44t8oiVGCjrj6XCVKqGSX9ClfpGpfC467rajqePGb",
                                       "2693346740-km3Ufby8r9BbYpyzcqwiHhss22h4YkmnPN4LnLM",
                                       "53R8GAAncFJ1aHA1yJe1OICfjqUbqwcMR38wSqvbzsQMB", 23, value)
        response2 = yield from future2
        yield from asyncio.sleep(5)
        print("fin du deux")

asyncio.Task(boucle_deux())

The purpose of the executor is to run slow or blocking operations (the processus is preempted by the kernel's scheduler) in a separate thread. StreamReader.read() never blocks. — Martin Richard, Jan 12 '15 at 20:56

score 0 · Answer 3 · answered Jun 15 '15 at 18:04

well since you are trying to read the value of 'line' right after you call read() you need that value at any cost...

if the coroutine wouldn't stop cause there are no data, you could get an AttributeError on the line.decode() call if 'line' then is None.

one thing you can do is to set a timeout on the blocking call and handle the timeout exception:

...
print("Waiting to read")
try:  # block at most for one second
    line = yield from asyncio.wait_for(reader.read(2**12), 1)
except asyncio.TimeoutError:
    continue
else:
    print(line.decode())
...

Non-blocking I/O with asyncio

3 Answers3

Linked