Make it possible to configure buffer sizes.

Document how backpressure and buffers work.

Refs #170.

Author: aaugustin

docs/deployment.rst

@@ -1,3 +1,76 @@
+Deployment
+==========
+
+Backpressure
+------------
+
+.. note::
+
+    This section discusses the concept of backpressure from the perspective of
+    a server but the concepts also apply to clients. The issue is symmetrical.
+
+With a naive implementation, if a server receives inputs faster than it can
+process them, or if it generates outputs faster than it can send them, data
+accumulates in buffers, eventually causing the server to run out of memory and
+crash.
+
+The solution to this problem is backpressure. Any part of the server that
+receives inputs faster than it can it can process them and send the outputs
+must propagate that information back to the previous part in the chain.
+
+``websockets`` is designed to make it easy to get backpressure right.
+
+For incoming data, ``websockets`` builds upon :class:`~asyncio.StreamReader`
+which propagates backpressure to its own buffer and to the TCP stream. Frames
+are parsed from the input stream and added to a bounded queue. If the queue
+fills up, parsing halts until some the application reads a frame.
+
+For outgoing data, ``websockets`` builds upon :class:`~asyncio.StreamWriter`
+which implements flow control. If the output buffers grow too large, it waits
+until they're drained. That's why all APIs that write frames are asynchronous
+in websockets (since version 2.0).
+
+Of course, it's still possible for an application to create its own unbounded
+buffers and break the backpressure. Be careful with queues.
+
+Buffers
+-------
+
+An asynchronous systems works best when its buffers are almost always empty.
+
+For example, if a client sends frames too fast for a server, the queue of
+incoming frames will be constantly full. The server will always be 32 frames
+(by default) behind the client. This consumes memory and adds latency for no
+good reason.
+
+If buffers are almost always full and that problem cannot be solved by adding
+capacity (typically because the system is bottlenecked by the output and
+constantly regulated by backpressure), reducing the size of buffers minimizes
+negative consequences.
+
+By default ``websockets`` has rather high limits. You can decrease them
+according to your application's characteristics.
+
+Bufferbloat can happen at every level in the stack where there is a buffer.
+The receiving side contains these buffers:
+
+- OS buffers: you shouldn't need to tune them in general.
+- :class:`~asyncio.StreamReader` bytes buffer: the default limit is 64kB.
+  You can set another limit by passing a ``read_limit`` keyword argument to
+  :func:`~websockets.client.connect` or :func:`~websockets.server.serve`.
+- ``websockets`` frame buffer: its size depends both on the size and the
+  number of frames it contains. By default the maximum size is 1MB and the
+  maximum number is 32. You can adjust these limits by setting the
+  ``max_size`` and ``max_queue`` keyword arguments of
+  :func:`~websockets.client.connect` or :func:`~websockets.server.serve`.
+
+The sending side contains these buffers:
+
+- :class:`~asyncio.StreamWriter` bytes buffer: the default size is 64kB.
+  You can set another limit by passing a ``write_limit`` keyword argument to
+  :func:`~websockets.client.connect` or :func:`~websockets.server.serve`.
+- OS buffers: you shouldn't need to tune them in general.
+
 Deployment
 ----------
 
@@ -34,7 +107,6 @@ Here's a full example (Unix-only):
 
 .. literalinclude:: ../example/shutdown.py
 
-
 It's more difficult to achieve the same effect on Windows. Some third-party
 projects try to help with this problem.
 

websockets/client.py

@@ -136,6 +136,7 @@ def handshake(self, wsuri,
 def connect(uri, *,
             klass=WebSocketClientProtocol,
             timeout=10, max_size=2 ** 20, max_queue=2 ** 5,
+            read_limit=2 ** 16, write_limit=2 ** 16,
             loop=None, legacy_recv=False,
             origin=None, subprotocols=None, extra_headers=None,
             **kwds):
@@ -154,9 +155,9 @@ def connect(uri, *,
     a ``wss://`` URI, if this argument isn't provided explicitly, it's set to
     ``True``, which means Python's default :class:`~ssl.SSLContext` is used.
 
-    The behavior of the ``timeout``, ``max_size``, and ``max_queue`` optional
-    arguments is described the documentation of
-    :class:`~websockets.protocol.WebSocketCommonProtocol`.
+    The behavior of the ``timeout``, ``max_size``, and ``max_queue``,
+    ``read_limit``, and ``write_limit`` optional arguments is described in the
+    documentation of :class:`~websockets.protocol.WebSocketCommonProtocol`.
 
     :func:`connect` also accepts the following optional arguments:
 
@@ -186,6 +187,7 @@ def connect(uri, *,
     factory = lambda: klass(
         host=wsuri.host, port=wsuri.port, secure=wsuri.secure,
         timeout=timeout, max_size=max_size, max_queue=max_queue,
+        read_limit=read_limit, write_limit=write_limit,
         loop=loop, legacy_recv=legacy_recv,
     )
 

websockets/protocol.py

@@ -78,6 +78,16 @@ class WebSocketCommonProtocol(asyncio.StreamReaderProtocol):
     this is 128MB. You may want to lower the limits, depending on your
     application's requirements.
 
+    The ``read_limit`` argument sets the high-water limit of the buffer for
+    incoming bytes. The low-water limit is half the high-water limit. The
+    default value is 64kB, half of asyncio's default (based on the current
+    implementation of :class:`~asyncio.StreamReader`).
+
+    The ``write_limit`` argument sets the high-water limit of the buffer for
+    outgoing bytes. The low-water limit is a quarter of the high-water limit.
+    The default value is 64kB, equal to asyncio's default (based on the
+    current implementation of ``_FlowControlMixin``).
+
     As soon as the HTTP request and response in the opening handshake are
     processed, the request path is available in the :attr:`path` attribute,
     and the request and response HTTP headers are available:
@@ -105,22 +115,27 @@ class WebSocketCommonProtocol(asyncio.StreamReaderProtocol):
     def __init__(self, *,
                  host=None, port=None, secure=None,
                  timeout=10, max_size=2 ** 20, max_queue=2 ** 5,
+                 read_limit=2 ** 16, write_limit=2 ** 16,
                  loop=None, legacy_recv=False):
         self.host = host
         self.port = port
         self.secure = secure
-
         self.timeout = timeout
         self.max_size = max_size
+        self.max_queue = max_queue
+        self.read_limit = read_limit
+        self.write_limit = write_limit
+
         # Store a reference to loop to avoid relying on self._loop, a private
-        # attribute of StreamReaderProtocol, inherited from FlowControlMixin.
+        # attribute of StreamReaderProtocol, inherited from _FlowControlMixin.
         if loop is None:
             loop = asyncio.get_event_loop()
         self.loop = loop
 
         self.legacy_recv = legacy_recv
 
-        stream_reader = asyncio.StreamReader(loop=loop)
+        # This limit is both the line length limit and half the buffer limit.
+        stream_reader = asyncio.StreamReader(limit=read_limit // 2, loop=loop)
         super().__init__(stream_reader, self.client_connected, loop)
 
         self.reader = None
@@ -636,6 +651,8 @@ def fail_connection(self, code=1011, reason=''):
     def client_connected(self, reader, writer):
         self.reader = reader
         self.writer = writer
+        # Configure write buffer limit.
+        self.writer._transport.set_write_buffer_limits(self.write_limit)
         # Start the task that handles incoming messages.
         self.worker_task = asyncio_ensure_future(self.run(), loop=self.loop)
 

websockets/server.py

@@ -388,6 +388,7 @@ def wait_closed(self):
 def serve(ws_handler, host=None, port=None, *,
           klass=WebSocketServerProtocol,
           timeout=10, max_size=2 ** 20, max_queue=2 ** 5,
+          read_limit=2 ** 16, write_limit=2 ** 16,
           loop=None, legacy_recv=False,
           origins=None, subprotocols=None, extra_headers=None,
           **kwds):
@@ -412,9 +413,9 @@ def serve(ws_handler, host=None, port=None, *,
     For example, you can set the ``ssl`` keyword argument to a
     :class:`~ssl.SSLContext` to enable TLS.
 
-    The behavior of the ``timeout``, ``max_size``, and ``max_queue`` optional
-    arguments is described the documentation of
-    :class:`~websockets.protocol.WebSocketCommonProtocol`.
+    The behavior of the ``timeout``, ``max_size``, and ``max_queue``,
+    ``read_limit``, and ``write_limit`` optional arguments is described in the
+    documentation of :class:`~websockets.protocol.WebSocketCommonProtocol`.
 
     :func:`serve` also accepts the following optional arguments:
 
@@ -451,6 +452,7 @@ def serve(ws_handler, host=None, port=None, *,
         ws_handler, ws_server,
         host=host, port=port, secure=secure,
         timeout=timeout, max_size=max_size, max_queue=max_queue,
+        read_limit=read_limit, write_limit=write_limit,
         loop=loop, legacy_recv=legacy_recv,
         origins=origins, subprotocols=subprotocols,
         extra_headers=extra_headers,