dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-28 02:54:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

catch panics to ensure reliability

Browse Source
This commit is contained in:
David Zhao
2021-01-11 00:07:11 -08:00
parent 6464ae3cd6
commit 21ac2413a2
7 changed files with 43 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/datatrack.go
+1 -4
View File
@@ -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()
}()
pkg/rtc/mediatrack.go
+19 -15
View File
@@ -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)
pkg/rtc/participant.go
+7 -43
View File
@@ -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 {
pkg/rtc/room.go
+4 -3
View File
@@ -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()
pkg/rtc/types/interfaces.go
-1
View File
@@ -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)
pkg/rtc/utils.go
+11
View File
@@ -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)
}
}
pkg/service/rtc.go
+1 -1
View File
@@ -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