CongestionSteer
The steer of one display's CBR target over congestion ticks, for both
transports. A tick carries two kinds of evidence: the queue standing on the
path (the WebSockets backpressure loop measures it from the display's frame
round trip, DataStreamingServer._steer_bitrate_to_link, and WebRTC's loop
from one-way delay, RTCDtlsTransport.take_twcc_window) and the loss the
receiver reported (WebRTC's loop, from the same transport-wide-cc feedback).
A standing queue backs the target off on the first tick that shows it: a delay verdict fires at a queue that is still small, where waiting a second tick lets it grow by the whole overshoot. The rate the path delivered while the queue stood is its capacity, so the target drops to HEADROOM of that rather than by a blind fraction, and lower still while the queue already written ahead drains: at the depth that empties it in DRAIN_S, never under DRAIN_FLOOR of the capacity, for as long as that takes at that rate. No cut follows until the drain and SETTLE_S more have passed, because the queue the cut is emptying reads as a queue meanwhile; cutting again on it would take the target to the floor while the path only drains. The target returns to HEADROOM of the capacity on the first tick past the drain that reads no queue and no loss past LOSS, and that return clears a loss strike: a queue the drain left keeps the drain's target through the hold, and the tick after the hold backs off again. Without a measured delivery the backoff is BACKOFF of the target, held for HOLD_S. A tick whose stream offered the path less than APP_LIMITED of the target, and whose path delivered at least APP_LIMITED_DELIVERED of it, a still screen's trickle, cuts nothing on a queue: what the path delivered is then what the stream sent rather than what the path carries, and a cut to it would put the next motion at the floor while draining nothing the stream queued. A stream the path delivers less of than it offers is filling the queue, whatever its target, and backs off. Where delivery is measured by acked frames, the offer compared is what the stream sent of those same frames over the span it sent them in: the tick's offer counts the tail of a burst still unacked, so a window switch over a trickle, whose queue drains as soon as the burst has crossed, read as a stream filling the queue.
Loss backs the target off only on the second lossy tick in a row: one tick of a thin stream is too few packets for its loss fraction to mean anything. That backoff is BACKOFF of the target, held for HOLD_S, so the recovery does not climb straight back onto the loss that caused it.
A clean tick outside a hold raises the target, by STEP a second while what
the path delivers is far from the last capacity it showed and by STEP_NEAR
once it is within NEAR of it, where the next queue waits: probing there
slowly keeps the overshoot that finds it small. A still screen the
transport padded up to the target (PathProbe, probed) has nothing to
show but the path, and climbs PROBE_STEP a second until the first queue
ends the probe: a cut desktop is back at its rate in seconds, at the cost of
the last tick's overshoot. Delivered rates rather than
targets are compared, since an encoder need not emit what it is asked for
(a CBR encoder with a small buffer runs under its target, and content it
cannot compress that far runs over). A path that delivers clear past the
old capacity with no queue forgets it, since it now carries more. A tick
whose stream offered under APP_LIMITED of the target tests nothing of the
path, so its step stops at the capacity the path last showed, or where the
target is when the steer knows none: stepping on a still screen's trickle
would climb to the ceiling and put the next motion there, past a slow
path's capacity, for a queue and a backoff every time the screen moves. A
tick that measured no offer is taken the same way, as a queue under one
backs off: over WebSockets the backpressure gate held it, which says the
page fell behind rather than that the target fits. A raise also follows
HEADROOM of the measured goodput where that is higher, never past the
ceiling.
A stall is not a slower path, and interrupts a rate the path was carrying, one in force through CARRIED_S of ticks that read no queue: before any there is none to restore (a session started over a path slower than its rate). A hold of seconds acks a window's frames late and together, a queue of STALL_QUEUE_S or more over a delivery under STALL_DELIVERY of the target, or one that grew by STALL_QUEUE_S since the last tick that read the delay (a queue a slower path builds is read while still small), and a cut on it keeps the target under what the stall let through, the capacity it showed, for as long as the screen stays still. A backoff on such a tick keeps the target the episode began at, before any backoff in the STALL_EPISODE_S ahead of it, as far as that carried rate, and the second tick in a row after it that reads no queue and no loss past LOSS restores that target and forgets the capacity (the first can come while the frames the stall held are still on their way, which the next one shows as the stall's queue); a backoff with no delivery measured keeps its target too, restored the same way past its hold. A queue the path delivered that target through, a window switch's burst on a fast path, keeps it, and so does a tick whose offer went unmeasured (the gate held it: what it delivered is the hold's release, not the path); one under it is a path that did slow and drops it. A path that did slow also shows a queue at the restored target: a backoff within STALL_EPISODE_S of a restore keeps nothing, so it backs off on the path's real delivery as any queue does, rather than restoring again each time the queue drains.
Attributes
attributeLOSS= 0.1attributeBACKOFF= 0.7attributeHOLD_S= 2.0attributeSTEP= 1.15attributeSTEP_NEAR= 1.05attributePROBE_STEP= 1.3attributeNEAR= 0.9attributeHEADROOM= 0.85attributeDRAIN_S= 1.0attributeDRAIN_FLOOR= 0.5attributeSETTLE_S= 0.5attributeAPP_LIMITED= 0.5attributeAPP_LIMITED_DELIVERED= 0.8attributeSTALL_QUEUE_S= 1.0attributeSTALL_DELIVERY= 0.25attributeSTALL_EPISODE_S= 5.0attributeCARRIED_S= 2.0attributestrikes= 0attributehold_until= 0.0attributecapacity_kbpsOptional[float]= Noneattributerestored= FalseFunctions
constructor__init__() -> NoneSource Code
def __init__(self) -> None:
self.strikes = 0
self.hold_until = 0.0
self.capacity_kbps: Optional[float] = None
self._cruise_kbps: Optional[float] = None
self._drain_until = 0.0
self._tick_at: Optional[float] = None
self._restore_kbps: Optional[float] = None
self._backoff_from: Optional[Tuple[float, float]] = None
self._restore_after = 0.0
self._restored_at = float('-inf')
self._last_queue_s: Optional[float] = None
# The rate in force on the last tick that read no queue and no loss through CARRIED_S of such
# ticks (`_clean_since` the first of them): what a stall interrupts.
self._clean_kbps: Optional[float] = None
self._clean_since: Optional[float] = None
# Whether the last tick restored the target a stall cut (for the transports' log).
self.restored = FalseReturns
Nonefunctarget(current, ceiling, floor, goodput_bps, loss, now, queue_s=None, offered_bps=None, acked_offered_bps=None, capacity_bps=None, probed=False) -> floatThe next target in kbps for one tick.
Source Code
def target(self, current: float, ceiling: float, floor: float,
goodput_bps: float, loss: float, now: float,
queue_s: Optional[float] = None,
offered_bps: Optional[float] = None,
acked_offered_bps: Optional[float] = None,
capacity_bps: Optional[float] = None, probed: bool = False) -> float:
"""The next target in kbps for one tick.
Args:
current: The target in force, in kbps.
ceiling: The configured target, which the steer never exceeds.
floor: The lowest target allowed.
goodput_bps: What the path delivered over the tick, or 0 when unmeasured.
loss: The loss fraction the receiver reported over the tick.
now: Monotonic time of the tick.
queue_s: The queue standing on the path through the tick, in seconds:
0 for a clean tick, None where the transport measures no delay.
offered_bps: What the stream sent over the tick, or None when
unmeasured.
acked_offered_bps: What the stream sent of the frames goodput_bps
counts, over the span it sent them in, where the transport
measures delivery by acked frames: the queue's app-limited
test takes it over offered_bps, since the tick's offer counts
a burst's tail still unacked, which the delivery does not.
capacity_bps: What the path carried while the queue built, where
the transport measured it apart from goodput_bps (a burst over
a trickle, whose window's rate the trickle dilutes); the
backoff takes the higher.
probed: Whether the transport padded the tick's stream up to the
target (`PathProbe`): a clean tick it filled to half the target
or more raises the target PROBE_STEP a second.
"""
dt = 1.0 if self._tick_at is None else min(max(now - self._tick_at, 0.0), 2.0)
self._tick_at = now
self.restored = False
last_queue_s = self._last_queue_s
if queue_s is not None:
self._last_queue_s = queue_s
clean_kbps = self._clean_kbps
if queue_s == 0 and loss <= self.LOSS:
if self._clean_since is None:
self._clean_since = now
if now - self._clean_since >= self.CARRIED_S:
self._clean_kbps = current
elif queue_s is not None or loss > self.LOSS:
self._clean_since = None
if (self._restore_kbps is not None and queue_s == 0 and last_queue_s == 0 and loss <= self.LOSS
and now >= self._restore_after):
restored = max(current, self._restore_kbps)
self._restore_kbps = self._cruise_kbps = self.capacity_kbps = None
self.hold_until = self._drain_until = now
self.strikes = 0
self.restored = restored > current
if self.restored:
self._restored_at = now
return max(floor, min(ceiling, restored))
drained = (
self._cruise_kbps is not None
and now >= self._drain_until
and not queue_s
and loss <= self.LOSS
)
if drained:
current, self._cruise_kbps = self._cruise_kbps, None
if queue_s:
self.strikes = 0
# Like with like: the delivery against what the stream sent of the same frames. A tick
# the gate held measured no offer, and backs off.
sent_bps = acked_offered_bps if offered_bps is not None and acked_offered_bps is not None \
else offered_bps
limited = (sent_bps is not None
and sent_bps < current * 1_000 * self.APP_LIMITED
and (goodput_bps <= 0 or sent_bps <= goodput_bps / self.APP_LIMITED_DELIVERED))
if now >= self.hold_until and not limited:
capacity = max(goodput_bps, capacity_bps or 0.0)
grew = last_queue_s is not None and queue_s - last_queue_s >= self.STALL_QUEUE_S
stalled = (queue_s >= self.STALL_QUEUE_S and now - self._restored_at > self.STALL_EPISODE_S
and (grew or capacity < current * 1_000 * self.STALL_DELIVERY))
began = self._backoff_from
if stalled or (capacity <= 0 and now - self._restored_at > self.STALL_EPISODE_S):
earlier = began[0] if began and now - began[1] <= self.STALL_EPISODE_S else 0.0
if self._restore_kbps is not None:
self._restore_kbps = max(self._restore_kbps, current, earlier)
elif clean_kbps is not None:
self._restore_kbps = min(max(current, earlier), clean_kbps)
elif (self._restore_kbps is not None and offered_bps is not None
and capacity < self._restore_kbps * 1_000):
self._restore_kbps = None
self._backoff_from = (current, now)
current = self._queue_backoff(current, capacity / 1_000, queue_s, now)
self._restore_after = now if stalled else self.hold_until
return max(floor, min(ceiling, current))
if drained:
self.strikes = 0
return max(floor, min(ceiling, current))
if loss > self.LOSS:
self.strikes += 1
if self.strikes < 2:
return max(floor, min(ceiling, current))
self.strikes = 0
self._cruise_kbps = None
self.hold_until = now + self.HOLD_S
return max(floor, min(ceiling, current * self.BACKOFF))
self.strikes = 0
if now < self.hold_until:
return max(floor, min(ceiling, current))
step = self.STEP
capacity = self.capacity_kbps
delivered = goodput_bps / 1_000
if capacity is not None:
if delivered * self.NEAR > capacity:
self.capacity_kbps = None
elif delivered >= capacity * self.NEAR:
step = self.STEP_NEAR
raised = current * step ** dt
filled = offered_bps is not None and offered_bps >= current * 1_000 * self.APP_LIMITED
if probed and filled:
raised = max(raised, current * self.PROBE_STEP ** dt)
if not filled:
raised = min(raised, max(current, self.capacity_kbps or current))
return max(floor, min(ceiling, max(raised, delivered * self.HEADROOM)))paramcurrentfloatThe target in force, in kbps.
paramceilingfloatThe configured target, which the steer never exceeds.
paramfloorfloatThe lowest target allowed.
paramgoodput_bpsfloatWhat the path delivered over the tick, or 0 when unmeasured.
paramlossfloatThe loss fraction the receiver reported over the tick.
paramnowfloatMonotonic time of the tick.
paramqueue_sOptional[float]= NoneThe queue standing on the path through the tick, in seconds: 0 for a clean tick, None where the transport measures no delay.
paramoffered_bpsOptional[float]= NoneWhat the stream sent over the tick, or None when unmeasured.
paramacked_offered_bpsOptional[float]= NoneWhat the stream sent of the frames goodput_bps counts, over the span it sent them in, where the transport measures delivery by acked frames: the queue's app-limited test takes it over offered_bps, since the tick's offer counts a burst's tail still unacked, which the delivery does not.
paramcapacity_bpsOptional[float]= NoneWhat the path carried while the queue built, where the transport measured it apart from goodput_bps (a burst over a trickle, whose window's rate the trickle dilutes); the backoff takes the higher.
paramprobedbool= FalseWhether the transport padded the tick's stream up to the
target (PathProbe): a clean tick it filled to half the target
or more raises the target PROBE_STEP a second.
Returns
floatfunc_queue_backoff(current, delivered_kbps, queue_s, now) -> floatThe target a standing queue of queue_s backs off to, with the drain it plans.
Source Code
def _queue_backoff(self, current: float, delivered_kbps: float, queue_s: float, now: float) -> float:
"""The target a standing queue of `queue_s` backs off to, with the drain it plans."""
if delivered_kbps <= 0:
self._cruise_kbps = None
self.hold_until = now + self.HOLD_S
return current * self.BACKOFF
self.capacity_kbps = delivered_kbps
fraction = max(self.DRAIN_FLOOR, self.HEADROOM - queue_s / self.DRAIN_S)
self._drain_until = now + queue_s / (1.0 - fraction)
self._cruise_kbps = delivered_kbps * self.HEADROOM
self.hold_until = self._drain_until + self.SETTLE_S
return min(current, delivered_kbps * fraction)paramcurrentfloatparamdelivered_kbpsfloatparamqueue_sfloatparamnowfloatReturns
float