UplinkGauge

Congestion verdicts for one client's bulk transfer, timed end to end over that client's own session websocket(s).

Nothing on this side can see where a transfer's queue stands: an upload's is in the client's kernel and first hop, a download's in whatever buffer sits ahead of the client's downlink — a reverse proxy, a tunnel, a bloated modem — and this host's own send queue reads empty while any of those absorbs its writes. But every session connection from that client (the data WebSocket, or the WebRTC signaling socket) crosses the same bottleneck as the transfer, in both directions. So the gauge sends a protocol ping every PING_INTERVAL on each session socket and times the pong: the ping queues behind a download on the way out and the pong behind an upload on the way back, and the round trip inflating past a per-session floor by more than INFLATION_US is the verdict a delay-based scavenger backs off on (LEDBAT's discipline: take what the link has spare, yield the moment anything else waits). Browsers answer pings in the network process, so no page or client-code cooperation is needed. The rejected alternatives: reading flat out to measure the client's rate, then settling on a fixed share of it, filled the very queue it existed to prevent, latched a one-shot estimate, and held idle-screen uploads to half the link; kernel srtt (TCP_INFO) and the unsent-queue ioctl on the transfer's own socket time or count only the first hop, so any relaying middlebox blinds them — and a fixed byte threshold on that queue reads every chunk written faster than the hop drains as congestion, which pinned downloads near the rate floor — where the pong crosses end to end.

The end-to-end trip also spans both event loops, so a server too loaded to serve the stream reads as congestion and sheds the transfer first — load protection for free. INFLATION_US sits above scheduler jitter and below what a session feels: the queue the sawtooth sustains stays imperceptible.

Ping payloads are per-session counters; the pong echoes its ping's payload (RFC 6455), so note_pong matches replies to send times through _UPLINK_SESSIONS and stray pongs — aiohttp's own heartbeat's among them — fall through unmatched. A ping failure drops that socket; a gauge with no sockets left reports not alive and the caller stops pacing — a vanished session leaves nothing to protect.

Attributes

Functions

def __init__(self, conns: List[List[Any]]) -> None:
    """``conns``: ``[websocket, session state, last-consumed seq]`` per
    session connection sharing the uploader's uplink."""
    self._conns = conns
    self._last_ping = 0.0
    self.inflation_us = 0

async def sample(self) -> Optional[bool]:
    """Ping on cadence and return one verdict across the gauged sockets.

    Returns:
        True when any session's round trip stands above its floor by more
        than `INFLATION_US`, False when at least one session answered
        fresh and none congested, None when no fresh pong arrived since
        the last call (the caller holds its rate). `inflation_us` keeps
        the largest fresh excess over the floor, the queue's depth in
        time, for the caller to size its step by.
    """
    now = time.monotonic()
    if now - self._last_ping >= self.PING_INTERVAL:
        self._last_ping = now
        for conn in list(self._conns):
            ws, state, _last = conn
            payload = state["next_ping"].to_bytes(8, "big")
            state["next_ping"] += 1
            pending = state["pending"]
            pending[payload] = now
            while len(pending) > self._PENDING_MAX:
                pending.pop(next(iter(pending)))
            try:
                await ws.ping(payload)
            except Exception:
                self._conns.remove(conn)
    verdict: Optional[bool] = None
    inflation = 0
    for conn in self._conns:
        _ws, state, last = conn
        if state["seq"] == last or state["rtt_us"] is None:
            continue
        conn[2] = state["seq"]
        rtt = state["rtt_us"]
        floor = _observe_rtt_floor(state, rtt, now)
        congested = rtt > floor + self.INFLATION_US
        inflation = max(inflation, rtt - floor)
        verdict = congested if verdict is None else (verdict or congested)
    if verdict is not None:
        self.inflation_us = inflation
    return verdict