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::{MediaWorkerId, RoomInstanceId, media_transport::MediaTransport},
23};
24
25/// Initial router placement context for one room instance.
26///
27/// [`Self::new_unassigned`] defers packet-worker selection until admission.
28#[derive(Debug, Clone, PartialEq, Eq)]
29pub struct RoomRuntimeContext {
30    instance: RoomInstanceId,
31    primary_router: RouterId,
32    initial_router_placements: Option<RouterPlacements>,
33}
34
35impl RoomRuntimeContext {
36    #[must_use]
37    pub fn new(
38        instance: RoomInstanceId,
39        primary: RouterPlacement,
40        spillover: Vec<RouterPlacement>,
41    ) -> Self {
42        Self {
43            instance,
44            primary_router: primary.router,
45            initial_router_placements: Some(RouterPlacements::new(primary, spillover)),
46        }
47    }
48
49    #[must_use]
50    pub const fn new_unassigned(instance: RoomInstanceId, primary_router: RouterId) -> Self {
51        Self {
52            instance,
53            primary_router,
54            initial_router_placements: None,
55        }
56    }
57
58    /// # Errors
59    ///
60    /// returns [`RouterPlacementsError::Empty`] when `placements` is empty
61    pub fn try_from_placements(
62        instance: RoomInstanceId,
63        placements: Vec<RouterPlacement>,
64    ) -> Result<Self, RouterPlacementsError> {
65        let routers = RouterPlacements::try_from_vec(placements)?;
66        Ok(Self {
67            instance,
68            primary_router: routers.primary().router,
69            initial_router_placements: Some(routers),
70        })
71    }
72
73    #[must_use]
74    pub const fn instance(&self) -> RoomInstanceId {
75        self.instance
76    }
77
78    #[must_use]
79    pub const fn primary_router(&self) -> RouterId {
80        self.primary_router
81    }
82
83    #[must_use]
84    pub fn initial_router_placements(&self) -> Option<&RouterPlacements> {
85        self.initial_router_placements.as_ref()
86    }
87}
88
89#[cfg(any(test, feature = "testing-transport"))]
90impl Room {
91    pub(super) async fn placement_usage_snapshot(&self) -> PlacementSnapshot {
92        self.state.read().await.placement_usage_snapshot()
93    }
94}
95
96enum PacketLoopDelaySource<'a> {
97    Transport(&'a MediaTransport),
98    #[cfg(any(test, feature = "testing-transport"))]
99    Fixed(Vec<Option<u64>>),
100}
101
102impl PacketLoopDelaySource<'_> {
103    fn snapshot(self) -> Vec<Option<u64>> {
104        match self {
105            Self::Transport(transport) => transport.packet_loop_delays_ms(),
106            #[cfg(any(test, feature = "testing-transport"))]
107            Self::Fixed(delays_ms) => delays_ms,
108        }
109    }
110}
111
112pub(super) struct JoinAdmissionTurn<'a, A = fn() -> RouterId> {
113    request: JoinUserRequest,
114    packet_loop_delays: PacketLoopDelaySource<'a>,
115    allocate_spillover_router: A,
116    #[cfg(any(test, feature = "testing-transport"))]
117    gate: Option<Arc<JoinPlacementTestGate>>,
118}
119
120impl JoinAdmissionTurn<'_> {
121    pub(super) fn from_factory<'a>(
122        request: JoinUserRequest,
123        media_transport: &'a MediaTransport,
124        factory: &'a RoomFactory,
125    ) -> JoinAdmissionTurn<'a, impl FnOnce() -> RouterId + 'a> {
126        JoinAdmissionTurn {
127            request,
128            packet_loop_delays: PacketLoopDelaySource::Transport(media_transport),
129            allocate_spillover_router: move || factory.allocate_spillover_router(),
130            #[cfg(any(test, feature = "testing-transport"))]
131            gate: None,
132        }
133    }
134
135    #[cfg(any(test, feature = "testing-transport"))]
136    pub(super) fn for_test(
137        request: JoinUserRequest,
138        delays_ms: Vec<Option<u64>>,
139        spillover_router_id: RouterId,
140    ) -> JoinAdmissionTurn<'static, impl FnOnce() -> RouterId> {
141        JoinAdmissionTurn {
142            request,
143            packet_loop_delays: PacketLoopDelaySource::Fixed(delays_ms),
144            allocate_spillover_router: move || spillover_router_id,
145            gate: None,
146        }
147    }
148}
149
150impl<A: FnOnce() -> RouterId> JoinAdmissionTurn<'_, A> {
151    #[cfg(any(test, feature = "testing-transport"))]
152    pub(super) fn with_gate(mut self, gate: Option<Arc<JoinPlacementTestGate>>) -> Self {
153        self.gate = gate;
154        self
155    }
156
157    pub(super) async fn commit(
158        self,
159        room: &Room,
160        joined_fanout: UserJoinedFanout,
161    ) -> Result<JoinCommit, RoomJoinError> {
162        #[cfg(any(test, feature = "testing-transport"))]
163        if let Some(gate) = &self.gate {
164            gate.wait_before_commit().await;
165        }
166        let mut state = room.state.write().await;
167        // Sample worker delay after reaching the serialized commit turn. The
168        // state guard makes each join select from placements committed earlier.
169        let delays_ms = self.packet_loop_delays.snapshot();
170        let worker_count = delays_ms.len().max(1);
171        let start_worker = room_worker_start(room.instance_id(), worker_count);
172        let placement = choose_placement(
173            &state.placement_usage_snapshot(),
174            room.room_worker_policy(),
175            &delays_ms,
176            start_worker,
177            self.allocate_spillover_router,
178        );
179        state.apply_join_on_placement(
180            &self.request.user_id,
181            self.request.permissions,
182            self.request.sender,
183            joined_fanout,
184            placement,
185        )
186    }
187}
188
189fn choose_placement(
190    room: &PlacementSnapshot,
191    policy: RoomWorkerPolicy,
192    delays_ms: &[Option<u64>],
193    start_worker: usize,
194    allocate_spillover_router: impl FnOnce() -> RouterId,
195) -> RouterPlacement {
196    let worker_count = delays_ms.len().max(1);
197    let threshold_ms = policy.packet_loop_delay_threshold_ms();
198    let assigned = room.assigned_placements();
199    let Some(primary) = assigned.first().copied() else {
200        return RouterPlacement {
201            router: room.primary(),
202            media_worker: choose_primary_worker(
203                delays_ms,
204                threshold_ms,
205                start_worker % worker_count,
206            ),
207        };
208    };
209    if policy.max_local_routers() == 1 {
210        return primary;
211    }
212    if let Some(placement) = assigned
213        .iter()
214        .filter(|placement| worker_is_healthy(delays_ms, placement.media_worker, threshold_ms))
215        .min_by_key(|placement| worker_delay(delays_ms, placement.media_worker))
216    {
217        return *placement;
218    }
219    let placement_cap = policy.max_local_routers().min(worker_count);
220    if assigned.len() < placement_cap
221        && let Some(media_worker) = cyclic_workers(start_worker, worker_count).find(|worker| {
222            worker_is_healthy(delays_ms, *worker, threshold_ms)
223                && assigned
224                    .iter()
225                    .all(|placement| placement.media_worker != *worker)
226        })
227    {
228        return RouterPlacement {
229            router: allocate_spillover_router(),
230            media_worker,
231        };
232    }
233    assigned
234        .iter()
235        .copied()
236        .min_by_key(|placement| worker_delay(delays_ms, placement.media_worker))
237        .unwrap_or(primary)
238}
239
240fn choose_primary_worker(
241    delays_ms: &[Option<u64>],
242    threshold_ms: u64,
243    start_worker: usize,
244) -> MediaWorkerId {
245    let worker_count = delays_ms.len().max(1);
246    cyclic_workers(start_worker, worker_count)
247        .find(|worker| worker_is_healthy(delays_ms, *worker, threshold_ms))
248        .or_else(|| {
249            cyclic_workers(start_worker, worker_count)
250                .min_by_key(|worker| worker_delay(delays_ms, *worker))
251        })
252        .unwrap_or_else(|| MediaWorkerId::from_raw(0))
253}
254
255fn cyclic_workers(start_worker: usize, worker_count: usize) -> impl Iterator<Item = MediaWorkerId> {
256    (0..worker_count).map(move |offset| {
257        MediaWorkerId::from_raw(start_worker.wrapping_add(offset) % worker_count)
258    })
259}
260
261fn worker_is_healthy(delays_ms: &[Option<u64>], worker: MediaWorkerId, threshold_ms: u64) -> bool {
262    worker_delay(delays_ms, worker) < threshold_ms
263}
264
265fn worker_delay(delays_ms: &[Option<u64>], worker: MediaWorkerId) -> u64 {
266    // Missing samples cannot qualify a worker as healthy. They retain the worst
267    // rank for the all-unhealthy fallback.
268    delays_ms
269        .get(worker.as_usize())
270        .copied()
271        .flatten()
272        .unwrap_or(u64::MAX)
273}
274
275fn room_worker_start(room_instance_id: RoomInstanceId, worker_count: usize) -> usize {
276    let worker_count = u64::try_from(worker_count.max(1)).unwrap_or(u64::MAX);
277    usize::try_from(room_instance_id.as_u64() % worker_count).unwrap_or_default()
278}
279
280#[cfg(any(test, feature = "testing-transport"))]
281#[derive(Debug)]
282pub struct JoinPlacementTestGate {
283    ready: Barrier,
284    release: Barrier,
285}
286
287#[cfg(any(test, feature = "testing-transport"))]
288impl JoinPlacementTestGate {
289    #[must_use]
290    pub fn new(expected: usize) -> Self {
291        Self {
292            ready: Barrier::new(expected + 1),
293            release: Barrier::new(expected + 1),
294        }
295    }
296
297    async fn wait_before_commit(&self) {
298        self.ready.wait().await;
299        self.release.wait().await;
300    }
301
302    pub async fn hold_all_ready(&self) {
303        self.ready.wait().await;
304    }
305
306    pub async fn release_all(&self) {
307        self.release.wait().await;
308    }
309}
310
311#[cfg(test)]
312#[path = "TESTS/placement.rs"]
313mod tests;
o_sfu_core/engine/room/placement.rs

o_sfu_core/engine/room/
placement.rs
