diff options
6 files changed, 517 insertions, 16 deletions
diff --git a/packages/meshbay-cli/src/meshbay_cli/client/sctp_fix.py b/packages/meshbay-cli/src/meshbay_cli/client/sctp_fix.py new file mode 100644 index 0000000..62835d3 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/client/sctp_fix.py @@ -0,0 +1,126 @@ +"""Fixes to aiortc's SCTP stack, applied on import (aiortc 1.15). + +The first two are the node's (`meshbay_node/transport/webrtc/sctp_fix.py`, where +why each is needed is written down); this is a copy because the client does not +depend on the node, and `meshbay-common` does not depend on aiortc (the hub uses +it without). `test_sctp_fix.py` in each package holds its copy to aiortc itself. + +In a download this side is the receiver of the node's 1 MiB `file_chunk` +messages, about 870 SCTP packets each, and two receiver costs set the ceiling. +Measured on loopback, 400 MB, one connection: + +- aiortc walked every fragment of the message in progress after each one + arrived, 153 million steps: two thirds of the client's CPU. Fixed by the + reassembly fix below — 20 MB/s became 28; +- it acknowledged every packet. RFC 4960 §6.2 asks for one SACK every second + packet (Chromium does that), and each SACK costs the node a DTLS decryption + and a parse on the one core that also sends: more of its time than sending + the data. One every second packet — 28 MB/s became 39, the node's own figure + for the transfer 49. + +A SACK still goes at once when something is out of order or duplicated (that is +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`. +""" + +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: + chunk._acked = False + self._flight_size += chunk._book_size + + +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-cli/src/meshbay_cli/client/session.py b/packages/meshbay-cli/src/meshbay_cli/client/session.py index d10acc8..b6fc9e2 100644 --- a/packages/meshbay-cli/src/meshbay_cli/client/session.py +++ b/packages/meshbay-cli/src/meshbay_cli/client/session.py @@ -63,6 +63,7 @@ from meshbay_common.handshake import ( from meshbay_common.join import join_transcript from ..errors import NODE, NODE_KEY, VERSION, CliError +from . import sctp_fix, udp_buffer # noqa: F401 (patch aiortc and aioice on import) from .account import Account from .keys import ( LEGACY_MAGIC, @@ -250,6 +251,13 @@ class GroupSession: raise CliError(NODE, "The group's index does not open under the group key; " "the session is ended.") from e + def open_delta(self, msg: dict) -> dict: + """A pushed `index_delta`, opened under the group key.""" + try: + return unseal(self.gek, PURPOSE_INDEX, "index_delta", self.group_id, msg) + except Exception as e: + raise CliError(NODE, "An index update does not open under the group key.") from e + async def close(self) -> None: with contextlib.suppress(Exception): await self.pc.close() diff --git a/packages/meshbay-cli/src/meshbay_cli/client/udp_buffer.py b/packages/meshbay-cli/src/meshbay_cli/client/udp_buffer.py new file mode 100644 index 0000000..c0045d8 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/client/udp_buffer.py @@ -0,0 +1,18 @@ +"""A larger receive buffer on the client's ICE sockets, applied on import. + +In a download the client is the receiver, and it reads its socket from Python +— on the same event loop as the terminal interface — more slowly than a node +sends in bursts. aioice asks the kernel for 256 KB per socket; what does not fit +is dropped by this machine's kernel (`UdpRcvbufErrors`), the node's SCTP sender +takes each drop for congestion and halves its window. Measured on loopback, 400 +MB: 3245 drops and 19 MB/s with 256 KB. The node does the same for the same +reason in the other direction (`meshbay_node/transport/webrtc/udp_buffer.py`). + +Linux grants no more than `net.core.rmem_max`; Windows grants what is asked. +""" + +import aioice.turn + +WANTED = 4 * 1024 * 1024 + +aioice.turn.UDP_SOCKET_BUFFER_SIZE = WANTED diff --git a/packages/meshbay-cli/tests/test_sctp_fix.py b/packages/meshbay-cli/tests/test_sctp_fix.py new file mode 100644 index 0000000..04638db --- /dev/null +++ b/packages/meshbay-cli/tests/test_sctp_fix.py @@ -0,0 +1,89 @@ +"""The client's copy of the node's aiortc fixes (client/sctp_fix.py). + +The copy exists because neither package may depend on the other; it is held to +the node's by comparing the code itself, docstrings aside, and the two effects +that set a download's speed here are replayed against aiortc. +""" + +import ast +import asyncio +from pathlib import Path + +import pytest +from aiortc.rtcsctptransport import ( + SCTP_DATA_FIRST_FRAG, + SCTP_DATA_LAST_FRAG, + DataChunk, + InboundStream, + RTCSctpTransport, + serialize_packet, +) +from meshbay_cli.client import sctp_fix + +NODE_COPY = (Path(__file__).resolve().parents[2] / "meshbay-node" / "src" / "meshbay_node" + / "transport" / "webrtc" / "sctp_fix.py") + + +def _code(path: Path) -> str: + tree = ast.parse(path.read_text(encoding="utf-8")) + tree.body = tree.body[1:] # the module docstring + return ast.dump(tree) + + +@pytest.mark.skipif(not NODE_COPY.exists(), reason="not in the source tree") +def test_the_code_is_the_nodes(): + assert _code(Path(sctp_fix.__file__)) == _code(NODE_COPY) + + +def _chunks(n_fragments: int, count: int, tsn: int = 1000) -> list[DataChunk]: + out = [] + for seq in range(count): + for i in range(n_fragments): + c = DataChunk() + c.flags = ((SCTP_DATA_FIRST_FRAG if i == 0 else 0) + | (SCTP_DATA_LAST_FRAG if i == n_fragments - 1 else 0)) + c.tsn, c.stream_id, c.stream_seq, c.protocol = tsn, 1, seq, 53 + c.user_data = b"x" + tsn += 1 + out.append(c) + return out + + +def test_a_megabyte_is_walked_once(monkeypatch): + walks = [] + original = sctp_fix._pop_messages + monkeypatch.setattr(sctp_fix, "_pop_messages", + lambda s: (walks.append(1), original(s))[1]) + s = InboundStream() + got = 0 + for c in _chunks(870, 3): + s.add_chunk(c) + got += len(list(s.pop_messages())) + assert got == 3 and len(walks) == 3 + + +def test_every_second_packet_is_acknowledged(): + class _Dtls: + state = "connected" + + def _register_data_receiver(self, *a): + pass + + async def go(): + 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 + + async def deliver(*_a): + pass + + s._send_sack, s._receive = send_sack, deliver + for c in _chunks(1, 6): + await s._handle_data(serialize_packet(1, 1, s._local_verification_tag, c)) + return sacks + assert asyncio.run(go()) == [1001, 1003, 1005] 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/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] |