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

Split ICE candidate queue. (#2885)

Browse Source 
Shared ICE candidate queue meant only one of PUBLISHER/SUBSCRIBER pc got
final candidate notification. Split the queue.
This commit is contained in:
Raja Subramanian
2024-07-20 10:25:06 +05:30
committed by GitHub
parent 73b3a91a5c
commit bd616d6074
4 changed files with 21 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/participant.go
+1 -1
View File
@@ -191,7 +191,7 @@ type ParticipantImpl struct {
*UpTrackManager
*SubscriptionManager
icQueue atomic.Pointer[webrtc.ICECandidate]
icQueue [2]atomic.Pointer[webrtc.ICECandidate]
// keeps track of unpublished tracks in order to reuse trackID
unpublishedTracks []*livekit.TrackInfo
pkg/rtc/participant_signal.go
+8 -1
View File
@@ -20,6 +20,7 @@ import (
"time"
"github.com/pion/webrtc/v3"
"go.uber.org/atomic"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
@@ -265,7 +266,13 @@ func (p *ParticipantImpl) sendDisconnectUpdatesForReconnect() error {
}
func (p *ParticipantImpl) sendICECandidate(ic *webrtc.ICECandidate, target livekit.SignalTarget) error {
prevIC := p.icQueue.Swap(ic)
var icQueue *atomic.Pointer[webrtc.ICECandidate]
if target == livekit.SignalTarget_PUBLISHER {
icQueue = &p.icQueue[0]
} else {
icQueue = &p.icQueue[1]
}
prevIC := icQueue.Swap(ic)
if prevIC == nil {
return nil
}
pkg/sfu/utils/wraparound.go
+4 -4
View File
@@ -157,7 +157,7 @@ func (w *WrapAround[T, ET]) GetExtendedHighest() ET {
}
func (w *WrapAround[T, ET]) updateExtendedHighest() {
w.extendedHighest = getExtendedHighest(w.cycles, w.highest)
w.extendedHighest = getExtended(w.cycles, w.highest)
}
func (w *WrapAround[T, ET]) maybeAdjustStart(val T) (result WrapAroundUpdateResult[ET]) {
@@ -172,7 +172,7 @@ func (w *WrapAround[T, ET]) maybeAdjustStart(val T) (result WrapAroundUpdateResu
cycles -= w.fullRange
}
result.PreExtendedHighest = w.extendedHighest
result.ExtendedVal = getExtendedHighest(cycles, val)
result.ExtendedVal = getExtended(cycles, val)
return
}
@@ -201,7 +201,7 @@ func (w *WrapAround[T, ET]) maybeAdjustStart(val T) (result WrapAroundUpdateResu
}
}
result.PreExtendedHighest = w.extendedHighest
result.ExtendedVal = getExtendedHighest(cycles, val)
result.ExtendedVal = getExtended(cycles, val)
return
}
@@ -211,6 +211,6 @@ func (w *WrapAround[T, ET]) isWrapBack(earlier T, later T) bool {
// ------------------------------------
func getExtendedHighest[T number, ET extendedNumber](cycles ET, val T) ET {
func getExtended[T number, ET extendedNumber](cycles ET, val T) ET {
return cycles + ET(val)
}
test/client/client.go
+8 -2
View File
@@ -69,7 +69,7 @@ type RTCClient struct {
signalRequestInterceptor SignalRequestInterceptor
signalResponseInterceptor SignalResponseInterceptor
icQueue atomic.Pointer[webrtc.ICECandidate]
icQueue [2]atomic.Pointer[webrtc.ICECandidate]
subscriberAsPrimary atomic.Bool
publisherFullyEstablished atomic.Bool
@@ -555,7 +555,13 @@ func (c *RTCClient) sendRequest(msg *livekit.SignalRequest) error {
}
func (c *RTCClient) SendIceCandidate(ic *webrtc.ICECandidate, target livekit.SignalTarget) error {
prevIC := c.icQueue.Swap(ic)
var icQueue *atomic.Pointer[webrtc.ICECandidate]
if target == livekit.SignalTarget_PUBLISHER {
icQueue = &c.icQueue[0]
} else {
icQueue = &c.icQueue[1]
}
prevIC := icQueue.Swap(ic)
if prevIC == nil {
return nil
}