fix possible deadlock when communicating with clients

go.mod

@@ -5,7 +5,7 @@ go 1.12
 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-20200829164650-64af75a6b5f6
+	github.com/aler9/gortsplib v0.0.0-20200831072723-57eae89cc552
 	github.com/aler9/sdp/v3 v3.0.0-20200719093237-2c3d108a7436
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/stretchr/testify v1.6.1

go.sum

@@ -2,8 +2,8 @@ github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 h1:JYp7IbQjafo
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d h1:UQZhZ2O0vMHr2cI+DC1Mbh0TJxzA3RcLoMsFw+aXw7E=
 github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
-github.com/aler9/gortsplib v0.0.0-20200829164650-64af75a6b5f6 h1:orZ3RyemnuPUk8K6Wiwga836KVKF64yQR4/gpA3xXu8=
-github.com/aler9/gortsplib v0.0.0-20200829164650-64af75a6b5f6/go.mod h1:kBMvjIdOHRjLdV+oT28JD72JUPpJuwxOc9u72GG8GpY=
+github.com/aler9/gortsplib v0.0.0-20200831072723-57eae89cc552 h1:iSF9Byglyx1yqa32kve+6V49wnfI1PB9ciSUBEUO0b0=
+github.com/aler9/gortsplib v0.0.0-20200831072723-57eae89cc552/go.mod h1:kBMvjIdOHRjLdV+oT28JD72JUPpJuwxOc9u72GG8GpY=
 github.com/aler9/sdp/v3 v3.0.0-20200719093237-2c3d108a7436 h1:W0iNErWKvSAyJBNVx+qQoyFrWOFVgS6f/WEME/D3EZc=
 github.com/aler9/sdp/v3 v3.0.0-20200719093237-2c3d108a7436/go.mod h1:OnlEK3QI7YtM+ShZWtGajmOHLZ3bjU80AcIS5e34i1U=
 github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=

main.go

@@ -43,7 +43,6 @@ type programEventClientNew struct {
 func (programEventClientNew) isProgramEvent() {}
 
 type programEventClientClose struct {
-	done   chan struct{}
 	client *client
 }
 
@@ -75,41 +74,18 @@ type programEventClientSetupPlay struct {
 
 func (programEventClientSetupPlay) isProgramEvent() {}
 
-type programEventClientSetupRecord struct {
-	res    chan error
-	client *client
-}
-
-func (programEventClientSetupRecord) isProgramEvent() {}
-
 type programEventClientPlay struct {
-	done   chan struct{}
 	client *client
 }
 
 func (programEventClientPlay) isProgramEvent() {}
 
-type programEventClientPlayStop struct {
-	done   chan struct{}
-	client *client
-}
-
-func (programEventClientPlayStop) isProgramEvent() {}
-
 type programEventClientRecord struct {
-	done   chan struct{}
 	client *client
 }
 
 func (programEventClientRecord) isProgramEvent() {}
 
-type programEventClientRecordStop struct {
-	done   chan struct{}
-	client *client
-}
-
-func (programEventClientRecordStop) isProgramEvent() {}
-
 type programEventClientFrameUdp struct {
 	addr       *net.UDPAddr
 	streamType gortsplib.StreamType
@@ -312,14 +288,10 @@ outer:
 				c.log("connected")
 
 			case programEventClientClose:
-				delete(p.clients, evt.client)
-
-				if evt.client.path != nil && evt.client.path.publisher == evt.client {
-					evt.client.path.onPublisherRemove()
+				if _, ok := p.clients[evt.client]; !ok {
+					continue
 				}
-
-				evt.client.log("disconnected")
-				close(evt.done)
+				p.closeClient(evt.client)
 
 			case programEventClientDescribe:
 				// create path if not exist
@@ -341,7 +313,12 @@ outer:
 					}
 				}
 
-				p.paths[evt.pathName].onPublisherNew(evt.client, evt.sdpText, evt.sdpParsed)
+				p.paths[evt.pathName].publisher = evt.client
+				p.paths[evt.pathName].publisherSdpText = evt.sdpText
+				p.paths[evt.pathName].publisherSdpParsed = evt.sdpParsed
+
+				evt.client.path = p.paths[evt.pathName]
+				evt.client.state = clientStatePreRecord
 				evt.res <- nil
 
 			case programEventClientSetupPlay:
@@ -360,19 +337,9 @@ outer:
 				evt.client.state = clientStatePrePlay
 				evt.res <- nil
 
-			case programEventClientSetupRecord:
-				evt.client.state = clientStatePreRecord
-				evt.res <- nil
-
 			case programEventClientPlay:
 				p.readerCount += 1
 				evt.client.state = clientStatePlay
-				close(evt.done)
-
-			case programEventClientPlayStop:
-				p.readerCount -= 1
-				evt.client.state = clientStatePrePlay
-				close(evt.done)
 
 			case programEventClientRecord:
 				p.publisherCount += 1
@@ -397,22 +364,6 @@ outer:
 				}
 
 				evt.client.path.onPublisherSetReady()
-				close(evt.done)
-
-			case programEventClientRecordStop:
-				p.publisherCount -= 1
-				evt.client.state = clientStatePreRecord
-				if evt.client.streamProtocol == gortsplib.StreamProtocolUdp {
-					for _, track := range evt.client.streamTracks {
-						key := makeUdpClientAddr(evt.client.ip(), track.rtpPort)
-						delete(p.udpClientsByAddr, key)
-
-						key = makeUdpClientAddr(evt.client.ip(), track.rtcpPort)
-						delete(p.udpClientsByAddr, key)
-					}
-				}
-				evt.client.path.onPublisherSetNotReady()
-				close(evt.done)
 
 			case programEventClientFrameUdp:
 				pub, ok := p.udpClientsByAddr[makeUdpClientAddr(evt.addr.IP, evt.addr.Port)]
@@ -454,32 +405,11 @@ outer:
 			case programEventMetrics:
 				evt.res <- nil
 
-			case programEventClientClose:
-				close(evt.done)
-
-			case programEventClientDescribe:
-				evt.client.describeRes <- describeRes{nil, fmt.Errorf("terminated")}
-
 			case programEventClientAnnounce:
 				evt.res <- fmt.Errorf("terminated")
 
 			case programEventClientSetupPlay:
 				evt.res <- fmt.Errorf("terminated")
-
-			case programEventClientSetupRecord:
-				evt.res <- fmt.Errorf("terminated")
-
-			case programEventClientPlay:
-				close(evt.done)
-
-			case programEventClientPlayStop:
-				close(evt.done)
-
-			case programEventClientRecord:
-				close(evt.done)
-
-			case programEventClientRecordStop:
-				close(evt.done)
 			}
 		}
 	}()
@@ -499,7 +429,7 @@ outer:
 	}
 
 	for c := range p.clients {
-		c.conn.NetConn().Close()
+		p.closeClient(c)
 		<-c.done
 	}
 
@@ -536,6 +466,38 @@ func (p *program) findConfForPathName(name string) *confPath {
 	return nil
 }
 
+func (p *program) closeClient(client *client) {
+	delete(p.clients, client)
+
+	switch client.state {
+	case clientStatePlay:
+		p.readerCount -= 1
+
+	case clientStateRecord:
+		p.publisherCount -= 1
+
+		if client.streamProtocol == gortsplib.StreamProtocolUdp {
+			for _, track := range client.streamTracks {
+				key := makeUdpClientAddr(client.ip(), track.rtpPort)
+				delete(p.udpClientsByAddr, key)
+
+				key = makeUdpClientAddr(client.ip(), track.rtcpPort)
+				delete(p.udpClientsByAddr, key)
+			}
+		}
+
+		client.path.onPublisherSetNotReady()
+	}
+
+	if client.path != nil && client.path.publisher == client {
+		client.path.onPublisherRemove()
+	}
+
+	close(client.terminate)
+
+	client.log("disconnected")
+}
+
 func (p *program) forwardFrame(path *path, trackId int, streamType gortsplib.StreamType, frame []byte) {
 	for c := range p.clients {
 		if c.path != path ||
@@ -570,12 +532,10 @@ func (p *program) forwardFrame(path *path, trackId int, streamType gortsplib.Str
 			}
 
 		} else {
-			c.events <- clientEventFrameTcp{
-				frame: &gortsplib.InterleavedFrame{
-					TrackId:    trackId,
-					StreamType: streamType,
-					Content:    frame,
-				},
+			c.tcpFrame <- &gortsplib.InterleavedFrame{
+				TrackId:    trackId,
+				StreamType: streamType,
+				Content:    frame,
 			}
 		}
 	}

path.go

@@ -76,7 +76,7 @@ func (pa *path) onInit() {
 
 func (pa *path) onClose() {
 	if pa.source != nil {
-		pa.source.events <- sourceEventTerminate{}
+		close(pa.source.terminate)
 		<-pa.source.done
 	}
 
@@ -91,6 +91,18 @@ func (pa *path) onClose() {
 		pa.onDemandCmd.Process.Signal(os.Interrupt)
 		pa.onDemandCmd.Wait()
 	}
+
+	for c := range pa.p.clients {
+		if c.path == pa {
+			if c.state == clientStateWaitDescription {
+				c.path = nil
+				c.state = clientStateInitial
+				c.describe <- describeRes{nil, fmt.Errorf("publisher of path '%s' has timed out", pa.name)}
+			} else {
+				pa.p.closeClient(c)
+			}
+		}
+	}
 }
 
 func (pa *path) hasClients() bool {
@@ -104,7 +116,7 @@ func (pa *path) hasClients() bool {
 
 func (pa *path) hasClientsWaitingDescribe() bool {
 	for c := range pa.p.clients {
-		if c.state == clientStateWaitingDescription && c.path == pa {
+		if c.state == clientStateWaitDescription && c.path == pa {
 			return true
 		}
 	}
@@ -125,11 +137,11 @@ func (pa *path) onCheck() {
 	if pa.hasClientsWaitingDescribe() &&
 		time.Since(pa.lastDescribeActivation) >= describeTimeout {
 		for c := range pa.p.clients {
-			if c.state == clientStateWaitingDescription &&
+			if c.state == clientStateWaitDescription &&
 				c.path == pa {
 				c.path = nil
 				c.state = clientStateInitial
-				c.describeRes <- describeRes{nil, fmt.Errorf("publisher of path '%s' has timed out", pa.name)}
+				c.describe <- describeRes{nil, fmt.Errorf("publisher of path '%s' has timed out", pa.name)}
 			}
 		}
 	}
@@ -142,7 +154,7 @@ func (pa *path) onCheck() {
 		time.Since(pa.lastDescribeReq) >= sourceStopAfterDescribeSecs {
 		pa.source.log("stopping since we're not requested anymore")
 		pa.source.state = sourceStateStopped
-		pa.source.events <- sourceEventApplyState{pa.source.state}
+		pa.source.setState <- pa.source.state
 	}
 
 	// stop on demand command if needed
@@ -164,17 +176,17 @@ func (pa *path) onCheck() {
 	}
 }
 
-func (pa *path) onPublisherNew(client *client, sdpText []byte, sdpParsed *sdp.SessionDescription) {
-	pa.publisher = client
-	pa.publisherSdpText = sdpText
-	pa.publisherSdpParsed = sdpParsed
-
-	client.path = pa
-	client.state = clientStateAnnounce
-}
-
 func (pa *path) onPublisherRemove() {
 	pa.publisher = nil
+
+	// close all clients that are reading or waiting for reading
+	for c := range pa.p.clients {
+		if c.path == pa &&
+			c.state != clientStateWaitDescription &&
+			c != pa.publisher {
+			pa.p.closeClient(c)
+		}
+	}
 }
 
 func (pa *path) onPublisherSetReady() {
@@ -182,11 +194,11 @@ func (pa *path) onPublisherSetReady() {
 
 	// reply to all clients that are waiting for a description
 	for c := range pa.p.clients {
-		if c.state == clientStateWaitingDescription &&
+		if c.state == clientStateWaitDescription &&
 			c.path == pa {
 			c.path = nil
 			c.state = clientStateInitial
-			c.describeRes <- describeRes{pa.publisherSdpText, nil}
+			c.describe <- describeRes{pa.publisherSdpText, nil}
 		}
 	}
 }
@@ -194,12 +206,12 @@ func (pa *path) onPublisherSetReady() {
 func (pa *path) onPublisherSetNotReady() {
 	pa.publisherReady = false
 
-	// close all clients that are reading
+	// close all clients that are reading or waiting for reading
 	for c := range pa.p.clients {
-		if c.state != clientStateWaitingDescription &&
-			c != pa.publisher &&
-			c.path == pa {
-			c.conn.NetConn().Close()
+		if c.path == pa &&
+			c.state != clientStateWaitDescription &&
+			c != pa.publisher {
+			pa.p.closeClient(c)
 		}
 	}
 }
@@ -228,11 +240,11 @@ func (pa *path) onDescribe(client *client) {
 			}
 
 			client.path = pa
-			client.state = clientStateWaitingDescription
+			client.state = clientStateWaitDescription
 
 			// no on-demand: reply with 404
 		} else {
-			client.describeRes <- describeRes{nil, fmt.Errorf("no one is publishing on path '%s'", pa.name)}
+			client.describe <- describeRes{nil, fmt.Errorf("no one is publishing on path '%s'", pa.name)}
 		}
 
 		// publisher was found but is not ready: put the client on hold
@@ -241,14 +253,14 @@ func (pa *path) onDescribe(client *client) {
 			pa.source.log("starting on demand")
 			pa.lastDescribeActivation = time.Now()
 			pa.source.state = sourceStateRunning
-			pa.source.events <- sourceEventApplyState{pa.source.state}
+			pa.source.setState <- pa.source.state
 		}
 
 		client.path = pa
-		client.state = clientStateWaitingDescription
+		client.state = clientStateWaitDescription
 
 		// publisher was found and is ready
 	} else {
-		client.describeRes <- describeRes{pa.publisherSdpText, nil}
+		client.describe <- describeRes{pa.publisherSdpText, nil}
 	}
 }

source.go

@@ -23,20 +23,6 @@ const (
 	sourceStateRunning
 )
 
-type sourceEvent interface {
-	isSourceEvent()
-}
-
-type sourceEventApplyState struct {
-	state sourceState
-}
-
-func (sourceEventApplyState) isSourceEvent() {}
-
-type sourceEventTerminate struct{}
-
-func (sourceEventTerminate) isSourceEvent() {}
-
 type source struct {
 	p      *program
 	path   *path
@@ -44,17 +30,19 @@ type source struct {
 	state  sourceState
 	tracks []*gortsplib.Track
 
-	events chan sourceEvent
-	done   chan struct{}
+	setState  chan sourceState
+	terminate chan struct{}
+	done      chan struct{}
 }
 
 func newSource(p *program, path *path, confp *confPath) *source {
 	s := &source{
-		p:      p,
-		path:   path,
-		confp:  confp,
-		events: make(chan sourceEvent),
-		done:   make(chan struct{}),
+		p:         p,
+		path:      path,
+		confp:     confp,
+		setState:  make(chan sourceState),
+		terminate: make(chan struct{}),
+		done:      make(chan struct{}),
 	}
 
 	if confp.SourceOnDemand {
@@ -99,12 +87,12 @@ func (s *source) run() {
 	applyState(s.state)
 
 outer:
-	for rawEvt := range s.events {
-		switch evt := rawEvt.(type) {
-		case sourceEventApplyState:
-			applyState(evt.state)
+	for {
+		select {
+		case state := <-s.setState:
+			applyState(state)
 
-		case sourceEventTerminate:
+		case <-s.terminate:
 			break outer
 		}
 	}
@@ -114,6 +102,7 @@ outer:
 		<-doDone
 	}
 
+	close(s.setState)
 	close(s.done)
 }