Fix some race conditions

Author: mengelbart

receiver/receiver.go

@@ -16,6 +16,7 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
+	"sync/atomic"
 	"time"
 
 	"github.com/pion/interceptor"
@@ -181,7 +182,7 @@ type trackInfo struct {
 }
 
 type trackStats struct {
-	rtpPacketsReceived int
+	rtpPacketsReceived atomic.Int64
 	framesAssembled    int
 	keyframesReceived  int
 	startTime          time.Time
@@ -297,7 +298,7 @@ func (r *Receiver) startStatsGoroutine(ctx context.Context, bytesReceivedChan ch
 				rate := bits / delta.Seconds()
 				mBitPerSecond := rate / float64(vnet.MBit)
 				r.log.Infof("throughput: %.2f Mb/s | RTP packets: %d | Frames: %d | Keyframes: %d",
-					mBitPerSecond, stats.rtpPacketsReceived, stats.framesAssembled, stats.keyframesReceived)
+					mBitPerSecond, stats.rtpPacketsReceived.Load(), stats.framesAssembled, stats.keyframesReceived)
 				bytesReceived = 0
 				last = now
 			case newBytesReceived := <-bytesReceivedChan:
@@ -351,7 +352,7 @@ func (r *Receiver) processPackets(ctx context.Context, trackRemote *webrtc.Track
 			}
 
 			bytesReceivedChan <- packet.MarshalSize()
-			stats.rtpPacketsReceived++
+			stats.rtpPacketsReceived.Add(1)
 
 			r.processVP8Packet(packet, trackInfo, frameAssembler, videoWidth, videoHeight, stats)
 		}

syncodec/statistical_codec.go

@@ -225,6 +225,9 @@ func (c *StatisticalCodec) GetTargetBitrate() int {
 // greater than c.rMax, bitrate will be set to c.rMax. If r is lower than
 // c.rMin, bitrate will be set to c.rMin.
 func (c *StatisticalCodec) SetTargetBitrate(r int) {
+	c.targetBitrateLock.Lock()
+	defer c.targetBitrateLock.Unlock()
+
 	if r < c.targetBitrateBps {
 		c.targetBitrateBps = maximum(r, c.rMin)
 
@@ -244,10 +247,14 @@ func (c *StatisticalCodec) nextFrame() Frame {
 		}
 	}
 
-	bytesPerFrame := c.targetBitrateBps / (8.0 * c.fps)
+	c.targetBitrateLock.Lock()
+	bps := c.targetBitrateBps
+	c.targetBitrateLock.Unlock()
+
+	bytesPerFrame := bps / (8.0 * c.fps)
 
 	if c.remainingBurstFrames > 0 {
-		size := (c.targetBitrateBps * c.burstFrameCount) / (c.burstFrameSize + (c.burstFrameCount - 1))
+		size := (bps * c.burstFrameCount) / (c.burstFrameSize + (c.burstFrameCount - 1))
 
 		return Frame{
 			Content:  make([]byte, size),

vnet/main.go

@@ -162,12 +162,7 @@ func (r *Runner) runVariableAvailableCapacitySingleFlow() error {
 	if err != nil {
 		return fmt.Errorf("setup simple flow: %w", err)
 	}
-	defer func(flow Flow) {
-		err = flow.Close()
-		if err != nil {
-			r.logger.Errorf("flow close: %v", err)
-		}
-	}(flow)
+	defer flow.Close()
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()