Add optional pacing for bursty Xiaomi CS2 video

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.
 

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
 			}
 		}

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
+}

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))
+}