asyncio: ConnectionResetError after StreamWriter is garbage collected

[rigens]

Bug report

Bug description:

The code below works perfectly on Python 3.11.4 but fails with ConnectionResetError on Python 3.11.5

import asyncio

HOST = 'ifconfig.me'
PORT = 80


async def connect() -> asyncio.Transport:
    reader, writer = await asyncio.open_connection(
        host=HOST,
        port=PORT,
    )
    return writer.transport  # type: ignore


async def fetch():
    loop = asyncio.get_running_loop()

    transport = await connect()
    # transport is already closed here on 3.11.5

    reader = asyncio.StreamReader(limit=2**16, loop=loop)
    protocol = asyncio.StreamReaderProtocol(reader, loop=loop)

    transport.set_protocol(protocol)
    loop.call_soon(protocol.connection_made, transport)
    loop.call_soon(transport.resume_reading)

    writer = asyncio.StreamWriter(
        transport=transport,
        protocol=protocol,
        reader=reader,
        loop=loop,
    )

    request = f'GET /ip HTTP/1.1\r\nHost: {HOST}\r\nConnection: close\r\n\r\n'.encode()
    writer.write(request)
    await writer.drain()

    response = await reader.read(-1)
    print(response)

    writer.close()


if __name__ == '__main__':
    asyncio.run(fetch())

Traceback (most recent call last):
  File "/home/rigens/aio_test1.py", line 50, in <module>
    asyncio.run(fetch())
  File "/usr/local/miniconda/envs/py311_5/lib/python3.11/asyncio/runners.py", line 190, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "/usr/local/miniconda/envs/py311_5/lib/python3.11/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/miniconda/envs/py311_5/lib/python3.11/asyncio/base_events.py", line 653, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "/home/rigens/aio_test1.py", line 41, in fetch
    await writer.drain()
  File "/usr/local/miniconda/envs/py311_5/lib/python3.11/asyncio/streams.py", line 378, in drain
    await self._protocol._drain_helper()
  File "/usr/local/miniconda/envs/py311_5/lib/python3.11/asyncio/streams.py", line 167, in _drain_helper
    raise ConnectionResetError('Connection lost')
ConnectionResetError: Connection lost

CPython versions tested on:

3.11

Operating systems tested on:

Linux

[kumaraditya303]

It is not recommended to manually instantiate StreamWriter ¹, StreamReader² objects and transports, so this is not a bug of asyncio.

Footnotes

1. https://docs.python.org/3/library/asyncio-stream.html#asyncio.StreamWriter ↩

2. https://docs.python.org/3/library/asyncio-stream.html#asyncio.StreamReader ↩

[kumaraditya303]

If you want to just use the transport then you should loop.create_connection API and not streaming APIs.

[rigens]

It is not recommended to manually instantiate StreamWriter 1, StreamReader...

It's not about manually instantiating the StreamWriter....
The code above is given only as an example

If you want to just use the transport then you should loop.create_connection API and not streaming APIs.

I want to reuse the transport received from some third-party component that uses the streams api (StreamReader, StreamWriter) to communicate with the remote host.

This approach worked fine until version 3.11.5 and this PR #107656

What was the purpose of closing the transport in the StreamWriter __del__ method, why not do it in the __del__ method of the transport itself?

This PR has already broken some libraries: romis2012/aiohttp-socks#27 (comment)

The mentioned PR at least breaks backward compatibility

[rigens]

@kumaraditya303, if you don't have any arguments, please reopen the issue.

[puddly]

The current asyncio API provides no way to attach StreamReader and StreamWriter to a transport other than UNIX sockets and TCP. I'm trying use a line-based API with a SerialTransport (wrapping a serial port).

#107656 unexpectedly closing the transport on __del__ breaks instances where set_protocol was used:

# Transport is already open
transport = ...

loop = asyncio.get_running_loop()
reader = asyncio.StreamReader(limit=65536, loop=loop)
protocol = asyncio.StreamReaderProtocol(reader, loop=loop)
writer = asyncio.StreamWriter(transport, protocol, reader, loop)

transport.set_protocol(protocol)

# Do line-based operations using `reader` and `writer`
...

# Change protocols
transport.set_protocol(new_protocol)

There is no way to detach the transport without digging into _transport and providing it a mock transport to "close":

class MockTransport:
    def close(self):
        pass

writer._transport = MockTransport()
del writer

It would be nice to have an API to clear writer._transport without closing.

[gvanrossum]

@puddly

It would be nice to have an API to clear writer._transport without closing.

If that's your main point, could you open a new issue please? While I'm not sure that this issue should remain closed, I don't think it's the place to discuss new feature requests.

[gvanrossum]

@rigens

This approach worked fine until version 3.11.5 and this PR #107656

The __del__ method got some improvements in #109538.

Is it still preventing you to do what you want to?

[gvanrossum]

@rigens

What was the purpose of closing the transport in the StreamWriter __del__ method, why not do it in the __del__ method of the transport itself?

AFAICT the reason is because the transport is not under our control -- there are many possible transports and we don't want to add a __del__ method to all of them.

FWIW: The asyncio streams implementation is not very flexible. There are a variety of design aspects that we wish we had done differently. But it's about a decade too late. We've contemplated deprecating it, but didn't have the energy to design and implement something better to replace it. But it looks like you are trying to do something unusual with it, and it's just hard to support that. The changes that broke your code, after all, were put in place to fix someone else's code. Alas.