From cd63d0d95a53e8b338532239a1d50d29b4630cb8 Mon Sep 17 00:00:00 2001 From: Trek H Date: Wed, 13 Nov 2019 19:56:13 +1030 Subject: [PATCH] alsa: unexported ringbuffer chunksize Chunksize can be calculated without needing an ALSA device instance. Added a DataSize function to pcm package for calculating size of pcm data given relevant attributes. Removed ChunkSize from config revid config struct. Changed NewByteLexer to accept an in rather than a pointer. --- cmd/audio-netsender/main.go | 4 ++-- codec/codecutil/lex.go | 20 +++++++---------- codec/codecutil/lex_test.go | 11 ++++++++-- codec/pcm/pcm.go | 11 ++++++++++ device/alsa/alsa.go | 43 ++++++++++--------------------------- device/alsa/alsa_test.go | 5 +++-- revid/config/config.go | 1 - revid/revid.go | 10 +++++++-- 8 files changed, 52 insertions(+), 53 deletions(-) diff --git a/cmd/audio-netsender/main.go b/cmd/audio-netsender/main.go index ade701aa..4cc9d0c6 100644 --- a/cmd/audio-netsender/main.go +++ b/cmd/audio-netsender/main.go @@ -151,9 +151,9 @@ func main() { Data: ab.Data, } - recSize := (((len(ac.pb.Data) / ac.dev.BufferFormat().Channels) * ac.channels) / ac.dev.BufferFormat().Rate) * ac.rate + cs := pcm.DataSize(ac.parameters.rate, ac.parameters.channels, ac.parameters.bits, float64(ac.parameters.period), 0) rbLen := rbDuration / ac.period - ac.rb = ring.NewBuffer(rbLen, recSize, rbTimeout) + ac.rb = ring.NewBuffer(rbLen, cs, rbTimeout) go ac.input() diff --git a/codec/codecutil/lex.go b/codec/codecutil/lex.go index e1498b96..61e71d4a 100644 --- a/codec/codecutil/lex.go +++ b/codec/codecutil/lex.go @@ -32,12 +32,15 @@ import ( // ByteLexer is used to lex bytes using a buffer size which is configured upon construction. type ByteLexer struct { - bufSize *int + bufSize int } // NewByteLexer returns a pointer to a ByteLexer with the given buffer size. -func NewByteLexer(bufSize *int) *ByteLexer { - return &ByteLexer{bufSize: bufSize} +func NewByteLexer(s int) (*ByteLexer, error) { + if s <= 0 { + return nil, fmt.Errorf("invalid buffer size: %v", s) + } + return &ByteLexer{bufSize: s}, nil } // zeroTicks can be used to create an instant ticker. @@ -48,15 +51,8 @@ func init() { close(zeroTicks) } -// Lex reads *l.bufSize bytes from src and writes them to dst every d seconds. +// Lex reads l.bufSize bytes from src and writes them to dst every d seconds. func (l *ByteLexer) Lex(dst io.Writer, src io.Reader, d time.Duration) error { - if l.bufSize == nil { - return fmt.Errorf("buffer size has not been set") - } - bufSize := *l.bufSize - if bufSize <= 0 { - return fmt.Errorf("invalid buffer size: %v", bufSize) - } if d < 0 { return fmt.Errorf("invalid delay: %v", d) } @@ -69,7 +65,7 @@ func (l *ByteLexer) Lex(dst io.Writer, src io.Reader, d time.Duration) error { defer ticker.Stop() } - buf := make([]byte, bufSize) + buf := make([]byte, l.bufSize) for { <-ticker.C off, err := src.Read(buf) diff --git a/codec/codecutil/lex_test.go b/codec/codecutil/lex_test.go index 70fd3d39..642c2c23 100644 --- a/codec/codecutil/lex_test.go +++ b/codec/codecutil/lex_test.go @@ -51,8 +51,15 @@ func TestByteLexer(t *testing.T) { for i, tt := range lexTests { t.Run(strconv.Itoa(i), func(t *testing.T) { dst := bytes.NewBuffer([]byte{}) - l := NewByteLexer(&tt.n) - err := l.Lex(dst, bytes.NewReader(tt.data), tt.t) + l, err := NewByteLexer(tt.n) + if err != nil { + if tt.isValid { + t.Errorf("unexpected error: %v", err) + } else { + t.Skip() + } + } + err = l.Lex(dst, bytes.NewReader(tt.data), tt.t) if err != nil && err != io.EOF { if tt.isValid { t.Errorf("unexpected error: %v", err) diff --git a/codec/pcm/pcm.go b/codec/pcm/pcm.go index 9c0b8c6e..98619f81 100644 --- a/codec/pcm/pcm.go +++ b/codec/pcm/pcm.go @@ -32,6 +32,8 @@ import ( "encoding/binary" "fmt" + "bitbucket.org/ausocean/av/codec/adpcm" + "bitbucket.org/ausocean/av/codec/codecutil" "github.com/pkg/errors" ) @@ -65,6 +67,15 @@ type Buffer struct { Data []byte } +// DataSize takes audio attributes describing audio data and returns the size of that data. +func DataSize(rate, channels, bitDepth int, period float64, codec uint8) int { + s := int(float64(channels) * float64(rate) * float64(bitDepth/8) * period) + if codec == codecutil.ADPCM { + s = adpcm.EncBytes(s) + } + return s +} + // Resample takes Buffer c and resamples the pcm audio data to 'rate' Hz and returns a Buffer with the resampled data. // Notes: // - Currently only downsampling is implemented and c's rate must be divisible by 'rate' or an error will occur. diff --git a/device/alsa/alsa.go b/device/alsa/alsa.go index f31ea383..60d6e476 100644 --- a/device/alsa/alsa.go +++ b/device/alsa/alsa.go @@ -63,15 +63,14 @@ const ( // An ALSA device holds everything we need to know about the audio input stream and implements io.Reader and device.AVDevice. type ALSA struct { - l Logger // Logger for device's routines to log to. - mode uint8 // Operating mode, either running, paused, or stopped. - mu sync.Mutex // Provides synchronisation when changing modes concurrently. - title string // Name of audio title, or empty for the default title. - dev *yalsa.Device // ALSA device's Audio input device. - pb pcm.Buffer // Buffer to contain the direct audio from ALSA. - rb *ring.Buffer // Ring buffer to contain processed audio ready to be read. - chunkSize int // This is the number of bytes that will be stored in rb at a time. - Config // Configuration parameters for this device. + l Logger // Logger for device's routines to log to. + mode uint8 // Operating mode, either running, paused, or stopped. + mu sync.Mutex // Provides synchronisation when changing modes concurrently. + title string // Name of audio title, or empty for the default title. + dev *yalsa.Device // ALSA device's Audio input device. + pb pcm.Buffer // Buffer to contain the direct audio from ALSA. + rb *ring.Buffer // Ring buffer to contain processed audio ready to be read. + Config // Configuration parameters for this device. } // Config provides parameters used by the ALSA device. @@ -93,7 +92,7 @@ type Logger interface { // OpenError is used to determine whether an error has originated from attempting to open a device. type OpenError error -// NewALSA initializes and returns an ALSA device which has its logger set as the given logger. +// New initializes and returns an ALSA device which has its logger set as the given logger. func New(l Logger) *ALSA { return &ALSA{l: l} } // Set will take a Config struct, check the validity of the relevant fields @@ -149,24 +148,9 @@ func (d *ALSA) Set(c config.Config) error { Data: ab.Data, } - // Account for channel conversion. - chunkSize := float64(len(d.pb.Data) / d.dev.BufferFormat().Channels * d.Channels) - - // Account for resampling. - chunkSize = (chunkSize / float64(d.dev.BufferFormat().Rate)) * float64(d.SampleRate) - if chunkSize < 1 { - return errors.New("given Config parameters are too small") - } - - // Account for codec conversion. - if d.Codec == codecutil.ADPCM { - d.chunkSize = adpcm.EncBytes(int(chunkSize)) - } else { - d.chunkSize = int(chunkSize) - } - // Create ring buffer with appropriate chunk size. - d.rb = ring.NewBuffer(rbLen, d.chunkSize, rbTimeout) + cs := pcm.DataSize(d.SampleRate, d.Channels, d.BitDepth, d.RecPeriod, 0) + d.rb = ring.NewBuffer(rbLen, cs, rbTimeout) // Start device in paused mode. d.mode = paused @@ -206,11 +190,6 @@ func (d *ALSA) Stop() error { return nil } -// ChunkSize returns the number of bytes written to the ringbuffer per d.RecPeriod. -func (d *ALSA) ChunkSize() int { - return d.chunkSize -} - // validate checks if Config parameters are valid and returns an error if they are not. func validate(c *Config) error { if c.SampleRate <= 0 { diff --git a/device/alsa/alsa_test.go b/device/alsa/alsa_test.go index 08be27a4..44e54e38 100644 --- a/device/alsa/alsa_test.go +++ b/device/alsa/alsa_test.go @@ -32,6 +32,7 @@ import ( "time" "bitbucket.org/ausocean/av/codec/codecutil" + "bitbucket.org/ausocean/av/codec/pcm" "bitbucket.org/ausocean/av/revid/config" "bitbucket.org/ausocean/utils/logger" ) @@ -63,8 +64,8 @@ func TestDevice(t *testing.T) { if err != nil { t.Error(err) } - chunkSize := ai.ChunkSize() - lexer := codecutil.NewByteLexer(&chunkSize) + cs := pcm.DataSize(c.SampleRate, c.Channels, c.BitDepth, c.RecPeriod, c.InputCodec) + lexer, err := codecutil.NewByteLexer(cs) go lexer.Lex(ioutil.Discard, ai, time.Duration(c.RecPeriod*float64(time.Second))) time.Sleep(time.Duration(c.RecPeriod*float64(time.Second)) * time.Duration(n)) ai.Stop() diff --git a/revid/config/config.go b/revid/config/config.go index 88228867..1448c9ef 100644 --- a/revid/config/config.go +++ b/revid/config/config.go @@ -237,7 +237,6 @@ type Config struct { RecPeriod float64 // How many seconds to record at a time. Channels int // Number of audio channels, 1 for mono, 2 for stereo. BitDepth int // Sample bit depth. - ChunkSize int // ChunkSize is the size of the chunks in the audio.Device's ringbuffer. RTPAddress string // RTPAddress defines the RTP output destination. BurstPeriod uint // BurstPeriod defines the revid burst period in seconds. diff --git a/revid/revid.go b/revid/revid.go index 4885bc1b..63072975 100644 --- a/revid/revid.go +++ b/revid/revid.go @@ -42,6 +42,7 @@ import ( "bitbucket.org/ausocean/av/codec/h264" "bitbucket.org/ausocean/av/codec/h265" "bitbucket.org/ausocean/av/codec/mjpeg" + "bitbucket.org/ausocean/av/codec/pcm" "bitbucket.org/ausocean/av/container/flv" "bitbucket.org/ausocean/av/container/mts" "bitbucket.org/ausocean/av/device" @@ -363,8 +364,13 @@ func (r *Revid) setupPipeline(mtsEnc func(dst io.WriteCloser, rate float64) (io. } r.input = alsa.New(r.cfg.Logger) - cs := r.input.(*alsa.ALSA).ChunkSize() - r.lexTo = codecutil.NewByteLexer(&cs).Lex + + l, err := codecutil.NewByteLexer(pcm.DataSize(r.cfg.SampleRate, r.cfg.Channels, r.cfg.BitDepth, r.cfg.RecPeriod, r.cfg.InputCodec)) + if err != nil { + return err + } + r.lexTo = l.Lex + } // Configure the input device. We know that defaults are set, so no need to