From 6b6f61b4df0aed3342181ec8123bfb057bff1bac Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 3 Aug 2022 23:03:10 +0530 Subject: [PATCH] Adding transport fallback (#866) * Adding transport fallback Commented out for now so that we can gather some data. * Promoting a few logs to info --- pkg/rtc/participant.go | 76 +++++++++++++++++++++++++++++------------- pkg/rtc/transport.go | 40 ++++++++++++++++++++++ pkg/sfu/forwarder.go | 2 +- 3 files changed, 94 insertions(+), 24 deletions(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 80f55d93e..bca5d56c1 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -277,8 +277,17 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { } else { p.activeCounter.Add(2) } + + primaryPC.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) { + p.handleICEStateChange(true, state) + }) + secondaryPC.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) { + p.handleICEStateChange(false, state) + }) + primaryPC.OnConnectionStateChange(p.handlePrimaryStateChange) secondaryPC.OnConnectionStateChange(p.handleSecondaryStateChange) + p.publisher.pc.OnTrack(p.onMediaTrack) p.publisher.pc.OnDataChannel(p.onDataChannel) @@ -989,7 +998,7 @@ func (p *ParticipantImpl) UpdateSubscribedTrackSettings(trackID livekit.TrackID, // AddSubscribedTrack adds a track to the participant's subscribed list func (p *ParticipantImpl) AddSubscribedTrack(subTrack types.SubscribedTrack) { - p.params.Logger.Debugw("added subscribedTrack", + p.params.Logger.Infow("added subscribedTrack", "publisherID", subTrack.PublisherID(), "publisherIdentity", subTrack.PublisherIdentity(), "trackID", subTrack.ID()) @@ -1031,7 +1040,7 @@ func (p *ParticipantImpl) AddSubscribedTrack(subTrack types.SubscribedTrack) { // RemoveSubscribedTrack removes a track to the participant's subscribed list func (p *ParticipantImpl) RemoveSubscribedTrack(subTrack types.SubscribedTrack) { - p.params.Logger.Debugw("removed subscribedTrack", + p.params.Logger.Infow("removed subscribedTrack", "publisherID", subTrack.PublisherID(), "publisherIdentity", subTrack.PublisherIdentity(), "trackID", subTrack.ID(), "kind", subTrack.DownTrack().Kind()) @@ -1283,38 +1292,57 @@ func (p *ParticipantImpl) handleDataMessage(kind livekit.DataPacket_Kind, data [ } } -func (p *ParticipantImpl) logSelectedCandidatePair(isPrimary bool) { +func (p *ParticipantImpl) getTransport(isPrimary bool) *PCTransport { pcTransport := p.publisher if (isPrimary && p.SubscriberAsPrimary()) || (!isPrimary && !p.SubscriberAsPrimary()) { pcTransport = p.subscriber } - sctp := pcTransport.pc.SCTP() - if sctp == nil { - return - } + return pcTransport +} - transport := sctp.Transport() - if transport == nil { - return - } +func (p *ParticipantImpl) handleICEConnected(isPrimary bool) { + pcTransport := p.getTransport(isPrimary) + pcTransport.SetICEConnectedAt(time.Now()) - iceTransport := transport.ICETransport() - if iceTransport == nil { - return - } - - selectedCandidatePair, err := iceTransport.GetSelectedCandidatePair() - if err != nil { + if pair, err := pcTransport.GetSelectedPair(); err != nil { pcTransport.Logger().Errorw("error getting selected ICE candidate pair", err) } else { - pcTransport.Logger().Infow("selected ICE candidate pair", "pair", selectedCandidatePair) + pcTransport.Logger().Infow("selected ICE candidate pair", "pair", pair) + } +} + +func (p *ParticipantImpl) handleConnectionFailed(isPrimary bool) { + pcTransport := p.getTransport(isPrimary) + isShort, duration := pcTransport.IsShortConnection(time.Now()) + if isShort { + // irrespective of which one fails, force TCP on both as the other one might + // fail at a different time and cause another disruption + pair, err := pcTransport.GetSelectedPair() + if err != nil { + pcTransport.Logger().Errorw("short ICE connection", err, "duration", duration) + } else { + pcTransport.Logger().Infow("short ICE connection", "pair", pair, "duration", duration) + } + /* + p.lock.Lock() + p.iceConfig.PreferSubTcp = true + p.iceConfig.PreferPubTcp = true + p.lock.Unlock() + */ + } +} + +func (p *ParticipantImpl) handleICEStateChange(isPrimary bool, state webrtc.ICEConnectionState) { + if state == webrtc.ICEConnectionStateConnected { + p.handleICEConnected(isPrimary) + } else if state == webrtc.ICEConnectionStateFailed { + p.handleConnectionFailed(isPrimary) } } func (p *ParticipantImpl) handlePrimaryStateChange(state webrtc.PeerConnectionState) { if state == webrtc.PeerConnectionStateConnected { - p.logSelectedCandidatePair(true) if !p.firstConnected.Swap(true) { p.setDowntracksConnected() } @@ -1324,6 +1352,8 @@ func (p *ParticipantImpl) handlePrimaryStateChange(state webrtc.PeerConnectionSt } p.incActiveCounter() } else if state == webrtc.PeerConnectionStateFailed { + p.handleConnectionFailed(true) + // clients support resuming of connections when websocket becomes disconnected p.closeSignalConnection() @@ -1361,9 +1391,9 @@ func (p *ParticipantImpl) handlePrimaryStateChange(state webrtc.PeerConnectionSt // for the secondary peer connection, we still need to handle when they become disconnected // instead of allowing them to silently fail. func (p *ParticipantImpl) handleSecondaryStateChange(state webrtc.PeerConnectionState) { - if state == webrtc.PeerConnectionStateConnected { - p.logSelectedCandidatePair(false) - } else if state == webrtc.PeerConnectionStateFailed { + if state == webrtc.PeerConnectionStateFailed { + p.handleConnectionFailed(false) + // clients support resuming of connections when websocket becomes disconnected p.closeSignalConnection() } diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 5049d6d23..225c80e93 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -35,6 +35,8 @@ const ( iceDisconnectedTimeout = 10 * time.Second // compatible for ice-lite with firefox client iceFailedTimeout = 25 * time.Second // pion's default iceKeepaliveInterval = 2 * time.Second // pion's default + + shortConnectionThreshold = 2 * time.Minute ) var ( @@ -61,6 +63,7 @@ type PCTransport struct { me *webrtc.MediaEngine lock sync.RWMutex + iceConnectedAt time.Time pendingCandidates []webrtc.ICECandidateInit debouncedNegotiate func(func()) negotiationPending map[livekit.ParticipantID]bool @@ -216,6 +219,43 @@ func (t *PCTransport) Logger() logger.Logger { return t.params.Logger } +func (t *PCTransport) SetICEConnectedAt(at time.Time) { + t.lock.Lock() + t.iceConnectedAt = at + t.lock.Unlock() +} + +func (t *PCTransport) IsShortConnection(at time.Time) (bool, time.Duration) { + t.lock.RLock() + defer t.lock.RUnlock() + + if t.iceConnectedAt.IsZero() { + return false, 0 + } + + duration := at.Sub(t.iceConnectedAt) + return duration < shortConnectionThreshold, duration +} + +func (t *PCTransport) GetSelectedPair() (*webrtc.ICECandidatePair, error) { + sctp := t.pc.SCTP() + if sctp == nil { + return nil, errors.New("no SCTP") + } + + dtlsTransport := sctp.Transport() + if dtlsTransport == nil { + return nil, errors.New("no DTLS transport") + } + + iceTransport := dtlsTransport.ICETransport() + if iceTransport == nil { + return nil, errors.New("no ICE transport") + } + + return iceTransport.GetSelectedCandidatePair() +} + func (t *PCTransport) createPeerConnection() error { var bwe cc.BandwidthEstimator pc, me, err := newPeerConnection(t.params, func(estimator cc.BandwidthEstimator) { diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index b4f4b172b..9685100bc 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -1275,7 +1275,7 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in if f.targetLayers.Spatial == layer { if extPkt.KeyFrame || tp.switchingToTargetLayer { // lock to target layer - f.logger.Debugw("locking to target layer", "current", f.currentLayers, "target", f.targetLayers) + f.logger.Infow("locking to target layer", "current", f.currentLayers, "target", f.targetLayers) f.currentLayers.Spatial = f.targetLayers.Spatial if !f.isTemporalSupported { f.currentLayers.Temporal = f.targetLayers.Temporal