diff options
Diffstat (limited to 'packages/meshbay-node')
9 files changed, 633 insertions, 30 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py index c5596f1..d46ab8c 100644 --- a/packages/meshbay-node/src/meshbay_node/config.py +++ b/packages/meshbay-node/src/meshbay_node/config.py @@ -74,13 +74,14 @@ max_concurrent_uploads = 8 # the partial file is deleted. max_upload_gb = 8 -# How many processes carry the connections of members. At 1 the node does it +# How many processes carry the connections of members. Left unset: one per core +# of this machine, up to four, about 85 MB of memory each. At 1 the node does it # all in one process, on one core: about 35 MB/s summed over every transfer in -# progress. Above 1 the node starts that many processes for the connections and -# keeps the rest for itself, so transfers use that many cores, plus part of one -# for the node. Takes effect for new connections, without a restart. Not above -# the number of cores of this machine. -connection_processes = 1 +# progress on Linux, and much less on Windows. Above 1 the node starts that many +# processes for the connections and keeps the rest for itself, so transfers use +# that many cores, plus part of one for the node. Takes effect for new +# connections, without a restart. Not above the number of cores of this machine. +# connection_processes = 4 # HEVC sources have no browser decoder on most platforms, so streaming one is # transcoded to H264 rather than the usual free copy — real CPU per viewer. @@ -202,8 +203,9 @@ class NodeConfig: # §15.3), so this is what stands between a writable root and a full disk. max_upload_gb: float = 8.0 # How many processes hold the peer connections. One means this process - # holds them itself, as it always did. See transport/webrtc/workers.py. - connection_processes: int = 1 + # holds them itself. See transport/webrtc/workers.py, and + # `default_connection_processes` for why it is not one by default. + connection_processes: int = field(default_factory=lambda: default_connection_processes()) # HEVC (and any future codec in media_probe.py's # BROWSER_INCOMPATIBLE_VIDEO_CODECS) has no decoder in most browsers, so # streaming it needs a real re-encode to H264 rather than the usual free @@ -305,6 +307,23 @@ def max_processes() -> int: return max(1, os.cpu_count() or 1) +# Above four, a single download no longer gains (docs/MESHBAY_DESIGN.md §5.5: +# 131 MB/s at four, 150 at eight with the receiving browser the limit), and each +# process is 85 MB held for the life of the node. +DEFAULT_CONNECTION_PROCESSES_MAX = 4 + + +def default_connection_processes() -> int: + """ + One per core, up to four. It was 1, which starts no process at all: the + several-cores path was measured, shipped and reached by no node whose + operator had not found the setting, and the lanes of a download all landed + on the node's one core — on Windows, where Python's UDP costs most, a + download over eight connections was as slow as over one. + """ + return min(DEFAULT_CONNECTION_PROCESSES_MAX, max_processes()) + + def _positive(value: object, default: int, name: str) -> int: """A count that must be at least one, or the default with a word about it. 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/sctp_fix.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/sctp_fix.py index 0d45a7d..a7e6f86 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/sctp_fix.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/sctp_fix.py @@ -1,12 +1,13 @@ -"""A fix to aiortc's SCTP sender, applied on import (aiortc 1.15). +"""Two fixes to aiortc's SCTP stack, applied on import (aiortc 1.15). -aiortc counts the bytes in flight and sends nothing new while that count is at -or above the congestion window. A chunk the peer acknowledged in a gap block is -marked `_acked` and taken out of the count. If the T3 timer then expires, every -chunk still queued is sent again, that one included, and goes back into the -count with `_acked` still set. When the cumulative ack finally covers it, the -SACK handler skips it because it reads as acknowledged already, and its bytes -stay counted for the life of the association. +**The sender leaked bytes in flight.** aiortc counts the bytes in flight and +sends nothing new while that count is at or above the congestion window. A chunk +the peer acknowledged in a gap block is marked `_acked` and taken out of the +count. If the T3 timer then expires, every chunk still queued is sent again, +that one included, and goes back into the count with `_acked` still set. When +the cumulative ack finally covers it, the SACK handler skips it because it reads +as acknowledged already, and its bytes stay counted for the life of the +association. Each such loss leaks a chunk. On a lossy path (a phone on Wi-Fi) the leak grows past the window, which T3 itself keeps small. From then on the node sends @@ -14,14 +15,49 @@ nothing on that connection: nothing is outstanding, so no timer runs to unstick it, and the association still reads as established. The peer keeps talking and the node keeps answering into a queue that never moves. Live symptom: the photo backup stalled after a few dozen photos, its slot granted and the grant never -delivered (docs/MESHBAY_DESIGN.md §9.12). +delivered (docs/MESHBAY_DESIGN.md §9.12). The fix is the invariant the code +assumed: a chunk being sent again is in flight, so it is not acknowledged. -The fix is the invariant the code assumed: a chunk being sent again is in -flight, so it is not acknowledged. `test_sctp_fix.py` replays the sequence -against aiortc itself. +**The receiver reassembled in quadratic time.** After every DATA chunk, +`InboundStream.pop_messages` walks the reassembly list from its start, through +every fragment of the message still being assembled. A message of n fragments +costs n²/2 steps, and a 1 MiB `file_chunk` is about 870 fragments: measured on a +receiving aiortc, 153 million steps for 400 MB, two thirds of its CPU, and the +whole of why it could not take more than 20 MB/s on loopback. No message can +come out of a reassembly holding no last fragment, so the walk is skipped until +one is there; when one is, the walk is aiortc's own, unchanged. + +The node receives uploads in 48 KB messages, where the walk is short; the +command-line client receives 1 MiB ones and carries the same fix +(`meshbay_cli/client/sctp_fix.py`). + +**The receiver acknowledged every packet.** RFC 4960 §6.2 asks for one SACK +every second packet, and Chromium does that; aiortc sent one at the end of +every packet of data. Each SACK costs its receiver a DTLS decryption and a parse +on the one core that also does everything else — on a node sending a download +to the command-line client, more time than sending the data. In an upload the +node is the receiver and pays the other half: the encryption and the send of a +SACK per packet. A SACK still goes at once when something is out of order or +duplicated (the loss signal the sender is waiting for) or the packet carried +anything other than data; a lone packet is acknowledged within `SACK_DELAY_S`. + +`test_sctp_fix.py` replays all three against aiortc itself. """ -from aiortc.rtcsctptransport import RTCSctpTransport +import asyncio +import contextlib + +from aiortc.rtcsctptransport import ( + SCTP_DATA_LAST_FRAG, + DataChunk, + InboundStream, + InitChunk, + RTCSctpTransport, + parse_packet, +) + +# RFC 4960 §6.2: at most 500 ms; 200 is what implementations use. +SACK_DELAY_S = 0.2 def _flight_size_increase(self, chunk) -> None: @@ -30,3 +66,82 @@ def _flight_size_increase(self, chunk) -> None: RTCSctpTransport._flight_size_increase = _flight_size_increase + + +_add_chunk = InboundStream.add_chunk +_pop_messages = InboundStream.pop_messages +_prune_chunks = InboundStream.prune_chunks + + +def add_chunk(self, chunk) -> None: + _add_chunk(self, chunk) + if chunk.flags & SCTP_DATA_LAST_FRAG: + self._last_frags = getattr(self, "_last_frags", 0) + 1 + + +def pop_messages(self): + # Only ever too high (a pruned last fragment), never too low: a message + # leaves the reassembly with exactly one last fragment, and only here. + if not getattr(self, "_last_frags", 0): + return + for message in _pop_messages(self): + self._last_frags -= 1 + yield message + + +def prune_chunks(self, tsn: int) -> int: + size = _prune_chunks(self, tsn) + self._last_frags = sum(1 for c in self.reassembly if c.flags & SCTP_DATA_LAST_FRAG) + return size + + +InboundStream.add_chunk = add_chunk +InboundStream.pop_messages = pop_messages +InboundStream.prune_chunks = prune_chunks + + +async def _handle_data(self, data: bytes) -> None: + """aiortc's own, down to the SACK at the end: that one may wait for a + second packet.""" + try: + _, _, verification_tag, chunks = parse_packet(data) + except ValueError: + return + if any(isinstance(x, InitChunk) for x in chunks): + assert len(chunks) == 1 + expected_tag = 0 + else: + expected_tag = self._local_verification_tag + if verification_tag != expected_tag: + return + for chunk in chunks: + await self._receive_chunk(chunk) + if not self._sack_needed: + return + self._sack_packets = getattr(self, "_sack_packets", 0) + 1 + if (self._sack_packets >= 2 or self._sack_misordered or self._sack_duplicates + or not all(isinstance(c, DataChunk) for c in chunks)): + await _sack_now(self) + elif getattr(self, "_sack_timer", None) is None: + self._sack_timer = self._loop.call_later(SACK_DELAY_S, _sack_later, self) + + +async def _sack_now(self) -> None: + timer = getattr(self, "_sack_timer", None) + if timer is not None: + timer.cancel() + self._sack_timer = None + self._sack_packets = 0 + if self._sack_needed and self._association_state != self.State.CLOSED: + # The DTLS transport may be gone under a timer that fired late. + with contextlib.suppress(ConnectionError): + await self._send_sack() + + +def _sack_later(self) -> None: + self._sack_timer = None + # Held: a task nobody references can be collected before it runs. + self._sack_task = asyncio.ensure_future(_sack_now(self)) + + +RTCSctpTransport._handle_data = _handle_data 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_connection_processes.py b/packages/meshbay-node/tests/test_connection_processes.py index ab9dfd3..778da5a 100644 --- a/packages/meshbay-node/tests/test_connection_processes.py +++ b/packages/meshbay-node/tests/test_connection_processes.py @@ -13,6 +13,7 @@ not carry. """ import asyncio +import re import struct from pathlib import Path @@ -231,11 +232,21 @@ async def test_one_process_starts_none(): # ── the setting ───────────────────────────────────────────────────────────── -def test_the_default_is_one_process(tmp_path): - assert node_settings_defaults()["connection_processes"] == 1 +@pytest.mark.parametrize("cores, expected", [(1, 1), (2, 2), (4, 4), (28, 4)]) +def test_the_default_is_one_per_core_up_to_four(tmp_path, monkeypatch, cores, expected): + """It was 1, which starts none: the path was shipped and reached by no node + whose operator had not found the setting.""" + monkeypatch.setattr("meshbay_node.config.max_processes", lambda: cores) + assert node_settings_defaults()["connection_processes"] == expected conf = tmp_path / "node.toml" conf.write_text('[hub]\nurl = "https://example.invalid"\nusername = "op"\n') - assert load_config(conf).node.connection_processes == 1 + assert load_config(conf).node.connection_processes == expected + + +def test_a_new_node_toml_leaves_it_to_the_default(tmp_path): + from meshbay_node.config import EXAMPLE_CONFIG + assert not re.search(r"^connection_processes\s*=", EXAMPLE_CONFIG, re.M), ( + "a value written by `init` is a value no later default reaches") def test_node_toml_cannot_ask_for_more_than_the_cores(tmp_path): @@ -296,4 +307,5 @@ async def test_a_number_the_machine_cannot_honour_is_refused(state, monkeypatch, with pytest.raises(ops.OpError): await ops.set_node_settings(state, {"connection_processes": value}) assert state["webrtc"].asked == [] - assert (await ops.get_node_settings(state))["connection_processes"] == 1 + assert ((await ops.get_node_settings(state))["connection_processes"] + == state["config"].node.connection_processes) diff --git a/packages/meshbay-node/tests/test_packaging_win.py b/packages/meshbay-node/tests/test_packaging_win.py index 6ee4891..0110b6c 100644 --- a/packages/meshbay-node/tests/test_packaging_win.py +++ b/packages/meshbay-node/tests/test_packaging_win.py @@ -1497,6 +1497,27 @@ def test_msix_gives_the_cli_an_execution_alias_and_the_windows_it_needs(): assert 'xmlns:uap5="http://schemas.microsoft.com/appx/manifest/uap/windows10/5"' in xml +def test_msix_gives_the_command_line_client_an_alias_of_its_own_application(): + """An Application may declare one windows.appExecutionAlias extension. + meshbay-node.exe has the app's; a second one beside it for meshbay-cli.exe + made makeappx refuse the whole package ("must only be declared once"), at + the very end of a Store build. The client is an Application of its own, + kept out of the Start menu, with its alias and its firewall rule.""" + xml = MSIX_MANIFEST_XML.read_text(encoding="utf-8") + ext = MSIX_EXTENSIONS_XML.read_text(encoding="utf-8") + assert ext.count('Category="windows.appExecutionAlias"') == 1 + apps = re.findall(r"<Application\b.*?</Application>", xml, re.S) + for app in apps: + assert app.count('Category="windows.appExecutionAlias"') <= 1, app[:80] + [cli] = [a for a in apps if 'Executable="app\\resources\\node-runtime\\meshbay-cli.exe"' + in a.split(">", 1)[0]] + assert 'AppListEntry="none"' in cli + assert 'Alias="meshbay-cli.exe"' in cli + package_level = xml.split("</Applications>", 1)[1] + rule = package_level.split('Executable="app\\resources\\node-runtime\\meshbay-cli.exe"', 1) + assert len(rule) == 2 and 'IPProtocol="UDP"' in rule[1].split("</desktop2:", 1)[0] + + def test_the_node_runtime_has_a_windowless_daemon_beside_the_cli(): spec = (WIN / "meshbay-node.spec").read_text(encoding="utf-8") windowless = spec.split('name="meshbay-nodew"', 1) 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"] diff --git a/packages/meshbay-node/tests/test_sctp_fix.py b/packages/meshbay-node/tests/test_sctp_fix.py index 0b2536a..c3fe622 100644 --- a/packages/meshbay-node/tests/test_sctp_fix.py +++ b/packages/meshbay-node/tests/test_sctp_fix.py @@ -1,4 +1,5 @@ -"""aiortc's SCTP sender must not leak bytes in flight (see sctp_fix.py). +"""aiortc's SCTP stack as sctp_fix.py leaves it: no bytes leaked in flight, +reassembly in linear time, one SACK every second packet. Replayed against aiortc's own RTCSctpTransport: three chunks sent, the first lost and the other two acknowledged in a gap block, T3 expires and they are @@ -8,10 +9,20 @@ enough of those stop the node from sending anything on that connection. """ import asyncio - -from aiortc.rtcsctptransport import RTCSctpTransport, SackChunk +import random import meshbay_node.transport.webrtc.core # noqa: F401 (applies the fix) +from aiortc.rtcsctptransport import ( + SCTP_DATA_FIRST_FRAG, + SCTP_DATA_LAST_FRAG, + SCTP_DATA_UNORDERED, + DataChunk, + InboundStream, + RTCSctpTransport, + SackChunk, + serialize_packet, +) +from meshbay_node.transport.webrtc import sctp_fix class _Dtls: @@ -57,3 +68,137 @@ def test_nothing_outstanding_means_nothing_in_flight(): s = asyncio.run(_replay()) assert not s._sent_queue assert s._flight_size == 0 + + +# ── Reassembly (see sctp_fix.py) ───────────────────────────────────────────── + + +def _fragments(messages: list[int], *, unordered: bool = False, tsn: int = 1000): + """DATA chunks for messages of the given fragment counts, in TSN order.""" + out = [] + for seq, n in enumerate(messages): + for i in range(n): + c = DataChunk() + c.flags = ((SCTP_DATA_FIRST_FRAG if i == 0 else 0) + | (SCTP_DATA_LAST_FRAG if i == n - 1 else 0) + | (SCTP_DATA_UNORDERED if unordered else 0)) + c.tsn, c.stream_id, c.stream_seq, c.protocol = tsn, 1, seq, 53 + c.user_data = f"{seq}.{i};".encode() + tsn += 1 + out.append(c) + return out + + +class _Original: + """aiortc's own reassembly, untouched, for the same arrivals.""" + + def __init__(self): + self.reassembly, self.sequence_number = [], 0 + + def add(self, chunk): + sctp_fix._add_chunk(self, chunk) + return list(sctp_fix._pop_messages(self)) + + +def test_reassembly_gives_what_aiortc_gives_for_any_arrival_order(): + rng = random.Random(7) + for trial in range(300): + messages = [rng.randint(1, 12) for _ in range(rng.randint(1, 8))] + chunks = _fragments(messages, unordered=trial % 3 == 0) + # Reordered within a short window, as a network reorders. + for i in range(len(chunks)): + j = min(len(chunks) - 1, i + rng.randint(0, 4)) + chunks[i], chunks[j] = chunks[j], chunks[i] + fixed, original = InboundStream(), _Original() + got, want = [], [] + for c in chunks: + fixed.add_chunk(c) + got.append(list(fixed.pop_messages())) + want.append(original.add(c)) + assert got == want, (trial, messages) + + +def test_a_large_message_is_walked_once_not_once_per_fragment(monkeypatch): + walks = [] + original = sctp_fix._pop_messages + + def counted(stream): + walks.append(len(stream.reassembly)) + return original(stream) + + monkeypatch.setattr(sctp_fix, "_pop_messages", counted) + s = InboundStream() + out = [] + for c in _fragments([1000, 1000]): + s.add_chunk(c) + out += list(s.pop_messages()) + assert len(out) == 2 + assert walks == [1000, 1000], "one walk per message, when its last fragment is in" + + +def test_pruning_a_last_fragment_does_not_stall_the_next_message(): + s = InboundStream() + first, second = _fragments([3]), _fragments([2], tsn=1003) + second[0].stream_seq = second[1].stream_seq = 1 + for c in first[1:]: # the first fragment never comes + s.add_chunk(c) + assert not list(s.pop_messages()) + s.prune_chunks(first[-1].tsn) # abandoned (FORWARD TSN) + s.sequence_number = 1 + for c in second: + s.add_chunk(c) + assert [m[2] for m in s.pop_messages()] == [b"0.0;0.1;"] + + +# ── Delayed acknowledgement (see sctp_fix.py) ──────────────────────────────── + +def _receiver(): + s = RTCSctpTransport(_Dtls()) + s._association_state = s.State.ESTABLISHED + s._last_received_tsn = 999 + sacks = [] + + async def send_sack(): + sacks.append(s._last_received_tsn) + s._sack_needed = False + s._sack_duplicates.clear() + + async def deliver(*_a): + pass + + s._send_sack = send_sack + s._receive = deliver + return s, sacks + + +def _packet(s, chunk) -> bytes: + return serialize_packet(5000, 5000, s._local_verification_tag, chunk) + + +def test_one_sack_every_second_packet_of_data(): + async def go(): + s, sacks = _receiver() + for c in _fragments([1] * 10): + await s._handle_data(_packet(s, c)) + return sacks + assert asyncio.run(go()) == [1001, 1003, 1005, 1007, 1009] + + +def test_a_gap_is_acknowledged_at_once(): + async def go(): + s, sacks = _receiver() + chunks = _fragments([1] * 4) + await s._handle_data(_packet(s, chunks[0])) + await s._handle_data(_packet(s, chunks[2])) # 1001 missing + return sacks + assert asyncio.run(go()) == [1000], "the gap is the loss signal the sender waits for" + + +def test_a_lone_packet_is_acknowledged_within_the_delay(): + async def go(): + s, sacks = _receiver() + await s._handle_data(_packet(s, _fragments([1])[0])) + assert sacks == [] + await asyncio.sleep(sctp_fix.SACK_DELAY_S + 0.1) + return sacks + assert asyncio.run(go()) == [1000] |