diff --git a/gumble/gumble/audio_sender.go b/gumble/gumble/audio_sender.go new file mode 100644 index 0000000..8848935 --- /dev/null +++ b/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) + } +} diff --git a/gumble/gumble/audio_sender_test.go b/gumble/gumble/audio_sender_test.go new file mode 100644 index 0000000..89084ee --- /dev/null +++ b/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: + } +} diff --git a/gumble/gumble/client.go b/gumble/gumble/client.go index 6b7705f..4d5a49e 100644 --- a/gumble/gumble/client.go +++ b/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) diff --git a/gumble/gumble/udp15.go b/gumble/gumble/udp15.go index cd8deba..a57a3aa 100644 --- a/gumble/gumble/udp15.go +++ b/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 { diff --git a/gumble/gumble/udp_audio_sequence_test.go b/gumble/gumble/udp_audio_sequence_test.go new file mode 100644 index 0000000..97d4d07 --- /dev/null +++ b/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) + } + } +} diff --git a/gumble/gumbleffmpeg/stream.go b/gumble/gumbleffmpeg/stream.go index bed0c14..7aafe5a 100644 --- a/gumble/gumbleffmpeg/stream.go +++ b/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)) } } } diff --git a/gumble/gumbleopenal/stream.go b/gumble/gumbleopenal/stream.go index b00ba24..0fa7d14 100644 --- a/gumble/gumbleopenal/stream.go +++ b/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) } diff --git a/tonetest.go b/tonetest.go index 95d929c..a9c8827 100644 --- a/tonetest.go +++ b/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)) } } }