Move probe observer to pacer (#3214)

* Probe ID pass

* WIP

* WIP

* WIP

* WIP

* WIP

* WIP

* WIP

* WIP

* WIP

* WIP

* WIP

* clean up

* typo

* populate desired bytes

* correct num probes calculation

* debug log

* remove unused constant

* log channel observer

* debug

* clear isInProbe flag on end

* clear probe flag on reset

* re-arrange

pkg/rtc/participant.go

@@ -1890,13 +1890,6 @@ func (p *ParticipantImpl) onDataMessage(kind livekit.DataPacket_Kind, data []byt
 		}
 		shouldForwardData = false
 		shouldForwardMetrics = true
-		// METRICS-TODO-QUESTIONS:
-		// 1. Should this record (and do processing/batching) metrics (i. e. publisher side) rather
-		//    than forwarding and recording/processing/batching at every subscriber (in this case
-		//    subscriber is defined as the other participants pushing this to edge client).
-		// 2. If the above is done, there could be two cadences, publisher side recording/processing/batching
-		//    and pushing it to all subscribers on some cadence and subscribers have their own cadence of
-		//    processing/batching and sending to edge clients.
 		p.metricTimestamper.Process(payload.Metrics)
 	case *livekit.DataPacket_RpcRequest:
 		if payload.RpcRequest == nil {

pkg/rtc/transport.go

@@ -452,13 +452,6 @@ func NewPCTransport(params TransportParams) (*PCTransport, error) {
 		lastNegotiate:            time.Now(),
 	}
 	if params.IsSendSide {
-		t.streamAllocator = streamallocator.NewStreamAllocator(streamallocator.StreamAllocatorParams{
-			Config: params.CongestionControlConfig.StreamAllocator,
-			Logger: params.Logger.WithComponent(utils.ComponentCongestionControl),
-		}, params.CongestionControlConfig.Enabled, params.CongestionControlConfig.AllowPause)
-		t.streamAllocator.OnStreamStateChange(params.Handler.OnStreamStateChange)
-		t.streamAllocator.Start()
-
 		if params.CongestionControlConfig.UseSendSideBWE || params.UseSendSideBWE {
 			params.Logger.Infow("using send side BWE")
 			t.bwe = sendsidebwe.NewSendSideBWE(sendsidebwe.SendSideBWEParams{
@@ -473,7 +466,15 @@ func NewPCTransport(params TransportParams) (*PCTransport, error) {
 			})
 			t.pacer = pacer.NewPassThrough(params.Logger, nil)
 		}
-		t.streamAllocator.SetBWE(t.bwe)
+
+		t.streamAllocator = streamallocator.NewStreamAllocator(streamallocator.StreamAllocatorParams{
+			Config: params.CongestionControlConfig.StreamAllocator,
+			BWE:    t.bwe,
+			Pacer:  t.pacer,
+			Logger: params.Logger.WithComponent(utils.ComponentCongestionControl),
+		}, params.CongestionControlConfig.Enabled, params.CongestionControlConfig.AllowPause)
+		t.streamAllocator.OnStreamStateChange(params.Handler.OnStreamStateChange)
+		t.streamAllocator.Start()
 	}
 
 	if err := t.createPeerConnection(); err != nil {

pkg/sfu/bwe/remotebwe/remote_bwe.go

@@ -24,10 +24,6 @@ import (
 	"github.com/livekit/protocol/utils/mono"
 )
 
-const (
-	ChannelCapacityInfinity = 100 * 1000 * 1000 // 100 Mbps
-)
-
 // ---------------------------------------------------------------------------
 
 type RemoteBWEConfig struct {
@@ -110,6 +106,7 @@ func (r *RemoteBWE) Reset() {
 	defer r.lock.Unlock()
 
 	r.channelObserver = r.newChannelObserverNonProbe()
+	r.isInProbe = false
 }
 
 func (r *RemoteBWE) Stop() {
@@ -205,7 +202,7 @@ func (r *RemoteBWE) estimateAvailableChannelCapacity(reason channelCongestionRea
 		"channel", r.channelObserver,
 	)
 	if estimateToCommit > commitThreshold {
-		ulgr.Debugw("remote bwe: channel congestion detected, skipping above commit threshold  channel capacity update")
+		ulgr.Debugw("remote bwe: channel congestion detected, skipping above commit threshold channel capacity update")
 		return false
 	}
 
@@ -298,7 +295,7 @@ func (r *RemoteBWE) ProbingEnd(isNotFailing bool, isGoalReached bool) {
 	// the send side is in full control of bandwidth estimation.
 	//
 	r.params.Logger.Debugw(
-		"probe done",
+		"remote bwe: probe done",
 		"isNotFailing", isNotFailing,
 		"isGoalReached", isGoalReached,
 		"committedEstimate", r.committedChannelCapacity,
@@ -306,6 +303,7 @@ func (r *RemoteBWE) ProbingEnd(isNotFailing bool, isGoalReached bool) {
 		"channel", r.channelObserver,
 	)
 	r.channelObserver = r.newChannelObserverNonProbe()
+	r.isInProbe = false
 	if !isNotFailing {
 		return
 	}

pkg/sfu/ccutils/prober.go

@@ -49,7 +49,7 @@
 //     are paused (could be due to downstream bandwidth
 //     constraints or the corresponding upstream tracks may
 //     have paused due to upstream bandwidth constraints).
-//     Another issue is not being to have tight control on
+//     Another issue is not being able to have tight control on
 //     probing window boundary as the packet forwarding path
 //     may not have a packet to forward. But, it should not
 //     be a major concern as long as some stream(s) is/are
@@ -126,14 +126,14 @@ import (
 
 	"github.com/gammazero/deque"
 	"go.uber.org/atomic"
+	"go.uber.org/zap/zapcore"
 
 	"github.com/livekit/protocol/logger"
 )
 
 type ProberListener interface {
 	OnSendProbe(bytesToSend int)
-	OnProbeClusterDone(info ProbeClusterInfo)
-	OnActiveChanged(isActive bool)
+	OnProbeClusterSwitch(probeClusterId ProbeClusterId, desiredBytes int)
 }
 
 type ProberParams struct {
@@ -146,11 +146,9 @@ type Prober struct {
 
 	clusterId atomic.Uint32
 
-	clustersMu                sync.RWMutex
-	clusters                  deque.Deque[*Cluster]
-	activeCluster             *Cluster
-	activeStateQueue          []bool
-	activeStateQueueInProcess atomic.Bool
+	clustersMu    sync.RWMutex
+	clusters      deque.Deque[*Cluster]
+	activeCluster *Cluster
 }
 
 func NewProber(params ProberParams) *Prober {
@@ -168,33 +166,27 @@ func (p *Prober) IsRunning() bool {
 	return p.clusters.Len() > 0
 }
 
-func (p *Prober) Reset() {
-	reset := false
-	var info ProbeClusterInfo
-
+func (p *Prober) Reset(info ProbeClusterInfo) {
 	p.clustersMu.Lock()
+	defer p.clustersMu.Unlock()
+
 	if p.activeCluster != nil {
-		p.params.Logger.Debugw("prober: resetting active cluster", "cluster", p.activeCluster.String())
-		reset = true
-		info = p.activeCluster.GetInfo()
+		if p.activeCluster.Id() == info.ProbeClusterId {
+			p.activeCluster.MarkCompleted(info)
+			p.params.Logger.Debugw("prober: resetting active cluster", "cluster", p.activeCluster)
+		}
 	}
 
 	p.clusters.Clear()
 	p.activeCluster = nil
-
-	p.activeStateQueue = append(p.activeStateQueue, false)
-	p.clustersMu.Unlock()
-
-	if reset {
-		if p.params.Listener != nil {
-			p.params.Listener.OnProbeClusterDone(info)
-		}
-	}
-
-	p.processActiveStateQueue()
 }
 
-func (p *Prober) AddCluster(mode ProbeClusterMode, desiredRateBps int, expectedRateBps int, minDuration time.Duration, maxDuration time.Duration) ProbeClusterId {
+func (p *Prober) AddCluster(
+	mode ProbeClusterMode,
+	desiredRateBps int,
+	expectedRateBps int,
+	duration time.Duration,
+) ProbeClusterId {
 	if desiredRateBps <= 0 {
 		return ProbeClusterIdInvalid
 	}
@@ -205,33 +197,38 @@ func (p *Prober) AddCluster(mode ProbeClusterMode, desiredRateBps int, expectedR
 		mode,
 		desiredRateBps,
 		expectedRateBps,
-		minDuration,
-		maxDuration,
+		duration,
 		p.params.Listener,
 	)
-	p.params.Logger.Debugw("cluster added", "cluster", cluster.String())
+	p.params.Logger.Debugw("cluster added", "cluster", cluster)
 
 	p.pushBackClusterAndMaybeStart(cluster)
 
 	return clusterId
 }
 
-func (p *Prober) PacketsSent(size int) {
+func (p *Prober) ClusterDone(info ProbeClusterInfo) {
 	cluster := p.getFrontCluster()
 	if cluster == nil {
 		return
 	}
 
-	cluster.PacketsSent(size)
+	if cluster.Id() == info.ProbeClusterId {
+		cluster.MarkCompleted(info)
+		p.params.Logger.Debugw("cluster done", "cluster", cluster)
+		p.popFrontCluster(cluster)
+	}
 }
 
-func (p *Prober) ProbeSent(size int) {
-	cluster := p.getFrontCluster()
-	if cluster == nil {
-		return
+func (p *Prober) GetActiveClusterId() ProbeClusterId {
+	p.clustersMu.RLock()
+	defer p.clustersMu.RUnlock()
+
+	if p.activeCluster != nil {
+		return p.activeCluster.Id()
 	}
 
-	cluster.ProbeSent(size)
+	return ProbeClusterIdInvalid
 }
 
 func (p *Prober) getFrontCluster() *Cluster {
@@ -268,12 +265,7 @@ func (p *Prober) popFrontCluster(cluster *Cluster) {
 		p.activeCluster = nil
 	}
 
-	if p.clusters.Len() == 0 {
-		p.activeStateQueue = append(p.activeStateQueue, false)
-	}
 	p.clustersMu.Unlock()
-
-	p.processActiveStateQueue()
 }
 
 func (p *Prober) pushBackClusterAndMaybeStart(cluster *Cluster) {
@@ -281,48 +273,19 @@ func (p *Prober) pushBackClusterAndMaybeStart(cluster *Cluster) {
 	p.clusters.PushBack(cluster)
 
 	if p.clusters.Len() == 1 {
-		p.activeStateQueue = append(p.activeStateQueue, true)
-
 		go p.run()
 	}
 	p.clustersMu.Unlock()
-
-	p.processActiveStateQueue()
-}
-
-func (p *Prober) processActiveStateQueue() {
-	if p.activeStateQueueInProcess.Swap(true) {
-		// processing queue
-		return
-	}
-
-	for {
-		p.clustersMu.Lock()
-		if len(p.activeStateQueue) == 0 {
-			p.clustersMu.Unlock()
-			break
-		}
-
-		isActive := p.activeStateQueue[0]
-		p.activeStateQueue = p.activeStateQueue[1:]
-		p.clustersMu.Unlock()
-
-		if p.params.Listener != nil {
-			p.params.Listener.OnActiveChanged(isActive)
-		}
-	}
-
-	p.activeStateQueueInProcess.Store(false)
 }
 
 func (p *Prober) run() {
-	// determine how long to sleep
 	cluster := p.getFrontCluster()
 	if cluster == nil {
 		return
 	}
 
 	timer := time.NewTimer(cluster.GetSleepDuration())
+	defer timer.Stop()
 	for {
 		<-timer.C
 
@@ -334,22 +297,6 @@ func (p *Prober) run() {
 
 		cluster.Process()
 
-		if cluster.IsFinished() {
-			p.params.Logger.Debugw("cluster finished", "cluster", cluster.String())
-
-			if p.params.Listener != nil {
-				p.params.Listener.OnProbeClusterDone(cluster.GetInfo())
-			}
-
-			p.popFrontCluster(cluster)
-		}
-
-		// determine how long to sleep
-		cluster := p.getFrontCluster()
-		if cluster == nil {
-			return
-		}
-
 		timer.Reset(cluster.GetSleepDuration())
 	}
 }
@@ -362,7 +309,7 @@ const (
 	ProbeClusterIdInvalid ProbeClusterId = 0
 
 	cBucketDuration  = 100 * time.Millisecond
-	cBytesPerProbe   = 1000
+	cBytesPerProbe   = 1100 // padding only packets are 255 bytes max + 20 byte header = 4 packets per probe
 	cMinProbeRateBps = 10000
 )
 
@@ -389,33 +336,74 @@ func (p ProbeClusterMode) String() string {
 // ---------------------------------------------------------------------------
 
 type ProbeClusterInfo struct {
-	Id        ProbeClusterId
-	BytesSent int
-	Duration  time.Duration
+	ProbeClusterId       ProbeClusterId
+	DesiredBytes         int
+	StartTime            int64
+	EndTime              int64
+	BytesProbe           int
+	BytesNonProbePrimary int
+	BytesNonProbeRTX     int
 }
 
-type clusterBucket struct {
-	desiredBytes       int
-	desiredElapsedTime time.Duration
-	sleepDuration      time.Duration
+var (
+	ProbeClusterInfoInvalid = ProbeClusterInfo{ProbeClusterId: ProbeClusterIdInvalid}
+)
+
+func (p ProbeClusterInfo) Bytes() int {
+	return p.BytesProbe + p.BytesNonProbePrimary + p.BytesNonProbeRTX
 }
 
+func (p ProbeClusterInfo) Duration() time.Duration {
+	return time.Duration(p.EndTime - p.StartTime)
+}
+
+func (p ProbeClusterInfo) MarshalLogObject(e zapcore.ObjectEncoder) error {
+	e.AddUint32("ProbeClusterId", uint32(p.ProbeClusterId))
+	e.AddInt("DesiredBytes", p.DesiredBytes)
+	e.AddTime("StartTime", time.Unix(0, p.StartTime))
+	e.AddTime("EndTime", time.Unix(0, p.EndTime))
+	e.AddDuration("Duration", p.Duration())
+	e.AddInt("BytesProbe", p.BytesProbe)
+	e.AddInt("BytesNonProbePrimary", p.BytesNonProbePrimary)
+	e.AddInt("BytesNonProbeRTX", p.BytesNonProbeRTX)
+	e.AddInt("Bytes", p.Bytes())
+	return nil
+}
+
+// ---------------------------------------------------------------------------
+
+type clusterBucket struct {
+	desiredNumProbes int
+	desiredBytes     int
+	sleepDuration    time.Duration
+}
+
+func (c clusterBucket) MarshalLogObject(e zapcore.ObjectEncoder) error {
+	e.AddInt("desiredNumProbes", c.desiredNumProbes)
+	e.AddInt("desiredBytes", c.desiredBytes)
+	e.AddDuration("sleepDuration", c.sleepDuration)
+	return nil
+}
+
+// ---------------------------------------------------------------------------
+
 type Cluster struct {
 	lock sync.RWMutex
 
-	id           ProbeClusterId
-	mode         ProbeClusterMode
-	listener     ProberListener
-	desiredBytes int
-	minDuration  time.Duration
-	maxDuration  time.Duration
+	id              ProbeClusterId
+	mode            ProbeClusterMode
+	desiredRateBps  int
+	expectedRateBps int
+	listener        ProberListener
+	desiredBytes    int
+	duration        time.Duration
 
 	buckets   []clusterBucket
 	bucketIdx int
 
-	bytesSentProbe    int
-	bytesSentNonProbe int
-	startTime         time.Time
+	numProbesSent    int
+	isComplete       bool
+	probeClusterInfo ProbeClusterInfo
 }
 
 func newCluster(
@@ -423,26 +411,26 @@ func newCluster(
 	mode ProbeClusterMode,
 	desiredRateBps int,
 	expectedRateBps int,
-	minDuration time.Duration,
-	maxDuration time.Duration,
+	duration time.Duration,
 	listener ProberListener,
 ) *Cluster {
 	c := &Cluster{
-		id:          id,
-		mode:        mode,
-		listener:    listener,
-		minDuration: minDuration,
-		maxDuration: maxDuration,
+		id:              id,
+		mode:            mode,
+		desiredRateBps:  desiredRateBps,
+		expectedRateBps: expectedRateBps,
+		listener:        listener,
+		duration:        duration,
 	}
-	c.initBuckets(desiredRateBps, expectedRateBps, minDuration)
+	c.initBuckets(desiredRateBps, expectedRateBps, duration)
 	c.desiredBytes = c.buckets[len(c.buckets)-1].desiredBytes
 	return c
 }
 
-func (c *Cluster) initBuckets(desiredRateBps int, expectedRateBps int, minDuration time.Duration) {
+func (c *Cluster) initBuckets(desiredRateBps int, expectedRateBps int, duration time.Duration) {
 	// split into granular buckets
 	// NOTE: splitting even if mode is unitform
-	numBuckets := int((minDuration.Milliseconds() + cBucketDuration.Milliseconds() - 1) / cBucketDuration.Milliseconds())
+	numBuckets := int((duration.Milliseconds() + cBucketDuration.Milliseconds() - 1) / cBucketDuration.Milliseconds())
 	if numBuckets < 1 {
 		numBuckets = 1
 	}
@@ -451,7 +439,8 @@ func (c *Cluster) initBuckets(desiredRateBps int, expectedRateBps int, minDurati
 	baseProbeRateBps := (desiredRateBps - expectedRateBps + numBuckets - 1) / numBuckets
 
 	runningDesiredBytes := 0
-	runningDesiredElapsedTime := time.Duration(0)
+	runningExpectedBytes := 0
+	runningNumProbes := 0
 
 	c.buckets = make([]clusterBucket, 0, numBuckets)
 	for bucketIdx := 0; bucketIdx < numBuckets; bucketIdx++ {
@@ -466,27 +455,29 @@ func (c *Cluster) initBuckets(desiredRateBps int, expectedRateBps int, minDurati
 		}
 		bucketProbeRateBytesPerSec := (bucketProbeRateBps + 7) / 8
 
-		// pace based on bytes per probe
-		numProbesPerSec := (bucketProbeRateBytesPerSec + cBytesPerProbe - 1) / cBytesPerProbe
-		sleepDurationMicroSeconds := int(float64(1_000_000)/float64(numProbesPerSec) + 0.5)
-
 		runningDesiredBytes += (((bucketProbeRateBytesPerSec + expectedRateBytesPerSec) * int(cBucketDuration.Milliseconds())) + 999) / 1000
-		runningDesiredElapsedTime += cBucketDuration
+		runningExpectedBytes += ((expectedRateBytesPerSec * int(cBucketDuration.Milliseconds())) + 999) / 1000
+		numProbesNeeded := ((runningDesiredBytes - runningExpectedBytes) + cBytesPerProbe - 1) / cBytesPerProbe
+
+		numProbesInBucket := numProbesNeeded - runningNumProbes
+		if numProbesInBucket <= 0 {
+			numProbesInBucket = 1
+		}
+		runningNumProbes += numProbesInBucket
+
+		sleepDurationMicroSeconds := int(float64(cBucketDuration.Microseconds())/float64(numProbesInBucket) + 0.5)
 
 		c.buckets = append(c.buckets, clusterBucket{
-			desiredBytes:       runningDesiredBytes,
-			desiredElapsedTime: runningDesiredElapsedTime,
-			sleepDuration:      time.Duration(sleepDurationMicroSeconds) * time.Microsecond,
+			desiredNumProbes: runningNumProbes,
+			desiredBytes:     runningDesiredBytes,
+			sleepDuration:    time.Duration(sleepDurationMicroSeconds) * time.Microsecond,
 		})
 	}
 }
 
 func (c *Cluster) Start() {
-	c.lock.Lock()
-	defer c.lock.Unlock()
-
-	if c.startTime.IsZero() {
-		c.startTime = time.Now()
+	if c.listener != nil {
+		c.listener.OnProbeClusterSwitch(c.id, c.desiredBytes)
 	}
 }
 
@@ -497,103 +488,57 @@ func (c *Cluster) GetSleepDuration() time.Duration {
 	return c.buckets[c.bucketIdx].sleepDuration
 }
 
-func (c *Cluster) PacketsSent(size int) {
+func (c *Cluster) Id() ProbeClusterId {
+	return c.id
+}
+
+func (c *Cluster) MarkCompleted(info ProbeClusterInfo) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
-	c.bytesSentNonProbe += size
-}
-
-func (c *Cluster) ProbeSent(size int) {
-	c.lock.Lock()
-	defer c.lock.Unlock()
-
-	c.bytesSentProbe += size
-}
-
-func (c *Cluster) IsFinished() bool {
-	c.lock.RLock()
-	defer c.lock.RUnlock()
-
-	// if already past deadline, end the cluster
-	timeElapsed := time.Since(c.startTime)
-	if timeElapsed > c.maxDuration {
-		return true
-	}
-
-	// do not end cluster until minDuration elapses even if rate is achieved.
-	// Ensures that the next cluster (if any) does not start early.
-	if (c.bytesSentProbe+c.bytesSentNonProbe) >= c.desiredBytes && timeElapsed >= c.minDuration {
-		return true
-	}
-
-	return false
-}
-
-func (c *Cluster) GetInfo() ProbeClusterInfo {
-	c.lock.RLock()
-	defer c.lock.RUnlock()
-
-	return ProbeClusterInfo{
-		Id:        c.id,
-		BytesSent: c.bytesSentProbe + c.bytesSentNonProbe,
-		Duration:  time.Since(c.startTime),
-	}
+	c.isComplete = true
+	c.probeClusterInfo = info
 }
 
 func (c *Cluster) Process() {
-	c.lock.RLock()
-	timeElapsed := time.Since(c.startTime)
-
-	// Calculate number of probe bytes that should have been sent since start.
-	// Overall goal is to send desired number of probe bytes in minDuration.
-	// However, it is possible that timeElapsed is more than minDuration due
-	// to scheduling variance. When overshooting time budget, use a capped
-	// short fall if there is a grace period given.
-	bytesShortFall := c.buckets[c.bucketIdx].desiredBytes - c.bytesSentProbe - c.bytesSentNonProbe
-	if bytesShortFall < 0 {
-		bytesShortFall = 0
+	c.lock.Lock()
+	if c.isComplete {
+		c.lock.Unlock()
+		return
 	}
-	// cap short fall to limit to 5 packets in an iteration
-	// 275 bytes per packet (255 max RTP padding payload + 20 bytes RTP header)
-	if bytesShortFall > (275 * 5) {
-		bytesShortFall = 275 * 5
-	}
-	// round up to packet size
-	bytesShortFall = ((bytesShortFall + 274) / 275) * 275
 
-	// move to next bucket if necessary
-	if timeElapsed > c.buckets[c.bucketIdx].desiredElapsedTime {
+	c.numProbesSent++
+	if c.numProbesSent >= c.buckets[c.bucketIdx].desiredNumProbes {
 		c.bucketIdx++
+		// stay in the last bucket till desired number of bytes are sent
 		if c.bucketIdx >= len(c.buckets) {
 			c.bucketIdx = len(c.buckets) - 1
 		}
 	}
-	c.lock.RUnlock()
+	c.lock.Unlock()
 
-	if bytesShortFall > 0 && c.listener != nil {
-		c.listener.OnSendProbe(bytesShortFall)
+	if c.listener != nil {
+		c.listener.OnSendProbe(cBytesPerProbe)
 	}
 
 	// STREAM-ALLOCATOR-TODO look at adapting sleep time based on how many bytes and how much time is left
 }
 
-func (c *Cluster) String() string {
-	activeTimeMs := int64(0)
-	if !c.startTime.IsZero() {
-		activeTimeMs = time.Since(c.startTime).Milliseconds()
+func (c *Cluster) MarshalLogObject(e zapcore.ObjectEncoder) error {
+	if c != nil {
+		e.AddUint32("id", uint32(c.id))
+		e.AddString("mode", c.mode.String())
+		e.AddInt("desiredRateBps", c.desiredRateBps)
+		e.AddInt("expectedRateBps", c.expectedRateBps)
+		e.AddInt("desiredBytes", c.desiredBytes)
+		e.AddDuration("duration", c.duration)
+		e.AddArray("buckets", logger.ObjectSlice(c.buckets))
+		e.AddInt("bucketIdx", c.bucketIdx)
+		e.AddInt("numProbesSent", c.numProbesSent)
+		e.AddBool("isComplete", c.isComplete)
+		e.AddObject("probeClusterInfo", c.probeClusterInfo)
 	}
-
-	return fmt.Sprintf("id: %d, mode: %s, bytes: desired %d / probe %d / non-probe %d / remaining: %d, time(ms): active %d / min %d / max %d",
-		c.id,
-		c.mode,
-		c.desiredBytes,
-		c.bytesSentProbe,
-		c.bytesSentNonProbe,
-		c.desiredBytes-c.bytesSentProbe-c.bytesSentNonProbe,
-		activeTimeMs,
-		c.minDuration.Milliseconds(),
-		c.maxDuration.Milliseconds())
+	return nil
 }
 
 // ----------------------------------------------------------------------

pkg/sfu/downtrack.go

@@ -37,6 +37,7 @@ import (
 	"github.com/livekit/protocol/utils/mono"
 
 	"github.com/livekit/livekit-server/pkg/sfu/buffer"
+	"github.com/livekit/livekit-server/pkg/sfu/ccutils"
 	"github.com/livekit/livekit-server/pkg/sfu/connectionquality"
 	"github.com/livekit/livekit-server/pkg/sfu/pacer"
 	act "github.com/livekit/livekit-server/pkg/sfu/rtpextension/abscapturetime"
@@ -190,9 +191,6 @@ type DownTrackStreamAllocatorListener interface {
 	// stream resumed
 	OnResume(dt *DownTrack)
 
-	// packet(s) sent
-	OnPacketsSent(dt *DownTrack, size int)
-
 	/* STREAM-ALLOCATOR-DATA
 	// NACKs received
 	OnNACK(dt *DownTrack, nackInfos []NackInfo)
@@ -291,14 +289,13 @@ type DownTrack struct {
 
 	activePaddingOnMuteUpTrack atomic.Bool
 
-	streamAllocatorLock             sync.RWMutex
-	streamAllocatorListener         DownTrackStreamAllocatorListener
-	streamAllocatorReportGeneration int
-	streamAllocatorBytesCounter     atomic.Uint32
+	streamAllocatorLock     sync.RWMutex
+	streamAllocatorListener DownTrackStreamAllocatorListener
 	/* STREAM-ALLOCATOR-DATA
 	bytesSent                       atomic.Uint32
 	bytesRetransmitted              atomic.Uint32
 	*/
+	probeClusterId atomic.Uint32
 
 	playoutDelay *PlayoutDelayController
 
@@ -627,48 +624,8 @@ func (d *DownTrack) getStreamAllocatorListener() DownTrackStreamAllocatorListene
 	return d.streamAllocatorListener
 }
 
-func (d *DownTrack) SetStreamAllocatorReportInterval(interval time.Duration) {
-	d.ClearStreamAllocatorReportInterval()
-
-	if interval == 0 {
-		return
-	}
-
-	d.streamAllocatorLock.Lock()
-	d.streamAllocatorBytesCounter.Store(0)
-
-	d.streamAllocatorReportGeneration++
-	gen := d.streamAllocatorReportGeneration
-	d.streamAllocatorLock.Unlock()
-
-	go func(generation int) {
-		timer := time.NewTimer(interval)
-		for {
-			<-timer.C
-
-			d.streamAllocatorLock.Lock()
-			if generation != d.streamAllocatorReportGeneration {
-				d.streamAllocatorLock.Unlock()
-				return
-			}
-
-			sal := d.streamAllocatorListener
-			bytes := d.streamAllocatorBytesCounter.Swap(0)
-			d.streamAllocatorLock.Unlock()
-
-			if sal != nil {
-				sal.OnPacketsSent(d, int(bytes))
-			}
-
-			timer.Reset(interval)
-		}
-	}(gen)
-}
-
-func (d *DownTrack) ClearStreamAllocatorReportInterval() {
-	d.streamAllocatorLock.Lock()
-	d.streamAllocatorReportGeneration++
-	d.streamAllocatorLock.Unlock()
+func (d *DownTrack) SetProbeClusterId(probeClusterId ccutils.ProbeClusterId) {
+	d.probeClusterId.Store(uint32(probeClusterId))
 }
 
 // ID is the unique identifier for this Track. This should be unique for the
@@ -942,18 +899,21 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
 		)
 	}
 
+	headerSize := hdr.MarshalSize()
 	d.updateStats(updateStatsParams{
 		packetTime:        extPkt.Arrival,
 		extSequenceNumber: tp.rtp.extSequenceNumber,
 		extTimestamp:      tp.rtp.extTimestamp,
 		isOutOfOrder:      extPkt.IsOutOfOrder,
-		headerSize:        hdr.MarshalSize(),
+		headerSize:        headerSize,
 		payloadSize:       len(payload),
 		marker:            hdr.Marker,
 	})
 	d.pacer.Enqueue(&pacer.Packet{
 		Header:             hdr,
+		HeaderSize:         headerSize,
 		Payload:            payload,
+		ProbeClusterId:     ccutils.ProbeClusterId(d.probeClusterId.Load()),
 		AbsSendTimeExtID:   uint8(d.absSendTimeExtID),
 		TransportWideExtID: uint8(d.transportWideExtID),
 		WriteStream:        d.writeStream,
@@ -986,7 +946,11 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
 
 // WritePaddingRTP tries to write as many padding only RTP packets as necessary
 // to satisfy given size to the DownTrack
-func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMarker bool) int {
+func (d *DownTrack) WritePaddingRTP(
+	bytesToSend int,
+	paddingOnMute bool,
+	forceMarker bool,
+) int {
 	if !d.writable.Load() {
 		return 0
 	}
@@ -1070,7 +1034,10 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMa
 		})
 		d.pacer.Enqueue(&pacer.Packet{
 			Header:             hdr,
+			HeaderSize:         hdrSize,
 			Payload:            payload,
+			ProbeClusterId:     ccutils.ProbeClusterId(d.probeClusterId.Load()),
+			IsProbe:            true,
 			AbsSendTimeExtID:   uint8(d.absSendTimeExtID),
 			TransportWideExtID: uint8(d.transportWideExtID),
 			WriteStream:        d.writeStream,
@@ -1227,8 +1194,6 @@ func (d *DownTrack) CloseWithFlush(flush bool) {
 	if onCloseHandler := d.getOnCloseHandler(); onCloseHandler != nil {
 		onCloseHandler(!flush)
 	}
-
-	d.ClearStreamAllocatorReportInterval()
 }
 
 func (d *DownTrack) SetMaxSpatialLayer(spatialLayer int32) {
@@ -1614,16 +1579,19 @@ func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan
 					return
 				}
 
+				headerSize := hdr.MarshalSize()
 				d.updateStats(updateStatsParams{
 					packetTime:        mono.UnixNano(),
 					extSequenceNumber: snts[i].extSequenceNumber,
 					extTimestamp:      snts[i].extTimestamp,
-					headerSize:        hdr.MarshalSize(),
+					headerSize:        headerSize,
 					payloadSize:       len(payload),
 				})
 				d.pacer.Enqueue(&pacer.Packet{
 					Header:             hdr,
+					HeaderSize:         headerSize,
 					Payload:            payload,
+					ProbeClusterId:     ccutils.ProbeClusterId(d.probeClusterId.Load()),
 					AbsSendTimeExtID:   uint8(d.absSendTimeExtID),
 					TransportWideExtID: uint8(d.transportWideExtID),
 					WriteStream:        d.writeStream,
@@ -1964,18 +1932,21 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
 		}
 		d.addDummyExtensions(&pkt.Header)
 
+		headerSize := pkt.Header.MarshalSize()
 		d.updateStats(updateStatsParams{
 			packetTime:        mono.UnixNano(),
 			extSequenceNumber: epm.extSequenceNumber,
 			extTimestamp:      epm.extTimestamp,
 			isOutOfOrder:      true,
-			headerSize:        pkt.Header.MarshalSize(),
+			headerSize:        headerSize,
 			payloadSize:       len(payload),
 			isRTX:             true,
 		})
 		d.pacer.Enqueue(&pacer.Packet{
 			Header:             &pkt.Header,
+			HeaderSize:         headerSize,
 			Payload:            payload,
+			ProbeClusterId:     ccutils.ProbeClusterId(d.probeClusterId.Load()),
 			IsRTX:              true,
 			AbsSendTimeExtID:   uint8(d.absSendTimeExtID),
 			TransportWideExtID: uint8(d.transportWideExtID),
@@ -2178,18 +2149,21 @@ func (d *DownTrack) sendSilentFrameOnMuteForOpus() {
 				return
 			}
 
+			headerSize := hdr.MarshalSize()
 			d.updateStats(updateStatsParams{
 				packetTime:        mono.UnixNano(),
 				extSequenceNumber: snts[i].extSequenceNumber,
 				extTimestamp:      snts[i].extTimestamp,
-				headerSize:        hdr.MarshalSize(),
+				headerSize:        headerSize,
 				payloadSize:       len(payload),
 				// although this is using empty frames, mark as padding as these are used to trigger Pion OnTrack only
 				isPadding: true,
 			})
 			d.pacer.Enqueue(&pacer.Packet{
 				Header:             hdr,
+				HeaderSize:         headerSize,
 				Payload:            payload,
+				ProbeClusterId:     ccutils.ProbeClusterId(d.probeClusterId.Load()),
 				AbsSendTimeExtID:   uint8(d.absSendTimeExtID),
 				TransportWideExtID: uint8(d.transportWideExtID),
 				WriteStream:        d.writeStream,
@@ -2235,9 +2209,6 @@ type updateStatsParams struct {
 
 func (d *DownTrack) updateStats(params updateStatsParams) {
 	if !params.disableCounter {
-		// STREAM-ALLOCATOR-TODO: remove this stream allocator bytes counter once stream allocator changes fully to pull bytes counter
-		size := uint32(params.headerSize + params.payloadSize)
-		d.streamAllocatorBytesCounter.Add(size)
 		/* STREAM-ALLOCATOR-DATA
 		if params.isRTX {
 			d.bytesRetransmitted.Add(size)

pkg/sfu/pacer/base.go

@@ -29,12 +29,15 @@ type Base struct {
 	logger logger.Logger
 
 	bwe bwe.BWE
+
+	*ProbeObserver
 }
 
 func NewBase(logger logger.Logger, bwe bwe.BWE) *Base {
 	return &Base{
-		logger: logger,
-		bwe:    bwe,
+		logger:        logger,
+		bwe:           bwe,
+		ProbeObserver: NewProbeObserver(logger),
 	}
 }
 
@@ -84,10 +87,11 @@ func (b *Base) patchRTPHeaderExtensions(p *Packet) error {
 		}
 	}
 
+	packetSize := p.HeaderSize + len(p.Payload)
 	if p.TransportWideExtID != 0 && b.bwe != nil {
 		twccSN := b.bwe.RecordPacketSendAndGetSequenceNumber(
 			sendingAt.UnixMicro(),
-			p.Header.MarshalSize()+len(p.Payload),
+			packetSize,
 			p.IsRTX,
 		)
 		twccExt := rtp.TransportCCExtension{
@@ -103,6 +107,7 @@ func (b *Base) patchRTPHeaderExtensions(p *Packet) error {
 		}
 	}
 
+	b.ProbeObserver.RecordPacket(packetSize, p.IsRTX, p.ProbeClusterId, p.IsProbe)
 	return nil
 }
 

pkg/sfu/pacer/pacer.go

@@ -18,14 +18,18 @@ import (
 	"sync"
 	"time"
 
+	"github.com/livekit/livekit-server/pkg/sfu/ccutils"
 	"github.com/pion/rtp"
 	"github.com/pion/webrtc/v4"
 )
 
 type Packet struct {
 	Header             *rtp.Header
+	HeaderSize         int
 	Payload            []byte
 	IsRTX              bool
+	ProbeClusterId     ccutils.ProbeClusterId
+	IsProbe            bool
 	AbsSendTimeExtID   uint8
 	TransportWideExtID uint8
 	WriteStream        webrtc.TrackLocalWriter
@@ -39,6 +43,14 @@ type Pacer interface {
 
 	SetInterval(interval time.Duration)
 	SetBitrate(bitrate int)
+
+	SetPacerProbeObserverListener(listener PacerProbeObserverListener)
+	StartProbeCluster(probeClusterId ccutils.ProbeClusterId, desiredBytes int)
+	EndProbeCluster(probeClusterId ccutils.ProbeClusterId) ccutils.ProbeClusterInfo
+}
+
+type PacerProbeObserverListener interface {
+	OnPacerProbeObserverClusterComplete(probeClusterId ccutils.ProbeClusterId)
 }
 
 // ------------------------------------------------

pkg/sfu/pacer/probe_observer.go

@@ -0,0 +1,154 @@
+// Copyright 2023 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package pacer
+
+import (
+	"sync"
+	"sync/atomic"
+
+	"github.com/livekit/livekit-server/pkg/sfu/ccutils"
+	"github.com/livekit/protocol/logger"
+	"github.com/livekit/protocol/utils/mono"
+)
+
+type ProbeObserver struct {
+	logger logger.Logger
+
+	listener PacerProbeObserverListener
+
+	isInProbe atomic.Bool
+
+	lock                     sync.Mutex
+	clusterStartTime         int64
+	activeProbeClusterId     ccutils.ProbeClusterId
+	desiredProbeClusterBytes int
+	bytesNonProbePrimary     int
+	bytesNonProbeRTX         int
+	bytesProbe               int
+	isActiveClusterDone      bool
+}
+
+func NewProbeObserver(logger logger.Logger) *ProbeObserver {
+	return &ProbeObserver{
+		logger: logger,
+	}
+}
+
+func (po *ProbeObserver) SetPacerProbeObserverListener(listener PacerProbeObserverListener) {
+	po.listener = listener
+}
+
+func (po *ProbeObserver) StartProbeCluster(probeClusterId ccutils.ProbeClusterId, desiredBytes int) {
+	if po.isInProbe.Load() {
+		po.logger.Warnw(
+			"ignoring start of a new probe cluster when already active", nil,
+			"probeClusterId", probeClusterId,
+			"desiredBytes", desiredBytes,
+		)
+		return
+	}
+
+	po.lock.Lock()
+	defer po.lock.Unlock()
+
+	po.clusterStartTime = mono.UnixNano()
+	po.activeProbeClusterId = probeClusterId
+	po.desiredProbeClusterBytes = desiredBytes
+	po.bytesNonProbePrimary = 0
+	po.bytesNonProbeRTX = 0
+	po.bytesProbe = 0
+	po.isActiveClusterDone = false
+
+	po.isInProbe.Store(true)
+}
+
+func (po *ProbeObserver) EndProbeCluster(probeClusterId ccutils.ProbeClusterId) ccutils.ProbeClusterInfo {
+	if !po.isInProbe.Load() {
+		// probe not active
+		if probeClusterId != ccutils.ProbeClusterIdInvalid {
+			po.logger.Debugw(
+				"ignoring end of a probe cluster when not active",
+				"probeClusterId", probeClusterId,
+			)
+		}
+		return ccutils.ProbeClusterInfoInvalid
+	}
+
+	po.lock.Lock()
+	defer po.lock.Unlock()
+
+	if po.activeProbeClusterId != probeClusterId {
+		// probe cluster id not active
+		po.logger.Warnw(
+			"ignoring end of a probe cluster of a non-active one", nil,
+			"probeClusterId", probeClusterId,
+			"active", po.activeProbeClusterId,
+		)
+		return ccutils.ProbeClusterInfoInvalid
+	}
+
+	clusterInfo := ccutils.ProbeClusterInfo{
+		ProbeClusterId:       po.activeProbeClusterId,
+		DesiredBytes:         po.desiredProbeClusterBytes,
+		StartTime:            po.clusterStartTime,
+		EndTime:              mono.UnixNano(),
+		BytesProbe:           po.bytesProbe,
+		BytesNonProbePrimary: po.bytesNonProbePrimary,
+		BytesNonProbeRTX:     po.bytesNonProbeRTX,
+	}
+
+	po.activeProbeClusterId = ccutils.ProbeClusterIdInvalid
+	po.isInProbe.Store(false)
+
+	return clusterInfo
+}
+
+func (po *ProbeObserver) RecordPacket(size int, isRTX bool, probeClusterId ccutils.ProbeClusterId, isProbe bool) {
+	if !po.isInProbe.Load() {
+		return
+	}
+
+	po.lock.Lock()
+	if probeClusterId != po.activeProbeClusterId || po.isActiveClusterDone {
+		po.lock.Unlock()
+		return
+	}
+
+	if isProbe {
+		po.bytesProbe += size
+	} else {
+		if isRTX {
+			po.bytesNonProbeRTX += size
+		} else {
+			po.bytesNonProbePrimary += size
+		}
+	}
+
+	notify := false
+	var clusterId ccutils.ProbeClusterId
+	if !po.isActiveClusterDone && po.bytesProbe+po.bytesNonProbePrimary+po.bytesNonProbeRTX >= po.desiredProbeClusterBytes {
+		po.isActiveClusterDone = true
+
+		notify = true
+		clusterId = po.activeProbeClusterId
+	}
+	po.lock.Unlock()
+
+	if notify && po.listener != nil {
+		po.listener.OnPacerProbeObserverClusterComplete(clusterId)
+	}
+}
+
+// ------------------------------------------------

pkg/sfu/streamallocator/probe_controller.go

@@ -20,6 +20,7 @@ import (
 
 	"github.com/livekit/livekit-server/pkg/sfu/bwe"
 	"github.com/livekit/livekit-server/pkg/sfu/ccutils"
+	"github.com/livekit/livekit-server/pkg/sfu/pacer"
 	"github.com/livekit/protocol/logger"
 )
 
@@ -39,7 +40,6 @@ type ProbeControllerConfig struct {
 	MinBps                 int64         `yaml:"min_bps,omitempty"`
 	MinDuration            time.Duration `yaml:"min_duration,omitempty"`
 	MaxDuration            time.Duration `yaml:"max_duration,omitempty"`
-	DurationOverflowFactor float64       `yaml:"duration_overflow_factor,omitempty"`
 	DurationIncreaseFactor float64       `yaml:"duration_increase_factor,omitempty"`
 }
 
@@ -58,7 +58,6 @@ var (
 		MinBps:                 200_000,
 		MinDuration:            200 * time.Millisecond,
 		MaxDuration:            20 * time.Second,
-		DurationOverflowFactor: 1.25,
 		DurationIncreaseFactor: 1.5,
 	}
 )
@@ -68,6 +67,8 @@ var (
 type ProbeControllerParams struct {
 	Config ProbeControllerConfig
 	Prober *ccutils.Prober
+	BWE    bwe.BWE
+	Pacer  pacer.Pacer
 	Logger logger.Logger
 }
 
@@ -75,7 +76,6 @@ type ProbeController struct {
 	params ProbeControllerParams
 
 	lock                      sync.RWMutex
-	bwe                       bwe.BWE
 	probeInterval             time.Duration
 	lastProbeStartTime        time.Time
 	probeGoalBps              int64
@@ -98,13 +98,6 @@ func NewProbeController(params ProbeControllerParams) *ProbeController {
 	return p
 }
 
-func (p *ProbeController) SetBWE(bwe bwe.BWE) {
-	p.lock.Lock()
-	defer p.lock.Unlock()
-
-	p.bwe = bwe
-}
-
 func (p *ProbeController) Reset() {
 	p.lock.Lock()
 	defer p.lock.Unlock()
@@ -118,14 +111,15 @@ func (p *ProbeController) Reset() {
 	p.clearProbeLocked()
 }
 
-func (p *ProbeController) ProbeClusterDone(info ccutils.ProbeClusterInfo) {
+func (p *ProbeController) ProbeClusterDone(probeClusterId ccutils.ProbeClusterId) {
 	p.lock.Lock()
 	defer p.lock.Unlock()
 
-	if p.probeClusterId != info.Id {
-		p.params.Logger.Debugw("not expected probe cluster", "probeClusterId", p.probeClusterId, "resetProbeClusterId", info.Id)
+	if p.probeClusterId != probeClusterId {
+		p.params.Logger.Debugw("not expected probe cluster", "probeClusterId", p.probeClusterId, "resetProbeClusterId", probeClusterId)
 	} else {
-		p.doneProbeClusterInfo = info
+		p.doneProbeClusterInfo = p.params.Pacer.EndProbeCluster(probeClusterId)
+		p.params.Prober.ClusterDone(p.doneProbeClusterInfo)
 	}
 }
 
@@ -149,14 +143,14 @@ func (p *ProbeController) MaybeFinalizeProbe(
 
 	if (isComplete || p.abortedProbeClusterId != ccutils.ProbeClusterIdInvalid) &&
 		p.probeEndTime.IsZero() &&
-		p.doneProbeClusterInfo.Id != ccutils.ProbeClusterIdInvalid && p.doneProbeClusterInfo.Id == p.probeClusterId {
+		p.doneProbeClusterInfo.ProbeClusterId != ccutils.ProbeClusterIdInvalid && p.doneProbeClusterInfo.ProbeClusterId == p.probeClusterId {
 		// ensure any queueing due to probing is flushed
 		// STREAM-ALLOCATOR-TODO: ProbeControllerConfig.SettleWait should actually be a certain number of RTTs.
 		expectedDuration := float64(0.0)
 		if lowestEstimate != 0 {
-			expectedDuration = float64(p.doneProbeClusterInfo.BytesSent*8*1000) / float64(lowestEstimate)
+			expectedDuration = float64(p.doneProbeClusterInfo.Bytes()*8*1000) / float64(lowestEstimate)
 		}
-		queueTime := expectedDuration - float64(p.doneProbeClusterInfo.Duration.Milliseconds())
+		queueTime := expectedDuration - float64(p.doneProbeClusterInfo.Duration().Milliseconds())
 		if queueTime < 0.0 {
 			queueTime = 0.0
 		}
@@ -164,7 +158,7 @@ func (p *ProbeController) MaybeFinalizeProbe(
 		if queueWait > p.params.Config.SettleWaitMax {
 			queueWait = p.params.Config.SettleWaitMax
 		}
-		p.probeEndTime = p.lastProbeStartTime.Add(queueWait + p.doneProbeClusterInfo.Duration)
+		p.probeEndTime = p.lastProbeStartTime.Add(queueWait + p.doneProbeClusterInfo.Duration())
 		p.params.Logger.Debugw(
 			"setting probe end time",
 			"probeClusterId", p.probeClusterId,
@@ -223,7 +217,7 @@ func (p *ProbeController) InitProbe(probeGoalDeltaBps int64, expectedBandwidthUs
 	}
 	p.probeGoalBps = expectedBandwidthUsage + desiredIncreaseBps
 
-	p.doneProbeClusterInfo = ccutils.ProbeClusterInfo{Id: ccutils.ProbeClusterIdInvalid}
+	p.doneProbeClusterInfo = ccutils.ProbeClusterInfoInvalid
 	p.abortedProbeClusterId = ccutils.ProbeClusterIdInvalid
 	p.goalReachedProbeClusterId = ccutils.ProbeClusterIdInvalid
 
@@ -236,7 +230,6 @@ func (p *ProbeController) InitProbe(probeGoalDeltaBps int64, expectedBandwidthUs
 		int(p.probeGoalBps),
 		int(expectedBandwidthUsage),
 		p.probeDuration,
-		time.Duration(float64(p.probeDuration.Milliseconds())*p.params.Config.DurationOverflowFactor)*time.Millisecond,
 	)
 
 	p.pollProbe(p.probeClusterId, expectedBandwidthUsage)
@@ -245,11 +238,7 @@ func (p *ProbeController) InitProbe(probeGoalDeltaBps int64, expectedBandwidthUs
 }
 
 func (p *ProbeController) pollProbe(probeClusterId ccutils.ProbeClusterId, expectedBandwidthUsage int64) {
-	if p.bwe == nil {
-		return
-	}
-
-	p.bwe.ProbingStart(expectedBandwidthUsage)
+	p.params.BWE.ProbingStart(expectedBandwidthUsage)
 
 	go func() {
 		for {
@@ -261,7 +250,7 @@ func (p *ProbeController) pollProbe(probeClusterId ccutils.ProbeClusterId, expec
 
 			done := false
 
-			_, trend, _, highestEstimate := p.bwe.GetProbeStatus()
+			_, trend, _, highestEstimate := p.params.BWE.GetProbeStatus()
 			if !p.probeTrendObserved && trend != bwe.ChannelTrendNeutral {
 				p.probeTrendObserved = true
 			}
@@ -315,7 +304,7 @@ func (p *ProbeController) pollProbe(probeClusterId ccutils.ProbeClusterId, expec
 
 func (p *ProbeController) clearProbeLocked() {
 	p.probeClusterId = ccutils.ProbeClusterIdInvalid
-	p.doneProbeClusterInfo = ccutils.ProbeClusterInfo{Id: ccutils.ProbeClusterIdInvalid}
+	p.doneProbeClusterInfo = ccutils.ProbeClusterInfoInvalid
 	p.abortedProbeClusterId = ccutils.ProbeClusterIdInvalid
 	p.goalReachedProbeClusterId = ccutils.ProbeClusterIdInvalid
 }
@@ -343,7 +332,7 @@ func (p *ProbeController) increaseProbeDurationLocked() {
 }
 
 func (p *ProbeController) StopProbe() {
-	p.params.Prober.Reset()
+	p.params.Prober.Reset(p.params.Pacer.EndProbeCluster(p.probeClusterId))
 }
 
 func (p *ProbeController) AbortProbe() {

pkg/sfu/streamallocator/streamallocator.go

@@ -32,6 +32,7 @@ import (
 	"github.com/livekit/livekit-server/pkg/sfu/buffer"
 	"github.com/livekit/livekit-server/pkg/sfu/bwe"
 	"github.com/livekit/livekit-server/pkg/sfu/ccutils"
+	"github.com/livekit/livekit-server/pkg/sfu/pacer"
 	"github.com/livekit/livekit-server/pkg/utils"
 )
 
@@ -82,13 +83,13 @@ const (
 	streamAllocatorSignalEstimate
 	streamAllocatorSignalPeriodicPing
 	streamAllocatorSignalSendProbe
-	streamAllocatorSignalProbeClusterDone
 	streamAllocatorSignalResume
 	streamAllocatorSignalSetAllowPause
 	streamAllocatorSignalSetChannelCapacity
 	// STREAM-ALLOCATOR-DATA streamAllocatorSignalNACK
 	// STREAM-ALLOCATOR-DATA streamAllocatorSignalRTCPReceiverReport
 	streamAllocatorSignalCongestionStateChange
+	streamAllocatorSignalPacerProbeObserverClusterComplete
 )
 
 func (s streamAllocatorSignal) String() string {
@@ -105,8 +106,6 @@ func (s streamAllocatorSignal) String() string {
 		return "PERIODIC_PING"
 	case streamAllocatorSignalSendProbe:
 		return "SEND_PROBE"
-	case streamAllocatorSignalProbeClusterDone:
-		return "PROBE_CLUSTER_DONE"
 	case streamAllocatorSignalResume:
 		return "RESUME"
 	case streamAllocatorSignalSetAllowPause:
@@ -121,6 +120,8 @@ func (s streamAllocatorSignal) String() string {
 		*/
 	case streamAllocatorSignalCongestionStateChange:
 		return "CONGESTION_STATE_CHANGE"
+	case streamAllocatorSignalPacerProbeObserverClusterComplete:
+		return "PACER_PROBE_OBSERVER_CLUSTER_COMPLETE"
 	default:
 		return fmt.Sprintf("%d", int(s))
 	}
@@ -168,6 +169,8 @@ var (
 
 type StreamAllocatorParams struct {
 	Config StreamAllocatorConfig
+	BWE    bwe.BWE
+	Pacer  pacer.Pacer
 	Logger logger.Logger
 }
 
@@ -176,7 +179,6 @@ type StreamAllocator struct {
 
 	onStreamStateChange func(update *StreamStateUpdate) error
 
-	bwe                    bwe.BWE
 	sendSideBWEInterceptor cc.BandwidthEstimator
 
 	enabled    bool
@@ -227,9 +229,14 @@ func NewStreamAllocator(params StreamAllocatorParams, enabled bool, allowPause b
 	s.probeController = NewProbeController(ProbeControllerParams{
 		Config: s.params.Config.ProbeController,
 		Prober: s.prober,
+		BWE:    s.params.BWE,
+		Pacer:  s.params.Pacer,
 		Logger: params.Logger,
 	})
 
+	s.params.BWE.SetBWEListener(s)
+	s.params.Pacer.SetPacerProbeObserverListener(s)
+
 	s.resetState()
 
 	return s
@@ -254,14 +261,6 @@ func (s *StreamAllocator) OnStreamStateChange(f func(update *StreamStateUpdate)
 	s.onStreamStateChange = f
 }
 
-func (s *StreamAllocator) SetBWE(bwe bwe.BWE) {
-	if bwe != nil {
-		bwe.SetBWEListener(s)
-	}
-	s.bwe = bwe
-	s.probeController.SetBWE(bwe)
-}
-
 func (s *StreamAllocator) SetSendSideBWEInterceptor(sendSideBWEInterceptor cc.BandwidthEstimator) {
 	if sendSideBWEInterceptor != nil {
 		sendSideBWEInterceptor.OnTargetBitrateChange(s.onTargetBitrateChange)
@@ -295,10 +294,7 @@ func (s *StreamAllocator) AddTrack(downTrack *sfu.DownTrack, params AddTrackPara
 	}
 
 	downTrack.SetStreamAllocatorListener(s)
-	if s.prober.IsRunning() {
-		// STREAM-ALLOCATOR-TODO: this can be changed to adapt to probe rate
-		downTrack.SetStreamAllocatorReportInterval(50 * time.Millisecond)
-	}
+	downTrack.SetProbeClusterId(s.prober.GetActiveClusterId())
 
 	s.maybePostEventAllocateTrack(downTrack)
 }
@@ -346,9 +342,7 @@ func (s *StreamAllocator) SetChannelCapacity(channelCapacity int64) {
 }
 
 func (s *StreamAllocator) resetState() {
-	if s.bwe != nil {
-		s.bwe.Reset()
-	}
+	s.params.BWE.Reset()
 	s.probeController.Reset()
 
 	s.state = streamAllocatorStateStable
@@ -440,9 +434,7 @@ func (s *StreamAllocator) OnTransportCCFeedback(downTrack *sfu.DownTrack, fb *rt
 		s.sendSideBWEInterceptor.WriteRTCP([]rtcp.Packet{fb}, nil)
 	}
 
-	if s.bwe != nil {
-		s.bwe.HandleTWCCFeedback(fb)
-	}
+	s.params.BWE.HandleTWCCFeedback(fb)
 }
 
 // called when target bitrate changes (send side bandwidth estimation)
@@ -463,10 +455,7 @@ type congestionStateChangeData struct {
 func (s *StreamAllocator) OnCongestionStateChange(congestionState bwe.CongestionState, estimatedAvailableChannelCapacity int64) {
 	s.postEvent(Event{
 		Signal: streamAllocatorSignalCongestionStateChange,
-		Data: congestionStateChangeData{
-			congestionState:                   congestionState,
-			estimatedAvailableChannelCapacity: estimatedAvailableChannelCapacity,
-		},
+		Data:   congestionStateChangeData{congestionState, estimatedAvailableChannelCapacity},
 	})
 }
 
@@ -522,11 +511,6 @@ func (s *StreamAllocator) OnResume(downTrack *sfu.DownTrack) {
 	})
 }
 
-// called by a video DownTrack to report packet send
-func (s *StreamAllocator) OnPacketsSent(downTrack *sfu.DownTrack, size int) {
-	s.prober.PacketsSent(size)
-}
-
 /* STREAM-ALLOCATOR-DATA
 // called by a video DownTrack when it processes NACKs
 func (s *StreamAllocator) OnNACK(downTrack *sfu.DownTrack, nackInfos []sfu.NackInfo) {
@@ -556,24 +540,21 @@ func (s *StreamAllocator) OnSendProbe(bytesToSend int) {
 	})
 }
 
-// called when prober finishes a probe cluster, could be called when prober is reset which stops an active cluster
-func (s *StreamAllocator) OnProbeClusterDone(info ccutils.ProbeClusterInfo) {
-	s.postEvent(Event{
-		Signal: streamAllocatorSignalProbeClusterDone,
-		Data:   info,
-	})
+// called when probe cluster changes
+func (s *StreamAllocator) OnProbeClusterSwitch(probeClusterId ccutils.ProbeClusterId, desiredBytes int) {
+	s.params.Pacer.StartProbeCluster(probeClusterId, desiredBytes)
+
+	for _, t := range s.getTracks() {
+		t.DownTrack().SetProbeClusterId(probeClusterId)
+	}
 }
 
-// called when prober active state changes
-func (s *StreamAllocator) OnActiveChanged(isActive bool) {
-	for _, t := range s.getTracks() {
-		if isActive {
-			// STREAM-ALLOCATOR-TODO: this can be changed to adapt to probe rate
-			t.DownTrack().SetStreamAllocatorReportInterval(50 * time.Millisecond)
-		} else {
-			t.DownTrack().ClearStreamAllocatorReportInterval()
-		}
-	}
+// called when pacer probe observer observes a cluster completion
+func (s *StreamAllocator) OnPacerProbeObserverClusterComplete(probeClusterId ccutils.ProbeClusterId) {
+	s.postEvent(Event{
+		Signal: streamAllocatorSignalPacerProbeObserverClusterComplete,
+		Data:   probeClusterId,
+	})
 }
 
 // called to check if track should participate in BWE
@@ -652,8 +633,6 @@ func (s *StreamAllocator) postEvent(event Event) {
 			event.handleSignalPeriodicPing(event)
 		case streamAllocatorSignalSendProbe:
 			event.handleSignalSendProbe(event)
-		case streamAllocatorSignalProbeClusterDone:
-			event.handleSignalProbeClusterDone(event)
 		case streamAllocatorSignalResume:
 			event.handleSignalResume(event)
 		case streamAllocatorSignalSetAllowPause:
@@ -668,6 +647,8 @@ func (s *StreamAllocator) postEvent(event Event) {
 			*/
 		case streamAllocatorSignalCongestionStateChange:
 			s.handleSignalCongestionStateChange(event)
+		case streamAllocatorSignalPacerProbeObserverClusterComplete:
+			event.handleSignalPacerProbeObserverClusterComplete(event)
 		}
 	}, event)
 }
@@ -705,29 +686,25 @@ func (s *StreamAllocator) handleSignalEstimate(event Event) {
 	// always update NACKs
 	packetDelta, repeatedNackDelta := s.getNackDelta()
 
-	if s.bwe != nil {
-		s.bwe.HandleREMB(
-			receivedEstimate,
-			s.probeController.DoesProbeNeedFinalize(), // waiting for goal reached OR aborted probe to finalize
-			s.getExpectedBandwidthUsage(),
-			packetDelta,
-			repeatedNackDelta,
-		)
-	}
+	s.params.BWE.HandleREMB(
+		receivedEstimate,
+		s.probeController.DoesProbeNeedFinalize(), // waiting for goal reached OR aborted probe to finalize
+		s.getExpectedBandwidthUsage(),
+		packetDelta,
+		repeatedNackDelta,
+	)
 }
 
 func (s *StreamAllocator) handleSignalPeriodicPing(Event) {
 	// finalize probe if necessary
-	if s.bwe != nil {
-		isValidSignal, trend, lowestEstimate, highestEstimate := s.bwe.GetProbeStatus()
-		isHandled, isNotFailing, isGoalReached := s.probeController.MaybeFinalizeProbe(
-			isValidSignal,
-			trend,
-			lowestEstimate,
-		)
-		if isHandled {
-			s.onProbeDone(isNotFailing, isGoalReached, highestEstimate)
-		}
+	isValidSignal, trend, lowestEstimate, highestEstimate := s.params.BWE.GetProbeStatus()
+	isHandled, isNotFailing, isGoalReached := s.probeController.MaybeFinalizeProbe(
+		isValidSignal,
+		trend,
+		lowestEstimate,
+	)
+	if isHandled {
+		s.onProbeDone(isNotFailing, isGoalReached, highestEstimate)
 	}
 
 	// probe if necessary and timing is right
@@ -756,15 +733,6 @@ func (s *StreamAllocator) handleSignalSendProbe(event Event) {
 			break
 		}
 	}
-
-	if bytesSent != 0 {
-		s.prober.ProbeSent(bytesSent)
-	}
-}
-
-func (s *StreamAllocator) handleSignalProbeClusterDone(event Event) {
-	info, _ := event.Data.(ccutils.ProbeClusterInfo)
-	s.probeController.ProbeClusterDone(info)
 }
 
 func (s *StreamAllocator) handleSignalResume(event Event) {
@@ -877,6 +845,11 @@ func (s *StreamAllocator) handleSignalCongestionStateChange(event Event) {
 	s.congestionState = cscd.congestionState
 }
 
+func (s *StreamAllocator) handleSignalPacerProbeObserverClusterComplete(event Event) {
+	probeClusterId, _ := event.Data.(ccutils.ProbeClusterId)
+	s.probeController.ProbeClusterDone(probeClusterId)
+}
+
 func (s *StreamAllocator) setState(state streamAllocatorState) {
 	if s.state == state {
 		return
@@ -889,14 +862,12 @@ func (s *StreamAllocator) setState(state streamAllocatorState) {
 	s.probeController.Reset()
 
 	// a fresh start after state transition to get clean data
-	if s.bwe != nil {
-		// BWE-TODO: ssbwe maybe should not reset like this as it might have useful state across
-		// BWE-TODO: state changes in this module, actually even remotebwe should also manage it
-		// BWE-TODO: internally, Reset should probably only be used if all managed tracks go away
-		// BWE-TODO: and we can get a clean start, mimicking existing behaviour till this can be
-		// BWE-TODO: evaluated more.
-		s.bwe.Reset()
-	}
+	// BWE-TODO: ssbwe maybe should not reset like this as it might have useful state across
+	// BWE-TODO: state changes in this module, actually even remotebwe should also manage it
+	// BWE-TODO: internally, Reset should probably only be used if all managed tracks go away
+	// BWE-TODO: and we can get a clean start, mimicking existing behaviour till this can be
+	// BWE-TODO: evaluated more.
+	s.params.BWE.Reset()
 }
 
 func (s *StreamAllocator) adjustState() {
@@ -1066,9 +1037,7 @@ func (s *StreamAllocator) allocateTrack(track *Track) {
 }
 
 func (s *StreamAllocator) onProbeDone(isNotFailing bool, isGoalReached bool, highestEstimate int64) {
-	if s.bwe != nil {
-		s.bwe.ProbingEnd(isNotFailing, isGoalReached)
-	}
+	s.params.BWE.ProbingEnd(isNotFailing, isGoalReached)
 
 	if !isNotFailing {
 		return