diff options
Diffstat (limited to 'packages/meshbay-node')
9 files changed, 237 insertions, 2 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transfers.py b/packages/meshbay-node/src/meshbay_node/transfers.py index 3345cb9..477ec92 100644 --- a/packages/meshbay-node/src/meshbay_node/transfers.py +++ b/packages/meshbay-node/src/meshbay_node/transfers.py @@ -102,11 +102,24 @@ class Lease: # later, for ever. Seen doing exactly that in a node's log, every 30 s, # minutes after the transfers involved had finished. missed_grants: int = 0 + # What moved under this lease and over how long, for the line the node logs + # when it ends. Counted here because nothing else spans a transfer: a + # download is a run of independent chunk requests. + moved: int = 0 + first_used: float | None = None + last_moved: float = 0.0 @property def member(self) -> tuple[str, str]: return (self.group_id, self.user_id) + @property + def elapsed(self) -> float: + """Seconds from the first request under this lease to the last byte.""" + if self.first_used is None: + return 0.0 + return max(0.0, self.last_moved - self.first_used) + @dataclass class TransferSlots: @@ -223,8 +236,18 @@ class TransferSlots: lease.used = True lease.missed_grants = 0 lease.last_seen = time.monotonic() if now is None else now + if lease.first_used is None: + lease.first_used = lease.last_seen return True + def account(self, tr: str, nbytes: int, now: float | None = None) -> None: + """`nbytes` were served or received under this lease.""" + lease = self.leases.get(tr) + if lease is None: + return + lease.moved += nbytes + lease.last_moved = time.monotonic() if now is None else now + def close(self, tr: str, reason: str = REASON_DONE, now: float | None = None) -> tuple[Lease | None, list[Lease]]: """Give a slot back. Returns (the closed lease, newly granted ones). 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 a44f097..c48e9ca 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py @@ -14,7 +14,10 @@ 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 sctp_fix # noqa: F401 (patches aiortc on import) +from meshbay_node.transport.webrtc import ( + sctp_fix, # noqa: F401 (patches aiortc on import) + sctp_stats, # noqa: F401 (patches aiortc on import) +) from meshbay_node.transport.webrtc.channel import ( _REPLY_TO, _DataChannelBuffer, diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py index 0434554..eafdaed 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py @@ -319,6 +319,8 @@ class FilesMixin: if self._channel is None or self._channel.readyState != "open": return self._send(chunk_data) + if leased: + self._account(tr, len(chunk_data.get("ct") or b"")) log.debug("dl: sent file=%s chunk=%s bytes=%s buffered=%s", file_id[:12], chunk_index, len(chunk_data.get("ct") or b""), getattr(self._channel, "bufferedAmount", "?")) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/sctp_stats.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/sctp_stats.py new file mode 100644 index 0000000..338ded9 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/sctp_stats.py @@ -0,0 +1,43 @@ +"""Counts what aiortc's SCTP sender does not, applied on import (aiortc 1.15). + +aiortc halves its congestion window on every loss and drops it to one packet +when the T3 timer expires, and keeps no record of either. Those two counts are +what say whether a slow transfer was held back by the path: a window that never +grows past a few hundred KB is losses, and the node could not tell. + +A loss is counted when the sender enters fast recovery, so a burst of packets +dropped together is one loss, which is also one halving of the window. +""" + +from aiortc.rtcsctptransport import RTCSctpTransport + +_receive_sack_chunk = RTCSctpTransport._receive_sack_chunk +_t3_expired = RTCSctpTransport._t3_expired + + +async def _counting_receive_sack_chunk(self, chunk) -> None: + recovering = self._fast_recovery_exit is not None + await _receive_sack_chunk(self, chunk) + if not recovering and self._fast_recovery_exit is not None: + self._mb_losses = getattr(self, "_mb_losses", 0) + 1 + + +def _counting_t3_expired(self) -> None: + self._mb_t3 = getattr(self, "_mb_t3", 0) + 1 + _t3_expired(self) + + +RTCSctpTransport._receive_sack_chunk = _counting_receive_sack_chunk +RTCSctpTransport._t3_expired = _counting_t3_expired + + +def describe(pc) -> str: + """The association of `pc` as a log suffix, or nothing when it has none.""" + sctp = getattr(pc, "sctp", None) + if sctp is None: + return "" + srtt = getattr(sctp, "_srtt", None) + rtt = f"{srtt * 1000:.0f} ms" if srtt else "?" + return (f" sctp cwnd={sctp._cwnd // 1024} KB rtt={rtt}" + f" loss={getattr(sctp, '_mb_losses', 0)}" + f" t3={getattr(sctp, '_mb_t3', 0)}") diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/transfer_handlers.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/transfer_handlers.py index 729c0a0..78c74d6 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/transfer_handlers.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/transfer_handlers.py @@ -11,6 +11,7 @@ from meshbay_common.protocol import MNP from meshbay_node import transfers as transfers_mod from meshbay_node.transfers import TransferSlots +from meshbay_node.transport.webrtc import sctp_stats from meshbay_node.transport.webrtc.limits import LEASE_GRANTED, LEASE_NONE, LEASE_QUEUED log = logging.getLogger("meshbay_node.transport.webrtc_server") @@ -25,6 +26,8 @@ if TYPE_CHECKING: # annotations only: the facade imports this module # noticing, so it trades latency for a timer that hardly ever runs. TRANSFER_SWEEP_SECS = 15 +MIB = 1024 * 1024 + class TransferMixin: def _slots(self) -> "TransferSlots": @@ -109,6 +112,29 @@ class TransferMixin: (self._user_id or "?")[:8], str(tr)[:8]) self._audit("transfer_unleased", str(tr)[:16]) + def _account(self, tr: str, nbytes: int) -> None: + slots = self._ctx.get("_transfer_slots") + if slots is not None: + slots.account(tr, nbytes) + + def _transfer_end_line(self, lease, reason: str) -> str: + """What a transfer moved, how fast, and what its association looked like. + + The node wrote nothing about throughput, so a slow download left no + trace in the journal and took a test bench to explain. The SCTP figures + are the association's totals since it opened, not this transfer's + share: a connection carries several transfers and they share one + window. + """ + line = (f"transfer: end {lease.kind} {lease.tr[:8]} {reason} " + f"{lease.moved / MIB:.1f} MB") + if lease.moved and lease.elapsed > 0: + line += (f" in {lease.elapsed:.1f} s " + f"({lease.moved / MIB / lease.elapsed:.2f} MB/s)") + owner = (self if lease.session_key == self._registry_key else + next(iter(self._sessions_everywhere(lease.session_key)), None)) + return line + sctp_stats.describe(getattr(owner, "_pc", None)) + def _transfer_state_msg(self, lease, state: str, reason: str = "") -> dict: slots = self._slots() out = { @@ -215,6 +241,7 @@ class TransferMixin: lease, granted = slots.close(tr, reason) if lease is not None: self._send(self._transfer_state_msg(lease, "closed", reason)) + log.info("%s", self._transfer_end_line(lease, reason)) self._announce(granted) def _release_transfers(self) -> None: @@ -223,6 +250,9 @@ class TransferMixin: if slots is None: return gone, granted = slots.release_session(self._registry_key) + for lease in gone: + log.info("%s", self._transfer_end_line( + lease, transfers_mod.REASON_SESSION_GONE)) if gone: log.info("transfer: session gone, released %d (%s)", len(gone), slots.summary()) @@ -251,6 +281,8 @@ class TransferMixin: ended, granted = slots.sweep() for lease, reason in ended: log.info("transfer: reclaimed %s (%s)", lease.tr[:8], reason) + if lease.state != "queued": + log.info("%s", self._transfer_end_line(lease, reason)) self._announce(granted, ended) log.debug("transfer: %s", slots.summary()) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py index b19846c..c052703 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py @@ -18,7 +18,7 @@ from meshbay_node.roots import ( shell_active, ) from meshbay_node.transport.webrtc.disk import _append_chunk, _free_bytes -from meshbay_node.transport.webrtc.limits import LEASE_NONE, LEASE_QUEUED +from meshbay_node.transport.webrtc.limits import LEASE_GRANTED, LEASE_NONE, LEASE_QUEUED log = logging.getLogger("meshbay_node.transport.webrtc_server") @@ -227,6 +227,8 @@ class UploadMixin: _refuse("Invalid chunk encoding", "bad_chunk_encoding") return chunk_bytes = bytes(data) + if tr and state == LEASE_GRANTED: + self._account(tr, len(chunk_bytes)) if not SAFE_UPLOAD_NAME.match(filename): _refuse("Invalid filename", "invalid_filename") diff --git a/packages/meshbay-node/tests/test_lease_enforcement.py b/packages/meshbay-node/tests/test_lease_enforcement.py index 2b1014b..7b636ac 100644 --- a/packages/meshbay-node/tests/test_lease_enforcement.py +++ b/packages/meshbay-node/tests/test_lease_enforcement.py @@ -273,3 +273,36 @@ async def test_the_sweeper_does_not_reclaim_a_transfer_that_is_moving(group): await _request(peer, _ids(group)[0], tr="t1") assert slots.leases["t1"].used is True await asyncio.sleep(0) + + +# ── the line the node logs when a transfer ends ───────────────────────────── + +async def test_a_served_chunk_is_counted_on_its_lease(group): + peer = _Session(group) + group["_peers"]["s1"] = peer + peer._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000}) + await _request(peer, _ids(group)[0], tr="t1") + await _request(peer, _ids(group)[0], tr="invented") + + lease = peer._slots().leases["t1"] + assert lease.moved == len(peer.chunks()[0]["ct"]) + line = peer._transfer_end_line(lease, "done") + assert line.startswith("transfer: end download t1 done 1.0 MB"), line + + +async def test_a_closed_transfer_is_logged_once(group): + peer = _Session(group) + group["_peers"]["s1"] = peer + lines: list[str] = [] + end_line = peer._transfer_end_line + peer._transfer_end_line = ( + lambda lease, reason: lines.append(end_line(lease, reason)) or lines[-1]) + + peer._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000}) + peer._do_transfer_open({"tr": "t2", "kind": DOWNLOAD, "bytes": 2_000_000}) + peer._do_transfer_close({"tr": "t1", "reason": "cancelled"}) + peer._release_transfers() + peer._release_transfers() + + assert [x.split()[3:5] for x in lines] == [ + ["t1", "cancelled"], ["t2", "session_gone"]], lines diff --git a/packages/meshbay-node/tests/test_sctp_stats.py b/packages/meshbay-node/tests/test_sctp_stats.py new file mode 100644 index 0000000..bf8970d --- /dev/null +++ b/packages/meshbay-node/tests/test_sctp_stats.py @@ -0,0 +1,74 @@ +"""Losses and T3 expirations are counted per association (see sctp_stats.py). + +Replayed against aiortc's own RTCSctpTransport: five chunks sent and the first +one lost, so three acknowledgements in a row report it missing and the sender +enters fast recovery, once. Then T3 expires, once. +""" + +import asyncio + +from aiortc.rtcsctptransport import RTCSctpTransport, SackChunk +from meshbay_node.transport.webrtc import sctp_stats + + +class _Dtls: + state = "connected" + + def _register_data_receiver(self, *a): + pass + + +class _Pc: + def __init__(self, sctp): + self.sctp = sctp + + +def _sack(cumulative_tsn: int, gaps: list) -> SackChunk: + sack = SackChunk() + sack.cumulative_tsn = cumulative_tsn % 2**32 + sack.gaps = gaps + sack.duplicates = [] + sack.advertised_rwnd = 1 << 20 + return sack + + +async def _replay() -> RTCSctpTransport: + s = RTCSctpTransport(_Dtls()) + s._association_state = s.State.ESTABLISHED + + async def sent(*_a, **_k): + pass + + s._send_chunk = sent + for name in ("_t3_start", "_t3_restart", "_t3_cancel"): + setattr(s, name, lambda: None) + s._cwnd = 100_000 + + for _ in range(5): + await s._send(1, 51, b"x" * 1000) + first = s._sent_queue[0].tsn + for last_seen in (2, 3, 4): + await s._receive_sack_chunk(_sack(first - 1, [(2, last_seen + 1)])) + # Still recovering from the same loss: one episode, not a second. + await s._receive_sack_chunk(_sack(first - 1, [(2, 5)])) + s._t3_expired() + return s + + +def test_a_loss_and_a_t3_expiry_are_each_counted_once(): + s = asyncio.run(_replay()) + assert s._mb_losses == 1 + assert s._mb_t3 == 1 + assert sctp_stats.describe(_Pc(s)).endswith("loss=1 t3=1") + + +def test_an_association_that_lost_nothing_says_so(): + async def fresh() -> RTCSctpTransport: + return RTCSctpTransport(_Dtls()) + + s = asyncio.run(fresh()) + assert sctp_stats.describe(_Pc(s)).endswith("rtt=? loss=0 t3=0") + + +def test_a_session_without_an_association_adds_nothing(): + assert sctp_stats.describe(None) == "" diff --git a/packages/meshbay-node/tests/test_transfer_slots.py b/packages/meshbay-node/tests/test_transfer_slots.py index 08e7e70..68d8858 100644 --- a/packages/meshbay-node/tests/test_transfer_slots.py +++ b/packages/meshbay-node/tests/test_transfer_slots.py @@ -369,3 +369,26 @@ def test_using_a_lease_forgives_its_earlier_misses(_=None): assert s.leases["slow"].missed_grants == 2 s.touch("slow", now=2 * GRANT_DEADLINE_SECS + 3) assert s.leases["slow"].missed_grants == 0 + + +# ── what the end-of-transfer line is built from ───────────────────────────── + +def test_a_lease_counts_what_moved_and_for_how_long(): + slots = TransferSlots() + slots.open(tr="t1", kind=DOWNLOAD, session_key="s1", user_id="alice", + group_id="g1", now=0.0) + slots.touch("t1", now=10.0) + slots.account("t1", 1_000_000, now=10.5) + slots.touch("t1", now=11.0) + slots.account("t1", 500_000, now=12.0) + lease = slots.leases["t1"] + assert lease.moved == 1_500_000 + assert lease.elapsed == 2.0 + + +def test_a_lease_nothing_moved_under_has_no_duration(): + slots = TransferSlots() + lease, _ = slots.open(tr="t1", kind=DOWNLOAD, session_key="s1", + user_id="alice", group_id="g1", now=0.0) + assert (lease.moved, lease.elapsed) == (0, 0.0) + slots.account("unknown", 10) |