Asyncio Server Client Callback Handler

March 26, 2024 Python Asyncio

The asyncio server accepts client connections that call a custom client callback coroutine.

Each time a client is connected to the asyncio server, the client callback coroutine is run in a new asyncio task and the task is not awaited by the server.

This means that each client connection is isolated from the server. In turn, it means that:

If a client callback task is canceled, the CancelledError will not propagate to the server task.
If a client callback task fails with an unhandled exception, the exception will not propagate to the server task.

This means that the task running the server is robust to failures that might occur within the client connection callback tasks.

In this tutorial, you will discover the client callback coroutine used by the asyncio server to manage connections.

Let's get started.

How to Create an Asyncio Server

An asyncio server accepts incoming TCP client connections.

It is represented in asyncio Python programs via the asyncio.Server class.

The create_server() method returns a Server instance, which wraps the sockets (or other network objects) used to accept requests.
-- PEP 3156 – Asynchronous IO Support Rebooted: the "asyncio" Module

We use an asyncio.Server in our asyncio programs when we want to accept connections from other programs.

We should not create an asyncio.Server instance directly. Instead, we can create a server using the asyncio.start_server() module function.

This function takes a coroutine handler used to manage each client connection. It also takes details about the server, such as the host address and port number.

For example:

...
# create an asyncio server
server = await asyncio.start_server(handler, '127.0.0.1', port=8888)

You can learn more about how to create an asyncio server in the tutorial:

How to Use Asyncio Stream Server

The handler must take two arguments, a asyncio.StreamReader for reading data from the client and an asyncio.StreamWriter for writing data to the client.

A handler coroutine may look as follows:

# handler for client connections
async def handler(reader, writer):
    pass

Once created, we can call the serve_forever() coroutine to accept connections forever.

For example:

...
# accept client connections forever
await server.serve_forever()

This can be performed within the asynchronous context manager of the server, ensuring the server is closed safely and all clients are disconnected if the server fails with an error.

For example:

...
# use the server
async with server:
    # accept client connections forever (kill via control-c)
    await server.serve_forever()
	# close automatically

You can learn more about how to use the asyncio server context manager in the tutorial:

Asyncio Server Context Manager

Now that we know how to create a server, let's consider the client callback handler in more detail.

About the Client Callback Handler

The client callback handler in the server is a coroutine that manages each new client connection.

The client_connected_cb callback is called whenever a new client connection is established. It receives a (reader, writer) pair as two arguments, instances of the StreamReader and StreamWriter classes.
-- Streams, Asyncio API Documentation.

When a client connects to the asyncio server, this creates a new task and runs the specified callback.

This is helpful to know because each client that is connected to the server will be running in a separate task.

client_connected_cb can be a plain callable or a coroutine function; if it is a coroutine function, it will be automatically scheduled as a Task.
-- Streams, Asyncio API Documentation.

Because each client connection is run in a separate task and this task is not awaited by the server task itself, this has a few implications, such as:

If a task running a client connection fails with an exception, the exception is not propagated back to the server.
If a task running a client connection is canceled, raising an asyncio.CancelledError, it is not propagated back to the server.

This means that client tasks are isolated from the server task.

Now that we understand some details about how client connections are managed by the server, let's look at some worked examples.

Example Client Callback Handler Are Separate Tasks

We can confirm that client connections are run as separate tasks in the server program.

In this case, we will develop a server, a client to connect the server, then run both programs and introspect all running tasks in the server after the client has connected.

We expect to see a separate class for the client connection within the server program.

Let's dive in.

Asyncio Server

The first step is to define the asyncio server.

We can start with the client connection handler.

In this case, the handler will report that a client is connected, sleep for 2 seconds to slow things down for us, and then report the details of all tasks running in the server asyncio event loop.

Finally, the handle will disconnect the client and wait for the client to disconnect.

The handler() coroutine below implements this.

# handler for client connections
async def handler(reader, writer):
    # report progress
    print('Client connected')
    # await a moment
    await asyncio.sleep(2)
    # report all tasks in the event loop
    for task in asyncio.all_tasks():
        print(task)
    # close the client connection
    await writer.close()
    # wait for the connection o close
    await writer.wait_closed()

Next, we can define the main coroutine.

This will create the server and server forever using the context manager interface.

We can use the walrus operator to await and assign the creation of the server within the context manager block.

For more on this trick, see the tutorial:

Asyncio Server Context Manager

The main() that implements this is listed below.

# main coroutine
async def main():
    # create the server and accept clients forever
    async with (server:= await asyncio.start_server(handler, '127.0.0.1', 8888)):
        # report progress
        print('Server is running')
        # accept client connections forever
        await server.serve_forever()

Finally, we can start the asyncio event loop.

...
# start the asyncio event loop
asyncio.run(main())

Tying this together, the complete example is listed below.

# SuperFastPython.com
# example showing client handlers are executed in separate tasks
import asyncio

# handler for client connections
async def handler(reader, writer):
    # report progress
    print('Client connected')
    # await a moment
    await asyncio.sleep(2)
    # report all tasks in the event loop
    for task in asyncio.all_tasks():
        print(task)
    # close the client connection
    await writer.close()
    # wait for the connection o close
    await writer.wait_closed()

# main coroutine
async def main():
    # create the server and accept clients forever
    async with (server:= await asyncio.start_server(handler, '127.0.0.1', 8888)):
        # report progress
        print('Server is running')
        # accept client connections forever
        await server.serve_forever()

# start the asyncio event loop
asyncio.run(main())

Next, let's develop a client for this server.

Asyncio Client

We can develop a simple client to connect to the server.

The client will open a connection and sleep for 10 seconds before closing the connection.

The complete example is listed below.

# SuperFastPython.com
# client for our test server
import asyncio

# test client
async def main():
    print('Connecting...')
    # open a connection to the server
    reader, writer = await asyncio.open_connection('127.0.0.1', 8888)
    # report message
    print('Connected')
    # wait around
    await asyncio.sleep(10)
    # close the connection
    print('Shutting down...')
    writer.close()
    await writer.wait_closed()

# run the event loop
asyncio.run(main())

Next, let's run both client and server and review an exchange.

Asyncio Client-Server Exchange

We can run the server, connect the client, and review the introspection performed in the server.

Firstly, we can save the server program to a file, such as server.py, and run it on the command line.

For example:

python server.py

Next, we can save the client program to a file, such as client.py, and run it on the command line.

For example:

python client.py

The client connects to the server and the server handles the connection.

The server reports a message, sleeps, then reports the details of all tasks running the event loop.

The server then disconnects the client, and later the client disconnects from the server.

The server can then be manually terminated, such as via the control-c key combination that sends a SIGINT signal to the server program, causing it to exit.

Below is a sample output from the server.

Server is running
Client connected
<Task pending name='Task-1' coro=<main() running at ...:26> wait_for=<Future pending cb=[Task.task_wakeup()]> cb=[_run_until_complete_cb() at .../asyncio/base_events.py:180]>
<Task pending name='Task-4' coro=<handler() running at ...:13>>

Below is a sample output from the client.

Connecting...
Connected
Shutting down...

We can see that within the server asyncio event loop there are two tasks running.

The first is running the main() coroutine and accepting client connections.

The second is the task running the single client that is connected, executing our handler() coroutine.

This highlights that client connections executing the client callback handler are executed in separate asyncio tasks within the server.

Next, let's explore the implications of this, starting with an exception raised in the client callback and its effect on the server task.

Example of Exception in Client Callback

We can explore the effect on the server if an unhandled exception is raised within a task handling a client connection.

The expectation is that the task is isolated and that it will not affect the server.

Let's dive in.

Asyncio Server

We can update the server to raise an unhandled exception.

...
# fail with an unhandled exception
raise Exception('Something bad happened')

The expectation is that this will terminate the task running the client connection.

The updated handler() coroutine with this change is listed below.

# handler for client connections
async def handler(reader, writer):
    # report progress
    print('Client connected')
    # await a moment
    await asyncio.sleep(2)
    # fail with an unhandled exception
    raise Exception('Something bad happened')
    # a line that is never reached
    print('Never reached')

Tying this together, the updated server program is listed below.

# SuperFastPython.com
# example of unhandled exception in client handler
import asyncio

# handler for client connections
async def handler(reader, writer):
    # report progress
    print('Client connected')
    # await a moment
    await asyncio.sleep(2)
    # fail with an unhandled exception
    raise Exception('Something bad happened')
    # a line that is never reached
    print('Never reached')

# main coroutine
async def main():
    # create the server and accept clients forever
    async with (server:= await asyncio.start_server(handler, '127.0.0.1', 8888)):
        # report progress
        print('Server is running')
        # accept client connections forever
        await server.serve_forever()

# start the asyncio event loop
asyncio.run(main())

Next, let's look at the client program.

Asyncio Client

The client program does not require any change, it is the same as in the above example.

The complete code listing is provided below.

# SuperFastPython.com
# client for our test server
import asyncio

# test client
async def main():
    print('Connecting...')
    # open a connection to the server
    reader, writer = await asyncio.open_connection('127.0.0.1', 8888)
    # report message
    print('Connected')
    # wait around
    await asyncio.sleep(10)
    # close the connection
    print('Shutting down...')
    writer.close()
    await writer.wait_closed()

# run the event loop
asyncio.run(main())

Next, let's look at an exchange between the client and the server.

Client-Server Exchange

We can run the server, connect the client, and review the output of the server.

Firstly, we can save the server program to a file, such as server.py, and run it on the command line.

For example:

python server.py

Next, we can save the client program to a file, such as client.py, and run it on the command line.

For example:

python client.py

The client connects to the server and the server handles the connection.

The server reports a message, sleeps, and then raises an unhandled exception. This immensely terminates the task running the client connection. Later the client stops blocking and disconnects.

We can then connect another client to the server and repeat the process.

The server can then be manually terminated, such as via the control-c key combination that sends a SIGINT signal to the server program, causing it to exit.

Below is a sample output from the server with a number of clients connecting over time.

Server is running
Client connected
Client connected
Client connected
...

Below is a sample output from a client.

Connecting...
Connected
Shutting down...

This highlights that although an unhandled exception can be raised within the client callback, it does not affect the server. The asyncio server is able to continue to accept connections from new clients.

This further highlights the isolated nature of tasks running client connections from the server task.

Takeaways

You now know about the client callback coroutine used by the asyncio server to manage connections.

If you enjoyed this tutorial, you will love my book: Python Asyncio Jump-Start. It covers everything you need to master the topic with hands-on examples and clear explanations.