WIP

Author: fantix

tests/test_tcp.py

@@ -2682,9 +2682,9 @@ def eof_server(sock):
             # send EOF
             sock.shutdown(socket.SHUT_WR)
 
-            # should receive all data
-            data = sock.recv_all(CHUNK * SIZE)
-            self.assertEqual(len(data), CHUNK * SIZE)
+            # should not receive any data
+            data = sock.recv(CHUNK * SIZE)
+            self.assertEqual(data, b'')
 
             sock.close()
 
@@ -2842,7 +2842,7 @@ def server(sock):
             sock.shutdown(socket.SHUT_WR)
             loop.call_soon_threadsafe(eof.set)
             # make sure we have enough time to reproduce the issue
-            assert sock.recv(1024) == b''
+            self.assertEqual(sock.recv(1024), b'')
             sock.close()
 
         class Protocol(asyncio.Protocol):
@@ -2875,13 +2875,18 @@ async def client(addr):
             tr.resume_reading()
             await pr.fut
             tr.close()
-            # extra data received after transport.close() is ignored
-            self.assertIsNone(extra)
+            if self.implementation != 'asyncio':
+                # extra data received after transport.close() should be
+                # ignored - this is likely a bug in asyncio
+                self.assertIsNone(extra)
 
         with self.tcp_server(server) as srv:
             loop.run_until_complete(client(srv.addr))
 
     def test_shutdown_while_pause_reading(self):
+        if self.implementation == 'asyncio':
+            raise unittest.SkipTest()
+
         loop = self.loop
         conn_made = loop.create_future()
         eof_recvd = loop.create_future()
@@ -2898,14 +2903,14 @@ def server(sock):
             while True:
                 try:
                     sslobj.do_handshake()
+                    sslobj.write(b'trailing data')
+                    break
                 except ssl.SSLWantReadError:
                     if outgoing.pending:
                         sock.send(outgoing.read())
                     incoming.write(sock.recv(16384))
-                else:
-                    if outgoing.pending:
-                        sock.send(outgoing.read())
-                    break
+            if outgoing.pending:
+                sock.send(outgoing.read())
 
             while True:
                 try:

uvloop/includes/stdlib.pxi

@@ -129,6 +129,7 @@ cdef ssl_MemoryBIO = ssl.MemoryBIO
 cdef ssl_create_default_context = ssl.create_default_context
 cdef ssl_SSLError = ssl.SSLError
 cdef ssl_SSLAgainErrors = (ssl.SSLWantReadError, ssl.SSLSyscallError)
+cdef ssl_SSLZeroReturnError = ssl.SSLZeroReturnError
 cdef ssl_CertificateError = ssl.CertificateError
 cdef int ssl_SSL_ERROR_WANT_READ = ssl.SSL_ERROR_WANT_READ
 cdef int ssl_SSL_ERROR_WANT_WRITE = ssl.SSL_ERROR_WANT_WRITE

uvloop/sslproto.pxd

@@ -65,7 +65,6 @@ cdef class SSLProtocol:
 
         bint _ssl_writing_paused
         bint _app_reading_paused
-        bint _eof_received
 
         size_t _incoming_high_water
         size_t _incoming_low_water
@@ -100,6 +99,7 @@ cdef class SSLProtocol:
 
     cdef _start_shutdown(self)
     cdef _check_shutdown_timeout(self)
+    cdef _do_read_into_void(self)
     cdef _do_flush(self)
     cdef _do_shutdown(self)
     cdef _on_shutdown_complete(self, shutdown_exc)

uvloop/sslproto.pyx

@@ -278,7 +278,6 @@ cdef class SSLProtocol:
         self._incoming_high_water = 0
         self._incoming_low_water = 0
         self._set_read_buffer_limits()
-        self._eof_received = False
 
         self._app_writing_paused = False
         self._outgoing_high_water = 0
@@ -392,27 +391,24 @@ cdef class SSLProtocol:
         will close itself.  If it returns a true value, closing the
         transport is up to the protocol.
         """
-        self._eof_received = True
         try:
             if self._loop.get_debug():
                 aio_logger.debug("%r received EOF", self)
 
             if self._state == DO_HANDSHAKE:
                 self._on_handshake_complete(ConnectionResetError)
 
-            elif self._state == WRAPPED:
-                self._set_state(FLUSHING)
-                self._do_write()
+            elif self._state == WRAPPED or self._state == FLUSHING:
+                # we treat low-level EOF as a critical situation similar as a
+                # broken connection - just send whatever in the buffer and
+                # close up. No application level eof_received() is called -
+                # because we don't want the user to think that this is a
+                # graceful shutdown triggered by SSL "close_notify".
                 self._set_state(SHUTDOWN)
-                self._do_shutdown()
-
-            elif self._state == FLUSHING:
-                self._do_write()
-                self._set_state(SHUTDOWN)
-                self._do_shutdown()
+                self._on_shutdown_complete(None)
 
             elif self._state == SHUTDOWN:
-                self._do_shutdown()
+                self._on_shutdown_complete(None)
 
         except Exception:
             self._transport.close()
@@ -443,6 +439,9 @@ cdef class SSLProtocol:
         elif self._state == WRAPPED and new_state == FLUSHING:
             allowed = True
 
+        elif self._state == WRAPPED and new_state == SHUTDOWN:
+            allowed = True
+
         elif self._state == FLUSHING and new_state == SHUTDOWN:
             allowed = True
 
@@ -560,54 +559,70 @@ cdef class SSLProtocol:
             self._transport._force_close(
                 aio_TimeoutError('SSL shutdown timed out'))
 
-    cdef _do_flush(self):
+    cdef _do_read_into_void(self):
+        """Consume and discard incoming application data.
+
+        If close_notify is received for the first time, call eof_received.
+        """
         cdef:
             bint close_notify = False
+        try:
+            while True:
+                if not self._sslobj_read(SSL_READ_MAX_SIZE):
+                    close_notify = True
+                    break
+        except ssl_SSLAgainErrors as exc:
+            pass
+        except ssl_SSLZeroReturnError:
+            close_notify = True
+        if close_notify:
+            self._call_eof_received()
 
+    cdef _do_flush(self):
+        """Flush the write backlog, discarding new data received.
+
+        We don't send close_notify in FLUSHING because we still want to send
+        the remaining data over SSL, even if we received a close_notify. Also,
+        no application-level resume_writing() or pause_writing() will be called
+        in FLUSHING, as we could fully manage the flow control internally.
+        """
         try:
-            try:
-                while True:
-                    if not self._sslobj_read(SSL_READ_MAX_SIZE):
-                        close_notify = True
-                        break
-            except ssl_SSLAgainErrors as exc:
-                pass
-            if close_notify:
-                self._call_eof_received()
-            if self._write_backlog:
-                self._do_write()
-            else:
-                self._process_outgoing()
+            self._do_read_into_void()
+            self._do_write()
+            self._process_outgoing()
             self._control_ssl_reading()
         except Exception as ex:
-            self._fatal_error(ex, 'Fatal error on SSL protocol')
+            self._on_shutdown_complete(ex)
         else:
             if not self._get_write_buffer_size():
                 self._set_state(SHUTDOWN)
                 self._do_shutdown()
 
     cdef _do_shutdown(self):
+        """Send close_notify and wait for the same from the peer."""
         try:
-            if not self._eof_received:
+            # we must skip all application data (if any) before unwrap
+            self._do_read_into_void()
+            try:
                 self._sslobj.unwrap()
-        except ssl_SSLAgainErrors as exc:
-            self._process_outgoing()
-        except ssl_SSLError as exc:
-            self._on_shutdown_complete(exc)
-        else:
-            self._process_outgoing()
-            self._call_eof_received()
-            self._on_shutdown_complete(None)
+            except ssl_SSLAgainErrors as exc:
+                self._process_outgoing()
+            else:
+                self._process_outgoing()
+                if not self._get_write_buffer_size():
+                    self._on_shutdown_complete(None)
+        except Exception as ex:
+            self._on_shutdown_complete(ex)
 
     cdef _on_shutdown_complete(self, shutdown_exc):
         if self._shutdown_timeout_handle is not None:
             self._shutdown_timeout_handle.cancel()
             self._shutdown_timeout_handle = None
 
         if shutdown_exc:
-            self._fatal_error(shutdown_exc)
+            self._fatal_error(shutdown_exc, 'Error occurred during shutdown')
         else:
-            self._loop.call_soon(self._transport.close)
+            self._transport.close()
 
     cdef _abort(self, exc):
         self._set_state(UNWRAPPED)
@@ -630,11 +645,14 @@ cdef class SSLProtocol:
         try:
             if self._state == WRAPPED:
                 self._do_write()
+                self._process_outgoing()
+                self._control_app_writing()
 
         except Exception as ex:
             self._fatal_error(ex, 'Fatal error on SSL protocol')
 
     cdef _do_write(self):
+        """Do SSL write, consumes write backlog and fills outgoing BIO."""
         cdef size_t data_len, count
         try:
             while self._write_backlog:
@@ -651,14 +669,13 @@ cdef class SSLProtocol:
                     self._write_buffer_size -= data_len
         except ssl_SSLAgainErrors as exc:
             pass
-        self._process_outgoing()
 
     cdef _process_outgoing(self):
+        """Send bytes from the outgoing BIO."""
         if not self._ssl_writing_paused:
             data = self._outgoing_read()
             if len(data):
                 self._transport.write(data)
-        self._control_app_writing()
 
     # Incoming flow
 
@@ -673,8 +690,8 @@ cdef class SSLProtocol:
                     self._do_read__copied()
                 if self._write_backlog:
                     self._do_write()
-                else:
-                    self._process_outgoing()
+                self._process_outgoing()
+                self._control_app_writing()
             self._control_ssl_reading()
         except Exception as ex:
             self._fatal_error(ex, 'Fatal error on SSL protocol')
@@ -860,7 +877,16 @@ cdef class SSLProtocol:
         """
         assert self._ssl_writing_paused
         self._ssl_writing_paused = False
-        self._process_outgoing()
+
+        if self._state == WRAPPED:
+            self._process_outgoing()
+            self._control_app_writing()
+
+        elif self._state == FLUSHING:
+            self._do_flush()
+
+        elif self._state == SHUTDOWN:
+            self._do_shutdown()
 
     cdef _fatal_error(self, exc, message='Fatal error on transport'):
         if self._transport: