From 7e5ba6a3b0e5d03d2c3e8fa6fb2ad899061f1d39 Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Wed, 1 Feb 2023 20:00:34 +0800 Subject: [PATCH] Improve connectivity check (#1366) * Add Timer to detect dtls failure quickly * Fix pc state check in timeout after ice * More strict conditions to switch candidate type * log for signal interuppt * typo --- pkg/rtc/signalhandler.go | 2 + pkg/rtc/transport.go | 69 ++++++++++++++++--- pkg/rtc/transportmanager.go | 33 ++++++++- pkg/rtc/types/interfaces.go | 1 + .../typesfakes/fake_local_participant.go | 30 ++++++++ pkg/service/rtcservice.go | 1 - 6 files changed, 123 insertions(+), 13 deletions(-) diff --git a/pkg/rtc/signalhandler.go b/pkg/rtc/signalhandler.go index ba73df4d6..163ee5ce6 100644 --- a/pkg/rtc/signalhandler.go +++ b/pkg/rtc/signalhandler.go @@ -8,6 +8,8 @@ import ( ) func HandleParticipantSignal(room types.Room, participant types.LocalParticipant, req *livekit.SignalRequest, pLogger logger.Logger) error { + participant.UpdateLastSeenSignal() + switch msg := req.Message.(type) { case *livekit.SignalRequest_Offer: participant.HandleOffer(FromProtoSessionDescription(msg.Offer)) diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 2117ca265..6a714efee 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -39,9 +39,11 @@ const ( negotiationFailedTimeout = 15 * time.Second dtlsRetransmissionInterval = 100 * time.Millisecond - 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 + 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 + minConnectTimeoutAfterICE = 5 * time.Second // min duration for waiting pc to connect after ICE is connected + maxConnectTimeoutAfterICE = 20 * time.Second // max duration for waiting pc to connect after ICE is connected shortConnectionThreshold = 90 * time.Second ) @@ -153,8 +155,10 @@ type PCTransport struct { lossyDCOpened bool onDataPacket func(kind livekit.DataPacket_Kind, data []byte) - iceConnectedAt time.Time - connectedAt time.Time + iceStartedAt time.Time + iceConnectedAt time.Time + connectedAt time.Time + connectAfterICETimer *time.Timer // timer to wait for pc to connect after ice connected onFullyEstablished func() @@ -397,6 +401,14 @@ func (t *PCTransport) createPeerConnection() error { return nil } +func (t *PCTransport) setICEStartedAt(at time.Time) { + t.lock.Lock() + if t.iceStartedAt.IsZero() { + t.iceStartedAt = at + } + t.lock.Unlock() +} + func (t *PCTransport) setICEConnectedAt(at time.Time) { t.lock.Lock() if t.iceConnectedAt.IsZero() { @@ -405,6 +417,24 @@ func (t *PCTransport) setICEConnectedAt(at time.Time) { // This prevents reset of connected at time if ICE goes `Connected` -> `Disconnected` -> `Connected`. // t.iceConnectedAt = at + + // set failure timer for dtls handshake + iceCost := at.Sub(t.iceStartedAt) + connTimeoutAfterICE := 3 * iceCost + if connTimeoutAfterICE < minConnectTimeoutAfterICE { + connTimeoutAfterICE = minConnectTimeoutAfterICE + } else if connTimeoutAfterICE > maxConnectTimeoutAfterICE { + connTimeoutAfterICE = maxConnectTimeoutAfterICE + } + t.params.Logger.Debugw("setting connection timer after ice connected", "timeout", connTimeoutAfterICE, "iceCost", iceCost) + t.connectAfterICETimer = time.AfterFunc(connTimeoutAfterICE, func() { + state := t.pc.ConnectionState() + // if pc is still checking or connected but not fully established after timeout, then fire connection fail + if state != webrtc.PeerConnectionStateClosed && state != webrtc.PeerConnectionStateFailed && !t.isFullyEstablished() { + t.params.Logger.Infow("connect timeout after ICE connected", "timeout", connTimeoutAfterICE, "iceCost", iceCost) + t.handleConnectionFailed() + } + }) } t.lock.Unlock() } @@ -503,6 +533,9 @@ func (t *PCTransport) onICEConnectionStateChange(state webrtc.ICEConnectionState } else { t.params.Logger.Infow("selected ICE candidate pair", "pair", pair) } + + case webrtc.ICEConnectionStateChecking: + t.setICEStartedAt(time.Now()) } } @@ -510,6 +543,7 @@ func (t *PCTransport) onPeerConnectionStateChange(state webrtc.PeerConnectionSta t.params.Logger.Debugw("peer connection state change", "state", state.String()) switch state { case webrtc.PeerConnectionStateConnected: + t.clearConnTimerAfterICE() isInitialConnection := t.setConnectedAt(time.Now()) if isInitialConnection { if onInitialConnected := t.getOnInitialConnected(); onInitialConnected != nil { @@ -520,6 +554,7 @@ func (t *PCTransport) onPeerConnectionStateChange(state webrtc.PeerConnectionSta } case webrtc.PeerConnectionStateFailed: t.params.Logger.Infow("peer connection failed") + t.clearConnTimerAfterICE() t.logICECandidates() t.handleConnectionFailed() } @@ -559,17 +594,20 @@ func (t *PCTransport) onDataChannel(dc *webrtc.DataChannel) { } func (t *PCTransport) maybeNotifyFullyEstablished() { - t.lock.RLock() - fullyEstablished := t.reliableDCOpened && t.lossyDCOpened && !t.connectedAt.IsZero() - t.lock.RUnlock() - - if fullyEstablished { + if t.isFullyEstablished() { if onFullyEstablished := t.getOnFullyEstablished(); onFullyEstablished != nil { onFullyEstablished() } } } +func (t *PCTransport) isFullyEstablished() bool { + t.lock.RLock() + fullyEstablished := t.reliableDCOpened && t.lossyDCOpened && !t.connectedAt.IsZero() + t.lock.RUnlock() + return fullyEstablished +} + func (t *PCTransport) SetPreferTCP(preferTCP bool) { t.preferTCP.Store(preferTCP) } @@ -782,6 +820,17 @@ func (t *PCTransport) Close() { } _ = t.pc.Close() + + t.clearConnTimerAfterICE() +} + +func (t *PCTransport) clearConnTimerAfterICE() { + t.lock.Lock() + defer t.lock.Unlock() + if t.connectAfterICETimer != nil { + t.connectAfterICETimer.Stop() + t.connectAfterICETimer = nil + } } func (t *PCTransport) HandleRemoteDescription(sd webrtc.SessionDescription) { diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index 45e22baee..7c4a1842b 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -4,6 +4,7 @@ import ( "math/bits" "strings" "sync" + "time" "github.com/pion/rtcp" "github.com/pion/sdp/v3" @@ -21,7 +22,8 @@ import ( ) const ( - failureCountThreshold = 2 + failureCountThreshold = 2 + preferNextByFailureWindow = time.Minute // when RR report loss percentage over this threshold, we consider it is a unstable event udpLossFracUnstable = 25 @@ -57,6 +59,8 @@ type TransportManager struct { subscriber *PCTransport failureCount int isTransportReconfigured bool + lastFailure time.Time + lastSignalAt time.Time pendingOfferPublisher *webrtc.SessionDescription pendingDataChannelsPublisher []*livekit.DataChannelInfo @@ -460,6 +464,7 @@ func (t *TransportManager) configureICE(iceConfig *livekit.ICEConfig, reset bool t.failureCount = 0 t.isTransportReconfigured = !reset t.udpLossUnstableCount = 0 + t.lastFailure = time.Time{} } if isEqual { @@ -512,6 +517,18 @@ func (t *TransportManager) handleConnectionFailed(isShortLived bool) { return } + lastSignalSince := time.Since(t.lastSignalAt) + if lastSignalSince > iceFailedTimeout { + // the failed might cause by network interrupt because we have not seen any signal in the time window too + // so don't switch to next candidate type + t.params.Logger.Infow("ignoring prefer candidate check by ICE failure because no signal received in the ice failed window", + "lastSignalSince", lastSignalSince) + t.failureCount = 0 + t.lastFailure = time.Time{} + t.lock.Unlock() + return + } + // // Checking only `PreferenceSubcriber` field although any connection failure (PUBLISHER OR SUBSCRIBER) will // flow through here. @@ -533,7 +550,9 @@ func (t *TransportManager) handleConnectionFailed(isShortLived bool) { preferNext = getNext(t.iceConfig) } else { t.failureCount++ - if t.failureCount < failureCountThreshold { + lastFailure := t.lastFailure + t.lastFailure = time.Now() + if t.failureCount < failureCountThreshold || time.Since(lastFailure) > preferNextByFailureWindow { t.lock.Unlock() return } @@ -668,5 +687,15 @@ func (t *TransportManager) UpdateRTT(rtt uint32, isUDP bool) { } } else { t.tcpRTT = rtt + + // TODO: considering using tcp rtt to calculate ice connection cost, if ice connection can't be established + // within 5 * tcp rtt(at least 5s), means udp traffic might be block/dropped, switch to tcp. + // Currently, most cases reported is that ice connected but subsequent connection, so left the thinking for now. } } + +func (t *TransportManager) UpdateLastSeenSignal() { + t.lock.Lock() + t.lastSignalAt = time.Now() + t.lock.Unlock() +} diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 6bd14bd1e..89f7f96a2 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -245,6 +245,7 @@ type LocalParticipant interface { SetResponseSink(sink routing.MessageSink) CloseSignalConnection() + UpdateLastSeenSignal() // permissions ClaimGrants() *auth.ClaimGrants diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 0dd26a810..044eba2a3 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -713,6 +713,10 @@ type FakeLocalParticipant struct { unsubscribeFromTrackArgsForCall []struct { arg1 livekit.TrackID } + UpdateLastSeenSignalStub func() + updateLastSeenSignalMutex sync.RWMutex + updateLastSeenSignalArgsForCall []struct { + } UpdateMediaLossStub func(livekit.NodeID, livekit.TrackID, uint32) error updateMediaLossMutex sync.RWMutex updateMediaLossArgsForCall []struct { @@ -4580,6 +4584,30 @@ func (fake *FakeLocalParticipant) UnsubscribeFromTrackArgsForCall(i int) livekit return argsForCall.arg1 } +func (fake *FakeLocalParticipant) UpdateLastSeenSignal() { + fake.updateLastSeenSignalMutex.Lock() + fake.updateLastSeenSignalArgsForCall = append(fake.updateLastSeenSignalArgsForCall, struct { + }{}) + stub := fake.UpdateLastSeenSignalStub + fake.recordInvocation("UpdateLastSeenSignal", []interface{}{}) + fake.updateLastSeenSignalMutex.Unlock() + if stub != nil { + fake.UpdateLastSeenSignalStub() + } +} + +func (fake *FakeLocalParticipant) UpdateLastSeenSignalCallCount() int { + fake.updateLastSeenSignalMutex.RLock() + defer fake.updateLastSeenSignalMutex.RUnlock() + return len(fake.updateLastSeenSignalArgsForCall) +} + +func (fake *FakeLocalParticipant) UpdateLastSeenSignalCalls(stub func()) { + fake.updateLastSeenSignalMutex.Lock() + defer fake.updateLastSeenSignalMutex.Unlock() + fake.UpdateLastSeenSignalStub = stub +} + func (fake *FakeLocalParticipant) UpdateMediaLoss(arg1 livekit.NodeID, arg2 livekit.TrackID, arg3 uint32) error { fake.updateMediaLossMutex.Lock() ret, specificReturn := fake.updateMediaLossReturnsOnCall[len(fake.updateMediaLossArgsForCall)] @@ -5158,6 +5186,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.uncacheDownTrackMutex.RUnlock() fake.unsubscribeFromTrackMutex.RLock() defer fake.unsubscribeFromTrackMutex.RUnlock() + fake.updateLastSeenSignalMutex.RLock() + defer fake.updateLastSeenSignalMutex.RUnlock() fake.updateMediaLossMutex.RLock() defer fake.updateMediaLossMutex.RUnlock() fake.updateRTTMutex.RLock() diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index 21f5198d6..f1b19a1bf 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -333,7 +333,6 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { if perr == nil && signalStats != nil { signalStats.AddBytes(uint64(count), true) } - continue } switch m := req.Message.(type) {