aiohttp ^C hangs when psycopg connection pool created

[thanatos]

Describe the bug

If I create a psycopg_pool.AsyncConnectionPool object, aiohttp's Ctrl+C handler will hang while trying to handle the ^C.

To Reproduce

The following minimal example reproduces the bug:

import logging

from aiohttp import web
import psycopg_pool


async def hello(request):
    return web.Response(text="Hello, world")


def main():
    logging.basicConfig(level=logging.DEBUG)

    app = web.Application()
    app.add_routes([
        web.get('/', hello),
    ])
    app.cleanup_ctx.append(db_pool)

    web.run_app(app)


async def db_pool(app):
    kwargs = {
        'hostname': 'localhost',
        'port': 5432,
    }
    pool = psycopg_pool.AsyncConnectionPool(
        kwargs=kwargs,
        open=False,
        min_size=0,
        max_size=4,
    )
    print('Pool opening.')
    await pool.open()
    print('Pool open.')
    #app['db_driver'] = db_driver
    try:
        yield
    finally:
        print('Cleanup ctx done.')
    await pool.close()


if __name__ == '__main__':
    main()

Run it, hit ^C:

» venv2/bin/python3 aiohttp_bug.py
DEBUG:asyncio:Using selector: EpollSelector
Pool opening.
Pool open.
======== Running on http://0.0.0.0:8080 ========
(Press CTRL+C to quit)
^C <hangs here for ~60s>

The hang is only after we create the pool. If we don't create the pool, no hang.

The reason we hang is this bit:

aiohttp/aiohttp/web.py

Lines 313 to 318 in 04b1212

	async def _wait(exclude: "WeakSet[asyncio.Task[object]]") -> None:
	t = asyncio.current_task()
	assert t is not None
	exclude.add(t)
	while tasks := asyncio.all_tasks().difference(exclude):
	await asyncio.wait(tasks)

Here, aiohttp attempts to wait on tasks to finish?

There's slightly more context here:

aiohttp/aiohttp/web.py

Lines 338 to 343 in 04b1212

	# On shutdown we want to avoid waiting on tasks which run forever.
	# It's very likely that all tasks which run forever will have been created by
	# the time we have completed the application startup (in runner.setup()),
	# so we just record all running tasks here and exclude them later.
	starting_tasks: "WeakSet[asyncio.Task[object]]" = WeakSet(asyncio.all_tasks())
	runner.shutdown_callback = partial(wait, starting_tasks, shutdown_timeout)

It's very likely that all tasks which run forever will have been created by the time we have completed the application startup

This just seems incorrect? In particular, psycopg_pool has a background task that is taking care of various async things that it appears to queue up for that task. That task has an event, which I presume is triggered when the queue has something new.

Its code looks like this:

    async def run(self) -> None:
        """Execute the events scheduled."""
        q = self._queue
        while True:
            async with self._lock:
                now = monotonic()
                task = q[0] if q else None
                if task:
                    if task.time <= now:
                        heappop(q)
                    else:
                        delay = task.time - now
                        task = None
                else:
                    delay = self.EMPTY_QUEUE_TIMEOUT
                self._event.clear()

            if task:
                if not task.action:
                    break
                try:
                    await task.action()
                except Exception as e:
                    logger.warning(
                        "scheduled task run %s failed: %s: %s",
                        task.action,
                        e.__class__.__name__,
                        e,
                    )
            else:
                # Block for the expected timeout or until a new task scheduled
                await self._event.wait_timeout(delay)

That last await self._event.wait_timeout(delay) is what I am talking about here. wait_timeout, when called, calls asyncio.wait_for, which creates a task.

This is done every time we hit the lock, and it happens after the app has started, too, so this event wait's task ends up in the tasks set that aiohttp waits on.

Note that aiohttp's wait on all other running tasks to finish occurs before we run the application cleanups, so the pool is still alive — i.e., we're not first giving the cleanup context an opportunity to close the pool.

Expected behavior

^C wraps up without hanging (for 60s).

Logs/tracebacks

(see above)

Python Version

$ python --version
Python 3.11.5

aiohttp Version

$ python -m pip show aiohttp
Name: aiohttp
Version: 3.9.5
Summary: Async http client/server framework (asyncio)
Home-page: https://github.com/aio-libs/aiohttp
Author: 
Author-email: 
License: Apache 2
Location: /home/royiv/code/groceries/venv2/lib/python3.11/site-packages
Requires: aiosignal, attrs, frozenlist, multidict, yarl
Required-by:

multidict Version

$ python -m pip show multidict
Name: multidict
Version: 6.0.5
Summary: multidict implementation
Home-page: https://github.com/aio-libs/multidict
Author: Andrew Svetlov
Author-email: andrew.svetlov@gmail.com
License: Apache 2
Location: /home/royiv/code/groceries/venv2/lib/python3.11/site-packages
Requires: 
Required-by: aiohttp, yarl

yarl Version

$ python -m pip show yarl
Name: yarl
Version: 1.9.4
Summary: Yet another URL library
Home-page: https://github.com/aio-libs/yarl
Author: Andrew Svetlov
Author-email: andrew.svetlov@gmail.com
License: Apache-2.0
Location: /home/royiv/code/groceries/venv2/lib/python3.11/site-packages
Requires: idna, multidict
Required-by: aiohttp

OS

Arch Linux

Related component

Server

Additional context

Name: psycopg-pool
Version: 3.2.1
Summary: Connection Pool for Psycopg
Home-page: https://psycopg.org/psycopg3/
Author: Daniele Varrazzo
Author-email: daniele.varrazzo@gmail.com
License: GNU Lesser General Public License v3 (LGPLv3)
Location: /home/royiv/code/groceries/venv2/lib/python3.11/site-packages
Requires: typing-extensions
Required-by: 

Code of Conduct

• I agree to follow the aio-libs Code of Conduct

[Dreamsorcerer]

Yeah, that one seems a bit awkward. Will have to relook and figure out if there's a better solution.