pimhwang
/
mediamtx
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

improve performance by using path instead of pathName for packet routing

pull/80/head
aler9 5 years ago
parent
2fe7004bc9
commit
1b4201aa76
4 changed files with 78 additions and 111 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 45
      client.go
  2. 84
      main.go
  3. 24
      path.go
  4. 12
      source.go

45
client.go
Unescape View File

@ -31,8 +31,7 @@ type udpClient struct {
} }
type udpClientAddr struct { type udpClientAddr struct {
// use a fixed-size array for ip comparison ip [net.IPv6len]byte // use a fixed-size array to enable the equality operator
ip [net.IPv6len]byte
port int port int
} }
@ -110,7 +109,7 @@ type client struct {
p *program p *program
conn *gortsplib.ConnServer conn *gortsplib.ConnServer
state clientState state clientState
pathName string path *path
authUser string authUser string
authPass string authPass string
authHelper *gortsplib.AuthServer authHelper *gortsplib.AuthServer
@ -481,8 +480,8 @@ func (c *client) handleRequest(req *gortsplib.Request) bool {
return true return true
} }
if c.pathName != "" && basePath != c.pathName { if c.path != nil && basePath != c.path.name {
c.writeResError(cseq, gortsplib.StatusBadRequest, fmt.Errorf("path has changed, was '%s', now is '%s'", c.pathName, basePath)) c.writeResError(cseq, gortsplib.StatusBadRequest, fmt.Errorf("path has changed, was '%s', now is '%s'", c.path.name, basePath))
return false return false
} }
@ -614,9 +613,9 @@ func (c *client) handleRequest(req *gortsplib.Request) bool {
return false return false
} }
// after ANNOUNCE, c.pathName is already set // after ANNOUNCE, c.path is already set
if basePath != c.pathName { if basePath != c.path.name {
c.writeResError(cseq, gortsplib.StatusBadRequest, fmt.Errorf("path has changed, was '%s', now is '%s'", c.pathName, basePath)) c.writeResError(cseq, gortsplib.StatusBadRequest, fmt.Errorf("path has changed, was '%s', now is '%s'", c.path.name, basePath))
return false return false
} }
@ -648,7 +647,7 @@ func (c *client) handleRequest(req *gortsplib.Request) bool {
return false return false
} }
if len(c.streamTracks) >= len(c.p.paths[c.pathName].publisherSdpParsed.MediaDescriptions) { if len(c.streamTracks) >= len(c.path.publisherSdpParsed.MediaDescriptions) {
c.writeResError(cseq, gortsplib.StatusBadRequest, fmt.Errorf("all the tracks have already been setup")) c.writeResError(cseq, gortsplib.StatusBadRequest, fmt.Errorf("all the tracks have already been setup"))
return false return false
} }
@ -706,7 +705,7 @@ func (c *client) handleRequest(req *gortsplib.Request) bool {
return false return false
} }
if len(c.streamTracks) >= len(c.p.paths[c.pathName].publisherSdpParsed.MediaDescriptions) { if len(c.streamTracks) >= len(c.path.publisherSdpParsed.MediaDescriptions) {
c.writeResError(cseq, gortsplib.StatusBadRequest, fmt.Errorf("all the tracks have already been setup")) c.writeResError(cseq, gortsplib.StatusBadRequest, fmt.Errorf("all the tracks have already been setup"))
return false return false
} }
@ -759,17 +758,13 @@ func (c *client) handleRequest(req *gortsplib.Request) bool {
// path can end with a slash, remove it // path can end with a slash, remove it
path = strings.TrimSuffix(path, "/") path = strings.TrimSuffix(path, "/")
if path != c.pathName { if path != c.path.name {
c.writeResError(cseq, gortsplib.StatusBadRequest, fmt.Errorf("path has changed, was '%s', now is '%s'", c.pathName, path)) c.writeResError(cseq, gortsplib.StatusBadRequest, fmt.Errorf("path has changed, was '%s', now is '%s'", c.path.name, path))
return false return false
} }
// check publisher existence if len(c.streamTracks) == 0 {
res := make(chan error) c.writeResError(cseq, gortsplib.StatusBadRequest, fmt.Errorf("no tracks have been setup"))
c.p.events <- programEventClientPlay1{res, c}
err := <-res
if err != nil {
c.writeResError(cseq, gortsplib.StatusBadRequest, err)
return false return false
} }
@ -797,12 +792,12 @@ func (c *client) handleRequest(req *gortsplib.Request) bool {
// path can end with a slash, remove it // path can end with a slash, remove it
path = strings.TrimSuffix(path, "/") path = strings.TrimSuffix(path, "/")
if path != c.pathName { if path != c.path.name {
c.writeResError(cseq, gortsplib.StatusBadRequest, fmt.Errorf("path has changed, was '%s', now is '%s'", c.pathName, path)) c.writeResError(cseq, gortsplib.StatusBadRequest, fmt.Errorf("path has changed, was '%s', now is '%s'", c.path.name, path))
return false return false
} }
if len(c.streamTracks) != len(c.p.paths[c.pathName].publisherSdpParsed.MediaDescriptions) { if len(c.streamTracks) != len(c.path.publisherSdpParsed.MediaDescriptions) {
c.writeResError(cseq, gortsplib.StatusBadRequest, fmt.Errorf("not all tracks have been setup")) c.writeResError(cseq, gortsplib.StatusBadRequest, fmt.Errorf("not all tracks have been setup"))
return false return false
} }
@ -837,10 +832,10 @@ func (c *client) runPlay(path string) {
// start sending frames only after sending the response to the PLAY request // start sending frames only after sending the response to the PLAY request
done := make(chan struct{}) done := make(chan struct{})
c.p.events <- programEventClientPlay2{done, c} c.p.events <- programEventClientPlay{done, c}
<-done <-done
c.log("is receiving on path '%s', %d %s via %s", c.pathName, len(c.streamTracks), func() string { c.log("is receiving on path '%s', %d %s via %s", c.path.name, len(c.streamTracks), func() string {
if len(c.streamTracks) == 1 { if len(c.streamTracks) == 1 {
return "track" return "track"
} }
@ -958,7 +953,7 @@ func (c *client) runRecord(path string) {
c.p.events <- programEventClientRecord{done, c} c.p.events <- programEventClientRecord{done, c}
<-done <-done
c.log("is publishing on path '%s', %d %s via %s", c.pathName, len(c.streamTracks), func() string { c.log("is publishing on path '%s', %d %s via %s", c.path.name, len(c.streamTracks), func() string {
if len(c.streamTracks) == 1 { if len(c.streamTracks) == 1 {
return "track" return "track"
} }
@ -1063,7 +1058,7 @@ func (c *client) runRecord(path string) {
c.rtcpReceivers[frame.TrackId].OnFrame(frame.StreamType, frame.Content) c.rtcpReceivers[frame.TrackId].OnFrame(frame.StreamType, frame.Content)
c.p.events <- programEventClientFrameTcp{ c.p.events <- programEventClientFrameTcp{
c.pathName, c.path,
frame.TrackId, frame.TrackId,
frame.StreamType, frame.StreamType,
frame.Content, frame.Content,

84
main.go
Unescape View File

@ -51,7 +51,7 @@ func (programEventClientClose) isProgramEvent() {}
type programEventClientDescribe struct { type programEventClientDescribe struct {
client *client client *client
path string pathName string
} }
func (programEventClientDescribe) isProgramEvent() {} func (programEventClientDescribe) isProgramEvent() {}
@ -59,7 +59,7 @@ func (programEventClientDescribe) isProgramEvent() {}
type programEventClientAnnounce struct { type programEventClientAnnounce struct {
res chan error res chan error
client *client client *client
path string pathName string
sdpText []byte sdpText []byte
sdpParsed *sdp.SessionDescription sdpParsed *sdp.SessionDescription
} }
@ -69,7 +69,7 @@ func (programEventClientAnnounce) isProgramEvent() {}
type programEventClientSetupPlay struct { type programEventClientSetupPlay struct {
res chan error res chan error
client *client client *client
path string pathName string
trackId int trackId int
} }
@ -82,19 +82,12 @@ type programEventClientSetupRecord struct {
func (programEventClientSetupRecord) isProgramEvent() {} func (programEventClientSetupRecord) isProgramEvent() {}
type programEventClientPlay1 struct { type programEventClientPlay struct {
res chan error
client *client
}
func (programEventClientPlay1) isProgramEvent() {}
type programEventClientPlay2 struct {
done chan struct{} done chan struct{}
client *client client *client
} }
func (programEventClientPlay2) isProgramEvent() {} func (programEventClientPlay) isProgramEvent() {}
type programEventClientPlayStop struct { type programEventClientPlayStop struct {
done chan struct{} done chan struct{}
@ -126,7 +119,7 @@ type programEventClientFrameUdp struct {
func (programEventClientFrameUdp) isProgramEvent() {} func (programEventClientFrameUdp) isProgramEvent() {}
type programEventClientFrameTcp struct { type programEventClientFrameTcp struct {
path string path *path
trackId int trackId int
streamType gortsplib.StreamType streamType gortsplib.StreamType
buf []byte buf []byte
@ -321,12 +314,8 @@ outer:
case programEventClientClose: case programEventClientClose:
delete(p.clients, evt.client) delete(p.clients, evt.client)
if evt.client.pathName != "" { if evt.client.path != nil && evt.client.path.publisher == evt.client {
if path, ok := p.paths[evt.client.pathName]; ok { evt.client.path.onPublisherRemove()
if path.publisher == evt.client {
path.onPublisherRemove()
}
}
} }
evt.client.log("disconnected") evt.client.log("disconnected")
@ -334,31 +323,31 @@ outer:
case programEventClientDescribe: case programEventClientDescribe:
// create path if not exist // create path if not exist
if _, ok := p.paths[evt.path]; !ok { if _, ok := p.paths[evt.pathName]; !ok {
p.paths[evt.path] = newPath(p, evt.path, p.findConfForPathName(evt.path), false) p.paths[evt.pathName] = newPath(p, evt.pathName, p.findConfForPathName(evt.pathName), false)
} }
p.paths[evt.path].onDescribe(evt.client) p.paths[evt.pathName].onDescribe(evt.client)
case programEventClientAnnounce: case programEventClientAnnounce:
// create path if not exist // create path if not exist
if path, ok := p.paths[evt.path]; !ok { if path, ok := p.paths[evt.pathName]; !ok {
p.paths[evt.path] = newPath(p, evt.path, p.findConfForPathName(evt.path), false) p.paths[evt.pathName] = newPath(p, evt.pathName, p.findConfForPathName(evt.pathName), false)
} else { } else {
if path.publisher != nil { if path.publisher != nil {
evt.res <- fmt.Errorf("someone is already publishing on path '%s'", evt.path) evt.res <- fmt.Errorf("someone is already publishing on path '%s'", evt.pathName)
continue continue
} }
} }
p.paths[evt.path].onPublisherNew(evt.client, evt.sdpText, evt.sdpParsed) p.paths[evt.pathName].onPublisherNew(evt.client, evt.sdpText, evt.sdpParsed)
evt.res <- nil evt.res <- nil
case programEventClientSetupPlay: case programEventClientSetupPlay:
path, ok := p.paths[evt.path] path, ok := p.paths[evt.pathName]
if !ok || !path.publisherReady { if !ok || !path.publisherReady {
evt.res <- fmt.Errorf("no one is publishing on path '%s'", evt.path) evt.res <- fmt.Errorf("no one is publishing on path '%s'", evt.pathName)
continue continue
} }
@ -367,7 +356,7 @@ outer:
continue continue
} }
evt.client.pathName = evt.path evt.client.path = path
evt.client.state = clientStatePrePlay evt.client.state = clientStatePrePlay
evt.res <- nil evt.res <- nil
@ -375,21 +364,7 @@ outer:
evt.client.state = clientStatePreRecord evt.client.state = clientStatePreRecord
evt.res <- nil evt.res <- nil
case programEventClientPlay1: case programEventClientPlay:
path, ok := p.paths[evt.client.pathName]
if !ok || !path.publisherReady {
evt.res <- fmt.Errorf("no one is publishing on path '%s'", evt.client.pathName)
continue
}
if len(evt.client.streamTracks) == 0 {
evt.res <- fmt.Errorf("no tracks have been setup")
continue
}
evt.res <- nil
case programEventClientPlay2:
p.readerCount += 1 p.readerCount += 1
evt.client.state = clientStatePlay evt.client.state = clientStatePlay
close(evt.done) close(evt.done)
@ -421,7 +396,7 @@ outer:
} }
} }
p.paths[evt.client.pathName].onPublisherSetReady() evt.client.path.onPublisherSetReady()
close(evt.done) close(evt.done)
case programEventClientRecordStop: case programEventClientRecordStop:
@ -436,7 +411,7 @@ outer:
delete(p.udpClientsByAddr, key) delete(p.udpClientsByAddr, key)
} }
} }
p.paths[evt.client.pathName].onPublisherSetNotReady() evt.client.path.onPublisherSetNotReady()
close(evt.done) close(evt.done)
case programEventClientFrameUdp: case programEventClientFrameUdp:
@ -451,21 +426,21 @@ outer:
} }
pub.client.rtcpReceivers[pub.trackId].OnFrame(evt.streamType, evt.buf) pub.client.rtcpReceivers[pub.trackId].OnFrame(evt.streamType, evt.buf)
p.forwardFrame(pub.client.pathName, pub.trackId, evt.streamType, evt.buf) p.forwardFrame(pub.client.path, pub.trackId, evt.streamType, evt.buf)
case programEventClientFrameTcp: case programEventClientFrameTcp:
p.forwardFrame(evt.path, evt.trackId, evt.streamType, evt.buf) p.forwardFrame(evt.path, evt.trackId, evt.streamType, evt.buf)
case programEventSourceReady: case programEventSourceReady:
evt.source.log("ready") evt.source.log("ready")
p.paths[evt.source.pathName].onPublisherSetReady() evt.source.path.onPublisherSetReady()
case programEventSourceNotReady: case programEventSourceNotReady:
evt.source.log("not ready") evt.source.log("not ready")
p.paths[evt.source.pathName].onPublisherSetNotReady() evt.source.path.onPublisherSetNotReady()
case programEventSourceFrame: case programEventSourceFrame:
p.forwardFrame(evt.source.pathName, evt.trackId, evt.streamType, evt.buf) p.forwardFrame(evt.source.path, evt.trackId, evt.streamType, evt.buf)
case programEventTerminate: case programEventTerminate:
break outer break outer
@ -494,10 +469,7 @@ outer:
case programEventClientSetupRecord: case programEventClientSetupRecord:
evt.res <- fmt.Errorf("terminated") evt.res <- fmt.Errorf("terminated")
case programEventClientPlay1: case programEventClientPlay:
evt.res <- fmt.Errorf("terminated")
case programEventClientPlay2:
close(evt.done) close(evt.done)
case programEventClientPlayStop: case programEventClientPlayStop:
@ -564,9 +536,9 @@ func (p *program) findConfForPathName(name string) *confPath {
return nil return nil
} }
func (p *program) forwardFrame(path string, trackId int, streamType gortsplib.StreamType, frame []byte) { func (p *program) forwardFrame(path *path, trackId int, streamType gortsplib.StreamType, frame []byte) {
for c := range p.clients { for c := range p.clients {
if c.pathName != path || if c.path != path ||
c.state != clientStatePlay { c.state != clientStatePlay {
continue continue
} }

24
path.go
Unescape View File

@ -45,7 +45,7 @@ func newPath(p *program, name string, confp *confPath, permanent bool) *path {
} }
if confp.Source != "record" { if confp.Source != "record" {
s := newSource(p, name, confp) s := newSource(p, pa, confp)
pa.source = s pa.source = s
pa.publisher = s pa.publisher = s
} }
@ -95,7 +95,7 @@ func (pa *path) onClose() {
func (pa *path) hasClients() bool { func (pa *path) hasClients() bool {
for c := range pa.p.clients { for c := range pa.p.clients {
if c.pathName == pa.name { if c.path == pa {
return true return true
} }
} }
@ -104,7 +104,7 @@ func (pa *path) hasClients() bool {
func (pa *path) hasClientsWaitingDescribe() bool { func (pa *path) hasClientsWaitingDescribe() bool {
for c := range pa.p.clients { for c := range pa.p.clients {
if c.state == clientStateWaitingDescription && c.pathName == pa.name { if c.state == clientStateWaitingDescription && c.path == pa {
return true return true
} }
} }
@ -113,7 +113,7 @@ func (pa *path) hasClientsWaitingDescribe() bool {
func (pa *path) hasClientReaders() bool { func (pa *path) hasClientReaders() bool {
for c := range pa.p.clients { for c := range pa.p.clients {
if c.pathName == pa.name && c != pa.publisher { if c.path == pa && c != pa.publisher {
return true return true
} }
} }
@ -126,8 +126,8 @@ func (pa *path) onCheck() {
time.Since(pa.lastDescribeActivation) >= describeTimeout { time.Since(pa.lastDescribeActivation) >= describeTimeout {
for c := range pa.p.clients { for c := range pa.p.clients {
if c.state == clientStateWaitingDescription && if c.state == clientStateWaitingDescription &&
c.pathName == pa.name { c.path == pa {
c.pathName = "" c.path = nil
c.state = clientStateInitial c.state = clientStateInitial
c.describeRes <- describeRes{nil, fmt.Errorf("publisher of path '%s' has timed out", pa.name)} c.describeRes <- describeRes{nil, fmt.Errorf("publisher of path '%s' has timed out", pa.name)}
} }
@ -169,7 +169,7 @@ func (pa *path) onPublisherNew(client *client, sdpText []byte, sdpParsed *sdp.Se
pa.publisherSdpText = sdpText pa.publisherSdpText = sdpText
pa.publisherSdpParsed = sdpParsed pa.publisherSdpParsed = sdpParsed
client.pathName = pa.name client.path = pa
client.state = clientStateAnnounce client.state = clientStateAnnounce
} }
@ -183,8 +183,8 @@ func (pa *path) onPublisherSetReady() {
// reply to all clients that are waiting for a description // reply to all clients that are waiting for a description
for c := range pa.p.clients { for c := range pa.p.clients {
if c.state == clientStateWaitingDescription && if c.state == clientStateWaitingDescription &&
c.pathName == pa.name { c.path == pa {
c.pathName = "" c.path = nil
c.state = clientStateInitial c.state = clientStateInitial
c.describeRes <- describeRes{pa.publisherSdpText, nil} c.describeRes <- describeRes{pa.publisherSdpText, nil}
} }
@ -198,7 +198,7 @@ func (pa *path) onPublisherSetNotReady() {
for c := range pa.p.clients { for c := range pa.p.clients {
if c.state != clientStateWaitingDescription && if c.state != clientStateWaitingDescription &&
c != pa.publisher && c != pa.publisher &&
c.pathName == pa.name { c.path == pa {
c.conn.NetConn().Close() c.conn.NetConn().Close()
} }
} }
@ -227,7 +227,7 @@ func (pa *path) onDescribe(client *client) {
} }
} }
client.pathName = pa.name client.path = pa
client.state = clientStateWaitingDescription client.state = clientStateWaitingDescription
// no on-demand: reply with 404 // no on-demand: reply with 404
@ -244,7 +244,7 @@ func (pa *path) onDescribe(client *client) {
pa.source.events <- sourceEventApplyState{pa.source.state} pa.source.events <- sourceEventApplyState{pa.source.state}
} }
client.pathName = pa.name client.path = pa
client.state = clientStateWaitingDescription client.state = clientStateWaitingDescription
// publisher was found and is ready // publisher was found and is ready

12
source.go
Unescape View File

@ -39,7 +39,7 @@ func (sourceEventTerminate) isSourceEvent() {}
type source struct { type source struct {
p *program p *program
pathName string path *path
confp *confPath confp *confPath
state sourceState state sourceState
tracks []*gortsplib.Track tracks []*gortsplib.Track
@ -48,10 +48,10 @@ type source struct {
done chan struct{} done chan struct{}
} }
func newSource(p *program, pathName string, confp *confPath) *source { func newSource(p *program, path *path, confp *confPath) *source {
s := &source{ s := &source{
p: p, p: p,
pathName: pathName, path: path,
confp: confp, confp: confp,
events: make(chan sourceEvent), events: make(chan sourceEvent),
done: make(chan struct{}), done: make(chan struct{}),
@ -67,7 +67,7 @@ func newSource(p *program, pathName string, confp *confPath) *source {
} }
func (s *source) log(format string, args ...interface{}) { func (s *source) log(format string, args ...interface{}) {
s.p.log("[source "+s.pathName+"] "+format, args...) s.p.log("[source "+s.path.name+"] "+format, args...)
} }
func (s *source) isPublisher() {} func (s *source) isPublisher() {}
@ -188,8 +188,8 @@ func (s *source) doInner(terminate chan struct{}) bool {
serverSdpParsed, serverSdpText := sdpForServer(tracks) serverSdpParsed, serverSdpText := sdpForServer(tracks)
s.tracks = tracks s.tracks = tracks
s.p.paths[s.pathName].publisherSdpText = serverSdpText s.path.publisherSdpText = serverSdpText
s.p.paths[s.pathName].publisherSdpParsed = serverSdpParsed s.path.publisherSdpParsed = serverSdpParsed
if s.confp.sourceProtocolParsed == gortsplib.StreamProtocolUdp { if s.confp.sourceProtocolParsed == gortsplib.StreamProtocolUdp {
return s.runUdp(terminate, conn) return s.runUdp(terminate, conn)

Write Preview
Loading…
Cancel
Save