Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/selkies/webrtc/pacer.py: 19%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# This Source Code Form is subject to the terms of the Mozilla Public
2# License, v. 2.0. If a copy of the MPL was not distributed with this
3# file, You can obtain one at https://mozilla.org/MPL/2.0/.
4#
5# Strict-priority packet pacer for the DTLS transport choke point.
6#
7# One RtpPacer per RTCDtlsTransport. Priority classes, highest first:
8# RTCP + audio (which bypass the token bucket entirely: they are protocol
9# rate-limited, tiny, and latency-critical), data-channel, video (incl. RTX and
10# FEC).
11#
12# Scheduling is an event-driven token bucket. A packet goes straight out when
13# nothing is queued and credit covers it; otherwise it queues and the drain
14# task sleeps to the exact instant credit covers the highest-priority head. An
15# enqueue that credit already covers pokes the drain awake, so neither sleep
16# overshoot nor coalescing adds latency to another class.
17#
18# Invariants:
19# * The burst budget always covers one max-size packet, otherwise a packet
20# could never become affordable and its class would wedge forever.
21# * The burst budget follows the link, never the pace alone: 5 ms of the
22# pace until transport-cc feedback typically shows the wire delivering
23# frames' leading bursts at twice the encoder's rate or more without loss,
24# then what that link delivers in 10 ms until it typically falls below 1.5
25# times the encoder's rate, loses a packet, delivers less than the encoder
26# makes, or goes unmeasured (`_apply_pace`, `set_link_bps`).
27# * Video is the only droppable class; every other class is bounded by
28# backpressure on the sender instead.
29# * The video queue budget is max(CAP_MIN_MS of wire time, IDR_FLOOR_FACTOR x
30# the largest of the last IDR_WINDOW keyframes). A budget below one IDR
31# truncates every keyframe under sustained congestion and breaks the
32# reference chain permanently, hence the sliding-window floor.
33# * Overflow drops the whole GOP and requests a throttled keyframe rather
34# than thinning arbitrary packet tails; a timeout resurrects video if no
35# keyframe arrives, so an unbound keyframe callback or a stuck encoder
36# cannot kill the class permanently (natural IDR cadence can be minutes).
37# Where the video sender can resync its peer without one (`resync`: the
38# encoder names each frame's reference and the peer reads the dependency
39# descriptor), the drop is handed to it instead: it holds back the frames
40# the purge left undecodable and has the encoder predict past them, and
41# video flows on, the shared stream carrying no keyframe for one peer.
42# * Rate control is AIMD on wire evidence: an internal overflow only proves
43# injection > drain, and with UDP sends the drain is the pace setting, not
44# the wire — once one brake drops the pace below the encoder rate, every
45# further overflow is self-inflicted. So an overflow resets the GOP but
46# sizes a brake only from a fresh goodput estimate, and only one measured
47# on a feedback window the wire cut in several places, since a window
48# delivered whole reads back the pacer's own output and one cut in a
49# single run is an outage (estimate - BYPASS_RESERVE_BPS, floored at
50# AIMD_FLOOR_FACTOR x the encoder target),
51# at most one brake per BRAKE_HOLD_S; recovery is +25%/s toward
52# PACE_FACTOR x the encoder target, and only a brake resets its clock, so
53# reset churn cannot hold the pace at the floor.
54import asyncio
55import logging
56import os
57import statistics
58import time
59from typing import Any, Awaitable, Callable, Dict, Deque, Optional
60from collections import deque
62logger = logging.getLogger("selkies_webrtc_pacer")
64# Priority classes (lower value wins). RTCP + audio share a class: both are
65# tiny absolute rates and must never queue behind anything else.
66CLASS_RTCP = 0
67CLASS_AUDIO = 0
68# CLASS_DC carries data-channel traffic (input, status, clipboard);
69# CLASS_VIDEO carries video RTP, RTX, and FEC.
70CLASS_DC = 1
71CLASS_VIDEO = 2
72# Classes that own a queue: class 0 bypasses the bucket and is never enqueued.
73_QUEUED_CLASSES = (CLASS_DC, CLASS_VIDEO)
75# Pace above the encoder target when the link allows.
76PACE_FACTOR = 2.5
77# Wire share reserved for bypass classes (audio + RTCP + DC) when braking
78# from an estimate.
79BYPASS_RESERVE_BPS = 500_000
80# AIMD: multiplicative pace recovery (+25%/s) in the absence of internal
81# overflow events.
82OVERFLOW_RECOVERY_PER_S = 0.25
83# Brake depth floor (x encoder target): chained overflow brakes must never
84# crater the stream into a starvation valley that +25%/s takes seconds to
85# climb out of.
86AIMD_FLOOR_FACTOR = 0.35
87# Burst credit window: how much wire time at the pace the bucket may hoard
88# while the link's room is unknown or short. Wider windows let video bursts
89# land in front of audio at a bottleneck.
90DEBT_WINDOW_S = 0.005
91# A link measured to have room takes the burst it delivers in BURST_LINK_S,
92# never more than libwebrtc's send burst interval at the pace under its byte
93# cap, which keeps a burst inside a socket's send buffer.
94BURST_LINK_S = 0.010
95BURST_WINDOW_S = 0.040
96MAX_BURST_BYTES = 63_000
97# Room is a link delivering a frame's leading burst at this many times the
98# encoder's rate or faster, typically over the last LINK_SAMPLES feedback
99# packets and measured within LINK_MAX_AGE_S; a link found to have room keeps
100# it down to LINK_KEEP_FACTOR times that rate.
101LINK_HEADROOM_FACTOR = 2.0
102LINK_KEEP_FACTOR = 1.5
103LINK_SAMPLES = 5
104LINK_MAX_AGE_S = 2.0
105# The share of the pace a still screen's cleanup may run at (`room_bps`): the pacer drains its
106# frames with room for audio, retransmissions and FEC beside them.
107CLEANUP_PACE_SHARE = 0.8
108# Floor under the burst budget. Credit saturates at the budget, so a budget
109# below one packet leaves that packet unaffordable forever; DTLS records here
110# are MTU-sized and this keeps headroom above them.
111BURST_FLOOR_BYTES = 2048
112# Video queue budget floor, time-denominated.
113CAP_MIN_MS = 120
114# IDR-size floor for the video queue budget, over a rolling window of recent
115# IDR sizes.
116IDR_FLOOR_FACTOR = 2.2
117IDR_WINDOW = 4
118KEYREQ_MIN_INTERVAL_S = 0.5
119# Sanity floor: a 0-goodput feedback window must never stall (or
120# divide-by-zero) the scheduler.
121MIN_PACE_BPS = 100_000
122# Defensive cap on drain sleeps; pokes handle the rest.
123DRAIN_MAX_SLEEP_S = 0.05
124# Frames whose end `frame_end` waits on at once; more than are ever in flight.
125FRAME_ENDS_MAX = 64
126# Post-enable grace: ignore goodput braking so the first (audio-only/startup)
127# feedback windows cannot slam the pace before video even ramps.
128GOODPUT_WARMUP_S = 5.0
129# At most one brake per hold: while the pace sits below the encoder rate the
130# queue refills from injection alone, and those overflows carry no wire signal.
131BRAKE_HOLD_S = 1.0
132# A brake only trusts an estimate this fresh; one bad feedback window (an
133# upload delaying feedback, a WiFi hiccup) must not size every later brake.
134GOODPUT_MAX_AGE_S = 2.0
135# If no keyframe resurrects a dead GOP within this of the last keyreq,
136# resurrect on queue-drain anyway.
137RESURRECT_TIMEOUT_S = 1.0
138# Data-channel backpressure: senders block once the backlog passes the high
139# water mark and are released again below the low water mark.
140DC_HIGH_WATER_BYTES = 2_000_000
141DC_LOW_WATER_BYTES = 1_000_000
142# Cap on one backpressure wait: a wedged wire must slow senders, never park
143# them forever.
144DC_BLOCK_TIMEOUT_S = 1.0
147def _stale_deadline_s() -> float:
148 """Read SELKIES_WEBRTC_PACER_STALE_MS as seconds; a malformed value disables
149 the deadline instead of killing the import."""
150 raw = os.environ.get("SELKIES_WEBRTC_PACER_STALE_MS", "0")
151 try:
152 return max(0.0, float(raw)) / 1000.0
153 except ValueError:
154 logger.warning("pacer: ignoring malformed "
155 "SELKIES_WEBRTC_PACER_STALE_MS=%r", raw)
156 return 0.0
159# Whole-GOP deadline for queued video, never tighter than one IDR's wire time.
160# Off by default: purging the backlog stalls the receiver's jitter buffer on the
161# resulting gap, which costs more delivered-video latency than the freshness it
162# buys on most links. Set SELKIES_WEBRTC_PACER_STALE_MS to enable.
163VIDEO_STALE_S = _stale_deadline_s()
164# Near-empty windows carry no rate signal.
165MIN_GOODPUT_SAMPLE_BYTES = 2048
166# What a feedback window must show for its delivery rate to size a brake: this
167# share of it lost, in at least this many separate runs. Less says the window
168# measured what the pacer sent; one run of any size is an outage.
169BRAKE_LOSS_FRACTION = 0.1
170BRAKE_LOSS_RUNS = 3
172SendNow = Callable[[bytes], Awaitable[None]]
175class RtpPacer:
176 """Token-bucket pacer with strict priority classes for one RTP/DTLS flow.
178 Owned by one RTCDtlsTransport. All classes share a single credit window.
179 Video packets that cannot fit the queue budget trigger a GOP reset: queued
180 video is purged, subsequent video is dropped until a fresh keyframe
181 resurrects the stream, and a throttled keyframe request is emitted.
182 Video is the only class ever dropped; data-channel senders are throttled by
183 backpressure instead, since their traffic is reliable.
184 """
186 def __init__(
187 self,
188 encoder_bps: int,
189 send_now: SendNow,
190 send_now_data: Optional[SendNow] = None,
191 request_keyframe: Optional[Callable[[], None]] = None,
192 loop: Optional[asyncio.AbstractEventLoop] = None,
193 on_dropped: Optional[Callable[[int], None]] = None,
194 resync: Optional[Callable[[Optional[int]], bool]] = None,
195 ) -> None:
196 self._encoder_bps = max(int(encoder_bps), 100_000)
197 self._goodput_bps: Optional[int] = None
198 self._goodput_at = 0.0
199 self._brake_hold_until = 0.0
200 self._send_now = send_now
201 self._send_now_data = send_now_data or send_now
202 self._request_keyframe = request_keyframe
203 # Told the tag of every video packet dropped, refused, or purged, so the
204 # transport's loss accounting can leave them out.
205 self._on_dropped = on_dropped
206 # Offered a GOP reset's oldest dropped tag; True when the video sender
207 # resyncs its peer itself, so no keyframe is asked for (module notes).
208 self._resync = resync
209 self._loop = loop or asyncio.get_running_loop()
210 # Tag of a frame's last packet -> what to call, and with what, once it is
211 # on the wire (`frame_end`); a dropped one is forgotten with its packet.
212 self.frame_ends: Dict[int, tuple] = {}
214 self._queues: Dict[int, Deque[bytes]] = {c: deque() for c in _QUEUED_CLASSES}
215 self._poke = asyncio.Event()
216 # Set while the backlog is below the low-water mark: data-channel
217 # senders wait on it once the backlog crosses the high-water mark.
218 self._drained = asyncio.Event()
219 self._drained.set()
220 self._make_class_table()
221 self._bytes_queued = 0
222 self._video_bytes = 0
223 # Enqueue-time mirror of the video queue: head age drives the
224 # latency-first stale-GOP deadline. Kept in lockstep at every
225 # mutation site (enqueue / trim / drain / wholesale clears).
226 self._video_ts: Deque[float] = deque()
227 self._video_tags: Deque[Optional[int]] = deque()
228 self.credit = 0.0
229 self._debt_cap = 0.0
230 self._narrow_cap = 0.0
231 self._link_bps: Optional[float] = None
232 self._link_at = 0.0
233 self._link_recent: Deque[float] = deque(maxlen=LINK_SAMPLES)
234 self._link_room = False
235 self._pace_bps = MIN_PACE_BPS
236 # Sentinel for "AIMD owns the pace now"; distinct from the recovery
237 # clock, which ticks on every pace update.
238 self._ever_overflowed = False
239 self._last_pace_update_at = 0.0
240 self._refresh_windows()
241 self._last = time.monotonic()
242 self._drain_task: Optional[asyncio.Task] = None
243 self._stopped = False
244 self._idr_sizes: Deque[int] = deque(maxlen=IDR_WINDOW)
245 self._idr_floor_bytes = 0
246 self._frame_floor_bytes = 0
247 self._gop_dead = False
248 self._last_keyreq = 0.0
249 self._keyreq_answered = True
250 self._enabled_at = self._last
251 self._gop_dead_at = 0.0
252 self._purged_first: Optional[int] = None
253 self._oversize_warned = False
254 self.stats = {
255 "video_dropped": 0, "keyreqs": 0, "gop_resets": 0, "resyncs": 0,
256 "idr_resurrects": 0, "timeout_resurrects": 0, "stale_resets": 0,
257 "paced_bytes": 0, "queue_max_bytes": 0, "fastpath_bytes": 0,
258 }
259 self._stats_prev: dict = {}
260 self._stats_timer: Optional[asyncio.TimerHandle] = None
261 self._arm_stats_log()
263 def _arm_stats_log(self) -> None:
264 if self._stopped:
265 return
266 self._stats_timer = self._loop.call_later(10.0, self._maybe_log_stats)
268 def _maybe_log_stats(self) -> None:
269 """Periodic counter deltas, at debug: a streaming link moves them every
270 interval, so this is one line per pacer per interval for as long as the
271 session lasts. The same numbers are exported as webrtc_pacer_* gauges,
272 which is where a running deployment should read them."""
273 changed = {
274 k: self.stats[k] - self._stats_prev.get(k, 0)
275 for k in self.stats
276 if self.stats[k] != self._stats_prev.get(k, 0)
277 }
278 self._stats_prev = dict(self.stats)
279 if changed:
280 logger.debug(
281 "pacer: +%s | total %s", changed, self.snapshot())
282 self._arm_stats_log()
284 # ------------------------------------------------------------------ rates
285 def set_encoder_bps(self, bps: int) -> None:
286 bps = max(int(bps), 100_000)
287 if bps != self._encoder_bps:
288 self._encoder_bps = bps
289 self._refresh_windows()
291 def set_goodput_bps(self, bps: Optional[float]) -> None:
292 # The estimate is CONTROL INPUT ONLY AT OVERFLOW EVENTS: braking on
293 # own-output measurement continuously is a feedback loop that always
294 # drifts (saturated wire = link, not capacity; unsaturated wire = own
295 # injection). AIMD: estimates merely size the step taken on overflow.
296 if bps is None:
297 return
298 now = time.monotonic()
299 if now - self._enabled_at < GOODPUT_WARMUP_S:
300 # The first feedback windows after enabling carry audio-only (or
301 # no) traffic: their throughput measures the offered load, not the
302 # link, and would size an early brake far below capacity.
303 return
304 bps = int(bps)
305 if bps > 0:
306 self._goodput_bps = bps
307 self._goodput_at = now
309 def set_link_bps(self, bps: Optional[float]) -> None:
310 """Take the rate the wire delivered a frame's leading burst at.
312 Args:
313 bps: The rate from the latest transport-cc feedback; 0 when that
314 feedback showed the wire losing packets, None when it held no
315 burst to measure, which leaves the last estimate to age out.
316 """
317 if bps is None:
318 return
319 self._link_bps = float(bps)
320 self._link_at = time.monotonic()
321 if bps > 0:
322 self._link_recent.append(float(bps))
323 else:
324 self._link_recent.clear()
325 self._apply_pace()
327 def room_bps(self) -> Optional[float]:
328 """The rate a still screen's cleanup may take through this pacer: what the wire delivered
329 frames' leading bursts at, typically (the median of the last LINK_SAMPLES estimates,
330 measured within LINK_MAX_AGE_S), within CLEANUP_PACE_SHARE of the pace; 0 where the
331 latest feedback showed the wire losing packets, None where none measured it."""
332 if self._link_bps is None or time.monotonic() - self._link_at > LINK_MAX_AGE_S:
333 return None
334 if not self._link_bps:
335 return 0.0
336 typical = statistics.median(self._link_recent) if self._link_recent else self._link_bps
337 return min(typical, CLEANUP_PACE_SHARE * self._pace_bps)
339 def set_frame_floor(self, size: int) -> None:
340 """Take a video frame of `size` bytes into the queue whole (`_video_cap_bytes`): a still
341 screen's cleanup frame at the rate `room_bps` gave."""
342 self._frame_floor_bytes = max(0, int(size))
344 @property
345 def burst_probe_bytes(self) -> float:
346 """How much of a frame leaves unpaced whatever the link's room: the
347 leading bytes whose arrivals measure the link rather than the pace."""
348 return self._narrow_cap
350 def _on_overflow(self) -> None:
351 """Brake on queue overflow, sized by wire evidence alone.
353 The overflow itself only proves injection > drain, and with UDP sends
354 the drain is the pace setting, not the wire — after one brake drops
355 the pace below the encoder rate, every further overflow is
356 self-inflicted. So a brake needs a fresh goodput estimate and runs at
357 most once per BRAKE_HOLD_S; without an estimate the GOP reset stands
358 alone. The rejected alternative — halve on every overflow — re-braked
359 on the self-inflicted ones faster than recovery could climb and
360 latched the pace at the floor until the page reloaded.
361 """
362 now = time.monotonic()
363 if now < self._brake_hold_until:
364 return
365 est = (self._goodput_bps
366 if now - self._goodput_at <= GOODPUT_MAX_AGE_S else None)
367 if not est:
368 return
369 target = max(MIN_PACE_BPS, est - BYPASS_RESERVE_BPS)
370 # Depth floor: never crater below a usable video share of the user's
371 # encoder target (starvation valleys take seconds to recover from).
372 target = max(target, int(AIMD_FLOOR_FACTOR * self._encoder_bps))
373 if target < self._pace_bps:
374 self._ever_overflowed = True
375 self._brake_hold_until = now + BRAKE_HOLD_S
376 self._last_pace_update_at = now
377 self._pace_bps = target
378 self._apply_pace()
380 def _maybe_recover_pace(self) -> None:
381 # +25%/s multiplicative recovery toward the encoder ceiling; only a
382 # brake resets the clock, so reset churn cannot stall the climb.
383 now = time.monotonic()
384 dt = now - self._last_pace_update_at
385 if dt < 0.25:
386 return
387 ceiling = PACE_FACTOR * self._encoder_bps
388 self._last_pace_update_at = now
389 if self._link_bps and now - self._link_at > LINK_MAX_AGE_S:
390 self._link_bps = None
391 self._link_recent.clear()
392 self._apply_pace()
393 if self._pace_bps >= ceiling:
394 return
395 grown = int(self._pace_bps * (1.0 + OVERFLOW_RECOVERY_PER_S) ** min(dt, 4.0))
396 self._pace_bps = min(int(ceiling), max(grown, MIN_PACE_BPS))
397 self._apply_pace()
399 def _make_class_table(self) -> None:
400 """Precomputed per-class (class, queue, sender) triples in priority
401 order: per-packet work must not re-derive what never changes."""
402 self._class_table = [
403 (CLASS_DC, self._queues[CLASS_DC], self._send_now_data),
404 (CLASS_VIDEO, self._queues[CLASS_VIDEO], self._send_now),
405 ]
407 def _apply_pace(self) -> None:
408 """Re-derive the burst budget from the pace and the link, and clamp
409 credit to it.
411 The budget is DEBT_WINDOW_S of the pace while the link's room is
412 unknown or short. A link the latest feedback showed delivering a
413 frame's leading burst at LINK_HEADROOM_FACTOR times the encoder's rate
414 or faster, with nothing lost on the wire and the pace not braked, takes
415 what it delivers in BURST_LINK_S, at most libwebrtc's burst, so a
416 frame leaves whole instead of trickling out at the pace behind its
417 first packets. Room is judged on the median of the last LINK_SAMPLES
418 estimates and kept while that stays at LINK_KEEP_FACTOR times the
419 encoder's rate or more: one burst's rate scatters with the receiver's
420 own timing, and a single estimate against a single threshold flips the
421 budget on every other feedback of a link close to it. The burst is
422 sized by the lower of the latest estimate and that median, and a
423 latest estimate below the encoder's rate closes the room at once, so
424 a link that drops is never burst into at its old rate. Sizing the burst
425 to the measured link rather than to the pace keeps the wide one off a
426 link near capacity, where it only moves the queue from the pacer to
427 the bottleneck, ahead of audio.
428 Credit saturates at the budget, so it is floored at
429 BURST_FLOOR_BYTES: below one packet's size, that packet could never
430 become affordable and its class would wedge."""
431 pace_bytes = self._pace_bps / 8.0
432 self._narrow_cap = cap = max(pace_bytes * DEBT_WINDOW_S, float(BURST_FLOOR_BYTES))
433 link = self._link_bps
434 typical = statistics.median(self._link_recent) if self._link_recent else 0.0
435 factor = LINK_KEEP_FACTOR if self._link_room else LINK_HEADROOM_FACTOR
436 self._link_room = bool(link and link >= self._encoder_bps
437 and typical >= factor * self._encoder_bps
438 and time.monotonic() - self._link_at <= LINK_MAX_AGE_S)
439 if self._link_room and self._pace_bps >= int(PACE_FACTOR * self._encoder_bps):
440 cap = max(cap, min(min(link, typical) / 8.0 * BURST_LINK_S, pace_bytes * BURST_WINDOW_S,
441 float(MAX_BURST_BYTES)))
442 self._debt_cap = cap
443 if self.credit > cap:
444 self.credit = cap
446 def _refresh_windows(self) -> None:
447 ceiling = PACE_FACTOR * self._encoder_bps
448 # Until the first brake the encoder setting owns the pace (config
449 # tracking); afterwards it can only CLAMP the AIMD-controlled pace.
450 if not self._ever_overflowed:
451 self._pace_bps = int(ceiling)
452 else:
453 self._pace_bps = min(self._pace_bps, int(ceiling))
454 self._pace_bps = max(self._pace_bps, MIN_PACE_BPS)
455 self._apply_pace()
457 def _video_cap_bytes(self) -> int:
458 base = int(self._pace_bps / 8.0 * CAP_MIN_MS / 1000)
459 return max(base, self._idr_floor_bytes, self._frame_floor_bytes)
461 # ----------------------------------------------------------------- inputs
462 def note_keyframe(self, total_payload_bytes: int, natural: bool = True) -> None:
463 """Feed the sliding-window IDR floor and resurrect the stream after a
464 GOP reset.
466 Only NATURAL keyframes enter the window: a forced keyframe is emitted at
467 the collapsed bitrate that made us ask for it, so letting it evict a
468 window entry would shrink the floor, shrink the cap, and trigger the next
469 reset — a self-reinforcing keyframe-churn cycle. A forced keyframe may
470 still RAISE the floor, so a cap too small to hold one IDR still grows out
471 of the churn.
473 A keyframe arriving within KEYREQ_MIN_INTERVAL_S of our own request is
474 treated as forced whatever the caller says: on the pre-encoded pack()
475 path the sender packetizes frames it did not encode, so it cannot tell
476 the two apart.
477 """
478 size = int(total_payload_bytes)
479 if natural and time.monotonic() - self._last_keyreq < KEYREQ_MIN_INTERVAL_S:
480 natural = False
481 if natural:
482 self._idr_sizes.append(size)
483 self._idr_floor_bytes = int(max(self._idr_sizes) * IDR_FLOOR_FACTOR)
484 else:
485 self._idr_floor_bytes = max(self._idr_floor_bytes,
486 int(size * IDR_FLOOR_FACTOR))
487 self._keyreq_answered = True
488 if self._gop_dead:
489 self._gop_dead = False
490 self.stats["idr_resurrects"] += 1
491 logger.debug("pacer: keyframe of %d bytes resurrects video %.0f ms after the reset",
492 size, (time.monotonic() - self._gop_dead_at) * 1000)
494 def request_keyframe_once(self) -> None:
495 """Ask for a keyframe, unless one was asked for within
496 KEYREQ_MIN_INTERVAL_S and has not arrived yet: that one resurrects the
497 stream when it comes, while a reset after it landed needs its own, or
498 video stays dead until the resurrect timeout."""
499 now = time.monotonic()
500 if not self._keyreq_answered and now - self._last_keyreq < KEYREQ_MIN_INTERVAL_S:
501 return
502 self._last_keyreq = now
503 self._keyreq_answered = False
504 if self._request_keyframe is None:
505 logger.warning("pacer: no keyframe callback bound; GOP reset "
506 "relies on the %.1fs timeout resurrect", RESURRECT_TIMEOUT_S)
507 return
508 self.stats["keyreqs"] += 1
509 try:
510 self._request_keyframe()
511 except Exception:
512 logger.debug("pacer keyframe request failed", exc_info=True)
514 # ------------------------------------------------------------------- send
515 def video_room(self) -> int:
516 """Bytes the video queue takes before a packet overflows it (`_reset_gop`)."""
517 return self._video_cap_bytes() - self._video_bytes
519 def drain_s(self) -> float:
520 """Seconds until what is queued now has left, at the current pace."""
521 return self._bytes_queued * 8 / max(self._pace_bps, 1)
523 def _accrue(self) -> None:
524 now = time.monotonic()
525 self.credit = min(self._debt_cap,
526 self.credit + (now - self._last) * self._pace_bps / 8.0)
527 self._last = now
529 async def send(self, data: bytes, cls: int, tag: Optional[int] = None) -> bool:
530 """Send `data` now, queue it, or drop it.
532 Args:
533 data: The datagram.
534 cls: Its priority class.
535 tag: What to report through `on_dropped` if it is dropped.
537 Returns:
538 True when the packet went out or queued; False when it was dropped,
539 which only video is: refused while its GOP is abandoned, or the
540 packet that overflowed the budget or found the queue stale and
541 abandoned the GOP itself.
542 """
543 if self._stopped:
544 raise ConnectionError("pacer stopped")
545 n = len(data)
547 # RTCP/audio bypass: protocol rate-limited and latency-critical, so
548 # they never queue and never consume video's credit.
549 if cls == CLASS_RTCP:
550 self.stats["fastpath_bytes"] += n
551 await self._send_now(data)
552 return True
554 self._maybe_recover_pace()
556 if cls != CLASS_VIDEO and self._bytes_queued >= DC_HIGH_WATER_BYTES:
557 await self._await_backlog_drain()
558 if self._stopped:
559 raise ConnectionError("pacer stopped")
561 if cls == CLASS_VIDEO and self._gop_dead:
562 # Safety net: if no keyframe resurrected us within the timeout of
563 # the last keyreq (broken/unbound callback, encoder stuck), let
564 # video flow again anyway. The decoder recovers from the reference
565 # break at the next keyframe; a permanently dead class does not.
566 if time.monotonic() - self._gop_dead_at >= RESURRECT_TIMEOUT_S:
567 self._gop_dead = False
568 self.stats["timeout_resurrects"] += 1
569 logger.info("pacer: no keyframe within %.1fs of reset; "
570 "resurrecting video optimistically", RESURRECT_TIMEOUT_S)
571 else:
572 self.stats["video_dropped"] += 1
573 self._drop(tag)
574 return False
576 # Latency-first stale deadline: video sitting in queue longer than
577 # the deadline is delivered-too-late by definition, so the whole GOP
578 # is purged and a fresh keyframe requested (fewer stale frames > more
579 # stale frames). The deadline is never tighter than one IDR's wire
580 # time — otherwise reset churn would deliver zero video.
581 if VIDEO_STALE_S > 0 and cls == CLASS_VIDEO and self._video_ts:
582 deadline = VIDEO_STALE_S
583 if self._idr_floor_bytes:
584 idr_time = self._idr_floor_bytes * 8.0 / max(self._pace_bps, 1)
585 if idr_time > deadline:
586 deadline = idr_time
587 now = time.monotonic()
588 if now - self._video_ts[0] > deadline:
589 self._stale_reset(deadline, tag)
590 self.stats["video_dropped"] += 1
591 self._drop(tag)
592 return False
594 # Fast path when nothing is buffered: anything credit covers goes
595 # straight out (~0 added delay below the rate).
596 if not self._bytes_queued:
597 self._accrue()
598 if n <= self.credit:
599 self.credit -= n
600 self.stats["fastpath_bytes"] += n
601 sender = self._send_now_data if cls == CLASS_DC else self._send_now
602 await sender(data)
603 if self.frame_ends and tag is not None:
604 self._left(tag)
605 return True
607 # Video queue budget: a packet the budget cannot hold abandons the GOP,
608 # queue and all. The cap is video-only: audio/DC never push an IDR out.
609 if cls == CLASS_VIDEO and self._video_bytes + n > self._video_cap_bytes():
610 self._reset_gop(tag=tag)
611 self.stats["video_dropped"] += 1
612 self._drop(tag)
613 return False
615 self._queues[cls].append(data)
616 if cls == CLASS_VIDEO:
617 self._video_bytes += n
618 self._video_ts.append(time.monotonic())
619 self._video_tags.append(tag)
620 self._bytes_queued += n
621 if self._bytes_queued > self.stats["queue_max_bytes"]:
622 self.stats["queue_max_bytes"] = self._bytes_queued
623 # Wake-on-affordable: if the drain is sleeping through a big head
624 # packet but credit already covers THIS newcomer, don't make it wait
625 # for the head's deadline.
626 self._accrue()
627 if n <= self.credit:
628 self._poke.set()
629 self._kick()
630 return True
632 def _stale_reset(self, deadline_s: float, tag: Optional[int] = None) -> None:
633 """Latency-first GOP reset for queued video that outlived its usefulness."""
634 self._reset_gop("video backlog stale (>%.0fms)" % (deadline_s * 1000), tag)
635 self.stats["stale_resets"] += 1
637 def frame_end(self, tag: int, sink: Callable[..., None], *args: Any) -> None:
638 """Call `sink(*args)` once the video packet tagged `tag`, the last of its
639 frame, leaves for the wire: how long a frame took to get there includes the
640 time it queued here. At most FRAME_ENDS_MAX are held; the oldest goes first."""
641 ends = self.frame_ends
642 ends[tag] = (sink, args)
643 if len(ends) > FRAME_ENDS_MAX:
644 del ends[next(iter(ends))]
646 def _left(self, tag: Optional[int]) -> None:
647 """A video packet went out: call what its frame's end was waiting on."""
648 end = self.frame_ends.pop(tag, None)
649 if end is not None:
650 end[0](*end[1])
652 def _drop(self, tag: Optional[int]) -> None:
653 if tag is not None:
654 self.frame_ends.pop(tag, None)
655 if self._on_dropped is not None:
656 self._on_dropped(tag)
658 def _purge_video(self) -> int:
659 """Drop the whole video queue, keeping the byte counters and the
660 enqueue-time and tag mirrors in lockstep with it; the packets dropped.
661 The oldest tag dropped is kept in `_purged_first`."""
662 dq = self._queues[CLASS_VIDEO]
663 n = len(dq)
664 if dq:
665 self.stats["video_dropped"] += n
666 dq.clear()
667 self._purged_first = next((t for t in self._video_tags if t is not None), None)
668 for tag in self._video_tags:
669 self._drop(tag)
670 self._video_tags.clear()
671 self._video_ts.clear()
672 self._bytes_queued -= self._video_bytes
673 self._video_bytes = 0
674 return n
676 def _reset_gop(self, reason: str = "video queue overflow", tag: Optional[int] = None) -> None:
677 """Abandon the GOP: purge the queued video, which nothing behind the
678 requested keyframe can use, refuse video until that keyframe, and
679 brake if wire evidence sizes one. Where the video sender resyncs its
680 peer itself (`_resync`), told the oldest packet lost (the queue's, else
681 `tag`, the packet at hand), the purge and the brake are all: video
682 flows on and no keyframe is asked for."""
683 purged = self._purge_video()
684 first = self._purged_first if self._purged_first is not None else tag
685 if not self._gop_dead and self._resync is not None and self._resync(first):
686 self._on_overflow()
687 self.stats["resyncs"] += 1
688 logger.info("pacer: %s => %d queued packets purged; the sender resyncs its peer",
689 reason, purged)
690 return
691 if not self._gop_dead:
692 self._gop_dead = True
693 self._gop_dead_at = time.monotonic()
694 self._on_overflow()
695 self.stats["gop_resets"] += 1
696 logger.info("pacer: %s => GOP reset, %d queued packets purged, keyframe requested",
697 reason, purged)
698 self.request_keyframe_once()
700 # ------------------------------------------------------------------ drain
701 def _kick(self) -> None:
702 task = self._drain_task
703 if self._stopped or (task is not None and not task.done()):
704 return
705 self._drain_task = self._loop.create_task(self._drain())
707 async def _await_backlog_drain(self) -> None:
708 """Backpressure for the reliable classes: hold the caller until the
709 backlog falls back to the low-water mark. Bounded so a wedged wire can
710 only slow a sender, never park it."""
711 self._drained.clear()
712 self._kick()
713 try:
714 await asyncio.wait_for(self._drained.wait(), DC_BLOCK_TIMEOUT_S)
715 except asyncio.TimeoutError:
716 logger.warning("pacer: %d bytes still queued after %.1fs of "
717 "backpressure; letting the sender through",
718 self._bytes_queued, DC_BLOCK_TIMEOUT_S)
720 def _release_senders(self) -> None:
721 if not self._drained.is_set():
722 self._drained.set()
724 async def _drain(self) -> None:
725 try:
726 while not self._stopped:
727 # Clear the poke BEFORE the pass, not right before the wait: a
728 # poke arriving mid-pass must survive into the wait below,
729 # otherwise a newly affordable packet sleeps until the head's
730 # deadline.
731 self._poke.clear()
732 self._maybe_recover_pace()
733 self._accrue()
734 for cls, dq, sender in self._class_table:
735 while dq:
736 size = len(dq[0])
737 if size > self.credit:
738 # A packet wider than the whole burst budget can
739 # never be covered by credit; release it once the
740 # bucket is full so its class cannot wedge.
741 if size <= self._debt_cap or self.credit < self._debt_cap:
742 break
743 if not self._oversize_warned:
744 self._oversize_warned = True
745 logger.warning(
746 "pacer: %d-byte packet exceeds the %d-byte burst "
747 "budget; releasing oversized packets on a full "
748 "bucket", size, int(self._debt_cap))
749 data = dq.popleft()
750 self._bytes_queued -= size
751 tag = None
752 if cls == CLASS_VIDEO:
753 self._video_bytes -= size
754 self._video_ts.popleft()
755 tag = self._video_tags.popleft()
756 self.credit -= size
757 try:
758 await sender(data)
759 except Exception:
760 logger.warning("pacer: send failed; dropping queue",
761 exc_info=True)
762 # The receiver's reference chain dies with the
763 # purged packets: mark video dead so nothing that
764 # depends on them is sent, and ask for a keyframe,
765 # unless the sender resyncs its peer itself.
766 self._reset_gop("send failed", tag)
767 self._queues = {c: deque() for c in _QUEUED_CLASSES}
768 self._make_class_table()
769 self._bytes_queued = self._video_bytes = 0
770 self._release_senders()
771 return
772 self.stats["paced_bytes"] += size
773 if tag is not None and self.frame_ends:
774 self._left(tag)
775 if self._bytes_queued <= DC_LOW_WATER_BYTES:
776 self._release_senders()
777 if not self._bytes_queued:
778 return
779 head = 0
780 for _, dq, _sender in self._class_table:
781 if dq:
782 head = len(dq[0])
783 break
784 if not head:
785 return
786 # Nothing more is affordable: sleep to the exact instant
787 # credit covers the highest-priority queued packet — but wake
788 # early if an enqueue pokes us (newly affordable packet) and
789 # never sleep beyond the defensive cap.
790 delay = min(
791 max((head - self.credit) * 8.0 / self._pace_bps, 0.0005),
792 DRAIN_MAX_SLEEP_S,
793 )
794 try:
795 await asyncio.wait_for(self._poke.wait(), delay)
796 except asyncio.TimeoutError:
797 pass
798 finally:
799 self._drain_task = None
801 async def close(self) -> None:
802 self._stopped = True
803 # Waiters re-check _stopped and raise; nothing may stay parked here.
804 self._release_senders()
805 if self._stats_timer is not None:
806 self._stats_timer.cancel()
807 self._stats_timer = None
808 task = self._drain_task
809 if task is not None and not task.done():
810 task.cancel()
811 try:
812 await task
813 except asyncio.CancelledError:
814 pass
815 except Exception:
816 pass
817 self._drain_task = None
818 logger.info("pacer closed: %s", self.snapshot())
820 def snapshot(self) -> Dict[str, int]:
821 out = dict(self.stats)
822 out["pace_bps"] = int(self._pace_bps)
823 out["burst_bytes"] = int(self._debt_cap)
824 out["queued_bytes"] = self._bytes_queued
825 out["video_bytes"] = self._video_bytes
826 out["idr_floor_bytes"] = int(self._idr_floor_bytes)
827 out["frame_floor_bytes"] = int(self._frame_floor_bytes)
828 out["gop_dead"] = int(self._gop_dead)
829 return out