diff --git a/go.mod b/go.mod index d402591f5..76ce8f0c9 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.20220907025839-ca55e87895c4 + github.com/livekit/protocol v1.0.2-0.20220909090645-6ec04e9ca47e github.com/livekit/rtcscore-go v0.0.0-20220815072451-20ee10ae1995 github.com/mackerelio/go-osstat v0.2.3 github.com/magefile/mage v1.13.0 diff --git a/go.sum b/go.sum index fc8a5724f..bed91ce47 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.20220907025839-ca55e87895c4 h1:hS766MKj69TBYAueHx2jrv8eGK9rvigARsoTveHU2D0= -github.com/livekit/protocol v1.0.2-0.20220907025839-ca55e87895c4/go.mod h1:ykRtMmaq4blqGyLWWPtYkB/74JYsyK7N2DAXLGnfSa4= +github.com/livekit/protocol v1.0.2-0.20220909090645-6ec04e9ca47e h1:9vfn+sBf1RQDNrBT21YJTcaXB70UI73W7it3S8pQOGU= +github.com/livekit/protocol v1.0.2-0.20220909090645-6ec04e9ca47e/go.mod h1:ykRtMmaq4blqGyLWWPtYkB/74JYsyK7N2DAXLGnfSa4= 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.3 h1:jAMXD5erlDE39kdX2CU7YwCGRcxIO33u/p8+Fhe5dJw= diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index 2680d5fd9..b7f5c60ed 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -202,7 +202,7 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr * // if cannot replace, find an unused transceiver or add new one if transceiver == nil { - if sub.ProtocolVersion().SupportsTransceiverReuse() && !sub.IsNegotiationPending(subTrack.PublisherID()) { + if sub.ProtocolVersion().SupportsTransceiverReuse() { // // AddTrack will create a new transceiver or re-use an unused one // if the attributes match. This prevents SDP from bloating @@ -423,7 +423,6 @@ func (t *MediaTrackSubscriptions) downTrackClosed( sub.RemoveSubscribedTrack(subTrack) if !willBeResumed { - sub.AddNegotiationPending(subTrack.PublisherID()) sub.Negotiate(false) } } diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 36bac1694..5e67d5f1d 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -615,14 +615,18 @@ func (p *ParticipantImpl) AddTrack(req *livekit.AddTrackRequest) { p.sendTrackPublished(req.Cid, ti) } -func (p *ParticipantImpl) SetMigrateInfo(previousAnswer *webrtc.SessionDescription, mediaTracks []*livekit.TrackPublishedResponse, dataChannels []*livekit.DataChannelInfo) { +func (p *ParticipantImpl) SetMigrateInfo( + previousOffer, previousAnswer *webrtc.SessionDescription, + mediaTracks []*livekit.TrackPublishedResponse, + dataChannels []*livekit.DataChannelInfo, +) { p.pendingTracksLock.Lock() for _, t := range mediaTracks { p.pendingTracks[t.GetCid()] = &pendingTrackInfo{trackInfos: []*livekit.TrackInfo{t.GetTrack()}, migrated: true} } p.pendingTracksLock.Unlock() - p.TransportManager.SetMigrateInfo(previousAnswer, dataChannels) + p.TransportManager.SetMigrateInfo(previousOffer, previousAnswer, dataChannels) } func (p *ParticipantImpl) Start() { diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 6c8fa853f..7a6699250 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -54,6 +54,7 @@ var ( ErrNoICECandidateHandler = errors.New("no ICE candidate handler") ErrNoOfferHandler = errors.New("no offer handler") ErrNoAnswerHandler = errors.New("no answer handler") + ErrMidNotFound = errors.New("mid not found") ) // ------------------------------------------------------------------------- @@ -134,6 +135,11 @@ type SimulcastTrackInfo struct { Rid string } +type trackDescription struct { + mid string + sender *webrtc.RTPSender +} + // PCTransport is a wrapper around PeerConnection, with some helper methods type PCTransport struct { params TransportParams @@ -169,6 +175,9 @@ type PCTransport struct { streamAllocator *sfu.StreamAllocator previousAnswer *webrtc.SessionDescription + // track id -> description map in previous offer sdp + previousTrackDescription map[string]*trackDescription + canReuseTransceiver bool preferTCP atomic.Bool isClosed atomic.Bool @@ -180,7 +189,6 @@ type PCTransport struct { cacheLocalCandidates bool cachedLocalCandidates []*webrtc.ICECandidate pendingRemoteCandidates []*webrtc.ICECandidateInit - negotiationPending map[livekit.ParticipantID]bool restartAfterGathering bool restartAtNextOffer bool negotiationState NegotiationState @@ -313,11 +321,12 @@ func newPeerConnection(params TransportParams, onBandwidthEstimator func(estimat func NewPCTransport(params TransportParams) (*PCTransport, error) { t := &PCTransport{ - params: params, - debouncedNegotiate: debounce.New(negotiationFrequency), - negotiationState: NegotiationStateNone, - negotiationPending: make(map[livekit.ParticipantID]bool), - eventCh: make(chan event, 50), + params: params, + debouncedNegotiate: debounce.New(negotiationFrequency), + negotiationState: NegotiationStateNone, + eventCh: make(chan event, 50), + previousTrackDescription: make(map[string]*trackDescription), + canReuseTransceiver: true, } if params.IsSendSide { t.streamAllocator = sfu.NewStreamAllocator(sfu.StreamAllocatorParams{ @@ -548,6 +557,28 @@ func (t *PCTransport) AddICECandidate(candidate webrtc.ICECandidateInit) { } func (t *PCTransport) AddTrack(trackLocal webrtc.TrackLocal) (sender *webrtc.RTPSender, transceiver *webrtc.RTPTransceiver, err error) { + t.lock.Lock() + canReuse := t.canReuseTransceiver + td, ok := t.previousTrackDescription[trackLocal.ID()] + if ok { + delete(t.previousTrackDescription, trackLocal.ID()) + } + t.lock.Unlock() + + // keep track use same mid after migration if possible + if td != nil && td.sender != nil { + for _, tr := range t.pc.GetTransceivers() { + if tr.Mid() == td.mid { + return td.sender, tr, tr.SetSender(td.sender, trackLocal) + } + } + } + + // if never negotiated with client, can't reuse transeiver for track not subscribed before migration + if !canReuse { + return t.AddTransceiverFromTrack(trackLocal) + } + sender, err = t.pc.AddTrack(trackLocal) if err != nil { return @@ -871,12 +902,6 @@ func (t *PCTransport) getOnNegotiationFailed() func() { return t.onNegotiationFailed } -func (t *PCTransport) AddNegotiationPending(publisherID livekit.ParticipantID) { - t.lock.Lock() - t.negotiationPending[publisherID] = true - t.lock.Unlock() -} - func (t *PCTransport) Negotiate(force bool) { if force { t.lock.Lock() @@ -907,12 +932,6 @@ func (t *PCTransport) Negotiate(force bool) { } } -func (t *PCTransport) IsNegotiationPending(publisherID livekit.ParticipantID) bool { - t.lock.RLock() - defer t.lock.RUnlock() - return t.negotiationPending[publisherID] -} - func (t *PCTransport) configureReceiverDTX(enableDTX bool) { // // DTX (Discontinuous Transmission) allows audio bandwidth saving @@ -1145,10 +1164,11 @@ func (t *PCTransport) preparePC(previousAnswer webrtc.SessionDescription) error return t.pc.SetRemoteDescription(ans) } -func (t *PCTransport) initPCWithPreviousAnswer(previousAnswer webrtc.SessionDescription) error { +func (t *PCTransport) initPCWithPreviousAnswer(previousAnswer webrtc.SessionDescription) (map[string]*webrtc.RTPSender, error) { + senders := make(map[string]*webrtc.RTPSender) parsed, err := previousAnswer.Unmarshal() if err != nil { - return err + return senders, err } for _, m := range parsed.MediaDescriptions { var codecType webrtc.RTPCodecType @@ -1164,31 +1184,37 @@ func (t *PCTransport) initPCWithPreviousAnswer(previousAnswer webrtc.SessionDesc // so use a dumb pc to negotiate sdp to fixed the datachannel's mid at same position with previous answer if err := t.preparePC(previousAnswer); err != nil { t.params.Logger.Errorw("prepare pc for migration failed", err) - return err + return senders, err } continue default: continue } - tr, err := t.pc.AddTransceiverFromKind(codecType, webrtc.RTPTransceiverInit{Direction: webrtc.RTPTransceiverDirectionRecvonly}) + tr, err := t.pc.AddTransceiverFromKind(codecType, webrtc.RTPTransceiverInit{Direction: webrtc.RTPTransceiverDirectionSendonly}) if err != nil { - return err + return senders, err } - tr.Stop() mid := lksdp.GetMidValue(m) if mid == "" { - return errors.New("mid value not found") + return senders, ErrMidNotFound } tr.SetMid(mid) + + // save mid -> senders for migration resue + sender := tr.Sender() + senders[mid] = sender + + // set transceiver to inactive + tr.SetSender(tr.Sender(), nil) } - return nil + return senders, nil } -func (t *PCTransport) SetPreviousAnswer(answer *webrtc.SessionDescription) { +func (t *PCTransport) SetPreviousSdp(offer, answer *webrtc.SessionDescription) { t.lock.Lock() if t.pc.RemoteDescription() == nil && t.previousAnswer == nil { t.previousAnswer = answer - if err := t.initPCWithPreviousAnswer(*t.previousAnswer); err != nil { + if senders, err := t.initPCWithPreviousAnswer(*t.previousAnswer); err != nil { t.params.Logger.Errorw("initPCWithPreviousAnswer failed", err) t.lock.Unlock() @@ -1196,11 +1222,45 @@ func (t *PCTransport) SetPreviousAnswer(answer *webrtc.SessionDescription) { onNegotiationFailed() } return + } else if offer != nil { + // in migration case, can't reuse tranceiver before negotiated except track subscribed at previous node + t.canReuseTransceiver = false + if err := t.parseTrackMid(*offer, senders); err != nil { + t.params.Logger.Errorw("parse previous offer failed", err, "offer", offer.SDP) + } } } t.lock.Unlock() } +func (t *PCTransport) parseTrackMid(offer webrtc.SessionDescription, senders map[string]*webrtc.RTPSender) error { + parsed, err := offer.Unmarshal() + if err != nil { + return err + } + + t.previousTrackDescription = make(map[string]*trackDescription) + for _, m := range parsed.MediaDescriptions { + msid, ok := m.Attribute(sdp.AttrKeyMsid) + if !ok { + continue + } + + if split := strings.Split(msid, " "); len(split) == 2 { + trackid := split[1] + mid := lksdp.GetMidValue(m) + if mid == "" { + return ErrMidNotFound + } + t.previousTrackDescription[trackid] = &trackDescription{ + mid: mid, + sender: senders[mid], + } + } + } + return nil +} + func (t *PCTransport) postEvent(event event) { t.eventChMu.RLock() if t.isClosed.Load() { @@ -1539,8 +1599,6 @@ func (t *PCTransport) createAndSendOffer(options *webrtc.OfferOptions) error { // indicate waiting for remote t.setNegotiationState(NegotiationStateRemote) - t.negotiationPending = make(map[livekit.ParticipantID]bool) - t.setupSignalStateCheckTimer() if onOffer := t.getOnOffer(); onOffer != nil { @@ -1603,6 +1661,13 @@ func (t *PCTransport) setRemoteDescription(sd webrtc.SessionDescription) error { } prometheus.ServiceOperationCounter.WithLabelValues(sdpType, "error", "remote_description").Add(1) return errors.Wrap(err, "setting remote description failed") + } else if sd.Type == webrtc.SDPTypeAnswer { + t.lock.Lock() + if !t.canReuseTransceiver { + t.canReuseTransceiver = true + t.previousTrackDescription = make(map[string]*trackDescription) + } + t.lock.Unlock() } for _, c := range t.pendingRemoteCandidates { diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index 75460bc67..7fe1fb08b 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -408,14 +408,6 @@ func (t *TransportManager) NegotiateSubscriber(force bool) { t.subscriber.Negotiate(force) } -func (t *TransportManager) AddNegotiationPending(publisherID livekit.ParticipantID) { - t.subscriber.AddNegotiationPending(publisherID) -} - -func (t *TransportManager) IsNegotiationPending(publisherID livekit.ParticipantID) bool { - return t.subscriber.IsNegotiationPending(publisherID) -} - func (t *TransportManager) ICERestart(iceConfig *types.IceConfig) { if iceConfig != nil { t.SetICEConfig(*iceConfig) @@ -496,7 +488,7 @@ func (t *TransportManager) handleConnectionFailed(isShortLived bool) { t.SetICEConfig(nextConfig) } -func (t *TransportManager) SetMigrateInfo(previousAnswer *webrtc.SessionDescription, dataChannels []*livekit.DataChannelInfo) { +func (t *TransportManager) SetMigrateInfo(previousOffer, previousAnswer *webrtc.SessionDescription, dataChannels []*livekit.DataChannelInfo) { t.lock.Lock() t.pendingDataChannelsPublisher = make([]*livekit.DataChannelInfo, 0, len(dataChannels)) pendingDataChannelsSubscriber := make([]*livekit.DataChannelInfo, 0, len(dataChannels)) @@ -515,7 +507,7 @@ func (t *TransportManager) SetMigrateInfo(previousAnswer *webrtc.SessionDescript } } - t.subscriber.SetPreviousAnswer(previousAnswer) + t.subscriber.SetPreviousSdp(previousOffer, previousAnswer) } func (t *TransportManager) ProcessPendingPublisherDataChannels() { diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 04d22c3b6..ffa656b8b 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -261,8 +261,6 @@ type LocalParticipant interface { HandleAnswer(sdp webrtc.SessionDescription) Negotiate(force bool) - AddNegotiationPending(publisherID livekit.ParticipantID) - IsNegotiationPending(publisherID livekit.ParticipantID) bool ICERestart(iceConfig *IceConfig) AddTrackToSubscriber(trackLocal webrtc.TrackLocal) (*webrtc.RTPSender, *webrtc.RTPTransceiver, error) AddTransceiverFromTrackToSubscriber(trackLocal webrtc.TrackLocal) (*webrtc.RTPSender, *webrtc.RTPTransceiver, error) @@ -309,7 +307,7 @@ type LocalParticipant interface { MaybeStartMigration(force bool, onStart func()) bool SetMigrateState(s MigrateState) MigrateState() MigrateState - SetMigrateInfo(previousAnswer *webrtc.SessionDescription, mediaTracks []*livekit.TrackPublishedResponse, dataChannels []*livekit.DataChannelInfo) + SetMigrateInfo(previousOffer, previousAnswer *webrtc.SessionDescription, mediaTracks []*livekit.TrackPublishedResponse, dataChannels []*livekit.DataChannelInfo) UpdateRTT(rtt uint32) diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 85b427835..312e77b7a 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -597,12 +597,13 @@ type FakeLocalParticipant struct { setMetadataArgsForCall []struct { arg1 string } - SetMigrateInfoStub func(*webrtc.SessionDescription, []*livekit.TrackPublishedResponse, []*livekit.DataChannelInfo) + SetMigrateInfoStub func(*webrtc.SessionDescription, *webrtc.SessionDescription, []*livekit.TrackPublishedResponse, []*livekit.DataChannelInfo) setMigrateInfoMutex sync.RWMutex setMigrateInfoArgsForCall []struct { arg1 *webrtc.SessionDescription - arg2 []*livekit.TrackPublishedResponse - arg3 []*livekit.DataChannelInfo + arg2 *webrtc.SessionDescription + arg3 []*livekit.TrackPublishedResponse + arg4 []*livekit.DataChannelInfo } SetMigrateStateStub func(types.MigrateState) setMigrateStateMutex sync.RWMutex @@ -3949,28 +3950,29 @@ func (fake *FakeLocalParticipant) SetMetadataArgsForCall(i int) string { return argsForCall.arg1 } -func (fake *FakeLocalParticipant) SetMigrateInfo(arg1 *webrtc.SessionDescription, arg2 []*livekit.TrackPublishedResponse, arg3 []*livekit.DataChannelInfo) { - var arg2Copy []*livekit.TrackPublishedResponse - if arg2 != nil { - arg2Copy = make([]*livekit.TrackPublishedResponse, len(arg2)) - copy(arg2Copy, arg2) - } - var arg3Copy []*livekit.DataChannelInfo +func (fake *FakeLocalParticipant) SetMigrateInfo(arg1 *webrtc.SessionDescription, arg2 *webrtc.SessionDescription, arg3 []*livekit.TrackPublishedResponse, arg4 []*livekit.DataChannelInfo) { + var arg3Copy []*livekit.TrackPublishedResponse if arg3 != nil { - arg3Copy = make([]*livekit.DataChannelInfo, len(arg3)) + arg3Copy = make([]*livekit.TrackPublishedResponse, len(arg3)) copy(arg3Copy, arg3) } + var arg4Copy []*livekit.DataChannelInfo + if arg4 != nil { + arg4Copy = make([]*livekit.DataChannelInfo, len(arg4)) + copy(arg4Copy, arg4) + } fake.setMigrateInfoMutex.Lock() fake.setMigrateInfoArgsForCall = append(fake.setMigrateInfoArgsForCall, struct { arg1 *webrtc.SessionDescription - arg2 []*livekit.TrackPublishedResponse - arg3 []*livekit.DataChannelInfo - }{arg1, arg2Copy, arg3Copy}) + arg2 *webrtc.SessionDescription + arg3 []*livekit.TrackPublishedResponse + arg4 []*livekit.DataChannelInfo + }{arg1, arg2, arg3Copy, arg4Copy}) stub := fake.SetMigrateInfoStub - fake.recordInvocation("SetMigrateInfo", []interface{}{arg1, arg2Copy, arg3Copy}) + fake.recordInvocation("SetMigrateInfo", []interface{}{arg1, arg2, arg3Copy, arg4Copy}) fake.setMigrateInfoMutex.Unlock() if stub != nil { - fake.SetMigrateInfoStub(arg1, arg2, arg3) + fake.SetMigrateInfoStub(arg1, arg2, arg3, arg4) } } @@ -3980,17 +3982,17 @@ func (fake *FakeLocalParticipant) SetMigrateInfoCallCount() int { return len(fake.setMigrateInfoArgsForCall) } -func (fake *FakeLocalParticipant) SetMigrateInfoCalls(stub func(*webrtc.SessionDescription, []*livekit.TrackPublishedResponse, []*livekit.DataChannelInfo)) { +func (fake *FakeLocalParticipant) SetMigrateInfoCalls(stub func(*webrtc.SessionDescription, *webrtc.SessionDescription, []*livekit.TrackPublishedResponse, []*livekit.DataChannelInfo)) { fake.setMigrateInfoMutex.Lock() defer fake.setMigrateInfoMutex.Unlock() fake.SetMigrateInfoStub = stub } -func (fake *FakeLocalParticipant) SetMigrateInfoArgsForCall(i int) (*webrtc.SessionDescription, []*livekit.TrackPublishedResponse, []*livekit.DataChannelInfo) { +func (fake *FakeLocalParticipant) SetMigrateInfoArgsForCall(i int) (*webrtc.SessionDescription, *webrtc.SessionDescription, []*livekit.TrackPublishedResponse, []*livekit.DataChannelInfo) { fake.setMigrateInfoMutex.RLock() defer fake.setMigrateInfoMutex.RUnlock() argsForCall := fake.setMigrateInfoArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4 } func (fake *FakeLocalParticipant) SetMigrateState(arg1 types.MigrateState) { diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index db04bb3f6..dcb823017 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -258,11 +258,6 @@ func (w *WebRTCReceiver) SetRTT(rtt uint32) { } } -func (w *WebRTCReceiver) SetTrackMeta(trackID livekit.TrackID, streamID string) { - w.streamID = streamID - w.trackID = trackID -} - func (w *WebRTCReceiver) StreamID() string { return w.streamID }