diff options
Diffstat (limited to 'packages/meshbay-node')
4 files changed, 293 insertions, 2 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py index d197044..07820b2 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py @@ -15,6 +15,7 @@ from meshbay_common.protocol import MNP from meshbay_node import ops from meshbay_node import transfers as transfers_mod from meshbay_node.transport.webrtc import ( + proactor_udp, # noqa: F401 (patches asyncio's Windows loop on import) sctp_fix, # noqa: F401 (patches aiortc on import) sctp_stats, # noqa: F401 (patches aiortc on import) udp_buffer, # noqa: F401 (patches aioice on import) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/proactor_udp.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/proactor_udp.py new file mode 100644 index 0000000..38cb97c --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/proactor_udp.py @@ -0,0 +1,139 @@ +"""Datagrams moved straight through the socket on Windows' event loop, applied +on import. + +The node runs on the ProactorEventLoop, the one that can start ffmpeg there +(`platform.configure_event_loop`), and so do its connection processes. Its +datagram transport keeps one overlapped operation in each direction and starts +the next only from the completion callback of the one before, which runs on a +later turn of the loop. So a socket sends **one datagram per turn of the loop, +and receives one**, however many are waiting, and a turn lasts as long as +everything else the loop has to do: the other connections, reading the file, +the hub. The busier the node, the slower every connection, with cores to spare. + +Measured on loopback, 1200-byte datagrams, with the loop otherwise busy 0.5 ms +a turn: 1,928 sent a second (2.3 MB/s), and 1,954 received, the kernel dropping +87 % of what arrived; straight through the socket, 88,000 sent, and every +datagram received at the 20,000 a second it was sent at. In the node holding the +connections itself, a browser's download over eight connections ran at 9.1 MB/s +with the node at 45 % of a core and an SCTP round trip of 745 ms on loopback: +that is the queue in front of each socket. In a connection process serving the +command-line client, 42 % of the time was `IocpProactor.sendto`. Linux never +paid for the sending half: the selector loop's transport tries the socket first +and queues only what it refuses. + +Sending: a datagram goes out with a non-blocking `sendto` when nothing is queued +and no overlapped send is pending, so order is kept; anything else, the socket +refusing included, goes through the proactor's own queue as before. + +Receiving: once an overlapped read completes, its datagram is delivered, then +whatever else the socket holds is read with non-blocking `recvfrom`, and only +then is the next overlapped read posted. Never while one is pending, which +could hand a later datagram to the completion port ahead of the ones read here. +The overlapped path turns the ICMP "port unreachable" a send to a silent +address brings back into an empty read; read directly it is a +ConnectionResetError, and it is skipped the same way. + +No effect off Windows, where the class is never used. +""" + +import sys +from asyncio import exceptions, proactor_events + +_Transport = proactor_events._ProactorDatagramTransport +_proactor_sendto = _Transport.sendto + +# Datagrams read straight from the socket after one completion, at most: a +# socket under a flood still lets the loop turn. +DRAIN_MAX = 256 + + +def sendto(self, data, addr=None) -> None: + if (self._write_fut is None and not self._buffer and not self._conn_lost + and isinstance(data, (bytes, bytearray, memoryview)) and data + and (self._address is None or addr in (None, self._address))): + try: + if self._address is not None: + self._sock.send(data) + else: + self._sock.sendto(data, addr) + return + except (BlockingIOError, InterruptedError): + pass + except OSError as exc: + # What the overlapped path does with a failed send. + self._protocol.error_received(exc) + return + _proactor_sendto(self, data, addr) + + +def _deliver(self, data, addr) -> None: + """To the protocol, and whatever it raises reported as a callback's would + be: in CPython's order the next read was posted first, and an exception + here must not end the reading either.""" + try: + self._protocol.datagram_received(data, addr) + except (SystemExit, KeyboardInterrupt): + raise + except BaseException as exc: + self._loop.call_exception_handler({ + "message": "Exception in datagram_received", "exception": exc, + "transport": self, "protocol": self._protocol}) + + +def _drain(self) -> None: + for _ in range(DRAIN_MAX): + if self._conn_lost or self._closing: + return + try: + if self._address is not None: + data, addr = self._sock.recv(self.max_size), self._address + else: + data, addr = self._sock.recvfrom(self.max_size) + except (BlockingIOError, InterruptedError): + return + except ConnectionResetError: + continue + except OSError as exc: + # The overlapped read posted next meets it again, and reports it + # the proactor's way. + self._protocol.error_received(exc) + return + if data: + _deliver(self, data, addr) + + +def _loop_reading(self, fut=None) -> None: + """CPython's own (3.12), but a completed read is delivered and the socket + drained before the next overlapped read is posted, not after.""" + try: + if self._conn_lost: + return + assert self._read_fut is fut or (self._read_fut is None and self._closing) + self._read_fut = None + if fut is not None: + res = fut.result() + if self._closing: + return + data, addr = (res, self._address) if self._address is not None else res + if data: + _deliver(self, data, addr) + _drain(self) + if self._conn_lost: + return + if self._address is not None: + self._read_fut = self._loop._proactor.recv(self._sock, self.max_size) + else: + self._read_fut = self._loop._proactor.recvfrom(self._sock, self.max_size) + except OSError as exc: + self._protocol.error_received(exc) + except exceptions.CancelledError: + if not self._closing: + raise + else: + if self._read_fut is not None: + self._read_fut.add_done_callback(self._loop_reading) + + +if sys.platform == "win32": + _Transport.sendto = sendto + _Transport._loop_reading = _loop_reading diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py index 832428a..c933aba 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py @@ -224,9 +224,10 @@ async def _main() -> None: root.handlers[:] = [_Forward(writer.write)] root.setLevel(hello.get("level", logging.INFO)) - # Patches aiortc on import, as the node does for the connections it holds - # itself. + # Patches aiortc, and asyncio's Windows loop, on import, as the node does + # for the connections it holds itself. from meshbay_node.transport.webrtc import ( # noqa: F401 + proactor_udp, sctp_fix, sctp_stats, udp_buffer, diff --git a/packages/meshbay-node/tests/test_proactor_udp.py b/packages/meshbay-node/tests/test_proactor_udp.py new file mode 100644 index 0000000..0b22b4e --- /dev/null +++ b/packages/meshbay-node/tests/test_proactor_udp.py @@ -0,0 +1,150 @@ +""" +On Windows the node's datagrams go straight through the socket, not one +overlapped operation per datagram and per turn of the loop (`proactor_udp.py`). + +Run on a real ProactorEventLoop, which only Windows has. +""" + +import asyncio +import socket +import sys +import time + +import pytest +from meshbay_node.transport.webrtc import proactor_udp + +pytestmark = pytest.mark.skipif(sys.platform != "win32", reason="the proactor is Windows'") + +N = 2000 + + +class _Collect(asyncio.DatagramProtocol): + def __init__(self, raise_on: int = -1): + self.got: list[int] = [] + self.raise_on = raise_on + self.all = asyncio.get_running_loop().create_future() + + def datagram_received(self, data, addr): + self.got.append(int.from_bytes(data[:4], "big")) + if len(self.got) == N and not self.all.done(): + self.all.set_result(None) + if len(self.got) == self.raise_on: + raise RuntimeError("a protocol's own bug") + + +class _Refusing(socket.socket): + """A socket that refuses a send every so often, as a full buffer does, and + can answer one read with the reset an ICMP port unreachable leaves. The + overlapped path works by handle and never calls these.""" + + def __init__(self, every: int = 0, reset_once: bool = False): + super().__init__(socket.AF_INET, socket.SOCK_DGRAM) + self.every, self.calls, self.reset_once = every, 0, reset_once + + def sendto(self, data, addr): + self.calls += 1 + if self.every and self.calls % self.every == 0: + raise BlockingIOError + return super().sendto(data, addr) + + def recvfrom(self, n): + if self.reset_once: + self.reset_once = False + raise ConnectionResetError(10054, "port unreachable") + return super().recvfrom(n) + + +def _busy_turns(ms: float): + """A loop with other work to do: every turn takes `ms` longer.""" + async def busy(): + while True: + end = time.perf_counter() + ms / 1000 + while time.perf_counter() < end: + pass + await asyncio.sleep(0) + return asyncio.ensure_future(busy()) + + +async def _exchange(*, refuse_every=0, reset_once=False, raise_on=-1, busy_ms=0.0): + """N datagrams from one proactor transport to another on loopback: what + arrived in what order, and how many overlapped sends and reads it took.""" + loop = asyncio.get_running_loop() + rsock = _Refusing(reset_once=reset_once) + rsock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 8 << 20) + rsock.bind(("127.0.0.1", 0)) + rtr, rx = await loop.create_datagram_endpoint(lambda: _Collect(raise_on), sock=rsock) + tsock = _Refusing(refuse_every) + tsock.bind(("127.0.0.1", 0)) + tx, _ = await loop.create_datagram_endpoint(asyncio.DatagramProtocol, sock=tsock) + counts = {"sendto": 0, "recvfrom": 0} + originals = {name: getattr(loop._proactor, name) for name in counts} + + def counting(name): + def call(*a, **kw): + counts[name] += 1 + return originals[name](*a, **kw) + return call + for name in counts: + setattr(loop._proactor, name, counting(name)) + errors = [] + loop.set_exception_handler(lambda _loop, ctx: errors.append(ctx)) + busy = _busy_turns(busy_ms) if busy_ms else None + try: + dest = rsock.getsockname() + for i in range(N): + tx.sendto(i.to_bytes(4, "big") * 50, dest) + if i % 50 == 49: + await asyncio.sleep(0) + await asyncio.wait_for(rx.all, 20) + return rx.got, counts, errors + finally: + if busy: + busy.cancel() + for name, fn in originals.items(): + setattr(loop._proactor, name, fn) + tx.close() + rtr.close() + + +def _run(**kw): + return asyncio.run(_exchange(**kw), loop_factory=asyncio.ProactorEventLoop) + + +def test_the_transport_is_patched(): + cls = asyncio.proactor_events._ProactorDatagramTransport + assert cls.sendto is proactor_udp.sendto + assert cls._loop_reading is proactor_udp._loop_reading + + +def test_datagrams_go_straight_out_in_order(): + got, counts, _ = _run() + assert got == list(range(N)) + assert counts["sendto"] == 0 + + +def test_what_the_socket_refuses_is_queued_and_order_is_kept(): + """A refusal sends that datagram the overlapped way, and the ones after it + wait behind it rather than overtake it.""" + got, counts, _ = _run(refuse_every=97) + assert got == list(range(N)) + assert counts["sendto"] >= N // 97 + + +def test_a_busy_loop_reads_everything_waiting_in_order(): + """Unpatched, this loop read one datagram a turn: about 2,000 a second + whatever was waiting, and the kernel dropped the rest.""" + got, counts, _ = _run(busy_ms=2) + assert got == list(range(N)) + assert counts["recvfrom"] < N // 10 + + +def test_a_port_unreachable_on_a_direct_read_is_skipped(): + got, _, errors = _run(reset_once=True, busy_ms=2) + assert got == list(range(N)) + assert not errors + + +def test_a_protocol_that_raises_does_not_end_the_reading(): + got, _, errors = _run(raise_on=10) + assert got == list(range(N)) + assert [str(e["exception"]) for e in errors] == ["a protocol's own bug"] |