From 21ac2413a22fe17f560da2879dce0a8f79807451 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Mon, 11 Jan 2021 00:07:11 -0800 Subject: [PATCH] catch panics to ensure reliability --- pkg/rtc/datatrack.go | 5 +--- pkg/rtc/mediatrack.go | 34 ++++++++++++++----------- pkg/rtc/participant.go | 50 ++++++------------------------------- pkg/rtc/room.go | 7 +++--- pkg/rtc/types/interfaces.go | 1 - pkg/rtc/utils.go | 11 ++++++++ pkg/service/rtc.go | 2 +- 7 files changed, 43 insertions(+), 67 deletions(-) diff --git a/pkg/rtc/datatrack.go b/pkg/rtc/datatrack.go index 4bde3e56e..103f0b6fb 100644 --- a/pkg/rtc/datatrack.go +++ b/pkg/rtc/datatrack.go @@ -67,10 +67,6 @@ func (t *DataTrack) Name() string { return t.name } -func (t *DataTrack) SetName(name string) { - t.name = name -} - // DataTrack cannot be muted func (t *DataTrack) IsMuted() bool { return false @@ -119,6 +115,7 @@ func (t *DataTrack) RemoveAllSubscribers() { } func (t *DataTrack) forwardWorker() { + defer Recover() defer func() { t.RemoveAllSubscribers() }() diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index d885ec978..891a88426 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -36,10 +36,11 @@ type MediaTrack struct { participantId string muted bool - ssrc webrtc.SSRC - name string - kind livekit.TrackType - codec webrtc.RTPCodecParameters + ssrc webrtc.SSRC + name string + streamID string + kind livekit.TrackType + codec webrtc.RTPCodecParameters // channel to send RTCP packets to the source rtcpCh chan []rtcp.Packet @@ -59,7 +60,7 @@ func NewMediaTrack(trackId string, pId string, rtcpCh chan []rtcp.Packet, track id: trackId, participantId: pId, ssrc: track.SSRC(), - name: track.StreamID(), + streamID: track.StreamID(), kind: ToProtoTrackKind(track.Kind()), codec: track.Codec(), rtcpCh: rtcpCh, @@ -92,10 +93,6 @@ func (t *MediaTrack) Name() string { return t.name } -func (t *MediaTrack) SetName(name string) { - t.name = name -} - func (t *MediaTrack) IsMuted() bool { return t.muted } @@ -114,7 +111,7 @@ func (t *MediaTrack) AddSubscriber(participant types.Participant) error { Channels: codec.Channels, SDPFmtpLine: codec.SDPFmtpLine, RTCPFeedback: feedbackTypes, - }, packedId, t.Name()) + }, packedId, t.streamID) if err != nil { return err } @@ -134,18 +131,23 @@ func (t *MediaTrack) AddSubscriber(participant types.Participant) error { t.handleRTCP(outTrack, pkt) }) } - go t.sendDownTrackBindingReports(participant.ID(), participant.RTCPChan()) + t.sendDownTrackBindingReports(participant.ID(), participant.RTCPChan()) }) outTrack.OnCloseHandler(func() { t.lock.Lock() delete(t.downtracks, participant.ID()) t.lock.Unlock() + // ignore if the subscribing participant disconnected if participant.PeerConnection().ConnectionState() == webrtc.PeerConnectionStateClosed { return } sender := transceiver.Sender() if sender != nil { + logger.GetLogger().Debugw("removing peerconnection track", + "track", t.id, + "srcParticipant", t.participantId, + "destParticipant", participant.ID()) if err := participant.PeerConnection().RemoveTrack(sender); err != nil { if _, ok := err.(*rtcerr.InvalidStateError); !ok { logger.GetLogger().Warnw("could not remove remoteTrack from forwarder", @@ -155,9 +157,9 @@ func (t *MediaTrack) AddSubscriber(participant types.Participant) error { } } - participant.RemoveDownTrack(t.Name(), outTrack) + participant.RemoveDownTrack(t.streamID, outTrack) }) - participant.AddDownTrack(t.Name(), outTrack) + participant.AddDownTrack(t.streamID, outTrack) t.lock.Lock() defer t.lock.Unlock() @@ -191,8 +193,8 @@ func (t *MediaTrack) sendDownTrackBindingReports(participantId string, rtcpCh ch var sd []rtcp.SourceDescriptionChunk t.lock.RLock() - defer t.lock.RUnlock() dt := t.downtracks[participantId] + t.lock.RUnlock() if !dt.IsBound() { return } @@ -206,6 +208,7 @@ func (t *MediaTrack) sendDownTrackBindingReports(participantId string, rtcpCh ch } go func() { + defer RecoverSilent() batch := pkts i := 0 for { @@ -221,8 +224,8 @@ func (t *MediaTrack) sendDownTrackBindingReports(participantId string, rtcpCh ch // b reads from the receiver and writes to each sender func (t *MediaTrack) forwardRTPWorker() { + defer Recover() defer func() { - logger.GetLogger().Debugw("stopping forward RTP worker") t.RemoveAllSubscribers() // TODO: send unpublished events? t.nackWorker.Stop() @@ -276,6 +279,7 @@ func (t *MediaTrack) forwardRTPWorker() { } func (t *MediaTrack) handleRTCP(dt *sfu.DownTrack, rtcpBuf []byte) { + defer Recover() pkts, err := rtcp.Unmarshal(rtcpBuf) if err != nil { logger.GetLogger().Warnw("could not decode RTCP packet", "err", err) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 4ef8c3cc8..1bf61556c 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -410,9 +410,6 @@ func (p *ParticipantImpl) AddDownTrack(streamId string, dt *sfu.DownTrack) { p.lock.Lock() p.subscribedTracks[streamId] = append(p.subscribedTracks[streamId], dt) p.lock.Unlock() - //dt.OnBind(func() { - // go p.sendDownTrackBindingReports(streamId) - //}) } func (p *ParticipantImpl) RemoveDownTrack(streamId string, dt *sfu.DownTrack) { @@ -481,6 +478,7 @@ func (p *ParticipantImpl) updateState(state livekit.ParticipantInfo_State) { if p.onStateChange != nil { go func() { + defer Recover() p.onStateChange(p, oldState) }() } @@ -498,6 +496,7 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w // create ReceiverImpl receiver := NewReceiver(p.rtcpCh, rtpReceiver, track, p.receiverConfig) mt := NewMediaTrack(ti.Sid, p.id, p.rtcpCh, track, receiver) + mt.name = ti.Name p.handleTrackPublished(mt) } @@ -515,6 +514,7 @@ func (p *ParticipantImpl) onDataChannel(dc *webrtc.DataChannel) { } dt := NewDataTrack(ti.Sid, p.id, dc) + dt.name = ti.Name p.handleTrackPublished(dt) } @@ -534,57 +534,20 @@ func (p *ParticipantImpl) popPendingTrack(clientId string) *livekit.TrackInfo { func (p *ParticipantImpl) handleTrackPublished(track types.PublishedTrack) { // fill in p.lock.Lock() - defer p.lock.Unlock() p.publishedTracks[track.ID()] = track + p.lock.Unlock() track.Start() if p.onTrackPublished != nil { - go p.onTrackPublished(p, track) + p.onTrackPublished(p, track) } } -func (p *ParticipantImpl) scheduleDownTrackBindingReports(streamId string) { - var sd []rtcp.SourceDescriptionChunk - - p.lock.RLock() - dts := p.subscribedTracks[streamId] - for _, dt := range dts { - if !dt.IsBound() { - continue - } - chunks := dt.CreateSourceDescriptionChunks() - if chunks != nil { - sd = append(sd, chunks...) - } - } - p.lock.RUnlock() - - pkts := []rtcp.Packet{ - &rtcp.SourceDescription{Chunks: sd}, - } - - go func() { - batch := pkts - i := 0 - for { - if err := p.peerConn.WriteRTCP(batch); err != nil { - logger.GetLogger().Debugw("error sending track binding reports", - "participant", p.id, - "err", err) - } - if i > 5 { - return - } - i++ - time.Sleep(20 * time.Millisecond) - } - }() -} - // downTracksRTCPWorker sends SenderReports periodically when the participant is subscribed to // other publishedTracks in the room. func (p *ParticipantImpl) downTracksRTCPWorker() { + defer Recover() for { time.Sleep(5 * time.Second) @@ -630,6 +593,7 @@ func (p *ParticipantImpl) downTracksRTCPWorker() { } func (p *ParticipantImpl) rtcpSendWorker() { + defer Recover() // read from rtcpChan for pkts := range p.rtcpCh { //for _, pkt := range pkts { diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 118a9712f..20606f437 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -90,6 +90,7 @@ func (r *Room) Join(participant types.Participant) error { p.Start() } }) + participant.OnTrackUpdated(r.onTrackUpdated) log.Infow("new participant joined", "id", participant.ID(), @@ -116,10 +117,9 @@ func (r *Room) RemoveParticipant(id string) { if p, ok := r.participants[id]; ok { // avoid blocking lock go func() { + Recover() // also stop connection if needed p.Close() - // update clients - r.broadcastParticipantState(p) }() } @@ -158,10 +158,11 @@ func (r *Room) onTrackAdded(participant types.Participant, track types.Published } } -func (r *Room) onTrackMuted(p types.Participant, track types.PublishedTrack) { +func (r *Room) onTrackUpdated(p types.Participant, track types.PublishedTrack) { r.broadcastParticipantState(p) } +// broadcast an update about participant p func (r *Room) broadcastParticipantState(p types.Participant) { r.lock.RLock() defer r.lock.RUnlock() diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index e1fbfc4a1..55b65896c 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -96,7 +96,6 @@ type PublishedTrack interface { ID() string Kind() livekit.TrackType Name() string - SetName(name string) IsMuted() bool AddSubscriber(participant Participant) error RemoveSubscriber(participantId string) diff --git a/pkg/rtc/utils.go b/pkg/rtc/utils.go index 7b333f0bd..3b2a8eb37 100644 --- a/pkg/rtc/utils.go +++ b/pkg/rtc/utils.go @@ -7,6 +7,7 @@ import ( "github.com/pion/webrtc/v3" + "github.com/livekit/livekit-server/pkg/logger" "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/proto/livekit" ) @@ -110,3 +111,13 @@ func ToProtoTrackKind(kind webrtc.RTPCodecType) livekit.TrackType { func IsEOF(err error) bool { return err == io.ErrClosedPipe || err == io.EOF } + +func RecoverSilent() { + recover() +} + +func Recover() { + if r := recover(); r != nil { + logger.GetLogger().Errorw("recovered panic", "err", r) + } +} diff --git a/pkg/service/rtc.go b/pkg/service/rtc.go index c47c23f1a..039bd79e5 100644 --- a/pkg/service/rtc.go +++ b/pkg/service/rtc.go @@ -106,7 +106,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { // remove peer from room upon disconnection room.RemoveParticipant(participant.ID()) participant.Close() - log.Infow("WS connection closed") + log.Infow("WS connection closed", "participant", participant.ID()) }() // read connection and wait for commands