Send Room updates when participant counts change (#1647)

Reduces the number of unneeded generation with ProtoProxy

go.mod

@@ -18,7 +18,7 @@ require (
 	github.com/jxskiss/base62 v1.1.0
 	github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1
 	github.com/livekit/mediatransportutil v0.0.0-20230326055817-ed569ca13d26
-	github.com/livekit/protocol v1.5.5-0.20230422131440-eb3ef0f4bf36
+	github.com/livekit/protocol v1.5.5
 	github.com/livekit/psrpc v0.3.0
 	github.com/mackerelio/go-osstat v0.2.4
 	github.com/magefile/mage v1.14.0

go.sum

@@ -121,8 +121,8 @@ github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkD
 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ=
 github.com/livekit/mediatransportutil v0.0.0-20230326055817-ed569ca13d26 h1:QlQFyMwCDgjyySsrgmrMcVbEBA6KZcyTzvK+z346tUA=
 github.com/livekit/mediatransportutil v0.0.0-20230326055817-ed569ca13d26/go.mod h1:eDA41kiySZoG+wy4Etsjb3w0jjLx69i/vAmSjG4bteA=
-github.com/livekit/protocol v1.5.5-0.20230422131440-eb3ef0f4bf36 h1:Qrl0N7dAeR2iLOk6u4WNelFhTh2OetyZzd/h8kbxLSI=
-github.com/livekit/protocol v1.5.5-0.20230422131440-eb3ef0f4bf36/go.mod h1:iZ289+6H5xn/9kP2iqpRvVWxuc8GXBMqN0qI7LdN9HI=
+github.com/livekit/protocol v1.5.5 h1:vuSU3TI/w58WnAWnyC59nMzY/JE+ZznU6W/iRgWw4JQ=
+github.com/livekit/protocol v1.5.5/go.mod h1:iZ289+6H5xn/9kP2iqpRvVWxuc8GXBMqN0qI7LdN9HI=
 github.com/livekit/psrpc v0.3.0 h1:giBZsfM3CWA0oIYXofsMITbVQtyW7u/ES9sQmVspHPM=
 github.com/livekit/psrpc v0.3.0/go.mod h1:n6JntEg+zT6Ji8InoyTpV7wusPNwGqqtxmHlkNhDN0U=
 github.com/mackerelio/go-osstat v0.2.4 h1:qxGbdPkFo65PXOb/F/nhDKpF2nGmGaCFDLXoZjJTtUs=

pkg/rtc/room.go

@@ -37,6 +37,7 @@ const (
 var (
 	// var to allow unit test override
 	RoomDepartureGrace uint32 = 20
+	roomUpdateInterval        = 5 * time.Second // frequency to update room participant counts
 )
 
 type broadcastOptions struct {
@@ -47,9 +48,10 @@ type broadcastOptions struct {
 type Room struct {
 	lock sync.RWMutex
 
-	protoRoom *livekit.Room
-	internal  *livekit.RoomInternal
-	Logger    logger.Logger
+	protoRoom  *livekit.Room
+	internal   *livekit.RoomInternal
+	protoProxy *utils.ProtoProxy[*livekit.Room]
+	Logger     logger.Logger
 
 	config         WebRTCConfig
 	audioConfig    *config.AudioConfig
@@ -76,7 +78,7 @@ type Room struct {
 	closed chan struct{}
 
 	onParticipantChanged func(p types.LocalParticipant)
-	onMetadataUpdate     func(metadata string)
+	onRoomUpdated        func()
 	onClose              func()
 }
 
@@ -110,6 +112,7 @@ func NewRoom(
 		batchedUpdates:            make(map[livekit.ParticipantIdentity]*livekit.ParticipantInfo),
 		closed:                    make(chan struct{}),
 	}
+	r.protoProxy = utils.NewProtoProxy[*livekit.Room](roomUpdateInterval, r.updateProto)
 	if r.protoRoom.EmptyTimeout == 0 {
 		r.protoRoom.EmptyTimeout = DefaultEmptyTimeout
 	}
@@ -119,16 +122,13 @@ func NewRoom(
 
 	go r.audioUpdateWorker()
 	go r.connectionQualityWorker()
-	go r.subscriberBroadcastWorker()
+	go r.changeUpdateWorker()
 
 	return r
 }
 
 func (r *Room) ToProto() *livekit.Room {
-	r.lock.RLock()
-	defer r.lock.RUnlock()
-
-	return proto.Clone(r.protoRoom).(*livekit.Room)
+	return r.protoProxy.Get()
 }
 
 func (r *Room) Name() livekit.RoomName {
@@ -244,14 +244,13 @@ func (r *Room) Join(participant types.LocalParticipant, requestSource routing.Me
 	}
 
 	if r.protoRoom.MaxParticipants > 0 && !participant.IsRecorder() {
-		participantCount := 0
+		numParticipants := uint32(0)
 		for _, p := range r.participants {
 			if !p.IsRecorder() {
-				participantCount++
+				numParticipants++
 			}
 		}
-
-		if participantCount >= int(r.protoRoom.MaxParticipants) {
+		if numParticipants >= r.protoRoom.MaxParticipants {
 			return ErrMaxParticipantsExceeded
 		}
 	}
@@ -259,9 +258,6 @@ func (r *Room) Join(participant types.LocalParticipant, requestSource routing.Me
 	if r.FirstJoinedAt() == 0 {
 		r.joinedAt.Store(time.Now().Unix())
 	}
-	if !participant.Hidden() {
-		r.protoRoom.NumParticipants++
-	}
 
 	// it's important to set this before connection, we don't want to miss out on any published tracks
 	participant.OnTrackPublished(r.onTrackPublished)
@@ -340,7 +336,9 @@ func (r *Room) Join(participant types.LocalParticipant, requestSource routing.Me
 
 	if participant.IsRecorder() && !r.protoRoom.ActiveRecording {
 		r.protoRoom.ActiveRecording = true
-		r.sendRoomUpdateLocked()
+		r.protoProxy.MarkDirty(true)
+	} else {
+		r.protoProxy.MarkDirty(false)
 	}
 
 	r.participants[participant.Identity()] = participant
@@ -419,10 +417,7 @@ func (r *Room) ResumeParticipant(p types.LocalParticipant, requestSource routing
 		return err
 	}
 
-	r.lock.RLock()
-	p.SendRoomUpdate(r.protoRoom)
-	r.lock.RUnlock()
-
+	p.SendRoomUpdate(r.ToProto())
 	p.ICERestart(nil)
 	return nil
 }
@@ -445,6 +440,7 @@ func (r *Room) RemoveParticipant(identity livekit.ParticipantIdentity, pID livek
 		}
 	}
 
+	immediateChange := false
 	if (p != nil && p.IsRecorder()) || r.protoRoom.ActiveRecording {
 		activeRecording := false
 		for _, op := range r.participants {
@@ -456,10 +452,12 @@ func (r *Room) RemoveParticipant(identity livekit.ParticipantIdentity, pID livek
 
 		if r.protoRoom.ActiveRecording != activeRecording {
 			r.protoRoom.ActiveRecording = activeRecording
-			r.sendRoomUpdateLocked()
+			immediateChange = true
+
 		}
 	}
 	r.lock.Unlock()
+	r.protoProxy.MarkDirty(immediateChange)
 
 	if !ok {
 		return
@@ -634,6 +632,7 @@ func (r *Room) Close() {
 	for _, p := range r.GetParticipants() {
 		_ = p.Close(true, types.ParticipantCloseReasonRoomClose)
 	}
+	r.protoProxy.Stop()
 	if r.onClose != nil {
 		r.onClose()
 	}
@@ -661,32 +660,26 @@ func (r *Room) SetMetadata(metadata string) {
 	r.lock.Lock()
 	r.protoRoom.Metadata = metadata
 	r.lock.Unlock()
-
-	r.lock.RLock()
-	r.sendRoomUpdateLocked()
-	r.lock.RUnlock()
-
-	if r.onMetadataUpdate != nil {
-		r.onMetadataUpdate(metadata)
-	}
+	r.protoProxy.MarkDirty(true)
 }
 
-func (r *Room) sendRoomUpdateLocked() {
+func (r *Room) sendRoomUpdate() {
+	roomInfo := r.ToProto()
 	// Send update to participants
-	for _, p := range r.participants {
+	for _, p := range r.GetParticipants() {
 		if !p.IsReady() {
 			continue
 		}
 
-		err := p.SendRoomUpdate(r.protoRoom)
+		err := p.SendRoomUpdate(roomInfo)
 		if err != nil {
 			r.Logger.Warnw("failed to send room update", err, "participant", p.Identity())
 		}
 	}
 }
 
-func (r *Room) OnMetadataUpdate(f func(metadata string)) {
-	r.onMetadataUpdate = f
+func (r *Room) OnRoomUpdated(f func()) {
+	r.onRoomUpdated = f
 }
 
 func (r *Room) SimulateScenario(participant types.LocalParticipant, simulateScenario *livekit.SimulateScenario) error {
@@ -760,11 +753,9 @@ func (r *Room) createJoinResponseLocked(participant types.LocalParticipant, iceS
 	}
 
 	return &livekit.JoinResponse{
-		Room:              r.protoRoom,
+		Room:              r.ToProto(),
 		Participant:       participant.ToProto(),
 		OtherParticipants: otherParticipants,
-		ServerVersion:     r.serverInfo.Version,
-		ServerRegion:      r.serverInfo.Region,
 		IceServers:        iceServers,
 		// indicates both server and client support subscriber as primary
 		SubscriberPrimary:   participant.SubscriberAsPrimary(),
@@ -1003,15 +994,39 @@ func (r *Room) pushAndDequeueUpdates(pi *livekit.ParticipantInfo, isImmediate bo
 	return updates
 }
 
-func (r *Room) subscriberBroadcastWorker() {
-	ticker := time.NewTicker(subscriberUpdateInterval)
-	defer ticker.Stop()
+func (r *Room) updateProto() *livekit.Room {
+	r.lock.RLock()
+	room := proto.Clone(r.protoRoom).(*livekit.Room)
+	r.lock.RUnlock()
+
+	room.NumPublishers = 0
+	room.NumParticipants = 0
+	for _, p := range r.GetParticipants() {
+		if !p.IsRecorder() {
+			room.NumParticipants++
+		}
+		if p.IsPublisher() {
+			room.NumPublishers++
+		}
+	}
+
+	return room
+}
+
+func (r *Room) changeUpdateWorker() {
+	subTicker := time.NewTicker(subscriberUpdateInterval)
+	defer subTicker.Stop()
 
 	for !r.IsClosed() {
 		select {
 		case <-r.closed:
 			return
-		case <-ticker.C:
+		case <-r.protoProxy.Updated():
+			if r.onRoomUpdated != nil {
+				r.onRoomUpdated()
+			}
+			r.sendRoomUpdate()
+		case <-subTicker.C:
 			r.batchedUpdatesMu.Lock()
 			updatesMap := r.batchedUpdates
 			r.batchedUpdates = make(map[livekit.ParticipantIdentity]*livekit.ParticipantInfo)

pkg/rtc/room_test.go

@@ -35,6 +35,7 @@ func init() {
 	})
 	// allow immediate closure in testing
 	RoomDepartureGrace = 1
+	roomUpdateInterval = defaultDelay
 }
 
 var iceServersForRoom = []*livekit.ICEServer{{Urls: []string{"stun:stun.l.google.com:19302"}}}
@@ -645,7 +646,7 @@ func TestHiddenParticipants(t *testing.T) {
 		require.Len(t, res.OtherParticipants, 2)
 		require.Len(t, rm.GetParticipants(), 4)
 		require.NotEmpty(t, res.IceServers)
-		require.Equal(t, "testregion", res.ServerRegion)
+		require.Equal(t, "testregion", res.ServerInfo.Region)
 	})
 
 	t.Run("hidden participant subscribes to tracks", func(t *testing.T) {
@@ -665,15 +666,35 @@ func TestHiddenParticipants(t *testing.T) {
 }
 
 func TestRoomUpdate(t *testing.T) {
+	t.Run("updates are sent when participant joined", func(t *testing.T) {
+		rm := newRoomWithParticipants(t, testRoomOpts{num: 1})
+		defer rm.Close()
+
+		p1 := rm.GetParticipants()[0].(*typesfakes.FakeLocalParticipant)
+		require.Equal(t, 0, p1.SendRoomUpdateCallCount())
+
+		p2 := newMockParticipant("p2", types.CurrentProtocol, false, false)
+		require.NoError(t, rm.Join(p2, nil, nil, iceServersForRoom))
+
+		// p1 should have received an update
+		time.Sleep(2 * defaultDelay)
+		require.Equal(t, 1, p1.SendRoomUpdateCallCount())
+		require.EqualValues(t, 2, p1.SendRoomUpdateArgsForCall(0).NumParticipants)
+	})
+
 	t.Run("participants should receive metadata update", func(t *testing.T) {
 		rm := newRoomWithParticipants(t, testRoomOpts{num: 2})
 		defer rm.Close()
 
 		rm.SetMetadata("test metadata...")
 
+		// callbacks are updated from goroutine
+		time.Sleep(2 * defaultDelay)
+
 		for _, op := range rm.GetParticipants() {
 			fp := op.(*typesfakes.FakeLocalParticipant)
-			require.Equal(t, 1, fp.SendRoomUpdateCallCount())
+			// room updates are now sent for both participant joining and room metadata
+			require.GreaterOrEqual(t, fp.SendRoomUpdateCallCount(), 1)
 		}
 	})
 }

pkg/rtc/types/interfaces.go

@@ -181,6 +181,7 @@ type Participant interface {
 	SetName(name string)
 	SetMetadata(metadata string)
 
+	IsPublisher() bool
 	GetPublishedTrack(sid livekit.TrackID) MediaTrack
 	GetPublishedTracks() []MediaTrack
 	RemovePublishedTrack(track MediaTrack, willBeResumed bool, shouldClose bool)
@@ -284,7 +285,6 @@ type LocalParticipant interface {
 	// returns list of participant identities that the current participant is subscribed to
 	GetSubscribedParticipants() []livekit.ParticipantID
 	IsSubscribedTo(sid livekit.ParticipantID) bool
-	IsPublisher() bool
 
 	GetAudioLevel() (smoothedLevel float64, active bool)
 	GetConnectionQuality() *livekit.ConnectionQualityInfo

pkg/rtc/types/typesfakes/fake_participant.go

@@ -95,6 +95,16 @@ type FakeParticipant struct {
 	identityReturnsOnCall map[int]struct {
 		result1 livekit.ParticipantIdentity
 	}
+	IsPublisherStub        func() bool
+	isPublisherMutex       sync.RWMutex
+	isPublisherArgsForCall []struct {
+	}
+	isPublisherReturns struct {
+		result1 bool
+	}
+	isPublisherReturnsOnCall map[int]struct {
+		result1 bool
+	}
 	IsRecorderStub        func() bool
 	isRecorderMutex       sync.RWMutex
 	isRecorderArgsForCall []struct {
@@ -637,6 +647,59 @@ func (fake *FakeParticipant) IdentityReturnsOnCall(i int, result1 livekit.Partic
 	}{result1}
 }
 
+func (fake *FakeParticipant) IsPublisher() bool {
+	fake.isPublisherMutex.Lock()
+	ret, specificReturn := fake.isPublisherReturnsOnCall[len(fake.isPublisherArgsForCall)]
+	fake.isPublisherArgsForCall = append(fake.isPublisherArgsForCall, struct {
+	}{})
+	stub := fake.IsPublisherStub
+	fakeReturns := fake.isPublisherReturns
+	fake.recordInvocation("IsPublisher", []interface{}{})
+	fake.isPublisherMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *FakeParticipant) IsPublisherCallCount() int {
+	fake.isPublisherMutex.RLock()
+	defer fake.isPublisherMutex.RUnlock()
+	return len(fake.isPublisherArgsForCall)
+}
+
+func (fake *FakeParticipant) IsPublisherCalls(stub func() bool) {
+	fake.isPublisherMutex.Lock()
+	defer fake.isPublisherMutex.Unlock()
+	fake.IsPublisherStub = stub
+}
+
+func (fake *FakeParticipant) IsPublisherReturns(result1 bool) {
+	fake.isPublisherMutex.Lock()
+	defer fake.isPublisherMutex.Unlock()
+	fake.IsPublisherStub = nil
+	fake.isPublisherReturns = struct {
+		result1 bool
+	}{result1}
+}
+
+func (fake *FakeParticipant) IsPublisherReturnsOnCall(i int, result1 bool) {
+	fake.isPublisherMutex.Lock()
+	defer fake.isPublisherMutex.Unlock()
+	fake.IsPublisherStub = nil
+	if fake.isPublisherReturnsOnCall == nil {
+		fake.isPublisherReturnsOnCall = make(map[int]struct {
+			result1 bool
+		})
+	}
+	fake.isPublisherReturnsOnCall[i] = struct {
+		result1 bool
+	}{result1}
+}
+
 func (fake *FakeParticipant) IsRecorder() bool {
 	fake.isRecorderMutex.Lock()
 	ret, specificReturn := fake.isRecorderReturnsOnCall[len(fake.isRecorderArgsForCall)]
@@ -1118,6 +1181,8 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} {
 	defer fake.iDMutex.RUnlock()
 	fake.identityMutex.RLock()
 	defer fake.identityMutex.RUnlock()
+	fake.isPublisherMutex.RLock()
+	defer fake.isPublisherMutex.RUnlock()
 	fake.isRecorderMutex.RLock()
 	defer fake.isRecorderMutex.RUnlock()
 	fake.removePublishedTrackMutex.RLock()

pkg/service/roommanager.go

@@ -451,7 +451,7 @@ func (r *RoomManager) getOrCreateRoom(ctx context.Context, roomName livekit.Room
 		newRoom.Logger.Infow("room closed")
 	})
 
-	newRoom.OnMetadataUpdate(func(metadata string) {
+	newRoom.OnRoomUpdated(func() {
 		if err := r.roomStore.StoreRoom(ctx, newRoom.ToProto(), newRoom.Internal()); err != nil {
 			newRoom.Logger.Errorw("could not handle metadata update", err)
 		}