From c526017f2fa3760831f48360b4adec84a0813dae Mon Sep 17 00:00:00 2001 From: HoneyHazard <8847050+HoneyHazard@users.noreply.github.com> Date: Wed, 5 Aug 2026 01:45:40 -0700 Subject: [PATCH 1/4] Add max_concurrent_captures to cap and rotate simultaneous peak captures Each node wiremix monitors for peak levels gets its own dedicated PipeWire capture stream (wirehose/stream.rs::capture_node). Every one of those is a real client object the session manager has to track and policy-link, and that cost scales with how many exist at once - not with anything CPU-throttleable, since it's driven by stream count, not per-quantum processing. lazy_capture already limits this to on-screen nodes, but on views where many nodes are visible simultaneously (a busy Output Devices tab, a tall terminal), that alone doesn't bound the concurrent stream count. Adds max_concurrent_captures: Option (unset = current unbounded behavior). When set and more nodes are eligible for capture than the cap allows, which ones are actually captured rotates on a fixed 3s interval (deliberately much slower than render cadence - rotating every frame would create more stream churn than not capping at all) so every eligible node eventually gets sampled rather than whichever ones happened to become eligible first holding their slot indefinitely. Meters for nodes outside the active window keep showing their last captured value until their next turn, rather than resetting to zero. The cap is enforced as a hard invariant directly in start_capture() (not just in the rotation logic), so it can never be transiently exceeded even if several nodes become eligible at once before the next rotation tick runs. Verified live against the real PipeWire graph (not just unit tests): with --max-concurrent-captures 2, `pw-dump` showed exactly 2 wiremix-capture streams at any moment, and the actual target node IDs fully changed after the 3s interval elapsed - confirming both the cap and the rotation are real, not just passing in isolation. Tested: cargo test (148/148, including 4 new tests covering the cap being enforced by start_capture, rotation respecting the cap, the active window actually advancing between rotations, and the no-op case where fewer nodes are eligible than the cap), cargo fmt --check / cargo clippy -- -D warnings / cargo doc (matching wiremix's CI) all clean. --- src/app.rs | 210 ++++++++++++++++++++++++++++++++++++++++++++++++++ src/config.rs | 9 +++ src/opt.rs | 8 ++ wiremix.toml | 12 +++ 4 files changed, 239 insertions(+) diff --git a/src/app.rs b/src/app.rs index 6c973d2..522729a 100644 --- a/src/app.rs +++ b/src/app.rs @@ -214,6 +214,12 @@ 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, + /// When the capture rotation window last advanced. + last_capture_rotation: Instant, } macro_rules! current_list { @@ -261,6 +267,8 @@ impl<'a> App<'a> { peak_processor: Arc::new(peak_processor), capturable_objects: HashSet::new(), capturing_objects: HashSet::new(), + capture_rotation_start: 0, + last_capture_rotation: Instant::now(), } } @@ -301,6 +309,12 @@ impl<'a> App<'a> { self.update_capturing(); } + // Runs every iteration (not just on visibility change) since + // rotation is time-driven, not scroll-driven. Cheap to call when + // there's nothing to do - an elapsed-time check and, usually, an + // early return. + self.rotate_capturing(); + if needs_render && pacer.is_time_to_render() { needs_render = false; @@ -353,6 +367,18 @@ 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; + } + } + let Some(node) = self.state.nodes.get(&object_id) else { return; }; @@ -417,6 +443,81 @@ impl<'a> App<'a> { } } + /// How often to swap which nodes are actively captured when + /// max_concurrent_captures limits capture to fewer than are eligible. + /// Deliberately much slower than typical UI/render cadence - rotating on + /// every frame would create more PipeWire stream churn than not + /// rotating at all, defeating the purpose of capping concurrency in the + /// first place. + const CAPTURE_ROTATION_INTERVAL: Duration = Duration::from_secs(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). + fn rotate_capturing(&mut self) { + let Some(max) = self.config.max_concurrent_captures else { + return; + }; + + if self.last_capture_rotation.elapsed() + < Self::CAPTURE_ROTATION_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 last_capture_rotation alone so a rotation isn't + // "owed" the moment the eligible set grows past max again. + return; + } + + self.last_capture_rotation = Instant::now(); + + // 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 + max) % 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 +983,7 @@ mod tests { tabs: vec![TabKind::Playback], lazy_capture: Default::default(), filters: Default::default(), + max_concurrent_captures: Default::default(), }; let mut app = App::new(wirehose, event_rx, config); @@ -983,6 +1085,7 @@ mod tests { ], lazy_capture: Default::default(), filters: Default::default(), + max_concurrent_captures: Default::default(), }; let mut app = App::new(&wirehose, event_rx, config); @@ -1271,4 +1374,111 @@ mod tests { Some(mock::MockCommand::NodeCaptureStop(id)) ); } + + /// Back-dates last_capture_rotation so the next rotate_capturing() call + /// doesn't have to wait out the real interval. + fn force_rotation_due(app: &mut App<'_>) { + app.last_capture_rotation = std::time::Instant::now() + .checked_sub(App::CAPTURE_ROTATION_INTERVAL * 2) + .unwrap(); + } + + #[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); + } + + #[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..a1ae269 100644 --- a/src/config.rs +++ b/src/config.rs @@ -44,6 +44,7 @@ pub struct Config { pub tab: usize, pub tabs: Vec, pub lazy_capture: bool, + pub max_concurrent_captures: Option, pub filters: Vec, } @@ -88,6 +89,7 @@ struct ConfigFile { tabs: Vec, #[serde(default = "default_lazy_capture")] lazy_capture: bool, + max_concurrent_captures: Option, #[serde(default = "Filter::defaults", deserialize_with = "Filter::merge")] filters: Vec, } @@ -328,6 +330,10 @@ 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); + } } } @@ -396,6 +402,7 @@ impl TryFrom for Config { tab, tabs: config_file.tabs, lazy_capture: config_file.lazy_capture, + max_concurrent_captures: config_file.max_concurrent_captures, filters, }) } @@ -481,6 +488,7 @@ pub mod strict { tab: Option, tabs: Vec, lazy_capture: bool, + max_concurrent_captures: Option, filters: Vec, } @@ -502,6 +510,7 @@ pub mod strict { tab: strict.tab, tabs: strict.tabs, lazy_capture: strict.lazy_capture, + max_concurrent_captures: strict.max_concurrent_captures, filters: strict.filters, } } diff --git a/src/opt.rs b/src/opt.rs index 534d130..a5212c0 100644 --- a/src/opt.rs +++ b/src/opt.rs @@ -78,6 +78,14 @@ 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, + #[cfg(debug_assertions)] #[clap(short, long)] pub dump_events: bool, diff --git a/wiremix.toml b/wiremix.toml index 218653b..3dae2ab 100644 --- a/wiremix.toml +++ b/wiremix.toml @@ -44,6 +44,18 @@ 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 every few seconds 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 + # Keybindings # From 3363932e3accf92c51bf69a2d6bd168d143de404 Mon Sep 17 00:00:00 2001 From: HoneyHazard <8847050+HoneyHazard@users.noreply.github.com> Date: Thu, 6 Aug 2026 16:07:39 -0700 Subject: [PATCH 2/4] Disconnect evicted capture streams before dropping them StreamRegistry::add_stream() evicted an existing stream from its map without calling disconnect() on it first, unlike remove(), which already did. Dropping the evicted StreamRc alone destroys the client-side stream object (pw_stream_destroy) without ever calling pw_stream_disconnect, so the corresponding PipeWire node can be left registered in the graph until something else tears it down. This path is hit every time a capture is renewed after CaptureEligibility::NeedsRestart, since that calls start_capture() again for an object_id that already has an active stream. --- src/wirehose/stream_registry.rs | 15 +++++++++++++++ 1 file changed, 15 insertions(+) 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); From cbb7f35d8f0e0487002baf46cf1c42d8baa71d9e Mon Sep 17 00:00:00 2001 From: HoneyHazard <8847050+HoneyHazard@users.noreply.github.com> Date: Thu, 6 Aug 2026 16:24:50 -0700 Subject: [PATCH 3/4] Add max_concurrent_captures_global to cap peak captures across all instances Extends max_concurrent_captures with a second, optional cap that counts "wiremix-capture" nodes across the whole PipeWire graph rather than just this instance's own captures - a best-effort budget shared by every running wiremix instance that also sets this option. The naive approach of gating solely on the live graph reading undercounts this instance's own just-issued captures: node_capture_start() is fire-and-forget over an async channel to a separate PipeWire thread, and handle_events() drains a whole burst of eligibility events synchronously before any of that burst's own captures can round-trip back into local state. Combining the graph reading with capturing_objects.len() (always exact and lag-free, since it updates the instant this instance decides to capture something) closes that gap without changing the inherent, documented cross-instance best-effort behavior. Depends on the capture-stream disconnect fix (separate PR) - without it, repeated CaptureEligibility::NeedsRestart churn can leave orphaned capture nodes in the graph that inflate what every instance perceives as the current global count. --- src/app.rs | 151 ++++++++++++++++++++++++++++++++++++++++++++++++++ src/config.rs | 14 +++++ src/opt.rs | 7 +++ wiremix.toml | 13 +++++ 4 files changed, 185 insertions(+) diff --git a/src/app.rs b/src/app.rs index 522729a..5399e54 100644 --- a/src/app.rs +++ b/src/app.rs @@ -360,6 +360,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) @@ -379,6 +401,58 @@ impl<'a> App<'a> { } } + // 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; }; @@ -984,6 +1058,7 @@ 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); @@ -1086,6 +1161,7 @@ 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); @@ -1403,6 +1479,81 @@ mod tests { 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()); diff --git a/src/config.rs b/src/config.rs index a1ae269..ce31038 100644 --- a/src/config.rs +++ b/src/config.rs @@ -45,6 +45,7 @@ pub struct Config { pub tabs: Vec, pub lazy_capture: bool, pub max_concurrent_captures: Option, + pub max_concurrent_captures_global: Option, pub filters: Vec, } @@ -90,6 +91,7 @@ struct ConfigFile { #[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, } @@ -334,6 +336,13 @@ impl ConfigFile { 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); + } } } @@ -403,6 +412,8 @@ impl TryFrom for Config { 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, }) } @@ -489,6 +500,7 @@ pub mod strict { tabs: Vec, lazy_capture: bool, max_concurrent_captures: Option, + max_concurrent_captures_global: Option, filters: Vec, } @@ -511,6 +523,8 @@ pub mod strict { 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 a5212c0..cb489dc 100644 --- a/src/opt.rs +++ b/src/opt.rs @@ -86,6 +86,13 @@ pub struct Opt { #[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/wiremix.toml b/wiremix.toml index 3dae2ab..340aa2a 100644 --- a/wiremix.toml +++ b/wiremix.toml @@ -56,6 +56,19 @@ lazy_capture = false # 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 # From 1ec4dfba3ccfce712962e15fbc65ee051234bb88 Mon Sep 17 00:00:00 2001 From: HoneyHazard <8847050+HoneyHazard@users.noreply.github.com> Date: Fri, 7 Aug 2026 14:56:50 -0700 Subject: [PATCH 4/4] Rotate capture window one object at a time, tied to render cadence The previous design jumped the entire max-sized capture window forward every fixed 3-second wall-clock interval: every currently-captured object went stale for the full interval, then all of them changed at once. That felt frozen, then jumpy, rather than real-time - and the 3-second constant had no relationship to fps or eligible-object count, so coverage latency could be far worse than the interval itself suggested (a fixed max per rotation, however many objects are waiting). Replaced with two changes: - Slide the window by exactly one object per tick instead of jumping it by the full window size, so at most one capture starts and one stops per tick and everything else is left alone - a continuous trickle instead of a batch swap. - Tie the tick itself to the render loop's own frame cadence (every Nth actually-rendered frame) instead of an independent wall-clock Duration, so rotation speed - and the PipeWire stream churn it costs - scales automatically with whatever fps a user has configured, with no second unrelated timer to reason about. rotate_capturing() is now only ever called in step with a rendered frame (moved its call site accordingly), so counting frames is exact. Verified live against the real PipeWire graph: sampled which specific objects a running instance was capturing every ~300ms and confirmed the pair slides by one each time rather than jumping to an unrelated pair. --- src/app.rs | 75 +++++++++++++++++++++++++++++++++------------------- wiremix.toml | 9 ++++--- 2 files changed, 53 insertions(+), 31 deletions(-) diff --git a/src/app.rs b/src/app.rs index 5399e54..e891033 100644 --- a/src/app.rs +++ b/src/app.rs @@ -218,8 +218,10 @@ pub struct App<'a> { /// rotation window should start. Only meaningful when /// config.max_concurrent_captures is set. capture_rotation_start: usize, - /// When the capture rotation window last advanced. - last_capture_rotation: Instant, + /// 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 { @@ -268,7 +270,7 @@ impl<'a> App<'a> { capturable_objects: HashSet::new(), capturing_objects: HashSet::new(), capture_rotation_start: 0, - last_capture_rotation: Instant::now(), + frames_since_rotation: 0, } } @@ -309,15 +311,16 @@ impl<'a> App<'a> { self.update_capturing(); } - // Runs every iteration (not just on visibility change) since - // rotation is time-driven, not scroll-driven. Cheap to call when - // there's nothing to do - an elapsed-time check and, usually, an - // early return. - self.rotate_capturing(); - 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| { @@ -517,27 +520,46 @@ impl<'a> App<'a> { } } - /// How often to swap which nodes are actively captured when + /// How many rendered frames pass between capture rotation slides, when /// max_concurrent_captures limits capture to fewer than are eligible. - /// Deliberately much slower than typical UI/render cadence - rotating on - /// every frame would create more PipeWire stream churn than not - /// rotating at all, defeating the purpose of capping concurrency in the - /// first place. - const CAPTURE_ROTATION_INTERVAL: Duration = Duration::from_secs(3); + /// 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; }; - if self.last_capture_rotation.elapsed() - < Self::CAPTURE_ROTATION_INTERVAL - { + self.frames_since_rotation = + self.frames_since_rotation.saturating_add(1); + if self.frames_since_rotation < Self::ROTATION_FRAME_INTERVAL { return; } @@ -555,12 +577,13 @@ impl<'a> App<'a> { if eligible.len() <= max { // Everything eligible already fits under the cap - nothing to - // rotate. Leave last_capture_rotation alone so a rotation isn't - // "owed" the moment the eligible set grows past max again. + // 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.last_capture_rotation = Instant::now(); + self.frames_since_rotation = 0; // Sorting gives a stable, deterministic rotation order across ticks // (HashSet iteration order isn't stable) so each tick advances @@ -572,7 +595,7 @@ impl<'a> App<'a> { 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 + max) % n; + self.capture_rotation_start = (start + 1) % n; let need_to_stop: Vec<_> = self .capturing_objects @@ -1451,12 +1474,10 @@ mod tests { ); } - /// Back-dates last_capture_rotation so the next rotate_capturing() call - /// doesn't have to wait out the real interval. + /// 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.last_capture_rotation = std::time::Instant::now() - .checked_sub(App::CAPTURE_ROTATION_INTERVAL * 2) - .unwrap(); + app.frames_since_rotation = App::ROTATION_FRAME_INTERVAL; } #[test] diff --git a/wiremix.toml b/wiremix.toml index 340aa2a..8ffaf11 100644 --- a/wiremix.toml +++ b/wiremix.toml @@ -47,10 +47,11 @@ 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 every few seconds 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 +# 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.