o_sfu_core/engine/room/source_policy/
input.rs

1use std::{
2    cmp::Reverse,
3    collections::{BTreeMap, BTreeSet},
4};
5
6use o_sfu_router::MediaKind;
7
8use super::action::FeaturedUserUpdate;
9use crate::{
10    Bitrate, RoomMediaLimits, VideoAdaptationTuning,
11    engine::{
12        ConnectionId, UserId,
13        media_transport::{
14            ActiveSpeakerSource, ReceiverBandwidthSnapshot, ReceiverBweTargetUpdate,
15            TransportBitrateSnapshot, TransportMediaId,
16        },
17        room::{
18            media_graph::ConsumerRouteView,
19            state::{ActiveUser, RoomState},
20        },
21    },
22};
23
24const ACTIVE_SPEAKER_FEATURED_CLEAR_LIMIT: usize = 5;
25
26#[derive(Debug)]
27pub(super) struct SourcePolicySnapshot<'a> {
28    pub(super) routes: Vec<ConsumerRouteView<'a>>,
29    pub(super) receiver_bwe_targets: BTreeMap<UserId, ReceiverBweTargetUpdate>,
30    pub(super) receiver_bandwidth_by_connection: BTreeMap<ConnectionId, Bitrate>,
31    pub(super) source_bitrate_by_media: BTreeMap<TransportMediaId, Bitrate>,
32    pub(super) active_speaker_media_ids: BTreeSet<TransportMediaId>,
33    pub(super) admitted_audio_media_ids: BTreeSet<TransportMediaId>,
34    pub(super) deaf_receiver_connection_ids: BTreeSet<ConnectionId>,
35    pub(super) featured_source_user_ids: BTreeSet<UserId>,
36    pub(super) active_speaker_rank_by_user: BTreeMap<UserId, usize>,
37    pub(super) featured_user_updates: Vec<FeaturedUserUpdate>,
38    pub(super) user_count: usize,
39    pub(super) media_limits: RoomMediaLimits,
40    pub(super) video_adaptation_tuning: VideoAdaptationTuning,
41    pub(super) audio_reserve_by_connection: BTreeMap<ConnectionId, Bitrate>,
42}
43
44impl<'a> SourcePolicySnapshot<'a> {
45    pub(super) fn from_state(
46        room: &'a RoomState,
47        active_speaker_sources: &[ActiveSpeakerSource],
48        receiver_bandwidth_snapshot: &ReceiverBandwidthSnapshot,
49        source_bitrate_snapshot: &TransportBitrateSnapshot,
50    ) -> Self {
51        let ranked_sources = rank_room_active_speakers(room, active_speaker_sources);
52        let media_limits = room.media_limits;
53        let tuning = room.video_adaptation_tuning;
54        let active_speakers = active_speaker_media_ids(&ranked_sources);
55        let admitted_audio_speakers = admitted_audio_media_ids(
56            room,
57            &ranked_sources,
58            media_limits.max_active_audio_speakers(),
59        );
60        let deaf_receiver_connection_ids = deaf_receiver_connection_ids(room);
61        let featured_source_user_ids = featured_source_user_ids(room, &ranked_sources);
62        let active_speaker_rank_by_user = active_speaker_rank_by_user(room, &ranked_sources);
63        let desired_featured_user_id = ranked_sources.iter().find_map(|source| {
64            featured_source_owner_for_active_speaker_source(room, source.transport_media_id())
65        });
66        let featured_user_updates = featured_user_updates(room, desired_featured_user_id.as_ref());
67        // Include policy-paused routes so later turns can resume them. Filtering
68        // on `delivery_active()` would make a policy pause self-perpetuating.
69        let routes = room
70            .committed_consumer_routes()
71            .filter(|route| route.source.active && route.selection.active())
72            .collect::<Vec<_>>();
73        let audio_reserve_by_connection = audio_reserve_by_connection(
74            &routes,
75            &admitted_audio_speakers,
76            &deaf_receiver_connection_ids,
77            tuning.audio_reserve_per_speaker,
78        );
79        Self {
80            routes,
81            receiver_bwe_targets: receiver_bwe_targets(room, &audio_reserve_by_connection),
82            receiver_bandwidth_by_connection: receiver_bandwidth_by_connection(
83                receiver_bandwidth_snapshot,
84            ),
85            source_bitrate_by_media: source_bitrate_snapshot.per_media.iter().copied().collect(),
86            active_speaker_media_ids: active_speakers,
87            admitted_audio_media_ids: admitted_audio_speakers,
88            deaf_receiver_connection_ids,
89            featured_source_user_ids,
90            active_speaker_rank_by_user,
91            featured_user_updates,
92            user_count: room.user_count(),
93            media_limits,
94            video_adaptation_tuning: tuning,
95            audio_reserve_by_connection,
96        }
97    }
98}
99
100/// Bandwidth reserved for admitted audio before video budgeting, per receiver
101/// connection.
102///
103/// Each receiver reserves `per_speaker` for every admitted audio route it
104/// actually consumes, so a receiver that disabled audio, deafened itself (or a
105/// publisher with no consumer routes) reserves nothing and keeps its full video
106/// budget. The reserve is fixed per route, so it is deterministic and
107/// independent of policy-turn cadence. A zero per-speaker rate disables the
108/// reservation and returns an empty map.
109fn audio_reserve_by_connection(
110    routes: &[ConsumerRouteView<'_>],
111    admitted_audio_media_ids: &BTreeSet<TransportMediaId>,
112    deaf_receiver_connection_ids: &BTreeSet<ConnectionId>,
113    per_speaker: Bitrate,
114) -> BTreeMap<ConnectionId, Bitrate> {
115    if per_speaker.as_bps() == 0 {
116        return BTreeMap::new();
117    }
118    let mut reserve_by_connection = BTreeMap::new();
119    for route in routes {
120        if route.source.descriptor.media_kind() != MediaKind::Audio
121            || !admitted_audio_media_ids.contains(&route.route.source_transport_media_id())
122        {
123            continue;
124        }
125        let connection_id = route.route.consumer_session_key().connection_id();
126        if deaf_receiver_connection_ids.contains(&connection_id) {
127            continue;
128        }
129        let reserve = reserve_by_connection
130            .entry(connection_id)
131            .or_insert_with(Bitrate::zero);
132        *reserve = reserve.saturating_add(per_speaker);
133    }
134    reserve_by_connection
135}
136
137fn receiver_bwe_targets(
138    room: &RoomState,
139    audio_reserve_by_connection: &BTreeMap<ConnectionId, Bitrate>,
140) -> BTreeMap<UserId, ReceiverBweTargetUpdate> {
141    // Seed every receiver, including one with no selected media. Otherwise a
142    // previous nonzero desired bitrate remains installed in str0m's BWE controller.
143    room.transport_user_entries()
144        .map(|(user_id, connection_id)| {
145            let session = room.transport_user_key(user_id, connection_id);
146            let audio_reserve = audio_reserve_by_connection
147                .get(&connection_id)
148                .copied()
149                .unwrap_or_else(Bitrate::zero);
150            (
151                user_id.clone(),
152                ReceiverBweTargetUpdate::new(session, audio_reserve),
153            )
154        })
155        .collect()
156}
157
158fn receiver_bandwidth_by_connection(
159    snapshot: &ReceiverBandwidthSnapshot,
160) -> BTreeMap<ConnectionId, Bitrate> {
161    snapshot
162        .per_session
163        .iter()
164        .map(|(session, estimate)| (session.connection_id(), *estimate))
165        .collect()
166}
167
168/// Filters and ranks a list of active speaker sources for a room.
169///
170/// **Ranking Criteria:**
171/// 1. **Recency:** Most recently active first (highest `observed_at`).
172/// 2. **Loudness:** Highest audio level first (`last_audio_level_dbov`).
173/// 3. **Tie-breaker:** Transport media ID.
174///
175/// Only retains sources that are active and present in the current room topology.
176fn rank_room_active_speakers(
177    room: &RoomState,
178    sources: &[ActiveSpeakerSource],
179) -> Vec<ActiveSpeakerSource> {
180    let mut sources = sources.to_vec();
181    sources.retain(|source| {
182        room.topology
183            .source_for_transport_media(source.transport_media_id())
184            .is_some_and(|source| source.active)
185    });
186    sources.sort_unstable_by_key(|source| {
187        (
188            Reverse(source.observed_at()),
189            Reverse(source.last_audio_level_dbov().unwrap_or(i8::MIN)),
190            source.transport_media_id().as_u64(),
191        )
192    });
193    sources
194}
195
196fn active_speaker_media_ids(sources: &[ActiveSpeakerSource]) -> BTreeSet<TransportMediaId> {
197    sources
198        .iter()
199        .map(|source| source.transport_media_id())
200        .collect()
201}
202
203fn user_for_source<'a>(
204    room: &'a RoomState,
205    source: &ActiveSpeakerSource,
206) -> Option<&'a ActiveUser> {
207    room.topology
208        .source_for_transport_media(source.transport_media_id())
209        .and_then(|published_source| {
210            room.users
211                .get(published_source.descriptor.owner().user_id())
212        })
213}
214
215/// Takes audio media IDs from `sources` up to the provided `limit`,
216/// prioritizing participants who are currently screen sharing.
217fn admitted_audio_media_ids(
218    room: &RoomState,
219    sources: &[ActiveSpeakerSource],
220    limit: usize,
221) -> BTreeSet<TransportMediaId> {
222    let mut admitted = BTreeSet::new();
223    let mut deferred = Vec::with_capacity(limit);
224    for source in sources {
225        if admitted.len() == limit {
226            break;
227        }
228        let media_id = source.transport_media_id();
229        // prioritize participants who are currently screen sharing
230        if user_for_source(room, source).is_some_and(ActiveUser::is_screensharing) {
231            admitted.insert(media_id);
232        } else if deferred.len() < limit - admitted.len() {
233            deferred.push(media_id);
234        }
235    }
236    admitted.extend(deferred.into_iter().take(limit - admitted.len()));
237    admitted
238}
239
240fn deaf_receiver_connection_ids(room: &RoomState) -> BTreeSet<ConnectionId> {
241    room.users
242        .values()
243        .filter(|user| user.is_deaf())
244        .map(|user| user.connection_id)
245        .collect()
246}
247
248fn featured_source_user_ids(room: &RoomState, sources: &[ActiveSpeakerSource]) -> BTreeSet<UserId> {
249    sources
250        .iter()
251        .filter_map(|source| {
252            featured_source_owner_for_active_speaker_source(room, source.transport_media_id())
253        })
254        .take(ACTIVE_SPEAKER_FEATURED_CLEAR_LIMIT)
255        .collect()
256}
257
258fn active_speaker_rank_by_user(
259    room: &RoomState,
260    sources: &[ActiveSpeakerSource],
261) -> BTreeMap<UserId, usize> {
262    let mut ranks = BTreeMap::new();
263    for source in sources {
264        let Some(user_id) =
265            featured_source_owner_for_active_speaker_source(room, source.transport_media_id())
266        else {
267            continue;
268        };
269        let next_rank = ranks.len();
270        ranks.entry(user_id).or_insert(next_rank);
271    }
272    ranks
273}
274
275fn featured_source_owner_for_active_speaker_source(
276    room: &RoomState,
277    transport_media_id: TransportMediaId,
278) -> Option<UserId> {
279    room.topology
280        .active_speaker_detector_owner(transport_media_id)
281}
282
283fn featured_user_updates(
284    room: &RoomState,
285    desired_featured_user_id: Option<&UserId>,
286) -> Vec<FeaturedUserUpdate> {
287    if desired_featured_user_id.is_none()
288        && !room.users.values().any(|user| user.featured().is_some())
289    {
290        return Vec::new();
291    }
292    room.users
293        .iter()
294        .filter_map(|(user_id, user)| {
295            let current_featured = user.featured();
296            let desired_featured = match desired_featured_user_id {
297                Some(featured_user_id) => Some(featured_user_id == user_id),
298                None if current_featured.is_some() => Some(false),
299                None => None,
300            };
301            (desired_featured != current_featured).then(|| {
302                FeaturedUserUpdate::new(user_id.clone(), user.connection_id, desired_featured)
303            })
304        })
305        .collect()
306}
o_sfu_core/engine/room/source_policy/input.rs

o_sfu_core/engine/room/source_policy/
input.rs
