Support simulating subscriber bandwidth. (livekit#1609)

* Support simualting subscriber bandwidth.

When non-zero, a full allocation is triggered.
Also, probes are stopped.

When set to zero, normal probing mechanism should catch up.

Adding `allowPause` override which can be a connection option.

* fix log

* allowPause in participant params

pkg/rtc/participant.go

@@ -88,6 +88,7 @@ type ParticipantParams struct {
 	VersionGenerator             utils.TimedVersionGenerator
 	TrackResolver                types.MediaTrackResolver
 	DisableDynacast              bool
+	SubscriberAllowPause         bool
 }
 
 type ParticipantImpl struct {
@@ -1035,6 +1036,8 @@ func (p *ParticipantImpl) setupTransportManager() error {
 	tm.OnAnyTransportNegotiationFailed(p.onAnyTransportNegotiationFailed)
 
 	tm.OnDataMessage(p.onDataMessage)
+
+	tm.SetSubscriberAllowPause(p.params.SubscriberAllowPause)
 	p.TransportManager = tm
 	return nil
 }

pkg/rtc/room.go

@@ -692,7 +692,7 @@ func (r *Room) OnMetadataUpdate(f func(metadata string)) {
 func (r *Room) SimulateScenario(participant types.LocalParticipant, simulateScenario *livekit.SimulateScenario) error {
 	switch scenario := simulateScenario.Scenario.(type) {
 	case *livekit.SimulateScenario_SpeakerUpdate:
-		r.Logger.Infow("simulating speaker update", "participant", participant.Identity())
+		r.Logger.Infow("simulating speaker update", "participant", participant.Identity(), "duration", scenario.SpeakerUpdate)
 		go func() {
 			<-time.After(time.Duration(scenario.SpeakerUpdate) * time.Second)
 			r.sendSpeakerChanges([]*livekit.SpeakerInfo{{
@@ -723,13 +723,19 @@ func (r *Room) SimulateScenario(participant types.LocalParticipant, simulateScen
 		if err := participant.Close(true, types.ParticipantCloseReasonSimulateServerLeave); err != nil {
 			return err
 		}
-
 	case *livekit.SimulateScenario_SwitchCandidateProtocol:
 		r.Logger.Infow("simulating switch candidate protocol", "participant", participant.Identity())
 		participant.ICERestart(&livekit.ICEConfig{
 			PreferenceSubscriber: livekit.ICECandidateType(scenario.SwitchCandidateProtocol),
 			PreferencePublisher:  livekit.ICECandidateType(scenario.SwitchCandidateProtocol),
 		})
+	case *livekit.SimulateScenario_SubscriberBandwidth:
+		if scenario.SubscriberBandwidth > 0 {
+			r.Logger.Infow("simulating subscriber bandwidth start", "participant", participant.Identity(), "bandwidth", scenario.SubscriberBandwidth)
+		} else {
+			r.Logger.Infow("simulating subscriber bandwidth end", "participant", participant.Identity())
+		}
+		participant.SetSubscriberChannelCapacity(scenario.SubscriberBandwidth)
 	}
 	return nil
 }

pkg/rtc/transport.go

@@ -1118,6 +1118,22 @@ func (t *PCTransport) RemoveTrackFromStreamAllocator(subTrack types.SubscribedTr
 	t.streamAllocator.RemoveTrack(subTrack.DownTrack())
 }
 
+func (t *PCTransport) SetAllowPauseOfStreamAllocator(allowPause bool) {
+	if t.streamAllocator == nil {
+		return
+	}
+
+	t.streamAllocator.SetAllowPause(allowPause)
+}
+
+func (t *PCTransport) SetChannelCapacityOfStreamAllocator(channelCapacity int64) {
+	if t.streamAllocator == nil {
+		return
+	}
+
+	t.streamAllocator.SetChannelCapacity(channelCapacity)
+}
+
 func (t *PCTransport) GetICEConnectionType() types.ICEConnectionType {
 	unknown := types.ICEConnectionTypeUnknown
 	if t.pc == nil {

pkg/rtc/transportmanager.go

@@ -782,3 +782,11 @@ func (t *TransportManager) SetSignalSourceValid(valid bool) {
 	t.signalSourceValid.Store(valid)
 	t.params.Logger.Debugw("signal source valid", "valid", valid)
 }
+
+func (t *TransportManager) SetSubscriberAllowPause(allowPause bool) {
+	t.subscriber.SetAllowPauseOfStreamAllocator(allowPause)
+}
+
+func (t *TransportManager) SetSubscriberChannelCapacity(channelCapacity int64) {
+	t.subscriber.SetChannelCapacityOfStreamAllocator(channelCapacity)
+}

pkg/rtc/types/interfaces.go

@@ -333,6 +333,10 @@ type LocalParticipant interface {
 
 	UpdateSubscribedQuality(nodeID livekit.NodeID, trackID livekit.TrackID, maxQualities []SubscribedCodecQuality) error
 	UpdateMediaLoss(nodeID livekit.NodeID, trackID livekit.TrackID, fractionalLoss uint32) error
+
+	// down stream bandwidth management
+	SetSubscriberAllowPause(allowPause bool)
+	SetSubscriberChannelCapacity(channelCapacity int64)
 }
 
 // Room is a container of participants, and can provide room-level actions

pkg/service/roommanager.go

@@ -336,6 +336,7 @@ func (r *RoomManager) StartSession(
 		ReconnectOnSubscriptionError: reconnectOnSubscriptionError,
 		VersionGenerator:             r.versionGenerator,
 		TrackResolver:                room.ResolveMediaTrackForSubscriber,
+		SubscriberAllowPause:         r.config.RTC.CongestionControl.AllowPause,
 	})
 	if err != nil {
 		return err