New websockets (sanic-org#2158)

* First attempt at new Websockets implementation based on websockets >= 9.0, with sans-i/o features. Requires more work.

* Update sanic/websocket.py

Co-authored-by: Adam Hopkins <adam@amhopkins.com>

* Update sanic/websocket.py

Co-authored-by: Adam Hopkins <adam@amhopkins.com>

* Update sanic/websocket.py

Co-authored-by: Adam Hopkins <adam@amhopkins.com>

* wip, update websockets code to new Sans/IO API

* Refactored new websockets impl into own modules
Incorporated other suggestions made by team

* Another round of work on the new websockets impl
* Added websocket_timeout support (matching previous/legacy support)
* Lots more comments
* Incorporated suggested changes from previous round of review
* Changed RuntimeError usage to ServerError
* Changed SanicException usage to ServerError
* Removed some redundant asserts
* Change remaining asserts to ServerErrors
* Fixed some timeout handling issues
* Fixed websocket.close() handling, and made it more robust
* Made auto_close task smarter and more error-resilient
* Made fail_connection routine smarter and more error-resilient

* Further new websockets impl fixes
* Update compatibility with Websockets v10
* Track server connection state in a more precise way
* Try to handle the shutdown process more gracefully
* Add a new end_connection() helper, to use as an alterative to close() or fail_connection()
* Kill the auto-close task and keepalive-timeout task when sanic is shutdown
* Deprecate WEBSOCKET_READ_LIMIT and WEBSOCKET_WRITE_LIMIT configs, they are not used in this implementation.

* Change a warning message to debug level
Remove default values for deprecated websocket parameters

* Fix flake8 errors

* Fix a couple of missed failing tests

* remove websocket bench from examples

* Integrate suggestions from code reviews
Use Optional[T] instead of union[T,None]
Fix mypy type logic errors
change "is not None" to truthy checks where appropriate
change "is None" to falsy checks were appropriate
Add more debug logging when debug mode is on
Change to using sanic.logger for debug logging rather than error_logger.

* Fix long line lengths of debug messages
Add some new debug messages when websocket IO is paused and unpaused for flow control
Fix websocket example to use app.static()

* remove unused import in websocket example app

* re-run isort after Flake8 fixes

Co-authored-by: Adam Hopkins <adam@amhopkins.com>
Co-authored-by: Adam Hopkins <admhpkns@gmail.com>

examples/websocket.py

@@ -1,13 +1,14 @@
 from sanic import Sanic
-from sanic.response import file
+from sanic.response import redirect
 
 app = Sanic(__name__)
 
 
-@app.route('/')
-async def index(request):
-    return await file('websocket.html')
+app.static('index.html', "websocket.html")
 
+@app.route('/')
+def index(request):
+    return redirect("index.html")
 
 @app.websocket('/feed')
 async def feed(request, ws):

sanic/app.py

@@ -74,9 +74,10 @@
 from sanic.server import AsyncioServer, HttpProtocol
 from sanic.server import Signal as ServerSignal
 from sanic.server import serve, serve_multiple, serve_single
+from sanic.server.protocols.websocket_protocol import WebSocketProtocol
+from sanic.server.websockets.impl import ConnectionClosed
 from sanic.signals import Signal, SignalRouter
 from sanic.touchup import TouchUp, TouchUpMeta
-from sanic.websocket import ConnectionClosed, WebSocketProtocol
 
 
 class Sanic(BaseSanic, metaclass=TouchUpMeta):
@@ -871,39 +872,31 @@ async def handle_request(self, request: Request):  # no cov
     async def _websocket_handler(
         self, handler, request, *args, subprotocols=None, **kwargs
     ):
-        request.app = self
-        if not getattr(handler, "__blueprintname__", False):
-            request._name = handler.__name__
-        else:
-            request._name = (
-                getattr(handler, "__blueprintname__", "") + handler.__name__
-            )
-
-            pass
-
         if self.asgi:
             ws = request.transport.get_websocket_connection()
             await ws.accept(subprotocols)
         else:
             protocol = request.transport.get_protocol()
-            protocol.app = self
-
             ws = await protocol.websocket_handshake(request, subprotocols)
 
         # schedule the application handler
         # its future is kept in self.websocket_tasks in case it
         # needs to be cancelled due to the server being stopped
         fut = ensure_future(handler(request, ws, *args, **kwargs))
         self.websocket_tasks.add(fut)
+        cancelled = False
         try:
             await fut
         except Exception as e:
             self.error_handler.log(request, e)
         except (CancelledError, ConnectionClosed):
-            pass
+            cancelled = True
         finally:
             self.websocket_tasks.remove(fut)
-            await ws.close()
+            if cancelled:
+                ws.end_connection(1000)
+            else:
+                await ws.close()
 
     # -------------------------------------------------------------------- #
     # Testing

sanic/asgi.py

@@ -10,7 +10,7 @@
 from sanic.models.asgi import ASGIReceive, ASGIScope, ASGISend, MockTransport
 from sanic.request import Request
 from sanic.server import ConnInfo
-from sanic.websocket import WebSocketConnection
+from sanic.server.websockets.connection import WebSocketConnection
 
 
 class Lifespan:

sanic/config.py

@@ -35,12 +35,9 @@
     "REQUEST_MAX_SIZE": 100000000,  # 100 megabytes
     "REQUEST_TIMEOUT": 60,  # 60 seconds
     "RESPONSE_TIMEOUT": 60,  # 60 seconds
-    "WEBSOCKET_MAX_QUEUE": 32,
     "WEBSOCKET_MAX_SIZE": 2 ** 20,  # 1 megabyte
     "WEBSOCKET_PING_INTERVAL": 20,
     "WEBSOCKET_PING_TIMEOUT": 20,
-    "WEBSOCKET_READ_LIMIT": 2 ** 16,
-    "WEBSOCKET_WRITE_LIMIT": 2 ** 16,
 }
 
 
@@ -62,12 +59,9 @@ class Config(dict):
     REQUEST_MAX_SIZE: int
     REQUEST_TIMEOUT: int
     RESPONSE_TIMEOUT: int
-    WEBSOCKET_MAX_QUEUE: int
     WEBSOCKET_MAX_SIZE: int
     WEBSOCKET_PING_INTERVAL: int
     WEBSOCKET_PING_TIMEOUT: int
-    WEBSOCKET_READ_LIMIT: int
-    WEBSOCKET_WRITE_LIMIT: int
 
     def __init__(
         self,

sanic/mixins/routes.py

@@ -121,8 +121,11 @@ def decorator(handler):
                         "Expected either string or Iterable of host strings, "
                         "not %s" % host
                     )
-
-            if isinstance(subprotocols, (list, tuple, set)):
+            if isinstance(subprotocols, list):
+                # Ordered subprotocols, maintain order
+                subprotocols = tuple(subprotocols)
+            elif isinstance(subprotocols, set):
+                # subprotocol is unordered, keep it unordered
                 subprotocols = frozenset(subprotocols)
 
             route = FutureRoute(

sanic/models/asgi.py

@@ -3,7 +3,7 @@
 from typing import Any, Awaitable, Callable, MutableMapping, Optional, Union
 
 from sanic.exceptions import InvalidUsage
-from sanic.websocket import WebSocketConnection
+from sanic.server.websockets.connection import WebSocketConnection
 
 
 ASGIScope = MutableMapping[str, Any]

sanic/server/protocols/websocket_protocol.py

@@ -0,0 +1,165 @@
+from typing import TYPE_CHECKING, Optional, Sequence
+
+from websockets.connection import CLOSED, CLOSING, OPEN
+from websockets.server import ServerConnection
+
+from sanic.exceptions import ServerError
+from sanic.log import error_logger
+from sanic.server import HttpProtocol
+
+from ..websockets.impl import WebsocketImplProtocol
+
+
+if TYPE_CHECKING:
+    from websockets import http11
+
+
+class WebSocketProtocol(HttpProtocol):
+
+    websocket: Optional[WebsocketImplProtocol]
+    websocket_timeout: float
+    websocket_max_size = Optional[int]
+    websocket_ping_interval = Optional[float]
+    websocket_ping_timeout = Optional[float]
+
+    def __init__(
+        self,
+        *args,
+        websocket_timeout: float = 10.0,
+        websocket_max_size: Optional[int] = None,
+        websocket_max_queue: Optional[int] = None,  # max_queue is deprecated
+        websocket_read_limit: Optional[int] = None,  # read_limit is deprecated
+        websocket_write_limit: Optional[int] = None,  # write_limit deprecated
+        websocket_ping_interval: Optional[float] = 20.0,
+        websocket_ping_timeout: Optional[float] = 20.0,
+        **kwargs,
+    ):
+        super().__init__(*args, **kwargs)
+        self.websocket = None
+        self.websocket_timeout = websocket_timeout
+        self.websocket_max_size = websocket_max_size
+        if websocket_max_queue is not None and websocket_max_queue > 0:
+            # TODO: Reminder remove this warning in v22.3
+            error_logger.warning(
+                DeprecationWarning(
+                    "Websocket no longer uses queueing, so websocket_max_queue"
+                    " is no longer required."
+                )
+            )
+        if websocket_read_limit is not None and websocket_read_limit > 0:
+            # TODO: Reminder remove this warning in v22.3
+            error_logger.warning(
+                DeprecationWarning(
+                    "Websocket no longer uses read buffers, so "
+                    "websocket_read_limit is not required."
+                )
+            )
+        if websocket_write_limit is not None and websocket_write_limit > 0:
+            # TODO: Reminder remove this warning in v22.3
+            error_logger.warning(
+                DeprecationWarning(
+                    "Websocket no longer uses write buffers, so "
+                    "websocket_write_limit is not required."
+                )
+            )
+        self.websocket_ping_interval = websocket_ping_interval
+        self.websocket_ping_timeout = websocket_ping_timeout
+
+    def connection_lost(self, exc):
+        if self.websocket is not None:
+            self.websocket.connection_lost(exc)
+        super().connection_lost(exc)
+
+    def data_received(self, data):
+        if self.websocket is not None:
+            self.websocket.data_received(data)
+        else:
+            # Pass it to HttpProtocol handler first
+            # That will (hopefully) upgrade it to a websocket.
+            super().data_received(data)
+
+    def eof_received(self) -> Optional[bool]:
+        if self.websocket is not None:
+            return self.websocket.eof_received()
+        else:
+            return False
+
+    def close(self, timeout: Optional[float] = None):
+        # Called by HttpProtocol at the end of connection_task
+        # If we've upgraded to websocket, we do our own closing
+        if self.websocket is not None:
+            # Note, we don't want to use websocket.close()
+            # That is used for user's application code to send a
+            # websocket close packet. This is different.
+            self.websocket.end_connection(1001)
+        else:
+            super().close()
+
+    def close_if_idle(self):
+        # Called by Sanic Server when shutting down
+        # If we've upgraded to websocket, shut it down
+        if self.websocket is not None:
+            if self.websocket.connection.state in (CLOSING, CLOSED):
+                return True
+            elif self.websocket.loop is not None:
+                self.websocket.loop.create_task(self.websocket.close(1001))
+            else:
+                self.websocket.end_connection(1001)
+        else:
+            return super().close_if_idle()
+
+    async def websocket_handshake(
+        self, request, subprotocols=Optional[Sequence[str]]
+    ):
+        # let the websockets package do the handshake with the client
+        try:
+            if subprotocols is not None:
+                # subprotocols can be a set or frozenset,
+                # but ServerConnection needs a list
+                subprotocols = list(subprotocols)
+            ws_conn = ServerConnection(
+                max_size=self.websocket_max_size,
+                subprotocols=subprotocols,
+                state=OPEN,
+                logger=error_logger,
+            )
+            resp: "http11.Response" = ws_conn.accept(request)
+        except Exception:
+            msg = (
+                "Failed to open a WebSocket connection.\n"
+                "See server log for more information.\n"
+            )
+            raise ServerError(msg, status_code=500)
+        if 100 <= resp.status_code <= 299:
+            rbody = "".join(
+                [
+                    "HTTP/1.1 ",
+                    str(resp.status_code),
+                    " ",
+                    resp.reason_phrase,
+                    "\r\n",
+                ]
+            )
+            rbody += "".join(f"{k}: {v}\r\n" for k, v in resp.headers.items())
+            if resp.body is not None:
+                rbody += f"\r\n{resp.body}\r\n\r\n"
+            else:
+                rbody += "\r\n"
+            await super().send(rbody.encode())
+        else:
+            raise ServerError(resp.body, resp.status_code)
+
+        self.websocket = WebsocketImplProtocol(
+            ws_conn,
+            ping_interval=self.websocket_ping_interval,
+            ping_timeout=self.websocket_ping_timeout,
+            close_timeout=self.websocket_timeout,
+        )
+        loop = (
+            request.transport.loop
+            if hasattr(request, "transport")
+            and hasattr(request.transport, "loop")
+            else None
+        )
+        await self.websocket.connection_made(self, loop=loop)
+        return self.websocket

sanic/server/runners.py

@@ -175,15 +175,11 @@ def serve(
 
         # Force close non-idle connection after waiting for
         # graceful_shutdown_timeout
-        coros = []
         for conn in connections:
             if hasattr(conn, "websocket") and conn.websocket:
-                coros.append(conn.websocket.close_connection())
+                conn.websocket.fail_connection(code=1001)
             else:
                 conn.abort()
-
-        _shutdown = asyncio.gather(*coros)
-        loop.run_until_complete(_shutdown)
         loop.run_until_complete(app._server_event("shutdown", "after"))
 
         remove_unix_socket(unix)
@@ -278,9 +274,6 @@ def _build_protocol_kwargs(
     if hasattr(protocol, "websocket_handshake"):
         return {
             "websocket_max_size": config.WEBSOCKET_MAX_SIZE,
-            "websocket_max_queue": config.WEBSOCKET_MAX_QUEUE,
-            "websocket_read_limit": config.WEBSOCKET_READ_LIMIT,
-            "websocket_write_limit": config.WEBSOCKET_WRITE_LIMIT,
             "websocket_ping_timeout": config.WEBSOCKET_PING_TIMEOUT,
             "websocket_ping_interval": config.WEBSOCKET_PING_INTERVAL,
         }