Cythonize sslproto and improve performance.

Author: fantix

examples/bench/echoserver.py

@@ -74,6 +74,21 @@ def data_received(self, data):
         self.transport.write(data)
 
 
+class EchoBufferedProtocol(asyncio.BufferedProtocol):
+    def connection_made(self, transport):
+        self.transport = transport
+        self.buffer = bytearray(256 * 1024)
+
+    def connection_lost(self, exc):
+        self.transport = None
+
+    def get_buffer(self, sizehint):
+        return self.buffer
+
+    def buffer_updated(self, nbytes):
+        self.transport.write(self.buffer[:nbytes])
+
+
 async def print_debug(loop):
     while True:
         print(chr(27) + "[2J")  # clear screen
@@ -89,6 +104,7 @@ async def print_debug(loop):
     parser.add_argument('--addr', default='127.0.0.1:25000', type=str)
     parser.add_argument('--print', default=False, action='store_true')
     parser.add_argument('--ssl', default=False, action='store_true')
+    parser.add_argument('--buffered', default=False, action='store_true')
     args = parser.parse_args()
 
     if args.uvloop:
@@ -140,6 +156,10 @@ async def print_debug(loop):
             print('cannot use --stream and --proto simultaneously')
             exit(1)
 
+        if args.buffered:
+            print('cannot use --stream and --buffered simultaneously')
+            exit(1)
+
         print('using asyncio/streams')
         if unix:
             coro = asyncio.start_unix_server(echo_client_streams,
@@ -155,12 +175,18 @@ async def print_debug(loop):
             print('cannot use --stream and --proto simultaneously')
             exit(1)
 
-        print('using simple protocol')
+        if args.buffered:
+            print('using buffered protocol')
+            protocol = EchoBufferedProtocol
+        else:
+            print('using simple protocol')
+            protocol = EchoProtocol
+
         if unix:
-            coro = loop.create_unix_server(EchoProtocol, addr,
+            coro = loop.create_unix_server(protocol, addr,
                                            ssl=server_context)
         else:
-            coro = loop.create_server(EchoProtocol, *addr,
+            coro = loop.create_server(protocol, *addr,
                                       ssl=server_context)
         srv = loop.run_until_complete(coro)
     else:

tests/test_tcp.py

@@ -2345,7 +2345,7 @@ async def client(addr):
                 writer.write(b'I AM WRITING NOWHERE2' * 100)
 
             self.assertEqual(
-                len(writer.transport._ssl_protocol._write_backlog), 0)
+                writer.transport.get_write_buffer_size(), 0)
 
             await future
 
@@ -2548,8 +2548,7 @@ async def client(addr):
             # fill write backlog in a hacky way - renegotiation won't help
             ssl_protocol = writer.transport._ssl_protocol
             for _ in range(SIZE):
-                ssl_protocol._write_backlog.append(b'x' * CHUNK)
-                ssl_protocol._write_buffer_size += CHUNK
+                ssl_protocol._append_write_backlog(b'x' * CHUNK)
 
             try:
                 data = await reader.read()

uvloop/includes/python.pxd

@@ -14,6 +14,13 @@ cdef extern from "Python.h":
     int _PyImport_ReleaseLock()
     void _Py_RestoreSignals()
 
+    object PyMemoryView_FromMemory(char *mem, ssize_t size, int flags)
+    object PyMemoryView_FromObject(object obj)
+    int PyMemoryView_Check(object obj)
+
+    cdef enum:
+        PyBUF_WRITE
+
 
 cdef extern from "includes/compat.h":
     ctypedef struct PyContext

uvloop/includes/stdlib.pxi

@@ -120,6 +120,7 @@ cdef ssl_SSLContext = ssl.SSLContext
 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_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/loop.pxd

@@ -227,6 +227,7 @@ include "handles/pipe.pxd"
 include "handles/process.pxd"
 
 include "request.pxd"
+include "sslproto.pxd"
 
 include "handles/udp.pxd"
 

uvloop/loop.pyx

@@ -18,7 +18,9 @@ from .includes.python cimport PY_VERSION_HEX, \
                               PyContext, \
                               PyContext_CopyCurrent, \
                               PyContext_Enter, \
-                              PyContext_Exit
+                              PyContext_Exit, \
+                              PyMemoryView_FromMemory, PyBUF_WRITE, \
+                              PyMemoryView_FromObject, PyMemoryView_Check
 
 from libc.stdint cimport uint64_t
 from libc.string cimport memset, strerror, memcpy
@@ -1566,7 +1568,7 @@ cdef class Loop:
             resume_cb.cancel()
             raise
 
-        return ssl_protocol._app_transport
+        return (<SSLProtocol>ssl_protocol)._app_transport
 
     @cython.iterable_coroutine
     async def create_server(self, protocol_factory, host=None, port=None,
@@ -1939,7 +1941,7 @@ cdef class Loop:
             except Exception:
                 tr._close()
                 raise
-            return protocol._app_transport, app_protocol
+            return (<SSLProtocol>protocol)._app_transport, app_protocol
         else:
             return tr, protocol
 
@@ -2169,7 +2171,7 @@ cdef class Loop:
             except Exception:
                 tr._close()
                 raise
-            return protocol._app_transport, app_protocol
+            return (<SSLProtocol>protocol)._app_transport, app_protocol
         else:
             return tr, protocol
 
@@ -2533,7 +2535,7 @@ cdef class Loop:
             raise
 
         if ssl:
-            return protocol._app_transport, protocol
+            return (<SSLProtocol>protocol)._app_transport, protocol
         else:
             return transport, protocol
 

uvloop/sslproto.pxd

@@ -0,0 +1,113 @@
+cdef enum ProtocolState:
+    _UNWRAPPED = 0
+    _DO_HANDSHAKE = 1
+    _WRAPPED = 2
+    _FLUSHING = 3
+    _SHUTDOWN = 4
+
+
+cdef class SSLProtocol:
+    cdef:
+        bint _server_side
+        str _server_hostname
+        object _sslcontext
+
+        object _extra
+
+        object _write_backlog
+        size_t _write_buffer_size
+
+        object _waiter
+        object _loop
+        object _app_transport
+
+        object _transport
+        bint _call_connection_made
+        object _ssl_handshake_timeout
+        object _ssl_shutdown_timeout
+
+        object _sslobj
+        object _sslobj_read
+        object _sslobj_write
+        object _incoming
+        object _incoming_write
+        object _outgoing
+        object _outgoing_read
+        char* _ssl_buffer
+        size_t _ssl_buffer_len
+        object _ssl_buffer_view
+        ProtocolState _state
+        size_t _conn_lost
+        bint _eof_received
+
+        bint _ssl_writing_paused
+        bint _app_reading_paused
+
+        size_t _incoming_high_water
+        size_t _incoming_low_water
+        bint _ssl_reading_paused
+
+        bint _app_writing_paused
+        size_t _outgoing_high_water
+        size_t _outgoing_low_water
+
+        object _app_protocol
+        bint _app_protocol_is_buffer
+        object _app_protocol_get_buffer
+        object _app_protocol_buffer_updated
+
+        object _handshake_start_time
+        object _handshake_timeout_handle
+        object _shutdown_timeout_handle
+
+    cdef _set_app_protocol(self, app_protocol)
+    cdef _wakeup_waiter(self, exc=*)
+    cdef _get_extra_info(self, name, default=*)
+    cdef _set_state(self, ProtocolState new_state)
+
+    # Handshake flow
+
+    cdef _start_handshake(self)
+    cdef _check_handshake_timeout(self)
+    cdef _do_handshake(self)
+    cdef _on_handshake_complete(self, handshake_exc)
+
+    # Shutdown flow
+
+    cdef _start_shutdown(self)
+    cdef _check_shutdown_timeout(self)
+    cdef _do_flush(self)
+    cdef _do_shutdown(self)
+    cdef _on_shutdown_complete(self, shutdown_exc)
+    cdef _abort(self, exc)
+
+    # Outgoing flow
+
+    cdef _write_appdata(self, list_of_data)
+    cdef _do_write(self)
+    cdef _process_outgoing(self)
+
+    # Incoming flow
+
+    cdef _do_read(self)
+    cdef _do_read__buffered(self)
+    cdef _do_read__copied(self)
+    cdef _call_eof_received(self)
+
+    # Flow control for writes from APP socket
+
+    cdef _control_app_writing(self)
+    cdef size_t _get_write_buffer_size(self)
+    cdef _set_write_buffer_limits(self, high=*, low=*)
+
+    # Flow control for reads to APP socket
+
+    cdef _pause_reading(self)
+    cdef _resume_reading(self)
+
+    # Flow control for reads from SSL socket
+
+    cdef _control_ssl_reading(self)
+    cdef _set_read_buffer_limits(self, high=*, low=*)
+    cdef size_t _get_read_buffer_size(self)
+    cdef _fatal_error(self, exc, message=*)