diff --git a/internal/core/path.go b/internal/core/path.go index 3bfdbfd3..a429a7b8 100644 --- a/internal/core/path.go +++ b/internal/core/path.go @@ -89,7 +89,7 @@ type path struct { ctx context.Context ctxCancel func() - pendingRequests *int64 + pendingRequests atomic.Int64 confMutex sync.RWMutex source defs.Source stream *stream.Stream @@ -129,7 +129,6 @@ func (pa *path) initialize() { pa.confName = pa.conf.Name pa.ctx = ctx pa.ctxCancel = ctxCancel - pa.pendingRequests = new(int64) pa.readers = make(map[defs.Reader]struct{}) pa.onDemandStaticSourceReadyTimer = emptyTimer() pa.onDemandStaticSourceCloseTimer = emptyTimer() @@ -304,7 +303,7 @@ func (pa *path) runInner() error { case req := <-pa.chDescribe: pa.doDescribe(req) - atomic.AddInt64(pa.pendingRequests, -1) + pa.pendingRequests.Add(-1) if pa.shouldClose() { pa.parent.closePathIfIdle(pa) @@ -313,7 +312,7 @@ func (pa *path) runInner() error { case req := <-pa.chAddPublisher: pa.doAddPublisher(req) - atomic.AddInt64(pa.pendingRequests, -1) + pa.pendingRequests.Add(-1) if pa.shouldClose() { pa.parent.closePathIfIdle(pa) @@ -329,7 +328,7 @@ func (pa *path) runInner() error { case req := <-pa.chAddReader: pa.doAddReader(req) - atomic.AddInt64(pa.pendingRequests, -1) + pa.pendingRequests.Add(-1) if pa.shouldClose() { pa.parent.closePathIfIdle(pa) diff --git a/internal/core/path_manager.go b/internal/core/path_manager.go index 459c006c..2f10e97f 100644 --- a/internal/core/path_manager.go +++ b/internal/core/path_manager.go @@ -6,7 +6,6 @@ import ( "maps" "sort" "sync" - "sync/atomic" "github.com/bluenviron/mediamtx/internal/auth" "github.com/bluenviron/mediamtx/internal/conf" @@ -174,7 +173,7 @@ outer: } case pa := <-pm.chClosePathIfIdle: - if atomic.LoadInt64(pa.pendingRequests) == 0 { + if pa.pendingRequests.Load() == 0 { pm.doClosePath(pa) } @@ -357,7 +356,7 @@ func (pm *pathManager) doDescribe(req defs.PathDescribeReq) { pa := pm.paths[req.AccessRequest.Name] - atomic.AddInt64(pa.pendingRequests, 1) + pa.pendingRequests.Add(1) req.Res <- defs.PathDescribeRes{Path: pa} } @@ -387,7 +386,7 @@ func (pm *pathManager) doAddReader(req defs.PathAddReaderReq) { pa := pm.paths[req.AccessRequest.Name] - atomic.AddInt64(pa.pendingRequests, 1) + pa.pendingRequests.Add(1) req.Res <- defs.PathAddReaderRes{ Path: pa, @@ -425,7 +424,7 @@ func (pm *pathManager) doAddPublisher(req defs.PathAddPublisherReq) { pa := pm.paths[req.AccessRequest.Name] - atomic.AddInt64(pa.pendingRequests, 1) + pa.pendingRequests.Add(1) req.Res <- defs.PathAddPublisherRes{ Path: pa, diff --git a/internal/protocols/webrtc/peer_connection.go b/internal/protocols/webrtc/peer_connection.go index 28aecd87..c3ebc82f 100644 --- a/internal/protocols/webrtc/peer_connection.go +++ b/internal/protocols/webrtc/peer_connection.go @@ -168,7 +168,7 @@ type PeerConnection struct { wr *webrtc.PeerConnection ctx context.Context ctxCancel context.CancelFunc - readingStarted *int64 + readingStarted atomic.Int64 incomingTracks []*IncomingTrack statsInterceptor *statsInterceptor @@ -321,8 +321,6 @@ func (co *PeerConnection) Start() error { co.ctx, co.ctxCancel = context.WithCancel(context.Background()) - co.readingStarted = new(int64) - co.newLocalCandidate = make(chan *webrtc.ICECandidateInit) co.connected = make(chan struct{}) co.failed = make(chan struct{}) @@ -471,7 +469,7 @@ func (co *PeerConnection) run() { for _, track := range co.incomingTracks { track.start() } - atomic.StoreInt64(co.readingStarted, 1) + co.readingStarted.Store(1) case <-co.ctx.Done(): return @@ -837,7 +835,7 @@ func (co *PeerConnection) Stats() *Stats { packetsSent := uint64(0) packetsLost := uint64(0) - if atomic.LoadInt64(co.readingStarted) == 1 { + if co.readingStarted.Load() == 1 { for _, tr := range co.incomingTracks { if recvStats := tr.rtpReceiver.Stats(); recvStats != nil { v += recvStats.Jitter @@ -868,7 +866,7 @@ func (co *PeerConnection) Stats() *Stats { RTPPacketsSent: packetsSent, RTPPacketsLost: packetsLost, RTPPacketsJitter: rtpPacketsJitter, - RTCPPacketsReceived: atomic.LoadUint64(co.statsInterceptor.rtcpPacketsReceived), - RTCPPacketsSent: atomic.LoadUint64(co.statsInterceptor.rtcpPacketsSent), + RTCPPacketsReceived: co.statsInterceptor.rtcpPacketsReceived.Load(), + RTCPPacketsSent: co.statsInterceptor.rtcpPacketsSent.Load(), } } diff --git a/internal/protocols/webrtc/stats_interceptor.go b/internal/protocols/webrtc/stats_interceptor.go index bd754ea4..8f7edb05 100644 --- a/internal/protocols/webrtc/stats_interceptor.go +++ b/internal/protocols/webrtc/stats_interceptor.go @@ -8,8 +8,8 @@ import ( ) type statsInterceptor struct { - rtcpPacketsSent *uint64 - rtcpPacketsReceived *uint64 + rtcpPacketsSent atomic.Uint64 + rtcpPacketsReceived atomic.Uint64 } func (*statsInterceptor) Close() error { @@ -24,7 +24,7 @@ func (s *statsInterceptor) BindRTCPReader(reader interceptor.RTCPReader) interce pkts, err2 := attrs.GetRTCPPackets(bytes) if err2 == nil { - atomic.AddUint64(s.rtcpPacketsReceived, uint64(len(pkts))) + s.rtcpPacketsReceived.Add(uint64(len(pkts))) } return n, attrs, err @@ -33,7 +33,7 @@ func (s *statsInterceptor) BindRTCPReader(reader interceptor.RTCPReader) interce func (s *statsInterceptor) BindRTCPWriter(writer interceptor.RTCPWriter) interceptor.RTCPWriter { return interceptor.RTCPWriterFunc(func(pkts []rtcp.Packet, attributes interceptor.Attributes) (int, error) { - atomic.AddUint64(s.rtcpPacketsSent, uint64(len(pkts))) + s.rtcpPacketsSent.Add(uint64(len(pkts))) return writer.Write(pkts, attributes) }) } @@ -59,10 +59,7 @@ type statsInterceptorFactory struct { } func (f *statsInterceptorFactory) NewInterceptor(_ string) (interceptor.Interceptor, error) { - s := &statsInterceptor{ - rtcpPacketsSent: new(uint64), - rtcpPacketsReceived: new(uint64), - } + s := &statsInterceptor{} f.onCreate(s) diff --git a/internal/servers/hls/muxer.go b/internal/servers/hls/muxer.go index d173ab79..8d73994b 100644 --- a/internal/servers/hls/muxer.go +++ b/internal/servers/hls/muxer.go @@ -22,12 +22,6 @@ const ( recreatePause = 10 * time.Second ) -func ptrOf[T any](v T) *T { - p := new(T) - *p = v - return p -} - func emptyTimer() *time.Timer { t := time.NewTimer(0) <-t.C @@ -36,12 +30,12 @@ func emptyTimer() *time.Timer { type responseWriterWithCounter struct { http.ResponseWriter - bytesSent *uint64 + bytesSent *atomic.Uint64 } func (w *responseWriterWithCounter) Write(p []byte) (int, error) { n, err := w.ResponseWriter.Write(p) - atomic.AddUint64(w.bytesSent, uint64(n)) + w.bytesSent.Add(uint64(n)) return n, err } @@ -74,8 +68,8 @@ type muxer struct { ctxCancel func() created time.Time path defs.Path - lastRequestTime *int64 - bytesSent *uint64 + lastRequestTime atomic.Int64 + bytesSent atomic.Uint64 // in chGetInstance chan muxerGetInstanceReq @@ -87,8 +81,8 @@ func (m *muxer) initialize() { m.ctx = ctx m.ctxCancel = ctxCancel m.created = time.Now() - m.lastRequestTime = ptrOf(time.Now().UnixNano()) - m.bytesSent = new(uint64) + m.lastRequestTime.Store(time.Now().UnixNano()) + m.bytesSent.Store(0) m.chGetInstance = make(chan muxerGetInstanceReq) m.Log(logger.Info, "created %s", func() string { @@ -212,7 +206,7 @@ func (m *muxer) runInner() error { } case <-activityCheckTimer.C: - t := time.Unix(0, atomic.LoadInt64(m.lastRequestTime)) + t := time.Unix(0, m.lastRequestTime.Load()) if time.Since(t) >= time.Duration(m.closeAfter) { return fmt.Errorf("not used anymore") } @@ -261,7 +255,7 @@ func (m *muxer) APIReaderDescribe() *defs.APIPathReader { } func (m *muxer) handleRequest(ctx *gin.Context) { - atomic.StoreInt64(m.lastRequestTime, time.Now().UnixNano()) + m.lastRequestTime.Store(time.Now().UnixNano()) res := m.getInstance() if res.instance == nil { @@ -271,7 +265,7 @@ func (m *muxer) handleRequest(ctx *gin.Context) { w := &responseWriterWithCounter{ ResponseWriter: ctx.Writer, - bytesSent: m.bytesSent, + bytesSent: &m.bytesSent, } res.instance.handleRequest(w, ctx.Request) @@ -288,9 +282,9 @@ func (m *muxer) apiItem() *defs.APIHLSMuxer { return &defs.APIHLSMuxer{ Path: m.pathName, Created: m.created, - LastRequest: time.Unix(0, atomic.LoadInt64(m.lastRequestTime)), - OutboundBytes: atomic.LoadUint64(m.bytesSent), + LastRequest: time.Unix(0, m.lastRequestTime.Load()), + OutboundBytes: m.bytesSent.Load(), OutboundFramesDiscarded: outboundFramesDiscarded, - BytesSent: atomic.LoadUint64(m.bytesSent), + BytesSent: m.bytesSent.Load(), } } diff --git a/internal/servers/rtsp/server_test.go b/internal/servers/rtsp/server_test.go index 9c84ebda..211a1922 100644 --- a/internal/servers/rtsp/server_test.go +++ b/internal/servers/rtsp/server_test.go @@ -629,7 +629,7 @@ func TestAuthError(t *testing.T) { }, } - n := new(int64) + var n atomic.Int64 done := make(chan struct{}) s := &Server{ @@ -640,7 +640,7 @@ func TestAuthError(t *testing.T) { PathManager: pathManager, Parent: test.Logger(func(l logger.Level, s string, i ...any) { if l == logger.Info { - if atomic.AddInt64(n, 1) == 3 { + if n.Add(1) == 3 { require.Regexp(t, "authentication failed: auth error$", fmt.Sprintf(s, i...)) close(done) } diff --git a/internal/stream/stream.go b/internal/stream/stream.go index 7aeddd74..57d2d353 100644 --- a/internal/stream/stream.go +++ b/internal/stream/stream.go @@ -325,8 +325,8 @@ type Stream struct { mutex sync.RWMutex subStream *SubStream offlineSubStream *offlineSubStream - inboundBytes *uint64 - outboundBytes *uint64 + inboundBytes atomic.Uint64 + outboundBytes atomic.Uint64 medias map[*description.Media]*streamMedia rtspStream *gortsplib.ServerStream rtspsStream *gortsplib.ServerStream @@ -371,8 +371,6 @@ func (s *Stream) Initialize() error { s.Desc = cloneDesc(s.offlineDesc) } - s.inboundBytes = new(uint64) - s.outboundBytes = new(uint64) s.medias = make(map[*description.Media]*streamMedia) s.readers = make(map[*Reader]struct{}) s.hasReaders = make(chan struct{}) @@ -461,12 +459,12 @@ func (s *Stream) StartOfflineSubStream() error { // InboundBytes returns received bytes. func (s *Stream) InboundBytes() uint64 { - return atomic.LoadUint64(s.inboundBytes) + return s.inboundBytes.Load() } // OutboundBytes returns sent bytes. func (s *Stream) OutboundBytes() uint64 { - outboundBytes := atomic.LoadUint64(s.outboundBytes) + outboundBytes := s.outboundBytes.Load() s.mutex.RLock() defer s.mutex.RUnlock() @@ -577,11 +575,11 @@ func (s *Stream) WaitForReaders() { } func (s *Stream) addInboundBytes(v uint64) { - atomic.AddUint64(s.inboundBytes, v) + s.inboundBytes.Add(v) } func (s *Stream) addOutboundBytes(v uint64) { - atomic.AddUint64(s.outboundBytes, v) + s.outboundBytes.Add(v) } func (s *Stream) updateLastTime(pts time.Duration) {