Wait for file playback workers before restart

This commit is contained in:
Brandon McGinty (chatgpt)
2026-08-09 18:23:19 -04:00
committed by Brandon McGinty
parent f629ce0359
commit 7ebe764d82
3 changed files with 44 additions and 13 deletions
+19 -12
View File
@@ -22,7 +22,9 @@ type Player struct {
ctx context.Context ctx context.Context
cancel context.CancelFunc cancel context.CancelFunc
mutex sync.Mutex mutex sync.Mutex
wg sync.WaitGroup
playing bool playing bool
stopping bool
errorFunc func(error) errorFunc func(error)
localPlayback func([]byte) localPlayback func([]byte)
@@ -75,8 +77,10 @@ func (p *Player) PlayFile(filename string) error {
// Start the file reading goroutine // Start the file reading goroutine
p.playing = true p.playing = true
p.stopping = false
p.stopChan = make(chan struct{}) p.stopChan = make(chan struct{})
p.ctx, p.cancel = context.WithCancel(context.Background()) p.ctx, p.cancel = context.WithCancel(context.Background())
p.wg.Add(1)
go p.readFileAudio() go p.readFileAudio()
return nil return nil
@@ -85,30 +89,32 @@ func (p *Player) PlayFile(filename string) error {
// Stop stops the currently playing file // Stop stops the currently playing file
func (p *Player) Stop() error { func (p *Player) Stop() error {
p.mutex.Lock() p.mutex.Lock()
defer p.mutex.Unlock()
if !p.playing { if !p.playing {
p.mutex.Unlock()
return errors.New("no file playing") return errors.New("no file playing")
} }
if !p.stopping {
close(p.stopChan) p.stopping = true
if p.cancel != nil { close(p.stopChan)
p.cancel() if p.cancel != nil {
p.cancel = nil p.cancel()
}
} }
p.playing = false
localPlayback := p.localPlayback
p.mutex.Unlock() p.mutex.Unlock()
// 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 = false, false, nil
localPlayback := p.localPlayback
p.mutex.Unlock()
if localPlayback != nil { if localPlayback != nil {
localPlayback(nil) localPlayback(nil)
} }
// Drain the audio channel.
for len(p.audioChan) > 0 { for len(p.audioChan) > 0 {
<-p.audioChan <-p.audioChan
} }
p.mutex.Lock()
return nil return nil
} }
@@ -140,6 +146,7 @@ func (p *Player) playLocalAudio(data []byte) {
// readFileAudio reads audio from the file via ffmpeg // readFileAudio reads audio from the file via ffmpeg
func (p *Player) readFileAudio() { func (p *Player) readFileAudio() {
defer p.wg.Done()
interval := p.client.Config.AudioInterval interval := p.client.Config.AudioInterval
frameSize := p.client.Config.AudioFrameSize() frameSize := p.client.Config.AudioFrameSize()
+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)
}
}
+1 -1
View File
@@ -99,7 +99,7 @@ Priority 1: transport, lifecycle, and correctness
overwrites b.Stream. Destroy/stop the old resources before reconnecting; overwrites b.Stream. Destroy/stop the old resources before reconnecting;
make disconnect cleanup idempotent. make disconnect cleanup idempotent.
12. File player sessions race each other [x] 12. File player sessions race each other
File: fileplayback/player.go File: fileplayback/player.go
readFileAudio repeatedly reads mutable Player stopChan/ctx/audioChan. readFileAudio repeatedly reads mutable Player stopChan/ctx/audioChan.
Stop can return and a new PlayFile can replace them while the old ffmpeg Stop can return and a new PlayFile can replace them while the old ffmpeg