Browse Source

remove count of readers and publishers from log lines

pull/643/head
aler9 5 years ago
parent
commit
af63360079
  1. 24
      internal/core/core.go
  2. 21
      internal/core/path.go
  3. 4
      internal/core/path_manager.go
  4. 23
      internal/core/stats.go

24
internal/core/core.go

@ -5,7 +5,6 @@ import (
"fmt" "fmt"
"os" "os"
"reflect" "reflect"
"sync/atomic"
"github.com/aler9/gortsplib" "github.com/aler9/gortsplib"
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
@ -26,7 +25,6 @@ type Core struct {
confPath string confPath string
conf *conf.Conf conf *conf.Conf
confFound bool confFound bool
stats *stats
logger *logger.Logger logger *logger.Logger
metrics *metrics metrics *metrics
pprof *pprof pprof *pprof
@ -118,12 +116,7 @@ func (p *Core) Wait() {
// Log is the main logging function. // Log is the main logging function.
func (p *Core) Log(level logger.Level, format string, args ...interface{}) { func (p *Core) Log(level logger.Level, format string, args ...interface{}) {
countPublishers := atomic.LoadInt64(p.stats.CountPublishers) p.logger.Log(level, format, args...)
countReaders := atomic.LoadInt64(p.stats.CountReaders)
p.logger.Log(level, "[%d/%d] "+format, append([]interface{}{
countPublishers, countReaders,
}, args...)...)
} }
func (p *Core) run() { func (p *Core) run() {
@ -180,10 +173,6 @@ outer:
func (p *Core) createResources(initial bool) error { func (p *Core) createResources(initial bool) error {
var err error var err error
if p.stats == nil {
p.stats = newStats()
}
if p.logger == nil { if p.logger == nil {
p.logger, err = logger.New( p.logger, err = logger.New(
logger.Level(p.conf.LogLevel), logger.Level(p.conf.LogLevel),
@ -232,7 +221,6 @@ func (p *Core) createResources(initial bool) error {
p.conf.ReadBufferCount, p.conf.ReadBufferCount,
p.conf.ReadBufferSize, p.conf.ReadBufferSize,
p.conf.Paths, p.conf.Paths,
p.stats,
p.metrics, p.metrics,
p) p)
} }
@ -367,11 +355,6 @@ func (p *Core) createResources(initial bool) error {
} }
func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
closeStats := false
if newConf == nil {
closeStats = true
}
closeLogger := false closeLogger := false
if newConf == nil || if newConf == nil ||
!reflect.DeepEqual(newConf.LogDestinations, p.conf.LogDestinations) || !reflect.DeepEqual(newConf.LogDestinations, p.conf.LogDestinations) ||
@ -400,7 +383,6 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
newConf.WriteTimeout != p.conf.WriteTimeout || newConf.WriteTimeout != p.conf.WriteTimeout ||
newConf.ReadBufferCount != p.conf.ReadBufferCount || newConf.ReadBufferCount != p.conf.ReadBufferCount ||
newConf.ReadBufferSize != p.conf.ReadBufferSize || newConf.ReadBufferSize != p.conf.ReadBufferSize ||
closeStats ||
closeMetrics { closeMetrics {
closePathManager = true closePathManager = true
} else if !reflect.DeepEqual(newConf.Paths, p.conf.Paths) { } else if !reflect.DeepEqual(newConf.Paths, p.conf.Paths) {
@ -538,10 +520,6 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
p.logger.Close() p.logger.Close()
p.logger = nil p.logger = nil
} }
if closeStats && p.stats != nil {
p.stats.close()
}
} }
func (p *Core) reloadConf(newConf *conf.Conf, calledByAPI bool) error { func (p *Core) reloadConf(newConf *conf.Conf, calledByAPI bool) error {

21
internal/core/path.go

@ -6,7 +6,6 @@ import (
"net" "net"
"strings" "strings"
"sync" "sync"
"sync/atomic"
"time" "time"
"github.com/aler9/gortsplib" "github.com/aler9/gortsplib"
@ -190,7 +189,6 @@ type path struct {
conf *conf.PathConf conf *conf.PathConf
name string name string
wg *sync.WaitGroup wg *sync.WaitGroup
stats *stats
parent pathParent parent pathParent
ctx context.Context ctx context.Context
@ -234,7 +232,6 @@ func newPath(
conf *conf.PathConf, conf *conf.PathConf,
name string, name string,
wg *sync.WaitGroup, wg *sync.WaitGroup,
stats *stats,
parent pathParent) *path { parent pathParent) *path {
ctx, ctxCancel := context.WithCancel(parentCtx) ctx, ctxCancel := context.WithCancel(parentCtx)
@ -248,7 +245,6 @@ func newPath(
conf: conf, conf: conf,
name: name, name: name,
wg: wg, wg: wg,
stats: stats,
parent: parent, parent: parent,
ctx: ctx, ctx: ctx,
ctxCancel: ctxCancel, ctxCancel: ctxCancel,
@ -431,10 +427,7 @@ outer:
req.Res <- pathReaderSetupPlayRes{Err: fmt.Errorf("terminated")} req.Res <- pathReaderSetupPlayRes{Err: fmt.Errorf("terminated")}
} }
for rp, state := range pa.readers { for rp := range pa.readers {
if state == pathReaderStatePlay {
atomic.AddInt64(pa.stats.CountReaders, -1)
}
rp.close() rp.close()
} }
@ -447,9 +440,6 @@ outer:
source.close() source.close()
pa.sourceStaticWg.Wait() pa.sourceStaticWg.Wait()
} else if source, ok := pa.source.(publisher); ok { } else if source, ok := pa.source.(publisher); ok {
if pa.sourceReady {
atomic.AddInt64(pa.stats.CountPublishers, -1)
}
source.close() source.close()
} }
} }
@ -634,7 +624,6 @@ func (pa *path) doReaderRemove(r reader) {
state := pa.readers[r] state := pa.readers[r]
if state == pathReaderStatePlay { if state == pathReaderStatePlay {
atomic.AddInt64(pa.stats.CountReaders, -1)
pa.stream.readerRemove(r) pa.stream.readerRemove(r)
} }
@ -643,8 +632,6 @@ func (pa *path) doReaderRemove(r reader) {
func (pa *path) doPublisherRemove() { func (pa *path) doPublisherRemove() {
if pa.sourceReady { if pa.sourceReady {
atomic.AddInt64(pa.stats.CountPublishers, -1)
if pa.isOnDemand() && pa.onDemandState != pathOnDemandStateInitial { if pa.isOnDemand() && pa.onDemandState != pathOnDemandStateInitial {
pa.onDemandCloseSource() pa.onDemandCloseSource()
} else { } else {
@ -738,8 +725,6 @@ func (pa *path) handlePublisherRecord(req pathPublisherRecordReq) {
return return
} }
atomic.AddInt64(pa.stats.CountPublishers, 1)
req.Author.onPublisherAccepted(len(req.Tracks)) req.Author.onPublisherAccepted(len(req.Tracks))
pa.sourceSetReady(req.Tracks) pa.sourceSetReady(req.Tracks)
@ -758,8 +743,6 @@ func (pa *path) handlePublisherRecord(req pathPublisherRecordReq) {
func (pa *path) handlePublisherPause(req pathPublisherPauseReq) { func (pa *path) handlePublisherPause(req pathPublisherPauseReq) {
if req.Author == pa.source && pa.sourceReady { if req.Author == pa.source && pa.sourceReady {
atomic.AddInt64(pa.stats.CountPublishers, -1)
if pa.isOnDemand() && pa.onDemandState != pathOnDemandStateInitial { if pa.isOnDemand() && pa.onDemandState != pathOnDemandStateInitial {
pa.onDemandCloseSource() pa.onDemandCloseSource()
} else { } else {
@ -815,7 +798,6 @@ func (pa *path) handleReaderSetupPlayPost(req pathReaderSetupPlayReq) {
} }
func (pa *path) handleReaderPlay(req pathReaderPlayReq) { func (pa *path) handleReaderPlay(req pathReaderPlayReq) {
atomic.AddInt64(pa.stats.CountReaders, 1)
pa.readers[req.Author] = pathReaderStatePlay pa.readers[req.Author] = pathReaderStatePlay
pa.stream.readerAdd(req.Author) pa.stream.readerAdd(req.Author)
@ -827,7 +809,6 @@ func (pa *path) handleReaderPlay(req pathReaderPlayReq) {
func (pa *path) handleReaderPause(req pathReaderPauseReq) { func (pa *path) handleReaderPause(req pathReaderPauseReq) {
if state, ok := pa.readers[req.Author]; ok && state == pathReaderStatePlay { if state, ok := pa.readers[req.Author]; ok && state == pathReaderStatePlay {
atomic.AddInt64(pa.stats.CountReaders, -1)
pa.readers[req.Author] = pathReaderStatePrePlay pa.readers[req.Author] = pathReaderStatePrePlay
pa.stream.readerRemove(req.Author) pa.stream.readerRemove(req.Author)
} }

4
internal/core/path_manager.go

@ -27,7 +27,6 @@ type pathManager struct {
readBufferCount int readBufferCount int
readBufferSize int readBufferSize int
pathConfs map[string]*conf.PathConf pathConfs map[string]*conf.PathConf
stats *stats
metrics *metrics metrics *metrics
parent pathManagerParent parent pathManagerParent
@ -56,7 +55,6 @@ func newPathManager(
readBufferCount int, readBufferCount int,
readBufferSize int, readBufferSize int,
pathConfs map[string]*conf.PathConf, pathConfs map[string]*conf.PathConf,
stats *stats,
metrics *metrics, metrics *metrics,
parent pathManagerParent) *pathManager { parent pathManagerParent) *pathManager {
ctx, ctxCancel := context.WithCancel(parentCtx) ctx, ctxCancel := context.WithCancel(parentCtx)
@ -68,7 +66,6 @@ func newPathManager(
readBufferCount: readBufferCount, readBufferCount: readBufferCount,
readBufferSize: readBufferSize, readBufferSize: readBufferSize,
pathConfs: pathConfs, pathConfs: pathConfs,
stats: stats,
metrics: metrics, metrics: metrics,
parent: parent, parent: parent,
ctx: ctx, ctx: ctx,
@ -285,7 +282,6 @@ func (pm *pathManager) createPath(confName string, conf *conf.PathConf, name str
conf, conf,
name, name,
&pm.wg, &pm.wg,
pm.stats,
pm) pm)
} }

23
internal/core/stats.go

@ -1,23 +0,0 @@
package core
func ptrInt64() *int64 {
v := int64(0)
return &v
}
type stats struct {
// use pointers to avoid a crash on 32bit platforms
// https://github.com/golang/go/issues/9959
CountPublishers *int64
CountReaders *int64
}
func newStats() *stats {
return &stats{
CountPublishers: ptrInt64(),
CountReaders: ptrInt64(),
}
}
func (s *stats) close() {
}
Loading…
Cancel
Save