From d74d371bd7c0ed25f60754e475a78975bf2dc4ee Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Fri, 16 Oct 2026 10:41:17 +0200 Subject: perf: linear SCTP reassembly, one SACK every second packet, 4 MB buffer in the CLI aiortc walked a message's fragments after each one arrived (n^2/2 steps for a 1 MiB chunk) and acknowledged every packet. Measured on loopback, 400 MB to the CLI: 19 MB/s before, 39 after (49 by the node's count); uploads from the CLI 25 -> 29 MB/s. Co-Authored-By: Claude Opus 5.5 --- .../meshbay-cli/src/meshbay_cli/client/sctp_fix.py | 126 +++++++++++++++++++++ .../meshbay-cli/src/meshbay_cli/client/session.py | 8 ++ .../src/meshbay_cli/client/udp_buffer.py | 18 +++ 3 files changed, 152 insertions(+) create mode 100644 packages/meshbay-cli/src/meshbay_cli/client/sctp_fix.py create mode 100644 packages/meshbay-cli/src/meshbay_cli/client/udp_buffer.py (limited to 'packages/meshbay-cli/src/meshbay_cli') 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 -- cgit v1.2.3 From 4802be13580de1529b7824a2337d132d64849801 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Fri, 16 Oct 2026 10:41:17 +0200 Subject: fix(cli): apply index updates in place instead of fetching the whole index The interactive mode fetched and re-parsed the full index on every index_delta, on the loop receiving the download: 67 times in 80 s while a node enriched a library. Co-Authored-By: Claude Opus 5.5 --- .../meshbay-cli/src/meshbay_cli/client/tree.py | 33 +++++++++++++++++ .../meshbay-cli/src/meshbay_cli/tui/screens.py | 30 ++++++++++++++- packages/meshbay-cli/tests/fake_node.py | 22 ++++++++++- packages/meshbay-cli/tests/test_interactive.py | 43 ++++++++++++++++++++++ 4 files changed, 126 insertions(+), 2 deletions(-) (limited to 'packages/meshbay-cli/src/meshbay_cli') diff --git a/packages/meshbay-cli/src/meshbay_cli/client/tree.py b/packages/meshbay-cli/src/meshbay_cli/client/tree.py index 6e537ef..dd3acc8 100644 --- a/packages/meshbay-cli/src/meshbay_cli/client/tree.py +++ b/packages/meshbay-cli/src/meshbay_cli/client/tree.py @@ -57,6 +57,39 @@ class Tree: dirs |= set(roots) return cls(roots, dirs, files) + def apply_delta(self, delta: dict) -> bool: + """ + Bring the tree up to an `index_delta`, in place. True when the delta + changed what the tree holds — a file added or removed, a root's state — + rather than only the fields of files already there. + + A delta carries no `dirs`, so an empty folder made elsewhere is not in + it: a structural change is where the caller asks for the whole index + again, once. The common delta is enrichment filling in a file's fields, + about one a second while a library is being enriched, and asking for + the whole index on each of those took the download's own event loop: + 67 full indexes in the 80 s of one download. + """ + updates = {e["id"]: e for e in delta.get("updates", [])} + gone = set(delta.get("deletions", [])) + for parent, entries in self.files.items(): + if gone or any(e["id"] in updates for e in entries): + self.files[parent] = [updates.get(e["id"], e) for e in entries + if e["id"] not in gone] + known = {e["id"] for entries in self.files.values() for e in entries} + added = [e for e in delta.get("additions", []) if e["id"] not in known] + for e in added: + self.files.setdefault(e["path"], []).append(e) + parts = e["path"].split("/") + for i in range(1, len(parts) + 1): + self.dirs.add("/".join(parts[:i])) + roots = {r["name"]: r for r in delta.get("roots", [])} + moved_roots = bool(roots) and roots != self.roots + if moved_roots: + self.roots = roots + self.dirs |= set(roots) + return bool(gone or added or moved_roots) + def root_available(self, path: str) -> bool: root = self.roots.get(path.split("/", 1)[0]) return bool(root.get("available", True)) if root else True diff --git a/packages/meshbay-cli/src/meshbay_cli/tui/screens.py b/packages/meshbay-cli/src/meshbay_cli/tui/screens.py index 39b5df3..b6fd347 100644 --- a/packages/meshbay-cli/src/meshbay_cli/tui/screens.py +++ b/packages/meshbay-cli/src/meshbay_cli/tui/screens.py @@ -41,6 +41,9 @@ from .panes import LocalPane, RemotePane from .widgets import Choice, TransferStrip PER_GROUP = 2 +# A burst of index updates that added or removed files is answered by one full +# index, this long after the last of them. +REFRESH_AFTER_DELTA_S = 2.0 class SignInScreen(Screen): @@ -118,6 +121,7 @@ class MainScreen(Screen): self._trees: dict[str, Tree] = {} self._gates: dict[str, asyncio.Semaphore] = {} self._watching: set[str] = set() + self._refreshes: dict[str, asyncio.TimerHandle] = {} def compose(self) -> ComposeResult: with Horizontal(): @@ -150,12 +154,36 @@ class MainScreen(Screen): self._trees[group["id"]] = Tree.from_index(await session.index()) if group["id"] not in self._watching: self._watching.add(group["id"]) - session.ch.on_push("index_delta", lambda _msg, g=group: self._changed(g)) + session.ch.on_push( + "index_delta", lambda msg, g=group, s=session: self._delta(g, s, msg)) return self._trees[group["id"]] def forget_tree(self, group: dict) -> None: self._trees.pop(group["id"], None) + def _delta(self, group: dict, session, msg: dict) -> None: + """A pushed index update, applied to the tree in place. Only one that + adds or removes something is drawn, and asks for the whole index — once + for a burst, not once per message.""" + tree = self._trees.get(group["id"]) + if tree is None: + return # the next look asks for the whole index anyway + try: + structural = tree.apply_delta(session.open_delta(msg)) + except CliError: + structural = True + if not structural: + return + pending = self._refreshes.pop(group["id"], None) + if pending is not None: + pending.cancel() + self._refreshes[group["id"]] = asyncio.get_running_loop().call_later( + REFRESH_AFTER_DELTA_S, self._refreshed, group) + + def _refreshed(self, group: dict) -> None: + self._refreshes.pop(group["id"], None) + self._changed(group) + def _changed(self, group: dict) -> None: """Something changed in the group: the index is asked for again, and the pane redrawn if it is showing that group.""" diff --git a/packages/meshbay-cli/tests/fake_node.py b/packages/meshbay-cli/tests/fake_node.py index c618ee4..47f82b3 100644 --- a/packages/meshbay-cli/tests/fake_node.py +++ b/packages/meshbay-cli/tests/fake_node.py @@ -79,6 +79,8 @@ class FakeNode: self.members: dict[str, Identity] = {} # sender → the device it signs with self.unresponsive = False # pings go unanswered, as a dead link's self.admin_ops: dict[str, dict] = {} + self.channels: list[FakeChannel] = [] + self.index_syncs = 0 # full indexes asked for def add(self, path: str, name: str, data: bytes, uploader_id: str = "") -> dict: fid = hashlib.blake2b(data + path.encode() + name.encode(), digest_size=32).hexdigest() @@ -92,9 +94,26 @@ class FakeNode: return entry def session(self) -> GroupSession: - return GroupSession(None, FakeChannel(self), GROUP_ID, "node-1", NODE_PK, self.gek, + ch = FakeChannel(self) + self.channels.append(ch) + return GroupSession(None, ch, GROUP_ID, "node-1", NODE_PK, self.gek, dict(self.ack), self.identity, self.nonce, self.role) + def push_delta(self, *, additions=(), updates=(), deletions=()) -> None: + """An `index_delta` to every session, sealed as the node seals one. The + entries are applied here first, so a later `index_sync` agrees.""" + for e in (*additions, *updates): + self.files[e["id"]] = e + for fid in deletions: + self.files.pop(fid, None) + wire = lambda es: [{k: v for k, v in e.items() if k != "data"} for e in es] # noqa: E731 + payload = {"base_version": 1, "version": 1, "additions": wire(additions), + "deletions": list(deletions), "updates": wire(updates)} + msg = {"type": "index_delta", "group_id": GROUP_ID, + **seal(self.gek, PURPOSE_INDEX, "index_delta", GROUP_ID, payload)} + for ch in self.channels: + ch._emit(msg) + def list_device(self, sender: str, device: Identity, *, countersigned_by: Identity | None = None, proof: bool = True) -> None: """A roster row for `sender`'s device: a root, or countersigned by another @@ -189,6 +208,7 @@ class FakeChannel: def _index_sync(self, msg): n = self.node + n.index_syncs += 1 payload = {"version": 1, "roots": n.roots, "dirs": sorted(n.dirs), "entries": [{k: v for k, v in e.items() if k != "data"} for e in n.files.values()]} diff --git a/packages/meshbay-cli/tests/test_interactive.py b/packages/meshbay-cli/tests/test_interactive.py index 36b7f47..9960533 100644 --- a/packages/meshbay-cli/tests/test_interactive.py +++ b/packages/meshbay-cli/tests/test_interactive.py @@ -444,3 +444,46 @@ async def test_put_sends_a_file_into_the_chat_folder_and_get_saves_one_here( await say("/get ../share/x") await _until(pilot, lambda: "name a file of the chat's folder" in _said(room)) await pilot.press("escape") + + +async def test_index_updates_are_applied_not_asked_for_again(world, monkeypatch): + """ + A library being enriched pushes an `index_delta` about once a second, each + one filling in a file's fields. The panes used to answer every one with the + whole index — 67 full indexes in the 80 s of one download on a real node, + on the event loop the download was being received on. Updates are applied + where they land; only a delta that adds or removes a file asks again, once + for a burst. + """ + from meshbay_cli.tui import screens + monkeypatch.setattr(screens, "REFRESH_AFTER_DELTA_S", 0.2) + hub, node, work = world + film = node.add("share", "film.mkv", b"F" * 5000) + assert await asyncio.to_thread(cli.main, ["--hub", "https://hub.test", "-u", USERNAME, + "--password", PASSPHRASE, "login"]) == 0 + app = _app() + async with app.run_test(size=(140, 40)) as pilot: + await _until(pilot, lambda: isinstance(app.screen, MainScreen)) + remote = app.screen.query_one(RemotePane) + await _open(pilot, remote, f"Family@{USERNAME}/") + await _open(pilot, remote, "share/") + await _until(pilot, lambda: "film.mkv" in _names(remote)) + asked = node.index_syncs + + for i in range(20): + node.push_delta(updates=[{**film, "duration": 100 + i}]) + await pilot.pause(0.6) + assert node.index_syncs == asked, "an update must not fetch the whole index" + tree = app.screen._trees["g1"] + assert tree.file("share/film.mkv")["duration"] == 119 + + extra = [node.add("share", f"new{i}.txt", b"n%d" % i) for i in range(3)] + node.files = {k: v for k, v in node.files.items() if v not in extra} + for e in extra: + node.push_delta(additions=[e]) + await _until(pilot, lambda: "new2.txt" in _names(remote)) + await pilot.pause(0.6) + assert node.index_syncs == asked + 1, "a burst of additions asks again once" + + node.push_delta(deletions=[film["id"]]) + await _until(pilot, lambda: "film.mkv" not in _names(remote)) -- cgit v1.2.3