Search⌘ K
AI Features
Home
Courses
Python Concurrency for Senior Engineering Interviews

... continued

Explore how to implement a concurrent chat server in Python using the Async.io module. Understand the use of coroutines, event loops, and asynchronous callbacks to manage client connections without threading. Learn how to handle client messaging and server operations efficiently within a single-threaded asynchronous environment.

We'll cover the following...

The code widget below runs a simulation of the chat server with four users that message each other (including themselves) randomly.

Python 3.5
from threading import Thread
from threading import Lock
import socket, time, random
class ChatServer:
    def __init__(self, port):
        self.port = port
        self.lock = Lock()
        self.clients = {}
    def handle_client(self, client_socket):
        user = "unknown"
        while True:
            data = client_socket.recv(4096).decode()
            command, param = data.split(",")
            # register handler
            if command == "register":
                print("\n{0} registered\n".format(param))
                with self.lock:
                    self.clients[param] = client_socket
                user = param
                client_socket.send("ack".encode())
            # list handler
            if command == "list":
                with self.lock:
                    names = self.clients.keys()
                names = ",".join(names)
                client_socket.send(names.encode())
            # chat handler
            if command == "chat":
                to_socket = None
                with self.lock:
                    if param in self.clients:
                        to_socket = self.clients[param]
                if to_socket is not None:
                    to_socket.send(("{0} says hi\n".format(user)).encode())
                else:
                    print("\nNo user by the name <{0}>\n".format(param))
    def run_server(self):
        # networking stuff to setup the connection, that the
        # can ignore
        socket_connection = socket.socket()
        socket_connection.bind(('', self.port))
        socket_connection.listen(5)
        # perpetually listen for new connections
        while True:
            client_socket, addr = socket_connection.accept()
            # spawn a thread to deal with a new client and immediately go back to
            # listening for new incoming connections
            Thread(target=self.handle_client, args=(client_socket,), daemon=True).start()
class User:
    def __init__(self, name, server_host, server_port):
        self.name = name
        self.server_port = server_port
        self.server_host = server_host
    def receive_messages(self, server_socket):
        while True:
            print("\n{0} received: {1}\n".format(self.name, server_socket.recv(4096).decode()))
    def run_cleint(self):
        server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server_socket.connect((self.server_host, self.server_port))
        # register and receive ack
        server_socket.send("register,{0}".format(self.name).encode())
        server_socket.recv(4096).decode()
        # wait for friends to join
        time.sleep(3)
        # get list of friends
        server_socket.send("list,friends".encode())
        list_of_friends = server_socket.recv(4096).decode().split(",")
        num_friends = len(list_of_friends)
        # start listening for incoming messages
        Thread(target=self.receive_messages, args=(server_socket,), daemon=True).start()
        while True:
            # randomly select a friend and send a message
            friend = list_of_friends[random.randint(0, num_friends - 1)]
            server_socket.send("chat,{0}".format(friend).encode())
            time.sleep(random.randint(2, 6))
if __name__ == "__main__":
    server_port = random.randint(10000, 65000)
    server_host = "127.0.0.1"
    server = ChatServer(server_port)
    # start server
    Thread(target=server.run_server, daemon=True).start()
    time.sleep(1)
    # jane gets online
    jane = User("Jane", server_host, server_port)
    Thread(target=jane.run_cleint, daemon=True).start()
    zak = User("Zak", server_host, server_port)
    Thread(target=zak.run_cleint, daemon=True).start()
    abhishek = User("Abhishek", server_host, server_port)
    Thread(target=abhishek.run_cleint, daemon=True).start()
    igor = User("Natasha", server_host, server_port)
    Thread(target=igor.run_cleint, daemon=True).start()
    time.sleep(30)

Note that our implementation makes generous use of daemon threads so that when the main program thread exits, all the threads are killed and the code-widget doesn't time out. The output will show various users receiving hi messages from other users.

Next, we'll pivot our implementation to an asynchronous one.

Asynchronous Implementation

The asynchronous paradigm to implement a chat server involves using a single thread, usually called the event loop. In the case of Python, we'll not implement an event loop from scratch rather leverage asyncio module for the job.

Similar to the multithreaded approach, we need an entity to continuously listen for new incoming connections. We'll use the asyncio's start_server() coroutine. The coroutine takes in the server's hostname and a port to listen for incoming connections on. Once a client connects with the server, the coroutine invokes a user-specified callback to invoke. The callback includes the details to read and write from the connected client. The code snippet to start the server would look like as follows:

    server_port = random.randint(10000, 65000)
    server_host = "127.0.0.1"
    chat_server = ChatServer(server_port)
...