Join capture worker before stopping OpenAL
This commit is contained in:
committed by
Brandon McGinty
parent
28a59832c4
commit
cc62f8e4a7
@@ -77,7 +77,7 @@ Priority 1: transport, lifecycle, and correctness
|
|||||||
failed. Mark UDP active only after an authenticated UDP response/audio
|
failed. Mark UDP active only after an authenticated UDP response/audio
|
||||||
packet (or retain TCP until confirmed), and define fallback/recovery rules.
|
packet (or retain TCP until confirmed), and define fallback/recovery rules.
|
||||||
|
|
||||||
9. Stream capture shutdown/startup races OpenAL
|
[x] 9. Stream capture shutdown/startup races OpenAL
|
||||||
File: gumble/gumbleopenal/stream.go
|
File: gumble/gumbleopenal/stream.go
|
||||||
StopSource closes a channel but does not wait for sourceRoutine. Destroy
|
StopSource closes a channel but does not wait for sourceRoutine. Destroy
|
||||||
immediately closes the capture device, so the routine can access a closed
|
immediately closes the capture device, so the routine can access a closed
|
||||||
|
|||||||
@@ -59,7 +59,9 @@ type Stream struct {
|
|||||||
sourceChannels int
|
sourceChannels int
|
||||||
sourceFrameSize int
|
sourceFrameSize int
|
||||||
micVolume atomic.Uint32 // float32 stored as bits
|
micVolume atomic.Uint32 // float32 stored as bits
|
||||||
|
sourceMu sync.Mutex
|
||||||
sourceStop chan bool
|
sourceStop chan bool
|
||||||
|
sourceDone chan struct{}
|
||||||
|
|
||||||
deviceSink *openal.Device
|
deviceSink *openal.Device
|
||||||
contextSink *openal.Context
|
contextSink *openal.Context
|
||||||
@@ -323,6 +325,8 @@ func (s *Stream) Destroy() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (s *Stream) StartSource(inputDevice *string) error {
|
func (s *Stream) StartSource(inputDevice *string) error {
|
||||||
|
s.sourceMu.Lock()
|
||||||
|
defer s.sourceMu.Unlock()
|
||||||
if s.sourceStop != nil {
|
if s.sourceStop != nil {
|
||||||
return ErrState
|
return ErrState
|
||||||
}
|
}
|
||||||
@@ -330,21 +334,29 @@ func (s *Stream) StartSource(inputDevice *string) error {
|
|||||||
return ErrMic
|
return ErrMic
|
||||||
}
|
}
|
||||||
s.deviceSource.CaptureStart()
|
s.deviceSource.CaptureStart()
|
||||||
s.sourceStop = make(chan bool)
|
stop := make(chan bool)
|
||||||
go s.sourceRoutine(inputDevice)
|
done := make(chan struct{})
|
||||||
|
s.sourceStop, s.sourceDone = stop, done
|
||||||
|
go s.sourceRoutine(inputDevice, stop, done)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *Stream) StopSource() error {
|
func (s *Stream) StopSource() error {
|
||||||
|
s.sourceMu.Lock()
|
||||||
|
if s.sourceStop == nil {
|
||||||
|
s.sourceMu.Unlock()
|
||||||
|
return ErrState
|
||||||
|
}
|
||||||
|
stop, done := s.sourceStop, s.sourceDone
|
||||||
|
s.sourceStop, s.sourceDone = nil, nil
|
||||||
|
close(stop)
|
||||||
|
s.sourceMu.Unlock()
|
||||||
|
// The routine owns capture access; wait for it before closing/reusing it.
|
||||||
|
<-done
|
||||||
if s.deviceSource == nil {
|
if s.deviceSource == nil {
|
||||||
return ErrMic
|
return ErrMic
|
||||||
}
|
}
|
||||||
s.deviceSource.CaptureStop()
|
s.deviceSource.CaptureStop()
|
||||||
if s.sourceStop == nil {
|
|
||||||
return ErrState
|
|
||||||
}
|
|
||||||
close(s.sourceStop)
|
|
||||||
s.sourceStop = nil
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -644,7 +656,8 @@ func (s *Stream) processAudioPacket(packet *gumble.AudioPacket, user *gumble.Use
|
|||||||
return emptyBufs
|
return emptyBufs
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *Stream) sourceRoutine(inputDevice *string) {
|
func (s *Stream) sourceRoutine(inputDevice *string, stop chan bool, done chan struct{}) {
|
||||||
|
defer close(done)
|
||||||
log.Info("source routine started: interval=%v frameSize=%d channels=%d",
|
log.Info("source routine started: interval=%v frameSize=%d channels=%d",
|
||||||
s.client.Config.AudioInterval, s.client.Config.AudioFrameSize(), s.sourceChannels)
|
s.client.Config.AudioInterval, s.client.Config.AudioFrameSize(), s.sourceChannels)
|
||||||
interval := s.client.Config.AudioInterval
|
interval := s.client.Config.AudioInterval
|
||||||
@@ -679,8 +692,6 @@ func (s *Stream) sourceRoutine(inputDevice *string) {
|
|||||||
ticker := time.NewTicker(interval)
|
ticker := time.NewTicker(interval)
|
||||||
defer ticker.Stop()
|
defer ticker.Stop()
|
||||||
|
|
||||||
stop := s.sourceStop
|
|
||||||
|
|
||||||
outgoing := s.client.AudioOutgoing()
|
outgoing := s.client.AudioOutgoing()
|
||||||
defer close(outgoing)
|
defer close(outgoing)
|
||||||
|
|
||||||
|
|||||||
@@ -4,6 +4,22 @@ import "testing"
|
|||||||
|
|
||||||
// Regression: audio cleanup could send a final render command after Destroy
|
// Regression: audio cleanup could send a final render command after Destroy
|
||||||
// had closed renderCh, panicking instead of safely discarding that work.
|
// had closed renderCh, panicking instead of safely discarding that work.
|
||||||
|
// Regression: StopSource returned before the capture worker ended, allowing
|
||||||
|
// Destroy to close the device while that worker still used it.
|
||||||
|
func TestStopSourceWaitsForWorker(t *testing.T) {
|
||||||
|
stop, done := make(chan bool), make(chan struct{})
|
||||||
|
s := &Stream{sourceStop: stop, sourceDone: done}
|
||||||
|
returned := make(chan struct{})
|
||||||
|
go func() { _ = s.StopSource(); close(returned) }()
|
||||||
|
select {
|
||||||
|
case <-returned:
|
||||||
|
t.Fatal("StopSource returned before worker")
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
close(done)
|
||||||
|
<-returned
|
||||||
|
}
|
||||||
|
|
||||||
func TestRenderRejectsWorkAfterShutdown(t *testing.T) {
|
func TestRenderRejectsWorkAfterShutdown(t *testing.T) {
|
||||||
s := &Stream{renderClosed: true}
|
s := &Stream{renderClosed: true}
|
||||||
called := false
|
called := false
|
||||||
|
|||||||
Reference in New Issue
Block a user