livekit/pkg/sfu/bwe/sendsidebwe/traffic_stats.go

// 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 sendsidebwe

import (
	"math"
	"time"

	"github.com/livekit/protocol/logger"
	"go.uber.org/zap/zapcore"
)

// -----------------------------------------------------------

type WeightedLossConfig struct {
	MinDurationForLossValidity time.Duration `yaml:"min_duration_for_loss_validity,omitempty"`
	BaseDuration               time.Duration `yaml:"base_duration,omitempty"`
	BasePPS                    int           `yaml:"base_pps,omitempty"`
	LossPenaltyFactor          float64       `yaml:"loss_penalty_factor,omitempty"`
}

var (
	defaultWeightedLossConfig = WeightedLossConfig{
		MinDurationForLossValidity: 100 * time.Millisecond,
		BaseDuration:               500 * time.Millisecond,
		BasePPS:                    30,
		LossPenaltyFactor:          0.25,
	}
)

// -----------------------------------------------------------

type trafficStatsParams struct {
	Config WeightedLossConfig
	Logger logger.Logger
}

type trafficStats struct {
	params trafficStatsParams

	minSendTime  int64
	maxSendTime  int64
	sendDelta    int64
	recvDelta    int64
	ackedPackets int
	ackedBytes   int
	lostPackets  int
	lostBytes    int
}

func newTrafficStats(params trafficStatsParams) *trafficStats {
	return &trafficStats{
		params: params,
	}
}

func (ts *trafficStats) Merge(rhs *trafficStats) {
	if ts.minSendTime == 0 || rhs.minSendTime < ts.minSendTime {
		ts.minSendTime = rhs.minSendTime
	}
	if rhs.maxSendTime > ts.maxSendTime {
		ts.maxSendTime = rhs.maxSendTime
	}
	ts.sendDelta += rhs.sendDelta
	ts.recvDelta += rhs.recvDelta
	ts.ackedPackets += rhs.ackedPackets
	ts.ackedBytes += rhs.ackedBytes
	ts.lostPackets += rhs.lostPackets
	ts.lostBytes += rhs.lostBytes
}

func (ts *trafficStats) NumBytes() int {
	return ts.ackedBytes + ts.lostBytes
}

func (ts *trafficStats) Duration() int64 {
	return ts.maxSendTime - ts.minSendTime
}

func (ts *trafficStats) AcknowledgedBitrate() int64 {
	duration := ts.Duration()
	if duration == 0 {
		return 0
	}

	ackedBitrate := float64(ts.ackedBytes) * 8 * 1e6 / float64(ts.Duration())
	return int64(ackedBitrate * ts.CapturedTrafficRatio())
}

func (ts *trafficStats) CapturedTrafficRatio() float64 {
	if ts.recvDelta == 0 {
		return 0.0
	}

	// apply a penalty for lost packets,
	// the rationale being packet dropping is a strategy to relieve congestion
	// and if they were not dropped, they would have increased queuing delay,
	// as it is not possible to know the reason for the losses,
	// apply a small penalty to receive delta aggregate to simulate those packets
	// building up queuing delay.
	return min(1.0, float64(ts.sendDelta)/float64(ts.recvDelta+ts.lossPenalty()))
}

func (ts *trafficStats) WeightedLoss() float64 {
	durationMicro := ts.Duration()
	if time.Duration(durationMicro*1000) < ts.params.Config.MinDurationForLossValidity {
		return 0.0
	}

	totalPackets := float64(ts.lostPackets + ts.ackedPackets)
	pps := totalPackets * 1e6 / float64(durationMicro)

	// longer duration, i. e. more time resolution, lower pps is acceptable as the measurement is more stable
	deltaDuration := time.Duration(durationMicro*1000) - ts.params.Config.BaseDuration
	if deltaDuration < 0 {
		deltaDuration = 0
	}
	threshold := math.Exp(-deltaDuration.Seconds()) * float64(ts.params.Config.BasePPS)
	if pps < threshold {
		return 0.0
	}

	lossRatio := float64(0.0)
	if totalPackets != 0 {
		lossRatio = float64(ts.lostPackets) / totalPackets
	}

	// Log10 is used to give higher weight for the same loss ratio at higher packet rates,
	// for e.g.
	//    - 10% loss at 20 pps = 0.1 * log10(20) = 0.130
	//    - 10% loss at 100 pps = 0.1 * log10(100) = 0.2
	//    - 10% loss at 1000 pps = 0.1 * log10(1000) = 0.3
	return lossRatio * math.Log10(pps)
}

func (ts *trafficStats) lossPenalty() int64 {
	return int64(float64(ts.recvDelta) * ts.WeightedLoss() * ts.params.Config.LossPenaltyFactor)
}

func (ts *trafficStats) MarshalLogObject(e zapcore.ObjectEncoder) error {
	if ts == nil {
		return nil
	}

	e.AddInt64("minSendTime", ts.minSendTime)
	e.AddInt64("maxSendTime", ts.maxSendTime)
	duration := time.Duration(ts.Duration() * 1000)
	e.AddDuration("duration", duration)

	e.AddInt("ackedPackets", ts.ackedPackets)
	e.AddInt("ackedBytes", ts.ackedBytes)
	e.AddInt("lostPackets", ts.lostPackets)
	e.AddInt("lostBytes", ts.lostBytes)

	bitrate := float64(0)
	if duration != 0 {
		bitrate = float64(ts.ackedBytes*8) / duration.Seconds()
		e.AddFloat64("bitrate", bitrate)
	}

	e.AddInt64("sendDelta", ts.sendDelta)
	e.AddInt64("recvDelta", ts.recvDelta)
	e.AddInt64("groupDelay", ts.recvDelta-ts.sendDelta)

	totalPackets := ts.lostPackets + ts.ackedPackets
	if duration != 0 {
		e.AddFloat64("pps", float64(totalPackets)/duration.Seconds())
	}
	if (totalPackets) != 0 {
		e.AddFloat64("rawLoss", float64(ts.lostPackets)/float64(totalPackets))
	}
	e.AddFloat64("weightedLoss", ts.WeightedLoss())
	e.AddInt64("lossPenalty", ts.lossPenalty())

	capturedTrafficRatio := ts.CapturedTrafficRatio()
	e.AddFloat64("capturedTrafficRatio", capturedTrafficRatio)
	e.AddFloat64("estimatedAvailableChannelCapacity", bitrate*capturedTrafficRatio)
	return nil
}