Trying the New Tornado — A Simple UDP Server

Tornado was the framework I used when learning Python — groundbreaking for using epoll back when the big all-inclusive Django reigned (gevent existed too, of course). Two things I like about Tornado: first, the colorful terminal logs; second, the autoreload feature. These two made development feel like a pleasure.

The pity was that my understanding of async and epoll was insufficient at the time, and I completely didn’t understand coroutines — I only focused on those two surface things. Afterwards I didn’t use it for a long time. A few days ago I was writing another simple UDP server, so I tried it out — unexpectedly Tornado had already reached 6.0. Reading the docs I discovered that since 5.0, Tornado has integrated asyncio — meaning libraries using asyncio can already be used with Tornado.

My Requirements

Actually I only had two requirements:

1. The server must use the UDP protocol, because the client only supports UDP.
2. After handling a request, the server needs to use Redis; I want this part to be asynchronous, with as little impact on program performance as possible.

Some Thoughts

I’m no expert either; writing a program from scratch is quite hard. I need to find a demo to borrow from, then modify it slowly until the program meets my expected function. This should be one of the fastest ways.

The rough process was like this:

1. First search the Tornado source repository — I found tcpserver and httpserver related code, but no udpserver related code. What could I do? Quite despairing.
2. Luckily, when searching Google for tornado udp server, the first result appeared: click to view the gist code. Although it’s 5-year-old code, having something beats having nothing — a quick read for reference is good too.

The Main Logic in the Code

In the gist code above, there are two main logic points:

1. Create a new udp socket and listen on the corresponding port.
2. To use Tornado’s IOLoop, the socket must be registered into the io_loop: io_loop.add_handler(sock.fileno(), accept_handler, IOLoop.READ) This way, when a new request arrives, accept_handler is called to process it.

The UDPServer class in the program is just icing on the cake, but it makes your code more extensible.

Looking at It with the New Tornado

By “new version” here I mean Tornado after asyncio.get_event_loop() is used, where you can customize with the async and await keywords.

I took the chance to also read Tornado’s tcpserver and IOLoop. At the code level, the difference between udp and tcp is that different parameters are used at creation; then tcp must accept before transmission can proceed, while UDP doesn’t — only plain recv and send.

In io_loop.py there is actually the following piece of code (I added comments; best to start from the main function and read upwards):

import errno
import functools
import socket

import tornado.ioloop
from tornado.iostream import IOStream

# This part of the code is already handled by asyncio's io_loop; those interested can also look at IOStream's functions.
# But I think I've achieved my goal: after getting data, how to hand control to asyncio for processing.
# You can see the method is by means of the `io_loop.spawn_callback` below
async def handle_connection(connection, address):
    stream = IOStream(connection)
    message = await stream.read_until_close()
    print("message from client:", message.decode().strip())

def connection_ready(sock, fd, events):
    while True:
        try:
            connection, address = sock.accept()
        except socket.error as e:
            if e.args[0] not in (errno.EWOULDBLOCK, errno.EAGAIN):
                raise
            return
        connection.setblocking(0)
        io_loop = tornado.ioloop.IOLoop.current()
        # This is the key: after receiving a connection request, don't call handle_connection directly (that wouldn't be async);
        # instead put the request into the io_loop and let the io_loop schedule this code.
        io_loop.spawn_callback(handle_connection, connection, address)

if __name__ == '__main__':
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.setblocking(0)
    sock.bind(("", 8888))
    sock.listen(128)

    io_loop = tornado.ioloop.IOLoop.current()
    callback = functools.partial(connection_ready, sock)

    # Add the callback function, similar to the earlier UDP server
    io_loop.add_handler(sock.fileno(), callback, io_loop.READ)
    io_loop.start()

This is a simple TCP service test program; seeing it, I knew the udp code could be modified accordingly.

How to Modify the Earlier UDP server

In the comments above I already said: after receiving a connection, use io_loop.spawn_callback to hand control to asyncio. For a UDP server, as soon as data is received — i.e. a connection is received — control should be handed over. The program should be changed to look like this (this function uses a closure, so functools.partial isn’t needed).

def add_accept_handler(sock, callback, io_loop=None):
    if io_loop is None:
        io_loop = tornado.ioloop.IOLoop.current()

    def accept_handler(fd, events):
        while True:
            try:
                data, address = sock.recvfrom(2500)
            except socket.error as e:
                if e.args[0] in (errno.EWOULDBLOCK, errno.EAGAIN):
                    return
                raise
            io_loop.spawn_callback(callback, sock, address, data)
    io_loop.add_handler(sock.fileno(), accept_handler, IOLoop.READ)


## In the UDPServer class, add_accept_handler is called like this
# add_accept_handler(sock, self._on_recive, io_loop=self.io_loop)
# So just change `_on_recive` into an async-form function; on_receive is as follows:

class UDPServer():
    # ...
    async def _on_recive(self, sock, address, data):
        # some process and generate reply
        # respond to the udp request
        # sock.sendto(reply, address)
        await redis_work(...)
    # ...

Here the udp request needs a response as soon as possible, while the redis storage can be slowed down, so the await operation is placed last — asyncio waits for redis’s return, which won’t affect the current user’s experience; scheduling is fully handed over to asyncio.

Using aioredis and Existing Problems

hiredis is not an asyncio-supporting library; while searching I found aioredis, so I used aioredis. Later, while writing this blog, I saw aredis, written by a fellow Chinese developer — from the Github homepage apparently some big shot at Bilibili; the library is also under maintenance. But my program was already finished, and I didn’t want to switch. I hope everyone tries it when using these, and supports domestic Python libraries too.

The aioredis homepage has a simple usage example that looks directly usable.

import asyncio
import aioredis

loop = asyncio.get_event_loop()

async def go():
    conn = await aioredis.create_connection(
        'redis://localhost', loop=loop)
    await conn.execute('set', 'my-key', 'value')
    val = await conn.execute('get', 'my-key')
    print(val)
    conn.close()
    await conn.wait_closed()
loop.run_until_complete(go())

I hit some pitfalls in use — namely the line loop = asyncio.get_event_loop(). Pursuing formal uniformity in the program, I used loop = IOLoop.current() to get the current ioloop in Tornado — which is actually obtained from asyncio too — but the following error occurred:

AttributeError: 'AsyncIOMainLoop' object has no attribute 'create_future'

Later I stopped fussing about it; since everything uses asyncio.get_event_loop anyway, I just wrote it like the example in the program. From a developer’s perspective, I believe Tornado will soon consider merging the issue, and this error may not appear anymore.

Summary

Looking back at this program, the logic itself isn’t complex, but to combine Tornado with UDP request handling, you need to read some source code to be able to operate it.

Of course, native asyncio can also handle UDP requests; for details see the Gist. My choice of Tornado was really largely for the beautiful terminal logs, plus simple code and strong extensibility. Thanks for reading.