From d315c55a0846ed70d9e07c6de1211c163ac6187d Mon Sep 17 00:00:00 2001
From: Dahuage <Dahuage@users.noreply.github.com>
Date: Fri, 16 Oct 2020 15:14:20 +0800
Subject: [PATCH] #4587 Always check transport is not closing before reuse
 (#4778)

* #4587 fix by dahua

* #4587 add unit test

Co-authored-by: dahua <liuhua@xiaoyezi.com>
Co-authored-by: Andrew Svetlov <andrew.svetlov@gmail.com>
---
 CHANGES/4587.bugfix     | 10 ++++++++++
 CONTRIBUTORS.txt        |  1 +
 aiohttp/client_proto.py |  2 +-
 aiohttp/connector.py    | 10 ++++++++++
 tests/test_connector.py | 32 ++++++++++++++++++++++++++++++++
 5 files changed, 54 insertions(+), 1 deletion(-)
 create mode 100644 CHANGES/4587.bugfix

diff --git a/CHANGES/4587.bugfix b/CHANGES/4587.bugfix
new file mode 100644
index 00000000000..f413dfac7c0
--- /dev/null
+++ b/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.
diff --git a/CONTRIBUTORS.txt b/CONTRIBUTORS.txt
index b9e0aee2c72..77daf329b99 100644
--- a/CONTRIBUTORS.txt
+++ b/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
diff --git a/aiohttp/client_proto.py b/aiohttp/client_proto.py
index a44e6454234..bc6d5af3a8d 100644
--- a/aiohttp/client_proto.py
+++ b/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()
diff --git a/aiohttp/connector.py b/aiohttp/connector.py
index cce2451bd9d..3a80bf590f9 100644
--- a/aiohttp/connector.py
+++ b/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]
diff --git a/tests/test_connector.py b/tests/test_connector.py
index 1d8273be19c..a8b334304f3 100644
--- a/tests/test_connector.py
+++ b/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)