o_sfu_core/engine/media_transport/
build.rs

1//! media transport construction and startup validation
2
3use std::{io, net::TcpListener, sync::Arc};
4
5use thiserror::Error;
6use tracing::info;
7
8use super::{
9    MediaTransport, SourcePolicySignal,
10    config::{MediaTransportConfig, MediaTransportDeps},
11    rtc::{RtcWorker, RtpProfile, TcpAcceptor},
12};
13use crate::{
14    MediaWorkerId, RtcUdpIoBackend,
15    engine::media_transport::rtc::{UfragWorkerMap, WorkerAssignment},
16};
17
18/// Per-worker [`TransportMediaId`](crate::engine::media_transport::TransportMediaId)
19/// allocation stride.
20///
21/// IDs have no worker namespace. Each `RtcWorker` may allocate at most this
22/// many IDs during its lifetime within one `MediaTransport`.
23const MEDIA_ID_STRIDE: u64 = 1_000_000_000;
24
25impl MediaTransport {
26    /// builds the runtime media transport from owner configuration and process services
27    ///
28    /// validation completes before worker startup and every worker has a bound
29    /// socket when this function returns
30    ///
31    /// # Errors
32    ///
33    /// returns [`MediaTransportBuildError`] when worker topology or the
34    /// code-controlled RTP profile is invalid, the selected UDP backend is
35    /// unavailable, the media-quality interval exceeds 24 hours or a worker
36    /// cannot start
37    pub fn build(
38        mut config: MediaTransportConfig,
39        deps: MediaTransportDeps,
40    ) -> Result<Self, MediaTransportBuildError> {
41        if config.rtc_udp_io_backend != RtcUdpIoBackend::Tokio && !cfg!(target_os = "linux") {
42            return Err(MediaTransportBuildError::UnsupportedUdpIoBackend {
43                backend: config.rtc_udp_io_backend,
44            });
45        }
46        if config.worker_count == 0 {
47            return Err(MediaTransportBuildError::InvalidWorkerCount);
48        }
49        // str0m adds the stats interval to Instant, so validate before starting workers.
50        if config
51            .media_quality_interval
52            .is_some_and(|interval| interval > MediaTransportConfig::MAX_MEDIA_QUALITY_INTERVAL)
53        {
54            return Err(MediaTransportBuildError::InvalidMediaQualityInterval);
55        }
56        config.media_quality_interval = config
57            .media_quality_interval
58            .filter(|value| !value.is_zero());
59        let worker_ranges = config
60            .rtc_port_range
61            .split_for_workers(config.worker_count)
62            .ok_or(MediaTransportBuildError::InvalidPortSplit {
63                worker_count: config.worker_count,
64                port_count: config.rtc_port_range.port_count(),
65            })?;
66        let profile = Arc::new(
67            RtpProfile::compile(config.codec_flags, config.codec_preferences)
68                .map_err(|_error| MediaTransportBuildError::InvalidRtpProfile)?,
69        );
70        let source_policy_signal = SourcePolicySignal::default();
71        let ufrag_worker_map = UfragWorkerMap::default();
72        let workers: Arc<[_]> = (0_u16..u16::MAX)
73            .zip(worker_ranges)
74            .map(|(worker_index, range)| {
75                // `start` returns only after socket binding, so the completed
76                // transport cannot publish a worker whose first command races I/O setup.
77                RtcWorker::start(
78                    &config,
79                    Arc::clone(&profile),
80                    WorkerAssignment {
81                        rtc_port_range: range,
82                        media_id_base: u64::from(worker_index) * MEDIA_ID_STRIDE,
83                        media_worker_id: MediaWorkerId::from_raw(usize::from(worker_index)),
84                    },
85                    &deps,
86                    source_policy_signal.clone(),
87                    ufrag_worker_map.clone(),
88                )
89                .map_err(|_error| MediaTransportBuildError::WorkerStartup {
90                    worker_index: usize::from(worker_index),
91                })
92            })
93            .collect::<Result<_, _>>()?;
94        let tcp_acceptor = config
95            .rtc_tcp_config
96            .map(|c| {
97                let addr = c.bind_addr;
98                let listener = TcpListener::bind(addr)?;
99                listener.set_nonblocking(true)?;
100                let acceptor: Result<Arc<TcpAcceptor>, io::Error> = TcpAcceptor::spawn(
101                    listener,
102                    c.announced_addr,
103                    ufrag_worker_map,
104                    Arc::clone(&workers),
105                    c.max_conn,
106                )
107                .map(Arc::new);
108                info!(
109                    "booted rtc TCP listener bind_addr={} candidate_addr={}",
110                    addr, c.announced_addr,
111                );
112                acceptor
113            })
114            .transpose()
115            .map_err(|error| MediaTransportBuildError::TcpAcceptorStartup { kind: error.kind() })?;
116        Ok(Self {
117            workers,
118            profile,
119            metrics: deps.metrics,
120            #[cfg(test)]
121            media_control_batches: Arc::default(),
122            #[cfg(any(test, feature = "testing-transport"))]
123            source_diagnostics_requests: Arc::default(),
124            source_policy_signal,
125            _tcp_acceptor: tcp_acceptor,
126        })
127    }
128}
129
130/// invalid construction inputs for the media transport
131#[derive(Debug, Error, Clone, Copy, PartialEq, Eq)]
132pub enum MediaTransportBuildError {
133    /// a transport cannot be built without at least one RTC worker
134    #[error("media transport worker count must be at least one")]
135    InvalidWorkerCount,
136    /// the configured UDP range cannot provide one port to every worker
137    #[error(
138        "media transport cannot split {port_count} UDP ports across {worker_count} media workers"
139    )]
140    InvalidPortSplit {
141        worker_count: usize,
142        port_count: u16,
143    },
144    /// the sampling interval exceeds the operational deadline bound
145    #[error("media-quality interval must not exceed 24 hours")]
146    InvalidMediaQualityInterval,
147    /// the selected UDP I/O backend is not available on this build target
148    #[error("rtc UDP I/O backend `{backend}` is not supported on this target")]
149    UnsupportedUdpIoBackend { backend: RtcUdpIoBackend },
150    /// the code-controlled RTC profile cannot be projected for router policy
151    #[error("media transport RTP profile is invalid")]
152    InvalidRtpProfile,
153    /// one worker could not create its runtime or bind its assigned UDP range
154    #[error("media transport worker {worker_index} failed to start")]
155    WorkerStartup { worker_index: usize },
156    /// the TCP listener could not be bound or the acceptor thread could not start
157    #[error("media transport failed to start the TCP acceptor: {kind}")]
158    TcpAcceptorStartup { kind: io::ErrorKind },
159}
o_sfu_core/engine/media_transport/build.rs

o_sfu_core/engine/media_transport/
build.rs
