Cannot cleanly shut down an asyncio based server

[CendioOssman]

Bug report

Bug description:

When writing an asyncio based service, you basically have this sequence:

1. Create an event loop
2. Register a SIGTERM handler
3. Start your server
4. loop.run_forever()
5. SIGTERM causes a loop.stop()
6. Close the server
7. Close event loop

If there are any connections active at this point, then they don't get discarded until interpreter shutdown, with the result that you get a bunch of ResourceWarnings (and cleanup code might not run).

It would be very useful if there was a Server.close_clients() or something like that. Even a Server.all_transports() would be useful, as then you could do something similar as when doing a Task.cancel() on what you get from loop.all_tasks().

We could poke at Server._transports, but that is something internal that might change in the future.

There is Server.wait_closed(), but that hangs until all clients have gracefully disconnected. It doesn't help us when we want to shut down the service now.

CPython versions tested on:

3.12

Operating systems tested on:

Linux

Linked PRs

• gh-113538: Don't crash in stream reader protocol callback when task is cancelled #113690
• [3.12] gh-113538: Don't error in stream reader protocol callback when task is cancelled (GH-113690) #113713
• [3.11] gh-113538: Don't error in stream reader protocol callback when task is cancelled (GH-113690) #113714
• gh-113538: Allow client connections to be closed #114432
• gh-113538: Revert "gh-113538: Add asycio.Server.{close,abort}_clients (#114432)" #116632
• gh-113538: Allow client connections to be closed #116784

[CendioOssman]

This might be a symptom of a more general issue, that there is a need for cancelling everything that's dependent on the event loop we are shutting down.

Currently, it is possible to cancel tasks, async generators, and executors. The issue here is (sort-of) about shutting down pending file descriptors. But it might also be necessary to be able to cancel pending futures, and possibly more things?

[gvanrossum]

I've always assumed that it is up to the application to keep track of connections (etc.) in a form that is suitable for shutting down. After all in a typical application there's more to clean up than the client connections -- there may be database connections, and who know what else to handle.

I would definitely not mess with pending futures -- those are sufficiently low-level that you cannot know who is waiting for them and why, and in which order they should be cancelled.

The key point of control you have is probably cancelling tasks -- tasks can do cleanup in response to a cancellation request, and if your tasks form a tree (or even a DAG), they will be cancelled recursively (if task t1 is waiting for task t2, cancelling t1 first cancels t2). So perhaps your application should keep track of the main task per connection?

[gvanrossum]

Another thought is that this might be something that a higher-level framework or library could offer. E.g. gunicorn or aio-http? (Note that I've never used either, so I may be way off base here.)

[CendioOssman]

I've always assumed that it is up to the application to keep track of connections (etc.) in a form that is suitable for shutting down. After all in a typical application there's more to clean up than the client connections -- there may be database connections, and who know what else to handle.

asyncio unfortunately makes this difficult, as it is the one in control of creating the connections. It is done by the Server class, which you cannot override or extend. At best, you can try to track things in the protocol factory callback. But that requires you to also have handling in each protocol class, which is messy and fragile.

Clean up should hopefully be possible using the normal signals from the transports, i.e. connection_lost().

I would definitely not mess with pending futures -- those are sufficiently low-level that you cannot know who is waiting for them and why, and in which order they should be cancelled.

You are probably right. But it does make me uneasy that there isn't a way to centrally force a clean-up on shutdown. If you have to build your own everywhere, then it's easy to make a mistake and overlook something.

The key point of control you have is probably cancelling tasks -- tasks can do cleanup in response to a cancellation request, and if your tasks form a tree (or even a DAG), they will be cancelled recursively (if task t1 is waiting for task t2, cancelling t1 first cancels t2). So perhaps your application should keep track of the main task per connection?

Unfortunately, there isn't always a task involved. Sometimes there is just a protocol, or a construct using futures and callbacks.

Another thought is that this might be something that a higher-level framework or library could offer. E.g. gunicorn or aio-http? (Note that I've never used either, so I may be way off base here.)

Maybe. Those two specific projects are not sufficient, though, as they are just for HTTP.

[gvanrossum]

Okay, let's dive into the original problem a bit more. I think I would like to see a "scaled-down" example of a realistic server that we might be able to use to demonstrate the problem, reason about it, and evaluate potential solutions. It should be small enough that I can understand the code in 5-10 minutes (my typical attention span :-). Would you be willing to code up a first attempt at such an example? We can iterate on it together.

Regarding Futures, these are used all over the place (internally in asyncio) with specific semantics, sometimes involving cancellation. If we were to start cancelling Futures automatically when a loop is stopped that could have all sorts of unexpected consequences -- probably worse than leaving them alone. So let's declare this out of scope (or if it really bugs you, please open a separate issue, as it has nothing to do with closing servers).

[CendioOssman]

This should be a rough mockup of the situation:

#!/usr/bin/python3

import asyncio

class EchoServerProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        peername = transport.get_extra_info('peername')
        print('Connection from {}'.format(peername))
        self.transport = transport

    def data_received(self, data):
        message = data.decode()
        print('Data received: {!r}'.format(message))

        print('Send: {!r}'.format(message))
        self.transport.write(data)

        print('Close the client socket')
        self.transport.close()

def stop():
    loop = asyncio.get_running_loop()
    loop.stop()

def server_up(fut):
    global server
    server = fut.result()

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

fut = asyncio.ensure_future(loop.create_server(
        lambda: EchoServerProtocol(),
        '127.0.0.1', 8888))
fut.add_done_callback(server_up)

# Simulate termination
loop.call_later(2.5, stop)

try:
    loop.run_forever()
finally:
    server.close()
    # Can't run this as it will hang:
    #loop.run_until_complete(server.wait_closed())
    loop.close()

If you run this in dev mode and connect a client, you see the issue:

$ python3.12 -X dev echosrv.py 
Connection from ('127.0.0.1', 56290)
sys:1: ResourceWarning: unclosed <socket.socket fd=7, family=2, type=1, proto=6, laddr=('127.0.0.1', 8888), raddr=('127.0.0.1', 56290)>
ResourceWarning: Enable tracemalloc to get the object allocation traceback
/usr/lib64/python3.12/asyncio/selector_events.py:875: ResourceWarning: unclosed transport <_SelectorSocketTransport fd=7>
  _warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
ResourceWarning: Enable tracemalloc to get the object allocation traceback

[jsundahl]

This wasn't a problem until 3.11.5. Here's a simpler example of what I think is the same root issue:

#!/usr/bin/env python3

import asyncio

async def main():
    async def on_connection(reader, writer):
        while True:
            buf = await reader.read(1024)
            print("got bytes", buf)


    server = await asyncio.start_unix_server(on_connection, path="/tmp/sock")
    cr, cw = await asyncio.open_unix_connection(path="/tmp/sock")
    print("sending hi")
    cw.write(b"hi")
    await cw.drain()

    await asyncio.sleep(0.2)

    server.close()
    await server.wait_closed()


asyncio.run(main())

then using pyenv to set the python version and run it we get the following:

$ echo "3.11.4" > .python-version 
$ ./t.py 
sending hi
got bytes b'hi'

I have yet to look at the underlying code but I'm guessing the server implementation in <=3.11.4 is cancelling the reader.read coroutine and ignoring the cancellation error as I'd expect.

$ echo "3.11.5" > .python-version 
$ ./t.py 
sending hi
got bytes b'hi'
Exception ignored in: <function StreamWriter.__del__ at 0x7fe69d0f2c00>
Traceback (most recent call last):
  File "/home/jsundahl/.pyenv/versions/3.11.5/lib/python3.11/asyncio/streams.py", line 396, in __del__
    self.close()
  File "/home/jsundahl/.pyenv/versions/3.11.5/lib/python3.11/asyncio/streams.py", line 344, in close
    return self._transport.close()
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/jsundahl/.pyenv/versions/3.11.5/lib/python3.11/asyncio/selector_events.py", line 860, in close
    self._loop.call_soon(self._call_connection_lost, None)
  File "/home/jsundahl/.pyenv/versions/3.11.5/lib/python3.11/asyncio/base_events.py", line 761, in call_soon
    self._check_closed()
  File "/home/jsundahl/.pyenv/versions/3.11.5/lib/python3.11/asyncio/base_events.py", line 519, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

skipping 3.11.6, it's identical to 3.11.5.
3.11.7 is slightly different:

$ echo "3.11.7" > .python-version 
$ ./t.py 
sending hi
got bytes b'hi'
Exception in callback StreamReaderProtocol.connection_made.<locals>.callback(<Task cancell...oon/./t.py:6>>) at /home/jsundahl/.pyenv/versions/3.11.7/lib/python3.11/asyncio/streams.py:248
handle: <Handle StreamReaderProtocol.connection_made.<locals>.callback(<Task cancell...oon/./t.py:6>>) at /home/jsundahl/.pyenv/versions/3.11.7/lib/python3.11/asyncio/streams.py:248>
Traceback (most recent call last):
  File "/home/jsundahl/.pyenv/versions/3.11.7/lib/python3.11/asyncio/events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "/home/jsundahl/.pyenv/versions/3.11.7/lib/python3.11/asyncio/streams.py", line 249, in callback
    exc = task.exception()
          ^^^^^^^^^^^^^^^^
  File "/home/jsundahl/scripts/callsoon/./t.py", line 8, in on_connection
    buf = await reader.read(1024)
          ^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/jsundahl/.pyenv/versions/3.11.7/lib/python3.11/asyncio/streams.py", line 708, in read
    await self._wait_for_data('read')
  File "/home/jsundahl/.pyenv/versions/3.11.7/lib/python3.11/asyncio/streams.py", line 540, in _wait_for_data
    await self._waiter
asyncio.exceptions.CancelledError

then 3.12.0 and 3.12.1 hang indefinitely on wait_closed and have the same issue in StreamWriter.__del__ that 3.11.5 has on SIGINT.

[jsundahl]

looks like this is related to #109538

[gvanrossum]

With @CendioOssman's example (connecting a client that doesn't send anything, just hangs) I get different results -- at first I thought I saw what they saw, but then I couldn't repro it again. Weird, let's say that's my fault.

With @jsundahl's example I get exactly what they got; main is the same as 3.12. Let me do some spelunking.

[gvanrossum]

A bit more...

• In 3.10 and before, server.wait_closed() was a no-op, unless you called it before server.close(), in a task (asyncio.create_task(server.wait_closed())). The unclosed connection was just getting abandoned.
• I attempted to fix this in 3.12.0, and backported the fix (I think this ended up in 3.11.5).
• Something was wrong with the fix and an improved fix landed in 3.12.1. I think this is what you have in 3.11.7 (3.11.5 regression: StreamWriter.__del__ fails if event loop is already closed #109538).

Alas, the fix waits until all handlers complete, which in this example is never. Hence the hang. I guess the correct way to write the handler is to put a timeout on the await reader.read(1024) call.

I have to think about this more, but I'm out of time for today. Sorry.

[CendioOssman]

I think the use case for server.wait_closed() is perfectly reasonable for some scenarios. So I don't think we should change that. If you have the option of trusting your clients to finish in a reasonable time, then it works great. What's missing is a method for when you don't trust your clients.

E.g. a clean version of:

    for fd, transport in loop._transports.items():
        transport.abort()
    loop.run_until_complete(server.wait_closed())