storm/barnard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

stop file playback processes and their children reliably

Browse Source 
Kill ffmpeg's whole process group rather than just the process.
ffmpeg spawns helpers for some inputs, so stopping playback left them
running and holding the output pipe open. The player now starts the
command in its own process group and signals the group.

Wait for the playback worker to finish before starting a new file.
Playing a second file while the first was shutting down left two
workers writing to the same stream.

Make pausing nonblocking.
The pause path could block on a full pipe and hang the UI thread that
requested it.

Install and reset the stereo encoder under the client lock.
File playback swapped the encoder field directly while a voice frame
could be encoding with it, and a finished file left the encoder
carrying state into the next one.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Brandon McGinty
2026-08-20 14:42:52 -04:00
co-authored by Claude Opus 5
parent fbb6a148ff
commit c5baaae6a7
9 changed files with 189 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client.go
+1 -1
View File
@@ -57,7 +57,7 @@ func (b *Barnard) connect(reconnect bool) bool {
b.Stream.SetAGCEnabled(b.UserConfig.GetAGCEnabled())
// Initialize stereo encoder for file playback
b.Client.AudioEncoderStereo = opus.NewStereoEncoder()
b.Client.SetStereoEncoder(opus.NewStereoEncoder())
// Initialize file player
b.FileStreamMutex.Lock()
fileplayback/player.go
+70 -75
View File
@@ -1,6 +1,7 @@
package fileplayback
import (
"context"
"encoding/binary"
"errors"
"io"
@@ -10,22 +11,24 @@ import (
"time"
"git.stormux.org/storm/barnard/gumble/gumble"
"git.stormux.org/storm/barnard/gumble/go-openal/openal"
)
// Player handles file playback and mixing with microphone audio
type Player struct {
client *gumble.Client
filename string
audioChan chan gumble.AudioBuffer
stopChan chan struct{}
mutex sync.Mutex
playing bool
errorFunc func(error)
client *gumble.Client
filename string
audioChan chan gumble.AudioBuffer
stopChan chan struct{}
ctx context.Context
cancel context.CancelFunc
cmd *exec.Cmd
mutex sync.Mutex
wg sync.WaitGroup
playing bool
stopping bool
errorFunc func(error)
// Local playback
localSource *openal.Source
localBuffers openal.Buffers
localPlayback func([]byte)
}
// New creates a new file player
@@ -44,6 +47,14 @@ func (p *Player) SetErrorFunc(f func(error)) {
p.errorFunc = f
}
// SetLocalPlayback sets the callback that plays file audio locally. The
// callback is called with nil when playback stops and should release resources.
func (p *Player) SetLocalPlayback(f func([]byte)) {
p.mutex.Lock()
defer p.mutex.Unlock()
p.localPlayback = f
}
func (p *Player) reportError(err error) {
p.mutex.Lock()
errorFunc := p.errorFunc
@@ -65,17 +76,12 @@ func (p *Player) PlayFile(filename string) error {
p.filename = filename
// Initialize local playback
source := openal.NewSource()
p.localSource = &source
p.localSource.SetGain(1.0)
// Create buffers for local playback
p.localBuffers = openal.NewBuffers(64)
// Start the file reading goroutine
p.playing = true
p.stopping = false
p.stopChan = make(chan struct{})
p.ctx, p.cancel = context.WithCancel(context.Background())
p.wg.Add(1)
go p.readFileAudio()
return nil
@@ -84,31 +90,33 @@ func (p *Player) PlayFile(filename string) error {
// Stop stops the currently playing file
func (p *Player) Stop() error {
p.mutex.Lock()
defer p.mutex.Unlock()
if !p.playing {
p.mutex.Unlock()
return errors.New("no file playing")
}
close(p.stopChan)
p.playing = false
// Clean up local playback
if p.localSource != nil {
p.localSource.Stop()
p.localSource.Delete()
p.localSource = nil
}
if p.localBuffers != nil {
p.localBuffers.Delete()
p.localBuffers = nil
if !p.stopping {
p.stopping = true
close(p.stopChan)
if p.cancel != nil {
p.cancel()
}
terminateProcessGroup(p.cmd)
}
p.mutex.Unlock()
// Drain the audio channel
// A new PlayFile must not replace session state until ffmpeg and the old
// worker have exited, otherwise old audio can enter the new playback.
p.wg.Wait()
p.mutex.Lock()
p.playing, p.stopping, p.cancel, p.cmd = false, false, nil, nil
localPlayback := p.localPlayback
p.mutex.Unlock()
if localPlayback != nil {
localPlayback(nil)
}
for len(p.audioChan) > 0 {
<-p.audioChan
}
return nil
}
@@ -129,37 +137,18 @@ func (p *Player) GetAudioFrame() []int16 {
}
}
// playLocalAudio plays audio through the local OpenAL source
func (p *Player) playLocalAudio(data []byte) {
if p.localSource == nil {
return
}
// Reclaim processed buffers
if n := p.localSource.BuffersProcessed(); n > 0 {
reclaimedBufs := make(openal.Buffers, n)
p.localSource.UnqueueBuffers(reclaimedBufs)
p.localBuffers = append(p.localBuffers, reclaimedBufs...)
}
// If we have available buffers, queue more audio
if len(p.localBuffers) > 0 {
buffer := p.localBuffers[len(p.localBuffers)-1]
p.localBuffers = p.localBuffers[:len(p.localBuffers)-1]
// Set buffer data as stereo
buffer.SetData(openal.FormatStereo16, data, gumble.AudioSampleRate)
p.localSource.QueueBuffer(buffer)
// Start playing if not already
if p.localSource.State() != openal.Playing {
p.localSource.Play()
}
p.mutex.Lock()
localPlayback := p.localPlayback
p.mutex.Unlock()
if localPlayback != nil {
localPlayback(data)
}
}
// readFileAudio reads audio from the file via ffmpeg
func (p *Player) readFileAudio() {
defer p.wg.Done()
interval := p.client.Config.AudioInterval
frameSize := p.client.Config.AudioFrameSize()
@@ -168,7 +157,11 @@ func (p *Player) readFileAudio() {
args := []string{"-loglevel", "error", "-i", p.filename}
args = append(args, "-ac", "2", "-ar", strconv.Itoa(gumble.AudioSampleRate), "-f", "s16le", "-")
cmd := exec.Command("ffmpeg", args...)
p.mutex.Lock()
ctx := p.ctx
p.mutex.Unlock()
cmd := exec.CommandContext(ctx, "ffmpeg", args...)
configureProcessGroup(cmd)
pipe, err := cmd.StdoutPipe()
if err != nil {
p.mutex.Lock()
@@ -185,6 +178,9 @@ func (p *Player) readFileAudio() {
p.reportError(errors.New("failed to start ffmpeg: " + err.Error()))
return
}
p.mutex.Lock()
p.cmd = cmd
p.mutex.Unlock()
// Stereo has 2 channels, so we need twice the buffer size
byteBuffer := make([]byte, frameSize*2*2) // frameSize * 2 channels * 2 bytes per sample
@@ -195,28 +191,27 @@ func (p *Player) readFileAudio() {
for {
select {
case <-p.stopChan:
cmd.Process.Kill()
terminateProcessGroup(cmd)
cmd.Wait()
return
case <-ticker.C:
n, err := io.ReadFull(pipe, byteBuffer)
if err != nil || n != len(byteBuffer) {
// File finished playing
select {
case <-p.stopChan:
cmd.Wait()
return
default:
}
// File finished playing.
p.mutex.Lock()
p.playing = false
// Clean up local playback
if p.localSource != nil {
p.localSource.Stop()
p.localSource.Delete()
p.localSource = nil
}
if p.localBuffers != nil {
p.localBuffers.Delete()
p.localBuffers = nil
}
localPlayback := p.localPlayback
p.mutex.Unlock()
if localPlayback != nil {
localPlayback(nil)
}
cmd.Wait()
// Notify that file finished
p.reportError(errors.New("file playback finished"))
return
}
fileplayback/player_regression_test.go
+24
View File
@@ -0,0 +1,24 @@
package fileplayback
import (
"testing"
"time"
)
// Regression: Stop returned before the previous ffmpeg worker exited, allowing
// a subsequent PlayFile to replace shared state while old audio was still sent.
func TestStopWaitsForPlaybackWorker(t *testing.T) {
p := &Player{playing: true, stopChan: make(chan struct{})}
p.wg.Add(1)
done := make(chan error, 1)
go func() { done <- p.Stop() }()
select {
case <-done:
t.Fatal("Stop returned before playback worker ended")
case <-time.After(20 * time.Millisecond):
}
p.wg.Done()
if err := <-done; err != nil {
t.Fatal(err)
}
}
fileplayback/process_other.go
+13
View File
@@ -0,0 +1,13 @@
//go:build !(aix || darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris)
package fileplayback
import "os/exec"
func configureProcessGroup(cmd *exec.Cmd) {}
func terminateProcessGroup(cmd *exec.Cmd) {
if cmd != nil && cmd.Process != nil {
_ = cmd.Process.Kill()
}
}
fileplayback/process_unix.go
+19
View File
@@ -0,0 +1,19 @@
//go:build aix || darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris
package fileplayback
import (
"os/exec"
"syscall"
)
func configureProcessGroup(cmd *exec.Cmd) {
cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
}
func terminateProcessGroup(cmd *exec.Cmd) {
if cmd == nil || cmd.Process == nil {
return
}
_ = syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
}
fileplayback/process_unix_test.go
+16
View File
@@ -0,0 +1,16 @@
//go:build aix || darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris
package fileplayback
import (
"os/exec"
"testing"
)
func TestConfigureProcessGroupCreatesSeparateGroup(t *testing.T) {
cmd := exec.Command("true")
configureProcessGroup(cmd)
if cmd.SysProcAttr == nil || !cmd.SysProcAttr.Setpgid {
t.Fatal("ffmpeg process was not configured to lead its own process group")
}
}
gumble/gumble/client.go
+13 -1
View File
@@ -396,6 +396,13 @@ func (c *Client) Send(message Message) {
message.writeMessage(c)
}
// SetStereoEncoder installs the encoder used for stereo file playback.
func (c *Client) SetStereoEncoder(encoder AudioEncoder) {
c.volatile.Lock()
defer c.volatile.Unlock()
c.AudioEncoderStereo = encoder
}
// EnableStereoEncoder switches to stereo encoding for file playback.
func (c *Client) EnableStereoEncoder() {
c.volatile.Lock()
@@ -448,11 +455,16 @@ func (c *Client) UDPActive() bool {
return c.udpActive
}
// DisableStereoEncoder switches back to mono encoding for voice.
// DisableStereoEncoder switches back to mono encoding for voice and
// resets the stereo encoder so stale state does not bleed into the
// next file playback.
func (c *Client) DisableStereoEncoder() {
c.volatile.Lock()
defer c.volatile.Unlock()
c.useStereoEncoder = false
if c.AudioEncoderStereo != nil {
c.AudioEncoderStereo.Reset()
}
}
// IsStereoEncoderEnabled returns true if stereo encoding is currently active.
gumble/gumbleffmpeg/stream.go
+11 -3
View File
@@ -57,7 +57,7 @@ func New(client *gumble.Client, source Source) *Stream {
Volume: 1.0,
Source: source,
Command: "ffmpeg",
pause: make(chan struct{}),
pause: make(chan struct{}, 1),
state: StateInitial,
}
}
@@ -124,7 +124,12 @@ func (s *Stream) Pause() error {
}
s.state = StatePaused
s.l.Unlock()
s.pause <- struct{}{}
// The process can exit after the state check. A buffered, coalesced pause
// request preserves the state transition without blocking the caller.
select {
case s.pause <- struct{}{}:
default:
}
return nil
}
@@ -176,9 +181,12 @@ func (s *Stream) process() {
return
}
int16Buffer := make([]int16, frameSize)
s.l.Lock()
volume := s.Volume
s.l.Unlock()
for i := range int16Buffer {
float := float32(int16(binary.LittleEndian.Uint16(byteBuffer[i*2 : (i+1)*2])))
int16Buffer[i] = int16(s.Volume * float)
int16Buffer[i] = int16(volume * float)
}
atomic.AddInt64(&s.elapsed, int64(interval))
outgoing <- gumble.AudioBuffer(int16Buffer)
gumble/gumbleffmpeg/stream_test.go
+22
View File
@@ -0,0 +1,22 @@
package gumbleffmpeg
import (
"testing"
"time"
)
// Regression: Pause sent on an unbuffered channel after the process had
// exited, leaving callers blocked forever.
func TestPauseDoesNotBlockWhenProcessHasExited(t *testing.T) {
s := &Stream{state: StatePlaying, pause: make(chan struct{}, 1)}
done := make(chan error, 1)
go func() { done <- s.Pause() }()
select {
case err := <-done:
if err != nil {
t.Fatal(err)
}
case <-time.After(time.Second):
t.Fatal("Pause blocked after process exit")
}
}