dandri/livekit
1
0 Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-10-23 10:52:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

skip out of order participant state updates (#3583)

Browse Source 
* skip out of order participant state updates

* avoid nil last active pointer

* cleanup

* protect state change callbacks

* gen

* test
This commit is contained in:
Paul Wells authored and GitHub committed 2025-04-06 06:44:50 -07:00
1 parent 15a8d9a266
commit ee08aede5b
5 files changed
+38 -25
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

pkg/rtc/participant.go
+22 -14
View File
@@ -39,6 +39,7 @@ import (
"github.com/livekit/protocol/logger"
"github.com/livekit/protocol/utils"
"github.com/livekit/protocol/utils/guid"
"github.com/livekit/protocol/utils/pointer"
"github.com/livekit/livekit-server/pkg/config"
"github.com/livekit/livekit-server/pkg/metric"
@@ -239,7 +240,7 @@ type ParticipantImpl struct {
onTrackPublished func(types.LocalParticipant, types.MediaTrack)
onTrackUpdated func(types.LocalParticipant, types.MediaTrack)
onTrackUnpublished func(types.LocalParticipant, types.MediaTrack)
onStateChange func(p types.LocalParticipant, state livekit.ParticipantInfo_State)
onStateChange func(p types.LocalParticipant)
onMigrateStateChange func(p types.LocalParticipant, migrateState types.MigrateState)
onParticipantUpdate func(types.LocalParticipant)
onDataPacket func(types.LocalParticipant, livekit.DataPacket_Kind, *livekit.DataPacket)
@@ -739,13 +740,13 @@ func (p *ParticipantImpl) getOnTrackUnpublished() func(types.LocalParticipant, t
return p.onTrackUnpublished
}
func (p *ParticipantImpl) OnStateChange(callback func(p types.LocalParticipant, state livekit.ParticipantInfo_State)) {
func (p *ParticipantImpl) OnStateChange(callback func(p types.LocalParticipant)) {
p.lock.Lock()
p.onStateChange = callback
p.lock.Unlock()
}
func (p *ParticipantImpl) getOnStateChange() func(p types.LocalParticipant, state livekit.ParticipantInfo_State) {
func (p *ParticipantImpl) getOnStateChange() func(p types.LocalParticipant) {
p.lock.RLock()
defer p.lock.RUnlock()
return p.onStateChange
@@ -1738,23 +1739,30 @@ func (p *ParticipantImpl) setupMetrics() {
}
func (p *ParticipantImpl) updateState(state livekit.ParticipantInfo_State) {
if state == livekit.ParticipantInfo_ACTIVE {
t := time.Now()
p.lastActiveAt.CompareAndSwap(nil, &t)
}
oldState := p.state.Swap(state).(livekit.ParticipantInfo_State)
if oldState == state {
return
var oldState livekit.ParticipantInfo_State
for {
oldState = p.state.Load().(livekit.ParticipantInfo_State)
if state <= oldState {
p.params.Logger.Debugw("ignoring out of order participant state", "state", state.String())
return
}
if state == livekit.ParticipantInfo_ACTIVE {
p.lastActiveAt.CompareAndSwap(nil, pointer.To(time.Now()))
}
if p.state.CompareAndSwap(oldState, state) {
break
}
}
if state == livekit.ParticipantInfo_DISCONNECTED && oldState == livekit.ParticipantInfo_ACTIVE {
prometheus.RecordSessionDuration(int(p.ProtocolVersion()), time.Since(*p.lastActiveAt.Load()))
}
p.params.Logger.Debugw("updating participant state", "state", state.String())
p.dirty.Store(true)
if onStateChange := p.getOnStateChange(); onStateChange != nil {
go onStateChange(p, state)
go onStateChange(p)
}
if state == livekit.ParticipantInfo_DISCONNECTED && oldState == livekit.ParticipantInfo_ACTIVE {
prometheus.RecordSessionDuration(int(p.ProtocolVersion()), time.Since(*p.lastActiveAt.Load()))
}
}
pkg/rtc/room.go
+5 -2
View File
@@ -444,13 +444,16 @@ func (r *Room) Join(participant types.LocalParticipant, requestSource routing.Me
r.joinedAt.Store(time.Now().Unix())
}
participant.OnStateChange(func(p types.LocalParticipant, state livekit.ParticipantInfo_State) {
var onStateChangeMu sync.Mutex
participant.OnStateChange(func(p types.LocalParticipant) {
if r.onParticipantChanged != nil {
r.onParticipantChanged(p)
}
r.broadcastParticipantState(p, broadcastOptions{skipSource: true})
if state == livekit.ParticipantInfo_ACTIVE {
onStateChangeMu.Lock()
defer onStateChangeMu.Unlock()
if state := p.State(); state == livekit.ParticipantInfo_ACTIVE {
// subscribe participant to existing published tracks
r.subscribeToExistingTracks(p)
pkg/rtc/room_test.go
+4 -2
View File
@@ -111,7 +111,8 @@ func TestRoomJoin(t *testing.T) {
stateChangeCB := p.OnStateChangeArgsForCall(0)
require.NotNil(t, stateChangeCB)
stateChangeCB(p, livekit.ParticipantInfo_ACTIVE)
p.StateReturns(livekit.ParticipantInfo_ACTIVE)
stateChangeCB(p)
// it should become a subscriber when connectivity changes
numTracks := 0
@@ -747,7 +748,8 @@ func TestHiddenParticipants(t *testing.T) {
stateChangeCB := hidden.OnStateChangeArgsForCall(0)
require.NotNil(t, stateChangeCB)
stateChangeCB(hidden, livekit.ParticipantInfo_ACTIVE)
hidden.StateReturns(livekit.ParticipantInfo_ACTIVE)
stateChangeCB(hidden)
require.Eventually(t, func() bool { return hidden.SubscribeToTrackCallCount() == 2 }, 5*time.Second, 10*time.Millisecond)
})
pkg/rtc/types/interfaces.go
+1 -1
View File
@@ -414,7 +414,7 @@ type LocalParticipant interface {
IssueFullReconnect(reason ParticipantCloseReason)
// callbacks
OnStateChange(func(p LocalParticipant, state livekit.ParticipantInfo_State))
OnStateChange(func(p LocalParticipant))
OnMigrateStateChange(func(p LocalParticipant, migrateState MigrateState))
// OnTrackPublished - remote added a track
OnTrackPublished(func(LocalParticipant, MediaTrack))
pkg/rtc/types/typesfakes/fake_local_participant.go
+6 -6
View File
@@ -726,10 +726,10 @@ type FakeLocalParticipant struct {
onParticipantUpdateArgsForCall []struct {
arg1 func(types.LocalParticipant)
}
OnStateChangeStub func(func(p types.LocalParticipant, state livekit.ParticipantInfo_State))
OnStateChangeStub func(func(p types.LocalParticipant))
onStateChangeMutex sync.RWMutex
onStateChangeArgsForCall []struct {
arg1 func(p types.LocalParticipant, state livekit.ParticipantInfo_State)
arg1 func(p types.LocalParticipant)
}
OnSubscribeStatusChangedStub func(func(publisherID livekit.ParticipantID, subscribed bool))
onSubscribeStatusChangedMutex sync.RWMutex
@@ -4975,10 +4975,10 @@ func (fake *FakeLocalParticipant) OnParticipantUpdateArgsForCall(i int) func(typ
return argsForCall.arg1
}
func (fake *FakeLocalParticipant) OnStateChange(arg1 func(p types.LocalParticipant, state livekit.ParticipantInfo_State)) {
func (fake *FakeLocalParticipant) OnStateChange(arg1 func(p types.LocalParticipant)) {
fake.onStateChangeMutex.Lock()
fake.onStateChangeArgsForCall = append(fake.onStateChangeArgsForCall, struct {
arg1 func(p types.LocalParticipant, state livekit.ParticipantInfo_State)
arg1 func(p types.LocalParticipant)
}{arg1})
stub := fake.OnStateChangeStub
fake.recordInvocation("OnStateChange", []interface{}{arg1})
@@ -4994,13 +4994,13 @@ func (fake *FakeLocalParticipant) OnStateChangeCallCount() int {
return len(fake.onStateChangeArgsForCall)
}
func (fake *FakeLocalParticipant) OnStateChangeCalls(stub func(func(p types.LocalParticipant, state livekit.ParticipantInfo_State))) {
func (fake *FakeLocalParticipant) OnStateChangeCalls(stub func(func(p types.LocalParticipant))) {
fake.onStateChangeMutex.Lock()
defer fake.onStateChangeMutex.Unlock()
fake.OnStateChangeStub = stub
}
func (fake *FakeLocalParticipant) OnStateChangeArgsForCall(i int) func(p types.LocalParticipant, state livekit.ParticipantInfo_State) {
func (fake *FakeLocalParticipant) OnStateChangeArgsForCall(i int) func(p types.LocalParticipant) {
fake.onStateChangeMutex.RLock()
defer fake.onStateChangeMutex.RUnlock()
argsForCall := fake.onStateChangeArgsForCall[i]