OPC/mediamtx
2
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

use safer atomic structs instead of atomic functions (#5622)

Browse Source
This commit is contained in:
Alessandro Ros
2026-03-31 11:30:50 +02:00
committed by GitHub
parent 19f66ed06a
commit 0eb7089ed6
7 changed files with 38 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/core/path.go
+4 -5
View File
@@ -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)
internal/core/path_manager.go
+4 -5
View File
@@ -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,
internal/protocols/webrtc/peer_connection.go
+5 -7
View File
@@ -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(),
}
}
internal/protocols/webrtc/stats_interceptor.go
+5 -8
View File
@@ -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)
internal/servers/hls/muxer.go
+12 -18
View File
@@ -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(),
}
}
internal/servers/rtsp/server_test.go
+2 -2
View File
@@ -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)
}
internal/stream/stream.go
+6 -8
View File
@@ -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) {