Better logging for packet stability and hopefully latency control.

gumble/gumble/audio_sender.go

@@ -0,0 +1,95 @@
+package gumble
+
+import (
+	"math"
+	"sync/atomic"
+
+	"git.stormux.org/storm/barnard/log"
+)
+
+type outgoingAudioFrame struct {
+	buffer   AudioBuffer
+	sequence int64
+}
+
+type audioFrameWriter func(AudioBuffer, int64, bool) error
+
+// AudioSender decouples real-time audio production from encoding and network
+// writes while keeping no more than a small, fixed amount of stale audio.
+// AudioSender has one producer and must not be used after Close.
+type AudioSender struct {
+	frames    chan outgoingAudioFrame
+	frameStep int64
+	next      int64
+	dropped   atomic.Uint64
+	write     audioFrameWriter
+}
+
+func newAudioSender(frameStep int64, queueFrames int, write audioFrameWriter) *AudioSender {
+	s := &AudioSender{
+		frames:    make(chan outgoingAudioFrame, queueFrames),
+		frameStep: frameStep,
+		write:     write,
+	}
+	go s.run()
+	return s
+}
+
+// Send queues one interval of audio without blocking. It returns false when
+// an older queued frame had to be discarded to keep latency bounded.
+func (s *AudioSender) Send(buffer AudioBuffer) bool {
+	frame := outgoingAudioFrame{buffer: buffer, sequence: s.next}
+	s.next = (s.next + s.frameStep) % math.MaxInt32
+
+	select {
+	case s.frames <- frame:
+		return true
+	default:
+	}
+
+	// The consumer may have freed capacity since the first select. Only remove
+	// a frame if one is still queued; either way, the consumer can only make
+	// more room before the replacement send.
+	didDrop := false
+	select {
+	case <-s.frames:
+		didDrop = true
+	default:
+	}
+	s.frames <- frame
+	if !didDrop {
+		return true
+	}
+	dropped := s.dropped.Add(1)
+	if dropped == 1 || dropped%100 == 0 {
+		log.Warn("outgoing audio queue congested: dropped %d stale frame(s)", dropped)
+	}
+	return false
+}
+
+// Dropped returns the number of stale frames discarded by Send.
+func (s *AudioSender) Dropped() uint64 {
+	return s.dropped.Load()
+}
+
+// Close finishes the stream. The most recent queued frame is marked as the
+// final packet so the receiver can close the talk burst cleanly.
+func (s *AudioSender) Close() {
+	close(s.frames)
+}
+
+func (s *AudioSender) run() {
+	previous, ok := <-s.frames
+	if !ok {
+		return
+	}
+	for frame := range s.frames {
+		if err := s.write(previous.buffer, previous.sequence, false); err != nil {
+			log.Error("outgoing audio write failed: %v", err)
+		}
+		previous = frame
+	}
+	if err := s.write(previous.buffer, previous.sequence, true); err != nil {
+		log.Error("final outgoing audio write failed: %v", err)
+	}
+}

gumble/gumble/audio_sender_test.go

@@ -0,0 +1,62 @@
+package gumble
+
+import (
+	"sync"
+	"testing"
+)
+
+func TestAudioSenderDropsOldestWithoutCompressingSequence(t *testing.T) {
+	type writeCall struct {
+		sequence int64
+		final    bool
+	}
+
+	started := make(chan struct{})
+	release := make(chan struct{})
+	var once sync.Once
+	writes := make(chan writeCall, 4)
+	sender := newAudioSender(1, 2, func(_ AudioBuffer, sequence int64, final bool) error {
+		once.Do(func() {
+			close(started)
+			<-release
+		})
+		writes <- writeCall{sequence: sequence, final: final}
+		return nil
+	})
+
+	sender.Send(AudioBuffer{0})
+	sender.Send(AudioBuffer{1})
+	<-started
+	sender.Send(AudioBuffer{2})
+	sender.Send(AudioBuffer{3})
+	if sender.Send(AudioBuffer{4}) {
+		t.Fatal("congested send did not report a dropped frame")
+	}
+	if got := sender.Dropped(); got != 1 {
+		t.Fatalf("dropped frames = %d, want 1", got)
+	}
+	sender.Close()
+	close(release)
+
+	want := []writeCall{{sequence: 0}, {sequence: 1}, {sequence: 3}, {sequence: 4, final: true}}
+	for i, expected := range want {
+		got := <-writes
+		if got != expected {
+			t.Fatalf("write %d = %+v, want %+v", i, got, expected)
+		}
+	}
+}
+
+func TestAudioSenderCloseWithoutFrames(t *testing.T) {
+	writes := make(chan struct{}, 1)
+	sender := newAudioSender(1, 1, func(AudioBuffer, int64, bool) error {
+		writes <- struct{}{}
+		return nil
+	})
+	sender.Close()
+	select {
+	case <-writes:
+		t.Fatal("empty stream wrote an audio packet")
+	default:
+	}
+}

gumble/gumble/client.go

@@ -80,6 +80,8 @@ type Client struct {
 	udpCryptoOut      *cryptState15
 	udpCryptoIn       *cryptState15
 	udpFrameNumber    uint64
+	udpAudioSequence  int64
+	udpSequenceValid  bool
 	udpProtobuf       bool
 	udpFallbackLogged atomic.Bool
 	udpFirstRecv      atomic.Bool
@@ -287,6 +289,29 @@ func (c *Client) AudioOutgoing() chan<- AudioBuffer {
 	return ch
 }
 
+// NewAudioSender creates a latency-bounded outgoing audio stream. Unlike
+// AudioOutgoing, Send never makes the capture goroutine wait for Opus encoding
+// or a congested network socket. When the encoder falls behind, the oldest
+// queued frame is discarded and sequence numbers retain the resulting gap so
+// receivers can apply packet-loss concealment.
+func (c *Client) NewAudioSender() *AudioSender {
+	frameStep := int64(c.Config.AudioFrameSize() / AudioDefaultFrameSize)
+	if frameStep < 1 {
+		frameStep = 1
+	}
+	interval := c.Config.AudioInterval
+	if interval <= 0 {
+		interval = AudioDefaultInterval
+	}
+	queueFrames := int((80 * time.Millisecond) / interval)
+	if queueFrames < 1 {
+		queueFrames = 1
+	}
+	return newAudioSender(frameStep, queueFrames, func(buffer AudioBuffer, sequence int64, final bool) error {
+		return buffer.writeAudio(c, sequence, final)
+	})
+}
+
 // pingRoutine sends ping packets to the server at regular intervals.
 func (c *Client) pingRoutine() {
 	ticker := time.NewTicker(time.Second * 5)

gumble/gumble/udp15.go

@@ -607,6 +607,8 @@ func (c *Client) setUDP15Crypto(key, clientNonce, serverNonce []byte) error {
 	c.udpCryptoOut = outbound
 	c.udpCryptoIn = inbound
 	c.udpFrameNumber = 0
+	c.udpAudioSequence = 0
+	c.udpSequenceValid = false
 	c.udpLastGood = time.Now()
 	c.udpLastRequest = time.Time{}
 	c.udpMu.Unlock()
@@ -650,7 +652,7 @@ func (c *Client) WriteAudioUDP15(format byte, target uint32, sequence int64, dat
 	frameNum := c.udpFrameNumber
 	protobuf := c.udpProtobuf
 	if protobuf {
-		c.udpFrameNumber++
+		frameNum = c.nextUDPFrameNumberLocked(sequence, final)
 	}
 	c.udpMu.Unlock()
 	if cs == nil || udpConn == nil {
@@ -682,6 +684,26 @@ func (c *Client) WriteAudioUDP15(format byte, target uint32, sequence int64, dat
 	return true, nil
 }
 
+// nextUDPFrameNumberLocked maps a producer's per-burst audio timestamp onto
+// the connection-wide Mumble 1.5 frame timeline. c.udpMu must be held.
+func (c *Client) nextUDPFrameNumberLocked(sequence int64, final bool) uint64 {
+	frameNum := c.udpFrameNumber
+	// The public audio sequence is a timestamp in 10 ms units. Preserve gaps
+	// introduced by a latency-bounded producer so the remote decoder can
+	// conceal missing audio instead of compressing time. Keep the wire frame
+	// number monotonic across separate talk bursts.
+	if c.udpSequenceValid {
+		delta := sequence - c.udpAudioSequence
+		if delta > 1 && delta < 1000 {
+			frameNum += uint64(delta - 1)
+		}
+	}
+	c.udpFrameNumber = frameNum + 1
+	c.udpAudioSequence = sequence
+	c.udpSequenceValid = !final
+	return frameNum
+}
+
 // HandleUDPPacket15 processes an incoming Mumble 1.5 native UDP packet.
 func (c *Client) HandleUDPPacket15(packet []byte, pktNum uint64) {
 	if len(packet) < udp15HeaderSize {

gumble/gumble/udp_audio_sequence_test.go

@@ -0,0 +1,29 @@
+package gumble
+
+import "testing"
+
+func TestUDPFrameNumberPreservesProducerSequenceGaps(t *testing.T) {
+	c := &Client{udpProtobuf: true}
+
+	next := func(sequence int64, final bool) uint64 {
+		c.udpMu.Lock()
+		defer c.udpMu.Unlock()
+		return c.nextUDPFrameNumberLocked(sequence, final)
+	}
+
+	for i, test := range []struct {
+		sequence int64
+		final    bool
+		want     uint64
+	}{
+		{sequence: 0, want: 0},
+		{sequence: 1, want: 1},
+		{sequence: 3, want: 3}, // sequence 2 was discarded
+		{sequence: 4, final: true, want: 4},
+		{sequence: 0, want: 5}, // a new talk burst stays monotonic
+	} {
+		if got := next(test.sequence, test.final); got != test.want {
+			t.Fatalf("case %d: frame number = %d, want %d", i, got, test.want)
+		}
+	}
+}

gumble/gumbleffmpeg/stream.go

@@ -164,8 +164,8 @@ func (s *Stream) process() {
 
 	byteBuffer := make([]byte, frameSize*2)
 
-	outgoing := s.client.AudioOutgoing()
-	defer close(outgoing)
+	outgoing := s.client.NewAudioSender()
+	defer outgoing.Close()
 
 	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
@@ -189,7 +189,7 @@ func (s *Stream) process() {
 				int16Buffer[i] = int16(volume * float)
 			}
 			atomic.AddInt64(&s.elapsed, int64(interval))
-			outgoing <- gumble.AudioBuffer(int16Buffer)
+			outgoing.Send(gumble.AudioBuffer(int16Buffer))
 		}
 	}
 }

gumble/gumbleopenal/stream.go

@@ -941,8 +941,8 @@ func (s *Stream) sourceRoutine(inputDevice *string, stop chan bool, done chan st
 	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
 
-	outgoing := s.client.AudioOutgoing()
-	defer close(outgoing)
+	outgoing := s.client.NewAudioSender()
+	defer outgoing.Close()
 
 	var micFailed bool
 	for {
@@ -1059,7 +1059,7 @@ func (s *Stream) sourceRoutine(inputDevice *string, stop chan bool, done chan st
 			// Determine what to send
 			if hasFileAudio {
 				// Send stereo buffer when file is playing
-				outgoing <- gumble.AudioBuffer(outputBuffer)
+				outgoing.Send(gumble.AudioBuffer(outputBuffer))
 				if recorder := s.getRecorder(); recorder != nil {
 					recorder.RecordAudioFrame(recorderOutgoingSource, outputBuffer, true)
 				}
@@ -1076,7 +1076,7 @@ func (s *Stream) sourceRoutine(inputDevice *string, stop chan bool, done chan st
 					}
 					outBuf = monoBuf
 				}
-				outgoing <- gumble.AudioBuffer(outBuf)
+				outgoing.Send(gumble.AudioBuffer(outBuf))
 				if recorder := s.getRecorder(); recorder != nil {
 					recorder.RecordAudioFrame(recorderOutgoingSource, outBuf, false)
 				}

tonetest.go

@@ -32,8 +32,8 @@ func StartToneGenerator(client *gumble.Client, stop <-chan struct{}) {
 	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
 
-	outgoing := client.AudioOutgoing()
-	defer close(outgoing)
+	outgoing := client.NewAudioSender()
+	defer outgoing.Close()
 
 	fmt.Fprintf(os.Stderr, "tonetest: starting 440 Hz tone generator (frameSize=%d, interval=%v)\n",
 		frameSize, interval)
@@ -53,7 +53,7 @@ func StartToneGenerator(client *gumble.Client, stop <-chan struct{}) {
 					phase -= 2.0 * math.Pi
 				}
 			}
-			outgoing <- gumble.AudioBuffer(buf)
+			outgoing.Send(gumble.AudioBuffer(buf))
 		}
 	}
 }