diff --git a/config-sample.yaml b/config-sample.yaml index 5aca3bea7..059de7beb 100644 --- a/config-sample.yaml +++ b/config-sample.yaml @@ -117,6 +117,16 @@ rtc: # # in the unlikely event of highly congested networks, SFU may choose to pause some tracks # # in order to allow others to stream smoothly. You can disable this behavior here # allow_pause: true + # # use send-side bandwidth estimation instead of relying on receiver reports, default false + # use_send_side_bwe: false + # # how egress is paced when use_send_side_bwe is enabled. one of: + # # no-queue - hand packets to a send worker that drains them without rate + # # limiting, so the sender never blocks on a write (default) + # # pass-through - send each packet inline on the calling goroutine + # # leaky-bucket - accepted, but not currently wired to the bandwidth estimate; it + # # logs a warning and falls back to no-queue. see livekit/livekit#4724 + # # an unrecognized value is rejected at startup. + # send_side_bwe_pacer: no-queue # # allows automatic connection fallback to TCP and TURN/TLS (if configured) when UDP has been unstable, default true # allow_tcp_fallback: true # # signaling RTT (in milliseconds) below which ICE/TCP is attempted on a UDP failure; at or above it, diff --git a/pkg/config/config.go b/pkg/config/config.go index 5dc2c91fd..2f5e5f8fe 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -220,6 +220,22 @@ type CongestionControlConfig struct { SendSideBWE sendsidebwe.SendSideBWEConfig `yaml:"send_side_bwe,omitempty"` } +// Validate reports configuration that would otherwise be ignored at runtime. +func (c CongestionControlConfig) Validate() error { + switch pacer.PacerBehavior(c.SendSideBWEPacer) { + case "", pacer.PacerBehaviorPassThrough, pacer.PacerBehaviorNoQueue, pacer.PacerBehaviorLeakybucket: + return nil + default: + return fmt.Errorf( + "invalid send_side_bwe_pacer %q, must be one of %q, %q, %q", + c.SendSideBWEPacer, + pacer.PacerBehaviorPassThrough, + pacer.PacerBehaviorNoQueue, + pacer.PacerBehaviorLeakybucket, + ) + } +} + type PlayoutDelayConfig struct { Enabled bool `yaml:"enabled,omitempty"` Min int `yaml:"min,omitempty"` @@ -671,6 +687,10 @@ func NewConfig(confString string, strictMode bool, c *cli.Command, baseFlags []c return nil, fmt.Errorf("could not validate RTC config: %v", err) } + if err := conf.RTC.CongestionControl.Validate(); err != nil { + return nil, fmt.Errorf("could not validate congestion control config: %v", err) + } + conf.NormalizeTURNTTLs() // expand env vars in filenames diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index 0b6439945..d4117cea5 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -23,6 +23,7 @@ import ( "github.com/urfave/cli/v3" "github.com/livekit/livekit-server/pkg/config/configtest" + "github.com/livekit/livekit-server/pkg/sfu/pacer" ) func TestConfig_UnmarshalKeys(t *testing.T) { @@ -182,6 +183,40 @@ func TestNewConfigNormalizesTURNTTL(t *testing.T) { require.Equal(t, DefaultTURNTTLSeconds, conf.TURN.TTLSeconds) } +func TestNewConfigValidatesSendSideBWEPacer(t *testing.T) { + cases := []struct { + name string + pacer string + wantErr bool + }{ + {name: "unset keeps the default", pacer: "", wantErr: false}, + {name: "pass-through", pacer: string(pacer.PacerBehaviorPassThrough), wantErr: false}, + {name: "no-queue", pacer: string(pacer.PacerBehaviorNoQueue), wantErr: false}, + {name: "leaky-bucket", pacer: string(pacer.PacerBehaviorLeakybucket), wantErr: false}, + {name: "typo is rejected", pacer: "leakybucket", wantErr: true}, + {name: "unknown value is rejected", pacer: "not-a-pacer", wantErr: true}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + content := "rtc:\n congestion_control:\n send_side_bwe_pacer: " + tc.pacer + conf, err := NewConfig(content, true, nil, nil) + if tc.wantErr { + require.Error(t, err) + require.Contains(t, err.Error(), tc.pacer) + return + } + + require.NoError(t, err) + if tc.pacer == "" { + require.Equal(t, string(pacer.PacerBehaviorNoQueue), conf.RTC.CongestionControl.SendSideBWEPacer) + return + } + require.Equal(t, tc.pacer, conf.RTC.CongestionControl.SendSideBWEPacer) + }) + } +} + func writeSecretFile(t *testing.T, content string) string { t.Helper() path := filepath.Join(t.TempDir(), "secret") diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index c00f629dd..4baf2e6e0 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -582,12 +582,25 @@ func NewPCTransport(params TransportParams) (*PCTransport, error) { Config: params.CongestionControlConfig.SendSideBWE, Logger: params.Logger, }) - switch pacer.PacerBehavior(params.CongestionControlConfig.SendSideBWEPacer) { + switch behavior := pacer.PacerBehavior(params.CongestionControlConfig.SendSideBWEPacer); behavior { case pacer.PacerBehaviorPassThrough: t.pacer = pacer.NewPassThrough(params.Logger, t.bwe) case pacer.PacerBehaviorNoQueue: t.pacer = pacer.NewNoQueue(params.Logger, t.bwe) default: + // leaky-bucket lands here: it is a recognized value, but LeakyBucket paces + // to a bitrate set through Pacer.SetBitrate and nothing drives that from + // the bandwidth estimate, so it would pace at a fixed rate and ignore + // congestion. Unknown values are rejected by CongestionControlConfig. + // Validate, so they only reach here from callers building TransportParams + // directly. + if behavior != "" { + params.Logger.Warnw( + "send side BWE pacer unavailable, falling back", nil, + "pacerBehavior", behavior, + "fallback", pacer.PacerBehaviorNoQueue, + ) + } t.pacer = pacer.NewNoQueue(params.Logger, t.bwe) } } else {