ConnectionResetError: Cannot write to closing transport

[soyasoya5]

Python version: Python 3.8.0

Sample Code

import aiohttp
import asyncio
from aiohttp_socks import ProxyConnector

class Example:
    async def create_session(self):
        connector = ProxyConnector.from_url('socks4://13.0.0.2:1080')
        self.s = aiohttp.ClientSession(connector=connector)

    async def close_session(self):
        await self.s.close()

    async def send_request(self):
        async with self.s.get('https://google.com') as r:
            print(r.status)

async def task_helper(example, s_time):
    await asyncio.sleep(s_time)
    await example.send_request()

async def main():
    example = Example()
    try:
        await example.create_session()
        tasks = [asyncio.create_task(task_helper(example, i)) for i in range(0,15,3)]
        await asyncio.gather(*tasks)
    finally:
        await example.close_session()

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

Error

An open stream object is being garbage collected; call "stream.close()" explicitly.
Traceback (most recent call last):
  File "/home/user/pyvenv/aiohttp/lib/python3.8/site-packages/aiohttp/client.py", line 542, in _request
    resp = await req.send(conn)
  File "/home/user/pyvenv/aiohttp/lib/python3.8/site-packages/aiohttp/client_reqrep.py", line 668, in send
    await writer.write_headers(status_line, self.headers)
  File "/home/user/pyvenv/aiohttp/lib/python3.8/site-packages/aiohttp/http_writer.py", line 119, in write_headers
    self._write(buf)
  File "/home/user/pyvenv/aiohttp/lib/python3.8/site-packages/aiohttp/http_writer.py", line 67, in _write
    raise ConnectionResetError("Cannot write to closing transport")
ConnectionResetError: Cannot write to closing transport

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "test.py", line 31, in <module>
    asyncio.run(main())
  File "/usr/lib/python3.8/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "/usr/lib/python3.8/asyncio/base_events.py", line 608, in run_until_complete
    return future.result()
  File "test.py", line 26, in main
    await asyncio.gather(*tasks)
  File "test.py", line 19, in task_helper
    await example.send_request()
  File "test.py", line 14, in send_request
    async with self.s.get('https://google.com') as r:
  File "/home/user/pyvenv/aiohttp/lib/python3.8/site-packages/aiohttp/client.py", line 1117, in __aenter__
    self._resp = await self._coro
  File "/home/user/pyvenv/aiohttp/lib/python3.8/site-packages/aiohttp/client.py", line 554, in _request
    raise ClientOSError(*exc.args) from exc
aiohttp.client_exceptions.ClientOSError: Cannot write to closing transport

I believe the socks proxy works, as the same async code works using httpx, and synchronous code works using requests-socks

[romis2012]

Your code works perfectly with alive socks. Try using another proxy

[romis2012]

Another possible reason is that google added your socks to the blacklist. Try another target resource too

[soyasoya5]

I'm getting the error above no matter what proxy I use, the proxies are verified to be working by using curl --proxy
I'm running Ubuntu 18.04.1 LTS

Here's my pip freeze output if that helps

aiohttp==3.7.4.post0
aiohttp-retry==2.4.6
aiohttp-socks==0.6.0
anyio==3.3.2
async-timeout==3.0.1
attrs==21.2.0
certifi==2021.5.30
chardet==4.0.0
charset-normalizer==2.0.6
h11==0.12.0
httpcore==0.13.7
httpx==0.19.0
httpx-socks==0.4.1
idna==3.2
multidict==5.1.0
pkg-resources==0.0.0
PySocks==1.7.1
python-dotenv==0.19.0
python-socks==1.2.4
redis==3.5.3
requests==2.26.0
rfc3986==1.5.0
sniffio==1.2.0
typing-extensions==3.10.0.2
urllib3==1.26.7
uvloop==0.16.0
yarl==1.6.3

[soyasoya5]

Logs from proxy server when running the script above

danted[198997]: info: pass(1): tcp/accept [: 13.0.0.x.57038 13.0.0.2.1080
danted[195200]: info: pass(1): tcp/accept ]: 6660 -> 13.0.0.x.57038 13.0.0.2.1080 -> 597: local client closed.  Session duration: 0s

[romis2012]

I tested your code with different types of free proxies from here http://free-proxy.cz. This works perfectly for me. Please provide the simplest code sample that fails with any proxy.

[soyasoya5]

I'm using the exact same example code, the same error is returned from any proxy. Any ideas on how to proceed from here?

[romis2012]

I have no idea. Your code works correctly for me with any proxy.

[iosakurov]

У меня такая же проблема, прокси рабочие, но вот конкретно через эту библиотеку не работают. Хз куда копать

[khoben]

This error occurs when accessing some sites via proxyconnector (http, socks5), but if specify http proxy directly in the request all is OK.

[Rongronggg9]

@khoben Are you using CPython 3.11.5? It seems to be a regression in 3.11.5.

import asyncio
import aiohttp
from aiohttp_socks import ProxyConnector


async def fetch(url):
    connector = ProxyConnector.from_url('socks5://127.0.0.1:1080')

    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get(url) as response:
            return f'{response.status} {response.reason}'


if __name__ == '__main__':
    print(asyncio.run(fetch('http://example.com')))

$ docker run --rm -v $PWD/test.py:/test.py --network host python:3.11.4-slim sh -c 'pip install -qq aiohttp_socks; python test.py'
200 OK
$ docker run --rm -v $PWD/test.py:/test.py --network host python:3.11.5-slim sh -c 'pip install -qq aiohttp_socks; python test.py'
Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/aiohttp/client.py", line 558, in _request
    resp = await req.send(conn)
           ^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/aiohttp/client_reqrep.py", line 670, in send
    await writer.write_headers(status_line, self.headers)
  File "/usr/local/lib/python3.11/site-packages/aiohttp/http_writer.py", line 130, in write_headers
    self._write(buf)
  File "/usr/local/lib/python3.11/site-packages/aiohttp/http_writer.py", line 75, in _write
    raise ConnectionResetError("Cannot write to closing transport")
ConnectionResetError: Cannot write to closing transport

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "//test.py", line 15, in <module>
    print(asyncio.run(fetch('http://example.com')))
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/asyncio/runners.py", line 190, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/asyncio/base_events.py", line 653, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "//test.py", line 10, in fetch
    async with session.get(url) as response:
  File "/usr/local/lib/python3.11/site-packages/aiohttp/client.py", line 1141, in __aenter__
    self._resp = await self._coro
                 ^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/aiohttp/client.py", line 572, in _request
    raise ClientOSError(*exc.args) from exc
aiohttp.client_exceptions.ClientOSError: Cannot write to closing transport

Not sure if python/cpython#107913 has fixed the regression. I've not done any tests yet.

@romis2012 Could you reopen the issue until the next CPython release? So that others could find it easily. Or I can open a new one if you'd prefer.

[Rongronggg9]

Not sure if python/cpython#107913 has fixed the regression. I've not done any tests yet.

Until the latest commit on the 3.11 branch (python/cpython@79f7a4c), the regression persists.

Before we can report the issue to CPython, we need to figure out a minimal reproducer, @romis2012, could you help?

[romis2012]

Before we can report the issue to CPython, we need to figure out a minimal reproducer, @romis2012, could you help?

I can't reproduce this issue, as I wrote above

[khoben]

@khoben Are you using CPython 3.11.5? It seems to be a regression in 3.11.5.

Yes, it looks like it broke after August 26 when the docker container with the base image python:3.11-slim-bullseye was rebuilt and updated to 3.11.5.

[Rongronggg9]

I can't reproduce this issue, as I wrote above

Even on CPython 3.11.5? That's quite weird. I've reproduced the issue on both my PC (Debian unstable) and my VPS (Debian bookworm), with both my own socks5 proxy and proxies from http://free-proxy.cz/.

The issue is only reproducible on CPython 3.11.5, 3.11.0~3.11.4 work fine (test.py: #27 (comment)):

$ for patch in {0..5}; do docker run --rm -v $PWD/test.py:/test.py --network host python:3.11.${patch}-slim sh -c 'pip install -qq aiohttp_socks; python test.py'; done
200 OK
200 OK
200 OK
200 OK
200 OK
Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/aiohttp/client.py", line 558, in _request
    resp = await req.send(conn)
           ^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/aiohttp/client_reqrep.py", line 670, in send
    await writer.write_headers(status_line, self.headers)
  File "/usr/local/lib/python3.11/site-packages/aiohttp/http_writer.py", line 130, in write_headers
    self._write(buf)
  File "/usr/local/lib/python3.11/site-packages/aiohttp/http_writer.py", line 75, in _write
    raise ConnectionResetError("Cannot write to closing transport")
ConnectionResetError: Cannot write to closing transport

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "//test.py", line 15, in <module>
    print(asyncio.run(fetch('http://example.com')))
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/asyncio/runners.py", line 190, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/asyncio/base_events.py", line 653, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "//test.py", line 10, in fetch
    async with session.get(url) as response:
  File "/usr/local/lib/python3.11/site-packages/aiohttp/client.py", line 1141, in __aenter__
    self._resp = await self._coro
                 ^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/aiohttp/client.py", line 572, in _request
    raise ClientOSError(*exc.args) from exc
aiohttp.client_exceptions.ClientOSError: Cannot write to closing transport

[romis2012]

Even on CPython 3.11.5

Oh, yes, on Python 3.11.5 I have the same error

[romis2012]

Fixed in v0.8.1

[Rongronggg9]

The fix in v0.8.1 may lead to a memory leak.

• The lifecycle of a session can be very long. And it will reuse the same connector before it gets closed.
  • Or, aiohttp permits reusing the same connector among sessions: https://docs.aiohttp.org/en/stable/client_advanced.html#connectors
• If the connection to the proxy server is lost somehow, it will be recreated automatically. In such a case _wrap_create_connection() will be called again.
• Then (Chain)ProxyConnector will retain all streams ever been created, even if they were lost.

import asyncio
import aiohttp
from aiohttp_socks import ProxyConnector


async def main():
    connector = ProxyConnector.from_url('socks5://127.0.0.1:1080')
    async with aiohttp.ClientSession(connector=connector) as session:
        for i in range(10):
            if connector._streams:
                # the connection to the proxy server is lost somehow
                # here we simulate the situation by manually closing the writer
                connector._streams[-1].writer.close()
            try:
                async with session.get('http://example.com') as response:
                    print(response.status)
            except Exception as e:
                print(e)

        print(len(connector._streams))  # 10
    print(len(connector._streams))  # 10


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

[Rongronggg9]

The fix in e1541cc is somehow infectious... During the lifecycle of a proxied session, the fix in python/cpython#107836 is discarded, making TLS connections leak again. Moreover, the connection to the proxy server is also leaked for about 30 seconds (maybe it is just closed by my proxy server instead of being garbage collected?). Both leakages were not observed in v0.8.1.

(The script is partly taken from python/cpython#106684, thx)

import os
import asyncio
import gc
import signal
import aiohttp
from aiohttp_socks import ProxyConnector

HOST = "cloudflare.com"  # will keep the connection alive for a few minutes at least
PROXY = 'socks5://127.0.0.1:1080'

BUF = ''
TIMES = 0


async def query():
    await asyncio.sleep(2)  # wait for socks()
    reader, writer = await asyncio.open_connection(HOST, 443, ssl=True)

    # No connection: close, remote side will keep the connection open
    writer.write(f"GET / HTTP/1.1\r\nHost: {HOST}\r\n\r\n".encode())
    await writer.drain()

    # only read the first header line
    try:
        return (await reader.readline()).decode()
    finally:
        # closing the writer will properly finalize the connection
        # writer.close()
        pass

    # reader and writer are now unreachable


async def socks():
    async with aiohttp.ClientSession(connector=ProxyConnector.from_url(PROXY)) as session:
        async with session.get(f'https://{HOST}') as response:
            # simulate a session of long lifecycle
            # StreamWriter.__del__() will keep unavailable during the period
            await asyncio.sleep(5)
            return response.status


def summarize():
    global BUF, TIMES
    if TIMES and BUF:
        print(f'... the above {len(BUF.splitlines())} line(s) were repeated {TIMES} time(s) ...')
        TIMES, BUF = 0, ''


async def lsof():
    global BUF, TIMES
    proc = await asyncio.create_subprocess_shell(f"lsof -np {os.getpid()} | grep TCP", stdout=asyncio.subprocess.PIPE)
    buf = (await proc.stdout.read()).decode().strip()
    if not buf:
        return
    if buf != BUF:
        summarize()
        print(buf)
        BUF = buf
    else:
        TIMES += 1


async def amain():
    await asyncio.gather(query(), socks())

    # The _SSLProtocolTransport object is kept in memory and the
    # connection won't be released until the remote side closes the connection
    for _ in range(200):
        # Just be sure everything is freed, just in case
        gc.collect()
        await asyncio.gather(asyncio.sleep(1), lsof())
    summarize()


def main():
    print(f"PID {os.getpid()}")
    task = asyncio.ensure_future(amain())

    loop = asyncio.get_event_loop()
    loop.add_signal_handler(signal.SIGTERM, task.cancel)
    loop.add_signal_handler(signal.SIGINT, task.cancel)
    loop.run_until_complete(task)


if __name__ == "__main__":
    main()

$ python3 test.py # v0.8.1
PID ****
$ python3 test.py # v0.8.2
PID ****
python  **** ****    6u     IPv4     ****      0t0     TCP 127.0.0.1:****->127.0.0.1:1080 (ESTABLISHED)
python  **** ****    8u     IPv6     ****      0t0     TCP [****]:****->[****]:https (ESTABLISHED)
... the above 2 line(s) were repeated 29 time(s) ...
python  **** ****    8u     IPv6     ****      0t0     TCP [****]:****->[****]:https (ESTABLISHED)
... the above 1 line(s) were repeated 169 time(s) ...

[romis2012]

You just have to close the writer manually...

In any case, closing the connection in the StreamWriter's __del__ method is not a good idea and will lead to problems in many projects

You can continue experimenting with version 0.8.3

[Rongronggg9]

You just have to close the writer manually...

Yes, CPython will raise a resource warning in 3.13 (python/cpython#107650). I just meant monkey-patching the stdlib is infectious...

In any case, closing the connection in the StreamWriter's del method is not a good idea and will lead to problems in many projects

I agree that more consideration is needed before backporting the fix to CPython 3.11. That's why I suggested reporting it to CPython. The leakage of connection, however, is a problem that can be minor or significant. It is hard to tell whether not breaking existing projects is more important than having the issue fixed. At least it is not a bad story since it helped us to find a connection leakage in aiohttp_socks.

You can continue experimenting with version 0.8.3

The infectivity has gone, but the connection to the proxy server is still leaked.

PID ****
python  **** ****    6u     IPv4     ****      0t0     TCP 127.0.0.1:****->127.0.0.1:1080 (ESTABLISHED)
... the above 1 line(s) were repeated 29 time(s) ...

I've also tried to run my script with CPython 3.11.4 and aiohttp_socks 0.8.0, which shows that the leakage of proxy connections is an existing-for-long issue. If a leakage was found, we'd better fix it. 05f5228 is a nice fix except that stream(s) are stored in a list. I am not pretty familiar with network stuff, what if just store a single stream as a trivial instance attribute? I suppose that, when _wrap_create_connection() is called twice or more, previous stream(s) (and their transports) should have been lost or closed, or else why would it be called again?

[romis2012]

Yes, CPython will raise a resource warning in 3.13 (python/cpython#107650)

And what? We only fix our "internal" writers, this will not affect the behavior of other writers in any way.

The infectivity has gone, but the connection to the proxy server is still leaked

Nothing leaks anywhere. Check your test code. In version 0.8.3 the connector behavior is completely equivalent to version 0.8.0 on Python<3.11.5

[romis2012]

I've also tried to run my script with CPython 3.11.4 and aiohttp_socks 0.8.0, which shows that the leakage of proxy connections is an existing-for-long issue

If such the issue exists, then this is an aiohttp issue, not an aiohttp-socks.
aiohttp itself should close the transport passed to it

[romis2012]

Before we can report the issue to CPython, we need to figure out a minimal reproducer, @romis2012, could you help?

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()
    # on Python 3.11.5 transport is already closed here

    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())

The code above works fine on Python < 3.11.5 but fails on 3.11.5

Traceback (most recent call last):
  File "/home/roman/projects/python/python-socks/usage_issue_27_reproducer.py", line 34, 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/roman/projects/python/python-socks/usage_issue_27_reproducer.py", line 27, 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

[romis2012]

05f5228 is a nice fix except that stream(s) are stored in a list. I am not pretty familiar with network stuff, what if just store a single stream as a trivial instance attribute? I suppose that, when _wrap_create_connection() is called twice or more, previous stream(s) (and their transports) should have been lost or closed, or else why would it be called again?

Just consider a use case like this:

async def fetch(session, url):
    async with session.get(url) as r:
        return await r.text()

connector = ProxyConnector.from_url(PROXY_URL)
async with ClientSession(connector=connector) as s:
    tasks = [fetch(s, 'https://google.com/'), fetch(s, 'https://check-host.net/ip')]
    result = await asyncio.gather(*tasks)
    print(result)