Overlord/vmic
1
1
Fork 0
Code Releases 1 Activity

Add support for --loopback-no-mix flag and topology synchronization

Browse Source 
- Introduced the `--loopback-no-mix` flag to control 2-node vs 4-node loopback topologies.
- Updated topology handling logic to recompute and migrate as needed based on loopback flags.
- Enhanced database schema to include `loopback_no_mix` with automatic migration for backward compatibility.
- Refactored `create`, `edit`, and `route` commands to integrate new topology management capabilities.
- Improved `list` command output to include architecture details.
This commit is contained in:
Overlord
2026-08-11 15:33:36 +02:00
parent be70c55728
commit 493f91bc28
10 changed files with 308 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/cli.rs
View File

@@ -74,6 +74,11 @@ pub struct EditArgs {
#[arg(short, long, value_name = "BOOL")]
pub loopback: Option<bool>,
/// Keep a mixed-in source audible in the loopback instead of isolating
/// it (2-node topology instead of 4).
#[arg(long = "loopback-no-mix", value_name = "BOOL")]
pub loopback_no_mix: Option<bool>,
/// Loopback volume: fraction (0.8) or percent (80).
#[arg(short, long, value_name = "PCT")]
pub volume: Option<f32>,

13
src/commands/create.rs
View File

@@ -21,7 +21,7 @@ pub fn run(args: CreateArgs) -> Result<()> {
let orig_sink = default_output("get-default-sink");
let orig_source = default_output("get-default-source");
let stages = loopback::create_stages(&graph, &names)?;
let stage = loopback::create_simple_stage(&graph, &names, &[], &[])?;
if let Some(sink) = &orig_sink {
let _ = Command::new("pactl").args(["set-default-sink", sink]).status();
@@ -38,16 +38,19 @@ pub fn run(args: CreateArgs) -> Result<()> {
ui::warn(&format!("removed {removed} unexpected monitor feedback link(s)."));
}
// A fresh vmic always starts as Simple2Node - no source can be mixed in
// yet, so there's nothing for a 4-node split to isolate (see
// commands::topology). mid_name/mix_name are still stored even though no
// live node backs them yet: cheap, and needed if this vmic later
// upgrades. mix_node_id/mix_pid stay at their Default (None).
let mut state = VmicState {
name: name.clone(),
sink_name: names.sink.clone(),
mid_name: names.mid,
mix_name: names.mix,
source_name: names.source.clone(),
sink_node_id: Some(stages.sink_node_id),
mix_node_id: Some(stages.mix_node_id),
sink_pid: Some(stages.sink_pid as i64),
mix_pid: Some(stages.mix_pid as i64),
sink_node_id: Some(stage.sink_node_id),
sink_pid: Some(stage.pid as i64),
..Default::default()
};

45
src/commands/edit.rs
View File

@@ -1,14 +1,19 @@
use crate::cli::EditArgs;
use crate::commands::to_pct;
use crate::commands::topology;
use crate::error::{Result, VmicError};
use crate::pw::{loopback, PwGraph};
use crate::state::{self, VmicState};
use crate::ui;
pub fn run(args: EditArgs) -> Result<()> {
if args.loopback.is_none() && args.volume.is_none() && args.source_volume.is_none() {
if args.loopback.is_none()
&& args.loopback_no_mix.is_none()
&& args.volume.is_none()
&& args.source_volume.is_none()
{
return Err(VmicError::NothingToDo(
"pass --loopback, --volume and/or --source-volume.".into(),
"pass --loopback, --loopback-no-mix, --volume and/or --source-volume.".into(),
));
}
@@ -21,6 +26,9 @@ pub fn run(args: EditArgs) -> Result<()> {
if let Some(enable) = args.loopback {
edit_loopback(&graph, &mut state, enable)?;
}
if let Some(no_mix) = args.loopback_no_mix {
edit_loopback_no_mix(&mut state, no_mix);
}
if let Some(volume) = args.volume {
edit_volume(&graph, &mut state, volume)?;
}
@@ -28,6 +36,12 @@ pub fn run(args: EditArgs) -> Result<()> {
edit_source_volume(&mut state, source_volume)?;
}
// Only --loopback/--loopback-no-mix can change the desired topology -
// recompute it and migrate if needed.
if args.loopback.is_some() || args.loopback_no_mix.is_some() {
topology::sync_topology(&graph, &mut state)?;
}
state.save(&conn)
}
@@ -40,7 +54,10 @@ fn warn_if_stale(graph: &PwGraph, state: &VmicState) {
"vmic '{}' appears stale (matching pw-loopback process not found); changes may not apply!",
state.name
));
} else if !alive(state.mix_node_id) {
} else if state.mix_pid.is_some() && !alive(state.mix_node_id) {
// A missing mix stage is only suspicious when one is supposed to
// exist (Pure4Node topology, see commands::topology) - a
// Simple2Node vmic never has one.
ui::warn(&format!(
"vmic '{}' mix loopback process not found; -sv changes may not apply!",
state.name
@@ -86,6 +103,28 @@ fn edit_loopback(graph: &PwGraph, state: &mut VmicState, enable: bool) -> Result
Ok(())
}
fn edit_loopback_no_mix(state: &mut VmicState, no_mix: bool) {
if state.loopback_no_mix == no_mix {
ui::warn(&format!(
"--loopback-no-mix is already {no_mix} for '{}'.",
state.name
));
return;
}
state.loopback_no_mix = no_mix;
if no_mix {
ui::ok(&format!(
"'{}' will keep a mixed-in source in the self-monitor loopback instead of isolating it.",
state.name
));
} else {
ui::ok(&format!(
"'{}' will isolate a mixed-in source from the self-monitor loopback again.",
state.name
));
}
}
fn edit_volume(graph: &PwGraph, state: &mut VmicState, value: f32) -> Result<()> {
let pct = to_pct(value).ok_or_else(|| VmicError::InvalidVolume(value.to_string()))?;
state.volume_pct = Some(pct);

10
src/commands/list.rs
View File

@@ -1,3 +1,4 @@
use crate::commands::topology;
use crate::error::Result;
use crate::pw::PwGraph;
use crate::state::{self, VmicState};
@@ -21,6 +22,7 @@ pub fn run() -> Result<()> {
println!(" output device: {}", state.sink_name);
println!(" mic device: {}", state.source_name);
println!(" status: {status_colored}");
println!(" architecture: {}", topology::architecture_label(state));
println!(
" self-monitor loopback: {}",
state
@@ -46,12 +48,20 @@ pub fn run() -> Result<()> {
}
/// Checks `sink_node_id`/`mix_node_id` against the live graph snapshot.
/// `mix_node_id` is only expected to be alive when this vmic is currently
/// running as `Pure4Node` (`mix_pid.is_some()`) - a `Simple2Node` vmic never
/// has one, and that's not a problem.
fn liveness(graph: Option<&PwGraph>, state: &VmicState) -> String {
let Some(graph) = graph else {
return "unknown (could not connect to PipeWire)".to_string();
};
let nodes = graph.nodes();
let sink_alive = state.sink_node_id.is_some_and(|id| nodes.iter().any(|n| n.id == id));
if topology::current_topology(state) == topology::Topology::Simple2Node {
return if sink_alive { "active".to_string() } else { "stale (matching pw-loopback process not found)".to_string() };
}
let mix_alive = state.mix_node_id.is_some_and(|id| nodes.iter().any(|n| n.id == id));
match (sink_alive, mix_alive) {
(true, true) => "active".to_string(),

7
src/commands/mod.rs
View File

@@ -3,6 +3,7 @@ pub mod delete;
pub mod edit;
pub mod list;
pub mod route;
pub mod topology;
pub mod wipe;
use crate::error::Result;
@@ -30,6 +31,10 @@ pub(crate) fn unsource(graph: &PwGraph, state: &mut VmicState) -> Result<()> {
let Some(mic) = state.mic_source.take() else {
return Ok(());
};
loopback::unlink_stage(graph, &mic, &state.mix_name)?;
// Target depends on the current topology: a Simple2Node vmic mixes the
// source straight into its sink, not a (nonexistent) mix stage.
let current = topology::current_topology(state);
let target = topology::mic_target(state, current).to_string();
loopback::unlink_stage(graph, &mic, &target)?;
Ok(())
}

32
src/commands/route.rs
View File

@@ -1,13 +1,9 @@
use crate::cli::RouteArgs;
use crate::commands::topology::{self, Topology};
use crate::error::{Result, VmicError};
use crate::pw::{loopback, text, PwGraph};
use crate::pw::{text, PwGraph};
use crate::state::{self, VmicState};
use crate::ui;
use std::time::Duration;
/// Shorter than stage-creation's timeout since the source already exists -
/// this is just retrying while its ports enumerate.
const SOURCE_LINK_TIMEOUT: Duration = Duration::from_secs(2);
pub fn run(args: RouteArgs) -> Result<()> {
let conn = state::open_db()?;
@@ -69,6 +65,10 @@ fn move_stream(kind: &str, move_cmd: &str, target: &str, filter: &str) -> Result
Ok(())
}
/// Sets or clears the mixed-in hardware source, then hands off to
/// `topology::sync_topology` to (re)link it and, if the loopback/no-mix
/// flags call for it, upgrade or downgrade between the 2-node and 4-node
/// shapes.
fn route_source(graph: &PwGraph, state: &mut VmicState, value: &str) -> Result<()> {
let lowered = value.to_lowercase();
if matches!(lowered.as_str(), "off" | "none" | "disable") {
@@ -77,6 +77,7 @@ fn route_source(graph: &PwGraph, state: &mut VmicState, value: &str) -> Result<(
return Ok(());
};
super::unsource(graph, state)?;
topology::sync_topology(graph, state)?;
ui::ok(&format!("Removed source mix '{mic}' from '{}'.", state.name));
return Ok(());
}
@@ -100,19 +101,16 @@ fn route_source(graph: &PwGraph, state: &mut VmicState, value: &str) -> Result<(
super::unsource(graph, state)?;
}
let linked = loopback::link_stage(graph, &mic, &state.mix_name, SOURCE_LINK_TIMEOUT)?;
state.mic_source = Some(mic.clone());
topology::sync_topology(graph, state)?;
// Insurance: make sure the new links did not create a monitor ring.
let prefix = format!("vmic_{}_", state.name.to_lowercase());
let rings = graph.break_feedback_rings(&prefix)?;
if rings > 0 {
ui::warn(&format!("removed {rings} unexpected monitor feedback link(s)."));
let note = match topology::current_topology(state) {
Topology::Pure4Node => "isolated mix stage; not audible in self-monitor",
Topology::Simple2Node if state.loopback_id.is_some() => {
"sink; also audible in self-monitor - see --loopback-no-mix"
}
ui::ok(&format!(
"Mixed '{mic}' into the mix stage of '{}' ({linked} link(s); not audible in self-monitor).",
state.name
));
Topology::Simple2Node => "sink",
};
ui::ok(&format!("Mixed '{mic}' into '{}' ({note}).", state.name));
Ok(())
}

16
src/pw/graph.rs
View File

@@ -160,6 +160,11 @@ impl PwGraph {
self.state.borrow().nodes.values().cloned().collect()
}
/// Must not be called on a node name that may transiently have two live
/// nodes (e.g. mid-migration in `commands::topology`, where an old and a
/// new node can briefly share a name) - it has no way to disambiguate
/// them. Callers in that situation must resolve the specific node id
/// first (see `wait_for_new_node`) and work with ids from then on.
pub fn ports_for(&self, node_name: &str) -> Vec<Port> {
let s = self.state.borrow();
let Some(node) = s.nodes.values().find(|n| n.name == node_name) else {
@@ -287,9 +292,18 @@ impl PwGraph {
/// Waits (bounded) for a node to appear in the graph.
pub fn wait_for_node(&self, name: &str, timeout: Duration) -> Result<Node> {
self.wait_for_new_node(name, &[], timeout)
}
/// Like `wait_for_node`, but ignores any node whose id is in `excluding`.
/// Needed during a topology migration (`commands::topology`), where a
/// freshly spawned node can transiently share a `node.name` with the
/// old one it's about to replace - a plain name match could otherwise
/// resolve to the dying node instead of the new one.
pub fn wait_for_new_node(&self, name: &str, excluding: &[u32], timeout: Duration) -> Result<Node> {
let deadline = Instant::now() + timeout;
loop {
if let Some(n) = self.nodes().into_iter().find(|n| n.name == name) {
if let Some(n) = self.nodes().into_iter().find(|n| n.name == name && !excluding.contains(&n.id)) {
return Ok(n);
}
if Instant::now() >= deadline {

100
src/pw/loopback.rs
View File

@@ -51,20 +51,100 @@ pub struct StageHandles {
pub mix_node_id: u32,
}
/// Creates the two chained loopback stages as detached `pw-loopback`
/// subprocesses (stdin is `/dev/null`, never waited on, so the OS
/// reparents them to init on exit), then links stage 1 into stage 2
/// (mid -> mix).
///
/// apps -> sink =(stage1)=> mid -> mix =(stage2)=> source -> recorders
/// ^ hw mic joins here (route --source)
pub fn create_stages(graph: &PwGraph, names: &NodeNames) -> Result<StageHandles> {
let fmt = format!("audio.rate={SAMPLE_RATE} audio.position=[FL FR]");
/// OS pid and sink PipeWire node id produced by [`create_simple_stage`].
pub struct SingleStageHandles {
pub pid: u32,
pub sink_node_id: u32,
}
fn log_file_for(names: &NodeNames) -> Result<(std::path::PathBuf, std::fs::File)> {
let log_path = std::env::temp_dir().join(format!(
"vmic_{}.log",
names.sink.trim_start_matches("vmic_").trim_end_matches("_sink")
));
let log_file = std::fs::OpenOptions::new().create(true).append(true).open(&log_path)?;
Ok((log_path, log_file))
}
/// Creates the baseline single-stage loopback: apps -> sink =(pw-loopback)=>
/// source_name -> recorders. Used whenever a 4-node split isn't needed (see
/// `commands::topology::desired_topology`) - the common case, and the reason
/// a fresh vmic only ever costs one process, not two.
///
/// `exclude_sink_ids`/`exclude_source_ids` matter only when this is called as
/// part of a topology migration (`commands::topology::migrate`), where a
/// same-named node from the topology being replaced may still be alive at
/// spawn time; see `PwGraph::wait_for_new_node`. Pass empty slices for a
/// brand-new vmic.
pub fn create_simple_stage(
graph: &PwGraph,
names: &NodeNames,
exclude_sink_ids: &[u32],
exclude_source_ids: &[u32],
) -> Result<SingleStageHandles> {
let fmt = format!("audio.rate={SAMPLE_RATE} audio.position=[FL FR]");
let (log_path, log_file) = log_file_for(names)?;
let mut child = spawn_loopback_stage(
&format!(
"media.class=Audio/Sink node.name={} node.description={} node.virtual=false {fmt}",
names.sink, names.sink_desc
),
&format!(
"media.class=Audio/Source node.name={} node.description={} node.virtual=false {fmt}",
names.source, names.source_desc
),
&log_file,
)?;
// Give the process a moment to fail fast (e.g. a name collision) before
// trusting it.
std::thread::sleep(STAGE_STARTUP_CHECK);
if child.try_wait()?.is_some() {
let _ = child.kill();
return Err(VmicError::PipeWire(format!(
"pw-loopback exited immediately, check '{}'.",
log_path.display()
)));
}
let pid = child.id();
// Unlike create_stages, there's no link_stage retry loop to implicitly
// wait through - both nodes must be waited for explicitly.
let sink_node_id = (|| -> Result<u32> {
let sink_node = graph.wait_for_new_node(&names.sink, exclude_sink_ids, NODE_WAIT_TIMEOUT)?;
graph.wait_for_new_node(&names.source, exclude_source_ids, NODE_WAIT_TIMEOUT)?;
Ok(sink_node.id)
})();
let sink_node_id = match sink_node_id {
Ok(id) => id,
Err(e) => {
let _ = child.kill();
return Err(e);
}
};
Ok(SingleStageHandles { pid, sink_node_id })
}
/// Creates the two chained loopback stages as detached `pw-loopback`
/// subprocesses (stdin is `/dev/null`, never waited on, so the OS
/// reparents them to init on exit), then links stage 1 into stage 2
/// (mid -> mix). Used only when a mixed-in source needs to be kept out of
/// the self-monitor loopback (see `commands::topology`).
///
/// apps -> sink =(stage1)=> mid -> mix =(stage2)=> source -> recorders
/// ^ hw mic joins here (route --source)
///
/// `exclude_sink_ids` is the same migration-safety mechanism as
/// `create_simple_stage`'s - `mid`/`mix` never need it, since those names are
/// exclusive to this topology and can't already exist across a migration
/// boundary.
pub fn create_stages(graph: &PwGraph, names: &NodeNames, exclude_sink_ids: &[u32]) -> Result<StageHandles> {
let fmt = format!("audio.rate={SAMPLE_RATE} audio.position=[FL FR]");
let (log_path, log_file) = log_file_for(names)?;
let mut sink_child = spawn_loopback_stage(
&format!(
@@ -106,7 +186,7 @@ pub fn create_stages(graph: &PwGraph, names: &NodeNames) -> Result<StageHandles>
let mix_pid = mix_child.id();
let stage_ids = (|| -> Result<(u32, u32)> {
let sink_node = graph.wait_for_node(&names.sink, NODE_WAIT_TIMEOUT)?;
let sink_node = graph.wait_for_new_node(&names.sink, exclude_sink_ids, NODE_WAIT_TIMEOUT)?;
let mix_node = graph.wait_for_node(&names.mix, NODE_WAIT_TIMEOUT)?;
link_stage(graph, &names.mid, &names.mix, LINK_RETRY_TIMEOUT)?;
Ok((sink_node.id, mix_node.id))

83
src/pw/text.rs
View File

@@ -188,6 +188,64 @@ fn parse_sink_input_for_module(text: &str, module_id: u32) -> Option<u32> {
None
}
/// Numeric pulse-side id for `name` within `kind` ("sinks" or "sources"),
/// via `pactl list short <kind>` - same id/name column layout used by
/// `match_source`'s numeric branch and `wipe::first_non_vmic`.
fn resolve_pulse_id(kind: &str, name: &str) -> Result<Option<u32>> {
let output = Command::new("pactl").args(["list", "short", kind]).output()?;
Ok(String::from_utf8_lossy(&output.stdout)
.lines()
.find_map(|l| {
let mut cols = l.split_whitespace();
let id = cols.next()?.parse::<u32>().ok()?;
(cols.next()? == name).then_some(id)
}))
}
/// Ids of every sink-input currently targeting `sink_name`. Used by
/// `commands::topology::migrate` to snapshot stream routing before a
/// migration recreates the sink node, so they can be moved back afterward.
pub fn sink_inputs_on(sink_name: &str) -> Result<Vec<u32>> {
let Some(id) = resolve_pulse_id("sinks", sink_name)? else {
return Ok(Vec::new());
};
let output = Command::new("pactl").args(["list", "sink-inputs"]).output()?;
Ok(parse_ids_matching(&String::from_utf8_lossy(&output.stdout), "Sink Input #", "Sink: ", id))
}
/// Ids of every source-output currently reading from `source_name`. See
/// `sink_inputs_on`.
pub fn source_outputs_on(source_name: &str) -> Result<Vec<u32>> {
let Some(id) = resolve_pulse_id("sources", source_name)? else {
return Ok(Vec::new());
};
let output = Command::new("pactl").args(["list", "source-outputs"]).output()?;
Ok(parse_ids_matching(&String::from_utf8_lossy(&output.stdout), "Source Output #", "Source: ", id))
}
/// Ids of every block (headed by `header_prefix<id>`) whose `key_prefix<id>`
/// property line equals `target_id`.
fn parse_ids_matching(text: &str, header_prefix: &str, key_prefix: &str, target_id: u32) -> Vec<u32> {
let mut ids = Vec::new();
let mut current_id: Option<u32> = None;
for line in text.lines() {
if let Some(rest) = line.strip_prefix(header_prefix) {
current_id = rest.trim().parse().ok();
continue;
}
let trimmed = line.trim_start();
if let Some(rest) = trimmed.strip_prefix(key_prefix) {
if rest.trim().parse::<u32>().ok() == Some(target_id) {
if let Some(id) = current_id {
ids.push(id);
}
}
}
}
ids
}
/// Every `module-loopback` whose `Argument:` line mentions a `vmic_` node,
/// tracked in state or not.
pub fn list_vmic_loopback_module_ids() -> Result<Vec<u32>> {
@@ -238,6 +296,31 @@ Sink Input #12
assert_eq!(parse_sink_input_for_module(text, 99), None);
}
#[test]
fn finds_sink_inputs_targeting_sink() {
let text = "\
Sink Input #10
\tSink: 65
Sink Input #11
\tSink: 99
Sink Input #12
\tSink: 65
";
assert_eq!(parse_ids_matching(text, "Sink Input #", "Sink: ", 65), vec![10, 12]);
assert_eq!(parse_ids_matching(text, "Sink Input #", "Sink: ", 1), Vec::<u32>::new());
}
#[test]
fn finds_source_outputs_targeting_source() {
let text = "\
Source Output #20
\tSource: 3
Source Output #21
\tSource: 4
";
assert_eq!(parse_ids_matching(text, "Source Output #", "Source: ", 3), vec![20]);
}
#[test]
fn lists_only_vmic_loopback_modules() {
let text = "\

37
src/state.rs
View File

@@ -29,6 +29,11 @@ pub struct VmicState {
// pitfalls here, unlike PipeWire's JSON dump (see pw::text).
pub mic_source: Option<String>,
pub mic_volume_pct: Option<u8>,
// Opt-in: keep the 2-node topology even when a source is mixed in while
// the self-monitor loopback is on, accepting that the mix becomes
// audible in the loopback instead of paying for a 4-node split.
pub loopback_no_mix: bool,
}
fn db_path() -> PathBuf {
@@ -90,12 +95,33 @@ pub fn open_db() -> Result<Connection> {
loopback_id INTEGER,
volume_pct INTEGER,
mic_source TEXT,
mic_volume_pct INTEGER
mic_volume_pct INTEGER,
loopback_no_mix INTEGER NOT NULL DEFAULT 0
);",
)?;
migrate_add_loopback_no_mix(&conn)?;
Ok(conn)
}
/// `CREATE TABLE IF NOT EXISTS` above only covers fresh databases; an
/// existing `vmic.db` from before `loopback_no_mix` existed needs the column
/// added explicitly. Safe to call unconditionally - a no-op once the column
/// is present either way.
fn migrate_add_loopback_no_mix(conn: &Connection) -> Result<()> {
let has_column = conn
.prepare("PRAGMA table_info(vmics)")?
.query_map([], |row| row.get::<_, String>(1))?
.filter_map(std::result::Result::ok)
.any(|c| c == "loopback_no_mix");
if !has_column {
conn.execute(
"ALTER TABLE vmics ADD COLUMN loopback_no_mix INTEGER NOT NULL DEFAULT 0",
[],
)?;
}
Ok(())
}
impl VmicState {
fn from_row(row: &rusqlite::Row) -> rusqlite::Result<Self> {
Ok(Self {
@@ -112,6 +138,7 @@ impl VmicState {
volume_pct: row.get("volume_pct")?,
mic_source: row.get("mic_source")?,
mic_volume_pct: row.get("mic_volume_pct")?,
loopback_no_mix: row.get("loopback_no_mix")?,
})
}
@@ -135,8 +162,8 @@ impl VmicState {
"INSERT INTO vmics (
name, sink_name, mid_name, mix_name, source_name,
sink_node_id, mix_node_id, sink_pid, mix_pid, loopback_id, volume_pct,
mic_source, mic_volume_pct
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)
mic_source, mic_volume_pct, loopback_no_mix
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)
ON CONFLICT(name) DO UPDATE SET
sink_name = excluded.sink_name,
mid_name = excluded.mid_name,
@@ -149,7 +176,8 @@ impl VmicState {
loopback_id = excluded.loopback_id,
volume_pct = excluded.volume_pct,
mic_source = excluded.mic_source,
mic_volume_pct = excluded.mic_volume_pct",
mic_volume_pct = excluded.mic_volume_pct,
loopback_no_mix = excluded.loopback_no_mix",
params![
self.name,
self.sink_name,
@@ -164,6 +192,7 @@ impl VmicState {
self.volume_pct,
self.mic_source,
self.mic_volume_pct,
self.loopback_no_mix,
],
)?;
Ok(())