o_sfu_core/engine/room/
placement.rs

1//! Router and packet-worker placement for room admission.
2//!
3//! [`RoomRuntimeContext`] seeds a room with assigned placements or an
4//! unassigned primary router. Admission selects a packet worker from current
5//! delay samples and may add a router within [`RoomWorkerPolicy`].
6
7use o_sfu_router::RouterId;
8pub use o_sfu_router::topology::{
9    PlacementSnapshot, RouterPlacement, RouterPlacements, RouterPlacementsError,
10};
11#[cfg(any(test, feature = "testing-transport"))]
12use {std::sync::Arc, tokio::sync::Barrier};
13
14use super::{
15    Room, RoomJoinError,
16    factory::RoomFactory,
17    membership::JoinUserRequest,
18    state::{JoinCommit, UserJoinedFanout},
19};
20use crate::{
21    RoomWorkerPolicy,
22    engine::{
23        MediaWorkerId, RoomInstanceId,
24        media_transport::{MediaTransport, WorkerPlacementState},
25    },
26};
27
28/// Initial router placement context for one room instance.
29///
30/// [`Self::new_unassigned`] defers packet-worker selection until admission.
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub struct RoomRuntimeContext {
33    instance: RoomInstanceId,
34    primary_router: RouterId,
35    initial_router_placements: Option<RouterPlacements>,
36}
37
38impl RoomRuntimeContext {
39    #[must_use]
40    pub fn new(
41        instance: RoomInstanceId,
42        primary: RouterPlacement,
43        spillover: Vec<RouterPlacement>,
44    ) -> Self {
45        Self {
46            instance,
47            primary_router: primary.router,
48            initial_router_placements: Some(RouterPlacements::new(primary, spillover)),
49        }
50    }
51
52    #[must_use]
53    pub const fn new_unassigned(instance: RoomInstanceId, primary_router: RouterId) -> Self {
54        Self {
55            instance,
56            primary_router,
57            initial_router_placements: None,
58        }
59    }
60
61    /// # Errors
62    ///
63    /// returns [`RouterPlacementsError::Empty`] when `placements` is empty
64    pub fn try_from_placements(
65        instance: RoomInstanceId,
66        placements: Vec<RouterPlacement>,
67    ) -> Result<Self, RouterPlacementsError> {
68        let routers = RouterPlacements::try_from_vec(placements)?;
69        Ok(Self {
70            instance,
71            primary_router: routers.primary().router,
72            initial_router_placements: Some(routers),
73        })
74    }
75
76    #[must_use]
77    pub const fn instance(&self) -> RoomInstanceId {
78        self.instance
79    }
80
81    #[must_use]
82    pub const fn primary_router(&self) -> RouterId {
83        self.primary_router
84    }
85
86    #[must_use]
87    pub fn initial_router_placements(&self) -> Option<&RouterPlacements> {
88        self.initial_router_placements.as_ref()
89    }
90}
91
92#[cfg(any(test, feature = "testing-transport"))]
93impl Room {
94    pub(super) async fn placement_usage_snapshot(&self) -> PlacementSnapshot {
95        self.state.read().await.placement_usage_snapshot()
96    }
97}
98
99enum WorkerPlacementSource<'a> {
100    Transport(&'a MediaTransport),
101    #[cfg(any(test, feature = "testing-transport"))]
102    Fixed(Vec<Option<u64>>),
103}
104
105impl WorkerPlacementSource<'_> {
106    fn snapshot(self) -> Vec<WorkerPlacementState> {
107        match self {
108            Self::Transport(transport) => transport.worker_placement_states(),
109            #[cfg(any(test, feature = "testing-transport"))]
110            Self::Fixed(delays_ms) => delays_ms
111                .into_iter()
112                .map(WorkerPlacementState::Running)
113                .collect(),
114        }
115    }
116}
117
118pub(super) struct JoinAdmissionTurn<'a, A = fn() -> RouterId> {
119    request: JoinUserRequest,
120    worker_states: WorkerPlacementSource<'a>,
121    allocate_spillover_router: A,
122    #[cfg(any(test, feature = "testing-transport"))]
123    gate: Option<Arc<JoinPlacementTestGate>>,
124}
125
126impl JoinAdmissionTurn<'_> {
127    pub(super) fn from_factory<'a>(
128        request: JoinUserRequest,
129        media_transport: &'a MediaTransport,
130        factory: &'a RoomFactory,
131    ) -> JoinAdmissionTurn<'a, impl FnOnce() -> RouterId + 'a> {
132        JoinAdmissionTurn {
133            request,
134            worker_states: WorkerPlacementSource::Transport(media_transport),
135            allocate_spillover_router: move || factory.allocate_spillover_router(),
136            #[cfg(any(test, feature = "testing-transport"))]
137            gate: None,
138        }
139    }
140
141    #[cfg(any(test, feature = "testing-transport"))]
142    pub(super) fn for_test(
143        request: JoinUserRequest,
144        delays_ms: Vec<Option<u64>>,
145        spillover_router_id: RouterId,
146    ) -> JoinAdmissionTurn<'static, impl FnOnce() -> RouterId> {
147        JoinAdmissionTurn {
148            request,
149            worker_states: WorkerPlacementSource::Fixed(delays_ms),
150            allocate_spillover_router: move || spillover_router_id,
151            gate: None,
152        }
153    }
154}
155
156impl<A: FnOnce() -> RouterId> JoinAdmissionTurn<'_, A> {
157    #[cfg(any(test, feature = "testing-transport"))]
158    pub(super) fn with_gate(mut self, gate: Option<Arc<JoinPlacementTestGate>>) -> Self {
159        self.gate = gate;
160        self
161    }
162
163    pub(super) async fn commit(
164        self,
165        room: &Room,
166        joined_fanout: UserJoinedFanout,
167    ) -> Result<JoinCommit, RoomJoinError> {
168        #[cfg(any(test, feature = "testing-transport"))]
169        if let Some(gate) = &self.gate {
170            gate.wait_before_commit().await;
171        }
172        let mut state = room.state.write().await;
173        // Sample worker availability and delay at the serialized commit turn.
174        // The state guard includes placements committed by earlier joins.
175        let worker_states = self.worker_states.snapshot();
176        let worker_count = worker_states.len().max(1);
177        let start_worker = room_worker_start(room.instance_id(), worker_count);
178        let placement = choose_placement(
179            &state.placement_usage_snapshot(),
180            room.room_worker_policy(),
181            &worker_states,
182            start_worker,
183            self.allocate_spillover_router,
184        )?;
185        state.apply_join_on_placement(
186            &self.request.user_id,
187            self.request.sender,
188            joined_fanout,
189            placement,
190        )
191    }
192}
193
194fn choose_placement(
195    room: &PlacementSnapshot,
196    policy: RoomWorkerPolicy,
197    workers: &[WorkerPlacementState],
198    start_worker: usize,
199    allocate_spillover_router: impl FnOnce() -> RouterId,
200) -> Result<RouterPlacement, RoomJoinError> {
201    if !workers
202        .iter()
203        .any(|worker| matches!(worker, WorkerPlacementState::Running(_)))
204    {
205        return Err(RoomJoinError::NoUsableWorker);
206    }
207    let worker_count = workers.len();
208    let threshold_ms = policy.packet_loop_delay_threshold_ms();
209    let assigned = room.assigned_placements();
210    if assigned.is_empty() {
211        return Ok(RouterPlacement {
212            router: room.primary(),
213            media_worker: choose_primary_worker(
214                workers,
215                threshold_ms,
216                start_worker % worker_count,
217            )?,
218        });
219    }
220    if let Some(placement) = assigned
221        .iter()
222        .filter(|placement| worker_is_healthy(workers, placement.media_worker, threshold_ms))
223        .min_by_key(|placement| worker_delay(workers, placement.media_worker))
224    {
225        return Ok(*placement);
226    }
227    let placement_cap = policy.max_local_routers().min(worker_count);
228    let usable_placements = assigned
229        .iter()
230        .filter(|placement| worker_is_usable(workers, placement.media_worker))
231        .count();
232    if usable_placements < placement_cap
233        && let Some(media_worker) = cyclic_workers(start_worker, worker_count).find(|worker| {
234            worker_is_healthy(workers, *worker, threshold_ms)
235                && assigned
236                    .iter()
237                    .all(|placement| placement.media_worker != *worker)
238        })
239    {
240        return Ok(RouterPlacement {
241            router: allocate_spillover_router(),
242            media_worker,
243        });
244    }
245    if let Some(placement) = assigned
246        .iter()
247        .copied()
248        .filter(|placement| worker_is_usable(workers, placement.media_worker))
249        .min_by_key(|placement| worker_delay(workers, placement.media_worker))
250    {
251        return Ok(placement);
252    }
253    if usable_placements < placement_cap
254        && let Some(media_worker) = cyclic_workers(start_worker, worker_count).find(|worker| {
255            worker_is_usable(workers, *worker)
256                && assigned
257                    .iter()
258                    .all(|placement| placement.media_worker != *worker)
259        })
260    {
261        return Ok(RouterPlacement {
262            router: allocate_spillover_router(),
263            media_worker,
264        });
265    }
266    Err(RoomJoinError::RouterState)
267}
268
269fn choose_primary_worker(
270    workers: &[WorkerPlacementState],
271    threshold_ms: u64,
272    start_worker: usize,
273) -> Result<MediaWorkerId, RoomJoinError> {
274    let worker_count = workers.len().max(1);
275    cyclic_workers(start_worker, worker_count)
276        .find(|worker| worker_is_healthy(workers, *worker, threshold_ms))
277        .or_else(|| {
278            cyclic_workers(start_worker, worker_count)
279                .filter(|worker| worker_is_usable(workers, *worker))
280                .min_by_key(|worker| worker_delay(workers, *worker))
281        })
282        .ok_or(RoomJoinError::NoUsableWorker)
283}
284
285fn cyclic_workers(start_worker: usize, worker_count: usize) -> impl Iterator<Item = MediaWorkerId> {
286    (0..worker_count).map(move |offset| {
287        MediaWorkerId::from_raw(start_worker.wrapping_add(offset) % worker_count)
288    })
289}
290
291fn worker_is_healthy(
292    workers: &[WorkerPlacementState],
293    worker: MediaWorkerId,
294    threshold_ms: u64,
295) -> bool {
296    matches!(
297        workers.get(worker.as_usize()),
298        Some(WorkerPlacementState::Running(Some(delay_ms))) if *delay_ms < threshold_ms
299    )
300}
301
302fn worker_is_usable(workers: &[WorkerPlacementState], worker: MediaWorkerId) -> bool {
303    matches!(
304        workers.get(worker.as_usize()),
305        Some(WorkerPlacementState::Running(_))
306    )
307}
308
309fn worker_delay(workers: &[WorkerPlacementState], worker: MediaWorkerId) -> u64 {
310    match workers.get(worker.as_usize()) {
311        Some(WorkerPlacementState::Running(Some(delay_ms))) => *delay_ms,
312        _ => u64::MAX,
313    }
314}
315
316fn room_worker_start(room_instance_id: RoomInstanceId, worker_count: usize) -> usize {
317    let worker_count = u64::try_from(worker_count.max(1)).unwrap_or(u64::MAX);
318    usize::try_from(room_instance_id.as_u64() % worker_count).unwrap_or_default()
319}
320
321#[cfg(any(test, feature = "testing-transport"))]
322#[derive(Debug)]
323pub struct JoinPlacementTestGate {
324    ready: Barrier,
325    release: Barrier,
326}
327
328#[cfg(any(test, feature = "testing-transport"))]
329impl JoinPlacementTestGate {
330    #[must_use]
331    pub fn new(expected: usize) -> Self {
332        Self {
333            ready: Barrier::new(expected + 1),
334            release: Barrier::new(expected + 1),
335        }
336    }
337
338    async fn wait_before_commit(&self) {
339        self.ready.wait().await;
340        self.release.wait().await;
341    }
342
343    pub async fn hold_all_ready(&self) {
344        self.ready.wait().await;
345    }
346
347    pub async fn release_all(&self) {
348        self.release.wait().await;
349    }
350}
351
352#[cfg(test)]
353#[path = "TESTS/placement.rs"]
354mod tests;
o_sfu_core/engine/room/placement.rs

o_sfu_core/engine/room/
placement.rs
