1use std::{
2 net::{IpAddr, SocketAddr},
3 num::{NonZeroU64, NonZeroUsize},
4 thread,
5 time::Duration,
6};
7
8use anyhow::{Result, anyhow, ensure};
9use o_sfu_core::prelude::{
10 Bitrate, RoomMediaLimits, RoomWorkerPolicy, RtcPortRange, RtcUdpIoBackend,
11 VideoAdaptationTuning, VideoAdaptationTuningError, VideoBitrateLimits,
12};
13use tokio::sync::Semaphore;
14
15use super::{
16 RtcTcpConfig, TransportConfig,
17 env::{Env, EnvParse, EnvValue, positive},
18};
19
20impl EnvParse for RtcUdpIoBackend {
21 fn parse(value: EnvValue) -> Result<Self> {
22 let key = value.key;
23 match value.raw.as_str() {
24 "tokio" => Ok(Self::Tokio),
25 "tokio_batch" => Ok(Self::TokioBatch),
26 "io_uring" => Ok(Self::IoUring),
27 other => Err(anyhow!(
28 "{key} must be one of tokio, tokio_batch or io_uring, got {other}"
29 )),
30 }
31 }
32}
33
34impl TransportConfig {
35 pub(super) fn from_env(env: &Env<'_>) -> Result<Self> {
36 let announced_ip = env
37 .var::<IpAddr>("ANNOUNCED_IP")
38 .alias("PUBLIC_IP")
39 .check(advertised_ip)
40 .required()?;
41 let rtc_min_port = env.var("RTC_MIN_PORT").default(40_000)?;
42 let max_bitrate_in = env
43 .var("MAX_BITRATE_IN")
44 .check(positive)
45 .default(Bitrate::from_mbps(8))?;
46 let max_bitrate_out = env
47 .var("MAX_BITRATE_OUT")
48 .check(positive)
49 .default(Bitrate::from_mbps(10))?;
50 let max_video_bitrate = env
51 .var("MAX_VIDEO_BITRATE")
52 .check(positive)
53 .default(VideoBitrateLimits::DEFAULT_MAX_VIDEO_BITRATE)?;
54 let rtc_max_port = env
55 .var("RTC_MAX_PORT")
56 .check(|key, value| {
57 ensure!(
58 value >= rtc_min_port,
59 "{key} must be greater than or equal to RTC_MIN_PORT"
60 );
61 Ok(value)
62 })
63 .default(49_999)?;
64 let rtc_port_range = RtcPortRange::try_new(rtc_min_port, rtc_max_port)?;
65 let rtc_udp_io_backend = env
66 .var("RTC_UDP_IO_BACKEND")
67 .check(supported_udp_io_backend)
68 .default(RtcUdpIoBackend::Tokio)?;
69 let rtc_media_worker_count = env
70 .var("RTC_MEDIA_WORKER_COUNT")
71 .check(positive)
72 .check(|key, value| {
73 ensure!(
74 value <= usize::from(rtc_port_range.port_count()),
75 "{key} must be less than or equal to the available RTC port count"
76 );
77 Ok(value)
78 })
79 .default(default_rtc_media_worker_count())?;
80 let room_max_local_routers = env
81 .var::<NonZeroUsize>("ROOM_MAX_LOCAL_ROUTERS")
82 .check(|key, value| {
83 ensure!(
84 value.get() <= rtc_media_worker_count,
85 "{key} must be less than or equal to RTC_MEDIA_WORKER_COUNT"
86 );
87 Ok(value)
88 })
89 .default(NonZeroUsize::MIN)?;
90 let default_packet_loop_delay_threshold =
91 NonZeroU64::try_from(RoomWorkerPolicy::DEFAULT_PACKET_LOOP_DELAY_THRESHOLD_MS)?;
92 let packet_loop_delay_threshold = env
93 .var("ROOM_SPILLOVER_PACKET_LOOP_DELAY_MS")
94 .default(default_packet_loop_delay_threshold)?;
95 let room_media_limits = room_media_limits_from_env(env)?;
96 let video_adaptation_tuning = video_adaptation_tuning_from_env(env)?;
97 let room_worker_policy =
98 RoomWorkerPolicy::new(room_max_local_routers, packet_loop_delay_threshold);
99 Ok(Self {
100 announced_ip,
101 max_bitrate_in,
102 max_bitrate_out,
103 video_bitrate_limits: VideoBitrateLimits::new(max_video_bitrate),
104 rtc_port_range,
105 rtc_udp_io_backend,
106 rtc_media_worker_count,
107 rtc_tcp_config: rtc_tcp_config_from_env(env, announced_ip, rtc_udp_io_backend)?,
108 room_worker_policy,
109 room_media_limits,
110 video_adaptation_tuning,
111 })
112 }
113}
114
115fn advertised_ip(key: &'static str, value: IpAddr) -> Result<IpAddr> {
116 ensure!(
117 !value.is_unspecified(),
118 "{key} must be a concrete advertised address"
119 );
120 ensure!(!value.is_multicast(), "{key} cannot be a multicast address");
121 Ok(value)
122}
123
124fn supported_udp_io_backend(key: &'static str, value: RtcUdpIoBackend) -> Result<RtcUdpIoBackend> {
125 ensure!(
126 value == RtcUdpIoBackend::Tokio || cfg!(target_os = "linux"),
127 "{key}={value} is only supported on Linux"
128 );
129 Ok(value)
130}
131
132fn room_media_limits_from_env(env: &Env<'_>) -> Result<RoomMediaLimits> {
133 let active_audio_speakers = env
134 .var("ROOM_MAX_ACTIVE_AUDIO_SPEAKERS")
135 .check(positive)
136 .default(RoomMediaLimits::DEFAULT_MAX_ACTIVE_AUDIO_SPEAKERS)?;
137 let video_downloads_per_receiver = env
138 .var("ROOM_MAX_VIDEO_DOWNLOADS_PER_RECEIVER")
139 .check(positive)
140 .default(RoomMediaLimits::DEFAULT_MAX_VIDEO_DOWNLOADS_PER_RECEIVER)?;
141 Ok(RoomMediaLimits::try_new(
142 active_audio_speakers,
143 video_downloads_per_receiver,
144 )?)
145}
146
147fn requires_tcp_bind_addr<T>(
148 bind_addr: Option<SocketAddr>,
149) -> impl Fn(&'static str, T) -> Result<T> {
150 move |key, value| {
151 ensure!(bind_addr.is_some(), "{key} requires TCP_BIND_ADDRESS");
152 Ok(value)
153 }
154}
155
156fn rtc_tcp_config_from_env(
157 env: &Env<'_>,
158 announced_ip: IpAddr,
159 io_backend: RtcUdpIoBackend,
160) -> Result<Option<RtcTcpConfig>> {
161 let rtc_tcp_bind_addr = env
162 .var::<SocketAddr>("TCP_BIND_ADDRESS")
163 .check(|key, value| {
164 ensure!(
165 io_backend != RtcUdpIoBackend::IoUring,
166 "{key} is not supported with RTC_UDP_IO_BACKEND=io_uring"
167 );
168 ensure!(value.port() > 0, "{key} must use a non-zero port");
169 Ok(value)
170 })
171 .optional()?;
172 let rtc_tcp_announced_addr = env
173 .var::<SocketAddr>("TCP_ANNOUNCED_ADDRESS")
174 .check(requires_tcp_bind_addr(rtc_tcp_bind_addr))
175 .check(|key, value| {
176 advertised_ip(key, value.ip())?;
177 Ok(value)
178 })
179 .optional()?;
180 let max_conn = env
181 .var::<usize>("TCP_MAX_CONNECTIONS")
182 .check(requires_tcp_bind_addr(rtc_tcp_bind_addr))
183 .check(|key, value| {
184 ensure!(
185 (1..=Semaphore::MAX_PERMITS).contains(&value),
186 "{key} must be between 1 and {}",
187 Semaphore::MAX_PERMITS
188 );
189 Ok(value)
190 })
191 .optional()?
192 .unwrap_or(1024);
193 Ok(rtc_tcp_bind_addr.map(|addr| {
194 let actual_announced_addr =
195 rtc_tcp_announced_addr.unwrap_or(SocketAddr::from((announced_ip, addr.port())));
196 RtcTcpConfig {
197 announced_addr: actual_announced_addr,
198 bind_addr: addr,
199 max_conn,
200 }
201 }))
202}
203
204fn video_adaptation_tuning_from_env(env: &Env<'_>) -> Result<VideoAdaptationTuning> {
205 let multiparty_scalable_video_threshold = env
206 .var("ROOM_MULTIPARTY_SCALABLE_VIDEO_THRESHOLD")
207 .check(positive)
208 .default(VideoAdaptationTuning::DEFAULT_MULTIPARTY_SCALABLE_VIDEO_THRESHOLD)?;
209 let thumbnail_budget_divisor = env
210 .var("ROOM_THUMBNAIL_BUDGET_DIVISOR")
211 .check(positive)
212 .default(VideoAdaptationTuning::DEFAULT_THUMBNAIL_BUDGET_DIVISOR)?;
213 let soft_pause_dwell = env
214 .var("ROOM_SOFT_PAUSE_DWELL_MS")
215 .check(positive)
216 .optional()?
217 .map_or(
218 VideoAdaptationTuning::DEFAULT_SOFT_PAUSE_DWELL,
219 Duration::from_millis,
220 );
221 let upgrade_dwell = env
222 .var("ROOM_UPGRADE_DWELL_MS")
223 .check(positive)
224 .optional()?
225 .map_or(
226 VideoAdaptationTuning::DEFAULT_UPGRADE_DWELL,
227 Duration::from_millis,
228 );
229 let receiver_budget_headroom_percent = env
230 .var("ROOM_RECEIVER_BUDGET_HEADROOM_PERCENT")
231 .default(VideoAdaptationTuning::DEFAULT_RECEIVER_BUDGET_HEADROOM_PERCENT)?;
232 let audio_reserve_per_speaker = env
233 .var("ROOM_AUDIO_RESERVE_PER_SPEAKER_BPS")
234 .default(VideoAdaptationTuning::DEFAULT_AUDIO_RESERVE_PER_SPEAKER)?;
235 VideoAdaptationTuning::try_new(
236 multiparty_scalable_video_threshold,
237 thumbnail_budget_divisor,
238 soft_pause_dwell,
239 upgrade_dwell,
240 receiver_budget_headroom_percent,
241 audio_reserve_per_speaker,
242 )
243 .map_err(video_adaptation_tuning_error)
244}
245
246pub fn default_rtc_media_worker_count() -> usize {
247 thread::available_parallelism().map_or(1, NonZeroUsize::get)
248}
249
250fn video_adaptation_tuning_error(error: VideoAdaptationTuningError) -> anyhow::Error {
251 match error {
252 VideoAdaptationTuningError::MultipartyScalableVideoThresholdZero => {
253 anyhow!("ROOM_MULTIPARTY_SCALABLE_VIDEO_THRESHOLD must be greater than zero")
254 }
255 VideoAdaptationTuningError::ThumbnailBudgetDivisorZero => {
256 anyhow!("ROOM_THUMBNAIL_BUDGET_DIVISOR must be greater than zero")
257 }
258 VideoAdaptationTuningError::SoftPauseDwellZero => {
259 anyhow!("ROOM_SOFT_PAUSE_DWELL_MS must be greater than zero")
260 }
261 VideoAdaptationTuningError::UpgradeDwellZero => {
262 anyhow!("ROOM_UPGRADE_DWELL_MS must be greater than zero")
263 }
264 VideoAdaptationTuningError::SoftPauseDwellTooLong => {
265 anyhow!(
266 "ROOM_SOFT_PAUSE_DWELL_MS must not exceed {}",
267 VideoAdaptationTuning::MAX_DWELL.as_millis()
268 )
269 }
270 VideoAdaptationTuningError::UpgradeDwellTooLong => {
271 anyhow!(
272 "ROOM_UPGRADE_DWELL_MS must not exceed {}",
273 VideoAdaptationTuning::MAX_DWELL.as_millis()
274 )
275 }
276 VideoAdaptationTuningError::ReceiverBudgetHeadroomPercentTooHigh => {
277 anyhow!("ROOM_RECEIVER_BUDGET_HEADROOM_PERCENT must not exceed 100")
278 }
279 }
280}
281
282#[cfg(test)]
283#[path = "TESTS/transport.rs"]
284mod tests;