High-Throughput Game Message Server with Python websockets
Tags: webdev, python, programming, gamedev
Published: 2020-12-05 22:14:02

---
title: High-Throughput Game Message Server with Python websockets
published: true
description: Tracking an error in my game, stressing over a fix, and the code to replace it all
tags: webdev, python, programming, gamedev
cover_image: https://dev-to-uploads.s3.amazonaws.com/i/x8fufyl7npgrohklhu3q.jpg
canonical_url: https://mortoray.com/2020/12/06/high-throughput-game-message-server-with-python-websockets/
---

An error came up during a competition with [my game](https://edaqasroom.com/game/carnival). One of the 80 players got stuck. Like really stuck: a breaking defect! The error should not have happened, could not have happened, yet it did. The only possibility was the Web Socket stack I use. Turns out that layer didn’t work as intended, and there was no way to fix it. Alas, I sought an alternate solution. 

Here I describe what I came up with: a new, more focused stack, using Python websockets. Lots of coroutines, asyncio, and queues. I have the complete working example at the end of this article. This is likely the last state where it can standalone; I’ll be changing it to fit even tighter with my game code.

# Logical Structure and the Problem

[My game](https://edaqasroom.com/game/carnival) is a multiplayer puzzle game coordinated by messages. I segregate each instance of the game from the others. For messages, I do this with the classic “rooms” concept. It’s the name that Flask-SocketIO uses as well, and that’s what my first implementation used.

Mostly, messages in the game don’t need a defined total order. They can arrive in a different order to the different clients. There are a few situations where this isn’t true however, places where I need some messages to have a defined total order. There aren’t many of them, and that’s likely why I didn’t notice the defect earlier.

When I first started the project I asked whether the library, if used on a single client system, echoing in order, could maintain an order to the clients. The answer was yes, but the truth is no. Given the load is mostly network bound, it usually maintains an order. Only in a few stress times, or because of dumb luck, it sends messages out-of-order.

Fine, I can probably put a queue around it. Ugh, the throughput drops to an abysmal 70msgs/s. Without the queue it was already a slow 1200msg/s, but that was enough for my game. After a bit of back-and-forth, me and the library author disagree on what is acceptable throughput.

So I grabbed the [websockets](https://github.com/aaugustin/websockets) library instead, whipped together a proof of concept, and got 12,000msgs /s. Yeah, that’s more like I’d expect.

>Actually, I’d expect even more. And long term, if I get enough traffic, I’ll rewrite this in C++. The throughput should be entirely network bound, but it’s still CPU bound on the server. I’ve done a lot of low-level networking before to know I can push it higher, but for my needs now, 12K/s is way more than enough. I’d likely scale the number of servers before worrying about optimizing one of them.

On to the code!

# A Python websockets Messaging server

The “websockets” module is a minimal implementation of [WebSockets](https://en.wikipedia.org/wiki/WebSocket). That sounded like what I wanted. I didn’t want to go to the low-level of handling the protocol. This left me writing all my high-level logic, in particular the client rooms.

The library is easy to use. I got a basic example working with little effort. Of course, then there are lots of details to cover. Here’s a quick list of things I needed to support, features first:

-Handle an incoming client where “handle” is mainly done by the library
-Allow a client to join a game room (each client can only join one room in my system, simplifying the code)
-Allow another client to join the same game room
-Allow other clients to join other rooms
-Allow a client to send a message
-Provide a total order to the message, with a message id
-Dispatch that message to all clients in the room

>In my final code I’ll persist the messages to a Redis store, then eventually to a MongoDB. That is not part of my example code.

And there are several situations, or errors, that I’d have to deal with.

-A client disconnects cleanly or abruptly
-The client sends crap
-The client is slow
-Cleanup a room if there are no more clients in it

# Structure

My server maintains a list of clients in a list of rooms:

```py
@dataclass
class Client:
	socket: Any # What type?
	id: int
	disconnected: bool = False
	
@dataclass
class Room:
	key: str
	clients: Dict[int,Client] = field(default_factory=dict)
	new_clients: List[Client] = field(default_factory=list)
	msg_id: int = 0
	event_queue: asyncio.Queue = field(default_factory=asyncio.Queue)
	listening: bool = False
	future: Any = None # What Type?
```

I use type annotations for my Python, along with MyPy to check the types. Alas, for several library classes I’m unsure of types.  Since many of them are created automatically, or are returned from other functions, it’s difficult to determine the type. I will eventually find out all the types.

In these data types, the `socket` is the only part directly connected to the “websockets” module. It tracks the incoming connection and used to send and receive data.

In brief, the `listen_room` function handles the incoming client connections. I push all messages onto the `event_queue` of the `Room`. The `listen_room` function listens to this queue and sends messages to all clients in the room.

## One listener per room

I initially had a single listening queue that handled all the rooms. When I eventually write the lower-level server, like in C++, I’d keep this structure. When you get low-enough level, you can control a lot more details, removing the need for coroutines entirely.

But in Python there are a few reasons I’m using one listener per room:

-**not** overhead
-Redis
-bad clients

Overhead is all my Python code, and the library code, surrounding the writing to the clients. It’s not a lot, but it can add up with a lot of activity. I suspect the JSON parsing and formatting is the biggest part of it. But this is not a reason I have one listener per room. Since the Python code is running as a single real thread, it is irrelevant whether this code happens in one listener, or many listeners. It’s all unavoidable computational load.

The first real reason, Redis, is the well-behaved motivator. For each outgoing message I have to create a unique message id. In the sample code I track this in Python, in the `Room` class. On my final server, I’ll track this in a Redis integer key. Additionally, I’ll store all messages in a Redis list.  A separate process will clear this regularly and persist the messages to a MongoDB.   The calls to Redis take time, time that the server could instead process messages for other rooms. Thus I want to segregate the rooms. While one room waits on Redis, the others can continue processing.

The second reason, bad clients, is an unfortunate need. It’s possible that a client gets disconnected, or fails to process messages quickly enough. For the most part, this is handled by buffers. The calls to `socket.send` are effectively asynchronous, at least until the queue fills up. When that happens, `send` will wait until there is space in the queue. While waiting all the other rooms will stall, being unable to send any messages. By having one queue per room, I limit the damage of a client to that room only.

This won’t likely happen. First off, the websockets library has a timeout feature. Unresponsive clients will be disconnected long-before the outgoing socket buffers get filled up. My game simply doesn’t generate enough messages to ever fill the buffers. Extrapolating from my stress test, with an estimated average message size, there is room for 25K game messages in the standard buffers. And a typical run-through of my game, with a team, generates only 3 to 4 thousand messages. 

In any case, it’s good protection to have.

## `clients`, `new_clients` and memory

One advantage of having a single real thread is not needing to worry about actual [data races](https://en.wikipedia.org/wiki/Race_condition#Data_race). They simply don’t happen as they would in a multi-threaded application. Yay! No memory corruption is possible. 

It doesn’t mean that race conditions, something different,  don’t happen. The logical concerns of concurrency still exist, though to a lesser degree. Thanks cooperative threading! The most significant concern in my code is with the `clients` object. The queue listener iterates over the clients. If the list is modified during the iteration, Python will throw a concurrent modification exception. That is a strict no-no, as the iterator has no idea what it should do.

There are three cases where the list needs to be modified:

-when a client disconnects in `listen_socket`
-when a client disconnects in `listen_room`
-when a new client joins the room

At first I handled disconnect in the `listen_socket` function, but through testing noticed it can be a `socket.send()` call that detects the disconnect first. Thus the disconnect happens in multiple places. In both cases, I merely mark the client as `disconnected` in the `Client` structure. The `listen_room` skips disconnected clients while sending messages. It’ll track them and safely remove them from the room after the iteration loop. 

When a new client joins the room, `listen_socket` adds it to the `new_clients` list. `listen_room` will then add new clients prior to each message loop. It does this just after retrieving a message to ensure that all new clients get the message. This means that the room messages can arrive at a client prior to the “joined” response from joining the room. In my game, getting this ordering, along with sending old messages, is important for clients getting a consistent game state. I’ll likely have to adjust this code a bit.

At no point does `listen_socket` know if it’s safe to work with `clients`, since it can’t tell if `listen_room` is inside or outside of the loop. A lock isn’t a bad idea, but it introduces an avoidable delay on the incoming listening side, and delays in the room listener. Why lock when I don’t have to?

In retrospect, it might be a disadvantage that much of the coroutine parallelism is implicit, especially if using something like eventlets. As a programmer, it’s less apparent where the logical thread switching happens. You just have to know that every await operation, every asyncio call, and every websocket call is a potential location for a thread switch. It’d be nice to say you should assume a switch is possible at any time, but then I couldn’t rely on it not switching in some places and would require a bunch of locks.

Use [locks](https://mortoray.com/2019/02/20/how-does-a-mutex-work-what-does-it-cost/) if you aren’t sure. Performance is irrelevant if your game breaks. **grumble**

## Stats and throughput

I added a simple `Stats` class to track throughput on the server. It emits timings for all the incoming and outgoing messages per room. The 12K/s is what happens if I have multiple clients connected to unique rooms. My machine hits that 12K limit with the server process pegged at 100% CPU use.

>Unfortunately, I must adjust my number down to 10K. Once I moved the rooms to individual listeners, I hit far more overhead. I’m not entirely sure why — I can’t imagine it’s the extra number of coroutines. Likely there are some tweaks in the async stuff to improve it, but it’s still fast enough that I’m not concerned.

As a curiosity, I measured a single client connected to the server. It’s getting slightly over 5Kmsgs/s. Since this is a client and server, I have two processes. They are both at 58% CPU use. Ideally they should be at 50% CPU use, since they send a message back and forth. That extra 8% is processing spent doing stuff other than handling the message.  Perhaps if I wrote the system in C++ it’d get closer to 50%, but never reach there completely. The throughput however should go up.

>When I say C++ would be faster, it’s because of my experience. I have better control of what happens and know how to use that control. It’s easy to get it wrong and up with a steaming pile that is worse than the clean Python version. Server code is hard!

The stats don’t directly measure response time. But knowing the ping pong nature of the tests, I can calculate roughly what that’d be. At fractional milliseconds per message, it’ll be noise compared to the true network overhead when deployed.

This statistics I calculate here aren’t great. If I were trying to write a truly high-performance server, I’d track averages, standard deviations, extremes, and record it all better. The numbers it’s showing are so over-powered for my game though that there’s no need for more.

# Next steps

Now I need to get this integrated with my existing server. I think I lose the ability to use a single port and single Python instance. That’s not a big loss, it’s something I was intending on doing at some point, anyway. The game server shouldn’t be the same as the web server. This is both for performance and stability.  Eventually, should I have enough load, I must run multiple game servers (multiple servers handling web socket connections).  I have a plan to scale that direction, but it’ll be a long time before I get there.

# Code

Below is the code for the server, the Python client, and a sample client for the browser. While the details may change slightly, the structure will stay close to this. Given the size of the code, it’s better not to spin this into a library. I’ll directly it adapt it for [my game](https://edaqasroom.com/game/carnival) server.

This code is fairly stable and should work as a starting point for your own needs. Though of course, I can’t make any promises of that. I’ll likely discover things in my application that require fixes.

## ws_server.py

```py
from typing import *
from dataclasses import dataclass, field
import asyncio, websockets, json, time
from collections import defaultdict

#sys.path.append('../server')
#from escape.live_game_state import LiveGameState

def encode_msg(msg: Dict) -> str:
	return json.dumps(msg, ensure_ascii=False)
	
def decode_msg(text: str) -> Dict:
	return json.loads(text)


@dataclass
class Client:
	socket: Any # What type?
	id: int
	disconnected: bool = False
	
@dataclass
class Room:
	key: str
	clients: Dict[int,Client] = field(default_factory=dict)
	new_clients: List[Client] = field(default_factory=list)
	msg_id: int = 0
	event_queue: asyncio.Queue = field(default_factory=asyncio.Queue)
	listening: bool = False
	future: Any = None # What Type?

	def client_count(self) -> int:
		return len([c.id for c in self.clients.values() if not c.disconnected])

client_id_count = 0

rooms: Dict[str, Room] = {}
	

# Used to get a basic idea of throughput
class Stats:
	def __init__(self, name):
		self._name = name
		self._count = 0
		self._time = time.monotonic()
		
	def incr(self, amount = 1):
		self._count += amount
		if self._count > 5000:
			end_time = time.monotonic()
			print( f'{self._name} {self._count / (end_time-self._time)}/s' )
			self._count = 0
			self._time = end_time
			
			
async def listen_room(room):
	if room.listening:
		raise Exception(f'Already listening to {room.key}')
		
	room.listening = True
	print(f'Listen Room {room.key}')
	stats = Stats(f'Outgoing {room.key}')
	while True:
		qevent = await room.event_queue.get()
		if qevent == None:
			break
			
		# Add any new clients that have shown up, this handler must control this to avoid it
		# happening inside the loop below
		if len(room.new_clients) > 0:
			for client in room.new_clients:
				room.clients[client.id] = client
			room.new_clients = []
		
		# In my game I'll track IDs in Redis, to survie unexpected failures.
		# The messages will also be pushed there, to be picked up by another process for DB storage
		room.msg_id += 1
		qevent['msg_id'] = room.msg_id
		
		count = 0
		disconnected: List[int] = []
		for client in room.clients.values():
			if client.disconnected:
				disconnected.append(client.id)
				continue
			count += 1
			
			# There's likely some asyncio technique to do this in parallel
			try:
				await client.socket.send(encode_msg(qevent))
			except websockets.ConnectionClosed:
				print("Lost client in send")
				client.disconnected = True
				# Hoping incoming will detect disconnected as well
			
		stats.incr(count)
		
		# Remove clients that aren't there anymore. I don't really need this in my game, but it's
		# good to not let long-lived rooms build-up cruft.
		for d in disconnected:
			# Check again since they may have reconnected in other loop
			if room.clients[d]:
				del room.clients[d]
			
	print(f'Unlisten Room {room.key}')
	room.listening = False


async def listen_socket(websocket, path):
	global rooms, client_id_count
	print("connect", path)
	client_id_count += 1
	room: Optional[Room] = None
	client = Client(id=client_id_count, socket=websocket)
	
	stats = Stats('Incoming')
	try:
		async for message_raw in websocket:
			message = decode_msg(message_raw)
			if message['type'] == 'join':
				# Get/create room
				room_key = message['room']
				if not room_key in rooms:
					room = Room(key=room_key)
					rooms[room_key] = room
					
					room.future = asyncio.ensure_future(listen_room(room))
				else:
					room = rooms[room_key]
					
				# Add client to the room
				room.new_clients.append(client)
				
				# Tell the client which id they are.
				await websocket.send(encode_msg({
					'type': 'joined',
					'client_id': client.id
				}))
				
			elif room:
				# Identify message and pass it off to the room queue
				message['client_id'] = client.id
				await room.event_queue.put(message)
			else:
				# Behave as trival echo server if not in room (will be removed in my final version)
				await websocket.send(encode_msg(message))
			stats.incr()
	except websockets.ConnectionClosed:
		pass
	except Exception as e:
		# In case something else happens we want to ditch this client.  This won't come from
		# websockets, but likely the code above, like having a broken JSON message
		print(e)
		pass
	
	# Only mark disconnected for queue loop on clients isn't broken
	client.disconnected = True
	if room is not None:
		# Though if zero we can kill the listener and clean up fully
		if room.client_count() == 0:
			await room.event_queue.put(None)
			del rooms[room.key]
			await room.future
			print(f"Cleaned Room {room.key}")
			
	print("disconnect", rooms)


def main() -> None:
	start_server = websockets.serve(listen_socket, "localhost", 8765, ping_interval=5, ping_timeout=5)

	asyncio.get_event_loop().run_until_complete(start_server)

	asyncio.get_event_loop().run_forever()

	
main()
```

## ws_client.py

A simple client that validates the correct ordering of messages. Provide a room on the command line.

There's an option to slow this client which forces the server to disconnect it when the buffers fill up.

You'll note my client test code is rougher than my server code, and lacking many type definitions. This code will not be used long-term, but the server code will be.

```py
from typing import *
import asyncio, json, websockets, time, sys

if len(sys.argv) < 2:
	print(f"Sytnax {sys.argv[0]} room (delay)" )
	sys.exit(-1)
	
room = sys.argv[1]
# A non-zero slow creates a client that can't keep up. If there are other clients in the room
# it will end up breaking, causing the server to disconnect it.
slow = 0.0
if len(sys.argv) > 2:
	slow = float(sys.argv[2])

def encode_msg(msg: Dict) -> str:
	return json.dumps(msg, ensure_ascii=False)
	
def decode_msg(text: str) -> Dict:
	return json.loads(text)

# An even simpler stats tracker than the server	
trigger_count = 5000.0
if slow > 0:
	trigger_count /= (1+slow) * 100
	
	
async def reader(websocket):
	count = 0
	seq = 0
	last_time = time.monotonic()
	client_id = None
	last_msg_id = None
	
	async for message_raw in websocket:
		count += 1
		msg = decode_msg(message_raw)
		
		if msg['type'] == 'joined':
			client_id = msg['client_id']
		else:
			# Ensure the messages have a single total order
			msg_id = msg['msg_id']
			if last_msg_id is None:
				last_msg_id == msg_id
			else:
				if msg_id != (last_msg_id+1):
					print(last_msg_id, msg_id)
					raise Exception("bad msg sequence")
			
		if msg['type'] == 'ping' and client_id == msg['client_id']:
			# Ensure our own measures retain the order we sent them
			if msg['seq'] != seq:
				print(seq, message_raw)
				raise Exception("bad message seq")
			
		# Track rough throughput
		if count >= trigger_count:
			next_time = time.monotonic()
			print( f'{count /(next_time - last_time)}/s {room}' )
			last_time = time.monotonic()
			count = 0
			
		if client_id == msg['client_id']:
			seq += 1
			await websocket.send(encode_msg({'type': 'ping', 'seq': seq }))
			
		if slow > 0:
			await asyncio.sleep(slow)
		
		
async def hello():
	uri = "ws://localhost:8765"
	async with websockets.connect(uri) as websocket:
		print("Connect")
		await websocket.send( encode_msg({ 'type': 'join', 'room': room }) )
		consumer_task = asyncio.ensure_future(
			reader(websocket))
		done = await asyncio.wait(
			[consumer_task],
			return_when=asyncio.FIRST_COMPLETED,
		)
		

asyncio.get_event_loop().run_until_complete(hello())
```

## ws_client.html

A simple web browser client, to prove that it's working where I need it.

```py
<html>
<script>
function loaded() {

	console.log("Loaded")
	const socket = new WebSocket('ws://localhost:8765');

	socket.addEventListener('open', function (event) {
		socket.send(encode_msg({
			type: 'join',
			room: 'js-room',
		}));
		console.log("Opened")
	});

	socket.addEventListener('message', function (event) {
		console.log('Message from server ', event.data);
	});
	
	window.addEventListener('beforeunload', () => {
		console.log("UNLOAD")
		socket.close()
	})
	console.log("LOADED")
	
	let count = 0
	setInterval(() => {
		count = count+1
		socket.send(encode_msg({
			type: 'ping',
			seq: count,
		}))
	}, 500)
}

function encode_msg(msg) { 
	return JSON.stringify(msg)
}

function decode_msg(text) {
	return JSON.parse(text)
}

</script>
<body onload="loaded()">
	<p>Text</p>
</body>
</html>
```