From f086b624c1584defd01d8821f6dc0a3c71f98f3f Mon Sep 17 00:00:00 2001 From: aler9 <46489434+aler9@users.noreply.github.com> Date: Tue, 11 May 2021 17:20:32 +0200 Subject: [PATCH] use hierarchical contexts --- go.mod | 2 +- go.sum | 4 ++-- internal/hlsconverter/converter.go | 5 ++++- internal/hlsserver/server.go | 4 +++- internal/path/path.go | 5 ++++- internal/pathman/pathman.go | 8 +++----- internal/rtmpconn/conn.go | 5 +++-- internal/rtmpserver/server.go | 4 +++- internal/rtmpsource/source.go | 16 +++++++++------- internal/rtspserver/server.go | 3 ++- internal/rtspsource/source.go | 3 ++- main.go | 5 +++++ 12 files changed, 41 insertions(+), 23 deletions(-) diff --git a/go.mod b/go.mod index 55225783..d44ef08e 100644 --- a/go.mod +++ b/go.mod @@ -5,7 +5,7 @@ go 1.15 require ( github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 // indirect github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d // indirect - github.com/aler9/gortsplib v0.0.0-20210511094934-fa2830eb2251 + github.com/aler9/gortsplib v0.0.0-20210511125241-4d1c2d18319b github.com/asticode/go-astits v0.0.0-00010101000000-000000000000 github.com/davecgh/go-spew v1.1.1 // indirect github.com/fsnotify/fsnotify v1.4.9 diff --git a/go.sum b/go.sum index 03e12022..bc7477e9 100644 --- a/go.sum +++ b/go.sum @@ -4,8 +4,8 @@ github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d h1:UQZhZ2O0vMHr2c github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho= github.com/aler9/go-astits v0.0.0-20210423195926-582b09ed7c04 h1:CXgQLsU4uxWAmsXNOjGLbj0A+0IlRcpZpMgI13fmVwo= github.com/aler9/go-astits v0.0.0-20210423195926-582b09ed7c04/go.mod h1:DkOWmBNQpnr9mv24KfZjq4JawCFX1FCqjLVGvO0DygQ= -github.com/aler9/gortsplib v0.0.0-20210511094934-fa2830eb2251 h1:pG9MnD5MOIpqcOBccQ5L4ZAhJGBcN7LKO87ilADzIYI= -github.com/aler9/gortsplib v0.0.0-20210511094934-fa2830eb2251/go.mod h1:zVCg+TQX445hh1pC5QgAuuBvvXZMWLY1XYz626dGFqY= +github.com/aler9/gortsplib v0.0.0-20210511125241-4d1c2d18319b h1:meeL4Le2z7zG0qO+X+2kG20/ys0MEph6RvamSYHqlAE= +github.com/aler9/gortsplib v0.0.0-20210511125241-4d1c2d18319b/go.mod h1:zVCg+TQX445hh1pC5QgAuuBvvXZMWLY1XYz626dGFqY= github.com/aler9/rtmp v0.0.0-20210403095203-3be4a5535927 h1:95mXJ5fUCYpBRdSOnLAQAdJHHKxxxJrVCiaqDi965YQ= github.com/aler9/rtmp v0.0.0-20210403095203-3be4a5535927/go.mod h1:vzuE21rowz+lT1NGsWbreIvYulgBpCGnQyeTyFblUHc= github.com/asticode/go-astikit v0.20.0 h1:+7N+J4E4lWx2QOkRdOf6DafWJMv6O4RRfgClwQokrH8= diff --git a/internal/hlsconverter/converter.go b/internal/hlsconverter/converter.go index 1f578081..05c85ef7 100644 --- a/internal/hlsconverter/converter.go +++ b/internal/hlsconverter/converter.go @@ -152,6 +152,7 @@ type Converter struct { // New allocates a Converter. func New( + ctxParent context.Context, hlsSegmentCount int, hlsSegmentDuration time.Duration, readBufferCount int, @@ -161,7 +162,7 @@ func New( pathMan PathMan, parent Parent) *Converter { - ctx, ctxCancel := context.WithCancel(context.Background()) + ctx, ctxCancel := context.WithCancel(ctxParent) c := &Converter{ hlsSegmentCount: hlsSegmentCount, @@ -495,6 +496,8 @@ func (c *Converter) runInner(innerCtx context.Context) error { return err case <-innerCtx.Done(): + c.ringBuffer.Close() + <-writerDone return nil } } diff --git a/internal/hlsserver/server.go b/internal/hlsserver/server.go index 28a32249..8543254d 100644 --- a/internal/hlsserver/server.go +++ b/internal/hlsserver/server.go @@ -42,6 +42,7 @@ type Server struct { // New allocates a Server. func New( + ctxParent context.Context, address string, hlsSegmentCount int, hlsSegmentDuration time.Duration, @@ -56,7 +57,7 @@ func New( return nil, err } - ctx, ctxCancel := context.WithCancel(context.Background()) + ctx, ctxCancel := context.WithCancel(ctxParent) s := &Server{ hlsSegmentCount: hlsSegmentCount, @@ -105,6 +106,7 @@ outer: c, ok := s.converters[req.Path] if !ok { c = hlsconverter.New( + s.ctx, s.hlsSegmentCount, s.hlsSegmentDuration, s.readBufferCount, diff --git a/internal/path/path.go b/internal/path/path.go index 788725ea..8a0af5e0 100644 --- a/internal/path/path.go +++ b/internal/path/path.go @@ -105,6 +105,7 @@ type Path struct { // New allocates a Path. func New( + ctxParent context.Context, rtspAddress string, readTimeout time.Duration, writeTimeout time.Duration, @@ -117,7 +118,7 @@ func New( stats *stats.Stats, parent Parent) *Path { - ctx, ctxCancel := context.WithCancel(context.Background()) + ctx, ctxCancel := context.WithCancel(ctxParent) pa := &Path{ rtspAddress: rtspAddress, @@ -327,6 +328,7 @@ func (pa *Path) startExternalSource() { if strings.HasPrefix(pa.conf.Source, "rtsp://") || strings.HasPrefix(pa.conf.Source, "rtsps://") { pa.source = rtspsource.New( + pa.ctx, pa.conf.Source, pa.conf.SourceProtocolParsed, pa.conf.SourceFingerprint, @@ -340,6 +342,7 @@ func (pa *Path) startExternalSource() { } else if strings.HasPrefix(pa.conf.Source, "rtmp://") { pa.source = rtmpsource.New( + pa.ctx, pa.conf.Source, pa.readTimeout, pa.writeTimeout, diff --git a/internal/pathman/pathman.go b/internal/pathman/pathman.go index 4809ea41..85147c25 100644 --- a/internal/pathman/pathman.go +++ b/internal/pathman/pathman.go @@ -66,6 +66,7 @@ type PathManager struct { // New allocates a PathManager. func New( + ctxParent context.Context, rtspAddress string, readTimeout time.Duration, writeTimeout time.Duration, @@ -76,7 +77,7 @@ func New( stats *stats.Stats, parent Parent) *PathManager { - ctx, ctxCancel := context.WithCancel(context.Background()) + ctx, ctxCancel := context.WithCancel(ctxParent) pm := &PathManager{ rtspAddress: rtspAddress, @@ -251,14 +252,11 @@ outer: } pm.ctxCancel() - - for _, pa := range pm.paths { - pa.Close() - } } func (pm *PathManager) createPath(confName string, conf *conf.PathConf, name string) { pm.paths[name] = path.New( + pm.ctx, pm.rtspAddress, pm.readTimeout, pm.writeTimeout, diff --git a/internal/rtmpconn/conn.go b/internal/rtmpconn/conn.go index 5b5273bc..839b9a2b 100644 --- a/internal/rtmpconn/conn.go +++ b/internal/rtmpconn/conn.go @@ -83,6 +83,7 @@ type Conn struct { // New allocates a Conn. func New( + ctxParent context.Context, rtspAddress string, readTimeout time.Duration, writeTimeout time.Duration, @@ -95,7 +96,7 @@ func New( pathMan PathMan, parent Parent) *Conn { - ctx, ctxCancel := context.WithCancel(context.Background()) + ctx, ctxCancel := context.WithCancel(ctxParent) c := &Conn{ rtspAddress: rtspAddress, @@ -157,7 +158,7 @@ func (c *Conn) run() { defer onConnectCmd.Close() } - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithCancel(c.ctx) runErr := make(chan error) go func() { runErr <- c.runInner(ctx) diff --git a/internal/rtmpserver/server.go b/internal/rtmpserver/server.go index cec50b31..be6b451f 100644 --- a/internal/rtmpserver/server.go +++ b/internal/rtmpserver/server.go @@ -41,6 +41,7 @@ type Server struct { // New allocates a Server. func New( + ctxParent context.Context, address string, readTimeout time.Duration, writeTimeout time.Duration, @@ -57,7 +58,7 @@ func New( return nil, err } - ctx, ctxCancel := context.WithCancel(context.Background()) + ctx, ctxCancel := context.WithCancel(ctxParent) s := &Server{ readTimeout: readTimeout, @@ -133,6 +134,7 @@ outer: case nconn := <-connNew: c := rtmpconn.New( + s.ctx, s.rtspAddress, s.readTimeout, s.writeTimeout, diff --git a/internal/rtmpsource/source.go b/internal/rtmpsource/source.go index 7921f285..09aeb612 100644 --- a/internal/rtmpsource/source.go +++ b/internal/rtmpsource/source.go @@ -45,14 +45,16 @@ type Source struct { } // New allocates a Source. -func New(ur string, +func New( + ctxParent context.Context, + ur string, readTimeout time.Duration, writeTimeout time.Duration, wg *sync.WaitGroup, stats *stats.Stats, parent Parent) *Source { - ctx, ctxCancel := context.WithCancel(context.Background()) + ctx, ctxCancel := context.WithCancel(ctxParent) s := &Source{ ur: ur, @@ -116,14 +118,14 @@ func (s *Source) run() { } func (s *Source) runInner() bool { - ctx, cancel := context.WithCancel(context.Background()) + innerCtx, innerCtxCancel := context.WithCancel(s.ctx) runErr := make(chan error) go func() { runErr <- func() error { s.log(logger.Debug, "connecting") - ctx2, cancel2 := context.WithTimeout(ctx, s.readTimeout) + ctx2, cancel2 := context.WithTimeout(innerCtx, s.readTimeout) defer cancel2() conn, err := rtmp.DialContext(ctx2, s.ur) @@ -258,7 +260,7 @@ func (s *Source) runInner() bool { conn.NetConn().Close() return err - case <-ctx.Done(): + case <-innerCtx.Done(): conn.NetConn().Close() <-readDone return nil @@ -268,12 +270,12 @@ func (s *Source) runInner() bool { select { case err := <-runErr: - cancel() + innerCtxCancel() s.log(logger.Info, "ERR: %s", err) return true case <-s.ctx.Done(): - cancel() + innerCtxCancel() <-runErr return false } diff --git a/internal/rtspserver/server.go b/internal/rtspserver/server.go index dc656d31..1a2d09f8 100644 --- a/internal/rtspserver/server.go +++ b/internal/rtspserver/server.go @@ -71,6 +71,7 @@ type Server struct { // New allocates a Server. func New( + ctxParent context.Context, address string, readTimeout time.Duration, writeTimeout time.Duration, @@ -90,7 +91,7 @@ func New( pathMan *pathman.PathManager, parent Parent) (*Server, error) { - ctx, ctxCancel := context.WithCancel(context.Background()) + ctx, ctxCancel := context.WithCancel(ctxParent) s := &Server{ readTimeout: readTimeout, diff --git a/internal/rtspsource/source.go b/internal/rtspsource/source.go index 0b7f8f28..c3e1b478 100644 --- a/internal/rtspsource/source.go +++ b/internal/rtspsource/source.go @@ -49,6 +49,7 @@ type Source struct { // New allocates a Source. func New( + ctxParent context.Context, ur string, proto *gortsplib.StreamProtocol, fingerprint string, @@ -60,7 +61,7 @@ func New( stats *stats.Stats, parent Parent) *Source { - ctx, ctxCancel := context.WithCancel(context.Background()) + ctx, ctxCancel := context.WithCancel(ctxParent) s := &Source{ ur: ur, diff --git a/main.go b/main.go index 8fccf117..0a816b2a 100644 --- a/main.go +++ b/main.go @@ -198,6 +198,7 @@ func (p *program) createResources(initial bool) error { if p.pathMan == nil { p.pathMan = pathman.New( + p.ctx, p.conf.RTSPAddress, p.conf.ReadTimeout, p.conf.WriteTimeout, @@ -215,6 +216,7 @@ func (p *program) createResources(initial bool) error { if p.serverRTSPPlain == nil { _, useUDP := p.conf.ProtocolsParsed[gortsplib.StreamProtocolUDP] p.serverRTSPPlain, err = rtspserver.New( + p.ctx, p.conf.RTSPAddress, p.conf.ReadTimeout, p.conf.WriteTimeout, @@ -244,6 +246,7 @@ func (p *program) createResources(initial bool) error { p.conf.EncryptionParsed == conf.EncryptionOptional) { if p.serverRTSPTLS == nil { p.serverRTSPTLS, err = rtspserver.New( + p.ctx, p.conf.RTSPSAddress, p.conf.ReadTimeout, p.conf.WriteTimeout, @@ -271,6 +274,7 @@ func (p *program) createResources(initial bool) error { if !p.conf.RTMPDisable { if p.serverRTMP == nil { p.serverRTMP, err = rtmpserver.New( + p.ctx, p.conf.RTMPAddress, p.conf.ReadTimeout, p.conf.WriteTimeout, @@ -290,6 +294,7 @@ func (p *program) createResources(initial bool) error { if !p.conf.HLSDisable { if p.serverHLS == nil { p.serverHLS, err = hlsserver.New( + p.ctx, p.conf.HLSAddress, p.conf.HLSSegmentCount, p.conf.HLSSegmentDuration,