Repository navigation
How to do more than one cleanup task in cancelled state? #772
Description
- asyncpg version: 0.23.0
- PostgreSQL version: 12.6
- Do you use a PostgreSQL SaaS? If so, which? Can you reproduce
the issue with a local PostgreSQL install?: no/ yes - Python version:3.9.5
- Platform: Fedora Linux
- Do you use pgbouncer?: no
- Did you install asyncpg with pip?: yes
- If you built asyncpg locally, which version of Cython did you use?: n/a
- Can the issue be reproduced under both asyncio and
uvloop?: have only tested w/ asyncio directly
We have a user illustrating the case that a task being cancelled will allow us to reach a finally: block where we can at most run only one awaitable cleanup task on asyncpg in order to close out the connection. if we have more than one thing to await, such as emitting a ROLLBACK or anything else, we don't get the chance to close() the connection. It then seems to go into some place where we no longer have any reference to this connection yet asyncpg still leaves it opened; our own GC handlers that are supposed to take care of this are never called.
One way to illustrate it is the use case in such a way that indicates how I'm looking for "how to solve this problem?", of having two separate asycnpg connections that suppose we are doing some kind of work on separately in the same awaitable. if a cancel() is called, I can reach the finally: block, and I can then close at most one of the connections, but not both. in the real case, we are using only one connection but we are trying to emit a ROLLBACK and also do other awaitable things before we get to the .close().
What I dont understand is why gc isn't collecting these connections or why they aren't getting closed.
import asyncio
from asyncio import current_task
import asyncpg
async def get_and_cancel():
c1 = await asyncpg.connect(
user="scott", password="tiger", host="localhost", database="test"
)
c2 = await asyncpg.connect(
user="scott", password="tiger", host="localhost", database="test"
)
try:
r1 = await c1.fetch("SELECT 1")
r2 = await c2.fetch("SELECT 1")
current_task().cancel()
finally:
# we get here...
# this seems to affect the asyncpg connection, the await is
# honored....
await c1.close()
# but we never get here. connection leaks. canonical way to
# solve this issue?
await c2.close()
async def main():
while True:
try:
await get_and_cancel()
except asyncio.exceptions.CancelledError:
pass
asyncio.run(main())the stack trace is that we've run out of connections:
Traceback (most recent call last):
File "/home/classic/dev/sqlalchemy/test4.py", line 39, in <module>
asyncio.run(main())
File "/usr/lib64/python3.9/asyncio/runners.py", line 44, in run
return loop.run_until_complete(main)
File "/usr/lib64/python3.9/asyncio/base_events.py", line 642, in run_until_complete
return future.result()
File "/home/classic/dev/sqlalchemy/test4.py", line 34, in main
await get_and_cancel()
File "/home/classic/dev/sqlalchemy/test4.py", line 11, in get_and_cancel
c2 = await asyncpg.connect(
File "/home/classic/.venv3/lib64/python3.9/site-packages/asyncpg/connection.py", line 1981, in connect
return await connect_utils._connect(
File "/home/classic/.venv3/lib64/python3.9/site-packages/asyncpg/connect_utils.py", line 732, in _connect
con = await _connect_addr(
File "/home/classic/.venv3/lib64/python3.9/site-packages/asyncpg/connect_utils.py", line 632, in _connect_addr
return await __connect_addr(params, timeout, True, *args)
File "/home/classic/.venv3/lib64/python3.9/site-packages/asyncpg/connect_utils.py", line 682, in __connect_addr
await compat.wait_for(connected, timeout=timeout)
File "/home/classic/.venv3/lib64/python3.9/site-packages/asyncpg/compat.py", line 103, in wait_for
return await asyncio.wait_for(fut, timeout)
File "/usr/lib64/python3.9/asyncio/tasks.py", line 481, in wait_for
return fut.result()
asyncpg.exceptions.TooManyConnectionsError: remaining connection slots are reserved for non-replication superuser connections
Activity
zzzeek commentedon Jun 18, 2021
good news magicstack, I found the issue on our end where we were catching an Exception rather than a BaseException, which is as you know where CancelledError derives. /me drives off in his clown car
zzzeek commentedon Jun 18, 2021
although FTR I am still not following yet where "c2" above goes and why it is seemingly not garbage collected.
fantix commentedon Jun 18, 2021
Hi Mike! That is a very interesting issue. I think the reason is that, there is no
awaitbetween the cancellation and the end of thetry:block, therefore the actual cancellation happens during the firstawaitin this task after that cancellation, which is the firstclose()in thefinallyblock, and the firstclose()got cancelled.In short,
current_task().cancel()doesn't raise aCancelledError- the firstawaitafter it does.zzzeek commentedon Jun 18, 2021
@fantix if you run the above program, it runs out of connections. There are like hundreds that are all opened. Where are they ? c2 only points to one of them at a time. does asyncpg have some kind of strong-referencing of connections that aren't closed going on or otherwise some logic that doesn't end the socket when the connection is de-referenced?
fantix commentedon Jun 18, 2021
Ah I see what you mean. Looks like without the cancellation, the connection will be recycled properly. 🤔 Looking into it now
Hmm no, as soon as the the connection is used, it's not recycled without an explicit
close()🤔fantix commentedon Jun 19, 2021
Aha, looks like there is an asyncio timer handle (300s) to clear some statement cache (
_on_entry_expired) that has a callback handle_on_removewhich is a bound method of the connection object - let me try fixing that.Break connection internal circular reference
Break connection internal circular reference (#774)