Feature/audio sync

README.MD

@@ -51,12 +51,12 @@ go build
 
 ### Dependencies 
 
-You need to have portaudio-dev, pulseaudio and lib-fdk-aac and go runtime installed to build this program
+You need to have portaudio-dev, pulseaudio, libsamplerate and lib-fdk-aac and go runtime installed to build this program
 
 #### Ubuntu 21.4 build
 
 ````shell
-sudo apt install golang-go libfdk-aac-dev
+sudo apt install golang-go libfdk-aac-dev libsamplerate-dev
 ````
 
 #### Rasbpian (buster)
@@ -71,12 +71,27 @@ apt-get install libfdk-aac-dev
 
 You also need to install go from https://golang.org/dl/ as version 1.16 is not supplied by buster (they only support 1.11)
 
+##### Raspberry 1 (arm6l)
+
+You can not rely on deb-multimedia, they only ship armv7 binary
+
+You will have to compile manually lib-fdk-aac
+
+````
+git clone --depth 1 https://github.com/mstorsjo/fdk-aac.git ~/ffmpeg-libraries/fdk-aac \
+  && cd ~/ffmpeg-libraries/fdk-aac \
+  && autoreconf -fiv \
+  && ./configure \
+  && make -j$(nproc) \
+  && sudo make install
+````
+
 #### Mac os build 
 
 * Having xcode properly installed (clang needed)
 
 ````shell
-brew install portaudio fdk-aac go 
+brew install portaudio fdk-aac libsamplerate go 
 ````
 
 ### Docker image
@@ -131,8 +146,6 @@ docker run \
 
 Note that mDNS by design will only work with networking mode "host" (recommended for beginners) or (mac/ip)vlan.
 
-#### Acknowledgments  
-
 ## Run
 
 - goplay2 by default run only on the ipv4 interface (because [this issue](https://github.com/golang/go/issues/31024) on ipv6 parsing) 
@@ -161,10 +174,16 @@ Ex: It takes around 60ms on my mac to launch the audio stream at the **Anchor Ti
 
 `sink` (pulse audio sink name) to replace default sink
 
+`nosync` (No Audio sync) disable experimental audio sync
+
 Example : 
 ```shell
-./goplay2 -sink alsa_output -i en0 -n aiwa
+./goplay2 -sink alsa_output -i en0 -n aiwa -nosync
 ```
 
+#### Misc
+
+patch used for qemu until it is merged
+
 
 By [AlbanSeurat](https://github.com/AlbanSeurat)

audio/clock.go

@@ -1,40 +1,57 @@
 package audio
 
 import (
+	"goplay2/codec"
 	"goplay2/ptp"
 	"time"
 )
 
 type Clock struct {
-	virtualClock   *ptp.VirtualClock
-	realAnchorTime time.Time
-	anchorRtpTime  int64
-	currentRtpTime int64
+	networkClock *ptp.VirtualClock
+	// audio times
+	realAudioAnchorTime time.Time
+	audioAnchorTime     time.Duration
+	// network times
+	realNetworkAnchorTime time.Time
+	anchorRtpTime         int64
+	networkAnchorTime     int64
 }
 
-// AnchorTime setup anchor time using the real clock
+func NewClock(networkClock *ptp.VirtualClock) *Clock {
+	return &Clock{
+		networkClock: networkClock,
+	}
+}
+
+func (c *Clock) virtualToRealTime(elapsed int64) time.Time {
+	return time.Unix(0, time.Now().UnixNano()+(elapsed-c.networkClock.Now().UnixNano()))
+}
+
+// SetAnchorTime setup anchor time using the real clock
 // virtualAnchorTime - network Time (using PTP) of the anchorTime
 // rtpTime Timestamp - monotonic counter of frames
-func (c *Clock) AnchorTime(virtualAnchorTime int64, rtpTime int64) {
-
+func (c *Clock) SetAnchorTime(virtualAnchorTime int64, rtpTime int64) {
 	c.anchorRtpTime = rtpTime
-	c.currentRtpTime = rtpTime
-
-	realNowTime := time.Now()
-	virtualNowTime := c.virtualClock.Now()
-
-	c.realAnchorTime = time.Unix(0, realNowTime.UnixNano()+(virtualAnchorTime-virtualNowTime.UnixNano()))
+	c.networkAnchorTime = virtualAnchorTime
+	c.realNetworkAnchorTime = c.virtualToRealTime(virtualAnchorTime)
 }
 
 // PacketTime returns the real clock time at which time a RTP packet should be played
 func (c *Clock) PacketTime(frameRtpTime int64) time.Time {
-	return time.Unix(0, 0)
+	elapsedTime := float64(frameRtpTime-c.anchorRtpTime) * 1.0 / float64(codec.SampleRate) * 1e9 // convert to nanoseconds
+	return c.virtualToRealTime(c.networkAnchorTime + int64(elapsedTime))
+}
+
+func (c *Clock) AudioTime(anchorAudioTime time.Duration, realAnchorTime time.Time) {
+	c.audioAnchorTime = anchorAudioTime
+	c.realAudioAnchorTime = realAnchorTime
 }
 
-func (c *Clock) CurrentRtpTime() int64 {
-	return c.currentRtpTime
+func (c *Clock) PlayTime(nowAudioTime time.Duration, playbackTime time.Duration) time.Time {
+	now := time.Now()
+	return now.Add(playbackTime - nowAudioTime)
 }
 
-func (c *Clock) IncRtpTime() {
-	c.currentRtpTime += 1024
+func (c *Clock) AnchorTime() time.Time {
+	return c.realNetworkAnchorTime
 }

audio/filter.go

@@ -0,0 +1,8 @@
+package audio
+
+import "time"
+
+type Filter interface {
+	Reset(*Clock)
+	Apply(audioStream Stream, samples []int16, playTime time.Time, sequence uint32, startTs uint32) (int, error)
+}

audio/player.go

@@ -1,11 +1,10 @@
 package audio
 
 import (
-	"bytes"
-	"encoding/binary"
-	codec2 "goplay2/codec"
+	"goplay2/codec"
+	"goplay2/config"
 	"goplay2/globals"
-	"goplay2/ptp"
+	"goplay2/rtp"
 	"time"
 )
 
@@ -19,32 +18,53 @@ const (
 type Player struct {
 	ControlChannel chan globals.ControlMessage
 	clock          *Clock
+	filter         Filter
 	Status         PlaybackStatus
-	stream         codec2.Stream
-	ringBuffer     *Ring
+	stream         codec.Stream
+	ring           *Ring
+	aacDecoder     *codec.AacDecoder
+	untilSeq       uint32
 }
 
-func NewPlayer(clock *ptp.VirtualClock, ring *Ring) *Player {
-
-	return &Player{
-		clock:          &Clock{clock, time.Now(), 0, 0},
+func NewPlayer(audioClock *Clock, filter Filter) *Player {
+	aacDecoder := codec.NewAacDecoder()
+	asc := []byte{0x12, 0x10}
+	if err := aacDecoder.InitRaw(asc); err != nil {
+		globals.ErrLog.Panicf("init decoder failed, err is %s", err)
+	}
+	player := &Player{
+		clock:          audioClock,
+		filter:         filter,
 		ControlChannel: make(chan globals.ControlMessage, 100),
-		stream:         codec2.NewStream(),
+		aacDecoder:     aacDecoder,
+		stream:         codec.NewStream(config.Config.Volume),
 		Status:         STOPPED,
-		ringBuffer:     ring,
+		ring:           New(globals.BufferSize / 2048),
+		untilSeq:       0,
 	}
+	return player
 }
 
-func (p *Player) callBack(out []int16, currentTime time.Duration, outputBufferDacTime time.Duration) {
-	frame, err := p.ringBuffer.TryPop()
+func (p *Player) audioSync(reader Stream, samples []int16, nextTime time.Time, sequence uint32, startTs uint32) (int, error) {
+	if sequence <= p.untilSeq {
+		return 0, nil
+	}
+	if config.Config.DisableAudioSync {
+		return reader.Read(samples)
+	} else {
+		return p.filter.Apply(reader, samples, nextTime, sequence, startTs)
+	}
+}
+
+func (p *Player) callBack(out []int16, currentTime time.Duration, outputBufferDacTime time.Duration) int {
+	playTime := p.clock.PlayTime(currentTime, outputBufferDacTime)
+	size, err := p.ring.TryRead(out, playTime, p.audioSync)
 	if err == ErrIsEmpty {
 		p.fillSilence(out)
-	} else {
-		err = binary.Read(bytes.NewReader(frame.(*PCMFrame).pcmData), binary.LittleEndian, out)
-		if err != nil {
-			globals.ErrLog.Printf("error reading data : %v\n", err)
-		}
+	} else if size < len(out) {
+		p.fillSilence(out[size:])
 	}
+	return p.stream.FilterVolume(out)
 }
 
 func (p *Player) Run() {
@@ -53,10 +73,20 @@ func (p *Player) Run() {
 		globals.ErrLog.Fatalln("Audio Stream init error:", err)
 	}
 	defer p.stream.Close()
+	p.clock.AudioTime(p.stream.AudioTime(), time.Now())
 	for {
 		select {
 		case msg := <-p.ControlChannel:
 			switch msg.MType {
+			case globals.STOP:
+				if p.Status == PLAYING {
+					if err := p.stream.Stop(); err != nil {
+						globals.ErrLog.Printf("error pausing audio :%v\n", err)
+						return
+					}
+				}
+				p.Reset()
+				p.Status = STOPPED
 			case globals.PAUSE:
 				if p.Status == PLAYING {
 					if err := p.stream.Stop(); err != nil {
@@ -74,7 +104,8 @@ func (p *Player) Run() {
 					}
 				}
 				p.Status = PLAYING
-				p.clock.AnchorTime(msg.Param1, msg.Param2)
+				p.clock.SetAnchorTime(msg.Param1, msg.Param2)
+				p.filter.Reset(p.clock)
 			case globals.SKIP:
 				p.skipUntil(msg.Param1, msg.Param2)
 			case globals.VOLUME:
@@ -86,23 +117,30 @@ func (p *Player) Run() {
 	}
 }
 
-func (p *Player) skipUntil(fromSeq int64, UntilSeq int64) {
-	p.ringBuffer.Flush(func(value interface{}) bool {
-		frame := value.(*PCMFrame)
-		return frame.SequenceNumber < uint32(fromSeq) || frame.SequenceNumber > uint32(UntilSeq)
+func (p *Player) skipUntil(fromSeq int64, untilSeq int64) {
+	// TODO : use also timestamp to have better precision
+	p.ring.Filter(func(sequence uint32, startTs uint32) bool {
+		return sequence > uint32(fromSeq) && sequence < uint32(untilSeq)
 	})
+	// some data are possibly not yet in the buffer - reader should skip them afterwards (during async callback)
+	p.untilSeq = uint32(untilSeq)
 }
 
-func (p *Player) Push(frame interface{}) {
-	p.ringBuffer.Push(frame)
+func (p *Player) Push(frame *rtp.Frame) {
+	var pcmBuffer = make([]int16, 2048)
+	_, err := frame.PcmData(p.aacDecoder, pcmBuffer)
+	if err != nil {
+		globals.ErrLog.Printf("error decoding the packet %v", err)
+	}
+	p.ring.Write(pcmBuffer, frame.SequenceNumber, frame.Timestamp)
 }
 
 func (p *Player) Reset() {
-	p.ringBuffer.Reset()
+	p.ring.Reset()
+	p.untilSeq = 0
 }
 
 func (p *Player) fillSilence(out []int16) {
-	globals.ErrLog.Printf("warning : filling audio buffer with silence")
 	for i := range out {
 		out[i] = 0
 	}