Building Redis from scratch with Python Part II — The asynchronous TCP server

This article speaks about my learnings on implementing an asynchronous, single-threaded caching DB with python that is fully compliant with redis based clients .

Adopted from Arpit Bhayani’s Redis Internals playlist available on Youtube. Arpit, if you’re reading this, I’m truly thankful for all the knowledge I’ve gained over the years from your articles and videos.

Introduction

Part I of this article talked about writing a synchronous TCP server using python for our own redis server.

This part will focus on building an async TCP server from scratch, and some of the gotchas I faced while building it.

Before jumping into the details, let’s understand what makes a single threaded server asynchronous — Non Blocking IO using EPOLL

What is non-blocking IO and EPOLL ? Arpit explains it wonderfully in this video

Essentially, when our Redis server will evaluate and process all the commands it receives first, and then allow the clients to connect to it. The moment there are no clients to connect, the server will jump back to handling the remaining commands sent by the connected clients.

All of the above code will occur under a single for loop. That’s the beauty of event loops. They’re simple yet brilliant.

The Async TCP Server

And here’s the actual code:

import select
import socket
import logging
from server.processing import read_command, respond, respondError

logger = logging.getLogger()

BACKLOG = 128
RECV_BUFFER = 4096

def RunAsyncTcpServer(host = '0.0.0.0', port = 7379):

    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listening_socket:
        listening_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        listening_socket.bind((host, int(port)))
        listening_socket.listen(BACKLOG)
        listening_socket.setblocking(False)

        epoll = select.epoll()
        epoll.register(listening_socket.fileno(), select.EPOLLIN)
        fd_to_socket: dict[int, socket.socket] = {}
        conn_clients = 0
        try:
            while True:
                events = epoll.poll(None, 10)
                for event in events:
                    fd, event_mask = event
                    if fd == listening_socket.fileno():
                        while True:
                            try: 
                                client,a = listening_socket.accept()
                                client.setblocking(False)
                                fd_to_socket[client.fileno()] = client
                                epoll.register(client.fileno(), select.EPOLLIN)
                                conn_clients+=1
                            except BlockingIOError as e:
                                logger.error('Encountered BlockingIOError')
                                break 

                        logger.info(f"Accepted connection from {a}, fd={fd}")
                        continue

                    if event_mask & select.EPOLLIN:
                        client = fd_to_socket[fd]
                        data = client.recv(RECV_BUFFER)
                        data, e = read_command(data)
                        if not data: 
                            respondError(e if e else "Some error occured", client)
                        else:
                            respond(data, client)

                    if event_mask & (select.EPOLLHUP | select.EPOLLERR):
                        # Socket disconnected - Perform cleanup.
                        epoll.unregister(fd)
                        fd_to_socket[fd].close()
                        del fd_to_socket[fd]
                        conn_clients-=1
                        continue
        finally:
            epoll.close()
            listening_socket.close()

Here’s the complete PR: https://github.com/priyank-R/python-redis/pull/2

Look at the code under branch async_tcp if you want to walk through the code.

Priyank Rupareliya is a Senior Software Engineer focused on architecting solutions revolving around Cloud, DevOps, Containerization, Backend and Frontend.