[3.7] #4587 Always check transport is not closing before reuse (#4778)

CHANGES/4587.bugfix

@@ -0,0 +1,10 @@
+Always make sure transport is not closing before reuse a connection.
+
+Reuse a protocol based on keepalive in headers is unreliable.
+For example, uWSGI will not support keepalive even it serves a
+http1.1 request, except explicitly configure uWSGI with a
+`--http-keepalive` option.
+
+Servers designed like uWSGI could cause aiohttp intermittently
+raise a ConnectionResetException when the protocol poll runs
+out and some protocol is reused.

CONTRIBUTORS.txt

@@ -161,6 +161,7 @@ Kirill Malovitsa
 Kyrylo Perevozchikov
 Kyungmin Lee
 Lars P. Søndergaard
+Liu Hua
 Louis-Philippe Huberdeau
 Loïc Lajeanne
 Lu Gong

aiohttp/client_proto.py

@@ -65,7 +65,7 @@ def close(self) -> None:
             self._drop_timeout()
 
     def is_connected(self) -> bool:
-        return self.transport is not None
+        return self.transport is not None and not self.transport.is_closing()
 
     def connection_lost(self, exc: Optional[BaseException]) -> None:
         self._drop_timeout()

aiohttp/connector.py

@@ -349,6 +349,11 @@ def _cleanup(self) -> None:
                                     transport)
                         else:
                             alive.append((proto, use_time))
+                    else:
+                        transport = proto.transport
+                        proto.close()
+                        if key.is_ssl and not self._cleanup_closed_disabled:
+                            self._cleanup_closed_transports.append(transport)
 
                 if alive:
                     connections[key] = alive
@@ -569,6 +574,11 @@ def _get(self, key: 'ConnectionKey') -> Optional[ResponseHandler]:
                         # The very last connection was reclaimed: drop the key
                         del self._conns[key]
                     return proto
+            else:
+                transport = proto.transport
+                proto.close()
+                if key.is_ssl and not self._cleanup_closed_disabled:
+                    self._cleanup_closed_transports.append(transport)
 
         # No more connections: drop the key
         del self._conns[key]

tests/test_connector.py

@@ -287,6 +287,38 @@ async def test_get(loop) -> None:
     conn.close()
 
 
+async def test_get_unconnected_proto(loop) -> None:
+    conn = aiohttp.BaseConnector()
+    key = ConnectionKey('localhost', 80, False, None, None, None, None)
+    assert conn._get(key) is None
+
+    proto = create_mocked_conn(loop)
+    conn._conns[key] = [(proto, loop.time())]
+    assert conn._get(key) == proto
+
+    assert conn._get(key) is None
+    conn._conns[key] = [(proto, loop.time())]
+    proto.is_connected = lambda *args: False
+    assert conn._get(key) is None
+    await conn.close()
+
+
+async def test_get_unconnected_proto_ssl(loop) -> None:
+    conn = aiohttp.BaseConnector()
+    key = ConnectionKey('localhost', 80, True, None, None, None, None)
+    assert conn._get(key) is None
+
+    proto = create_mocked_conn(loop)
+    conn._conns[key] = [(proto, loop.time())]
+    assert conn._get(key) == proto
+
+    assert conn._get(key) is None
+    conn._conns[key] = [(proto, loop.time())]
+    proto.is_connected = lambda *args: False
+    assert conn._get(key) is None
+    await conn.close()
+
+
 async def test_get_expired(loop) -> None:
     conn = aiohttp.BaseConnector(loop=loop)
     key = ConnectionKey('localhost', 80, False, None, None, None, None)