From 1350400c3aac4c68c1cf447d690b0e6f949cfe67 Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Wed, 24 Aug 2022 20:42:56 +0800 Subject: [PATCH] fallback to turn over tls when tcp short connection happen (#950) * fallback to tls when tcp failed * go mod * magefile --- go.mod | 2 +- go.sum | 4 ++-- pkg/rtc/participant.go | 12 ++++++++-- pkg/rtc/room.go | 4 ++-- pkg/rtc/transport.go | 2 -- pkg/rtc/transportmanager.go | 46 +++++++++++++++++++++++++------------ pkg/rtc/types/interfaces.go | 12 ++++++++-- pkg/service/roommanager.go | 18 ++++++++++----- 8 files changed, 68 insertions(+), 32 deletions(-) diff --git a/go.mod b/go.mod index a74625db4..6bfa9d4cc 100644 --- a/go.mod +++ b/go.mod @@ -16,7 +16,7 @@ require ( github.com/gorilla/websocket v1.5.0 github.com/hashicorp/go-version v1.6.0 github.com/hashicorp/golang-lru v0.5.4 - github.com/livekit/protocol v1.0.2-0.20220819200418-cd39f4ca4d10 + github.com/livekit/protocol v1.0.2-0.20220824112019-a1c2809ddc67 github.com/livekit/rtcscore-go v0.0.0-20220815072451-20ee10ae1995 github.com/mackerelio/go-osstat v0.2.2 github.com/magefile/mage v1.13.0 diff --git a/go.sum b/go.sum index 81df09224..982a88b5c 100644 --- a/go.sum +++ b/go.sum @@ -240,8 +240,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/lithammer/shortuuid/v3 v3.0.7 h1:trX0KTHy4Pbwo/6ia8fscyHoGA+mf1jWbPJVuvyJQQ8= github.com/lithammer/shortuuid/v3 v3.0.7/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts= -github.com/livekit/protocol v1.0.2-0.20220819200418-cd39f4ca4d10 h1:6W0YbYSIa8xpcmYaNlYj86tnwCZTMrl8cs1Pd2y/uc4= -github.com/livekit/protocol v1.0.2-0.20220819200418-cd39f4ca4d10/go.mod h1:M+JZ29cEbUozppVg3LlDC9rZPE5d218Dft3h1qBP1qM= +github.com/livekit/protocol v1.0.2-0.20220824112019-a1c2809ddc67 h1:mh2hHoW2nMaqrEXPCy0G8utMTv8AbqFvzpetJ4YzYzw= +github.com/livekit/protocol v1.0.2-0.20220824112019-a1c2809ddc67/go.mod h1:M+JZ29cEbUozppVg3LlDC9rZPE5d218Dft3h1qBP1qM= github.com/livekit/rtcscore-go v0.0.0-20220815072451-20ee10ae1995 h1:vOaY2qvfLihDyeZtnGGN1Law9wRrw8BMGCr1TygTvMw= github.com/livekit/rtcscore-go v0.0.0-20220815072451-20ee10ae1995/go.mod h1:116ych8UaEs9vfIE8n6iZCZ30iagUFTls0vRmC+Ix5U= github.com/mackerelio/go-osstat v0.2.2 h1:7jVyXGXTkQL3+6lDVUDBY+Fpo8VQPfyOkZeXxxsXX4c= diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index b74d253cb..b5c82918f 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -233,6 +233,8 @@ func (p *ParticipantImpl) ConnectedAt() time.Time { } func (p *ParticipantImpl) GetClientConfiguration() *livekit.ClientConfiguration { + p.lock.RLock() + defer p.lock.RUnlock() return p.params.ClientConf } @@ -1035,9 +1037,15 @@ func (p *ParticipantImpl) setupTransportManager() error { } tm.OnICEConfigChanged(func(iceConfig types.IceConfig) { - p.lock.RLock() + p.lock.Lock() onICEConfigChanged := p.onICEConfigChanged - p.lock.RUnlock() + if iceConfig.PreferSub == types.PreferTls { + if p.params.ClientConf == nil { + p.params.ClientConf = &livekit.ClientConfiguration{} + } + p.params.ClientConf.ForceRelay = livekit.ClientConfigSetting_ENABLED + } + p.lock.Unlock() if onICEConfigChanged != nil { onICEConfigChanged(p, iceConfig) diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index b27663f52..cdad3a0d5 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -637,8 +637,8 @@ func (r *Room) SimulateScenario(participant types.LocalParticipant, simulateScen case *livekit.SimulateScenario_SwitchCandidateProtocol: r.Logger.Infow("simulating switch candidate protocol", "participant", participant.Identity()) participant.ICERestart(&types.IceConfig{ - PreferSubTcp: scenario.SwitchCandidateProtocol == livekit.CandidateProtocol_TCP, - PreferPubTcp: scenario.SwitchCandidateProtocol == livekit.CandidateProtocol_TCP, + PreferSub: types.PreferCandidateType(scenario.SwitchCandidateProtocol), + PreferPub: types.PreferCandidateType(scenario.SwitchCandidateProtocol), }) } return nil diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 1f0131761..817438a74 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -473,8 +473,6 @@ func (t *PCTransport) onICEConnectionStateChange(state webrtc.ICEConnectionState } else { t.params.Logger.Infow("selected ICE candidate pair", "pair", pair) } - case webrtc.ICEConnectionStateFailed: - t.handleConnectionFailed() } } diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index 0b018f67f..47bc07149 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -46,6 +46,7 @@ type TransportManager struct { pendingOfferPublisher *webrtc.SessionDescription pendingDataChannelsPublisher []*livekit.DataChannelInfo lastPublisherAnswer atomic.Value + iceConfig types.IceConfig onPublisherInitialConnected func() onSubscriberInitialConnected func() @@ -101,7 +102,6 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro } }) t.publisher.OnFailed(func(isShortLived bool) { - t.handleConnectionFailed(isShortLived) if t.onAnyTransportFailed != nil { t.onAnyTransportFailed() } @@ -422,17 +422,14 @@ func (t *TransportManager) OnICEConfigChanged(f func(iceConfig types.IceConfig)) func (t *TransportManager) SetICEConfig(iceConfig types.IceConfig) { t.params.Logger.Infow("setting ICE config", "iceConfig", iceConfig) - if iceConfig.PreferPubTcp { - t.publisher.SetPreferTCP(true) - } - if iceConfig.PreferSubTcp { - t.subscriber.SetPreferTCP(true) - } + t.publisher.SetPreferTCP(iceConfig.PreferPub == types.PreferTcp) + t.subscriber.SetPreferTCP(iceConfig.PreferSub == types.PreferTcp) - t.lock.RLock() + t.lock.Lock() onICEConfigChanged := t.onICEConfigChanged - t.lock.RUnlock() + t.iceConfig = iceConfig + t.lock.Unlock() if onICEConfigChanged != nil { onICEConfigChanged(iceConfig) @@ -456,14 +453,33 @@ func (t *TransportManager) handleConnectionFailed(isShortLived bool) { if !t.params.AllowTCPFallback || !isShortLived { return } + t.lock.RLock() + iceConfig := t.iceConfig + t.lock.RUnlock() - // irrespective of which one fails, force TCP on both as the other one might + var nextConfig types.IceConfig + // irrespective of which one fails, force prefer candidate on both as the other one might // fail at a different time and cause another disruption - t.params.Logger.Infow("restricting transport to TCP on both peer connections") - t.SetICEConfig(types.IceConfig{ - PreferPubTcp: true, - PreferSubTcp: true, - }) + switch iceConfig.PreferSub { + case types.PreferNone: + t.params.Logger.Infow("restricting transport to TCP on both peer connections") + nextConfig = types.IceConfig{ + PreferPub: types.PreferTcp, + PreferSub: types.PreferTcp, + } + + case types.PreferTcp: + t.params.Logger.Infow("prefer transport to TLS on both peer connections") + nextConfig = types.IceConfig{ + PreferPub: types.PreferTls, + PreferSub: types.PreferTls, + } + + default: + return + } + + t.SetICEConfig(nextConfig) } func (t *TransportManager) SetMigrateInfo(previousAnswer *webrtc.SessionDescription, dataChannels []*livekit.DataChannelInfo) { diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 33ec9004c..d020a945f 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -195,9 +195,17 @@ type Participant interface { DebugInfo() map[string]interface{} } +type PreferCandidateType int + +const ( + PreferNone PreferCandidateType = iota + PreferTcp + PreferTls +) + type IceConfig struct { - PreferSubTcp bool - PreferPubTcp bool + PreferSub PreferCandidateType + PreferPub PreferCandidateType } //counterfeiter:generate . LocalParticipant diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index dd915eb83..da1861e6d 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -292,13 +292,13 @@ func (r *RoomManager) StartSession( if err != nil { return err } - r.setIceConfig(participant) + iceConfig := r.setIceConfig(participant) // join room opts := rtc.ParticipantOptions{ AutoSubscribe: pi.AutoSubscribe, } - if err = room.Join(participant, &opts, r.iceServersForRoom(protoRoom)); err != nil { + if err = room.Join(participant, &opts, r.iceServersForRoom(protoRoom, iceConfig.PreferSub == types.PreferTls)); err != nil { pLogger.Errorw("could not join room", err) _ = participant.Close(true, types.ParticipantCloseReasonJoinFailed) return err @@ -577,10 +577,15 @@ func (r *RoomManager) handleRTCMessage(ctx context.Context, roomName livekit.Roo } } -func (r *RoomManager) iceServersForRoom(ri *livekit.Room) []*livekit.ICEServer { +func (r *RoomManager) iceServersForRoom(ri *livekit.Room, tlsOnly bool) []*livekit.ICEServer { var iceServers []*livekit.ICEServer rtcConf := r.config.RTC + if tlsOnly && r.config.TURN.TLSPort == 0 { + logger.Warnw("tls only enabled but no turn tls config", nil) + tlsOnly = false + } + hasSTUN := false if r.config.TURN.Enabled { var urls []string @@ -590,7 +595,7 @@ func (r *RoomManager) iceServersForRoom(ri *livekit.Room) []*livekit.ICEServer { urls = append(urls, fmt.Sprintf("turn:%s:%d?transport=udp", r.config.RTC.NodeIP, r.config.TURN.UDPPort)) } if r.config.TURN.TLSPort > 0 { - urls = append(urls, fmt.Sprintf("turns:%s:443?transport=tcp", r.config.TURN.Domain)) + urls = append(urls, fmt.Sprintf("turns:%s:%d?transport=tcp", r.config.TURN.Domain, r.config.TURN.TLSPort)) } if len(urls) > 0 { iceServers = append(iceServers, &livekit.ICEServer{ @@ -654,17 +659,18 @@ func (r *RoomManager) refreshToken(participant types.LocalParticipant) error { return nil } -func (r *RoomManager) setIceConfig(participant types.LocalParticipant) { +func (r *RoomManager) setIceConfig(participant types.LocalParticipant) types.IceConfig { r.lock.Lock() iceConfigCacheEntry, ok := r.iceConfigCache[participant.Identity()] if !ok || time.Since(iceConfigCacheEntry.modifiedAt) > iceConfigTTL { delete(r.iceConfigCache, participant.Identity()) r.lock.Unlock() - return + return types.IceConfig{} } r.lock.Unlock() participant.SetICEConfig(iceConfigCacheEntry.iceConfig) + return iceConfigCacheEntry.iceConfig } // ------------------------------------