revid: concurrency and testing

fixed a situation where a deadlock can occur and also found and fixed some issues while testing different initialisations on AudioDevice.

revid/audio-input.go

@@ -52,14 +52,15 @@ const (
 	stopped
 )
 
-var rates = [8]int{8000, 16000, 32000, 44100, 48000, 88200, 96000, 192000}
+// Rates contains the audio sample rates used by revid.
+var Rates = [8]int{8000, 16000, 32000, 44100, 48000, 88200, 96000, 192000}
 
 var log *logger.Logger
 
-// audioDevice holds everything we need to know about the audio input stream.
+// AudioDevice holds everything we need to know about the audio input stream.
 // Note: At 44100 Hz sample rate, 2 channels and 16-bit samples, a period of 5 seconds
 // results in PCM data chunks of 882000 bytes. A longer period exceeds datastore's 1MB blob limit.
-type audioDevice struct {
+type AudioDevice struct {
 	mu     sync.Mutex
 	source string // Name of audio source, or empty for the default source.
 	mode   uint8  // Operating mode, either running, paused, or stopped.
@@ -72,7 +73,7 @@ type audioDevice struct {
 	*AudioConfig
 }
 
-// AudioConfig provides parameters used by audioDevice.
+// AudioConfig provides parameters used by AudioDevice.
 type AudioConfig struct {
 	SampleRate int
 	Channels   int
@@ -81,8 +82,8 @@ type AudioConfig struct {
 	Codec      uint8
 }
 
-// NewAudioDevice initializes and returns an audioDevice struct which can be started, read from, and stopped.
+// NewAudioDevice initializes and returns an AudioDevice struct which can be started, read from, and stopped.
-func NewAudioDevice(cfg *AudioConfig) *audioDevice {
+func NewAudioDevice(cfg *AudioConfig) *AudioDevice {
 	// Initialize logger.
 	logLevel := int(logger.Debug)
 	validLogLevel := true
@@ -97,7 +98,7 @@ func NewAudioDevice(cfg *AudioConfig) *audioDevice {
 		log.Log(logger.Error, "Invalid log level was defaulted to Info")
 	}
 
-	a := &audioDevice{}
+	a := &AudioDevice{}
 	a.AudioConfig = cfg
 
 	// Open the requested audio device.
@@ -107,17 +108,28 @@ func NewAudioDevice(cfg *AudioConfig) *audioDevice {
 	}
 
 	// Setup ring buffer to capture audio in periods of a.RecPeriod seconds, and buffer rbDuration seconds in total.
-	a.ab = a.dev.NewBufferDuration(time.Second * time.Duration(a.RecPeriod))
+	a.ab = a.dev.NewBufferDuration(time.Duration(a.RecPeriod * float64(time.Second)))
-	a.chunkSize = (((len(a.ab.Data) / a.dev.BufferFormat().Channels) * a.Channels) / a.dev.BufferFormat().Rate) * a.SampleRate
+	cs := (float64((len(a.ab.Data)/a.dev.BufferFormat().Channels)*a.Channels) / float64(a.dev.BufferFormat().Rate)) * float64(a.SampleRate)
+	if cs < 1 {
+		log.Log(logger.Fatal, "given AudioConfig parameters are too small")
+	}
+	a.chunkSize = int(cs)
 	a.rb = ring.NewBuffer(rbLen, a.chunkSize, rbTimeout)
-
+	if a.rb == nil {
+		fmt.Println("NEW:", "rb: NIL", a.mode)
+		fmt.Println(rbLen, a.chunkSize, rbTimeout)
+		fmt.Println(len(a.ab.Data), a.dev.BufferFormat().Channels, a.Channels, a.dev.BufferFormat().Rate, a.SampleRate)
+	} else {
+		fmt.Println("NEW:", "rb: VALID", a.mode)
+	}
 	a.mode = paused
 
 	return a
 }
 
 // Start will start recording audio and writing to the output.
-func (a *audioDevice) Start() {
+func (a *AudioDevice) Start() {
+	fmt.Println("start lock")
 	a.mu.Lock()
 	switch a.mode {
 	case paused:
@@ -136,10 +148,12 @@ func (a *audioDevice) Start() {
 		return
 	}
 	a.mu.Unlock()
+	fmt.Println("start unlock")
 }
 
 // Stop will stop recording audio and close the device
-func (a *audioDevice) Stop() {
+func (a *AudioDevice) Stop() {
+	fmt.Println("stop lock")
 	a.mu.Lock()
 	if a.dev != nil {
 		log.Log(logger.Debug, "Closing", "source", a.source)
@@ -148,17 +162,18 @@ func (a *audioDevice) Stop() {
 	}
 	a.mode = stopped
 	a.mu.Unlock()
+	fmt.Println("stop unlock")
 
 }
 
-// ChunkSize returns the audioDevice's chunkSize, ie. the number of bytes of audio written to output at a time.
+// ChunkSize returns the AudioDevice's chunkSize, ie. the number of bytes of audio written to output at a time.
-func (a *audioDevice) ChunkSize() int {
+func (a *AudioDevice) ChunkSize() int {
 	return a.chunkSize
 }
 
 // open or re-open the recording device with the given name and prepare it to record.
 // If name is empty, the first recording device is used.
-func (a *audioDevice) open() error {
+func (a *AudioDevice) open() error {
 	if a.dev != nil {
 		log.Log(logger.Debug, "Closing", "source", a.source)
 		a.dev.Close()
@@ -211,17 +226,17 @@ func (a *audioDevice) open() error {
 	// so that it can be easily downsampled to the wanted rate.
 	// Note: if a card thinks it can record at a rate but can't actually, this can cause a failure. Eg.
 	// the audioinjector is supposed to record at 8000Hz and 16000Hz but it can't due to a firmware issue,
-	// to fix this 8000 and 16000 must be removed from the rates slice.
+	// to fix this 8000 and 16000 must be removed from the Rates slice.
 	foundRate := false
-	for i := 0; i < len(rates) && !foundRate; i++ {
+	for i := 0; i < len(Rates) && !foundRate; i++ {
-		if rates[i] < a.SampleRate {
+		if Rates[i] < a.SampleRate {
 			continue
 		}
-		if rates[i]%a.SampleRate == 0 {
+		if Rates[i]%a.SampleRate == 0 {
-			_, err = a.dev.NegotiateRate(rates[i])
+			_, err = a.dev.NegotiateRate(Rates[i])
 			if err == nil {
 				foundRate = true
-				log.Log(logger.Debug, "Sample rate set", "rate", rates[i])
+				log.Log(logger.Debug, "Sample rate set", "rate", Rates[i])
 			}
 		}
 	}
@@ -265,63 +280,88 @@ func (a *audioDevice) open() error {
 
 // input continously records audio and writes it to the ringbuffer.
 // Re-opens the device and tries again if ASLA returns an error.
-func (a *audioDevice) input() {
+func (a *AudioDevice) input() {
 	for {
 		a.mu.Lock()
+		fmt.Println("input lock")
+		if a.dev == nil {
+			fmt.Println("INPUT:", "dev: NIL", a.mode)
+		} else {
+			fmt.Println("INPUT:", "dev: VALID", a.mode)
+		}
+		if a.rb == nil {
+			fmt.Println("INPUT:", "rb: NIL", a.mode)
+		} else {
+			fmt.Println("INPUT:", "rb: VALID", a.mode)
+		}
 		switch a.mode {
 		case paused:
 			a.mu.Unlock()
+			fmt.Println("input unlock")
 			time.Sleep(time.Duration(a.RecPeriod) * time.Second)
 			continue
 		case stopped:
 			a.mu.Unlock()
+			fmt.Println("input unlock")
 			return
 		}
 		log.Log(logger.Debug, "Recording audio for period", "seconds", a.RecPeriod)
+		fmt.Println("LEN:", len(a.ab.Data))
 		err := a.dev.Read(a.ab.Data)
-		a.mu.Unlock()
+		fmt.Println("input read")
 		if err != nil {
 			log.Log(logger.Debug, "Device.Read failed", "error", err.Error())
-			a.mu.Lock()
 			err = a.open() // re-open
 			if err != nil {
+				a.mu.Unlock()
+				fmt.Println("input unlock")
 				log.Log(logger.Fatal, "alsa.open failed", "error", err.Error())
 			}
 			a.mu.Unlock()
+			fmt.Println("input unlock")
 			continue
 		}
 
 		toWrite := a.formatBuffer()
+		fmt.Println("input point")
 
 		log.Log(logger.Debug, "Audio format conversion has been performed where needed")
 
 		var n int
 		n, err = a.rb.Write(toWrite.Data)
+		fmt.Println("input write")
 		switch err {
 		case nil:
 			log.Log(logger.Debug, "Wrote audio to ringbuffer", "length", n)
 		case ring.ErrDropped:
 			log.Log(logger.Warning, "Dropped audio")
 		default:
+			a.mu.Unlock()
+			fmt.Println("input unlock")
 			log.Log(logger.Error, "Unexpected ringbuffer error", "error", err.Error())
 			return
 		}
+		a.mu.Unlock()
+		fmt.Println("input unlock")
 	}
 }
 
 // Read reads a full PCM chunk from the ringbuffer, returning the number of bytes read upon success.
 // Any errors returned are unexpected and should be considered fatal.
-func (a *audioDevice) Read(p []byte) (n int, err error) {
+func (a *AudioDevice) Read(p []byte) (n int, err error) {
+	fmt.Println("read lock")
 	a.mu.Lock()
-	if a.rb == nil {
-		fmt.Println("READ: RB IS NIL")
-	}
 	switch a.mode {
 	case paused:
 		return 0, nil
 	case stopped:
 		return 0, nil
 	}
+	if a.rb == nil {
+		fmt.Println("READ:", "NIL", a.mode)
+	} else {
+		fmt.Println("READ:", "VALID", a.mode)
+	}
 	chunk, err := a.rb.Next(rbNextTimeout)
 	switch err {
 	case nil:
@@ -342,17 +382,16 @@ func (a *audioDevice) Read(p []byte) (n int, err error) {
 	}
 	log.Log(logger.Debug, "Read audio from ringbuffer", "length", n)
 	a.mu.Unlock()
+	fmt.Println("read unlock")
 	return n, nil
 }
 
 // formatBuffer returns an ALSA buffer that has the recording data from the ac's original ALSA buffer but stored
 // in the desired format specified by the ac's parameters.
-func (a *audioDevice) formatBuffer() alsa.Buffer {
+func (a *AudioDevice) formatBuffer() alsa.Buffer {
 	var err error
-	a.mu.Lock()
 	wantChannels := a.Channels
 	wantRate := a.SampleRate
-	a.mu.Unlock()
 
 	// If nothing needs to be changed, return the original.
 	if a.ab.Format.Channels == wantChannels && a.ab.Format.Rate == wantRate {

revid/audio-input_test.go

@@ -43,12 +43,12 @@ func checkDevice(ac *AudioConfig) error {
 		return err
 	}
 	foundRate := false
-	for i := 0; i < len(rates) && !foundRate; i++ {
+	for i := 0; i < len(Rates) && !foundRate; i++ {
-		if rates[i] < ac.SampleRate {
+		if Rates[i] < ac.SampleRate {
 			continue
 		}
-		if rates[i]%ac.SampleRate == 0 {
+		if Rates[i]%ac.SampleRate == 0 {
-			_, err = testDev.NegotiateRate(rates[i])
+			_, err = testDev.NegotiateRate(Rates[i])
 			if err == nil {
 				foundRate = true
 			}
@@ -91,7 +91,7 @@ func TestAudio(t *testing.T) {
 	ac := &AudioConfig{
 		SampleRate: 8000,
 		Channels:   1,
-		RecPeriod:  0.01,
+		RecPeriod:  0.1,
 		BitDepth:   16,
 		Codec:      ADPCM,
 	}
@@ -104,10 +104,10 @@ func TestAudio(t *testing.T) {
 
 	// Create a new audioDevice, start, read/lex, and then stop it.
 	ai := NewAudioDevice(ac)
-	dst := bytes.NewBuffer(make([]byte, 0, ai.ChunkSize()*4))
+	dst := bytes.NewBuffer(make([]byte, 0))
 	ai.Start()
-	go lex.ADPCM(dst, ai, time.Duration(ac.RecPeriod), ai.ChunkSize())
+	go lex.ADPCM(dst, ai, time.Duration(ac.RecPeriod*float64(time.Second)), ai.ChunkSize())
-	time.Sleep(time.Millisecond * 10)
+	time.Sleep(time.Millisecond * 30)
 	ai.Stop()
 
 }

revid/revid.go

@@ -628,7 +628,7 @@ func (r *Revid) startAudioDevice() (func() error, error) {
 	}
 	ai := NewAudioDevice(ac)
 	r.wg.Add(1)
-	go r.processFrom(ai, time.Second/time.Duration(r.config.WriteRate), ai.ChunkSize())
+	go r.processFrom(ai, time.Duration(float64(time.Second)/r.config.WriteRate), ai.ChunkSize())
 	return func() error {
 		ai.Stop()
 		return nil