Let’s write a simple Redis server using Python.
Redis is an in-memory key-value database. A key-value database is very simple.
At the very minimum, there are two kinds of queries: SET and GET. You can
SET a key to a value, and you can GET a key to return a value. Of course, in
practice, we need more capabilities than just SET and GET such as deletion,
timeouts, scanning, etc. However, I can argue that most of those operations
technically can be achieved by SET and GET operations. To perform delete, we
can just SET the key to null or a special value indicating that the key is
deleted. To add timeout, we can simply put the expiration time during the SET
operation and check it during the GET operation. But, most key-value databases
support those operations natively because it’s very commonly used by the users.
Redis is a very popular key-value database, and its protocol is very simple.
Let’s write a Redis server implementation in Python.
RESP
Redis documentation is really good and well-written. The wire protocol is also well documented here. Redis calls its protocol RESP, which stands for Redis Serialization Protocol. RESP was designed to be human-readable, which means, you can read the protocol without parsing it. It uses newline characters to split items in the protocol and represents numbers using text instead of its binary representation. But, even though RESP is human-readable, it was also designed to be fast to parse and simple to implement.
A Redis server runs over a TCP connection. A client can connect to the server by
creating a TCP connection and talks using RESP. By default, Redis servers use
port 6379 as the TCP port, but we will use a different port. RESP works using
request-response model: the client sends a request, and the server sends back
the response. In the context of Redis, the requests consist of a command and its
arguments. The list of Redis commands and their arguments are documented
here. We are going to implement four Redis
commands: GET, SET, SELECT, and COMMAND.
Let’s start writing a simple TCP server in Python:
1import socket
2import logging
3import dataclasses
4import asyncio
5
6@dataclasses.dataclass
7class Config:
8 bind: str
9 port: int
10
11async def run_server(config: Config):
12 tcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
13 server_address = (config.bind, config.port)
14 tcp_socket.bind(server_address)
15 tcp_socket.listen()
16 tcp_socket.setblocking(False)
17
18 loop = asyncio.get_event_loop()
19
20 logging.info("Server is listening to: %s:%d", config.bind, config.port)
21
22 while True:
23 connection, client = await loop.sock_accept(tcp_socket)
24 logging.info("New client is connected: %s", client)
25
26 loop.create_task(handle_client(connection, client))
27
28
29async def handle_client(connection: socket.socket, client: socket.AddressInfo):
30 loop = asyncio.get_event_loop()
31 while True:
32 ch = await loop.sock_recv(connection, 1)
33 await connection.sendall(ch)
34
35
36if __name__ == "__main__":
37 logging.basicConfig(level=logging.INFO)
38 config = Config(bind='0.0.0.0', port=5000)
39 try:
40 asyncio.run(run_server(config))
41 except KeyboardInterrupt:
42 logging.info("Stopping")
Data Types
There are a few data types supported in RESP. Some of them are String, Array,
Integer, Null, and Error. The String data type itself technically can be divided
into two types: Simple String and Bulk String, but we will just refer to them as
strings. The client requests are represented as an Array, where the first index
is a String containing the command to be executed, and the rest of the elements
are the command arguments. For example, the Array
["SET", "key_example", "value_example"] represents a client request to set a
key "key_example" to a String "value_example". The server responses are also
represented as one of the Redis data types. Depending on the command, some
commands return a simple String “Ok”, some return an array.
Encoding
Let’s start writing a Python implementation to encode the Redis data types. We
are going to represent the Redis data types using Python data types. We are
going to use Python list for Redis Array, Python bytes for Redis String,
Python int for Redis Integer, Python None for Redis Null and the Error
class below to represent Redis Error.
1import dataclasses
2
3@dataclasses.dataclass
4class Error:
5 code: str
6 message: str
The
documentation of RESP data types
is very clear. To encode a String, we can simply write the $ character
followed by the length of the string, followed by \r\n (a CRLF), followed by
the string itself, and followed by \r\n again. So, a String “hello” can be
encoded into $5\r\nhello\r\n, and an empty string can be encoded as
$0\r\n\r\n. Let’s write a Python implementation for that:
1async def write_string_value(connection: socket.socket, value: bytes):
2 loop = asyncio.get_event_loop()
3 await loop.sock_sendall(connection, b'$')
4 await loop.sock_sendall(connection, bytes(str(len(value)), 'ascii'))
5 await loop.sock_sendall(connection, b'\r\n')
6 await loop.sock_sendall(connection, value)
7 await loop.sock_sendall(connection, b'\r\n')
Interestingly, there is a special value in Redis called Null. Null means
nothing. And interestingly, we can encode Null by seeing it as a string with
-1 length. So we can encode a Null value as: $-1\r\n. Let’s code that:
1async def write_null_value(connection: socket.socket):
2 loop = asyncio.get_event_loop()
3 connection.sendall(b'$-1\r\n')
To encode a number, we need to use : character prefix, followed by the number
in text format, followed by \r\n characters. So a number 1023 can be encoded
into :1023\r\n. We can also encode a negative number as it is. For example, a
number -10 can be encoded into :-10\r\n. Let’s write a Python implementation
to encode numbers:
1async def write_number_value(connection: socket.socket, value: int):
2 loop = asyncio.get_event_loop()
3 await loop.sock_sendall(connection, b':')
4 await loop.sock_sendall(connection, bytes(str(value), 'ascii'))
5 await loop.sock_sendall(connection, b'\r\n')
Next, let’s encode an error type. Errors are quite similar to strings; they are
basically just a bunch of characters. But, unlike strings, errors have two
things in it: the error code and the error message. Both the error code and the
error message are just a simple string. To encode an error type, we can use the
- prefix, followed by the error code, followed by a space character, followed
by the error message, and followed by a \r\n. Let’s write Python
implementation to encode errors:
1async def write_error_value(connection: socket.socket, value: Error):
2 loop = asyncio.get_event_loop()
3 await loop.sock_sendall(connection, b'-')
4 await loop.sock_sendall(connection, bytes(value.code, 'ascii'))
5 await loop.sock_sendall(connection, b' ')
6 await loop.sock_sendall(connection, bytes(value.message, 'ascii'))
7 await loop.sock_sendall(connection, b'\r\n')
Finally, let’s encode an array type. Arrays are basically just a collection of
multiple values. An array can also contain other arrays inside it. Redis encodes
Array by specifying its length, followed by the encoded values of its elements.
To be precise, to encode an array, we need to write a * prefix, followed by
the length of the Array in text format, followed by a \r\n, and followed by
the encoded values of its elements. We can implement Array encoding recursively
like this:
1from typing import List, Any
2
3
4async def write_array_value(connection: socket.socket, value: List[Any]):
5 loop = asyncio.get_event_loop()
6 await loop.sock_sendall(connection, b'*')
7 await loop.sock_sendall(connection, bytes(str(len(value)), 'ascii'))
8 await loop.sock_sendall(connection, b'\r\n')
9 for element in value:
10 await write_value(connection, element)
Now that we have a way to encode various types of Values, let’s write a single function to encode a Value of any type. But before that, we need to know that Redis has more types of Values. We are only implementing a few types that we need to make this simple server work. In our implementation, we will just encode an error if we don’t recognize the Value type.
1async def write_value(connection: socket.socket, value: Any):
2 if value is None:
3 await write_null_value(connection)
4 elif isinstance(value, bytes):
5 await write_string_value(connection, value)
6 elif isinstance(value, int):
7 await write_number_value(connection, value)
8 elif isinstance(value, list):
9 await write_array_value(connection, value)
10 elif isinstance(value, Error):
11 await write_error_value(connection, value)
12 else:
13 logging.error(
14 "Found unrecognized value with type (%s): %s", type(value), value
15 )
16 value = Error(code="ERR", message="internal server error")
17 await write_error_value(connection, value)
Decoding
Now, we have the ability to write any values to a Redis client. But, we also need to read a value from a Redis client. To do that, we need to write the decoding logic to decode Redis Values. The decoding logic is basically just the inverse of the encoding logic. Let’s implement decoding logic in Python:
1from typing import List, Any, Union
2
3
4async def read_value(connection: socket.socket) -> Any:
5 loop = asyncio.get_event_loop()
6
7 marker = await loop.sock_recv(connection, 1)
8 if marker == b'*':
9 return await read_array(connection)
10 elif marker == b'$':
11 return await read_bulk_string(connection)
12 elif marker == b'+':
13 return await read_simple_string(connection)
14 elif marker == b':':
15 return await read_number(connection)
16 elif marker == b'':
17 raise EOFError
18
19
20async def read_array(connection: socket.socket) -> List[Any]:
21 array_len = await read_number(connection)
22 array_values = [None] * array_len
23 for i in range(array_len):
24 array_values[i] = await read_value(connection)
25 return array_values
26
27
28async def read_bulk_string(connection: socket.socket) -> Union[None, bytes]:
29 loop = asyncio.get_event_loop()
30
31 str_len = await read_number(connection)
32 if str_len < 0:
33 return None
34 str_value = await loop.sock_recv(connection, str_len)
35
36 b = await loop.sock_recv(connection, 2)
37 assert (b == b'\r\n')
38
39 return str_value
40
41
42async def read_simple_string(connection: socket.socket) -> bytes:
43 return await read_until_newline(connection)
44
45
46async def read_number(connection: socket.socket) -> int:
47 return int(await read_until_newline(connection))
48
49
50async def read_until_newline(connection: socket.socket) -> bytes:
51 loop = asyncio.get_event_loop()
52
53 result = bytes()
54 while True:
55 chr = await loop.sock_recv(connection, 1)
56 if chr == b'\r':
57 break
58 result += chr
59
60 b = await loop.sock_recv(connection, 1)
61 assert (b == b'\n')
62
63 return result
Command Processing
Now that we have the encoding and decoding logic, let’s write the actual client handler. Currently, our handler is very simple, we just read any bytes from the client, and write back the same bytes back to the client. The Redis server works in request-response pattern: the client sends a Redis Value as a command, and the server should write back a Redis Value as the response. Let’s write a simple client handler to read the command from the client. We won’t be processing the command yet, we just want to see what the command looks like.
1async def handle_client(connection: socket.socket, client: socket.AddressInfo):
2 while True:
3 try:
4 await handle_client_request(connection)
5 except EOFError:
6 logging.info("Client is disconnected: %s", client)
7 return
8
9
10async def handle_client_request(connection: socket.socket):
11 request = await read_value(connection)
12 print(request)
Let’s test our server implementation. Redis has a CLI tool called redis-cli
that we can use to talk with our server. Let’s run our server and redis-cli to
test it:
1$ redis-cli -h 127.0.0.1 -p 5000
If you look at your server, you will see something like this:
1INFO:root:Server is listening to: 0.0.0.0:5000
2INFO:root:New client is connected: ('127.0.0.1', 45184)
3[b'COMMAND']
In the third line, you’ll see an array with "COMMAND" string in it. When a
Redis client connects to a Redis server, they will send a command called
COMMAND. Yes, there is a command called COMMAND. This COMMAND command is
used to fetch all the supported commands from the server. As you can see, Redis
uses arrays to represent a command. The first element of the Array is the
command name, and the rest are the command arguments. Next, let’s write the
implementation of the COMMAND command.
COMMAND Command
The details of COMMAND command are documented
here in the Redis documentation. We need
to return a list of all commands we support, in this case we only support four
types of commands: COMMAND, GET, SET, and SELECT. A command is
represented as an array with this specification:
- Name
- Arity
- Flags
- First key
- Last key
- Step
- ACL categories (as of Redis 6.0)
- Tips (as of Redis 7.0)
- Key specifications (as of Redis 7.0)
- Subcommands (as of Redis 7.0)
The first element of the command is the command name. The second element is the number of command arguments. Some commands can have an infinite number of arguments and need special handling, but we are not going to implement that. The third element is the command flags. Command flags tell us various information of the command such as whether the command is slow, whether the command is read-only, whether the command is for processing string, etc. The fourth, fifth, and sixth elements are related to each other. They represent the argument index of the first key, the argument index of the last key, and the number of steps to get into the next key.
For simplicity, we are going to use only six of them. We are not going to implement the ACL, tips, key specifications, and subcommands. And we are not going to support multiple keys.
Since the return value of the COMMAND command is constant, we can just
represent it as a single global variable for the response.
1@dataclasses.dataclass
2class Command:
3 name: str
4 arguments_len: int
5 flags: List[str]
6 first_key: int
7 last_key: int
8 key_step: int
9
10commands = {
11 b"COMMAND": Command(
12 name=b"COMMAND",
13 arguments_len=1,
14 flags=[b"readonly", b"random"],
15 first_key=1,
16 last_key=1,
17 key_step=1,
18 ),
19 b"GET": Command(
20 name=b"GET",
21 arguments_len=2,
22 flags=[b"readonly", b"random"],
23 first_key=1,
24 last_key=1,
25 key_step=1,
26 ),
27 b"SET": Command(
28 name=b"SET",
29 arguments_len=1,
30 flags=[b"write", b"string", b"slow"],
31 first_key=1,
32 last_key=1,
33 key_step=1,
34 ),
35 b"SELECT": Command(
36 name=b"SELECT",
37 arguments_len=2,
38 flags=[b"write", b"string", b"slow"],
39 first_key=0,
40 last_key=0,
41 key_step=0,
42 )
43}
44
45
46async def handle_command_command() -> Any:
47 global commands
48
49 return [
50 [command.name, command.arguments_len, command.flags,
51 command.first_key, command.last_key, command.key_step]
52 for command in commands.values()
53 ]
Then, let’s connect our handle_command_command to our handle_client_req:
1async def handle_client_request(connection: socket.socket):
2 request = await read_value(connection)
3
4 if not isinstance(request, list):
5 logging.warn("Invalid request from client: %s", request)
6 return
7 if len(request) == 0:
8 logging.warn("Invalid request from client: %s", request)
9 return
10 if not isinstance(request[0], bytes):
11 logging.warn("Invalid request from client: %s", request)
12 return
13
14 logging.debug("Got request: %s", request)
15
16 command, args = request[0].upper(), request[1:]
17 if command == b'COMMAND':
18 result = await handle_command_command()
19 else:
20 msg = "unknown command `{}`, with args beginning with: {}".format(
21 command, args)
22 result = Error(code="ERR", message=msg)
23
24 await write_value(connection, result)
Now, we have handled our COMMAND command. If you try running redis-cli
again, you can try running some commands:
1$ redis-cli -h 127.0.0.1 -p 5000
2127.0.0.1:5000> GET 1
3(error) ERR unknown command `b'GET'`, with args beginning with: [b'1']
4127.0.0.1:5000> SET a b
5(error) ERR unknown command `b'SET'`, with args beginning with: [b'a', b'b']
SELECT Command
Redis server was designed to have multiple databases running at the same time.
Normally, there are 16 databases numbered from 0 to 15. When a client connects
to a Redis server, they will use database 0 by default. The client can use the
SELECT command to change the database by passing the database number. The
selected database is local to every client, which means two different clients
can select two different databases. To implement this, we need to save the
current selected database for every client. Let’s create a data structure called
Session to store any client-scoped information.
1@dataclasses.dataclass
2class Config:
3 bind: str
4 port: int
5 databases: int
6
7
8@dataclasses.dataclass
9class Session:
10 selected_database: int = 0
11
12
13if __name__ == "__main__":
14 ...
15 config = Config(bind='0.0.0.0', port=5000, databases=16)
16 ...
Note that we also added a databases field in the Config structure to
configure the maximum number of databases.
Now, we have to create and pass the Session to our command handler.
1async def run_server(config: Config):
2 ...
3 loop.create_task(handle_client(connection, client, config))
4
5
6async def handle_client(conn: socket.socket, client: socket.AddressInfo, config: Config):
7 session = Session()
8 while True:
9 ...
10 await handle_client_request(config, session, conn)
11 ...
12
13async def handle_client_request(config: Config, session: Session, connection: socket.socket):
14 ...
Now, let’s implement our SELECT command handler. You can check the detailed
specification of the SELECT command here.
Basically, it will just return a simple “OK” string if the operation succeeded
and returns an error when the operation failed.
1async def handle_client_request(config: Config, session: Session, connection: socket.socket):
2 ...
3 if command == b'COMMAND':
4 result = await handle_command_command()
5 elif command == b'SELECT':
6 result = await handle_select_command(config, session, args)
7 ...
8
9async def handle_select_command(config: Config, session: Session, args: Any) -> Any:
10 if not isinstance(args, list) or len(args) != 1:
11 return Error(code="ERR", message="wrong number of arguments for 'select' command")
12 target_db = args[0]
13 if not isinstance(target_db, bytes):
14 return Error(code="ERR", message="invalid DB index")
15 try:
16 target_db = int(target_db)
17 except:
18 return Error(code="ERR", message="invalid DB index")
19 if target_db >= config.databases:
20 return Error(code="ERR", message="DB index is out of range")
21
22 session.selected_database = target_db
23
24 return b'OK'
Now, if you run the redis-cli again, you can use the SELECT command to
change the database.
1$ redis-cli -h 127.0.0.1 -p 5000
2127.0.0.1:5000> SELECT 1
3"OK"
4127.0.0.1:5000[1]> SELECT 0
5"OK"
6127.0.0.1:5000> SELECT 16
7(error) ERR DB index is out of range
8127.0.0.1:5000> SELECT 1
9"OK"
SET Command
Let’s implement the SET command. SET command is technically quite complex,
but we will simplify it in our implementation. The details of Redis’s SET
operation are documented here. Redis supports
timeouts in its SET command, but we won’t support that. We will just accept
two arguments for SET, the key and the value, and then store it in our server.
To store the key-value pairs sent by the clients, we need to have a hash map.
Actually, we need to have multiple hash maps, one for every database. Remember,
Redis supports multiple databases.
Let’s create a global variable to store all of our hash maps. Additionally,
let’s also create a global mutex to guard our hash maps for concurrent
execution. We are also going to create a function init_database to initialize
the databases and the locks.
1databases: List[Dict[bytes, Any]] = None
2locks: List[asyncio.Lock] = None
3
4
5def init_database(config: Config):
6 global databases
7 global locks
8
9 databases = [{} for _ in range(config.databases)]
10 locks = [asyncio.Lock() for _ in range(config.databases)]
11
12
13if __name__ == "__main__":
14 ...
15 config = Config(...)
16 init_database(config)
17 ...
Now, let’s implement the SET command. Implementing the SET command is pretty
straightforward. What we need to do is just lock the current selected database
and put the key and value into the Python dictionary.
1async def handle_set_command(session: Session, args: Any) -> Any:
2 global databases
3 global locks
4
5 if not isinstance(args, list) or len(args) < 2:
6 return Error(code="ERR", message="wrong number of arguments for 'set' command")
7
8 key, value = args[0], args[1]
9 if not isinstance(key, bytes):
10 return Error(code="ERR", message="invalid key")
11 if not isinstance(value, bytes):
12 return Error(code="ERR", message="invalid value")
13
14 async with locks[session.selected_database]:
15 database = databases[session.selected_database]
16 database[key] = value
17
18 return b'OK'
19
20async def handle_client_request(...):
21 ...
22 if command == b'COMMAND':
23 ...
24 elif command == b'SET':
25 result = await handle_set_command(session, args)
26 ...
Now, if you run the redis-cli again, you can use the SET command to put a
key-value pair into the database:
1$ redis-cli -h 127.0.0.1 -p 5000
2127.0.0.1:5001> SET a b
3"OK"
4127.0.0.1:5001> SET a c
5"OK"
6127.0.0.1:5001> SET a
7(error) ERR wrong number of arguments for 'set' command
8127.0.0.1:5001> SET c
9(error) ERR wrong number of arguments for 'set' command
10127.0.0.1:5001> SET
11(error) ERR wrong number of arguments for 'set' command
GET Command
Finally, let’s implement the GET command. GET is quite simple: it accepts
one argument which is the key to be fetched, and returns the value of the key.
1async def handle_get_command(session: Session, args: Any) -> Any:
2 global databases
3 global locks
4
5 if not isinstance(args, list) or len(args) != 1:
6 return Error(code="ERR", message="wrong number of arguments for 'get' command")
7
8 key = args[0]
9 if not isinstance(key, bytes):
10 return Error(code="ERR", message="invalid key")
11
12 async with locks[session.selected_database]:
13 database = databases[session.selected_database]
14 return database.get(key)
15
16
17async def handle_client_request(...):
18 ...
19 if command == b'COMMAND':
20 ...
21 elif command == b'GET':
22 result = await handle_get_command(session, args)
23 ...
Now, you have a full implementation of a Redis server that supports SET,
GET, and SELECT commands. You can test your server by using redis-cli:
1$ redis-cli -h 127.0.0.1 -p 5000
2127.0.0.1:5002> GET key1
3(nil)
4127.0.0.1:5002> SET key1 value1
5"OK"
6127.0.0.1:5002> GET key1
7"value1"
8127.0.0.1:5002> SELECT 1
9"OK"
10127.0.0.1:5002[1]> GET key1
11(nil)
12127.0.0.1:5002[1]> SET key1 "value 1 from db 1"
13"OK"
14127.0.0.1:5002[1]> GET key1
15"value 1 from db 1"
16127.0.0.1:5002[1]> SELECT 0
17"OK"
18127.0.0.1:5002> GET key1
19"value1"
20127.0.0.1:5002> SELECT 1
21"OK"
22127.0.0.1:5002[1]> GET key1
23"value 1 from db 1"
24127.0.0.1:5002[1]>
You can find the full code in this gist. This implementation is no match for the actual Redis server implementation. The Redis server is written in C and is highly optimized for key-value store.