[PR #9672/afb5ebb backport][3.11] Reuse the oldest keep-alive connection first (#9680)

Author: bdraco

CHANGES/9672.bugfix.rst

@@ -0,0 +1,3 @@
+Fixed the keep-alive connection pool to be FIFO instead of LIFO -- by :user:`bdraco`.
+
+Keep-alive connections are more likely to be reused before they disconnect.

aiohttp/connector.py

@@ -5,7 +5,7 @@
 import sys
 import traceback
 import warnings
-from collections import OrderedDict, defaultdict
+from collections import OrderedDict, defaultdict, deque
 from contextlib import suppress
 from http import HTTPStatus
 from itertools import chain, cycle, islice
@@ -17,6 +17,7 @@
     Awaitable,
     Callable,
     DefaultDict,
+    Deque,
     Dict,
     Iterator,
     List,
@@ -255,7 +256,12 @@ def __init__(
         if loop.get_debug():
             self._source_traceback = traceback.extract_stack(sys._getframe(1))
 
-        self._conns: Dict[ConnectionKey, List[Tuple[ResponseHandler, float]]] = {}
+        # Connection pool of reusable connections.
+        # We use a deque to store connections because it has O(1) popleft()
+        # and O(1) append() operations to implement a FIFO queue.
+        self._conns: DefaultDict[
+            ConnectionKey, Deque[Tuple[ResponseHandler, float]]
+        ] = defaultdict(deque)
         self._limit = limit
         self._limit_per_host = limit_per_host
         self._acquired: Set[ResponseHandler] = set()
@@ -362,10 +368,10 @@ def _cleanup(self) -> None:
         timeout = self._keepalive_timeout
 
         if self._conns:
-            connections = {}
+            connections = defaultdict(deque)
             deadline = now - timeout
             for key, conns in self._conns.items():
-                alive: List[Tuple[ResponseHandler, float]] = []
+                alive: Deque[Tuple[ResponseHandler, float]] = deque()
                 for proto, use_time in conns:
                     if proto.is_connected() and use_time - deadline >= 0:
                         alive.append((proto, use_time))
@@ -590,14 +596,12 @@ async def _get(
 
         The connection will be marked as acquired.
         """
-        try:
-            conns = self._conns[key]
-        except KeyError:
+        if (conns := self._conns.get(key)) is None:
             return None
 
         t1 = monotonic()
         while conns:
-            proto, t0 = conns.pop()
+            proto, t0 = conns.popleft()
             # We will we reuse the connection if its connected and
             # the keepalive timeout has not been exceeded
             if proto.is_connected() and t1 - t0 <= self._keepalive_timeout:
@@ -689,10 +693,7 @@ def _release(
                 self._cleanup_closed_transports.append(transport)
             return
 
-        conns = self._conns.get(key)
-        if conns is None:
-            conns = self._conns[key] = []
-        conns.append((protocol, monotonic()))
+        self._conns[key].append((protocol, monotonic()))
 
         if self._cleanup_handle is None:
             self._cleanup_handle = helpers.weakref_handle(

tests/test_client_session.py

@@ -3,6 +3,7 @@
 import gc
 import io
 import json
+from collections import deque
 from http.cookies import SimpleCookie
 from typing import Any, Awaitable, Callable, List
 from unittest import mock
@@ -32,7 +33,7 @@ async def make_conn():
 
     conn = loop.run_until_complete(make_conn())
     proto = mock.Mock()
-    conn._conns["a"] = [(proto, 123)]
+    conn._conns["a"] = deque([(proto, 123)])
     yield conn
     loop.run_until_complete(conn.close())
 

tests/test_connector.py

@@ -7,9 +7,10 @@
 import ssl
 import sys
 import uuid
+from collections import defaultdict, deque
 from concurrent import futures
 from contextlib import closing, suppress
-from typing import Any, List, Literal, Optional, Sequence, Tuple
+from typing import Any, DefaultDict, Deque, List, Literal, Optional, Sequence, Tuple
 from unittest import mock
 
 import pytest
@@ -194,7 +195,7 @@ async def test_del_with_scheduled_cleanup(loop) -> None:
     loop.set_debug(True)
     conn = aiohttp.BaseConnector(loop=loop, keepalive_timeout=0.01)
     transp = mock.Mock()
-    conn._conns["a"] = [(transp, 123)]
+    conn._conns["a"] = deque([(transp, 123)])
 
     conns_impl = conn._conns
     exc_handler = mock.Mock()
@@ -224,7 +225,7 @@ async def make_conn():
 
     conn = loop.run_until_complete(make_conn())
     transp = mock.Mock()
-    conn._conns["a"] = [(transp, 123)]
+    conn._conns["a"] = deque([(transp, 123)])
 
     conns_impl = conn._conns
     exc_handler = mock.Mock()
@@ -283,7 +284,7 @@ async def test_close(loop) -> None:
 
     conn = aiohttp.BaseConnector(loop=loop)
     assert not conn.closed
-    conn._conns[("host", 8080, False)] = [(proto, object())]
+    conn._conns[("host", 8080, False)] = deque([(proto, object())])
     await conn.close()
 
     assert not conn._conns
@@ -296,7 +297,7 @@ async def test_get(loop: asyncio.AbstractEventLoop, key: ConnectionKey) -> None:
     assert await conn._get(key, []) is None
 
     proto = create_mocked_conn(loop)
-    conn._conns[key] = [(proto, loop.time())]
+    conn._conns[key] = deque([(proto, loop.time())])
     connection = await conn._get(key, [])
     assert connection is not None
     assert connection.protocol == proto
@@ -310,14 +311,14 @@ async def test_get_unconnected_proto(loop) -> None:
     assert await conn._get(key, []) is None
 
     proto = create_mocked_conn(loop)
-    conn._conns[key] = [(proto, loop.time())]
+    conn._conns[key] = deque([(proto, loop.time())])
     connection = await conn._get(key, [])
     assert connection is not None
     assert connection.protocol == proto
     connection.close()
 
     assert await conn._get(key, []) is None
-    conn._conns[key] = [(proto, loop.time())]
+    conn._conns[key] = deque([(proto, loop.time())])
     proto.is_connected = lambda *args: False
     assert await conn._get(key, []) is None
     await conn.close()
@@ -329,14 +330,14 @@ async def test_get_unconnected_proto_ssl(loop) -> None:
     assert await conn._get(key, []) is None
 
     proto = create_mocked_conn(loop)
-    conn._conns[key] = [(proto, loop.time())]
+    conn._conns[key] = deque([(proto, loop.time())])
     connection = await conn._get(key, [])
     assert connection is not None
     assert connection.protocol == proto
     connection.close()
 
     assert await conn._get(key, []) is None
-    conn._conns[key] = [(proto, loop.time())]
+    conn._conns[key] = deque([(proto, loop.time())])
     proto.is_connected = lambda *args: False
     assert await conn._get(key, []) is None
     await conn.close()
@@ -348,7 +349,7 @@ async def test_get_expired(loop: asyncio.AbstractEventLoop) -> None:
     assert await conn._get(key, []) is None
 
     proto = mock.Mock()
-    conn._conns[key] = [(proto, loop.time() - 1000)]
+    conn._conns[key] = deque([(proto, loop.time() - 1000)])
     assert await conn._get(key, []) is None
     assert not conn._conns
     await conn.close()
@@ -361,7 +362,7 @@ async def test_get_expired_ssl(loop: asyncio.AbstractEventLoop) -> None:
 
     proto = mock.Mock()
     transport = proto.transport
-    conn._conns[key] = [(proto, loop.time() - 1000)]
+    conn._conns[key] = deque([(proto, loop.time() - 1000)])
     assert await conn._get(key, []) is None
     assert not conn._conns
     assert conn._cleanup_closed_transports == [transport]
@@ -1411,12 +1412,12 @@ async def test_release_close_do_not_delete_existing_connections(key) -> None:
     proto1 = mock.Mock()
 
     conn = aiohttp.BaseConnector()
-    conn._conns[key] = [(proto1, 1)]
+    conn._conns[key] = deque([(proto1, 1)])
 
     proto = mock.Mock(should_close=True)
     conn._acquired.add(proto)
     conn._release(key, proto)
-    assert conn._conns[key] == [(proto1, 1)]
+    assert conn._conns[key] == deque([(proto1, 1)])
     assert proto.close.called
     await conn.close()
 
@@ -1451,7 +1452,7 @@ async def test_connect(loop, key) -> None:
     req = ClientRequest("GET", URL("http://localhost:80"), loop=loop)
 
     conn = aiohttp.BaseConnector(loop=loop)
-    conn._conns[key] = [(proto, loop.time())]
+    conn._conns[key] = deque([(proto, loop.time())])
     conn._create_connection = mock.Mock()
     conn._create_connection.return_value = loop.create_future()
     conn._create_connection.return_value.set_result(proto)
@@ -1687,12 +1688,15 @@ async def test_ctor_cleanup() -> None:
     assert conn._cleanup_closed_handle is not None
 
 
-async def test_cleanup(key) -> None:
-    testset = {
-        key: [(mock.Mock(), 10), (mock.Mock(), 300)],
-    }
-    testset[key][0][0].is_connected.return_value = True
-    testset[key][1][0].is_connected.return_value = False
+async def test_cleanup(key: ConnectionKey) -> None:
+    m1 = mock.Mock()
+    m2 = mock.Mock()
+    m1.is_connected.return_value = True
+    m2.is_connected.return_value = False
+    testset: DefaultDict[ConnectionKey, Deque[Tuple[ResponseHandler, float]]] = (
+        defaultdict(deque)
+    )
+    testset[key] = deque([(m1, 10), (m2, 300)])
 
     loop = mock.Mock()
     loop.time.return_value = 300
@@ -1710,7 +1714,10 @@ async def test_cleanup(key) -> None:
 async def test_cleanup_close_ssl_transport(ssl_key) -> None:
     proto = mock.Mock()
     transport = proto.transport
-    testset = {ssl_key: [(proto, 10)]}
+    testset: DefaultDict[ConnectionKey, Deque[Tuple[ResponseHandler, float]]] = (
+        defaultdict(deque)
+    )
+    testset[ssl_key] = deque([(proto, 10)])
 
     loop = mock.Mock()
     new_time = asyncio.get_event_loop().time() + 300
@@ -1727,9 +1734,13 @@ async def test_cleanup_close_ssl_transport(ssl_key) -> None:
     assert conn._cleanup_closed_transports == [transport]
 
 
-async def test_cleanup2() -> None:
-    testset = {1: [(mock.Mock(), 300)]}
-    testset[1][0][0].is_connected.return_value = True
+async def test_cleanup2(loop: asyncio.AbstractEventLoop, key: ConnectionKey) -> None:
+    m = create_mocked_conn()
+    m.is_connected.return_value = True
+    testset: DefaultDict[ConnectionKey, Deque[Tuple[ResponseHandler, float]]] = (
+        defaultdict(deque)
+    )
+    testset[key] = deque([(m, 300)])
 
     conn = aiohttp.BaseConnector(keepalive_timeout=10)
     conn._loop = mock.Mock()
@@ -1744,9 +1755,13 @@ async def test_cleanup2() -> None:
     await conn.close()
 
 
-async def test_cleanup3(key) -> None:
-    testset = {key: [(mock.Mock(), 290.1), (mock.Mock(), 305.1)]}
-    testset[key][0][0].is_connected.return_value = True
+async def test_cleanup3(loop: asyncio.AbstractEventLoop, key: ConnectionKey) -> None:
+    m = create_mocked_conn(loop)
+    m.is_connected.return_value = True
+    testset: DefaultDict[ConnectionKey, Deque[Tuple[ResponseHandler, float]]] = (
+        defaultdict(deque)
+    )
+    testset[key] = deque([(m, 290.1), (create_mocked_conn(loop), 305.1)])
 
     loop = mock.Mock()
     loop.time.return_value = 308.5
@@ -1757,7 +1772,7 @@ async def test_cleanup3(key) -> None:
     with mock.patch("aiohttp.connector.monotonic", return_value=308.5):
         conn._cleanup()
 
-    assert conn._conns == {key: [testset[key][1]]}
+    assert conn._conns == {key: deque([testset[key][1]])}
 
     assert conn._cleanup_handle is not None
     loop.call_at.assert_called_with(319, mock.ANY, mock.ANY)
@@ -1927,7 +1942,7 @@ async def test_close_twice(loop) -> None:
     proto = mock.Mock()
 
     conn = aiohttp.BaseConnector(loop=loop)
-    conn._conns[1] = [(proto, object())]
+    conn._conns[1] = deque([(proto, object())])
     await conn.close()
 
     assert not conn._conns
@@ -2324,7 +2339,7 @@ async def test_connect_with_limit(
     )
 
     conn = aiohttp.BaseConnector(loop=loop, limit=1)
-    conn._conns[key] = [(proto, loop.time())]
+    conn._conns[key] = deque([(proto, loop.time())])
     conn._create_connection = mock.Mock()
     conn._create_connection.return_value = loop.create_future()
     conn._create_connection.return_value.set_result(proto)
@@ -2380,7 +2395,7 @@ async def test_connect_queued_operation_tracing(loop, key) -> None:
     )
 
     conn = aiohttp.BaseConnector(loop=loop, limit=1)
-    conn._conns[key] = [(proto, loop.time())]
+    conn._conns[key] = deque([(proto, loop.time())])
     conn._create_connection = mock.Mock()
     conn._create_connection.return_value = loop.create_future()
     conn._create_connection.return_value.set_result(proto)
@@ -2424,7 +2439,7 @@ async def test_connect_reuseconn_tracing(loop, key) -> None:
     )
 
     conn = aiohttp.BaseConnector(loop=loop, limit=1)
-    conn._conns[key] = [(proto, loop.time())]
+    conn._conns[key] = deque([(proto, loop.time())])
     conn2 = await conn.connect(req, traces, ClientTimeout())
     conn2.release()
 
@@ -2441,7 +2456,7 @@ async def test_connect_with_limit_and_limit_per_host(loop, key) -> None:
     req = ClientRequest("GET", URL("http://localhost:80"), loop=loop)
 
     conn = aiohttp.BaseConnector(loop=loop, limit=1000, limit_per_host=1)
-    conn._conns[key] = [(proto, loop.time())]
+    conn._conns[key] = deque([(proto, loop.time())])
     conn._create_connection = mock.Mock()
     conn._create_connection.return_value = loop.create_future()
     conn._create_connection.return_value.set_result(proto)
@@ -2475,7 +2490,7 @@ async def test_connect_with_no_limit_and_limit_per_host(loop, key) -> None:
     req = ClientRequest("GET", URL("http://localhost1:80"), loop=loop)
 
     conn = aiohttp.BaseConnector(loop=loop, limit=0, limit_per_host=1)
-    conn._conns[key] = [(proto, loop.time())]
+    conn._conns[key] = deque([(proto, loop.time())])
     conn._create_connection = mock.Mock()
     conn._create_connection.return_value = loop.create_future()
     conn._create_connection.return_value.set_result(proto)
@@ -2507,7 +2522,7 @@ async def test_connect_with_no_limits(loop, key) -> None:
     req = ClientRequest("GET", URL("http://localhost:80"), loop=loop)
 
     conn = aiohttp.BaseConnector(loop=loop, limit=0, limit_per_host=0)
-    conn._conns[key] = [(proto, loop.time())]
+    conn._conns[key] = deque([(proto, loop.time())])
     conn._create_connection = mock.Mock()
     conn._create_connection.return_value = loop.create_future()
     conn._create_connection.return_value.set_result(proto)
@@ -2541,7 +2556,7 @@ async def test_connect_with_limit_cancelled(loop) -> None:
 
     conn = aiohttp.BaseConnector(loop=loop, limit=1)
     key = ("host", 80, False)
-    conn._conns[key] = [(proto, loop.time())]
+    conn._conns[key] = deque([(proto, loop.time())])
     conn._create_connection = mock.Mock()
     conn._create_connection.return_value = loop.create_future()
     conn._create_connection.return_value.set_result(proto)
@@ -2687,7 +2702,7 @@ async def test_close_with_acquired_connection(loop) -> None:
 
     conn = aiohttp.BaseConnector(loop=loop, limit=1)
     key = ("host", 80, False)
-    conn._conns[key] = [(proto, loop.time())]
+    conn._conns[key] = deque([(proto, loop.time())])
     conn._create_connection = mock.Mock()
     conn._create_connection.return_value = loop.create_future()
     conn._create_connection.return_value.set_result(proto)
@@ -3291,7 +3306,7 @@ async def allow_connection_and_add_dummy_waiter() -> None:
         spec_set=True,
         side_effect=[0, 1, 1, 1],
     ):
-        connector._conns[key] = [(proto, loop.time())]
+        connector._conns[key] = deque([(proto, loop.time())])
         with mock.patch.object(
             connector,
             "_create_connection",