PacedBuffer: add paced jitter buffer wrapper and tests - #24
Conversation
Introduces a paced wrapper around the jitter buffer that drains samples at media-clock cadence—absorbing bursts without blowing max timestamp drift. It can be configured to allow a lead up tiime so more sampels are pushed initially to prewarm subsequent processing stages.
biglittlebigben
left a comment
biglittlebigben
left a comment
There was a problem hiding this comment.
Thanks! Just a few remarks about switching to core.Fuse when possible/sensible as sticking with one mechanism with well understood semantics may help with maintenance long term.
| allowLead time.Duration // maximum lead allowed when pre-warming | ||
| maxLag time.Duration // maximum lag permitted before clamping | ||
|
|
||
| stop chan struct{} // closed to signal shutdown |
There was a problem hiding this comment.
We tend to use core.Fuse for this purpose as they tend to be safer (no panic on double close for instance)
|
|
||
| stop chan struct{} // closed to signal shutdown | ||
| start chan struct{} // closed when pacing should begin | ||
| done sync.WaitGroup |
There was a problem hiding this comment.
If we never add more than 1 entry to the WaitGroup, would a core.Fuse work here as well?
| close(pb.samples) | ||
| } | ||
|
|
||
| func (pb *PacedBuffer) handleSample(sample []*rtp.Packet) { |
There was a problem hiding this comment.
guessing this needs to change to take jitter.ExtPacket as jitter buffer emits those in the latest media-sdk.
There was a problem hiding this comment.
yes - the branch was created before these changes - I am going to integrate that - thanks for pointing out.
| return pb | ||
| } | ||
|
|
||
| func (pb *PacedBuffer) Samples() <-chan []*rtp.Packet { |
There was a problem hiding this comment.
I believe @dennwc changed the jitter buffer interface to use a callback rather than a channel. Should this follow the same pattern?
There was a problem hiding this comment.
makes sense - will update it.
| case <-p.timer.C: | ||
| default: |
There was a problem hiding this comment.
what's the purpose of this?
this will not wait
There was a problem hiding this comment.
It’s there to drain the timer’s channel when Stop reports that the timer has already fired. time.Timer.Stop() returns false if the timer either has fired or was never active; in the “fired” case the value may still be sitting in p.timer.C. That non‑blocking select snags the pending tick so the channel is empty before we call Reset again. Because of the default branch it never blocks—if there’s nothing to drain we just fall through. This pattern avoids stale wakeups or panics from resetting a timer whose channel still holds an unread tick.
| opts..., | ||
| ) | ||
|
|
||
| pb.done.Add(1) |
There was a problem hiding this comment.
where do we call .Done() on this wait group?
Maybe it's better to use errgroup?
There was a problem hiding this comment.
we call it here:
https://github.com/livekit/media-sdk/blob/add-pacer-buffer/jitter/paced_buffer.go#L260
Since we’re only tracking this single goroutine, the plain sync.WaitGroup already does the job; IMO switching to an errgroup wouldn’t add value unless we need aggregated error handling / multiple goroutines. But it will do the job as well.
4 participants
Introduces a paced wrapper around the jitter buffer that drains samples at media-clock cadence—absorbing bursts without blowing max timestamp drift. It can be configured to allow a lead up tiime so more sampels are pushed initially to prewarm subsequent processing stages.