revid: simplified audio device read write concurrency

revid/audio-input.go

@@ -62,7 +62,11 @@ type AudioDevice struct {
 	l      Logger
 	mu     sync.Mutex
 	source string // Name of audio source, or empty for the default source.
-	mode   uint8  // Operating mode, either running, paused, or stopped.
+	// Operating mode, either running, paused, or stopped.
+	// "running" means the input goroutine is reading from the ALSA device and writing to the ringbuffer.
+	// "paused" means the input routine is sleeping until unpaused or stopped.
+	// "stopped" means the input routine is stopped and the ALSA device is closed.
+	mode uint8
 
 	dev       *alsa.Device // Audio input device.
 	ab        alsa.Buffer  // ALSA's buffer.
@@ -82,7 +86,7 @@ type AudioConfig struct {
 }
 
 // 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, error) {
 	a := &AudioDevice{}
 	a.AudioConfig = cfg
 
@@ -103,89 +107,80 @@ func NewAudioDevice(cfg *AudioConfig) *AudioDevice {
 	// Open the requested audio device.
 	err := a.open()
 	if err != nil {
-		a.l.Log(logger.Fatal, "alsa.open failed", "error", err.Error())
+		a.l.Log(logger.Error, "failed to open audio device", "error", err.Error())
+		return nil, errors.New("failed to open audio device")
 	}
 
-	// Setup ring buffer to capture audio in periods of a.RecPeriod seconds, and buffer rbDuration seconds in total.
+	// Setup ring buffer to capture audio in periods of a.RecPeriod seconds and buffer rbDuration seconds in total.
 	a.ab = a.dev.NewBufferDuration(time.Duration(a.RecPeriod * float64(time.Second)))
 	cs := (float64((len(a.ab.Data)/a.dev.BufferFormat().Channels)*a.Channels) / float64(a.dev.BufferFormat().Rate)) * float64(a.SampleRate)
 	if cs < 1 {
-		a.l.Log(logger.Fatal, "given AudioConfig parameters are too small")
+		a.l.Log(logger.Error, "given AudioConfig parameters are too small", "error", err.Error())
+		return nil, errors.New("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
+	go a.input()
 
-	return a
+	return a, nil
 }
 
-// Start will start recording audio and writing to the output.
-func (a *AudioDevice) Start() {
-	fmt.Println("start lock")
+// Start will start recording audio and writing to the ringbuffer.
+func (a *AudioDevice) Start() error {
 	a.mu.Lock()
-	switch a.mode {
-	case paused:
-		// Start Recording
-		go a.input()
-		a.mode = running
-	case stopped:
-		// Open the audio device and start recording.
-		err := a.open()
-		if err != nil {
-			a.l.Log(logger.Fatal, "alsa.open failed", "error", err.Error())
-		}
-		go a.input()
-		a.mode = running
-	case running:
-		return
-	}
+	mode := a.mode
 	a.mu.Unlock()
-	fmt.Println("start unlock")
+	switch mode {
+	case paused:
+		a.mu.Lock()
+		a.mode = running
+		a.mu.Unlock()
+		return nil
+	case stopped:
+		// TODO(Trek): Make this reopen device and start recording.
+		return errors.New("device is stopped")
+	case running:
+		return nil
+	default:
+		return errors.New("invalid mode")
+	}
+	return nil
 }
 
-// Stop will stop recording audio and close the device
+// Stop will stop recording audio and close the device.
 func (a *AudioDevice) Stop() {
-	fmt.Println("stop lock")
 	a.mu.Lock()
-	if a.dev != nil {
-		a.l.Log(logger.Debug, "Closing", "source", a.source)
-		a.dev.Close()
-		a.dev = nil
-	}
 	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 number of bytes written to the ringbuffer per a.RecPeriod.
 func (a *AudioDevice) ChunkSize() int {
 	return a.chunkSize
 }
 
-// open or re-open the recording device with the given name and prepare it to record.
+// 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 {
+	// Close any existing device.
 	if a.dev != nil {
-		a.l.Log(logger.Debug, "Closing", "source", a.source)
+		a.l.Log(logger.Debug, "closing device", "source", a.source)
 		a.dev.Close()
 		a.dev = nil
 	}
-	a.l.Log(logger.Debug, "Opening", "source", a.source)
 
+	// Open sound card and open recording device.
+	a.l.Log(logger.Debug, "opening sound card")
 	cards, err := alsa.OpenCards()
 	if err != nil {
+		a.l.Log(logger.Debug, "failed to open sound card")
 		return err
 	}
 	defer alsa.CloseCards(cards)
 
+	a.l.Log(logger.Debug, "finding audio device")
 	for _, card := range cards {
 		devices, err := card.Devices()
 		if err != nil {
@@ -201,18 +196,17 @@ func (a *AudioDevice) open() error {
 			}
 		}
 	}
-
 	if a.dev == nil {
-		return errors.New("No audio source found")
+		a.l.Log(logger.Debug, "failed to find audio device")
+		return errors.New("no audio device found")
 	}
-	a.l.Log(logger.Debug, "Found audio source", "source", a.dev.Title)
 
-	// ToDo: time out if Open takes too long.
+	a.l.Log(logger.Debug, "opening audio device", "source", a.dev.Title)
 	err = a.dev.Open()
 	if err != nil {
+		a.l.Log(logger.Debug, "failed to open audio device")
 		return err
 	}
-	a.l.Log(logger.Debug, "Opened audio source")
 
 	// 2 channels is what most devices need to record in. If mono is requested,
 	// the recording will be converted in formatBuffer().
@@ -223,9 +217,9 @@ func (a *AudioDevice) open() error {
 
 	// Try to negotiate a rate to record in that is divisible by the wanted rate
 	// 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.
+	// Note: if a card thinks it can record at a rate but can't actually, this can cause a failure.
+	// Eg. the audioinjector sound card is supposed to record at 8000Hz and 16000Hz but it can't due to a firmware issue,
+	// a fix for this is to remove 8000 and 16000 from the Rates slice.
 	foundRate := false
 	for i := 0; i < len(Rates) && !foundRate; i++ {
 		if Rates[i] < a.SampleRate {
@@ -281,107 +275,77 @@ func (a *AudioDevice) open() error {
 // Re-opens the device and tries again if ASLA returns an error.
 func (a *AudioDevice) input() {
 	for {
+		// Check mode.
 		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 {
+		mode := a.mode
+		a.mu.Unlock()
+		switch 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")
+			if a.dev != nil {
+				a.l.Log(logger.Debug, "closing audio device", "source", a.source)
+				a.dev.Close()
+				a.dev = nil
+			}
 			return
 		}
-		a.l.Log(logger.Debug, "Recording audio for period", "seconds", a.RecPeriod)
-		fmt.Println("LEN:", len(a.ab.Data))
+
+		// Read from audio device.
+		a.l.Log(logger.Debug, "recording audio for period", "seconds", a.RecPeriod)
 		err := a.dev.Read(a.ab.Data)
-		fmt.Println("input read")
 		if err != nil {
-			a.l.Log(logger.Debug, "Device.Read failed", "error", err.Error())
+			a.l.Log(logger.Debug, "read failed", "error", err.Error())
 			err = a.open() // re-open
 			if err != nil {
-				a.mu.Unlock()
-				fmt.Println("input unlock")
-				a.l.Log(logger.Fatal, "alsa.open failed", "error", err.Error())
+				a.l.Log(logger.Fatal, "reopening device failed", "error", err.Error())
+				return
 			}
-			a.mu.Unlock()
-			fmt.Println("input unlock")
 			continue
 		}
 
+		// Process audio.
+		a.l.Log(logger.Debug, "processing audio")
 		toWrite := a.formatBuffer()
-		fmt.Println("input point")
 
-		a.l.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")
+		// Write audio to ringbuffer.
+		n, err := a.rb.Write(toWrite.Data)
 		switch err {
 		case nil:
-			a.l.Log(logger.Debug, "Wrote audio to ringbuffer", "length", n)
+			a.l.Log(logger.Debug, "wrote audio to ringbuffer", "length", n)
 		case ring.ErrDropped:
-			a.l.Log(logger.Warning, "Dropped audio")
+			a.l.Log(logger.Warning, "old audio data overwritten")
 		default:
-			a.mu.Unlock()
-			fmt.Println("input unlock")
-			a.l.Log(logger.Error, "Unexpected ringbuffer error", "error", err.Error())
+			a.l.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) {
-	fmt.Println("read lock")
-	a.mu.Lock()
-	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)
+	// Ready ringbuffer for read.
+	_, err = a.rb.Next(rbNextTimeout)
 	switch err {
 	case nil:
-		// Do nothing.
 	case ring.ErrTimeout:
 		return 0, nil
-	case io.EOF:
-		a.l.Log(logger.Error, "Unexpected EOF from ring.Next")
-		return 0, io.ErrUnexpectedEOF
 	default:
-		a.l.Log(logger.Error, "Unexpected error from ring.Next", "error", err.Error())
 		return 0, err
 	}
-	n, err = io.ReadFull(a.rb, p[:chunk.Len()])
-	if err != nil {
-		a.l.Log(logger.Error, "Unexpected error from ring.Read", "error", err.Error())
-		return n, err
+
+	// Read from ring buffer.
+	n, err = a.rb.Read(p)
+	switch err {
+	case nil:
+	case io.EOF:
+		return 0, nil
+	default:
+		return 0, err
 	}
-	a.l.Log(logger.Debug, "Read audio from ringbuffer", "length", n)
-	a.mu.Unlock()
-	fmt.Println("read unlock")
+
 	return n, nil
 }
 
@@ -402,9 +366,11 @@ func (a *AudioDevice) formatBuffer() alsa.Buffer {
 	if a.ab.Format.Channels != wantChannels {
 
 		// Convert channels.
+		// TODO(Trek): Make this work for conversions other than stereo to mono.
 		if a.ab.Format.Channels == 2 && wantChannels == 1 {
-			if formatted.Data, err = pcm.StereoToMono(a.ab); err != nil {
-				a.l.Log(logger.Warning, "Channel conversion failed, audio has remained stereo", "error", err.Error())
+			formatted.Data, err = pcm.StereoToMono(a.ab)
+			if err != nil {
+				a.l.Log(logger.Warning, "channel conversion failed, audio has remained stereo", "error", err.Error())
 			} else {
 				formatted.Format.Channels = 1
 			}
@@ -421,7 +387,7 @@ func (a *AudioDevice) formatBuffer() alsa.Buffer {
 			formatted.Data, err = pcm.Resample(a.ab, wantRate)
 		}
 		if err != nil {
-			a.l.Log(logger.Warning, "Rate conversion failed, audio has remained original rate", "error", err.Error())
+			a.l.Log(logger.Warning, "rate conversion failed, audio has remained original rate", "error", err.Error())
 		} else {
 			formatted.Format.Rate = wantRate
 		}

revid/audio-input_test.go

@@ -3,6 +3,7 @@ package revid
 import (
 	"bytes"
 	"errors"
+	"io/ioutil"
 	"testing"
 	"time"
 
@@ -91,7 +92,7 @@ func TestAudio(t *testing.T) {
 	ac := &AudioConfig{
 		SampleRate: 8000,
 		Channels:   1,
-		RecPeriod:  0.1,
+		RecPeriod:  1,
 		BitDepth:   16,
 		Codec:      ADPCM,
 	}
@@ -103,11 +104,18 @@ func TestAudio(t *testing.T) {
 	}
 
 	// Create a new audioDevice, start, read/lex, and then stop it.
-	ai := NewAudioDevice(ac)
+	ai, err := NewAudioDevice(ac)
+	if err != nil {
+		t.Error(err)
+	}
 	dst := bytes.NewBuffer(make([]byte, 0))
-	ai.Start()
+	err = ai.Start()
+	if err != nil {
+		t.Error(err)
+	}
 	num := 3 // How many 'ac.RecPeriod's to record.
 	go lex.ADPCM(dst, ai, time.Duration(ac.RecPeriod*float64(time.Second)), ai.ChunkSize())
-	time.Sleep(time.Millisecond * 100 * time.Duration(num))
+	time.Sleep(time.Millisecond * 1000 * time.Duration(num))
 	ai.Stop()
+	err = ioutil.WriteFile("./testout", dst.Bytes(), 0644)
 }

revid/revid.go

@@ -619,6 +619,7 @@ func (r *Revid) setupInputForFile() (func() error, error) {
 
 // startAudioDevice is used to start capturing audio from an audio device and processing it.
 func (r *Revid) startAudioDevice() (func() error, error) {
+	// Create audio device.
 	ac := &AudioConfig{
 		SampleRate: r.config.SampleRate,
 		Channels:   r.config.Channels,
@@ -626,7 +627,18 @@ func (r *Revid) startAudioDevice() (func() error, error) {
 		BitDepth:   r.config.BitDepth,
 		Codec:      r.config.InputCodec,
 	}
-	ai := NewAudioDevice(ac)
+	ai, err := NewAudioDevice(ac)
+	if err != nil {
+		r.config.Logger.Log(logger.Fatal, pkg+"failed to create audio device", "error", err.Error())
+	}
+
+	// Start audio device
+	err = ai.Start()
+	if err != nil {
+		r.config.Logger.Log(logger.Fatal, pkg+"failed to start audio device", "error", err.Error())
+	}
+
+	// Process output from audio device.
 	r.wg.Add(1)
 	go r.processFrom(ai, time.Duration(float64(time.Second)/r.config.WriteRate), ai.ChunkSize())
 	return func() error {