Add configurable incoming audio jitter buffer

README.md

@@ -154,6 +154,21 @@ barnard --audio-interval 20
 Supported values are `10`, `20`, `40`, and `60` milliseconds. Try `20` ms
 first; use `40` ms only if the connection remains unreliable.
 
+## Incoming Audio Jitter Buffer
+
+Barnard holds 40 ms of audio separately for each speaker before starting
+playback. This prevents brief delayed UDP packets from draining OpenAL's audio
+queue, which otherwise produces clicks or pops. To adjust this tradeoff between
+resilience and added incoming latency:
+
+```sh
+barnard --jitter-buffer 60
+```
+
+Supported values are `0`, `20`, `40` (default), and `60` milliseconds. Try
+`60` ms for a lossy or jittery connection. Use `0` only when minimizing latency
+is more important than avoiding playback underruns.
+
 ## Audio Devices
 
 You can set the default input and output devices in the config file as well.

client_notification_test.go

@@ -86,6 +86,22 @@ func TestAudioIntervalDuration(t *testing.T) {
 	}
 }
 
+func TestJitterBufferDuration(t *testing.T) {
+	for _, milliseconds := range []int{0, 20, 40, 60} {
+		got, err := jitterBufferDuration(milliseconds)
+		if err != nil {
+			t.Errorf("jitterBufferDuration(%d): %v", milliseconds, err)
+			continue
+		}
+		if got != time.Duration(milliseconds)*time.Millisecond {
+			t.Errorf("jitterBufferDuration(%d) = %v", milliseconds, got)
+		}
+	}
+	if _, err := jitterBufferDuration(10); err == nil {
+		t.Fatal("jitterBufferDuration accepted unsupported duration")
+	}
+}
+
 func TestServerAddressDefaultsPortWithoutBreakingIPv6(t *testing.T) {
 	for input, want := range map[string]string{
 		"server":          "server:64738",

gumble/gumble/config.go

@@ -25,6 +25,9 @@ type Config struct {
 	AudioInterval time.Duration
 	// AudioDataBytes is the number of bytes that an audio frame can use.
 	AudioDataBytes int
+	// IncomingAudioBuffer is the amount of per-speaker audio retained before
+	// playback starts, absorbing jitter in incoming UDP packet delivery.
+	IncomingAudioBuffer time.Duration
 
 	// DisableUDP forces all audio to use the TCP tunnel instead of UDP.
 	DisableUDP bool
@@ -38,9 +41,10 @@ type Config struct {
 // NewConfig returns a new Config struct with default values set.
 func NewConfig() *Config {
 	return &Config{
-		Buffers:        8,
-		AudioInterval:  AudioDefaultInterval,
-		AudioDataBytes: AudioDefaultDataBytes,
+		Buffers:             8,
+		AudioInterval:       AudioDefaultInterval,
+		AudioDataBytes:      AudioDefaultDataBytes,
+		IncomingAudioBuffer: 40 * time.Millisecond,
 	}
 }
 
@@ -54,6 +58,9 @@ func (c *Config) Validate() error {
 	if c.AudioDataBytes <= 0 {
 		return fmt.Errorf("gumble: AudioDataBytes must be positive")
 	}
+	if c.IncomingAudioBuffer < 0 {
+		return fmt.Errorf("gumble: IncomingAudioBuffer must not be negative")
+	}
 	if c.Buffers <= 0 {
 		return fmt.Errorf("gumble: Buffers must be positive")
 	}

gumble/gumble/handlers.go

@@ -210,7 +210,6 @@ func (c *Client) handleUDPTunnel(buffer []byte) error {
 		},
 		Sequence:    seq,
 		AudioBuffer: AudioBuffer(pcm),
-		Terminator:  isFinal,
 	}
 
 	if len(buffer)-audioLength == 3*4 {
@@ -227,6 +226,7 @@ func (c *Client) handleUDPTunnel(buffer []byte) error {
 	if isFinal {
 		decoder.Reset()
 		user.audioSequenceValid = false
+		c.dispatchAudio(user, &AudioPacket{Client: c, Sender: user, Terminator: true})
 	}
 	return nil
 }

gumble/gumble/udp15.go

@@ -861,7 +861,6 @@ func (c *Client) decodeAndDispatch(pktNum uint64, user *User, decoder AudioDecod
 		Target:           &VoiceTarget{ID: context},
 		Sequence:         frameNum,
 		AudioBuffer:      AudioBuffer(pcm),
-		Terminator:       terminator,
 		VolumeAdjustment: volumeAdjustment,
 	}
 	if position != nil {
@@ -873,6 +872,7 @@ func (c *Client) decodeAndDispatch(pktNum uint64, user *User, decoder AudioDecod
 		decoder.Reset()
 		user.audioSequenceValid = false
 		user.audioFrameStep = 0
+		c.dispatchAudio(user, &AudioPacket{Client: c, Sender: user, Terminator: true})
 	}
 }
 

gumble/gumbleopenal/stream.go

@@ -52,14 +52,22 @@ const recorderOutgoingSource uint32 = ^uint32(0)
 
 const (
 	maxBufferSize    = 11520 // Max frame size (2880) * bytes per stereo sample (4)
-	jitterMinPackets = 3
-	jitterMaxPackets = 10
+	jitterMaxPackets = 50
 )
 
-// jitterPlaybackReady holds the initial playout delay only once. Requiring
-// the minimum on every packet drains and refills the renderer in bursts.
-func jitterPlaybackReady(started bool, buffered int) bool {
-	return started || buffered >= jitterMinPackets
+// jitterPlaybackReady holds the requested initial playout delay only once.
+// Requiring the delay on every packet drains and refills the renderer in bursts.
+func jitterPlaybackReady(started bool, buffered, target time.Duration) bool {
+	return started || buffered >= target
+}
+
+func audioPacketDuration(packet *gumble.AudioPacket) time.Duration {
+	if packet == nil || len(packet.AudioBuffer) == 0 {
+		return 0
+	}
+	// Opus decoders deliver interleaved stereo PCM to this renderer.
+	frames := len(packet.AudioBuffer) / gumble.AudioChannels
+	return time.Duration(frames) * time.Second / gumble.AudioSampleRate
 }
 
 var (
@@ -484,11 +492,13 @@ func (s *Stream) OnAudioStream(e *gumble.AudioStreamEvent) {
 		// Jitter buffer: collects incoming packets, reorders by
 		// sequence number, and releases them after a small initial delay.
 		var jitterBuf []*gumble.AudioPacket
+		var jitterDuration time.Duration
 		var jitterNextSeq int64
 		var jitterInit, jitterStarted bool
 		var jitterDrainLogCounter, jitterAnomalyLogCounter int
 		resetJitter := func() {
 			jitterBuf = nil
+			jitterDuration = 0
 			jitterNextSeq = 0
 			jitterInit = false
 			jitterStarted = false
@@ -513,6 +523,7 @@ func (s *Stream) OnAudioStream(e *gumble.AudioStreamEvent) {
 			jitterBuf = append(jitterBuf, nil)
 			copy(jitterBuf[i+1:], jitterBuf[i:])
 			jitterBuf[i] = p
+			jitterDuration += audioPacketDuration(p)
 		}
 
 		// popNext removes and returns the packet with the expected next
@@ -523,6 +534,7 @@ func (s *Stream) OnAudioStream(e *gumble.AudioStreamEvent) {
 			}
 			p := jitterBuf[0]
 			jitterBuf = jitterBuf[1:]
+			jitterDuration -= audioPacketDuration(p)
 			// Frame numbers are Mumble timestamps in 10 ms units.
 			// Compute the actual step from the PCM sample count so we
 			// never skip a legitimate gap.
@@ -571,8 +583,8 @@ func (s *Stream) OnAudioStream(e *gumble.AudioStreamEvent) {
 
 			// Hold only the initial packets. Once playback starts, drain every
 			// ready packet so the renderer is fed continuously rather than in
-			// bursts of jitterMinPackets packets.
-			if !jitterPlaybackReady(jitterStarted, len(jitterBuf)) {
+			// bursts of packets.
+			if !jitterPlaybackReady(jitterStarted, jitterDuration, e.Client.Config.IncomingAudioBuffer) {
 				continue
 			}
 			jitterStarted = true
@@ -590,6 +602,7 @@ func (s *Stream) OnAudioStream(e *gumble.AudioStreamEvent) {
 								log.Debug("jitter: discarding late seq=%d for %s (next=%d buf=%d)",
 									jitterBuf[0].Sequence, e.User.Name, jitterNextSeq, len(jitterBuf))
 							}
+							jitterDuration -= audioPacketDuration(jitterBuf[0])
 							jitterBuf = jitterBuf[1:]
 							continue
 						}

gumble/gumbleopenal/stream_regression_test.go

@@ -4,8 +4,10 @@ import (
 	"errors"
 	"strings"
 	"testing"
+	"time"
 
 	"git.stormux.org/storm/barnard/gumble/go-openal/openal"
+	"git.stormux.org/storm/barnard/gumble/gumble"
 )
 
 // Regression: audio cleanup could send a final render command after Destroy
@@ -49,17 +51,24 @@ func TestStopSourceWaitsForWorker(t *testing.T) {
 }
 
 func TestJitterPlaybackDelayAppliesOnlyAtStartup(t *testing.T) {
-	if jitterPlaybackReady(false, jitterMinPackets-1) {
+	if jitterPlaybackReady(false, 20*time.Millisecond, 40*time.Millisecond) {
 		t.Fatal("jitter playback started before initial buffer filled")
 	}
-	if !jitterPlaybackReady(false, jitterMinPackets) {
+	if !jitterPlaybackReady(false, 40*time.Millisecond, 40*time.Millisecond) {
 		t.Fatal("jitter playback did not start after initial buffer filled")
 	}
-	if !jitterPlaybackReady(true, 1) {
+	if !jitterPlaybackReady(true, 0, 40*time.Millisecond) {
 		t.Fatal("jitter playback paused while refilling after startup")
 	}
 }
 
+func TestAudioPacketDurationUsesStereoFrameCount(t *testing.T) {
+	packet := &gumble.AudioPacket{AudioBuffer: make(gumble.AudioBuffer, 2*gumble.AudioDefaultFrameSize)}
+	if got := audioPacketDuration(packet); got != 10*time.Millisecond {
+		t.Fatalf("audioPacketDuration = %v, want 10ms", got)
+	}
+}
+
 func TestRenderRejectsWorkAfterShutdown(t *testing.T) {
 	s := &Stream{renderClosed: true}
 	called := false

main.go

@@ -86,6 +86,7 @@ func main() {
 	certificateSet := false
 	buffers := flag.Int("buffers", 16, "number of audio buffers to use")
 	audioInterval := flag.Int("audio-interval", 10, "outgoing audio packet duration in ms (10, 20, 40, or 60)")
+	jitterBuffer := flag.Int("jitter-buffer", 40, "incoming per-user audio buffer in ms (0, 20, 40, or 60)")
 	profile := flag.Bool("profile", false, "add http server to serve profiles")
 	noiseSuppressionEnabled := flag.Bool("noise-suppression", false, "enable noise suppression for microphone input")
 	autoTransmit := flag.Bool("auto-transmit", false, "start transmitting immediately on connect")
@@ -100,6 +101,10 @@ func main() {
 	if err != nil {
 		handle_raw_error(err)
 	}
+	selectedJitterBuffer, err := jitterBufferDuration(*jitterBuffer)
+	if err != nil {
+		handle_raw_error(err)
+	}
 
 	// Set up logging
 	var level barnlog.Level
@@ -199,6 +204,7 @@ func main() {
 	}
 	b.Config.Buffers = *buffers
 	b.Config.AudioInterval = selectedAudioInterval
+	b.Config.IncomingAudioBuffer = selectedJitterBuffer
 	b.Config.DisableUDP = *tcpOnly
 
 	b.Hotkeys = b.UserConfig.GetHotkeys()
@@ -257,6 +263,18 @@ func audioIntervalDuration(milliseconds int) (time.Duration, error) {
 	}
 }
 
+// jitterBufferDuration converts the requested incoming playout delay to a
+// supported duration. Zero starts playback without an initial safety buffer.
+func jitterBufferDuration(milliseconds int) (time.Duration, error) {
+	interval := time.Duration(milliseconds) * time.Millisecond
+	switch interval {
+	case 0, 20 * time.Millisecond, 40 * time.Millisecond, 60 * time.Millisecond:
+		return interval, nil
+	default:
+		return 0, fmt.Errorf("jitter buffer must be 0, 20, 40, or 60 ms, got %d", milliseconds)
+	}
+}
+
 // serverAddress adds Mumble's default port without corrupting an IPv6 literal.
 func serverAddress(address string) string {
 	if _, port, err := net.SplitHostPort(address); err == nil && port != "" {