Make ffmpeg pause nonblocking
This commit is contained in:
committed by
Brandon McGinty
parent
e564bb8583
commit
977b690eed
@@ -121,7 +121,7 @@ Priority 1: transport, lifecycle, and correctness
|
|||||||
output file creation fails, the tone goroutine continues. Create the saver
|
output file creation fails, the tone goroutine continues. Create the saver
|
||||||
first, or close/wait for the tone generator and reset Tx on every failure.
|
first, or close/wait for the tone generator and reset Tx on every failure.
|
||||||
|
|
||||||
15. Gumble ffmpeg Pause can block forever
|
[x] 15. Gumble ffmpeg Pause can block forever
|
||||||
File: gumble/gumbleffmpeg/stream.go
|
File: gumble/gumbleffmpeg/stream.go
|
||||||
Pause checks StatePlaying, releases the lock, then sends on an unbuffered
|
Pause checks StatePlaying, releases the lock, then sends on an unbuffered
|
||||||
pause channel. If process exits in between, no receiver remains. Redesign
|
pause channel. If process exits in between, no receiver remains. Redesign
|
||||||
|
|||||||
@@ -57,7 +57,7 @@ func New(client *gumble.Client, source Source) *Stream {
|
|||||||
Volume: 1.0,
|
Volume: 1.0,
|
||||||
Source: source,
|
Source: source,
|
||||||
Command: "ffmpeg",
|
Command: "ffmpeg",
|
||||||
pause: make(chan struct{}),
|
pause: make(chan struct{}, 1),
|
||||||
state: StateInitial,
|
state: StateInitial,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -124,7 +124,12 @@ func (s *Stream) Pause() error {
|
|||||||
}
|
}
|
||||||
s.state = StatePaused
|
s.state = StatePaused
|
||||||
s.l.Unlock()
|
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
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -176,9 +181,12 @@ func (s *Stream) process() {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
int16Buffer := make([]int16, frameSize)
|
int16Buffer := make([]int16, frameSize)
|
||||||
|
s.l.Lock()
|
||||||
|
volume := s.Volume
|
||||||
|
s.l.Unlock()
|
||||||
for i := range int16Buffer {
|
for i := range int16Buffer {
|
||||||
float := float32(int16(binary.LittleEndian.Uint16(byteBuffer[i*2 : (i+1)*2])))
|
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))
|
atomic.AddInt64(&s.elapsed, int64(interval))
|
||||||
outgoing <- gumble.AudioBuffer(int16Buffer)
|
outgoing <- gumble.AudioBuffer(int16Buffer)
|
||||||
|
|||||||
@@ -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")
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user