Socket module: multiple clients

In the previous topic, we created a server that accepts a client, talks to it until it disconnects, and only then moves on to the next one. For a small experiment, this is enough, but real servers usually have to talk to many clients all at once. In this topic, we'll use threads to make our server serve several clients simultaneously and discuss the problems that come with it.

One client at a time

Let's recall how the server from the previous topic works. It has two loops: the outer one accepts a new client, and the inner one exchanges data with this client until it disconnects:

while True:
    client_socket, client_address = server_socket.accept()

    while True:
        data = client_socket.recv(1024)
        if not data:
            break
        client_socket.sendall(data)

    client_socket.close()

The problem is that both accept() and recv() are blocking. While the server waits for the data from the first client inside the inner loop, it can't call accept() again. So, if the first client connects and stays silent for a minute, all other clients have to wait for that minute, even though the server is doing nothing.

In terms of our information office, there is only one consultant, and a queue of visitors stands in front of the table. The solution is to hire more consultants: let the one who sits at the entrance only greet the visitors, and let each visitor be served by a separate consultant. In Python, such "consultants" are threads.

A thread for each client

The idea is simple: the main thread does nothing but accept() new clients. As soon as a client connects, we hand its socket over to a new thread and immediately go back to accept(). All the communication with the client happens in that thread.

First, let's move the code that talks to the client into a function:

import socket
import threading


def handle_client(client_socket, client_address):
    with client_socket:
        while True:
            data = client_socket.recv(1024)
            if not data:
                break
            client_socket.sendall(data)

Then, in the main loop, we create and start a thread for each accepted client:

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server_socket:
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_socket.bind(('127.0.0.1', 9090))
    server_socket.listen(5)

    while True:
        client_socket, client_address = server_socket.accept()

        client_thread = threading.Thread(
            target=handle_client,
            args=(client_socket, client_address),
            daemon=True
        )
        client_thread.start()

Let's look at the details:

• target is the function that the thread will run, and args is a tuple with its arguments. Don't forget the comma if there is only one argument: args=(client_socket,);

• the thread starts working only after we call start(). If we called handle_client(...) directly instead, we would just get the same sequential server as before, since the main thread would wait for the function to finish;

• daemon=True makes the thread a daemon: it will not prevent the program from exiting. Without it, the program would not finish while at least one client thread is still running;

• the client socket is closed inside the thread by the with statement, so the main thread doesn't have to keep track of it.

Now the server can talk to several clients at once. If you connect with two clients, the second one will get the response even when the first one is connected but silent.

Sharing data between clients

Often, clients of the same server need to work with the same data: a list of messages, a counter of requests, a database of keys and values. All the threads of a program share the same memory, so they can access the same variables. However, it's not as safe as it sounds.

Let's add a counter of all the messages received from all the clients:

total_messages = 0

Increasing it looks like a single action: total_messages += 1. But in fact, Python has to read the current value, add one to it and write the result back. If two threads read the value at the same moment, both will add one to the same number, and one of the increments will be lost. Such a situation is called a race condition. Its results are unpredictable and hard to reproduce, which makes it a nasty bug.

To avoid it, we can use a lock: only one thread at a time can be inside the block protected by it, and the others wait for their turn.

total_messages = 0
total_messages_lock = threading.Lock()


def handle_client(client_socket, client_address):
    global total_messages

    with client_socket:
        while True:
            data = client_socket.recv(1024)
            if not data:
                break

            with total_messages_lock:
                total_messages += 1
                number = total_messages

            reply = f'#{number}: '.encode() + data
            client_socket.sendall(reply)

Pay attention to two things here. First, we use the lock as a context manager, so it is released automatically, even if an error occurs inside the block. Second, we copy the value to the local variable number inside the lock. If we read total_messages after the lock is released, another thread might have already changed it. Also, it's better to keep the protected block as short as possible: we send the reply outside of it, so that a slow client doesn't make the others wait.

The same approach is used for any data shared between the clients, for example, a dictionary: every access to it, both reading and writing, should be done under the same lock.

Messages and the TCP stream

There is one more thing that matters for a server with real clients. TCP is a stream of bytes: it doesn't know where one message ends and the next one begins. A single recv(1024) call may return only a part of a message, or two messages glued together, depending on how the data was split on its way.

That's why the server needs a rule that tells where a message ends. The simplest one is to end each message with a newline character \n. Then the server keeps the received data in a buffer and processes only the complete messages:

buffer = b''

while True:
    data = client_socket.recv(1024)
    if not data:
        break

    buffer += data

    while b'\n' in buffer:
        line, buffer = buffer.split(b'\n', 1)
        message = line.decode().strip()
        # process the complete message here

The inner while is needed because one recv() call can bring several complete messages at once. Whatever is left in the buffer after the last \n is an incomplete message that will be finished by the data from the next recv().

Of course, a newline is not the only possible choice. Some protocols send the length of the message before the message itself, and some use special separators. What matters is that the client and the server agree on the same rule.

Stopping the server

Our server runs in an infinite loop, so it can be stopped only by force, for example, with Ctrl+C. But sometimes a server should be able to stop by itself, say, when it receives a special command from a client.

For this purpose, we can use a threading.Event: an object that works as a flag shared between threads. One thread can set it with set(), and the others can check it with is_set():

stop_event = threading.Event()

The main loop will check the flag on each iteration: while not stop_event.is_set(). However, there is a catch. If the main thread is blocked on accept() and no client connects, it will never get to check the flag. To fix this, we set a timeout for the server socket:

server_socket.settimeout(0.5)

Now accept() waits for a client for at most half a second. If nobody connects during this time, it raises socket.timeout, which we can catch and simply continue the loop. This way, the flag is checked twice a second:

while not stop_event.is_set():
    try:
        client_socket, client_address = server_socket.accept()
    except socket.timeout:
        continue

    # starting a thread for the client

When the flag is set, the loop ends, the server socket is closed, and the program finishes. Since the client threads are daemons, they are stopped together with the main thread, so we don't have to wait for them.

Overview

Let's put everything together. Here is a server that serves several clients at the same time, receives the messages ending with a newline, numbers them using a counter shared by all the clients, and stops when someone sends the stop message:

import socket
import threading

total_messages = 0
total_messages_lock = threading.Lock()
stop_event = threading.Event()


def handle_client(client_socket, client_address):
    global total_messages

    buffer = b''

    try:
        with client_socket:
            while True:
                data = client_socket.recv(1024)
                if not data:
                    break

                buffer += data

                while b'\n' in buffer:
                    line, buffer = buffer.split(b'\n', 1)
                    message = line.decode().strip()

                    if message == 'stop':
                        stop_event.set()
                        return

                    with total_messages_lock:
                        total_messages += 1
                        number = total_messages

                    reply = f'#{number} from {client_address[1]}: {message}\n'
                    client_socket.sendall(reply.encode())
    except ConnectionError:
        print('Connection with', client_address, 'was lost')


with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server_socket:
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_socket.bind(('127.0.0.1', 9090))
    server_socket.listen(5)
    server_socket.settimeout(0.5)

    while not stop_event.is_set():
        try:
            client_socket, client_address = server_socket.accept()
        except socket.timeout:
            continue

        client_thread = threading.Thread(
            target=handle_client,
            args=(client_socket, client_address),
            daemon=True
        )
        client_thread.start()

To try it, run the server and then start two or more clients in different terminal windows. Here is a simple client that sends one message and prints the response (don't forget the \n at the end of the message):

import socket

with socket.socket() as client_socket:
    client_socket.connect(('127.0.0.1', 9090))
    client_socket.sendall('Wake up, Neo\n'.encode())
    print(client_socket.recv(1024).decode())

Each time you run the client, the number in the response grows, no matter which client sent the message. This shows that all the threads work with the same counter. To stop the server, send the stop message.

Note that a thread per client is not the only way to serve many clients. It's simple and works well for a moderate number of connections, but each thread takes resources, and a very large number of them can become a problem. For such cases, there are other approaches, for example, asynchronous programming with asyncio.

Conclusion

Let's go over the main points of the topic:

• a server that handles clients in a single thread can't accept new clients while it waits for the data from the current one, because accept() and recv() are blocking;

• to serve several clients at the same time, the main thread only calls accept(), and each accepted client socket is passed to a new threading.Thread that talks to the client;

• client threads are usually created as daemons, so they don't prevent the program from exiting;

• the data shared between the threads, such as counters or dictionaries, should be protected with a threading.Lock to avoid race conditions, and the protected block should be kept short;

• TCP is a stream of bytes, so one recv() can return a part of a message or several messages at once; a server should keep the received data in a buffer and process only complete messages, using a rule that tells where a message ends, for example, a newline;

• a threading.Event can be used as a flag to stop the server, and a timeout on the server socket lets a blocked accept() check this flag regularly.