Drop source frames during pause

Source reader should drop frames to catch up the latest frame.

.github/workflows/ci.yaml

@@ -30,15 +30,15 @@ jobs:
             libvpx-dev \
             libx264-dev
       - name: go vet
-        run: go vet -tags nolibopusfile ./...
+        run: go vet ./...
       - name: go build
-        run: go build -tags nolibopusfile ./...
+        run: go build ./...
       - name: go build without CGO
         run: go build . pkg/...
         env:
           CGO_ENABLED: 0
       - name: go test
-        run: go test -tags nolibopusfile ./... -v -race
+        run: go test ./... -v -race
       - name: go test without CGO
         run: go test . pkg/... -v
         env:

go.mod

@@ -5,9 +5,9 @@ go 1.13
 require (
 	github.com/blackjack/webcam v0.0.0-20200313125108-10ed912a8539
 	github.com/jfreymuth/pulse v0.0.0-20200817093420-a82ccdb5e8aa
-	github.com/lherman-cs/opus v0.0.0-20200925065115-26ea9d322d39
-	github.com/pion/webrtc/v2 v2.2.26
+	github.com/lherman-cs/opus v0.0.0-20200223204610-6a4b98199ea4
+	github.com/pion/webrtc/v2 v2.2.24
 	github.com/satori/go.uuid v1.2.0
-	golang.org/x/image v0.0.0-20200927104501-e162460cd6b5
+	golang.org/x/image v0.0.0-20200801110659-972c09e46d76
 	golang.org/x/sys v0.0.0-20200831180312-196b9ba8737a
 )

go.sum

@@ -22,8 +22,8 @@ github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORN
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
-github.com/lherman-cs/opus v0.0.0-20200925065115-26ea9d322d39 h1:WEYmSwg/uoPVmfmpXWPYplb1UUx/Jr4TXGNrPaI8Cj4=
-github.com/lherman-cs/opus v0.0.0-20200925065115-26ea9d322d39/go.mod h1:v9KQvlDYMuvlwniumBVMlrB0VHQvyTgxNvaXjPmTmps=
+github.com/lherman-cs/opus v0.0.0-20200223204610-6a4b98199ea4 h1:2ydMA2KbxRkYmIw3R8Me8dn90bejxBR4MKYXJ5THK3I=
+github.com/lherman-cs/opus v0.0.0-20200223204610-6a4b98199ea4/go.mod h1:v9KQvlDYMuvlwniumBVMlrB0VHQvyTgxNvaXjPmTmps=
 github.com/lucas-clemente/quic-go v0.7.1-0.20190401152353-907071221cf9 h1:tbuodUh2vuhOVZAdW3NEUvosFHUMJwUNl7jk/VSEiwc=
 github.com/lucas-clemente/quic-go v0.7.1-0.20190401152353-907071221cf9/go.mod h1:PpMmPfPKO9nKJ/psF49ESTAGQSdfXxlg1otPbEB2nOw=
 github.com/marten-seemann/qtls v0.2.3 h1:0yWJ43C62LsZt08vuQJDK1uC1czUc3FJeCLPoNAI4vA=
@@ -33,8 +33,8 @@ github.com/onsi/ginkgo v1.7.0 h1:WSHQ+IS43OoUrWtD1/bbclrwK8TTH5hzp+umCiuxHgs=
 github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/gomega v1.4.3 h1:RE1xgDvH7imwFD45h+u2SgIfERHlS2yNG4DObb5BSKU=
 github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
-github.com/pion/datachannel v1.4.21 h1:3ZvhNyfmxsAqltQrApLPQMhSFNA+aT87RqyCq4OXmf0=
-github.com/pion/datachannel v1.4.21/go.mod h1:oiNyP4gHx2DIwRzX/MFyH0Rz/Gz05OgBlayAI2hAWjg=
+github.com/pion/datachannel v1.4.20 h1:+uYUrxbhGuEt+9En81Necda5ul8M2h7mMsvGWkYZ/yI=
+github.com/pion/datachannel v1.4.20/go.mod h1:hsjWYdTW5fMmtM4hVIxUNYqViRPv2A6ixzkQFd82wSc=
 github.com/pion/dtls/v2 v2.0.1 h1:ddE7+V0faYRbyh4uPsRZ2vLdRrjVZn+wmCfI7jlBfaA=
 github.com/pion/dtls/v2 v2.0.1/go.mod h1:uMQkz2W0cSqY00xav7WByQ4Hb+18xeQh2oH2fRezr5U=
 github.com/pion/dtls/v2 v2.0.2 h1:FHCHTiM182Y8e15aFTiORroiATUI16ryHiQh8AIOJ1E=
@@ -54,8 +54,8 @@ github.com/pion/rtcp v1.2.3 h1:2wrhKnqgSz91Q5nzYTO07mQXztYPtxL8a0XOss4rJqA=
 github.com/pion/rtcp v1.2.3/go.mod h1:zGhIv0RPRF0Z1Wiij22pUt5W/c9fevqSzT4jje/oK7I=
 github.com/pion/rtp v1.6.0 h1:4Ssnl/T5W2LzxHj9ssYpGVEQh3YYhQFNVmSWO88MMwk=
 github.com/pion/rtp v1.6.0/go.mod h1:QgfogHsMBVE/RFNno467U/KBqfUywEH+HK+0rtnwsdI=
-github.com/pion/sctp v1.7.10 h1:o3p3/hZB5Cx12RMGyWmItevJtZ6o2cpuxaw6GOS4x+8=
-github.com/pion/sctp v1.7.10/go.mod h1:EhpTUQu1/lcK3xI+eriS6/96fWetHGCvBi9MSsnaBN0=
+github.com/pion/sctp v1.7.9 h1:n+A37cTMU08xL3Oodkz39XjtPReQliKyk01q96mGB5M=
+github.com/pion/sctp v1.7.9/go.mod h1:EhpTUQu1/lcK3xI+eriS6/96fWetHGCvBi9MSsnaBN0=
 github.com/pion/sdp/v2 v2.4.0 h1:luUtaETR5x2KNNpvEMv/r4Y+/kzImzbz4Lm1z8eQNQI=
 github.com/pion/sdp/v2 v2.4.0/go.mod h1:L2LxrOpSTJbAns244vfPChbciR/ReU1KWfG04OpkR7E=
 github.com/pion/srtp v1.5.1 h1:9Q3jAfslYZBt+C69SI/ZcONJh9049JUHZWYRRf5KEKw=
@@ -73,8 +73,8 @@ github.com/pion/turn/v2 v2.0.4 h1:oDguhEv2L/4rxwbL9clGLgtzQPjtuZwCdoM7Te8vQVk=
 github.com/pion/turn/v2 v2.0.4/go.mod h1:1812p4DcGVbYVBTiraUmP50XoKye++AMkbfp+N27mog=
 github.com/pion/udp v0.1.0 h1:uGxQsNyrqG3GLINv36Ff60covYmfrLoxzwnCsIYspXI=
 github.com/pion/udp v0.1.0/go.mod h1:BPELIjbwE9PRbd/zxI/KYBnbo7B6+oA6YuEaNE8lths=
-github.com/pion/webrtc/v2 v2.2.26 h1:01hWE26pL3LgqfxvQ1fr6O4ZtyRFFJmQEZK39pHWfFc=
-github.com/pion/webrtc/v2 v2.2.26/go.mod h1:XMZbZRNHyPDe1gzTIHFcQu02283YO45CbiwFgKvXnmc=
+github.com/pion/webrtc/v2 v2.2.24 h1:l7q/iO96tMTElxuE2XGdNhCzklGcd9aVZ00XufASp0g=
+github.com/pion/webrtc/v2 v2.2.24/go.mod h1:U/m+nvG1t8gInf8PwiDyJEDcd7qfl+jmGQXqTX2zGvo=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -98,8 +98,8 @@ golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 h1:psW17arqaxU48Z5kZ0CQnk
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200709230013-948cd5f35899 h1:DZhuSZLsGlFL4CmhA8BcRA0mnthyA/nZ00AqCUo7vHg=
 golang.org/x/crypto v0.0.0-20200709230013-948cd5f35899/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/image v0.0.0-20200927104501-e162460cd6b5 h1:QelT11PB4FXiDEXucrfNckHoFxwt8USGY1ajP1ZF5lM=
-golang.org/x/image v0.0.0-20200927104501-e162460cd6b5/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
+golang.org/x/image v0.0.0-20200801110659-972c09e46d76 h1:U7GPaoQyQmX+CBRWXKrvRzWTbd+slqeSh8uARsIyhAw=
+golang.org/x/image v0.0.0-20200801110659-972c09e46d76/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20191126235420-ef20fe5d7933/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=

pkg/io/broadcast.go

@@ -7,11 +7,11 @@ import (
 )
 
 const (
-	maskReading                = 1 << 63
-	defaultBroadcasterRingSize = 32
+	maskReading         = 1 << 63
+	broadcasterRingSize = 32
 	// TODO: If the data source has fps greater than 30, they'll see some
 	// 			 fps fluctuation. But, 30 fps should be enough for general cases.
-	defaultBroadcasterRingPollDuration = time.Millisecond * 33
+	broadcasterRingPollDuration = time.Millisecond * 33
 )
 
 var errEmptySource = fmt.Errorf("Source can't be nil")
@@ -23,16 +23,13 @@ type broadcasterData struct {
 }
 
 type broadcasterRing struct {
+	buffer []atomic.Value
 	// reading (1 bit) + reserved (31 bits) + data count (32 bits)
-	// IMPORTANT: state has to be the first element in struct, otherwise LoadUint64 will panic in 32 bits systems
-	//            due to unallignment
-	state        uint64
-	buffer       []atomic.Value
-	pollDuration time.Duration
+	state uint64
 }
 
-func newBroadcasterRing(size uint, pollDuration time.Duration) *broadcasterRing {
-	return &broadcasterRing{buffer: make([]atomic.Value, size), pollDuration: pollDuration}
+func newBroadcasterRing() *broadcasterRing {
+	return &broadcasterRing{buffer: make([]atomic.Value, broadcasterRingSize)}
 }
 
 func (ring *broadcasterRing) index(count uint32) int {
@@ -62,7 +59,7 @@ func (ring *broadcasterRing) get(count uint32) *broadcasterData {
 		// TODO: since it's lockless, it spends a lot of resources in the scheduling.
 		for atomic.LoadUint64(&ring.state) == reading {
 			// Yield current goroutine to let other goroutines to run instead
-			time.Sleep(ring.pollDuration)
+			time.Sleep(broadcasterRingPollDuration)
 		}
 
 		i := ring.index(count)
@@ -88,33 +85,10 @@ type Broadcaster struct {
 	buffer *broadcasterRing
 }
 
-// BroadcasterConfig is a config to control broadcaster behaviour
-type BroadcasterConfig struct {
-	// BufferSize configures the underlying ring buffer size that's being used
-	// to avoid data lost for late readers. The default value is 32.
-	BufferSize uint
-	// PollDuration configures the sleep duration in waiting for new data to come.
-	// The default value is 33 ms.
-	PollDuration time.Duration
-}
-
-// NewBroadcaster creates a new broadcaster. Source is expected to drop frames
-// when any of the readers is slower than the source.
-func NewBroadcaster(source Reader, config *BroadcasterConfig) *Broadcaster {
-	pollDuration := defaultBroadcasterRingPollDuration
-	var bufferSize uint = defaultBroadcasterRingSize
-	if config != nil {
-		if config.PollDuration != 0 {
-			pollDuration = config.PollDuration
-		}
-
-		if config.BufferSize != 0 {
-			bufferSize = config.BufferSize
-		}
-	}
-
+// NewNewBroadcaster creates a new broadcaster.
+func NewBroadcaster(source Reader) *Broadcaster {
 	var broadcaster Broadcaster
-	broadcaster.buffer = newBroadcasterRing(bufferSize, pollDuration)
+	broadcaster.buffer = newBroadcasterRing()
 	broadcaster.ReplaceSource(source)
 
 	return &broadcaster

pkg/io/video/broadcast.go

@@ -14,22 +14,11 @@ type Broadcaster struct {
 	ioBroadcaster *io.Broadcaster
 }
 
-type BroadcasterConfig struct {
-	Core *io.BroadcasterConfig
-}
-
-// NewBroadcaster creates a new broadcaster. Source is expected to drop frames
-// when any of the readers is slower than the source.
-func NewBroadcaster(source Reader, config *BroadcasterConfig) *Broadcaster {
-	var coreConfig *io.BroadcasterConfig
-
-	if config != nil {
-		coreConfig = config.Core
-	}
-
+// NewNewBroadcaster creates a new broadcaster.
+func NewBroadcaster(source Reader) *Broadcaster {
 	broadcaster := io.NewBroadcaster(io.ReaderFunc(func() (interface{}, error) {
 		return source.Read()
-	}), coreConfig)
+	}))
 
 	return &Broadcaster{broadcaster}
 }
@@ -65,7 +54,7 @@ func (broadcaster *Broadcaster) ReplaceSource(source Reader) error {
 	}))
 }
 
-// Source retrieves the underlying source. This operation is thread safe.
+// ReplaceSource retrieves the underlying source. This operation is thread safe.
 func (broadcaster *Broadcaster) Source() Reader {
 	source := broadcaster.ioBroadcaster.Source()
 	return ReaderFunc(func() (image.Image, error) {

pkg/io/video/broadcast_test.go

@@ -25,7 +25,7 @@ func BenchmarkBroadcast(b *testing.B) {
 
 		b.Run(fmt.Sprintf("Readers-%d", n), func(b *testing.B) {
 			b.SetParallelism(n)
-			broadcaster := NewBroadcaster(src, nil)
+			broadcaster := NewBroadcaster(src)
 			b.RunParallel(func(pb *testing.PB) {
 				reader := broadcaster.NewReader(false)
 				for pb.Next() {
@@ -109,7 +109,7 @@ func TestBroadcast(t *testing.T) {
 						frameSent++
 						return frame, nil
 					})
-					broadcaster := NewBroadcaster(src, nil)
+					broadcaster := NewBroadcaster(src)
 					var done uint32
 					duration := time.Second * 3
 					fpsChan := make(chan []float64)