Fixed clean shutdown and added test.

Fixed https://bugs.python.org/issue30698

Author: fantix

tests/test_tcp.py

@@ -2231,6 +2231,72 @@ async def start_server():
         for client in clients:
             client.stop()
 
+    def test_shutdown_cleanly(self):
+        if self.implementation == 'asyncio':
+            raise unittest.SkipTest()
+
+        CNT = 0
+        TOTAL_CNT = 25
+
+        A_DATA = b'A' * 1024 * 1024
+
+        sslctx = self._create_server_ssl_context(self.ONLYCERT, self.ONLYKEY)
+        client_sslctx = self._create_client_ssl_context()
+
+        def server(sock):
+            sock.starttls(
+                sslctx,
+                server_side=True)
+
+            data = sock.recv_all(len(A_DATA))
+            self.assertEqual(data, A_DATA)
+            sock.send(b'OK')
+
+            sock.unwrap()
+
+            sock.close()
+
+        async def client(addr):
+            extras = {}
+            if self.implementation != 'asyncio' or self.PY37:
+                extras = dict(ssl_handshake_timeout=10.0)
+
+            reader, writer = await asyncio.open_connection(
+                *addr,
+                ssl=client_sslctx,
+                server_hostname='',
+                loop=self.loop,
+                **extras)
+
+            writer.write(A_DATA)
+            self.assertEqual(await reader.readexactly(2), b'OK')
+
+            await writer.wait_closed()
+
+            nonlocal CNT
+            CNT += 1
+
+            writer.close()
+
+        def run(coro):
+            nonlocal CNT
+            CNT = 0
+
+            with self.tcp_server(server,
+                                 max_clients=TOTAL_CNT,
+                                 backlog=TOTAL_CNT) as srv:
+                tasks = []
+                for _ in range(TOTAL_CNT):
+                    tasks.append(coro(srv.addr))
+
+                self.loop.run_until_complete(
+                    asyncio.gather(*tasks, loop=self.loop))
+
+            self.assertEqual(CNT, TOTAL_CNT)
+
+        with self._silence_eof_received_warning():
+            run(client)
+
 
 class Test_UV_TCPSSL(_TestSSL, tb.UVTestCase):
     pass

uvloop/sslproto.pyx

@@ -305,14 +305,13 @@ class SSLProtocol(object):
         meaning a regular EOF is received or the connection was
         aborted or closed).
         """
-        if self._state in (_WRAPPED, _SHUTDOWN):
-            self._loop.call_soon(self._app_protocol.connection_lost, exc)
-        else:
-            # Most likely an exception occurred while in SSL handshake.
+        if self._state == _DO_HANDSHAKE:
             # Just mark the app transport as closed so that its __del__
             # doesn't complain.
             if self._app_transport is not None:
                 self._app_transport._closed = True
+        else:
+            self._loop.call_soon(self._app_protocol.connection_lost, exc)
         self._set_state(_UNWRAPPED)
         self._transport = None
         self._app_transport = None
@@ -498,7 +497,6 @@ class SSLProtocol(object):
         if shutdown_exc:
             self._fatal_error(shutdown_exc)
         else:
-            self._set_state(_UNWRAPPED)
             self._loop.call_soon(self._transport.close)
 
     def _abort(self):
@@ -582,6 +580,7 @@ class SSLProtocol(object):
         if offset:
             self._app_protocol.buffer_updated(offset)
         if not count:
+            # close_notify
             self._start_shutdown()
 
     def _do_read__copied(self):
@@ -601,6 +600,7 @@ class SSLProtocol(object):
         if data:
             self._app_protocol.data_received(b''.join(data))
         if not chunk:
+            # close_notify
             self._start_shutdown()
 
     # Flow control for writes from APP socket