dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-22 14:20:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

keep mid unchange after migration for subscribed track (#995)

Browse Source
This commit is contained in:
cnderrauber
2022-09-09 17:39:09 +08:00
committed by GitHub
parent 93da599059
commit f1915feb1a
9 changed files with 129 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -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
go.sum
+2 -2
View File
@@ -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=
pkg/rtc/mediatracksubscriptions.go
+1 -2
View File
@@ -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)
}
}
pkg/rtc/participant.go
+6 -2
View File
@@ -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() {
pkg/rtc/transport.go
+95 -30
View File
@@ -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 {
pkg/rtc/transportmanager.go
+2 -10
View File
@@ -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() {
pkg/rtc/types/interfaces.go
+1 -3
View File
@@ -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)
pkg/rtc/types/typesfakes/fake_local_participant.go
+21 -19
View File
@@ -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) {
pkg/sfu/receiver.go
-5
View File
@@ -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
}