o_sfu_core/engine/media_transport/
policy_invalidation.rs

1//! Coalesces transport observations into room source-policy wakeups.
2
3use std::{
4    collections::{BTreeMap, BTreeSet},
5    iter, mem,
6    sync::{Arc, Mutex},
7    time::Duration,
8};
9
10use tokio::{sync::Notify, time::sleep};
11
12use super::MediaTransport;
13use crate::{RoomInstanceId, engine::sync::lock_unpoisoned};
14
15const SOURCE_POLICY_FOLLOW_UP_DELAY: Duration = Duration::from_millis(250);
16
17#[derive(Debug, Default)]
18struct PendingSourcePolicyUpdates {
19    rooms: BTreeSet<RoomInstanceId>,
20    scheduled: BTreeMap<RoomInstanceId, u64>,
21    next_schedule_token: u64,
22}
23
24#[derive(Debug, Default)]
25struct SourcePolicyUpdates {
26    pending: Mutex<PendingSourcePolicyUpdates>,
27    notify: Notify,
28}
29
30impl SourcePolicyUpdates {
31    fn take(&self) -> BTreeSet<RoomInstanceId> {
32        mem::take(&mut lock_unpoisoned(&self.pending).rooms)
33    }
34
35    fn publish(&self, room: RoomInstanceId, token: u64) {
36        let mut pending = lock_unpoisoned(&self.pending);
37        if pending.scheduled.get(&room) != Some(&token) {
38            return;
39        }
40        pending.scheduled.remove(&room);
41        let notify = pending.rooms.is_empty();
42        pending.rooms.insert(room);
43        drop(pending);
44        if notify {
45            self.notify.notify_one();
46        }
47    }
48}
49
50/// Shared drain for coalesced room source-policy invalidations.
51///
52/// Clones share one drain. The runtime must assign all clones to a single
53/// consumer task.
54#[derive(Debug, Clone)]
55pub struct SourcePolicyUpdateSubscription(Arc<SourcePolicyUpdates>);
56
57impl SourcePolicyUpdateSubscription {
58    /// Waits until at least one room needs a source-policy pass.
59    pub async fn wait_for_update(&self) -> BTreeSet<RoomInstanceId> {
60        loop {
61            let rooms = self.0.take();
62            if !rooms.is_empty() {
63                return rooms;
64            }
65            self.0.notify.notified().await;
66        }
67    }
68
69    /// Drains updates published after the previous wait completed.
70    #[must_use]
71    pub fn take_pending_updates(&self) -> BTreeSet<RoomInstanceId> {
72        self.0.take()
73    }
74}
75
76/// Sender for coalesced room source-policy updates.
77#[derive(Debug, Clone, Default)]
78pub struct SourcePolicySignal(Arc<SourcePolicyUpdates>);
79
80impl SourcePolicySignal {
81    /// Creates the runtime's single-consumer subscription.
82    #[must_use]
83    pub fn subscribe(&self) -> SourcePolicyUpdateSubscription {
84        SourcePolicyUpdateSubscription(Arc::clone(&self.0))
85    }
86
87    /// Marks one room as needing a source-policy pass.
88    pub fn mark_dirty(&self, room_instance_id: RoomInstanceId) {
89        self.mark_dirty_rooms([room_instance_id]);
90    }
91
92    /// Marks rooms as needing a source-policy pass.
93    pub fn mark_dirty_rooms(&self, room_instance_ids: impl IntoIterator<Item = RoomInstanceId>) {
94        let mut room_instance_ids = room_instance_ids.into_iter();
95        let Some(first) = room_instance_ids.next() else {
96            return;
97        };
98        let mut pending = lock_unpoisoned(&self.0.pending);
99        // The single consumer drains `pending.rooms`, so only its
100        // empty-to-nonempty transition needs a wake.
101        let notify = pending.rooms.is_empty();
102        if pending.scheduled.is_empty() {
103            pending.rooms.insert(first);
104            pending.rooms.extend(room_instance_ids);
105        } else {
106            for room in iter::once(first).chain(room_instance_ids) {
107                pending.scheduled.remove(&room);
108                pending.rooms.insert(room);
109            }
110        }
111        drop(pending);
112        if notify {
113            self.0.notify.notify_one();
114        }
115    }
116
117    /// Schedules one room wake unless current work makes it immediately dirty.
118    ///
119    /// Tokens prevent stale spawned tasks from publishing after an immediate
120    /// wake cancels then reschedules the same room.
121    fn mark_dirty_after(&self, room: RoomInstanceId, delay: Duration) {
122        let token = {
123            let mut pending = lock_unpoisoned(&self.0.pending);
124            if pending.rooms.contains(&room) || pending.scheduled.contains_key(&room) {
125                return;
126            }
127            let token = pending.next_schedule_token;
128            pending.next_schedule_token = pending.next_schedule_token.wrapping_add(1);
129            pending.scheduled.insert(room, token);
130            token
131        };
132        let updates = Arc::clone(&self.0);
133        tokio::spawn(async move {
134            sleep(delay).await;
135            updates.publish(room, token);
136        });
137    }
138}
139
140impl MediaTransport {
141    /// Schedules another policy pass when adaptation hysteresis remains unresolved.
142    pub(in crate::engine) fn schedule_source_policy_follow_up(&self, room: RoomInstanceId) {
143        // Delay prevents one transport wake from consuming several consecutive
144        // observation thresholds. Per-room follow-ups coalesce.
145        self.source_policy_signal
146            .mark_dirty_after(room, SOURCE_POLICY_FOLLOW_UP_DELAY);
147    }
148}
149
150#[cfg(test)]
151#[path = "TESTS/policy_invalidation.rs"]
152mod tests;
o_sfu_core/engine/media_transport/policy_invalidation.rs

o_sfu_core/engine/media_transport/
policy_invalidation.rs
