diff --git a/src/app.rs b/src/app.rs index 6c973d2..e891033 100644 --- a/src/app.rs +++ b/src/app.rs @@ -214,6 +214,14 @@ pub struct App<'a> { capturable_objects: HashSet, /// Objects currently being captured. capturing_objects: HashSet, + /// Index into the sorted eligible-objects list where the next capture + /// rotation window should start. Only meaningful when + /// config.max_concurrent_captures is set. + capture_rotation_start: usize, + /// Rendered frames elapsed since the capture rotation window last + /// advanced. See ROTATION_FRAME_INTERVAL - deliberately a frame count, + /// not a wall-clock duration. + frames_since_rotation: u32, } macro_rules! current_list { @@ -261,6 +269,8 @@ impl<'a> App<'a> { peak_processor: Arc::new(peak_processor), capturable_objects: HashSet::new(), capturing_objects: HashSet::new(), + capture_rotation_start: 0, + frames_since_rotation: 0, } } @@ -304,6 +314,13 @@ impl<'a> App<'a> { if needs_render && pacer.is_time_to_render() { needs_render = false; + // Tied to the render cadence itself (see + // ROTATION_FRAME_INTERVAL) rather than a wall-clock timer of + // its own: rotate_capturing() is only ever called here, in + // step with an actually-rendered frame, so counting frames + // there is exact rather than an approximation of one. + self.rotate_capturing(); + self.mouse_areas.clear(); terminal.draw(|frame| { @@ -346,6 +363,28 @@ impl<'a> App<'a> { self.wirehose.node_capture_stop(object_id); } + /// Node name wiremix (fork or upstream) gives its own peak-monitoring + /// capture streams - see wirehose::stream::capture_node. Counting nodes + /// with this name in the shared PipeWire graph state is what makes + /// max_concurrent_captures_global possible without any custom IPC: the + /// graph itself is already the shared, live state every instance reads. + const CAPTURE_NODE_NAME: &'static str = "wiremix-capture"; + + /// Total number of "wiremix-capture" nodes currently visible in the + /// PipeWire graph, from any client - not just this instance's own + /// capturing_objects. See the comment at its call site in + /// start_capture() for the consistency caveats this implies. + fn global_capture_count(&self) -> usize { + self.state + .nodes + .values() + .filter(|node| { + node.props.node_name().map(String::as_str) + == Some(Self::CAPTURE_NODE_NAME) + }) + .count() + } + fn start_capture(&mut self, object_id: ObjectId) { if self.config.lazy_capture && !self.visible_objects.contains(&object_id) @@ -353,6 +392,70 @@ impl<'a> App<'a> { return; } + // Hard cap on concurrent captures, enforced here (not just in the + // rotation logic) so it can never be transiently exceeded - e.g. by + // several nodes becoming capture-eligible at once before the next + // rotation tick has a chance to run. + if let Some(max) = self.config.max_concurrent_captures { + if self.capturing_objects.len() >= max + && !self.capturing_objects.contains(&object_id) + { + return; + } + } + + // Same idea, but counting *every* "wiremix-capture" node currently + // visible in the PipeWire graph - including ones opened by other + // wiremix instances - rather than just this instance's own + // capturing_objects. This is necessarily best-effort: there's no + // cross-process locking, so two instances racing to start a new + // capture at the same moment can both observe room under the cap + // and both proceed, transiently exceeding it until the graph state + // settles. It also can't make non-cooperating processes (upstream + // wiremix without this option, or any other client also named + // "wiremix-capture") back off - it only makes instances that opt + // into this setting considerate of each other and of themselves. + // + // global_capture_count() alone is not enough: it only reflects + // captures that have round-tripped through the PipeWire server and + // come back as a state update over wirehose's event channel, but + // node_capture_start() (called at the bottom of this function) is + // fire-and-forget - it queues a command for a separate PipeWire + // thread and returns immediately, with no synchronous confirmation. + // handle_events() drains and processes every currently-queued event + // in one synchronous burst (see its `while let Ok(event) = + // self.rx.try_recv()` loop), which is exactly when a flood of + // CaptureEligibility::Eligible events (e.g. on startup, or when + // lazy_capture reveals many nodes at once) calls start_capture() + // for many objects back to back. None of those calls' own + // node_capture_start() commands can possibly have round-tripped + // back into self.state.nodes by the time the next call in the same + // burst runs this check - so global_capture_count() reports the + // same stale pre-burst number for the whole burst, and every call + // sees "room under the cap" even after dozens of this instance's + // own captures have already been started. Confirmed live: a single + // instance with max_concurrent_captures_global = 20 and no other + // instances contending ended up with 28 of its own active + // captures, purely from this self-lag - not from any cross-instance + // race, and not from CaptureEligibility::NeedsRestart (ruled out + // separately: capture streams are properly disconnected on renewal + // now that StreamRegistry::add_stream() calls disconnect() on the + // stream it evicts, matching what StreamRegistry::remove() already + // did - see that fix's own commit). self.capturing_objects.len() + // has no such lag - it's updated synchronously the instant this + // instance decides to capture something, before any round trip - + // so folding it in as a floor closes the self-lag gap without + // changing the inherent, already-documented cross-instance + // best-effort behavior above. + if let Some(max) = self.config.max_concurrent_captures_global { + let estimate = self + .global_capture_count() + .max(self.capturing_objects.len()); + if estimate >= max && !self.capturing_objects.contains(&object_id) { + return; + } + } + let Some(node) = self.state.nodes.get(&object_id) else { return; }; @@ -417,6 +520,101 @@ impl<'a> App<'a> { } } + /// How many rendered frames pass between capture rotation slides, when + /// max_concurrent_captures limits capture to fewer than are eligible. + /// Deliberately a multiple of the render cadence itself (see + /// RenderPacer and rotate_capturing()'s call site) rather than an + /// independent wall-clock timer of its own: rotation speed - and the + /// PipeWire stream churn it costs - then scales automatically with + /// whatever fps a user has configured, with no second unrelated timer + /// to reason about. Every frame would create more stream churn than not + /// rotating at all (see rotate_capturing()'s doc comment); every 3rd + /// frame keeps swap rates to a handful per second even at high fps, + /// while still giving near-real-time coverage at typical settings - + /// e.g. at 10-20fps, sweeping through ~10 eligible objects at + /// max_concurrent_captures=2 takes well under 3 seconds. + const ROTATION_FRAME_INTERVAL: u32 = 3; + + /// If max_concurrent_captures is set and more objects are eligible for + /// capture than that, periodically swap which subset is actually being + /// captured so every eligible object eventually gets sampled instead of + /// whichever ones happened to become eligible first (and then stay + /// captured forever, starving everything else). + /// + /// Slides the window by exactly one object per tick, rather than + /// jumping it forward by the full window size - so at most one capture + /// starts and one stops per tick, and everything else already in the + /// window is left alone. A full-window jump every tick would mean every + /// currently-captured object goes stale for the whole interval and then + /// *all* of them change at once; a one-at-a-time slide instead gives a + /// continuous, staggered trickle where something is always freshly + /// updated. + /// + /// Only ever called once per actually-rendered frame (see its call site + /// in run()), so counting frames via frames_since_rotation is exact. + fn rotate_capturing(&mut self) { + let Some(max) = self.config.max_concurrent_captures else { + return; + }; + + self.frames_since_rotation = + self.frames_since_rotation.saturating_add(1); + if self.frames_since_rotation < Self::ROTATION_FRAME_INTERVAL { + return; + } + + // Same eligibility rule start_capture()/update_capturing() already + // use: scoped to on-screen objects under lazy_capture, otherwise + // every capturable object regardless of visibility. + let mut eligible: Vec = if self.config.lazy_capture { + self.visible_objects + .intersection(&self.capturable_objects) + .copied() + .collect() + } else { + self.capturable_objects.iter().copied().collect() + }; + + if eligible.len() <= max { + // Everything eligible already fits under the cap - nothing to + // rotate. Leave frames_since_rotation alone (already at or past + // the threshold) so a rotation isn't "owed" extra delay the + // moment the eligible set grows past max again. + return; + } + + self.frames_since_rotation = 0; + + // Sorting gives a stable, deterministic rotation order across ticks + // (HashSet iteration order isn't stable) so each tick advances + // through the *same* sequence rather than picking an arbitrary new + // subset every time. + eligible.sort_unstable(); + + let n = eligible.len(); + let start = self.capture_rotation_start % n; + let window: HashSet = + (0..max.min(n)).map(|i| eligible[(start + i) % n]).collect(); + self.capture_rotation_start = (start + 1) % n; + + let need_to_stop: Vec<_> = self + .capturing_objects + .difference(&window) + .copied() + .collect(); + for object_id in need_to_stop { + self.stop_capture(object_id); + } + + let need_to_start: Vec<_> = window + .difference(&self.capturing_objects) + .copied() + .collect(); + for object_id in need_to_start { + self.start_capture(object_id); + } + } + fn set_capture_eligibility( &mut self, capture_eligibility: CaptureEligibility, @@ -882,6 +1080,8 @@ mod tests { tabs: vec![TabKind::Playback], lazy_capture: Default::default(), filters: Default::default(), + max_concurrent_captures: Default::default(), + max_concurrent_captures_global: Default::default(), }; let mut app = App::new(wirehose, event_rx, config); @@ -983,6 +1183,8 @@ mod tests { ], lazy_capture: Default::default(), filters: Default::default(), + max_concurrent_captures: Default::default(), + max_concurrent_captures_global: Default::default(), }; let mut app = App::new(&wirehose, event_rx, config); @@ -1271,4 +1473,184 @@ mod tests { Some(mock::MockCommand::NodeCaptureStop(id)) ); } + + /// Sets frames_since_rotation so the next rotate_capturing() call + /// doesn't have to wait out the real frame-count interval. + fn force_rotation_due(app: &mut App<'_>) { + app.frames_since_rotation = App::ROTATION_FRAME_INTERVAL; + } + + #[test] + fn start_capture_enforces_max_concurrent() { + let commands = RefCell::new(VecDeque::new()); + let wirehose = mock::WirehoseHandle::with_commands(&commands); + let (_, event_rx) = mpsc::channel(); + let config = Config::from_toml_str( + "lazy_capture = false\nmax_concurrent_captures = 2", + ); + let mut app = App::new(&wirehose, event_rx, config); + + for i in 1..=3 { + let id = ObjectId::from_raw_id(i); + add_capturable_node(&mut app, id); + app.set_capture_eligibility(CaptureEligibility::Eligible(id)); + } + + // Only 2 of the 3 eligible nodes should actually be capturing. + assert_eq!(app.capturing_objects.len(), 2); + } + + /// Adds a node with node.name = "wiremix-capture" but never marks it + /// eligible - simulates another wiremix instance's own capture stream, + /// which is visible in the shared PipeWire graph state but is not + /// something *this* instance would ever try to manage itself (the + /// default node.name = "wiremix-capture" filter in + /// config::filter::Filter::defaults() already keeps it out of + /// capturable_objects). + fn add_foreign_capture_node(app: &mut App<'_>, object_id: ObjectId) { + let mut props = PropertyStore::default(); + props.set_node_description(String::from("wiremix-capture")); + props.set_media_class(String::from("Stream/Input/Audio")); + props.set_node_name(String::from("wiremix-capture")); + props.set_object_serial(u32::from(object_id) as u64); + + StateEvent::NodeProperties { object_id, props } + .handle(app) + .unwrap(); + } + + #[test] + fn start_capture_enforces_global_max() { + let commands = RefCell::new(VecDeque::new()); + let wirehose = mock::WirehoseHandle::with_commands(&commands); + let (_, event_rx) = mpsc::channel(); + let config = Config::from_toml_str( + "lazy_capture = false\nmax_concurrent_captures_global = 2", + ); + let mut app = App::new(&wirehose, event_rx, config); + + // Two capture streams already open elsewhere (not ours) - the + // global budget is already fully spent. In the real running + // system, this instance's own started captures would also show up + // here (PipeWire reflects every client's streams to every + // listener, which is exactly why Filter::defaults() has to + // explicitly exclude "wiremix-capture" from capturable_objects - + // self-observation is real), self-consistently shrinking its own + // remaining budget as it captures more. The mock wirehose doesn't + // simulate that round trip, so this test instead fixes the global + // count at the cap via foreign nodes and checks the decline path. + add_foreign_capture_node(&mut app, ObjectId::from_raw_id(100)); + add_foreign_capture_node(&mut app, ObjectId::from_raw_id(101)); + + for i in 1..=3 { + let id = ObjectId::from_raw_id(i); + add_capturable_node(&mut app, id); + app.set_capture_eligibility(CaptureEligibility::Eligible(id)); + } + + // Global budget is already fully spent by the two foreign nodes, + // so none of the 3 locally-eligible nodes should start capturing - + // despite there being no local max_concurrent_captures at all. + assert_eq!(app.capturing_objects.len(), 0); + } + + #[test] + fn global_capture_count_counts_foreign_nodes_only_by_name() { + let commands = RefCell::new(VecDeque::new()); + let wirehose = mock::WirehoseHandle::with_commands(&commands); + let (_, event_rx) = mpsc::channel(); + let config = Config::from_toml_str(""); + let mut app = App::new(&wirehose, event_rx, config); + + assert_eq!(app.global_capture_count(), 0); + + add_foreign_capture_node(&mut app, ObjectId::from_raw_id(100)); + assert_eq!(app.global_capture_count(), 1); + + // A regular, differently-named node shouldn't be counted. + add_capturable_node(&mut app, ObjectId::from_raw_id(101)); + assert_eq!(app.global_capture_count(), 1); + + add_foreign_capture_node(&mut app, ObjectId::from_raw_id(102)); + assert_eq!(app.global_capture_count(), 2); + } + + #[test] + fn rotate_capturing_respects_max() { + let commands = RefCell::new(VecDeque::new()); + let wirehose = mock::WirehoseHandle::with_commands(&commands); + let (_, event_rx) = mpsc::channel(); + let config = Config::from_toml_str( + "lazy_capture = true\nmax_concurrent_captures = 2", + ); + let mut app = App::new(&wirehose, event_rx, config); + + for i in 1..=5 { + let id = ObjectId::from_raw_id(i); + add_capturable_node(&mut app, id); + app.capturable_objects.insert(id); + app.visible_objects.insert(id); + } + + force_rotation_due(&mut app); + app.rotate_capturing(); + + assert_eq!(app.capturing_objects.len(), 2); + } + + #[test] + fn rotate_capturing_advances_window() { + let commands = RefCell::new(VecDeque::new()); + let wirehose = mock::WirehoseHandle::with_commands(&commands); + let (_, event_rx) = mpsc::channel(); + let config = Config::from_toml_str( + "lazy_capture = true\nmax_concurrent_captures = 2", + ); + let mut app = App::new(&wirehose, event_rx, config); + + for i in 1..=5 { + let id = ObjectId::from_raw_id(i); + add_capturable_node(&mut app, id); + app.capturable_objects.insert(id); + app.visible_objects.insert(id); + } + + force_rotation_due(&mut app); + app.rotate_capturing(); + let first_window = app.capturing_objects.clone(); + + force_rotation_due(&mut app); + app.rotate_capturing(); + let second_window = app.capturing_objects.clone(); + + assert_eq!(first_window.len(), 2); + assert_eq!(second_window.len(), 2); + assert_ne!(first_window, second_window); + } + + #[test] + fn rotate_capturing_noop_under_cap() { + let commands = RefCell::new(VecDeque::new()); + let wirehose = mock::WirehoseHandle::with_commands(&commands); + let (_, event_rx) = mpsc::channel(); + let config = Config::from_toml_str( + "lazy_capture = true\nmax_concurrent_captures = 10", + ); + let mut app = App::new(&wirehose, event_rx, config); + + for i in 1..=3 { + let id = ObjectId::from_raw_id(i); + add_capturable_node(&mut app, id); + app.capturable_objects.insert(id); + app.visible_objects.insert(id); + app.set_capture_eligibility(CaptureEligibility::Eligible(id)); + } + + force_rotation_due(&mut app); + app.rotate_capturing(); + + // Fewer eligible nodes than the cap - everything stays captured, + // nothing gets rotated out. + assert_eq!(app.capturing_objects.len(), 3); + } } diff --git a/src/config.rs b/src/config.rs index ac871b0..ce31038 100644 --- a/src/config.rs +++ b/src/config.rs @@ -44,6 +44,8 @@ pub struct Config { pub tab: usize, pub tabs: Vec, pub lazy_capture: bool, + pub max_concurrent_captures: Option, + pub max_concurrent_captures_global: Option, pub filters: Vec, } @@ -88,6 +90,8 @@ struct ConfigFile { tabs: Vec, #[serde(default = "default_lazy_capture")] lazy_capture: bool, + max_concurrent_captures: Option, + max_concurrent_captures_global: Option, #[serde(default = "Filter::defaults", deserialize_with = "Filter::merge")] filters: Vec, } @@ -328,6 +332,17 @@ impl ConfigFile { if opt.lazy_capture { self.lazy_capture = true; } + + if let Some(max_concurrent_captures) = &opt.max_concurrent_captures { + self.max_concurrent_captures = Some(*max_concurrent_captures); + } + + if let Some(max_concurrent_captures_global) = + &opt.max_concurrent_captures_global + { + self.max_concurrent_captures_global = + Some(*max_concurrent_captures_global); + } } } @@ -396,6 +411,9 @@ impl TryFrom for Config { tab, tabs: config_file.tabs, lazy_capture: config_file.lazy_capture, + max_concurrent_captures: config_file.max_concurrent_captures, + max_concurrent_captures_global: config_file + .max_concurrent_captures_global, filters, }) } @@ -481,6 +499,8 @@ pub mod strict { tab: Option, tabs: Vec, lazy_capture: bool, + max_concurrent_captures: Option, + max_concurrent_captures_global: Option, filters: Vec, } @@ -502,6 +522,9 @@ pub mod strict { tab: strict.tab, tabs: strict.tabs, lazy_capture: strict.lazy_capture, + max_concurrent_captures: strict.max_concurrent_captures, + max_concurrent_captures_global: strict + .max_concurrent_captures_global, filters: strict.filters, } } diff --git a/src/opt.rs b/src/opt.rs index 534d130..cb489dc 100644 --- a/src/opt.rs +++ b/src/opt.rs @@ -78,6 +78,21 @@ pub struct Opt { #[clap(long, conflicts_with = "no_lazy_capture")] pub lazy_capture: bool, + /// Cap how many nodes have their peak levels actively monitored at once, + /// rotating which ones are captured every few seconds if more than this + /// many are eligible (further reduces CPU usage on systems with many + /// concurrent streams/devices, at the cost of meters updating less often + /// per node) + #[clap(long, value_name = "COUNT")] + pub max_concurrent_captures: Option, + + /// Cap how many peak-monitoring streams may exist system-wide across + /// all running wiremix instances at once (best-effort - see README for + /// caveats). Composes with --max-concurrent-captures, which caps this + /// instance alone + #[clap(long, value_name = "COUNT")] + pub max_concurrent_captures_global: Option, + #[cfg(debug_assertions)] #[clap(short, long)] pub dump_events: bool, diff --git a/src/wirehose/stream_registry.rs b/src/wirehose/stream_registry.rs index f3fb52b..c52edbe 100644 --- a/src/wirehose/stream_registry.rs +++ b/src/wirehose/stream_registry.rs @@ -60,6 +60,20 @@ impl StreamRegistry { } /// Register a stream and its listener, evicting any with the same ID. + /// + /// If a stream is already registered under `stream_id` (e.g. when a + /// capture is renewed after [`CaptureEligibility::NeedsRestart`]), the + /// evicted stream is explicitly disconnected before being handed to the + /// garbage collector - matching what [`Self::remove()`] already does. + /// Without this, dropping the evicted [`StreamRc`] alone destroys the + /// client-side stream object (see `StreamBox`'s `Drop` impl, which calls + /// `pw_stream_destroy` directly) without ever calling + /// `pw_stream_disconnect`, so the corresponding PipeWire node can be left + /// registered in the graph - orphaned from wiremix's own bookkeeping, + /// but still visible to every other client - until something else tears + /// it down. + /// + /// [`CaptureEligibility::NeedsRestart`]: crate::wirehose::state::CaptureEligibility::NeedsRestart pub fn add_stream( &mut self, stream_id: ObjectId, @@ -67,6 +81,7 @@ impl StreamRegistry { listener: StreamListener, ) { if let Some(old) = self.streams.insert(stream_id, stream) { + let _ = old.disconnect(); self.garbage_streams.push(old); if let Some(listeners) = self.listeners.get_mut(&stream_id) { self.garbage_listeners.append(listeners); diff --git a/wiremix.toml b/wiremix.toml index 218653b..8ffaf11 100644 --- a/wiremix.toml +++ b/wiremix.toml @@ -44,6 +44,32 @@ enforce_max_volume = false # If true, only monitor peak levels of visible nodes lazy_capture = false +# Cap how many nodes have their peak levels actively monitored at once. If +# more nodes than this are eligible for capture (all visible nodes under +# lazy_capture, or every capturable node otherwise), which ones are actually +# captured rotates a step at a time, in sync with the render frame rate, so +# each eventually gets sampled, rather than whichever nodes happened to +# become eligible first holding their slot indefinitely. Meters for nodes +# not currently in the active set keep showing their last captured value +# until their next turn. Unset by default +# (no cap) - each additional concurrent capture is a real PipeWire stream +# the session manager has to track, so on a system with many concurrent +# streams/devices this is the most direct way to bound that cost. +#max_concurrent_captures = 8 + +# Like max_concurrent_captures, but counts (and caps) "wiremix-capture" +# nodes across the whole PipeWire graph, not just this instance's own - +# i.e. a total shared across every running wiremix instance that also has +# this option set, rather than a per-instance limit. Best-effort: with no +# cross-process locking, instances racing to start a capture at the same +# moment can transiently exceed this before the graph state settles, and +# it can't make non-cooperating processes (upstream wiremix, or anything +# else sharing the "wiremix-capture" node name) back off - only instances +# that also set this option consider each other. Unset by default (no +# cap). Composes with max_concurrent_captures above - both are checked +# independently, so whichever is more restrictive applies. +#max_concurrent_captures_global = 8 + # Keybindings #