Repository: aaugustin/websockets
Pull request #986 (2021-06-06 18:02:45)

Everything about broadcast
Ref #653.


Patch:
diff --git a/docs/intro/index.rst b/docs/intro/index.rst
index 49e17b6..1df92ed 100644
--- a/docs/intro/index.rst
+++ b/docs/intro/index.rst
@@ -181,7 +181,7 @@ unregister them when they disconnect.
         connected.add(websocket)
         try:
             # Broadcast a message to all connected clients.
-            await asyncio.wait([ws.send("Hello!") for ws in connected])
+            websockets.broadcast(connected, "Hello!")
             await asyncio.sleep(10)
         finally:
             # Unregister.
diff --git a/docs/project/changelog.rst b/docs/project/changelog.rst
index a0cd8f0..c8553d1 100644
--- a/docs/project/changelog.rst
+++ b/docs/project/changelog.rst
@@ -51,6 +51,8 @@ They may change at any time.
 
 * Added compatibility with Python 3.10.
 
+* Added :func:`~websockets.broadcast` to send a message to many clients.
+
 * Added support for reconnecting automatically by using
   :func:`~legacy.client.connect` as an asynchronous iterator.
 
diff --git a/docs/reference/utilities.rst b/docs/reference/utilities.rst
index 198e928..f1d89ed 100644
--- a/docs/reference/utilities.rst
+++ b/docs/reference/utilities.rst
@@ -1,6 +1,11 @@
 Utilities
 =========
 
+Broadcast
+---------
+
+.. autofunction:: websockets.broadcast
+
 Data structures
 ---------------
 
diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt
index 8346ace..86a41ff 100644
--- a/docs/spelling_wordlist.txt
+++ b/docs/spelling_wordlist.txt
@@ -59,6 +59,7 @@ retransmit
 runtime
 scalable
 serializers
+stateful
 subclasses
 subclassing
 subprotocol
diff --git a/docs/topics/broadcast.rst b/docs/topics/broadcast.rst
new file mode 100644
index 0000000..c43a8fa
--- /dev/null
+++ b/docs/topics/broadcast.rst
@@ -0,0 +1,349 @@
+Broadcasting messages
+=====================
+
+.. currentmodule: websockets
+
+
+.. note::
+
+    If you just want to send a message to all connected clients, use
+    :func:`~websockets.broadcast`.
+
+    If you want to learn about its design in depth, continue reading this
+    document.
+
+WebSocket servers often send the same message to all connected clients to a
+subset of clients for which the message is relevant.
+
+Let's explore options for broadcasting a message, explain the design
+of :func:`~websockets.broadcast`, and discuss alternatives.
+
+For each option, we'll provide a connection handler called ``handler()`` and a
+function or coroutine called ``broadcast()`` that sends a message to all
+connected clients.
+
+Integrating them is left as an exercise for the reader. You could start with::
+
+    import asyncio
+    import websockets
+
+    async def handler(websocket, path):
+        ...
+
+    async def broadcast(message):
+        ...
+
+    async def broadcast_messages():
+        while True:
+            await asyncio.sleep(1)
+            message = ...  # your application logic goes here
+            await broadcast(message)
+
+    async def main():
+        async with websockets.serve(handler, "localhost", 8765):
+            await broadcast_messages()  # runs forever
+
+    if __name__ == "__main__":
+        asyncio.run(main())
+
+``broadcast_messages()`` must yield control to the event loop between each
+message, or else it will never let the server run. That's why it includes
+``await asyncio.sleep(1)``.
+
+A complete example is available in the `experiments/broadcast`_ directory.
+
+.. _experiments/broadcast: https://github.com/aaugustin/websockets/tree/main/experiments/broadcast
+
+The naive way
+-------------
+
+The most obvious way to send a message to all connected clients consists in
+keeping track of them and sending the message to each of them.
+
+Here's a connection handler that registers clients in a global variable::
+
+    CLIENTS = set()
+
+    async def handler(websocket, path):
+        CLIENTS.add(websocket)
+        try:
+            await websocket.wait_closed()
+        finally:
+            CLIENTS.remove(websocket)
+
+This implementation assumes that the client will never send any messages. If
+you'd rather not make this assumption, you can change::
+
+            await websocket.wait_closed()
+
+to::
+
+            async for _ in websocket:
+                pass
+
+Here's a coroutine that broadcasts a message to all clients::
+
+    async def broadcast(message):
+        for websocket in CLIENTS.copy():
+            try:
+                await websocket.send(message)
+            except websockets.ConnectionClosed:
+                pass
+
+There are two tricks in this version of ``broadcast()``.
+
+First, it makes a copy of ``CLIENTS`` before iterating it. Else, if a client
+connects or disconnects while ``broadcast()`` is running, the loop would fail
+with::
+
+    RuntimeError: Set changed size during iteration
+
+Second, it ignores :exc:`~exceptions.ConnectionClosed` exceptions because a
+client could disconnect between the moment ``broadcast()`` makes a copy of
+``CLIENTS`` and the moment it sends a message to this client. This is fine: a
+client that disconnected doesn't belongs to "all connected clients" anymore.
+
+The naive way can be very fast. Indeed, if all connections have enough free
+space in their write buffers, ``await websocket.send(message)`` writes the
+message and returns immediately, as it doesn't need to wait for the buffer to
+drain. In this case, ``broadcast()`` doesn't yield control to the event loop,
+which minimizes overhead.
+
+The naive way can also fail badly. If the write buffer of a connection reaches
+``write_limit``, ``broadcast()`` waits for the buffer to drain before sending
+the message to other clients. This can cause a massive drop in performance.
+
+As a consequence, this pattern works only when write buffers never fill up,
+which is usually outside of the control of the server.
+
+If you know for sure that you will never write more than ``write_limit`` bytes
+within ``ping_interval + ping_timeout``, then websockets will terminate slow
+connections before the write buffer has time to fill up.
+
+Don't set extreme ``write_limit``, ``ping_interval``, and ``ping_timeout``
+values to ensure that this condition holds. Set reasonable values and use the
+built-in :func:`~websockets.broadcast` function instead.
+
+The concurrent way
+------------------
+
+The naive way didn't work well because it serialized writes, while the whole
+point of asynchronous I/O is to perform I/O concurrently.
+
+Let's modify ``broadcast()`` to send messages concurrently::
+
+    async def send(websocket, message):
+        try:
+            await websocket.send(message)
+        except websockets.ConnectionClosed:
+            pass
+
+    def broadcast(message):
+        for websocket in CLIENTS:
+            asyncio.create_task(send(websocket, message))
+
+We move the error handling logic in a new coroutine and we schedule
+a :class:`~asyncio.Task` to run it instead of executing it immediately.
+
+Since ``broadcast()`` no longer awaits coroutines, we can make it a function
+rather than a coroutine and do away with the copy of ``CLIENTS``.
+
+This version of ``broadcast()`` makes clients independent from one another: a
+slow client won't block others. As a side effect, it makes messages
+independent from one another.
+
+If you broadcast several messages, there is no strong guarantee that they will
+be sent in the expected order. Fortunately, the event loop runs tasks in the
+order in which they are created, so the order is correct in practice.
+
+Technically, this is an implementation detail of the event loop. However, it
+seems unlikely for an event loop to run tasks in an order other than FIFO.
+
+If you wanted to enforce the order without relying this implementation detail,
+you could be tempted to wait until all clients have received the message::
+
+    async def broadcast(message):
+        if CLIENTS:  # asyncio.wait doesn't accept an empty list
+            await asyncio.wait([
+                asyncio.create_task(send(websocket, message))
+                for websocket in CLIENTS
+            ])
+
+However, this doesn't really work in practice. Quite often, it will block
+until the slowest client times out.
+
+Backpressure meets broadcast
+----------------------------
+
+At this point, it becomes apparent that backpressure, usually a good practice,
+doesn't work well when broadcasting a message to thousands of clients.
+
+When you're sending messages to a single client, you don't want to send them
+faster than the network can transfer them and the client accept them. This is
+why :meth:`~server.WebSocketServerProtocol.send` checks if the write buffer
+is full and, if it is, waits until it drain, giving the network and the
+client time to catch up. This provides backpressure.
+
+Without backpressure, you could pile up data in the write buffer until the
+server process runs out of memory and the operating system kills it.
+
+The :meth:`~server.WebSocketServerProtocol.send` API is designed to enforce
+backpressure by default. This helps users of websockets write robust programs
+even if they never heard about backpressure.
+
+For comparison, :class:`asyncio.StreamWriter` requires users to understand
+backpressure and to await :meth:`~asyncio.StreamWriter.drain` explicitly
+after each :meth:`~asyncio.StreamWriter.write`.
+
+When broadcasting messages, backpressure consists in slowing down all clients
+in an attempt to let the slowest client catch up. With thousands of clients,
+the slowest one is probably timing out and isn't going to receive the message
+anyway. So it doesn't make sense to synchronize with the slowest client.
+
+How do we avoid running out of memory when slow clients can't keep up with the
+broadcast rate, then? The most straightforward option is to disconnect them.
+
+If a client gets too far behind, eventually it reaches the limit defined by
+``ping_timeout`` and websockets terminates the connection. You can read the
+discussion of :doc:`keepalive and timeouts <./timeouts>` for details.
+
+How :func:`~websockets.broadcast` works
+---------------------------------------
+
+The built-in :func:`~websockets.broadcast` function is similar to the naive
+way. The main difference is that it doesn't apply backpressure.
+
+This provides the best performance by avoiding the overhead of scheduling and
+running one task per client.
+
+Also, when sending text messages, encoding to UTF-8 happens only once rather
+than once per client, providing a small performance gain.
+
+Per-client queues
+-----------------
+
+At this point, we deal with slow clients rather brutally: we disconnect then.
+
+Can we do better? For example, we could decide to skip or to batch messages,
+depending on how far behind a client is.
+
+To implement this logic, we can create a queue of messages for each client and
+run a task that gets messages from the queue and sends them to the client::
+
+    import asyncio
+
+    CLIENTS = set()
+
+    async def relay(queue, websocket):
+        while True:
+            # Implement custom logic based on queue.qsize() and
+            # websocket.transport.get_write_buffer_size() here.
+            message = await queue.get()
+            await websocket.send(message)
+
+    async def handler(websocket, path):
+        queue = asyncio.Queue()
+        relay_task = asyncio.create_task(relay(queue, websocket))
+        CLIENTS.add(queue)
+        try:
+            await websocket.wait_closed()
+        finally:
+            CLIENTS.remove(queue)
+            relay_task.cancel()
+
+Then we can broadcast a message by pushing it to all queues::
+
+    def broadcast(message):
+        for queue in CLIENTS:
+            queue.put_nowait(message)
+
+The queues provide an additional buffer between the ``broadcast()`` function
+and clients. This makes it easier to support slow clients without excessive
+memory usage because queued messages aren't duplicated to  write buffers
+until ``relay()`` processes them.
+
+Publish–subscribe
+-----------------
+
+Can we avoid centralizing the list of connected clients in a global variable?
+
+If each client subscribes to a stream a messages, then broadcasting becomes as
+simple as publishing a message to the stream.
+
+Here's a message stream that supports multiple consumers::
+
+    class PubSub:
+        def __init__(self):
+            self.waiter = asyncio.Future()
+
+        def publish(self, value):
+            waiter, self.waiter = self.waiter, asyncio.Future()
+            waiter.set_result((value, self.waiter))
+
+        async def subscribe(self):
+            waiter = self.waiter
+            while True:
+                value, waiter = await waiter
+                yield value
+
+        __aiter__ = subscribe
+
+    PUBSUB = PubSub()
+
+The stream is implemented as a linked list of futures. It isn't necessary to
+synchronize consumers. They can read the stream at their own pace,
+independently from one another. Once all consumers read a message, there are
+no references left, therefore the garbage collector deletes it.
+
+The connection handler subscribes to the stream and sends messages::
+
+    async def handler(websocket, path):
+        async for message in PUBSUB:
+            await websocket.send(message)
+
+The broadcast function publishes to the stream::
+
+    def broadcast(message):
+        PUBSUB.publish(message)
+
+Like per-client queues, this version supports slow clients with limited memory
+usage. Unlike per-client queues, it makes it difficult to tell how far behind
+a client is. The ``PubSub`` class could be extended or refactored to provide
+this information.
+
+The ``for`` loop is gone from this version of the ``broadcast()`` function.
+However, there's still a ``for`` loop iterating on all clients hidden deep
+inside :mod:`asyncio`. When ``publish()`` sets the result of the ``waiter``
+future, :mod:`asyncio` loops on callbacks registered with this future and
+schedules them. This is how connection handlers receive the next value from
+the asynchronous iterator returned by ``subscribe()``.
+
+Performance considerations
+--------------------------
+
+The built-in :func:`~websockets.broadcast` function sends all messages without
+yielding control to the event loop. So does the naive way when the network
+and clients are fast and reliable.
+
+For each client, a WebSocket frame is prepared and sent to the network. This
+is the minimum amount of work required to broadcast a message.
+
+It would be tempting to prepare a frame and reuse it for all connections.
+However, this isn't possible in general for two reasons:
+
+* Clients can negotiate different extensions. You would have to enforce the
+  same extensions with the same parameters. For example, you would have to
+  select some compression settings and reject clients that cannot support
+  these settings.
+
+* Extensions can be stateful, producing different encodings of the same
+  message depending on previous messages. For example, you would have to
+  disable context takeover to make compression stateless, resulting in poor
+  compression rates.
+
+All other patterns discussed above yield control to the event loop once per
+client because messages are sent by different tasks. This makes them slower
+than the built-in :func:`~websockets.broadcast` function.
+
+There is no major difference between the performance of per-message queues and
+publish–subscribe.
diff --git a/docs/topics/index.rst b/docs/topics/index.rst
index 9269c58..dc192a2 100644
--- a/docs/topics/index.rst
+++ b/docs/topics/index.rst
@@ -6,6 +6,7 @@ Topics
 
    deployment
    authentication
+   broadcast
    compression
    timeouts
    design
diff --git a/example/counter.py b/example/counter.py
index a9ed618..e41f6fa 100755
--- a/example/counter.py
+++ b/example/counter.py
@@ -22,23 +22,11 @@ def users_event():
     return json.dumps({"type": "users", "count": len(USERS)})
 
 
-async def broadcast(message):
-    # asyncio.wait doesn't accept an empty list
-    if not USERS:
-        return
-    # Ignore return value. If a user disconnects before we send
-    # the message to them, there's nothing we can do anyway.
-    await asyncio.wait([
-        asyncio.create_task(user.send(message))
-        for user in USERS
-    ])
-
-
 async def counter(websocket, path):
     try:
         # Register user
         USERS.add(websocket)
-        await broadcast(users_event())
+        websockets.broadcast(USERS, users_event())
         # Send current state to user
         await websocket.send(state_event())
         # Manage state changes
@@ -46,16 +34,16 @@ async def counter(websocket, path):
             data = json.loads(message)
             if data["action"] == "minus":
                 STATE["value"] -= 1
-                await broadcast(state_event())
+                websockets.broadcast(USERS, state_event())
             elif data["action"] == "plus":
                 STATE["value"] += 1
-                await broadcast(state_event())
+                websockets.broadcast(USERS, state_event())
             else:
                 logging.error("unsupported event: %s", data)
     finally:
         # Unregister user
         USERS.remove(websocket)
-        await broadcast(users_event())
+        websockets.broadcast(USERS, users_event())
 
 
 async def main():
diff --git a/example/django/notifications.py b/example/django/notifications.py
index 641643f..41fb719 100644
--- a/example/django/notifications.py
+++ b/example/django/notifications.py
@@ -55,9 +55,12 @@ async def process_events():
         payload = message["data"].decode()
         # Broadcast event to all users who have permissions to see it.
         event = json.loads(payload)
-        for websocket, connection in CONNECTIONS.items():
-            if event["content_type_id"] in connection["content_type_ids"]:
-                asyncio.create_task(websocket.send(payload))
+        recipients = (
+            websocket
+            for websocket, connection in CONNECTIONS.items()
+            if event["content_type_id"] in connection["content_type_ids"]
+        )
+        websockets.broadcast(recipients, payload)
 
 
 async def main():
diff --git a/experiments/broadcast/clients.py b/experiments/broadcast/clients.py
new file mode 100644
index 0000000..fe39dfe
--- /dev/null
+++ b/experiments/broadcast/clients.py
@@ -0,0 +1,61 @@
+#!/usr/bin/env python
+
+import asyncio
+import statistics
+import sys
+import time
+
+import websockets
+
+
+LATENCIES = {}
+
+
+async def log_latency(interval):
+    while True:
+        await asyncio.sleep(interval)
+        p = statistics.quantiles(LATENCIES.values(), n=100)
+        print(f"clients = {len(LATENCIES)}")
+        print(
+            f"p50 = {p[49] / 1e6:.1f}ms, "
+            f"p95 = {p[94] / 1e6:.1f}ms, "
+            f"p99 = {p[98] / 1e6:.1f}ms"
+        )
+        print()
+
+
+async def client():
+    try:
+        async with websockets.connect(
+            "ws://localhost:8765",
+            ping_timeout=None,
+        ) as websocket:
+            async for msg in websocket:
+                client_time = time.time_ns()
+                server_time = int(msg[:19].decode())
+                LATENCIES[websocket] = client_time - server_time
+    except Exception as exc:
+        print(exc)
+
+
+async def main(count, interval):
+    asyncio.create_task(log_latency(interval))
+    clients = []
+    for _ in range(count):
+        clients.append(asyncio.create_task(client()))
+        await asyncio.sleep(0.001)  # 1ms between each connection
+    await asyncio.wait(clients)
+
+
+if __name__ == "__main__":
+    try:
+        count = int(sys.argv[1])
+        interval = float(sys.argv[2])
+    except Exception as exc:
+        print(f"Usage: {sys.argv[0]} count interval")
+        print("    Connect <count> clients e.g. 1000")
+        print("    Report latency every <interval> seconds e.g. 1")
+        print()
+        print(exc)
+    else:
+        asyncio.run(main(count, interval))
diff --git a/experiments/broadcast/server.py b/experiments/broadcast/server.py
new file mode 100644
index 0000000..355d6b5
--- /dev/null
+++ b/experiments/broadcast/server.py
@@ -0,0 +1,153 @@
+#!/usr/bin/env python
+
+import asyncio
+import functools
+import os
+import sys
+import time
+
+import websockets
+
+
+CLIENTS = set()
+
+
+async def send(websocket, message):
+    try:
+        await websocket.send(message)
+    except websockets.ConnectionClosed:
+        pass
+
+
+async def relay(queue, websocket):
+    while True:
+        message = await queue.get()
+        await websocket.send(message)
+
+
+class PubSub:
+    def __init__(self):
+        self.waiter = asyncio.Future()
+
+    def publish(self, value):
+        waiter, self.waiter = self.waiter, asyncio.Future()
+        waiter.set_result((value, self.waiter))
+
+    async def subscribe(self):
+        waiter = self.waiter
+        while True:
+            value, waiter = await waiter
+            yield value
+
+    __aiter__ = subscribe
+
+
+PUBSUB = PubSub()
+
+
+async def handler(websocket, path, method=None):
+    if method in ["default", "naive", "task", "wait"]:
+        CLIENTS.add(websocket)
+        try:
+            await websocket.wait_closed()
+        finally:
+            CLIENTS.remove(websocket)
+    elif method == "queue":
+        queue = asyncio.Queue()
+        relay_task = asyncio.create_task(relay(queue, websocket))
+        CLIENTS.add(queue)
+        try:
+            await websocket.wait_closed()
+        finally:
+            CLIENTS.remove(queue)
+            relay_task.cancel()
+    elif method == "pubsub":
+        async for message in PUBSUB:
+            await websocket.send(message)
+    else:
+        raise NotImplementedError(f"unsupported method: {method}")
+
+
+async def broadcast(method, size, delay):
+    """Broadcast messages at regular intervals."""
+    load_average = 0
+    time_average = 0
+    pc1, pt1 = time.perf_counter_ns(), time.process_time_ns()
+    await asyncio.sleep(delay)
+    while True:
+        print(f"clients = {len(CLIENTS)}")
+        pc0, pt0 = time.perf_counter_ns(), time.process_time_ns()
+        load_average = 0.9 * load_average + 0.1 * (pt0 - pt1) / (pc0 - pc1)
+        print(
+            f"load = {(pt0 - pt1) / (pc0 - pc1) * 100:.1f}% / "
+            f"average = {load_average * 100:.1f}%, "
+            f"late = {(pc0 - pc1 - delay * 1e9) / 1e6:.1f} ms"
+        )
+        pc1, pt1 = pc0, pt0
+
+        assert size > 20
+        message = str(time.time_ns()).encode() + b" " + os.urandom(size - 20)
+
+        if method == "default":
+            websockets.broadcast(CLIENTS, message)
+        elif method == "naive":
+            # Since the loop can yield control, make a copy of CLIENTS
+            # to avoid: RuntimeError: Set changed size during iteration
+            for websocket in CLIENTS.copy():
+                await send(websocket, message)
+        elif method == "task":
+            for websocket in CLIENTS:
+                asyncio.create_task(send(websocket, message))
+        elif method == "wait":
+            if CLIENTS:  # asyncio.wait doesn't accept an empty list
+                await asyncio.wait(
+                    [
+                        asyncio.create_task(send(websocket, message))
+                        for websocket in CLIENTS
+                    ]
+                )
+        elif method == "queue":
+            for queue in CLIENTS:
+                queue.put_nowait(message)
+        elif method == "pubsub":
+            PUBSUB.publish(message)
+        else:
+            raise NotImplementedError(f"unsupported method: {method}")
+
+        pc2 = time.perf_counter_ns()
+        wait = delay + (pc1 - pc2) / 1e9
+        time_average = 0.9 * time_average + 0.1 * (pc2 - pc1)
+        print(
+            f"broadcast = {(pc2 - pc1) / 1e6:.1f}ms / "
+            f"average = {time_average / 1e6:.1f}ms, "
+            f"wait = {wait * 1e3:.1f}ms"
+        )
+        await asyncio.sleep(wait)
+        print()
+
+
+async def main(method, size, delay):
+    async with websockets.serve(
+        functools.partial(handler, method=method),
+        "localhost",
+        8765,
+        compression=None,
+        ping_timeout=None,
+    ):
+        await broadcast(method, size, delay)
+
+
+if __name__ == "__main__":
+    try:
+        method = sys.argv[1]
+        assert method in ["default", "naive", "task", "wait", "queue", "pubsub"]
+        size = int(sys.argv[2])
+        delay = float(sys.argv[3])
+    except Exception as exc:
+        print(f"Usage: {sys.argv[0]} method size delay")
+        print("    Start a server broadcasting messages with <method> e.g. naive")
+        print("    Send a payload of <size> bytes every <delay> seconds")
+        print()
+        print(exc)
+    else:
+        asyncio.run(main(method, size, delay))
diff --git a/src/websockets/__init__.py b/src/websockets/__init__.py
index 8c69a6d..6378b82 100644
--- a/src/websockets/__init__.py
+++ b/src/websockets/__init__.py
@@ -6,6 +6,7 @@ __all__ = [  # noqa
     "AbortHandshake",
     "basic_auth_protocol_factory",
     "BasicAuthWebSocketServerProtocol",
+    "broadcast",
     "ClientConnection",
     "connect",
     "ConnectionClosed",
@@ -56,6 +57,7 @@ lazy_import(
         "auth": ".legacy",
         "basic_auth_protocol_factory": ".legacy.auth",
         "BasicAuthWebSocketServerProtocol": ".legacy.auth",
+        "broadcast": ".legacy.protocol",
         "ClientConnection": ".client",
         "connect": ".legacy.client",
         "unix_connect": ".legacy.client",
diff --git a/src/websockets/legacy/protocol.py b/src/websockets/legacy/protocol.py
index c7dd4d2..8a95577 100644
--- a/src/websockets/legacy/protocol.py
+++ b/src/websockets/legacy/protocol.py
@@ -61,7 +61,7 @@ from .compatibility import loop_if_py_lt_38
 from .framing import Frame
 
 
-__all__ = ["WebSocketCommonProtocol"]
+__all__ = ["WebSocketCommonProtocol", "broadcast"]
 
 
 # In order to ensure consistency, the code always checks the current value of
@@ -974,15 +974,7 @@ class WebSocketCommonProtocol(asyncio.Protocol):
             self.logger.debug("< %s", frame)
         return frame
 
-    async def write_frame(
-        self, fin: bool, opcode: int, data: bytes, *, _expected_state: int = State.OPEN
-    ) -> None:
-        # Defensive assertion for protocol compliance.
-        if self.state is not _expected_state:  # pragma: no cover
-            raise InvalidState(
-                f"Cannot write to a WebSocket in the {self.state.name} state"
-            )
-
+    def write_frame_sync(self, fin: bool, opcode: int, data: bytes) -> None:
         frame = Frame(fin, Opcode(opcode), data)
         if self.debug:
             self.logger.debug("> %s", frame)
@@ -992,6 +984,7 @@ class WebSocketCommonProtocol(asyncio.Protocol):
             extensions=self.extensions,
         )
 
+    async def drain(self) -> None:
         try:
             # drain() cannot be called concurrently by multiple coroutines:
             # http://bugs.python.org/issue29930. Remove this lock when no
@@ -1006,6 +999,17 @@ class WebSocketCommonProtocol(asyncio.Protocol):
             # with the correct code and reason.
             await self.ensure_open()
 
+    async def write_frame(
+        self, fin: bool, opcode: int, data: bytes, *, _state: int = State.OPEN
+    ) -> None:
+        # Defensive assertion for protocol compliance.
+        if self.state is not _state:  # pragma: no cover
+            raise InvalidState(
+                f"Cannot write to a WebSocket in the {self.state.name} state"
+            )
+        self.write_frame_sync(fin, opcode, data)
+        await self.drain()
+
     async def write_close_frame(self, data: bytes = b"") -> None:
         """
         Write a close frame if and only if the connection state is OPEN.
@@ -1023,7 +1027,7 @@ class WebSocketCommonProtocol(asyncio.Protocol):
                 self.logger.debug("= connection is CLOSING")
 
             # 7.1.2. Start the WebSocket Closing Handshake
-            await self.write_frame(True, OP_CLOSE, data, _expected_state=State.CLOSING)
+            await self.write_frame(True, OP_CLOSE, data, _state=State.CLOSING)
 
     async def keepalive_ping(self) -> None:
         """
@@ -1371,3 +1375,50 @@ class WebSocketCommonProtocol(asyncio.Protocol):
 
         """
         self.reader.feed_eof()
+
+
+def broadcast(websockets: Iterable[WebSocketCommonProtocol], message: Data) -> None:
+    """
+    Broadcast a message to several WebSocket connections.
+
+    A string (:class:`str`) is sent as a `Text frame`_. A bytestring or
+    bytes-like object (:class:`bytes`, :class:`bytearray`, or
+    :class:`memoryview`) is sent as a `Binary frame`_.
+
+    .. _Text frame: https://tools.ietf.org/html/rfc6455#section-5.6
+    .. _Binary frame: https://tools.ietf.org/html/rfc6455#section-5.6
+
+    :func:`broadcast` pushes the message synchronously to all connections even
+    if their write buffers overflow ``write_limit``. There's no backpressure.
+
+    :func:`broadcast` skips silently connections that aren't open in order to
+    avoid errors on connections where the closing handshake is in progress.
+
+    If you broadcast messages faster than a connection can handle them,
+    messages will pile up in its write buffer until the connection times out.
+    Keep low values for ``ping_interval`` and ``ping_timeout`` to prevent
+    excessive memory use by slow connections when you use :func:`broadcast`.
+
+    Unlike :meth:`~websockets.server.WebSocketServerProtocol.send`,
+    :func:`broadcast` doesn't support sending fragmented messages. Indeed,
+    fragmentation is useful for sending large messages without buffering
+    them in memory, while :func:`broadcast` buffers one copy per connection
+    as fast as possible.
+
+    :raises RuntimeError: if a connection is busy sending a fragmented message
+    :raises TypeError: for unsupported inputs
+
+    """
+    if not isinstance(message, (str, bytes, bytearray, memoryview)):
+        raise TypeError("data must be bytes, str, or iterable")
+
+    opcode, data = prepare_data(message)
+
+    for websocket in websockets:
+        if websocket.state is not State.OPEN:
+            continue
+
+        if websocket._fragmented_message_waiter is not None:
+            raise RuntimeError("busy sending a fragmented message")
+
+        websocket.write_frame_sync(True, opcode, data)
diff --git a/src/websockets/legacy/server.py b/src/websockets/legacy/server.py
index 10416ab..ae749aa 100644
--- a/src/websockets/legacy/server.py
+++ b/src/websockets/legacy/server.py
@@ -836,8 +836,8 @@ class WebSocketServer:
         await asyncio.sleep(0, **loop_if_py_lt_38(self.loop))
 
         # Close OPEN connections with status code 1001. Since the server was
-        # closed, handshake() closes OPENING conections with a HTTP 503 error.
-        # Wait until all connections are closed.
+        # closed, handshake() closes OPENING connections with a HTTP 503
+        # error. Wait until all connections are closed.
 
         # asyncio.wait doesn't accept an empty first argument
         if self.websockets:


Test patch:
diff --git a/tests/legacy/test_protocol.py b/tests/legacy/test_protocol.py
index 11589ba..97aa015 100644
--- a/tests/legacy/test_protocol.py
+++ b/tests/legacy/test_protocol.py
@@ -18,7 +18,7 @@ from websockets.frames import (
 )
 from websockets.legacy.compatibility import loop_if_py_lt_38
 from websockets.legacy.framing import Frame
-from websockets.legacy.protocol import WebSocketCommonProtocol
+from websockets.legacy.protocol import WebSocketCommonProtocol, broadcast
 
 from .utils import MS, AsyncioTestCase
 
@@ -1388,6 +1388,52 @@ class CommonTests:
     # There is no test_local_close_during_send because this cannot really
     # happen, considering that writes are serialized.
 
+    def test_broadcast_text(self):
+        broadcast([self.protocol], "café")
+        self.assertOneFrameSent(True, OP_TEXT, "café".encode("utf-8"))
+
+    def test_broadcast_binary(self):
+        broadcast([self.protocol], b"tea")
+        self.assertOneFrameSent(True, OP_BINARY, b"tea")
+
+    def test_broadcast_type_error(self):
+        with self.assertRaises(TypeError):
+            broadcast([self.protocol], ["ca", "fé"])
+
+    def test_broadcast_no_clients(self):
+        broadcast([], "café")
+        self.assertNoFrameSent()
+
+    def test_broadcast_two_clients(self):
+        broadcast([self.protocol, self.protocol], "café")
+        self.assertFramesSent(
+            (True, OP_TEXT, "café".encode("utf-8")),
+            (True, OP_TEXT, "café".encode("utf-8")),
+        )
+
+    def test_broadcast_skips_closed_connection(self):
+        self.close_connection()
+
+        broadcast([self.protocol], "café")
+        self.assertNoFrameSent()
+
+    def test_broadcast_skips_closing_connection(self):
+        close_task = self.half_close_connection_local()
+
+        broadcast([self.protocol], "café")
+        self.assertNoFrameSent()
+
+        self.loop.run_until_complete(close_task)  # cleanup
+
+    def test_broadcast_within_fragmented_text(self):
+        self.make_drain_slow()
+        self.loop.create_task(self.protocol.send(["ca", "fé"]))
+        self.run_loop_once()
+        self.assertOneFrameSent(False, OP_TEXT, "ca".encode("utf-8"))
+
+        with self.assertRaises(RuntimeError):
+            broadcast([self.protocol], "café")
+
 
 class ServerTests(CommonTests, AsyncioTestCase):
     def setUp(self):