diff --git a/internal/xiaomi/README.md b/internal/xiaomi/README.md index 330183e8c..9f590d251 100644 --- a/internal/xiaomi/README.md +++ b/internal/xiaomi/README.md @@ -10,6 +10,20 @@ go2rtc supports two formats: `xiaomi/mess` and `xiaomi/legacy`. And multiple P2P protocols: `cs2+udp`, `cs2+tcp`, several versions of `tutk+udp`. Almost all cameras in the `xiaomi/mess` format and the `cs2` protocol work well. + +Some CS2 cameras deliver video frames in short bursts even though their media +timestamps are stable. An optional bounded pacing buffer can smooth those +bursts for strict real-time consumers such as HomeKit: + +```yaml +streams: + camera: xiaomi://...&pacing=400 +``` + +The value is the initial video jitter buffer in milliseconds (maximum 5000). +It is disabled by default. When the queue grows beyond the configured buffer, +frames are released at a gently accelerated cadence instead of being dumped or +dropped, preventing latency from growing without bound. Older `xiaomi/legacy` format cameras may have support issues. The `tutk` protocol is the worst thing that's ever happened to the P2P world. It works terribly. diff --git a/pkg/xiaomi/miss/producer.go b/pkg/xiaomi/miss/producer.go index dcea284e3..c3e3d616d 100644 --- a/pkg/xiaomi/miss/producer.go +++ b/pkg/xiaomi/miss/producer.go @@ -3,6 +3,7 @@ package miss import ( "fmt" "net/url" + "strconv" "time" "github.com/AlexxIT/go2rtc/pkg/core" @@ -14,7 +15,8 @@ import ( type Producer struct { core.Connection - client *Client + client *Client + videoPacing time.Duration } func Dial(rawURL string) (core.Producer, error) { @@ -38,6 +40,11 @@ func Dial(rawURL string) (core.Producer, error) { return nil, err } + pacingMS, _ := strconv.Atoi(query.Get("pacing")) + if pacingMS < 0 || pacingMS > 5000 { + pacingMS = 0 + } + return &Producer{ Connection: core.Connection{ ID: core.NewID(), @@ -48,7 +55,8 @@ func Dial(rawURL string) (core.Producer, error) { Medias: medias, Transport: client, }, - client: client, + client: client, + videoPacing: time.Duration(pacingMS) * time.Millisecond, }, nil } @@ -129,6 +137,11 @@ const timestamp40ms = 48000 * 0.040 func (p *Producer) Start() error { var audioTS uint32 + var pacer *videoPacer + if p.videoPacing > 0 { + pacer = newVideoPacer(p.videoPacing) + defer pacer.Close() + } for { _ = p.client.SetDeadline(time.Now().Add(10 * time.Second)) @@ -186,7 +199,11 @@ func (p *Producer) Start() error { for _, recv := range p.Receivers { if recv.Codec.Name == name { - recv.WriteRTP(pkt2) + if pacer != nil && (name == core.CodecH264 || name == core.CodecH265) { + pacer.Write(recv, pkt2) + } else { + recv.WriteRTP(pkt2) + } break } } diff --git a/pkg/xiaomi/miss/video_pacer.go b/pkg/xiaomi/miss/video_pacer.go new file mode 100644 index 000000000..2a2166cce --- /dev/null +++ b/pkg/xiaomi/miss/video_pacer.go @@ -0,0 +1,185 @@ +package miss + +import ( + "sync/atomic" + "time" + + "github.com/AlexxIT/go2rtc/pkg/core" +) + +const videoClockRate = 90000 + +type pacedVideoFrame struct { + timestamp uint32 + receiver *core.Receiver + packets []*core.Packet +} + +// videoPacer absorbs the bursty delivery used by some Xiaomi CS2 cameras. +// Packets with the same RTP timestamp form one encoded frame and are released +// together. Frames are then released according to the camera RTP clock. +type videoPacer struct { + buffer time.Duration + frames chan pacedVideoFrame + stop chan struct{} + done chan struct{} + + latestTimestamp atomic.Uint32 + current pacedVideoFrame +} + +func newVideoPacer(buffer time.Duration) *videoPacer { + p := &videoPacer{ + buffer: buffer, + frames: make(chan pacedVideoFrame, 1024), + stop: make(chan struct{}), + done: make(chan struct{}), + } + go p.run() + return p +} + +func (p *videoPacer) Write(receiver *core.Receiver, packet *core.Packet) { + if len(p.current.packets) == 0 { + p.current = pacedVideoFrame{ + timestamp: packet.Timestamp, + receiver: receiver, + packets: []*core.Packet{packet}, + } + return + } + + if p.current.timestamp == packet.Timestamp && p.current.receiver == receiver { + p.current.packets = append(p.current.packets, packet) + return + } + + p.enqueueCurrent() + p.current = pacedVideoFrame{ + timestamp: packet.Timestamp, + receiver: receiver, + packets: []*core.Packet{packet}, + } +} + +func (p *videoPacer) enqueueCurrent() { + if len(p.current.packets) == 0 { + return + } + p.latestTimestamp.Store(p.current.timestamp) + p.frames <- p.current + p.current = pacedVideoFrame{} +} + +func (p *videoPacer) Close() { + close(p.stop) + <-p.done +} + +func (p *videoPacer) run() { + defer close(p.done) + + // Let the source build a small jitter buffer before the first frame. The + // goroutine producing CS2 packets continues filling p.frames meanwhile. + if p.buffer > 0 { + timer := time.NewTimer(p.buffer) + select { + case <-timer.C: + case <-p.stop: + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + return + } + } + + var lastTimestamp uint32 + var lastRelease time.Time + var lastInterval = 100 * time.Millisecond + + for { + var frame pacedVideoFrame + select { + case frame = <-p.frames: + case <-p.stop: + return + } + now := time.Now() + if !lastRelease.IsZero() { + interval := rtpDuration(frame.timestamp - lastTimestamp) + if interval < 10*time.Millisecond || interval > 250*time.Millisecond { + interval = lastInterval + } + lastInterval = interval + + backlog := rtpDuration(p.latestTimestamp.Load() - frame.timestamp) + interval = pacedFrameInterval(interval, backlog, p.buffer) + target := lastRelease.Add(interval) + if wait := time.Until(target); wait > 0 { + timer := time.NewTimer(wait) + select { + case <-timer.C: + case <-p.stop: + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + return + } + } + now = time.Now() + lastRelease = pacedReleaseAnchor(lastRelease, now, interval) + } + + for _, packet := range frame.packets { + frame.receiver.WriteRTP(packet) + } + lastTimestamp = frame.timestamp + if lastRelease.IsZero() { + lastRelease = now + } + } +} + +func rtpDuration(delta uint32) time.Duration { + // A delta larger than a minute indicates an RTP timestamp reset rather + // than a real frame interval or queue depth. + if delta > videoClockRate*60 { + return 0 + } + return time.Duration(delta) * time.Second / videoClockRate +} + +func pacedFrameInterval(interval, backlog, buffer time.Duration) time.Duration { + if buffer <= 0 { + return interval + } + // Keep normal cadence around the requested buffer. If a recovered CS2 + // burst grows the queue, catch up gently instead of dumping all frames at + // once. No encoded prediction frames are discarded. + if backlog > buffer*5/2 { + return interval / 2 + } + if backlog > buffer*3/2 { + return interval * 3 / 4 + } + return interval +} + +// pacedReleaseAnchor advances the pacing clock by the requested media interval +// instead of anchoring every frame to the timer's actual wake-up time. Timer +// overshoot is expected and would otherwise accumulate into seconds of latency +// during a long-running prebuffer. Rebase only after falling more than one +// frame behind so a real scheduler stall is not released as a packet burst. +func pacedReleaseAnchor(lastRelease, actual time.Time, interval time.Duration) time.Time { + target := lastRelease.Add(interval) + if actual.Sub(target) > interval { + return actual + } + return target +} diff --git a/pkg/xiaomi/miss/video_pacer_test.go b/pkg/xiaomi/miss/video_pacer_test.go new file mode 100644 index 000000000..f1e39ce83 --- /dev/null +++ b/pkg/xiaomi/miss/video_pacer_test.go @@ -0,0 +1,43 @@ +package miss + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestRTPDuration(t *testing.T) { + require.Equal(t, 100*time.Millisecond, rtpDuration(9000)) + require.Equal(t, time.Duration(0), rtpDuration(videoClockRate*61)) +} + +func TestPacedFrameInterval(t *testing.T) { + buffer := 400 * time.Millisecond + interval := 100 * time.Millisecond + + require.Equal(t, interval, pacedFrameInterval(interval, 400*time.Millisecond, buffer)) + require.Equal(t, 75*time.Millisecond, pacedFrameInterval(interval, 700*time.Millisecond, buffer)) + require.Equal(t, 50*time.Millisecond, pacedFrameInterval(interval, 1100*time.Millisecond, buffer)) +} + +func TestPacedReleaseAnchorDoesNotAccumulateTimerOvershoot(t *testing.T) { + interval := 50 * time.Millisecond + start := time.Unix(100, 0) + anchor := start + + for range 12000 { // ten minutes at 20 fps + actual := anchor.Add(interval + time.Millisecond) + anchor = pacedReleaseAnchor(anchor, actual, interval) + } + + require.Equal(t, start.Add(10*time.Minute), anchor) +} + +func TestPacedReleaseAnchorRebasesAfterRealStall(t *testing.T) { + interval := 50 * time.Millisecond + last := time.Unix(100, 0) + actual := last.Add(2*interval + time.Nanosecond) + + require.Equal(t, actual, pacedReleaseAnchor(last, actual, interval)) +}