okooooooo/biliup
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(preview): 中转模式在有 RTT 的链路上频繁缓冲——旁路挪到解析点、掉队原地续播、缓冲分低延迟 2 s / 流畅 5 s 两档 (#1716)

Browse Source 
* fix(preview): stream-gears FLV 在解析点旁路，不等写盘循环整 GOP 落盘

httpflv 的写盘循环把 tag 攒在 flv_tags_cache 里，等下一个关键帧到来才把整个 GOP
一起写出——预览的旁路点原来就挂在这个写出循环里，于是预览端每个 GOP 间隔（B 站
常见 2–4 s）收到一次整 GOP 的突发、其余时间一个字节都不来，且整体落后一个 GOP。
播放器拿到的缓冲恰好在下一个突发到达时耗尽，零余量：同机 0 ms 下勉强连上，链路
上稍有 RTT / 抖动就 waiting；GOP 长于 mpegts.js 的追帧阈值时更是每个 GOP 都跳一次
再停一次（实测 GOP 4.2 s 的房间同机也有 85% 时间处于 waiting）。

把旁路挪到 tag 解析完、进缓存之前：预览按 tag 实时收到，与 mesio 路径（在条目进
管线前 tee）一致。写盘循环一个字节不变；旁路仍只是 push，不 await、不返回错误。

实测（arrival-probe，40 s）：同一斗鱼房间直连 CDN 到达连续（最大静默 0.24 s），
中转修前每 2.0 s 一个 ~940 KB 突发、20 个静默间隙全部 > 1 s。

Refs #1712 #1713

* feat(preview): 快照保留最近 6 s 的 GOP，新订阅者起播就有几秒缓冲

原来快照只含「当前 GOP」：新订阅者从零到一个 GOP 之间的缓冲起步，之后到达 = 消耗，
缓冲永远这么薄，链路上几十到几百毫秒的抖动就 waiting。浏览器直连 CDN 时 CDN 会先
给几秒的 GOP 缓存，中转得给同样的深度。

PreviewSink 里多一个已完成 GOP 的队列（按关键帧到达时刻计，保留 SNAPSHOT_WINDOW = 6 s
以内的），快照 = 文件头 + 序列头 + 队列里的 GOP + 当前 GOP，实际深度在 6 s 到 6 s +
一个 GOP 之间。字节上限 MAX_SNAPSHOT_BYTES = 16 MB（6 s × 20 Mbps），超过先丢最旧的
GOP；单个 GOP 超过 MAX_GOP_BYTES 作废时历史一并放掉（不再连续）；序列头内容变了或
fMP4 换了 init 时历史也放掉（旧 GOP 与新参数不配），内容相同的重发（stream-gears
分段时重放 onMetaData / 序列头）不影响。热路径只多几次 usize 比较与一次 VecDeque
push，仍不 await、不返回错误；无订阅者时同样保留（第一个订阅者也要立刻有缓冲），
每路多占约 6 s 的码率（1080p 2–5 MB）。

`with_snapshot_window(Duration::ZERO)` 给只关心分块判定的测试用。

Refs #1712 #1713

* fix(preview): 客户端掉队时从最近关键帧原地续播，不再断开重连

广播缓冲被覆盖（Lagged）时原来直接结束响应，浏览器看到「连接已断开，2 秒后重连」
的加载态，再花一次 HTTP 往返 + 等下一个关键帧才恢复；带宽长期偏紧的链路上就是
反复「触发加载」。

现在 hub 多一个 `resubscribe(previous, timeout)`：复用原订阅的连接许可（不占第二
个名额），向写入端再要一份从最近关键帧起的快照与新接收端；响应在同一条连接里续
发这份快照——FLV 去掉文件头（`Subscription::snapshot_after_header`，解码器不能中途
再收到文件头；序列头照给，掉队期间可能换过），fMP4 的 init segment 照发（MSE 接受
中途再来一份）。mpegts.js 的 remuxer 会把批次之间的时间戳跳变收平，画面表现为跳
过一段而不是转圈；一次响应最多 MAX_RESYNCS_PER_RESPONSE = 20 次，再多说明客户端
长期跟不上，结束响应交给播放器。每次重对齐 warn 日志带 skipped 与累计次数。

Refs #1712 #1713

* perf(preview): 客户端追赶积压时把广播缓冲里攒着的分块合并写出

实时转发每收到一个分块就写一个 chunked 帧（FLV 一个 tag 几百字节到几十 KB）。
客户端跟得上时缓冲里本来就只有一个，无从合并；追赶积压（起播快照刚发完、链路
抖动之后）时缓冲里攒着几十上百个，逐个写出是几十上百次 socket 写与 chunked 帧。
`coalesce` 在 `recv()` 到手后用 `try_recv` 把已经在缓冲里的后续分块合并成一个
响应块（上限 COALESCE_BYTES = 64 KB），缓冲空着就原样透传不复制，不引入任何等待；
合并途中撞上 Lagged 时把跳过数交回主循环按掉队处理，不吞掉。

Refs #1712 #1713

* fix(ui): 中转模式的播放器维持约 5 s 缓冲，用倍速而不是跳变追帧

mpegts.js 原来对中转、直连一律 liveBufferLatencyMaxLatency 3 s / MinRemain 0.5 s：
缓冲一超过 3 s 就跳到末尾前 0.5 s，等于把服务端刚给的几秒快照直接扔掉，之后只剩
0.5 s 余量，链路一抖就 waiting；GOP 长于 3 s 的流更是每个 GOP 跳一次再停一次。

按取流方式分两套参数（`Players` 新 prop `transport`，LivePreviewPlayer 按实际 source
传）：
- relay：liveSync 倍速追帧——落后 > 9 s 用 1.1× 追到 5 s 附近，≤ 5 s 恢复 1×；
  只在落后 > 20 s（长时间暂停后）才直接跳到末尾前 5 s。与服务端 6 s 的快照窗口配合，
  播放器起播就有约 5 s 缓冲并一直保持。
- direct：沿用原参数（CDN 自带 GOP 缓存，ForgQi 实测异地直连不卡，不动）。

fMP4 的 MediaSource 播放器（只在中转路径上）同样改成：起播对齐到末尾前 5 s、
落后 > 9 s 用 1.1× 追、> 20 s 才跳；另加空洞跳过——服务端掉队重对齐后分片时间戳
前跳，缓冲区出现空洞，停在空洞前沿时跳到下一段起点，不再原地转圈。

Refs #1712 #1713

* style(preview): cargo fmt 与 clippy（len_zero）

* feat(preview): /live 支持 snapshot_ms，起播快照按播放器要的缓冲深度回溯

新订阅者拿到多深的快照，播放器就有多深的起播缓冲，也就是多长的中转延迟。之前一律给整个 6 s
保留窗口，只适合「流畅」档；低延迟档要 2 s 就只该拿 2 s，而不是拿 6–10 s 再用 1.1 倍速追 40 s。

- PreviewHub::subscribe_with_depth(depth)：从当前 GOP 往前取，取到第一个关键帧到达时刻早于
  「现在 − depth」的 GOP 为止（含），起播缓冲落在 depth 到 depth + 一个 GOP 之间；Some(ZERO)
  只给当前 GOP，None 给整个窗口。Subscription 记住 depth，Lagged 重新对齐沿用。
- GET /v1/streamers/{id}/live?snapshot_ms=2000；不带参数行为不变。「开始直播预览」日志多一个
  snapshot_ms 字段。
- 保留窗口 SNAPSHOT_WINDOW 仍是 6 s，服务端内存不变；热路径只在有订阅请求时多一次
  VecDeque::make_contiguous（历史只有几个 GOP）。
- 单测 +2：深度切分的边界（零 / 跨线 / 大于全部历史 / 空历史）、带深度订阅与重对齐沿用深度。

Refs #1713 #1716

* feat(ui): 中转预览分「低延迟 / 流畅」两档，默认低延迟 2 s，卡了自动升档

ForgQi 在 #1713 下问「中转有办法不延迟 5 s 也不卡顿吗」。主因（整 GOP 成坨下发）修掉后 5 s 是
过度配置：同一条模拟链路（60 ms RTT ± 16 ms、丢包重传停顿、20 Mbps）上四路真实房间扫
1 / 1.5 / 2 / 3 / 5 s 目标缓冲各播 180 s，三路 FLV 从 1 s 起就 0 次 waiting；只有 HLS 来的
fMP4 要盖住一个分片（B 站 1–3 s），3 s 起 0 次、4 s 余量最低约 1–2 s。

- app/lib/live-buffer.ts：缓冲控制统一到一处（原来 mpegts.js 的 liveSync 管 FLV、fMP4 播放器
  自己一套 chase）：起播跳到「末尾 − target」，快照落地那几秒里超了再对齐（慢链路上几 MB 要
  几秒才到齐，每对齐一次多看 2 s、最多 15 s），稳态超 max 倍速 / 超 hard 跳 / 跳过重对齐留下的
  空洞，稳态里缓冲耗尽一次报 onStall。档位：low = FLV 2 s / fMP4·TS 4 s，smooth = 5 s
  （目标与原 0d41140 相同，倍速阈值 9 → 7）。mpegts.js 在中转模式下关掉自己的追帧，直连一行未动。
- /live 请求带 snapshot_ms = 目标缓冲，起播就是目标深度，不再从 6–10 s 追回来。
- 低延迟档在稳态里卡一次 ≥ 1 s 或 60 s 内两次，本次播放升到流畅档重连（更深的快照起播），
  角标「卡顿，已切到流畅档」；换房间或改偏好即复位。
- 弹层标题栏与监视器工具栏各一个「低延迟 / 流畅」开关，记 localStorage biliup.preview.latency，
  默认低延迟；直连模式下不显示。

实测（同一模拟链路，180 s，hub 最新时间戳 → 已渲染）：低延迟档三路 FLV 延迟中位 2.1–2.4 s、
fMP4 2.8–3.7 s，四路 0 次 waiting；流畅档 5.1 s。每 18 s 停 2.5 s 的链路上低延迟档卡两次后
自动升档，之后 130 s 0 次。

Refs #1713 #1716
This commit is contained in:
Noah authored and GitHub committed 2026-09-23 06:20:47 +00:00
1 parent 486676db29
commit 48dc8fd65b
10 files changed
+1145 -130
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

app/lib/live-buffer.ts
+169
View File
@@ -0,0 +1,169 @@
/**
* 中转直播流的缓冲控制：播放器该维持多深的缓冲、落后了怎么追。
*
* 中转流的「缓冲深度」就是用户看到的延迟（服务端收到 → 画面播出），两者只差一个链路传输时间：
* 服务端按实时节奏转发，播放器攒着不放的那几秒既是抵消抖动的余量，也是延迟。
* 所以起播时向服务端要多深的快照（`snapshot_ms`）、稳态维持多深的缓冲，是同一个数。
*
* 两个档位：`low` 低延迟——够吃掉几百毫秒的到达抖动（模拟 60 ms RTT ± 16 ms、丢包重传停顿的链路上
* 1 s 都不卡，2 s 留一倍余量）；`smooth` 流畅——链路差（RTT 高、重传多、带宽勉强够码率）时用，
* 多 3 s 延迟换零 waiting。低延迟档卡了会自动升到流畅档。
* HLS 来的流（fMP4 / TS）另算：服务端是整个分片到一坨，缓冲至少得盖住一个分片加取片抖动
* （B 站 hls_fmp4 分片 1–3 s），低延迟档给 4 s。
* 直连 CDN 的流不走这里（CDN 自带 GOP 缓存，mpegts.js 自己的追帧参数够用）。
*/
export interface LiveBufferPolicy {
/** 目标缓冲 = 中转延迟，秒 */
target: number
/** 落后到这以上用倍速慢慢追回目标（不跳，跳一次会把画面顿一下） */
max: number
/** 落后到这以上（长时间暂停、后台标签页）直接跳到目标位置 */
hard: number
/** 追帧倍速 */
rate: number
}
export type LatencyProfile = 'low' | 'smooth'
/** 按 tag 到达的流（FLV）：低延迟 2 s / 流畅 5 s */
export const RELAY_PROFILES: Record<LatencyProfile, LiveBufferPolicy> = {
low: { target: 2, max: 3.5, hard: 12, rate: 1.1 },
smooth: { target: 5, max: 7, hard: 20, rate: 1.1 },
}
/** 按 HLS 分片到达的流（fMP4 / TS）：低延迟也得盖住一个分片 */
export const RELAY_PROFILES_SEGMENTED: Record<LatencyProfile, LiveBufferPolicy> = {
low: { target: 4, max: 6, hard: 14, rate: 1.1 },
smooth: RELAY_PROFILES.smooth,
}
/** 该档位在该容器下的缓冲策略；容器未定按 FLV */
export function relayPolicy(profile: LatencyProfile, format: string | null | undefined): LiveBufferPolicy {
return format === 'fmp4' || format === 'mpegts' ? RELAY_PROFILES_SEGMENTED[profile] : RELAY_PROFILES[profile]
}
/** 播放器要维持 target 秒缓冲，起播快照就要 target 秒（服务端给到 target 到 target + 一个 GOP） */
export function snapshotMsFor(policy: LiveBufferPolicy): number {
return Math.round(policy.target * 1000)
}
/**
* 起播对齐后再看几秒：快照是一坨突发到达的，落地要一两秒（几 MB 在慢链路上更久），这期间缓冲还在长，
* 超了就再对齐一次；每对齐一次再多看 2 s，最多看 15 s，之后只做倍速 / 跳的稳态控制
*/
const SETTLE_MS = 3000
const SETTLE_EXTEND_MS = 2000
const SETTLE_MAX_MS = 15000
/** 缓冲比目标多出这么多才值得再对齐一次（起播阶段） */
const ALIGN_SLACK_S = 1
const TICK_MS = 250
export interface StallInfo {
/** 卡住了多久，秒 */
seconds: number
/** 卡住那一刻的缓冲余量，秒（真正的缓冲耗尽接近 0） */
bufferAhead: number
}
export interface LiveBufferHooks {
/** 稳态期间（起播对齐并稳定之后）一次 waiting → playing，告诉上层卡了多久 */
onStall?: (info: StallInfo) => void
}
function liveEdge(video: HTMLVideoElement): { start: number; end: number } | null {
const b = video.buffered
if (!b.length) return null
return { start: b.start(b.length - 1), end: b.end(b.length - 1) }
}
/**
* 把缓冲控制挂到 `<video>` 上：起播把播放位置对齐到「末尾 − target」，之后每 250 ms 看一次落后量，
* 超 `max` 倍速、超 `hard` 跳、掉队重对齐留下的缓冲空洞直接跳过。返回卸载函数。
* mpegts.js 的 `liveSync` / `liveBufferLatencyChasing` 与 fMP4 的 MediaSource 播放器都改用这一套，
* 两种容器的中转流行为一致，档位也能在播放中途改（`getPolicy` 每次 tick 现取）。
*/
export function attachLiveBufferControl(
video: HTMLVideoElement,
getPolicy: () => LiveBufferPolicy,
hooks: LiveBufferHooks = {}
): () => void {
let disposed = false
let aligned = false
let alignedAt = 0
let settleUntil = 0
let lastAlignAt = 0
let stallStartedAt: number | null = null
let stallAhead = 0
const seekTo = (t: number) => {
lastAlignAt = performance.now()
video.currentTime = t
}
const tick = () => {
if (disposed) return
const edge = liveEdge(video)
if (!edge) return
const policy = getPolicy()
const now = performance.now()
const latency = edge.end - video.currentTime
if (!aligned) {
// 起播：跳到「末尾 − target」（fMP4 的时间戳从一个很大的值起步，不跳进缓冲区根本放不起来）；
// 快照还没到齐就先从有的地方放，接下来几秒里缓冲长过头了再对齐
aligned = true
alignedAt = now
settleUntil = now + SETTLE_MS
seekTo(Math.max(edge.start, edge.end - policy.target))
return
}
if (now < settleUntil) {
if (latency > policy.target + ALIGN_SLACK_S && now - lastAlignAt >= 500) {
seekTo(edge.end - policy.target)
settleUntil = Math.min(alignedAt + SETTLE_MAX_MS, now + SETTLE_EXTEND_MS)
}
return
}
// 掉队重对齐后时间戳前跳，缓冲区出现空洞：停在空洞前沿就跳到下一段的起点
const b = video.buffered
for (let i = 0; i + 1 < b.length; i++) {
if (video.currentTime >= b.end(i) - 0.3 && video.currentTime < b.start(i + 1) && video.readyState < 3) {
seekTo(b.start(i + 1) + 0.05)
return
}
}
if (video.paused) return
if (latency > policy.hard) {
seekTo(edge.end - policy.target)
if (video.playbackRate !== 1) video.playbackRate = 1
} else if (latency > policy.max) {
if (video.playbackRate !== policy.rate) video.playbackRate = policy.rate
} else if (latency <= policy.target && video.playbackRate !== 1) {
video.playbackRate = 1
}
}
const onWaiting = () => {
// 只统计稳态期间真正的缓冲耗尽；对齐 / 跳空洞引起的 seeking 不算
if (disposed || !aligned || performance.now() < settleUntil || video.seeking) return
const edge = liveEdge(video)
stallStartedAt = performance.now()
stallAhead = edge ? Math.max(0, edge.end - video.currentTime) : 0
}
const onPlaying = () => {
if (stallStartedAt == null) return
const seconds = (performance.now() - stallStartedAt) / 1000
stallStartedAt = null
hooks.onStall?.({ seconds, bufferAhead: stallAhead })
}
video.addEventListener('waiting', onWaiting)
video.addEventListener('playing', onPlaying)
const timer = setInterval(tick, TICK_MS)
return () => {
disposed = true
clearInterval(timer)
video.removeEventListener('waiting', onWaiting)
video.removeEventListener('playing', onPlaying)
}
}
app/lib/use-dashboard.ts
+7 -3
View File
@@ -112,9 +112,13 @@ export function liveImageUrl(
return `${API_BASE}/v1/streamers/${id}/${kind}?v=${(h >>> 0).toString(36)}`
}
/** 正在录制的那一路流的同源地址（chunked FLV / MPEG-TS / fMP4），供页面内播放器直接拉取。 */
export function livePreviewUrl(id: number): string {
return `${API_BASE}/v1/streamers/${id}/live`
/**
* 正在录制的那一路流的同源地址（chunked FLV / MPEG-TS / fMP4），供页面内播放器直接拉取。
* `snapshotMs`：起播快照回溯多少毫秒的已完成 GOP（播放器要维持多深的缓冲就要多深）；不传给服务端的整个保留窗口。
*/
export function livePreviewUrl(id: number, snapshotMs?: number): string {
const base = `${API_BASE}/v1/streamers/${id}/live`
return snapshotMs === undefined ? base : `${base}?snapshot_ms=${Math.max(0, Math.round(snapshotMs))}`
}
/** 浏览器直连模式：向后端要当前录制中那条流的 CDN 直链。 */
app/lib/use-local-pref.ts
+13
View File
@@ -52,6 +52,19 @@ export function useBoolPref(key: string, fallback: boolean): [boolean, (v: boole
return [value, (v: boolean) => writePref(key, v ? '1' : '0')]
}
/** 只接受给定候选值的字符串偏好（档位一类的枚举）。 */
export function useEnumPref<T extends string>(key: string, options: readonly T[], fallback: T): [T, (v: T) => void] {
const value = useSyncExternalStore(
subscribe,
() => {
const raw = readPref(key)
return options.includes(raw as T) ? (raw as T) : fallback
},
() => fallback
)
return [value, (v: T) => writePref(key, v)]
}
/** 只接受给定候选值的数字偏好。 */
export function useChoicePref(key: string, options: number[], fallback: number): [number, (v: number) => void] {
const value = useSyncExternalStore(
app/ui/LiveMonitor.tsx
+27 -1
View File
@@ -13,10 +13,12 @@ import {
previewFormatLabel,
SLOW_REFRESH_MS,
STREAMERS_REFRESH_MS,
usePreviewTransport,
} from '@/app/lib/use-dashboard'
import { useBoolPref, useChoicePref } from '@/app/lib/use-local-pref'
import { useDanmakuFeed } from '@/app/lib/danmaku-feed'
import { LivePreviewPlayer } from './LivePreview'
import { LivePreviewPlayer, useLatencyProfile } from './LivePreview'
import { type LatencyProfile, RELAY_PROFILES, RELAY_PROFILES_SEGMENTED } from '@/app/lib/live-buffer'
import { LiveRateChart, SPARK_RATE_WINDOW_MS } from './LiveRateChart'
import styles from './live-monitor.module.scss'
@@ -91,6 +93,8 @@ export default function LiveMonitor() {
const options = tileOptions()
const [maxTiles, setMaxTiles] = useChoicePref(TILES_KEY, options, DEFAULT_MONITOR_TILES)
const [danmaku, setDanmaku] = useBoolPref(DANMAKU_KEY, false)
const [latency, setLatency] = useLatencyProfile()
const transport = usePreviewTransport()
const [rateChart, setRateChart] = useBoolPref(RATE_CHART_KEY, false)
const [selection, setSelection] = useState<Selection>({ order: [], stopped: [] })
@@ -166,6 +170,28 @@ export default function LiveMonitor() {
<Switch size="small" checked={danmaku} onChange={(v) => setDanmaku(!!v)} aria-label="弹幕" />
</span>
</Tooltip>
{transport === 'relay' ? (
<>
<Tooltip
content={`中转缓冲深度，也就是画面延迟。低延迟：FLV 约 ${RELAY_PROFILES.low.target} s、HLS 分片流约 ${RELAY_PROFILES_SEGMENTED.low.target} s，卡了的小窗自动切到流畅；流畅：约 ${RELAY_PROFILES.smooth.target} s`}
>
<Text type="tertiary" size="small">
延迟
</Text>
</Tooltip>
<Select
size="small"
value={latency}
onChange={(v) => setLatency(v as LatencyProfile)}
optionList={[
{ value: 'low', label: '低延迟' },
{ value: 'smooth', label: '流畅' },
]}
style={{ width: 96 }}
aria-label="中转延迟"
/>
</>
) : null}
<Tooltip content="每路小窗下方显示最近 60 秒的写盘速率折线（每秒向后端拉一次速率，所有小窗共用一次请求）">
<span className={styles.switchRow}>
<Text type="tertiary" size="small">
app/ui/LivePreview.tsx
+80 -6
View File
@@ -1,7 +1,7 @@
'use client'
import React, { useCallback, useEffect, useMemo, useRef, useState } from 'react'
import dynamic from 'next/dynamic'
import { Button, Modal, Switch, Tag, Toast, Tooltip, Typography } from '@douyinfe/semi-ui'
import { Button, Modal, Radio, RadioGroup, Switch, Tag, Toast, Tooltip, Typography } from '@douyinfe/semi-ui'
import { IconChevronDown, IconPlay, IconRefresh } from '@douyinfe/semi-icons'
import type { ButtonProps } from '@douyinfe/semi-ui/lib/es/button'
import { LiveStreamerEntity } from '@/app/lib/api-streamer'
@@ -17,8 +17,17 @@ import {
previewFormatLabel,
usePreviewTransport,
} from '@/app/lib/use-dashboard'
import { useBoolPref } from '@/app/lib/use-local-pref'
import { useBoolPref, useEnumPref } from '@/app/lib/use-local-pref'
import { type DanmakuFeed, useDanmakuFeed } from '@/app/lib/danmaku-feed'
import {
type LatencyProfile,
type LiveBufferPolicy,
RELAY_PROFILES,
RELAY_PROFILES_SEGMENTED,
relayPolicy,
snapshotMsFor,
type StallInfo,
} from '@/app/lib/live-buffer'
import { LiveRateChart, LiveRateSummary, MODAL_RATE_WINDOW_MS } from './LiveRateChart'
import styles from './live-preview.module.scss'
@@ -29,6 +38,17 @@ const MAX_AUTO_RECONNECT = 3
const RECONNECT_DELAY_MS = 2000
/** 弹层弹幕开关记在本地，默认开；监视器另有自己的开关（默认关） */
const MODAL_DANMAKU_KEY = 'biliup.preview.danmaku'
/** 中转延迟档位（弹层与监视器共用），默认低延迟；卡顿后本次播放自动升到流畅 */
const LATENCY_PROFILE_KEY = 'biliup.preview.latency'
const LATENCY_PROFILES: readonly LatencyProfile[] = ['low', 'smooth']
/** 低延迟档下，稳态里一次卡住这么久、或 60 s 内卡两次，就升到流畅档重连 */
const ESCALATE_STALL_S = 1
const ESCALATE_WINDOW_MS = 60_000
/** 中转延迟档位偏好 */
export function useLatencyProfile(): [LatencyProfile, (v: LatencyProfile) => void] {
return useEnumPref(LATENCY_PROFILE_KEY, LATENCY_PROFILES, 'low')
}
/** 弹层底部的码率折线默认展开，折叠状态记在本地 */
const MODAL_RATE_CHART_KEY = 'biliup.preview.rateChart'
@@ -118,9 +138,32 @@ export function LivePreviewPlayer({
return fetched?.key === sourceKey ? fetched.source : null
}, [override, fetched, sourceKey, wantDirect, transport, directReason])
// 中转延迟档位：用户偏好起步；低延迟档在稳态里卡了，本次播放（同房间、同偏好）升到流畅档重连，
// 用更深的快照起播。升档记 key，房间或偏好一变自然作废，不需要 effect 里重置
const relayFormat = streamer.preview?.format ?? undefined
const [profilePref] = useLatencyProfile()
const levelKey = `${streamer.id}:${profilePref}`
const [escalated, setEscalated] = useState<string | null>(null)
const level: LatencyProfile = escalated === levelKey ? 'smooth' : profilePref
const policy: LiveBufferPolicy = relayPolicy(level, relayFormat)
const stallsRef = useRef<number[]>([])
const handleStall = useCallback(
(info: StallInfo) => {
if (level !== 'low') return
const now = Date.now()
stallsRef.current = [...stallsRef.current.filter((t) => now - t < ESCALATE_WINDOW_MS), now]
if (info.seconds < ESCALATE_STALL_S && stallsRef.current.length < 2) return
stallsRef.current = []
setEscalated(levelKey)
setPhase('connecting')
setMessage(null)
setNonce((n) => n + 1)
},
[level, levelKey]
)
const codecs = streamer.preview?.codecs ?? null
const relayUrl = livePreviewUrl(streamer.id)
const relayUrl = livePreviewUrl(streamer.id, snapshotMsFor(policy))
const url = source?.kind === 'direct' ? source.url : relayUrl
const format = source?.kind === 'direct' ? (source.format ?? 'flv') : relayFormat
// 平台有弹幕客户端才装弹幕层；开关只控制订阅与显示
@@ -245,9 +288,16 @@ export function LivePreviewPlayer({
? { text: '直连 CDN', tone: 'direct' as const }
: source?.kind === 'relay' && source.fallbackReason
? { text: `已回落中转：${source.fallbackReason}`, tone: 'fallback' as const }
: null
: source?.kind === 'relay' && escalated === levelKey
? { text: '卡顿，已切到流畅档', tone: 'fallback' as const }
: null
return (
<div className={styles.player} data-phase={phase} data-source={source?.kind ?? 'pending'}>
<div
className={styles.player}
data-phase={phase}
data-source={source?.kind ?? 'pending'}
data-latency={source?.kind === 'relay' ? level : undefined}
>
{showPlayer ? (
<Players
key={`${source?.kind}-${url}-${nonce}`}
@@ -255,10 +305,13 @@ export function LivePreviewPlayer({
type={format}
codecs={codecs}
isLive
transport={source?.kind ?? 'relay'}
buffer={policy}
muted={muted}
autoplay
onEnded={handleEnded}
onError={handleError}
onStall={handleStall}
danmaku={danmakuLayer}
/>
) : null}
@@ -267,7 +320,9 @@ export function LivePreviewPlayer({
content={
badge.tone === 'direct'
? '浏览器用另取的直链直接向 CDN 拉流，不经 biliup 中转、不影响录制（全局配置 preview_transport = direct）'
: badge.text
: escalated === levelKey && source?.kind === 'relay' && !source.fallbackReason
? `低延迟档缓冲耗尽过，本次播放改用约 ${RELAY_PROFILES.smooth.target} s 缓冲；重新打开恢复低延迟`
: badge.text
}
>
<span className={styles.badge} data-tone={badge.tone} data-compact={compact || undefined}>
@@ -335,6 +390,7 @@ export function LivePreviewModal({
const danmakuFeed = useDanmakuFeed([streamer.id], danmakuOn)
const [chartOpen, setChartOpen] = useBoolPref(MODAL_RATE_CHART_KEY, true)
const transport = usePreviewTransport()
const [latency, setLatency] = useLatencyProfile()
const notifiedRef = useRef(false)
useEffect(() => {
if (!visible) notifiedRef.current = false
@@ -393,6 +449,24 @@ export function LivePreviewModal({
/>
</span>
</Tooltip>
{transport === 'relay' ? (
<Tooltip
content={`中转缓冲深度，也就是画面延迟。低延迟：FLV 约 ${RELAY_PROFILES.low.target} s、HLS 分片流（fMP4 / TS）约 ${RELAY_PROFILES_SEGMENTED.low.target} s，链路抖动大时可能偶发缓冲，卡了会自动切到流畅；流畅：约 ${RELAY_PROFILES.smooth.target} s。直连 CDN 时不适用`}
>
<span className={styles.danmakuSwitch}>
<RadioGroup
type="button"
buttonSize="small"
value={latency}
onChange={(e) => setLatency(e.target.value as LatencyProfile)}
aria-label="中转延迟"
>
<Radio value="low">低延迟</Radio>
<Radio value="smooth">流畅</Radio>
</RadioGroup>
</span>
</Tooltip>
) : null}
</div>
}
>
app/ui/Player.tsx
+96 -46
View File
@@ -3,6 +3,7 @@ import Artplayer from 'artplayer'
import artplayerPluginDanmuku from 'artplayer-plugin-danmuku'
import mpegts from 'mpegts.js'
import type { DanmakuFeed, DanmakuFrame } from '@/app/lib/danmaku-feed'
import { attachLiveBufferControl, type LiveBufferPolicy, RELAY_PROFILES, type StallInfo } from '@/app/lib/live-buffer'
type VideoPlayer = Artplayer | null
@@ -28,12 +29,21 @@ interface PlayerConfig {
codecs?: string | null
/** 直播：隐藏进度条、开启追帧，播放器随分块到达持续解码 */
isLive?: boolean
/**
* 直播的取流方式：`relay`（经 biliup 中转，缓冲深度由 `buffer` 决定，见 `live-buffer.ts`）
* 或 `direct`（浏览器直连 CDN，沿用 mpegts.js 的低延迟追帧）。不传按中转处理。
*/
transport?: LiveTransport
/** 中转流的缓冲策略；不传用「流畅」档。播放中途换档不重建播放器，下一次 tick 生效 */
buffer?: LiveBufferPolicy
muted?: boolean
autoplay?: boolean
/** 直播流结束 / 被服务端断开（如客户端掉队、录制换直链）时回调，调用方决定是否重连 */
onEnded?: () => void
/** mpegts.js 或探测阶段报错（网络 / 解码），附一句可展示的说明 */
onError?: (message: string) => void
/** 中转流稳态期间缓冲耗尽一次（waiting → playing），调用方据此决定要不要换更深的档 */
onStall?: (info: StallInfo) => void
/**
* 实时弹幕：`feed` 是页面持有的共享连接（见 `DanmakuFeed`），`id` 是本路直播间；
* `feed` 为 null 时弹幕层隐藏。传了这个 prop 就装 artplayer-plugin-danmuku，开关切换只
@@ -196,17 +206,25 @@ export function canPlayFmp4(codecs: string | null | undefined): boolean {
}
}
/** 直播追帧：缓冲末尾落后超过这么多秒就跳到末尾附近 */
const FMP4_MAX_LATENCY_S = 3
const FMP4_TARGET_REMAIN_S = 0.5
/** 直播预览的取流方式：决定播放器维持多深的缓冲 */
export type LiveTransport = 'relay' | 'direct'
/**
* 直连 CDN 的流沿用 mpegts.js 自己的追帧参数：落后超过 3 s 就跳到末尾前 0.5 s
* （CDN 自带 GOP 缓存，ForgQi 实测异地直连不卡）。中转流的缓冲策略在 `live-buffer.ts`。
*/
const DIRECT_MAX_LATENCY_S = 3
const DIRECT_TARGET_REMAIN_S = 0.5
/** 早于当前播放位置这么多秒的缓冲定期清掉，长时间观看不涨内存 */
const FMP4_KEEP_BACKWARD_S = 30
interface Fmp4Options {
codecs: string | null | undefined
autoplay: boolean
buffer: () => LiveBufferPolicy
onEnded?: () => void
onError?: (message: string) => void
onStall?: (info: StallInfo) => void
}
interface HlsOptions {
@@ -291,13 +309,13 @@ async function playWithHls(video: HTMLVideoElement, url: string, art: Artplayer,
* Artplayer customType：fMP4 直接交给 MediaSource。
* 服务端先发 init segment（ftyp + moov）再按 moof + mdat 分片广播，正是 MSE 的原生输入，不需要转封装：
* fetch 的 ReadableStream 按到达顺序进 appendBuffer 队列（`updateend` 串行），
* 首个缓冲区出现时把播放位置对齐到缓冲起点，之后落后超过阈值就追到末尾，并定期 remove 旧缓冲。
* 播放位置的对齐与追帧交给 `attachLiveBufferControl`（与 FLV 中转流同一套），并定期 remove 旧缓冲。
*/
function playWithMediaSource(
video: HTMLVideoElement,
url: string,
art: Artplayer,
{ codecs, autoplay, onEnded, onError }: Fmp4Options
{ codecs, autoplay, buffer, onEnded, onError, onStall }: Fmp4Options
) {
// 同 playWithMpegts：Artplayer 出错后的自动重连会在实例销毁后仍重设 url，忽略
if (art.isDestroy) return
@@ -321,8 +339,7 @@ function playWithMediaSource(
let sourceBuffer: SourceBuffer | null = null
let streamEnded = false
let disposed = false
let aligned = false
let chaser: ReturnType<typeof setInterval> | null = null
let detachBufferControl: (() => void) | null = null
const fail = (message: string) => {
if (disposed) return
@@ -359,29 +376,13 @@ function playWithMediaSource(
const cut = video.currentTime - FMP4_KEEP_BACKWARD_S
if (cut - start > 5) sourceBuffer.remove(start, cut)
}
const chase = () => {
if (disposed || !video.buffered.length) return
const end = video.buffered.end(video.buffered.length - 1)
if (!aligned) {
// 直播分片的时间戳从一个很大的值起步，播放位置得先跳进缓冲区
aligned = true
video.currentTime = Math.max(video.buffered.start(0), end - FMP4_TARGET_REMAIN_S)
if (autoplay) video.play().catch(() => {
video.muted = true
video.play().catch(() => {})
})
return
}
if (end - video.currentTime > FMP4_MAX_LATENCY_S && !video.paused) {
video.currentTime = end - FMP4_TARGET_REMAIN_S
}
}
const dispose = () => {
if (disposed) return
disposed = true
controller.abort()
if (chaser) clearInterval(chaser)
detachBufferControl?.()
detachBufferControl = null
try {
if (mediaSource.readyState === 'open') mediaSource.endOfStream()
} catch {
@@ -406,7 +407,16 @@ function playWithMediaSource(
pump()
})
sourceBuffer.addEventListener('error', () => fail('MSE 解码失败'))
chaser = setInterval(chase, 500)
// fMP4 只在中转路径上出现（直连的 HLS 走 hls.js），缓冲策略按中转流来；
// 直播分片的时间戳从一个很大的值起步，控制器的首次对齐会把播放位置跳进缓冲区，之后才真正播起来
detachBufferControl = attachLiveBufferControl(video, buffer, { onStall })
if (autoplay) {
video.play().catch(() => {
if (disposed) return
video.muted = true
video.play().catch(() => {})
})
}
fetch(url, { signal: controller.signal, cache: 'no-store' })
.then(async (res) => {
@@ -463,9 +473,36 @@ function describeMpegtsError(errorType: string, detail: string, info?: { code?:
interface MpegtsOptions {
type: MpegtsType
isLive: boolean
transport: LiveTransport
buffer: () => LiveBufferPolicy
autoplay: boolean
onEnded?: () => void
onError?: (message: string) => void
onStall?: (info: StallInfo) => void
}
/**
* mpegts.js 的直播缓冲参数。
* 中转：mpegts.js 自己的追帧 / liveSync 关掉，缓冲深度由 `attachLiveBufferControl` 按档位维持
* （跟 fMP4 中转流同一套，档位能在播放中途改）；直连：沿用原来的追帧参数。
*/
export function mpegtsLiveConfig(transport: LiveTransport): mpegts.Config {
const common = {
enableStashBuffer: false,
customLoader: AbortableFetchLoader,
autoCleanupSourceBuffer: true,
autoCleanupMaxBackwardDuration: 30,
autoCleanupMinBackwardDuration: 10,
}
if (transport === 'relay') {
return { ...common, liveBufferLatencyChasing: false, liveSync: false }
}
return {
...common,
liveBufferLatencyChasing: true,
liveBufferLatencyMaxLatency: DIRECT_MAX_LATENCY_S,
liveBufferLatencyMinRemain: DIRECT_TARGET_REMAIN_S,
}
}
/** Artplayer customType：用 mpegts.js 解封装 FLV / MPEG-TS 后喂给 MSE。 */
@@ -473,7 +510,7 @@ function playWithMpegts(
video: HTMLVideoElement,
url: string,
art: Artplayer,
{ type, isLive, autoplay, onEnded, onError }: MpegtsOptions
{ type, isLive, transport, buffer, autoplay, onEnded, onError, onStall }: MpegtsOptions
) {
if (!mpegts.isSupported()) {
art.notice.show = `当前浏览器不支持 MSE，无法播放 ${type}`
@@ -484,32 +521,28 @@ function playWithMpegts(
// 而且不看自己是否已被 destroy——那会在弹层关闭后凭空再拉一路流、占着服务端许可。
// 重连由 LivePreviewPlayer 自己管，这里对已销毁的实例直接不理。
if (art.isDestroy) return
const artWithMpegts = art as Artplayer & { mpegts?: mpegts.Player | null }
const artWithMpegts = art as Artplayer & { mpegts?: mpegts.Player | null; detachBuffer?: (() => void) | null }
if (artWithMpegts.mpegts) {
destroyMpegts(artWithMpegts.mpegts)
artWithMpegts.mpegts = null
}
artWithMpegts.detachBuffer?.()
artWithMpegts.detachBuffer = null
const player = mpegts.createPlayer(
// 直连 CDN 时是跨域请求：cors 模式、不带 cookie（这些 CDN 的 ACAO 是 *，带凭据反而会被拒）；
// 同源的 /live 走 same-origin 凭据，登录 cookie 照常带上（见 AbortableFetchLoader）
{ type, url, isLive, cors: true, withCredentials: false },
isLive
? {
// 直播：不攒缓冲、落后就追，源缓冲区用完即清理，长时间观看不涨内存
enableStashBuffer: false,
customLoader: AbortableFetchLoader,
liveBufferLatencyChasing: true,
liveBufferLatencyMaxLatency: 3,
liveBufferLatencyMinRemain: 0.5,
autoCleanupSourceBuffer: true,
autoCleanupMaxBackwardDuration: 30,
autoCleanupMinBackwardDuration: 10,
}
: {}
// 直播：不攒 IO 缓冲、源缓冲区用完即清理（长时间观看不涨内存），缓冲深度按取流方式定
isLive ? mpegtsLiveConfig(transport) : {}
)
artWithMpegts.mpegts = player
if (isLive && transport === 'relay') {
artWithMpegts.detachBuffer = attachLiveBufferControl(video, buffer, { onStall })
}
art.on('destroy', () => {
artWithMpegts.detachBuffer?.()
artWithMpegts.detachBuffer = null
if (artWithMpegts.mpegts) {
destroyMpegts(artWithMpegts.mpegts)
artWithMpegts.mpegts = null
@@ -548,10 +581,13 @@ const Players: React.FC<PlayerConfig> = ({
type,
codecs,
isLive = false,
transport = 'relay',
buffer,
muted = false,
autoplay = isLive,
onEnded,
onError,
onStall,
danmaku,
}) => {
const containerRef = useRef<HTMLDivElement>(null)
@@ -560,10 +596,12 @@ const Players: React.FC<PlayerConfig> = ({
const danmakuId = danmaku?.id
const danmakuFeed = danmaku?.feed ?? null
const danmakuFontSize = danmaku?.fontSize ?? 22
// 回调放进 ref：父组件每次渲染传入的新函数不应重建播放器
const callbacksRef = useRef({ onEnded, onError })
// 回调与缓冲策略放进 ref：父组件每次渲染传入的新函数 / 中途换档都不应重建播放器
const callbacksRef = useRef({ onEnded, onError, onStall })
const bufferRef = useRef<LiveBufferPolicy>(buffer ?? RELAY_PROFILES.smooth)
useEffect(() => {
callbacksRef.current = { onEnded, onError }
callbacksRef.current = { onEnded, onError, onStall }
bufferRef.current = buffer ?? RELAY_PROFILES.smooth
})
useEffect(() => {
@@ -605,6 +643,8 @@ const Players: React.FC<PlayerConfig> = ({
}
const onEndedCb = () => callbacksRef.current.onEnded?.()
const onErrorCb = (message: string) => callbacksRef.current.onError?.(message)
const onStallCb = (info: StallInfo) => callbacksRef.current.onStall?.(info)
const bufferPolicy = () => bufferRef.current
try {
if (mediaType === 'hls') {
const options: HlsOptions = { autoplay, onEnded: onEndedCb, onError: onErrorCb }
@@ -620,7 +660,14 @@ const Players: React.FC<PlayerConfig> = ({
},
})
} else if (mediaType === 'fmp4') {
const options: Fmp4Options = { codecs, autoplay, onEnded: onEndedCb, onError: onErrorCb }
const options: Fmp4Options = {
codecs,
autoplay,
buffer: bufferPolicy,
onEnded: onEndedCb,
onError: onErrorCb,
onStall: onStallCb,
}
playerRef.current = new Artplayer({
...base,
type: 'fmp4',
@@ -633,9 +680,12 @@ const Players: React.FC<PlayerConfig> = ({
const options: MpegtsOptions = {
type: mediaType,
isLive,
transport,
buffer: bufferPolicy,
autoplay,
onEnded: onEndedCb,
onError: onErrorCb,
onStall: onStallCb,
}
playerRef.current = new Artplayer({
...base,
@@ -671,7 +721,7 @@ const Players: React.FC<PlayerConfig> = ({
playerRef.current = null
}
}
}, [url, height, width, type, codecs, isLive, muted, autoplay, danmakuCapable, danmakuFontSize])
}, [url, height, width, type, codecs, isLive, transport, muted, autoplay, danmakuCapable, danmakuFontSize])
// 弹幕开关：有 feed → 订阅本路、显示弹幕层；没有 → 退订、隐藏。不重建播放器。
// 播放器可能还在探测 Content-Type（异步创建），所以轮询等到实例出现再挂。
crates/biliup-cli/src/server/api/live_preview.rs
+229 -30
View File
@@ -1,8 +1,9 @@
//! `GET /v1/streamers/{id}/live`：把正在录制的那一路流旁路给页面内的播放器。
//!
//! 响应是 chunked 的 `video/x-flv` 或 `video/mp2t` 字节流：先发「文件头 + 序列头 + 当前 GOP」
//! 的快照，再持续转发与写盘同步的实时分块。慢客户端落后到广播缓冲被覆盖（`Lagged`）时
//! 直接结束响应，由播放器重连拿新快照；写入端换代（断流重试 / 换直链）时同样结束响应。
//! 响应是 chunked 的 `video/x-flv` / `video/mp2t` / `video/mp4` 字节流：先发「文件头 + 序列头 +
//! 最近几秒的 GOP」的快照，再持续转发与写盘同步的实时分块。慢客户端落后到广播缓冲被覆盖
//! （`Lagged`）时不断开，而是在同一条响应里从最近的关键帧重新对齐（再要一份不含文件头的快照）；
//! 写入端换代（断流重试 / 换直链）时结束响应，由播放器重连。
//!
//! 连接数：每路 [`crate::server::common::download::PREVIEW_MAX_SUBSCRIBERS_PER_ROOM`]、
//! 进程 [`MAX_PREVIEW_CONNECTIONS`]，超出返回 429。
@@ -18,14 +19,14 @@ use axum::http::{HeaderValue, StatusCode, header};
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::response::{IntoResponse, Response};
use biliup::downloader::live::LiveStatus;
use biliup::downloader::preview::{PreviewHub, SubscribeError, Subscription};
use bytes::Bytes;
use biliup::downloader::preview::{PreviewFormat, PreviewHub, SubscribeError, Subscription};
use bytes::{Bytes, BytesMut};
use danmaku_client::DanmakuEvent;
use std::collections::VecDeque;
use std::sync::{Arc, OnceLock};
use std::time::Duration;
use tokio::sync::broadcast::error::RecvError;
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use tokio::sync::{OwnedSemaphorePermit, Semaphore, broadcast};
use tracing::{debug, info, warn};
/// 进程内同时允许的预览连接总数（所有直播间合计）。
@@ -38,11 +39,20 @@ fn connection_limit() -> &'static Arc<Semaphore> {
LIMIT.get_or_init(|| Arc::new(Semaphore::new(MAX_PREVIEW_CONNECTIONS)))
}
/// `GET /v1/streamers/{id}/live`
#[derive(Debug, Default, serde::Deserialize)]
pub struct LiveQuery {
/// 起播快照回溯多少毫秒的已完成 GOP（见 `PreviewHub::subscribe_with_depth`）：
/// 播放器按自己要维持的缓冲深度来要；不带则给整个保留窗口（`SNAPSHOT_WINDOW`）
pub snapshot_ms: Option<u64>,
}
/// `GET /v1/streamers/{id}/live?snapshot_ms=2000`
pub async fn get_live_stream(
State(managers): State<Arc<DownloadManager>>,
Path(id): Path<i64>,
Query(query): Query<LiveQuery>,
) -> Response {
let depth = query.snapshot_ms.map(Duration::from_millis);
let Some(worker) = managers.get_room_by_id(id).await else {
return (StatusCode::NOT_FOUND, "直播间不存在").into_response();
};
@@ -58,10 +68,17 @@ pub async fn get_live_stream(
)
.into_response();
};
match hub.subscribe(SUBSCRIBE_TIMEOUT).await {
match hub.subscribe_with_depth(depth, SUBSCRIBE_TIMEOUT).await {
Ok(subscription) => {
info!(id, format = subscription.format.as_str(), "开始直播预览");
live_response(subscription, global)
info!(
id,
format = subscription.format.as_str(),
snapshot_ms = depth.map(|d| d.as_millis() as u64),
snapshot_chunks = subscription.snapshot.len(),
snapshot_bytes = subscription.snapshot.iter().map(Bytes::len).sum::<usize>(),
"开始直播预览"
);
live_response(id, hub, subscription, global)
}
Err(error) => subscribe_error_response(id, error),
}
@@ -455,43 +472,130 @@ impl<S: futures::Stream + Unpin> futures::Stream for HoldPermit<S> {
}
}
/// 一次响应里最多容忍多少次掉队重对齐；再多说明客户端带宽长期跟不上，结束响应交给播放器
/// （它会显示原因并按自己的策略重连）。
pub const MAX_RESYNCS_PER_RESPONSE: u32 = 20;
/// 实时转发时把已经攒在广播缓冲里的多个分块合并成一个响应块的上限：客户端追赶积压时
/// 少写几次 socket / 少几个 chunked 帧；没有积压时一个分块就是一个响应块，不加任何延迟。
pub const COALESCE_BYTES: usize = 64 * 1024;
struct LiveBody {
id: i64,
hub: PreviewHub,
snapshot: VecDeque<Bytes>,
/// 持有该路的连接许可；`rx` 是实时接收端
subscription: Subscription,
/// 持有该路的连接许可；`rx` 是实时接收端。掉队重对齐时被 take 走再换新的
subscription: Option<Subscription>,
/// 合并分块时 `try_recv` 撞上的掉队，留到下一轮按掉队处理
pending_lag: Option<u64>,
resyncs: u32,
_global: OwnedSemaphorePermit,
}
/// 实时分块到手后，把广播缓冲里已经攒着的后续分块（客户端在追赶积压时才会有）
/// 一起合并成一个响应块，最多 [`COALESCE_BYTES`]；缓冲空着就原样返回、不复制。
/// 途中 `try_recv` 报告掉队时停止合并，把掉队数交回调用方处理。
fn coalesce(rx: &mut broadcast::Receiver<Bytes>, first: Bytes) -> (Bytes, Option<u64>) {
use tokio::sync::broadcast::error::TryRecvError;
if first.len() >= COALESCE_BYTES || rx.is_empty() {
return (first, None);
}
let mut out = BytesMut::with_capacity(COALESCE_BYTES);
out.extend_from_slice(&first);
loop {
match rx.try_recv() {
Ok(chunk) => {
out.extend_from_slice(&chunk);
if out.len() >= COALESCE_BYTES {
return (out.freeze(), None);
}
}
Err(TryRecvError::Lagged(skipped)) => return (out.freeze(), Some(skipped)),
Err(TryRecvError::Empty | TryRecvError::Closed) => return (out.freeze(), None),
}
}
}
impl Drop for LiveBody {
/// 客户端断开、掉队或写入端换代都走到这里：两个许可随之释放
fn drop(&mut self) {
debug!("直播预览连接结束，释放许可");
debug!(
id = self.id,
resyncs = self.resyncs,
"直播预览连接结束，释放许可"
);
}
}
/// 把订阅编成 chunked 响应：快照分块先发，之后逐个转发实时分块。
/// 把订阅编成 chunked 响应：快照分块先发，之后逐个转发实时分块；掉队时原地重对齐。
///
/// 两个许可（该路、进程）都随响应体一起活，客户端断开或响应结束即释放。
pub fn live_response(mut subscription: Subscription, global: OwnedSemaphorePermit) -> Response {
pub fn live_response(
id: i64,
hub: PreviewHub,
mut subscription: Subscription,
global: OwnedSemaphorePermit,
) -> Response {
let format = subscription.format;
let state = LiveBody {
id,
hub,
snapshot: VecDeque::from(std::mem::take(&mut subscription.snapshot)),
subscription,
subscription: Some(subscription),
pending_lag: None,
resyncs: 0,
_global: global,
};
let stream = futures::stream::unfold(state, |mut state| async move {
if let Some(chunk) = state.snapshot.pop_front() {
return Some((Ok::<Bytes, std::io::Error>(chunk), state));
}
match state.subscription.rx.recv().await {
Ok(chunk) => Some((Ok(chunk), state)),
Err(RecvError::Lagged(skipped)) => {
// 慢客户端：丢掉的数据没法补，结束响应让播放器重连拿新快照
warn!(skipped, "预览客户端落后于录制进度，断开以便其重连");
None
let stream = futures::stream::unfold(state, move |mut state| async move {
loop {
if let Some(chunk) = state.snapshot.pop_front() {
return Some((Ok::<Bytes, std::io::Error>(chunk), state));
}
let rx = &mut state.subscription.as_mut()?.rx;
let received = match state.pending_lag.take() {
Some(skipped) => Err(RecvError::Lagged(skipped)),
None => rx.recv().await,
};
match received {
Ok(chunk) => {
let (chunk, lagged) = coalesce(rx, chunk);
state.pending_lag = lagged;
return Some((Ok(chunk), state));
}
Err(RecvError::Lagged(skipped)) => {
// 慢客户端：丢掉的数据没法补，但不必断开——从最近的关键帧重新对齐，
// 快照里不再带文件头（fMP4 的 init segment 除外，MSE 接受中途再来一份）
state.resyncs += 1;
if state.resyncs > MAX_RESYNCS_PER_RESPONSE {
warn!(
id = state.id,
skipped,
resyncs = state.resyncs,
"预览客户端持续落后于录制进度，结束响应"
);
return None;
}
warn!(
id = state.id,
skipped,
resyncs = state.resyncs,
"预览客户端落后于录制进度，从最近的关键帧重新对齐"
);
let previous = state.subscription.take()?;
let again = state
.hub
.resubscribe(previous, SUBSCRIBE_TIMEOUT)
.await
.ok()?;
state.snapshot = if format == PreviewFormat::Fmp4 {
again.snapshot.iter().cloned().collect()
} else {
again.snapshot_after_header().iter().cloned().collect()
};
state.subscription = Some(again);
}
// 写入端换代（拉流结束 / 断流重试）：新连接会拿到新的序列头
Err(RecvError::Closed) => return None,
}
// 写入端换代（拉流结束 / 断流重试）：新连接会拿到新的序列头
Err(RecvError::Closed) => None,
}
});
let mut response = Body::from_stream(stream).into_response();
@@ -528,7 +632,7 @@ mod tests {
sink.push(ChunkKind::Media, Bytes::from_static(b"p2"));
let subscription = pending.await.unwrap().unwrap();
let global = connection_limit().clone().try_acquire_owned().unwrap();
let response = live_response(subscription, global);
let response = live_response(1, hub.clone(), subscription, global);
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
@@ -558,6 +662,101 @@ mod tests {
assert_eq!(&body[..], &expected[..]);
}
/// 掉队时不结束响应：从最近的关键帧重新对齐，续上的快照不再带 FLV 文件头，
/// 但带序列头（掉队期间可能换过）；响应体里文件头只出现一次。
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn lagged_client_is_realigned_in_place_instead_of_disconnected() {
use biliup::downloader::preview::BROADCAST_CAPACITY_FLV;
let hub = PreviewHub::new(4);
let mut sink = hub
.attach(PreviewFormat::Flv)
.with_snapshot_window(Duration::ZERO);
sink.push(ChunkKind::Header, Bytes::from_static(&flv::FILE_HEADER));
sink.push(ChunkKind::SequenceHeader(9), Bytes::from_static(b"avc"));
sink.push(ChunkKind::Keyframe, Bytes::from_static(b"K1"));
let pending = tokio::spawn({
let hub = hub.clone();
async move { hub.subscribe(Duration::from_secs(5)).await }
});
tokio::time::sleep(Duration::from_millis(20)).await;
sink.push(ChunkKind::Media, Bytes::from_static(b"p2"));
let subscription = pending.await.unwrap().unwrap();
let global = connection_limit().clone().try_acquire_owned().unwrap();
let response = live_response(1, hub.clone(), subscription, global);
// 响应体还没被读，写入端推满缓冲再多一些：接收端必然掉队
for _ in 0..(BROADCAST_CAPACITY_FLV + 100) {
sink.push(ChunkKind::Media, Bytes::from_static(b"x"));
}
// 读取任务：先拿到快照，撞上掉队 → 向写入端要新快照（要等下一个关键帧）
let reader = tokio::spawn(async move {
axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap()
});
tokio::time::sleep(Duration::from_millis(50)).await;
sink.push(ChunkKind::Keyframe, Bytes::from_static(b"K9"));
tokio::time::sleep(Duration::from_millis(20)).await;
sink.push(ChunkKind::Media, Bytes::from_static(b"p10"));
drop(sink);
let body = reader.await.unwrap();
let text = String::from_utf8_lossy(&body);
assert!(text.starts_with("FLV"), "{text}");
assert!(
text.ends_with("avcK9p10"),
"resync = seq header + new GOP, got {text}"
);
assert_eq!(text.matches("FLV").count(), 1, "file header only once");
assert!(
!text.contains("xx"),
"no stale chunks after the lag: {text}"
);
}
/// 合并：缓冲里攒着的分块合成一个响应块（不超过上限）；缓冲空时原样返回；
/// 途中掉队交回调用方。
#[tokio::test]
async fn coalesce_merges_backlog_and_reports_lag() {
let (tx, mut rx) = broadcast::channel::<Bytes>(4);
tx.send(Bytes::from_static(b"a")).unwrap();
let first = rx.recv().await.unwrap();
let (out, lag) = coalesce(&mut rx, first);
assert_eq!(&out[..], b"a");
assert!(lag.is_none());
tx.send(Bytes::from_static(b"b")).unwrap();
tx.send(Bytes::from_static(b"c")).unwrap();
tx.send(Bytes::from_static(b"d")).unwrap();
let first = rx.recv().await.unwrap();
let (out, lag) = coalesce(&mut rx, first);
assert_eq!(&out[..], b"bcd");
assert!(lag.is_none());
assert!(rx.is_empty());
// 上限：单个大分块不合并
let big = Bytes::from(vec![0u8; COALESCE_BYTES]);
tx.send(big.clone()).unwrap();
tx.send(Bytes::from_static(b"e")).unwrap();
let first = rx.recv().await.unwrap();
let (out, _) = coalesce(&mut rx, first);
assert_eq!(out.len(), COALESCE_BYTES);
assert_eq!(rx.recv().await.unwrap(), Bytes::from_static(b"e"));
// 掉队：合并到掉队处为止，把跳过数交回
for i in 0..8u8 {
tx.send(Bytes::from(vec![i])).unwrap();
}
assert!(matches!(rx.recv().await, Err(RecvError::Lagged(_))));
let first = rx.recv().await.unwrap();
tx.send(Bytes::from_static(b"f")).unwrap();
for _ in 0..6 {
tx.send(Bytes::from_static(b"g")).unwrap();
}
let (out, lag) = coalesce(&mut rx, first);
assert!(!out.is_empty());
assert!(lag.is_some(), "lag during coalescing must be reported");
}
#[test]
fn subscribe_errors_map_to_http_statuses() {
assert_eq!(
@@ -767,7 +966,7 @@ mod tests {
let subscription = pending.await.unwrap().unwrap();
let before = limit.available_permits();
let global = limit.clone().try_acquire_owned().unwrap();
let response = live_response(subscription, global);
let response = live_response(1, hub.clone(), subscription, global);
assert_eq!(
response.headers().get(header::CONTENT_TYPE).unwrap(),
"video/mp2t"
crates/biliup-cli/src/server/core/downloader/mesio.rs
+10 -3
View File
@@ -679,7 +679,10 @@ mod tests {
))
};
let hub = PreviewHub::new(4);
let mut sink = hub.attach(PreviewFormat::Flv);
// 这里只看 tee 的分块判定，快照只留当前 GOP（多 GOP 快照在 preview.rs 里测）
let mut sink = hub
.attach(PreviewFormat::Flv)
.with_snapshot_window(Duration::ZERO);
tee_flv(&mut sink, &FlvData::Header(FlvHeader::new(true, true)));
tee_flv(
&mut sink,
@@ -769,7 +772,9 @@ mod tests {
use biliup::downloader::preview::{PreviewFormat, PreviewHub};
let hub = PreviewHub::new(4);
let mut sink = hub.attach(PreviewFormat::MpegTs);
let mut sink = hub
.attach(PreviewFormat::MpegTs)
.with_snapshot_window(Duration::ZERO);
let segment = || m3u8_rs::MediaSegment {
uri: "1.ts".to_string(),
..Default::default()
@@ -805,7 +810,9 @@ mod tests {
out
}
let hub = PreviewHub::new(4);
let mut sink = hub.attach(PreviewFormat::Fmp4);
let mut sink = hub
.attach(PreviewFormat::Fmp4)
.with_snapshot_window(Duration::ZERO);
let segment = |uri: &str| m3u8_rs::MediaSegment {
uri: uri.to_string(),
..Default::default()
crates/biliup/src/downloader/httpflv.rs
+38 -25
View File
@@ -75,6 +75,22 @@ pub(crate) async fn parse_flv(
tag_data(tag_header.tag_type, tag_header.data_size as usize)(&bytes),
"tag data",
)?;
// 直播预览旁路：在解析点、进 GOP 缓存之前就推给 hub，让预览端按 tag 实时收到。
// 写盘循环要等到下一个关键帧才把整个 GOP 一起落盘，若在那里旁路，预览端会每个
// GOP 间隔收到一次整 GOP 的突发、其余时间颗粒不进，播放器缓冲刚好在下一个突发到达时
// 耗尽，链路上稍有 RTT / 抖动就 waiting（#1713 ForgQi 异地实测「非常卡」的根因）。
// 与写盘同一份字节，只是 push：不 await、不返回错误，预览端的状况不改变这里的控制流。
if let Some(sink) = preview.as_mut() {
let tag_type = tag_header.tag_type as u8;
sink.push(
preview::flv::classify(tag_type, &bytes),
preview::flv::tag_chunk_from_parts(
&preview::flv::tag_header(tag_type, tag_header.data_size, tag_header.timestamp),
&bytes,
&previous_tag_size,
),
);
}
let flv_tag = match flv_tag_data {
TagData::Audio(audio_data) => {
let packet_type = if audio_data.sound_format == SoundFormat::AAC {
@@ -173,23 +189,6 @@ pub(crate) async fn parse_flv(
);
}
out.write_tag(tag_header, flv_tag_data, previous_tag_size_bytes)?;
// 直播预览旁路：与写盘同一个 tag、同一份字节。只是 push，不 await、
// 不返回错误——预览端的任何状况都不改变这里的控制流。
if let Some(sink) = preview.as_mut() {
let tag_type = tag_header.tag_type as u8;
sink.push(
preview::flv::classify(tag_type, flv_tag_data),
preview::flv::tag_chunk_from_parts(
&preview::flv::tag_header(
tag_type,
tag_header.data_size,
tag_header.timestamp,
),
flv_tag_data,
previous_tag_size_bytes,
),
);
}
segment.increase_size((11 + tag_header.data_size + 4) as u64);
// downloaded_size += (11 + tag_header.data_size + 4) as u64;
prev_timestamp = tag_header.timestamp
@@ -463,10 +462,12 @@ mod tests {
chunks.iter().flat_map(|b| b.iter().copied()).collect()
}
/// 预览快照 + 实时分块拼起来，恰好等于「文件头 + 落盘的 tag 序列」的一段；
/// 跨多个分段文件（rolling）时连接不断、不重发 FLV 文件头。
/// 预览快照 + 实时分块拼起来，恰好等于源流「文件头 + 从第一个关键帧起的全部 tag」，
/// 即预览在解析点按 tag 实时旁路，不等写盘循环在下一个关键帧才整 GOP 落盘（那会让
/// 预览端每个 GOP 间隔收到一次突发，缓冲刚好在下一个突发到达时耗尽）；
/// 跨多个分段文件（rolling）时连接不断、不重发 FLV 文件头，分段本身照常切。
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn preview_follows_the_written_tags_across_rolling_segments()
async fn preview_follows_the_parsed_tags_across_rolling_segments()
-> Result<(), Box<dyn std::error::Error>> {
use crate::downloader::preview::{PreviewFormat, PreviewHub};
use crate::downloader::util::{LifecycleFile, Segmentable};
@@ -554,7 +555,15 @@ mod tests {
"existing subscribers must never get the FLV file header again"
);
// 落盘：多个分段，每个都以 FLV 头开始；预览收到的 tag 序列是落盘序列的后缀
// 预览 = 文件头 + 序列头 + 从第一个关键帧起的全部 tag = 源流逐字节（源流体不含 9 字节
// 文件头、含 PreviousTagSize0；也含写盘循环还留在缓存里、尚未落盘的最后一个 GOP——预览不等它）
assert_eq!(
&received[13..],
&body[4..],
"preview must reproduce the source tags byte for byte"
);
// 落盘不受影响：多个分段，每个都以 FLV 头开始，tag 序列是源流的子序列（分段处补了序列头）
let segments = segments.lock().unwrap();
assert!(
segments.len() > 5,
@@ -567,10 +576,14 @@ mod tests {
assert_eq!(&data[..13], &crate::downloader::preview::flv::FILE_HEADER);
written_tags.extend_from_slice(&data[13..]);
}
// 预览 = 快照里的序列头 + 从第一个关键帧起的全部 tag，与落盘的 tag 序列完全一致
let preview_tags = &received[13..];
assert_eq!(preview_tags.len(), written_tags.len());
assert!(written_tags.ends_with(preview_tags));
// 关键帧 NALU 的标记：写盘只有前 119 个（最后一个 GOP 留在缓存里没落盘），预览是全部 120 个
let keyframes = |data: &[u8]| {
data.windows(5)
.filter(|w| *w == [0x17, 0x01, 0, 0, 0])
.count()
};
assert_eq!(keyframes(&written_tags), 119);
assert_eq!(keyframes(&received), 120);
Ok(())
}
crates/biliup/src/downloader/preview.rs
+476 -16
View File
@@ -5,28 +5,42 @@
//! - 每路录制任务一个 [`PreviewHub`]，寿命与整场录制（`execute()`）相同，跨分段、跨断流重试。
//! - 下载器每次开始拉流时用 [`PreviewHub::attach`] 拿到一个 [`PreviewSink`]（写入端），
//! 在写盘点旁边调用 [`PreviewSink::push`]；拉流结束时 `PreviewSink` 随之 drop。
//! - HTTP 端点用 [`PreviewHub::subscribe`] 拿到 [`Subscription`]：先是一份「文件头 + 序列头 +
//! 当前 GOP」的快照，之后是与写盘同步的实时分块。
//! - HTTP 端点用 [`PreviewHub::subscribe`] / [`PreviewHub::subscribe_with_depth`] 拿到 [`Subscription`]：
//! 先是一份「文件头 + 序列头 + 最近一段时间内的完整 GOP + 当前 GOP」的快照（回溯多久由播放器按
//! 自己要维持的缓冲深度指定，最多 [`SNAPSHOT_WINDOW`]），之后是与写盘同步的实时分块。
//!
//! 写入端热路径上只做三件事：把分块的引用追加进当前 GOP 缓冲、`try_recv` 待处理的订阅请求、
//! `broadcast::send`。没有 `.await`、没有锁等待、没有可失败的返回值；订阅者的快慢只影响
//! 它自己（掉队即断开重连），不会传导回录制。
//! 它自己（掉队即从最近的关键帧重新对齐），不会传导回录制。
//!
//! 内存上限：GOP 快照最多 [`MAX_GOP_BYTES`]，超过就丢掉这一 GOP、等下一个关键帧；
//! 内存上限：单个 GOP 最多 [`MAX_GOP_BYTES`]，超过就丢掉这一 GOP、等下一个关键帧；整份快照
//! （已完成 GOP + 当前 GOP）最多 [`MAX_SNAPSHOT_BYTES`]，超过就先丢最旧的 GOP；
//! 广播缓冲按格式固定槽位数（[`BROADCAST_CAPACITY_FLV`] / [`BROADCAST_CAPACITY_SEGMENTED`]），
//! 每槽最多 [`MAX_CHUNK_BYTES`]（更大的分块会被切开），
//! 且只在有订阅者时才占用。订阅者数量由信号量限制（[`PreviewHub::new`] 的参数）。
use bytes::Bytes;
use std::collections::VecDeque;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::mpsc as std_mpsc;
use std::sync::{Arc, RwLock};
use std::time::Duration;
use std::time::{Duration, Instant};
use tokio::sync::{OwnedSemaphorePermit, Semaphore, broadcast, oneshot};
use tracing::debug;
/// 每路直播默认允许同时观看的预览连接数。
pub const DEFAULT_MAX_SUBSCRIBERS: usize = 4;
/// 快照里保留最近多少时间内（按到达时刻计）的已完成 GOP。
///
/// 新订阅者拿到的快照就是它起播时的全部缓冲：只给当前 GOP 时，播放器从零到一个 GOP 之间的
/// 缓冲起步，之后到达 = 消耗，缓冲永远这么薄，链路上几十到几百毫秒的抖动就 `waiting`
/// （浏览器直连 CDN 时 CDN 会先给几秒的 GOP 缓存，中转也得给同样的深度）。
/// 6 s 盖住前端最深的档位（「流畅」5 s 缓冲 + 起播对齐的余量）；更浅的档位（「低延迟」2 s）
/// 由播放器用 `snapshot_ms` 只要自己那一份，见 [`PreviewHub::subscribe_with_depth`]。
pub const SNAPSHOT_WINDOW: Duration = Duration::from_secs(6);
/// 整份快照（已完成 GOP + 当前 GOP）的字节上限，超过先丢最旧的 GOP。
/// 6 s × 20 Mbps = 15 MB，够到 4K 直播；1080p 常见的 3–6 Mbps 只用 2–5 MB。
pub const MAX_SNAPSHOT_BYTES: usize = 16 * 1024 * 1024;
/// FLV 的广播缓冲槽位数：一个分块就是一个 tag（音频几百字节、视频几 KB 到几十 KB），
/// 而 CDN 常常整 GOP 突发送达（实测斗鱼一次 ~320 个 tag / 2 MB 在 1 ms 内解析完），
/// 缓冲必须装得下一整个突发，否则刚建立的订阅者还没来得及读就 `Lagged`。
@@ -122,11 +136,15 @@ enum HubState {
struct SnapshotRequest {
reply: oneshot::Sender<SnapshotReply>,
/// 快照里已完成 GOP 最多回溯多久（见 [`PreviewHub::subscribe_with_depth`]）；`None` 给整个保留窗口
depth: Option<Duration>,
}
struct SnapshotReply {
format: PreviewFormat,
snapshot: Vec<Bytes>,
/// `snapshot` 开头有几个分块是文件头（0 或 1）
header_len: usize,
rx: broadcast::Receiver<Bytes>,
}
@@ -189,14 +207,25 @@ impl std::error::Error for SubscribeError {}
/// 一个预览订阅：先发 `snapshot` 里的分块，再从 `rx` 取实时分块。
///
/// 持有该路的一个连接许可，drop 即释放。
/// 持有该路的一个连接许可，drop 即释放；掉队后可用 [`PreviewHub::resubscribe`] 原地重新对齐。
pub struct Subscription {
pub format: PreviewFormat,
pub snapshot: Vec<Bytes>,
pub rx: broadcast::Receiver<Bytes>,
header_len: usize,
/// 订阅时要的快照深度，掉队重新对齐时沿用
depth: Option<Duration>,
_permit: OwnedSemaphorePermit,
}
impl Subscription {
/// 快照去掉开头的文件头：掉队后在同一条响应里续播时用——FLV 的解码器不能中途再收到
/// 文件头；序列头与 GOP 照常给（序列头可能在掉队期间换过）。
pub fn snapshot_after_header(&self) -> &[Bytes] {
&self.snapshot[self.header_len.min(self.snapshot.len())..]
}
}
impl PreviewHub {
/// 新建一个能 tee 的 hub，`max_subscribers` 为该路同时允许的预览连接数。
pub fn new(max_subscribers: usize) -> Self {
@@ -272,6 +301,10 @@ impl PreviewHub {
gop: Vec::new(),
gop_bytes: 0,
gop_ready: false,
gop_started: Instant::now(),
history: VecDeque::new(),
history_bytes: 0,
snapshot_window: SNAPSHOT_WINDOW,
fmp4_init: mp4::InitInfo::default(),
ring_used: false,
}
@@ -286,6 +319,18 @@ impl PreviewHub {
/// 严格衔接（不重复、不缺帧）；写入端会等到有完整关键帧起点时才回应，`timeout`
/// 需要长于一个 GOP / 一个 HLS 分片。
pub async fn subscribe(&self, timeout: Duration) -> Result<Subscription, SubscribeError> {
self.subscribe_with_depth(None, timeout).await
}
/// 同 [`subscribe`](Self::subscribe)，但快照里的已完成 GOP 只回溯 `depth`：
/// 从当前 GOP 往前，取到第一个关键帧到达时刻早于「现在 − depth」的 GOP 为止（含），
/// 所以起播缓冲在 `depth` 到 `depth + 一个 GOP` 之间；`Some(ZERO)` 只给当前 GOP，
/// `None` 给整个 [`SNAPSHOT_WINDOW`]。播放器要多深的缓冲就要多深的快照，多要的只会被追帧丢掉。
pub async fn subscribe_with_depth(
&self,
depth: Option<Duration>,
timeout: Duration,
) -> Result<Subscription, SubscribeError> {
if let HubState::Unavailable(reason) = &*self.0.state.read().unwrap() {
return Err(SubscribeError::Unavailable(reason.clone()));
}
@@ -295,19 +340,45 @@ impl PreviewHub {
.clone()
.try_acquire_owned()
.map_err(|_| SubscribeError::TooManySubscribers(self.0.max_subscribers))?;
self.subscribe_holding(permit, depth, timeout).await
}
/// 掉队（`Lagged`）后原地重新对齐：复用 `previous` 的连接许可与快照深度，向写入端再要一份从最近
/// 关键帧起的快照与新的接收端。丢掉的那段补不回来，但连接不断、播放器不必重连；
/// 调用方发快照时应去掉文件头（[`Subscription::snapshot_after_header`]）。
pub async fn resubscribe(
&self,
previous: Subscription,
timeout: Duration,
) -> Result<Subscription, SubscribeError> {
let Subscription { _permit, depth, .. } = previous;
self.subscribe_holding(_permit, depth, timeout).await
}
async fn subscribe_holding(
&self,
permit: OwnedSemaphorePermit,
depth: Option<Duration>,
timeout: Duration,
) -> Result<Subscription, SubscribeError> {
let request_tx = match &*self.0.requests.read().unwrap() {
Some((_, tx)) => tx.clone(),
None => return Err(SubscribeError::NotAttached),
};
let (reply_tx, reply_rx) = oneshot::channel();
request_tx
.send(SnapshotRequest { reply: reply_tx })
.send(SnapshotRequest {
reply: reply_tx,
depth,
})
.map_err(|_| SubscribeError::NotAttached)?;
match tokio::time::timeout(timeout, reply_rx).await {
Ok(Ok(reply)) => Ok(Subscription {
format: reply.format,
snapshot: reply.snapshot,
rx: reply.rx,
header_len: reply.header_len,
depth,
_permit: permit,
}),
// 写入端在回应前被 drop（拉流结束 / 重试），请求随之作废
@@ -332,17 +403,42 @@ pub struct PreviewSink {
gop_bytes: usize,
/// 当前 GOP 缓冲是否从关键帧起且未超上限；为 `false` 时订阅请求留到下一个关键帧再回应
gop_ready: bool,
/// 当前 GOP 的关键帧到达时刻
gop_started: Instant,
/// 已完成的 GOP，旧的在前；与当前 GOP 连续（中间没有被丢弃的 GOP、没有换过序列头）
history: VecDeque<Gop>,
history_bytes: usize,
/// 已完成 GOP 的保留时长，见 [`SNAPSHOT_WINDOW`]；零表示快照只含当前 GOP
snapshot_window: Duration,
/// fMP4：从 init segment 读出的视频轨信息，用于判定分片首帧是否关键帧
fmp4_init: mp4::InitInfo,
/// 广播缓冲里有没有数据；有且订阅者归零时换新通道释放
ring_used: bool,
}
/// 快照里一个已完成的 GOP。
struct Gop {
chunks: Vec<Bytes>,
bytes: usize,
started: Instant,
}
impl PreviewSink {
pub fn format(&self) -> PreviewFormat {
self.format
}
/// 改快照里已完成 GOP 的保留时长（默认 [`SNAPSHOT_WINDOW`]）；`Duration::ZERO` 只保留当前 GOP。
pub fn with_snapshot_window(mut self, window: Duration) -> Self {
self.snapshot_window = window;
self
}
/// 快照当前的字节数（已完成 GOP + 当前 GOP，不含文件头与序列头）。
pub fn snapshot_bytes(&self) -> usize {
self.history_bytes + self.gop_bytes
}
/// 当前有多少实时订阅者在收。
pub fn receiver_count(&self) -> usize {
self.tx.receiver_count()
@@ -389,7 +485,11 @@ impl PreviewSink {
if self.format == PreviewFormat::Fmp4 {
// init segment：每次 attach 只出现一次（上游换初始化分片时才会再来），
// 解出编码串供前端 addSourceBuffer，记下视频轨用于分片的关键帧判定；
// MSE 允许中途追加新 init，照常广播
// MSE 允许中途追加新 init，照常广播。换了 init 的分片参数可能变了，
// 旧 GOP 不能再和新 init 一起给新订阅者
if self.header.as_ref().is_some_and(|old| *old != chunk) {
self.drop_history();
}
self.fmp4_init = mp4::init_info(&chunk);
let codecs = mp4::codecs_from_init(&chunk);
if let HubState::Available { codecs: slot, .. } =
@@ -406,15 +506,23 @@ impl PreviewSink {
}
ChunkKind::SequenceHeader(slot) => {
match self.sequence_headers.iter_mut().find(|(s, _)| *s == slot) {
Some(entry) => entry.1 = chunk.clone(),
Some(entry) => {
// 序列头内容变了（换分辨率 / 编码参数）：之前的 GOP 与新序列头不配，
// 不能再放进快照；当前 GOP 沿用既有行为，与新序列头一起给出
if entry.1 != chunk {
entry.1 = chunk.clone();
self.drop_history();
}
}
None => self.sequence_headers.push((slot, chunk.clone())),
}
self.broadcast(chunk);
}
ChunkKind::Keyframe => {
self.gop.clear();
self.gop_bytes = 0;
let now = Instant::now();
self.archive_gop(now);
self.gop_ready = true;
self.gop_started = now;
self.retain(chunk.clone());
self.broadcast(chunk);
}
@@ -430,6 +538,45 @@ impl PreviewSink {
}
}
/// 关键帧到来：把当前 GOP 收进历史，再按时长与字节上限修剪历史。
fn archive_gop(&mut self, now: Instant) {
if self.gop_ready && !self.gop.is_empty() && !self.snapshot_window.is_zero() {
self.history_bytes += self.gop_bytes;
self.history.push_back(Gop {
chunks: std::mem::take(&mut self.gop),
bytes: self.gop_bytes,
started: self.gop_started,
});
} else {
self.gop.clear();
}
self.gop_bytes = 0;
while let Some(oldest) = self.history.front()
&& now.duration_since(oldest.started) > self.snapshot_window
{
self.pop_oldest_gop();
}
self.trim_history_bytes();
}
fn pop_oldest_gop(&mut self) {
if let Some(gop) = self.history.pop_front() {
self.history_bytes -= gop.bytes;
}
}
/// 整份快照超过字节上限时先丢最旧的 GOP；当前 GOP 自己的上限由 `retain` 管。
fn trim_history_bytes(&mut self) {
while !self.history.is_empty() && self.history_bytes + self.gop_bytes > MAX_SNAPSHOT_BYTES {
self.pop_oldest_gop();
}
}
fn drop_history(&mut self) {
self.history.clear();
self.history_bytes = 0;
}
fn retain(&mut self, chunk: Bytes) {
self.gop_bytes += chunk.len();
if self.gop_bytes > MAX_GOP_BYTES {
@@ -437,11 +584,14 @@ impl PreviewSink {
bytes = self.gop_bytes,
"preview GOP exceeds the snapshot limit, waiting for the next keyframe"
);
// 当前 GOP 作废，历史与之不再连续，一并放掉
self.gop.clear();
self.gop_bytes = 0;
self.gop_ready = false;
self.drop_history();
} else {
self.gop.push(chunk);
self.trim_history_bytes();
}
}
@@ -471,21 +621,48 @@ impl PreviewSink {
/// 先广播再处理请求：新接收端只会收到本分块之后的数据，而本分块已在快照里。
fn serve_requests(&mut self) {
let mut contiguous = false;
while let Ok(request) = self.requests.try_recv() {
if !contiguous {
// 历史只有几个 GOP，整理成连续切片的代价可以忽略；没有请求时不做
self.history.make_contiguous();
contiguous = true;
}
let rx = self.tx.subscribe();
let mut snapshot = Vec::with_capacity(1 + self.sequence_headers.len() + self.gop.len());
let history =
history_for_depth(self.history.as_slices().0, request.depth, Instant::now());
let history_chunks: usize = history.iter().map(|g| g.chunks.len()).sum();
let mut snapshot = Vec::with_capacity(
1 + self.sequence_headers.len() + history_chunks + self.gop.len(),
);
snapshot.extend(self.header.iter().cloned());
snapshot.extend(self.sequence_headers.iter().map(|(_, b)| b.clone()));
snapshot.extend(history.iter().flat_map(|g| g.chunks.iter().cloned()));
snapshot.extend(self.gop.iter().cloned());
let _ = request.reply.send(SnapshotReply {
format: self.format,
snapshot,
header_len: usize::from(self.header.is_some()),
rx,
});
}
}
}
/// 按要求的深度从（连续的）历史里取已完成的 GOP：`None` 全给；`Some(d)` 从最新往前取，
/// 取到第一个关键帧到达时刻早于 `now - d` 的 GOP 为止（含它，起播缓冲才不少于 `d`）。
fn history_for_depth(history: &[Gop], depth: Option<Duration>, now: Instant) -> &[Gop] {
let Some(depth) = depth else { return history };
if depth.is_zero() {
return &[];
}
let start = history
.iter()
.rposition(|g| now.duration_since(g.started) >= depth)
.unwrap_or(0);
&history[start..]
}
impl Drop for PreviewSink {
fn drop(&mut self) {
let mut requests = self.hub.requests.write().unwrap();
@@ -1194,7 +1371,10 @@ mod tests {
#[tokio::test]
async fn broadcast_ring_is_released_when_the_last_subscriber_leaves() {
let hub = PreviewHub::new(4);
let mut sink = hub.attach(PreviewFormat::Flv);
// 只看当前 GOP，快照历史另有测试
let mut sink = hub
.attach(PreviewFormat::Flv)
.with_snapshot_window(Duration::ZERO);
sink.push(ChunkKind::Header, Bytes::from_static(b"FLV"));
sink.push(ChunkKind::Keyframe, Bytes::from_static(b"K0"));
let pending = tokio::spawn({
@@ -1233,7 +1413,9 @@ mod tests {
#[tokio::test]
async fn a_whole_gop_burst_does_not_lag_a_fresh_flv_subscriber() {
let hub = PreviewHub::new(4);
let mut sink = hub.attach(PreviewFormat::Flv);
let mut sink = hub
.attach(PreviewFormat::Flv)
.with_snapshot_window(Duration::ZERO);
sink.push(ChunkKind::Header, Bytes::from_static(b"FLV"));
sink.push(ChunkKind::Keyframe, Bytes::from_static(b"K0"));
let pending = tokio::spawn({
@@ -1465,8 +1647,8 @@ mod tests {
let segment2 = Bytes::from_static(b"moof+mdat #2");
sink.push(ChunkKind::Keyframe, segment2.clone());
let mut sub = pending.await.unwrap().unwrap();
// 快照 = init + 最近一个完整分片
assert_eq!(sub.snapshot, vec![init.clone(), segment2]);
// 快照 = init + 窗口内的关键帧分片（两个都刚到，都在）
assert_eq!(sub.snapshot, vec![init.clone(), segment, segment2]);
// 上游换 init 时，已有订阅者也会收到新的 init（MSE 允许中途追加）
sink.push(ChunkKind::Header, init.clone());
assert_eq!(sub.rx.recv().await.unwrap(), init);
@@ -1678,6 +1860,227 @@ mod tests {
assert_eq!(sink.gop, vec![key(2)]);
}
/// 快照 = 文件头 + 序列头 + 窗口内的已完成 GOP + 当前 GOP，随后与实时分块严格衔接。
/// 新订阅者由此从起播就有几秒缓冲，而不是从零到一个 GOP 之间起步。
#[tokio::test]
async fn snapshot_keeps_recent_gops_within_the_window() {
let hub = PreviewHub::new(4);
let mut sink = hub.attach(PreviewFormat::Flv);
sink.push(ChunkKind::Header, Bytes::from_static(&flv::FILE_HEADER));
sink.push(ChunkKind::SequenceHeader(9), Bytes::from_static(b"avc"));
sink.push(ChunkKind::Media, inter(0)); // 关键帧之前的不算
sink.push(ChunkKind::Keyframe, key(1));
sink.push(ChunkKind::Media, inter(2));
sink.push(ChunkKind::Keyframe, key(3));
sink.push(ChunkKind::Media, inter(4));
sink.push(ChunkKind::Keyframe, key(5));
assert_eq!(sink.history.len(), 2);
assert_eq!(sink.snapshot_bytes(), 3 * key(0).len() + 2 * inter(0).len());
let pending = tokio::spawn({
let hub = hub.clone();
async move { hub.subscribe(Duration::from_secs(5)).await }
});
tokio::time::sleep(Duration::from_millis(20)).await;
sink.push(ChunkKind::Media, inter(6));
let mut sub = pending.await.unwrap().unwrap();
assert_eq!(
sub.snapshot,
vec![
Bytes::from_static(&flv::FILE_HEADER),
Bytes::from_static(b"avc"),
key(1),
inter(2),
key(3),
inter(4),
key(5),
inter(6),
]
);
sink.push(ChunkKind::Keyframe, key(7));
assert_eq!(sub.rx.recv().await.unwrap(), key(7));
}
/// 按深度取历史：从最新往前，取到第一个早于「现在 − 深度」的 GOP 为止（含）；
/// 零深度只给当前 GOP；`None` 给全部；深度大于全部历史时也给全部。
#[test]
fn history_is_cut_by_requested_depth() {
let now = Instant::now();
let gop = |age_ms: u64, k: u8| Gop {
chunks: vec![key(k)],
bytes: key(k).len(),
started: now - Duration::from_millis(age_ms),
};
// 关键帧分别在 5.0 / 3.0 / 1.0 s 前到达（当前 GOP 不在历史里）
let history = vec![gop(5000, 1), gop(3000, 3), gop(1000, 5)];
let picked = |depth: Option<Duration>| -> Vec<Bytes> {
history_for_depth(&history, depth, now)
.iter()
.map(|g| g.chunks[0].clone())
.collect()
};
assert_eq!(picked(None), vec![key(1), key(3), key(5)]);
assert_eq!(picked(Some(Duration::ZERO)), Vec::<Bytes>::new());
// 要 2 s：K5 只有 1 s，再往前 K3（3 s 前）跨过了 2 s 的线，取到它为止
assert_eq!(picked(Some(Duration::from_secs(2))), vec![key(3), key(5)]);
// 要 3 s：K3 正好 3 s 前到达（>=），取到它
assert_eq!(picked(Some(Duration::from_secs(3))), vec![key(3), key(5)]);
assert_eq!(
picked(Some(Duration::from_millis(3500))),
vec![key(1), key(3), key(5)]
);
assert_eq!(
picked(Some(Duration::from_secs(30))),
vec![key(1), key(3), key(5)]
);
assert_eq!(picked(Some(Duration::from_millis(500))), vec![key(5)]);
assert!(history_for_depth(&[], Some(Duration::from_secs(2)), now).is_empty());
}
/// 端到端：带深度订阅拿到的快照只含回溯范围内的 GOP，掉队重对齐时沿用同一深度。
#[tokio::test]
async fn subscribe_with_depth_trims_the_snapshot() {
let hub = PreviewHub::new(4);
let mut sink = hub.attach(PreviewFormat::Flv);
sink.push(ChunkKind::Header, Bytes::from_static(&flv::FILE_HEADER));
sink.push(ChunkKind::Keyframe, key(1));
sink.push(ChunkKind::Media, inter(2));
tokio::time::sleep(Duration::from_millis(60)).await;
sink.push(ChunkKind::Keyframe, key(3));
sink.push(ChunkKind::Media, inter(4));
tokio::time::sleep(Duration::from_millis(60)).await;
sink.push(ChunkKind::Keyframe, key(5));
assert_eq!(sink.history.len(), 2);
// 要 30 ms：K3 的 GOP（60 ms 前开始）是第一个跨过线的，K1 不给
let pending = tokio::spawn({
let hub = hub.clone();
async move {
hub.subscribe_with_depth(Some(Duration::from_millis(30)), Duration::from_secs(5))
.await
}
});
tokio::time::sleep(Duration::from_millis(20)).await;
sink.push(ChunkKind::Media, inter(6));
let sub = pending.await.unwrap().unwrap();
assert_eq!(
sub.snapshot,
vec![
Bytes::from_static(&flv::FILE_HEADER),
key(3),
inter(4),
key(5),
inter(6)
]
);
assert_eq!(sub.depth, Some(Duration::from_millis(30)));
// 零深度：只有当前 GOP
let pending = tokio::spawn({
let hub = hub.clone();
async move {
hub.subscribe_with_depth(Some(Duration::ZERO), Duration::from_secs(5))
.await
}
});
tokio::time::sleep(Duration::from_millis(20)).await;
sink.push(ChunkKind::Media, inter(7));
let sub0 = pending.await.unwrap().unwrap();
assert_eq!(
sub0.snapshot,
vec![
Bytes::from_static(&flv::FILE_HEADER),
key(5),
inter(6),
inter(7)
]
);
// 重对齐沿用深度
let pending = tokio::spawn({
let hub = hub.clone();
async move { hub.resubscribe(sub0, Duration::from_secs(5)).await }
});
tokio::time::sleep(Duration::from_millis(20)).await;
sink.push(ChunkKind::Media, inter(8));
let again = pending.await.unwrap().unwrap();
assert_eq!(again.depth, Some(Duration::ZERO));
assert_eq!(
again.snapshot_after_header(),
&[key(5), inter(6), inter(7), inter(8)]
);
}
/// 超出窗口时长的 GOP 在下一个关键帧到来时被丢掉（按关键帧到达时刻算）。
#[test]
fn gops_older_than_the_window_are_dropped_at_the_next_keyframe() {
let hub = PreviewHub::new(4);
let mut sink = hub
.attach(PreviewFormat::Flv)
.with_snapshot_window(Duration::from_millis(30));
sink.push(ChunkKind::Keyframe, key(1));
sink.push(ChunkKind::Media, inter(2));
std::thread::sleep(Duration::from_millis(80));
sink.push(ChunkKind::Keyframe, key(3));
// K1 的 GOP 已经 80 ms 前开始，超出 30 ms 的窗口
assert!(sink.history.is_empty());
assert_eq!(sink.snapshot_bytes(), key(3).len());
sink.push(ChunkKind::Keyframe, key(4));
// K3 刚开始不久，留下
assert_eq!(sink.history.len(), 1);
assert_eq!(sink.history[0].chunks, vec![key(3)]);
}
/// 整份快照超过字节上限时先丢最旧的 GOP，当前 GOP 保留。
#[test]
fn snapshot_history_is_bounded_by_bytes() {
let hub = PreviewHub::new(4);
let mut sink = hub.attach(PreviewFormat::Flv);
let big = Bytes::from(vec![0u8; MAX_SNAPSHOT_BYTES / 4 + 1]);
for i in 0..6u8 {
sink.push(ChunkKind::Keyframe, key(i));
sink.push(ChunkKind::Media, big.clone());
}
// 每个 GOP 略大于 1/4 上限：最多 3 个能同时在快照里
assert!(sink.snapshot_bytes() <= MAX_SNAPSHOT_BYTES);
assert_eq!(sink.history.len(), 2);
assert_eq!(sink.history[0].chunks[0], key(3));
assert_eq!(sink.gop[0], key(5));
}
/// 序列头内容变了（换分辨率）：之前的 GOP 与新序列头不配，从快照里去掉；
/// 内容相同的重发（stream-gears 分段时重放 onMetaData / 序列头）不影响历史。
#[test]
fn changed_sequence_header_drops_older_gops_but_a_repeat_does_not() {
let hub = PreviewHub::new(4);
let mut sink = hub.attach(PreviewFormat::Flv);
sink.push(ChunkKind::SequenceHeader(9), Bytes::from_static(b"avc"));
sink.push(ChunkKind::Keyframe, key(1));
sink.push(ChunkKind::Keyframe, key(2));
sink.push(ChunkKind::SequenceHeader(9), Bytes::from_static(b"avc"));
assert_eq!(sink.history.len(), 1, "identical re-send keeps history");
sink.push(
ChunkKind::SequenceHeader(9),
Bytes::from_static(b"avc-1080p"),
);
assert!(sink.history.is_empty(), "changed header drops history");
assert_eq!(sink.gop, vec![key(2)], "current GOP is kept as before");
}
/// 当前 GOP 超限作废时历史与之不再连续，一并放掉，等下一个关键帧从头攒。
#[test]
fn oversized_gop_also_drops_the_history() {
let hub = PreviewHub::new(4);
let mut sink = hub.attach(PreviewFormat::Flv);
sink.push(ChunkKind::Keyframe, key(1));
sink.push(ChunkKind::Keyframe, key(2));
assert_eq!(sink.history.len(), 1);
sink.push(ChunkKind::Media, Bytes::from(vec![0u8; MAX_GOP_BYTES + 1]));
assert!(!sink.gop_ready);
assert!(sink.history.is_empty());
assert_eq!(sink.snapshot_bytes(), 0);
sink.push(ChunkKind::Keyframe, key(3));
assert!(sink.gop_ready);
assert!(sink.history.is_empty());
assert_eq!(sink.gop, vec![key(3)]);
}
#[tokio::test]
async fn large_chunks_are_split_for_broadcast_but_kept_whole_in_snapshot() {
let hub = PreviewHub::new(4);
@@ -1707,6 +2110,63 @@ mod tests {
assert!(sub.rx.try_recv().is_err());
}
/// 掉队后 `resubscribe`：复用同一个许可（不占第二个名额），拿到从最近关键帧起的新快照
/// 与新接收端；`snapshot_after_header` 去掉文件头供同一条响应续播。
#[tokio::test]
async fn resubscribe_reuses_the_permit_and_realigns_at_a_keyframe() {
let hub = PreviewHub::new(1);
let mut sink = hub
.attach(PreviewFormat::Flv)
.with_snapshot_window(Duration::ZERO);
sink.push(ChunkKind::Header, Bytes::from_static(&flv::FILE_HEADER));
sink.push(ChunkKind::SequenceHeader(9), Bytes::from_static(b"avc"));
sink.push(ChunkKind::Keyframe, key(1));
let pending = tokio::spawn({
let hub = hub.clone();
async move { hub.subscribe(Duration::from_secs(5)).await }
});
tokio::time::sleep(Duration::from_millis(20)).await;
sink.push(ChunkKind::Media, inter(2));
let mut sub = pending.await.unwrap().unwrap();
assert_eq!(hub.0.subscribers.available_permits(), 0);
// 名额已满，第二个订阅被拒
assert_eq!(
hub.subscribe(Duration::from_secs(1)).await.err(),
Some(SubscribeError::TooManySubscribers(1))
);
// 订阅者不读，写入端推满整个缓冲再多一些 → 掉队
for i in 0..(BROADCAST_CAPACITY_FLV + 300) {
sink.push(ChunkKind::Media, inter((i % 200) as u8));
}
assert!(matches!(sub.rx.recv().await, Err(RecvError::Lagged(_))));
let pending = tokio::spawn({
let hub = hub.clone();
async move { hub.resubscribe(sub, Duration::from_secs(5)).await }
});
tokio::time::sleep(Duration::from_millis(20)).await;
sink.push(ChunkKind::Keyframe, key(9));
let mut again = pending.await.unwrap().unwrap();
assert_eq!(hub.0.subscribers.available_permits(), 0, "same permit");
assert_eq!(sink.receiver_count(), 1, "old receiver is gone");
assert_eq!(
again.snapshot,
vec![
Bytes::from_static(&flv::FILE_HEADER),
Bytes::from_static(b"avc"),
key(9)
]
);
assert_eq!(
again.snapshot_after_header(),
&[Bytes::from_static(b"avc"), key(9)]
);
sink.push(ChunkKind::Media, inter(10));
assert_eq!(again.rx.recv().await.unwrap(), inter(10));
drop(again);
assert_eq!(hub.0.subscribers.available_permits(), 1);
}
/// 订阅者上限：超出的订阅立刻被拒，释放一个后又能进。
#[tokio::test]
async fn subscriber_limit_is_enforced_per_hub() {