Abandoned StreamWriter isn't reliably detected

[CendioOssman]

Bug report

Bug description:

A StreamWriter should be close():d when you are done with it. There is code in the destructor for StreamWriter to detect when this is overlooked and trigger a ResourceWarning. However, the current code often maintains a strong reference to the writer, preventing it from being garbage collected and hence no warning.

Test case:

#!/usr/bin/python3

import asyncio
import gc
import socket

async def handle_echo(reader, writer):
    addr = writer.get_extra_info('peername')
    print(f"Connection from {addr!r}")
    # Forgetting to close the writer
    #writer.close()
    #await writer.wait_closed()

async def main():
    server = await asyncio.start_server(
        handle_echo, '127.0.0.1', 8888)

    addrs = ', '.join(str(sock.getsockname()) for sock in server.sockets)
    print(f'Serving on {addrs}')

    client = socket.create_connection(('127.0.0.1', 8888))
    client.send(b'a' * 64 * 1024)

    async with server:
        for i in range(25):
            await asyncio.sleep(0.1)
            gc.collect()
        print('Exiting')

    print('Done serving')

    client.close()

asyncio.run(main())

Test case locks up waiting for the client connection, when instead I would expect a ResourceWarning and everything exiting nicely.

CPython versions tested on:

3.12, CPython main branch

Operating systems tested on:

Linux

Linked PRs

• gh-114914: Avoid keeping dead StreamWriter alive #115661

[CendioOssman]

The issue is because StreamReaderProtocol maintains a strong reference to the writer. The protocol is in turn referenced by the transport, and the transport from the event loop (because it has called add_reader()).

This chain isn't always active, though, so things sometimes work:

1. If more than 128 KiB of unread data is buffered in the StreamReader, then the transport unregisters from the event loop and the whole group of objects can be garbage collected. This can easily be tested by changing the test case above to send 256 KiB of data instead of 64 KiB.
2. The reference from the protocol to the writer is not always set up. It is only active when either create_server() is used, or start_tls(). So a fairly standard outgoing connection will not be affected.

[CendioOssman]

The reference between the protocol and the writer is never actually used for anything, so my suggestion would be to just remove it:

diff --git a/Lib/asyncio/streams.py b/Lib/asyncio/streams.py
index df58b7a799..82a5daaf50 100644
--- a/Lib/asyncio/streams.py
+++ b/Lib/asyncio/streams.py
@@ -201,7 +201,6 @@ def __init__(self, stream_reader, client_connected_cb=None, loop=None):
             # is established.
             self._strong_reader = stream_reader
         self._reject_connection = False
-        self._stream_writer = None
         self._task = None
         self._transport = None
         self._client_connected_cb = client_connected_cb
@@ -214,10 +213,8 @@ def _stream_reader(self):
             return None
         return self._stream_reader_wr()
 
-    def _replace_writer(self, writer):
+    def _replace_transport(self, transport):
         loop = self._loop
-        transport = writer.transport
-        self._stream_writer = writer
         self._transport = transport
         self._over_ssl = transport.get_extra_info('sslcontext') is not None
 
@@ -239,11 +236,8 @@ def connection_made(self, transport):
             reader.set_transport(transport)
         self._over_ssl = transport.get_extra_info('sslcontext') is not None
         if self._client_connected_cb is not None:
-            self._stream_writer = StreamWriter(transport, self,
-                                               reader,
-                                               self._loop)
-            res = self._client_connected_cb(reader,
-                                            self._stream_writer)
+            writer = StreamWriter(transport, self, reader, self._loop)
+            res = self._client_connected_cb(reader, writer)
             if coroutines.iscoroutine(res):
                 def callback(task):
                     if task.cancelled():
@@ -405,7 +399,7 @@ async def start_tls(self, sslcontext, *,
             ssl_handshake_timeout=ssl_handshake_timeout,
             ssl_shutdown_timeout=ssl_shutdown_timeout)
         self._transport = new_transport
-        protocol._replace_writer(self)
+        protocol._replace_transport(new_transport)
 
     def __del__(self, warnings=warnings):
         if not self._transport.is_closing():

The above change resolves the issue for me here.

[gvanrossum]

Yeah, it does look like that _stream_writer attribute is never used. And it was this way since start_server() was added, in 1540b16. This code is over a decade old, so I am willing to believe that it was never thought through carefully.

Interestingly, the ResourceWarning itself was only added 6 months ago (in gh-107650, and backported to 3.12 and 3.11), with a further refinement a few months later in gh-111983 (also backported).

Also interesting, your example does not hang for me in 3.11! So what happened to streams.py between 3.11 and 3.12 that affects your example? I don't see anything that looks relevant in the diff for streams.py between 3.11 and 3.12, but the rest of asyncio changed extensively. Let me run a git bisect command.

[gvanrossum]

And the winner is... gh-98582, the cursed fix to wait_closed(). :-(

In case you don't recall what happened in that PR, in 3.11 and before, await server.wait_closed() after calling server.close() was a no-op, due to a bug that made it return immediately if the server was closed. This was fixed, but broke code that had unclosed connections. Your example is one of those -- you're using async with server and there's a wait_closed() call in AbstractServer.__aexit__().

Anyway, I think we should just fix this (and probably backport the fix to 3.12 and 3.11) along the lines of your diff. Do you want to submit a PR?

[CendioOssman]

Sure, let me give it a try.